<commit_before>package worker

import (
	"sort"
	"sync"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/pborman/uuid"
	"github.com/travis-ci/worker/backend"
	"github.com/travis-ci/worker/context"
	gocontext "golang.org/x/net/context"
)

// A ProcessorPool spins up multiple Processors handling build jobs from the
// same queue.
type ProcessorPool struct {
	Context   gocontext.Context
	Provider  backend.Provider
	Generator BuildScriptGenerator
	Canceller Canceller
	Hostname  string

	HardTimeout, LogTimeout, ScriptUploadTimeout, StartupTimeout time.Duration

	SkipShutdownOnLogTimeout bool

	queue          JobQueue
	poolErrors     []error
	processorsLock *sync.Mutex
	processors     []*Processor
	processorsWG   *sync.WaitGroup
	pauseCount     int
}

type ProcessorPoolConfig struct {
	Hostname string
	Context  gocontext.Context

	HardTimeout, LogTimeout, ScriptUploadTimeout, StartupTimeout time.Duration
}

// NewProcessorPool creates a new processor pool using the given arguments.
func NewProcessorPool(ppc *ProcessorPoolConfig,
	provider backend.Provider, generator BuildScriptGenerator,
	canceller Canceller) *ProcessorPool {

	return &ProcessorPool{
		Hostname: ppc.Hostname,
		Context:  ppc.Context,

		HardTimeout:         ppc.HardTimeout,
		LogTimeout:          ppc.LogTimeout,
		ScriptUploadTimeout: ppc.ScriptUploadTimeout,
		StartupTimeout:      ppc.StartupTimeout,

		Provider:  provider,
		Generator: generator,
		Canceller: canceller,

		processorsLock: &sync.Mutex{},
		processorsWG:   &sync.WaitGroup{},
	}
}

// Each loops through all the processors in the pool and calls the given
// function for each of them, passing in the index and the processor. The order
// of the processors is the same for the same set of processors.
func (p *ProcessorPool) Each(f func(int, *Processor)) {
	procIDs := []string{}
	procsByID := map[string]*Processor{}

	for _, proc := range p.processors {
		id := proc.ID.String()
		procIDs = append(procIDs, id)
		procsByID[id] = proc
	}

	sort.Strings(procIDs)

	for i, procID := range procIDs {
		f(i, procsByID[procID])
	}
}

// Size returns the number of processors in the pool
func (p *ProcessorPool) Size() int {
	return len(p.processors)
}

// Run starts up a number of processors and connects them to the given queue.
// This method stalls until all processors have finished.
func (p *ProcessorPool) Run(poolSize int, queue JobQueue) error {
	p.queue = queue
	p.poolErrors = []error{}

	for i := 0; i < poolSize; i++ {
		p.Incr()
	}

	if len(p.poolErrors) > 0 {
		context.LoggerFromContext(p.Context).WithFields(logrus.Fields{
			"pool_errors": p.poolErrors,
		}).Panic("failed to populate pool")
	}

	p.processorsWG.Wait()

	return nil
}

// GracefulShutdown causes each processor in the pool to start its graceful
// shutdown.
func (p *ProcessorPool) GracefulShutdown(togglePause bool) {
	p.processorsLock.Lock()
	defer p.processorsLock.Unlock()

	log := context.LoggerFromContext(p.Context)

	if togglePause {
		p.pauseCount++

		if p.pauseCount == 1 {
			log.Info("incrementing wait group for pause")
			p.processorsWG.Add(1)
		} else if p.pauseCount == 2 {
			log.Info("finishing wait group to unpause")
			p.processorsWG.Done()
		} else if p.pauseCount > 2 {
			return
		}
	}

	for _, processor := range p.processors {
		processor.GracefulShutdown()
	}
}

// Incr adds a single running processor to the pool
func (p *ProcessorPool) Incr() {
	p.processorsWG.Add(1)
	go func() {
		defer p.processorsWG.Done()
		err := p.runProcessor(p.queue)
		if err != nil {
			p.poolErrors = append(p.poolErrors, err)
			return
		}
	}()
}

// Decr pops a processor out of the pool and issues a graceful shutdown
func (p *ProcessorPool) Decr() {
	if len(p.processors) == 0 {
		return
	}

	var proc *Processor
	proc, p.processors = p.processors[len(p.processors)-1], p.processors[:len(p.processors)-1]
	proc.GracefulShutdown()
}

func (p *ProcessorPool) runProcessor(queue JobQueue) error {
	processorUUID := uuid.NewRandom()
	ctx := context.FromProcessor(p.Context, processorUUID.String())

	jobsChan, err := queue.Jobs(ctx)
	if err != nil {
		context.LoggerFromContext(p.Context).WithField("err", err).Error("couldn't create jobs channel")
		return err
	}

	proc, err := NewProcessor(ctx, p.Hostname,
		jobsChan, p.Provider, p.Generator, p.Canceller,
		p.HardTimeout, p.LogTimeout, p.ScriptUploadTimeout, p.StartupTimeout)

	if err != nil {
		context.LoggerFromContext(p.Context).WithField("err", err).Error("couldn't create processor")
		return err
	}

	proc.SkipShutdownOnLogTimeout = p.SkipShutdownOnLogTimeout

	p.processorsLock.Lock()
	p.processors = append(p.processors, proc)
	p.processorsLock.Unlock()

	proc.Run()
	return nil
}
<commit_msg>Roll back struct/pointer bits :fr: :fries:<commit_after>package worker

import (
	"sort"
	"sync"
	"time"

	"github.com/Sirupsen/logrus"
	"github.com/pborman/uuid"
	"github.com/travis-ci/worker/backend"
	"github.com/travis-ci/worker/context"
	gocontext "golang.org/x/net/context"
)

// A ProcessorPool spins up multiple Processors handling build jobs from the
// same queue.
type ProcessorPool struct {
	Context   gocontext.Context
	Provider  backend.Provider
	Generator BuildScriptGenerator
	Canceller Canceller
	Hostname  string

	HardTimeout, LogTimeout, ScriptUploadTimeout, StartupTimeout time.Duration

	SkipShutdownOnLogTimeout bool

	queue          JobQueue
	poolErrors     []error
	processorsLock sync.Mutex
	processors     []*Processor
	processorsWG   sync.WaitGroup
	pauseCount     int
}

type ProcessorPoolConfig struct {
	Hostname string
	Context  gocontext.Context

	HardTimeout, LogTimeout, ScriptUploadTimeout, StartupTimeout time.Duration
}

// NewProcessorPool creates a new processor pool using the given arguments.
func NewProcessorPool(ppc *ProcessorPoolConfig,
	provider backend.Provider, generator BuildScriptGenerator,
	canceller Canceller) *ProcessorPool {

	return &ProcessorPool{
		Hostname: ppc.Hostname,
		Context:  ppc.Context,

		HardTimeout:         ppc.HardTimeout,
		LogTimeout:          ppc.LogTimeout,
		ScriptUploadTimeout: ppc.ScriptUploadTimeout,
		StartupTimeout:      ppc.StartupTimeout,

		Provider:  provider,
		Generator: generator,
		Canceller: canceller,
	}
}

// Each loops through all the processors in the pool and calls the given
// function for each of them, passing in the index and the processor. The order
// of the processors is the same for the same set of processors.
func (p *ProcessorPool) Each(f func(int, *Processor)) {
	procIDs := []string{}
	procsByID := map[string]*Processor{}

	for _, proc := range p.processors {
		id := proc.ID.String()
		procIDs = append(procIDs, id)
		procsByID[id] = proc
	}

	sort.Strings(procIDs)

	for i, procID := range procIDs {
		f(i, procsByID[procID])
	}
}

// Size returns the number of processors in the pool
func (p *ProcessorPool) Size() int {
	return len(p.processors)
}

// Run starts up a number of processors and connects them to the given queue.
// This method stalls until all processors have finished.
func (p *ProcessorPool) Run(poolSize int, queue JobQueue) error {
	p.queue = queue
	p.poolErrors = []error{}

	for i := 0; i < poolSize; i++ {
		p.Incr()
	}

	if len(p.poolErrors) > 0 {
		context.LoggerFromContext(p.Context).WithFields(logrus.Fields{
			"pool_errors": p.poolErrors,
		}).Panic("failed to populate pool")
	}

	p.processorsWG.Wait()

	return nil
}

// GracefulShutdown causes each processor in the pool to start its graceful
// shutdown.
func (p *ProcessorPool) GracefulShutdown(togglePause bool) {
	p.processorsLock.Lock()
	defer p.processorsLock.Unlock()

	log := context.LoggerFromContext(p.Context)

	if togglePause {
		p.pauseCount++

		if p.pauseCount == 1 {
			log.Info("incrementing wait group for pause")
			p.processorsWG.Add(1)
		} else if p.pauseCount == 2 {
			log.Info("finishing wait group to unpause")
			p.processorsWG.Done()
		} else if p.pauseCount > 2 {
			return
		}
	}

	for _, processor := range p.processors {
		processor.GracefulShutdown()
	}
}

// Incr adds a single running processor to the pool
func (p *ProcessorPool) Incr() {
	p.processorsWG.Add(1)
	go func() {
		defer p.processorsWG.Done()
		err := p.runProcessor(p.queue)
		if err != nil {
			p.poolErrors = append(p.poolErrors, err)
			return
		}
	}()
}

// Decr pops a processor out of the pool and issues a graceful shutdown
func (p *ProcessorPool) Decr() {
	if len(p.processors) == 0 {
		return
	}

	var proc *Processor
	proc, p.processors = p.processors[len(p.processors)-1], p.processors[:len(p.processors)-1]
	proc.GracefulShutdown()
}

func (p *ProcessorPool) runProcessor(queue JobQueue) error {
	processorUUID := uuid.NewRandom()
	ctx := context.FromProcessor(p.Context, processorUUID.String())

	jobsChan, err := queue.Jobs(ctx)
	if err != nil {
		context.LoggerFromContext(p.Context).WithField("err", err).Error("couldn't create jobs channel")
		return err
	}

	proc, err := NewProcessor(ctx, p.Hostname,
		jobsChan, p.Provider, p.Generator, p.Canceller,
		p.HardTimeout, p.LogTimeout, p.ScriptUploadTimeout, p.StartupTimeout)

	if err != nil {
		context.LoggerFromContext(p.Context).WithField("err", err).Error("couldn't create processor")
		return err
	}

	proc.SkipShutdownOnLogTimeout = p.SkipShutdownOnLogTimeout

	p.processorsLock.Lock()
	p.processors = append(p.processors, proc)
	p.processorsLock.Unlock()

	proc.Run()
	return nil
}

{'fraction_non_alphanumeric': 0.11076693014313181, 'fraction_numerical': 0.0021362956633198035, 'mean_word_length': 4.094124047878128, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 7, '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': '4778190', 'n_tokens_mistral': 3285, 'n_tokens_neox': 3084, 'n_words': 833}