<commit_before>// Copyright 2012, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

/*
The agent listens on an action node for new actions to perform.

It passes them off to a separate action process. Even though some
actions could be completed inline very quickly, the external process
makes it easy to track and interrupt complex actions that may wedge
due to external circumstances.
*/

package tabletmanager

import (
	"fmt"
	"net"
	"os"
	"os/exec"
	"path"
	"sync"

	log "github.com/golang/glog"
	"github.com/youtube/vitess/go/netutil"
	"github.com/youtube/vitess/go/vt/dbconfigs"
	"github.com/youtube/vitess/go/vt/env"
	"github.com/youtube/vitess/go/vt/logutil"
	"github.com/youtube/vitess/go/vt/tabletserver"
	"github.com/youtube/vitess/go/vt/topo"
)

// Each TabletChangeCallback must be idempotent and "threadsafe".  The
// agent will execute these in a new goroutine each time a change is
// triggered. We won't run two in parallel.
type TabletChangeCallback func(oldTablet, newTablet topo.Tablet)

type tabletChangeItem struct {
	oldTablet topo.Tablet
	newTablet topo.Tablet
	context   string
}

type ActionAgent struct {
	ts                topo.Server
	tabletAlias       topo.TabletAlias
	vtActionBinFile   string // path to vtaction binary
	MycnfFile         string // my.cnf file
	DbCredentialsFile string // File that contains db credentials

	done chan struct{} // closed when we are done.

	// actionMutex is there to run only one action at a time. If
	// both agent.actionMutex and agent.mutex needs to be taken,
	// take actionMutex first.
	actionMutex sync.Mutex // to run only one action at a time

	// mutex is protecting the rest of the members
	mutex           sync.Mutex
	changeCallbacks []TabletChangeCallback
	changeItems     chan tabletChangeItem
	_tablet         *topo.TabletInfo
}

func NewActionAgent(topoServer topo.Server, tabletAlias topo.TabletAlias, mycnfFile, dbCredentialsFile string) (*ActionAgent, error) {
	return &ActionAgent{
		ts:                topoServer,
		tabletAlias:       tabletAlias,
		MycnfFile:         mycnfFile,
		DbCredentialsFile: dbCredentialsFile,
		done:              make(chan struct{}),
		changeCallbacks:   make([]TabletChangeCallback, 0, 8),
		changeItems:       make(chan tabletChangeItem, 100),
	}, nil
}

func (agent *ActionAgent) AddChangeCallback(f TabletChangeCallback) {
	agent.mutex.Lock()
	agent.changeCallbacks = append(agent.changeCallbacks, f)
	agent.mutex.Unlock()
}

func (agent *ActionAgent) runChangeCallbacks(oldTablet *topo.Tablet, context string) {
	agent.mutex.Lock()
	// Access directly since we have the lock.
	newTablet := agent._tablet.Tablet
	agent.changeItems <- tabletChangeItem{oldTablet: *oldTablet, newTablet: *newTablet, context: context}
	log.Infof("Queued tablet callback: %v", context)
	agent.mutex.Unlock()
}

func (agent *ActionAgent) executeCallbacksLoop() {
	for {
		select {
		case changeItem := <-agent.changeItems:
			var wg sync.WaitGroup
			agent.mutex.Lock()
			for _, f := range agent.changeCallbacks {
				log.Infof("Running tablet callback: %v %v", changeItem.context, f)
				wg.Add(1)
				go func(f TabletChangeCallback, oldTablet, newTablet topo.Tablet) {
					defer wg.Done()
					f(oldTablet, newTablet)
				}(f, changeItem.oldTablet, changeItem.newTablet)
			}
			agent.mutex.Unlock()
			wg.Wait()
		case <-agent.done:
			return
		}
	}
}

func (agent *ActionAgent) readTablet() error {
	tablet, err := agent.ts.GetTablet(agent.tabletAlias)
	if err != nil {
		return err
	}
	agent.mutex.Lock()
	agent._tablet = tablet
	agent.mutex.Unlock()
	return nil<commit_msg>Show time delayed from enqueued to executed for vttablet after-action callback<commit_after>// Copyright 2012, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

/*
The agent listens on an action node for new actions to perform.

It passes them off to a separate action process. Even though some
actions could be completed inline very quickly, the external process
makes it easy to track and interrupt complex actions that may wedge
due to external circumstances.
*/

package tabletmanager

import (
	"fmt"
	"net"
	"os"
	"os/exec"
	"path"
	"sync"
	"time"

	log "github.com/golang/glog"
	"github.com/youtube/vitess/go/netutil"
	"github.com/youtube/vitess/go/vt/dbconfigs"
	"github.com/youtube/vitess/go/vt/env"
	"github.com/youtube/vitess/go/vt/logutil"
	"github.com/youtube/vitess/go/vt/tabletserver"
	"github.com/youtube/vitess/go/vt/topo"
)

// Each TabletChangeCallback must be idempotent and "threadsafe".  The
// agent will execute these in a new goroutine each time a change is
// triggered. We won't run two in parallel.
type TabletChangeCallback func(oldTablet, newTablet topo.Tablet)

type tabletChangeItem struct {
	oldTablet  topo.Tablet
	newTablet  topo.Tablet
	context    string
	queuedTime time.Time
}

type ActionAgent struct {
	ts                topo.Server
	tabletAlias       topo.TabletAlias
	vtActionBinFile   string // path to vtaction binary
	MycnfFile         string // my.cnf file
	DbCredentialsFile string // File that contains db credentials

	done chan struct{} // closed when we are done.

	// actionMutex is there to run only one action at a time. If
	// both agent.actionMutex and agent.mutex needs to be taken,
	// take actionMutex first.
	actionMutex sync.Mutex // to run only one action at a time

	// mutex is protecting the rest of the members
	mutex           sync.Mutex
	changeCallbacks []TabletChangeCallback
	changeItems     chan tabletChangeItem
	_tablet         *topo.TabletInfo
}

func NewActionAgent(topoServer topo.Server, tabletAlias topo.TabletAlias, mycnfFile, dbCredentialsFile string) (*ActionAgent, error) {
	return &ActionAgent{
		ts:                topoServer,
		tabletAlias:       tabletAlias,
		MycnfFile:         mycnfFile,
		DbCredentialsFile: dbCredentialsFile,
		done:              make(chan struct{}),
		changeCallbacks:   make([]TabletChangeCallback, 0, 8),
		changeItems:       make(chan tabletChangeItem, 100),
	}, nil
}

func (agent *ActionAgent) AddChangeCallback(f TabletChangeCallback) {
	agent.mutex.Lock()
	agent.changeCallbacks = append(agent.changeCallbacks, f)
	agent.mutex.Unlock()
}

func (agent *ActionAgent) runChangeCallbacks(oldTablet *topo.Tablet, context string) {
	agent.mutex.Lock()
	// Access directly since we have the lock.
	newTablet := agent._tablet.Tablet
	agent.changeItems <- tabletChangeItem{oldTablet: *oldTablet, newTablet: *newTablet, context: context, queuedTime: time.Now()}
	log.Infof("Queued tablet callback: %v", context)
	agent.mutex.Unlock()
}

func (agent *ActionAgent) executeCallbacksLoop() {
	for {
		select {
		case changeItem := <-agent.changeItems:
			var wg sync.WaitGroup
			agent.mutex.Lock()
			for _, f := range agent.changeCallbacks {
				log.Infof("Running tablet callback after %v: %v %v", time.Now().Sub(changeItem.queuedTime), changeItem.context, f)
				wg.Add(1)
				go func(f TabletChangeCallback, oldTablet, newTablet topo.Tablet) {
					defer wg.Done()
					f(oldTablet, newTablet)
				}(f, changeItem.oldTablet, changeItem.newTablet)
			}
			agent.mutex.Unlock()
			wg.Wait()
		case <-agent.done:
			return
		}
	}
}

func (agent *ActionAgent) readTablet() error {
	tablet, err := agent.ts.GetTablet(agent.tabletAlias)
	if err != nil {
		return err
	}
	agent.mutex.Lock()
	agent._tablet = tablet
	agent.mutex.Unlock()
	return nil
{'fraction_non_alphanumeric': 0.09730738469741403, 'fraction_numerical': 0.002665955745134631, 'mean_word_length': 3.671855541718555, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1049162', 'n_tokens_mistral': 2551, 'n_tokens_neox': 2413, 'n_words': 721}