<commit_before>package tq

import (
	"fmt"
	"sync"
	"time"

	"github.com/git-lfs/git-lfs/errors"
	"github.com/rubyist/tracerx"
)

const (
	// objectExpirationToTransfer is the duration we expect to have passed
	// from the time that the object's expires_at property is checked to
	// when the transfer is executed.
	objectExpirationToTransfer = 5 * time.Second
)

// adapterBase implements the common functionality for core adapters which
// process transfers with N workers handling an oid each, and which wait for
// authentication to succeed on one worker before proceeding
type adapterBase struct {
	name         string
	direction    Direction
	transferImpl transferImplementation
	jobChan      chan *job
	cb           ProgressCallback
	// WaitGroup to sync the completion of all workers
	workerWait sync.WaitGroup
	// WaitGroup to sync the completion of all in-flight jobs
	jobWait *sync.WaitGroup
	// WaitGroup to serialise the first transfer response to perform login if needed
	authWait sync.WaitGroup
}

// transferImplementation must be implemented to provide the actual upload/download
// implementation for all core transfer approaches that use adapterBase for
// convenience. This function will be called on multiple goroutines so it
// must be either stateless or thread safe. However it will never be called
// for the same oid in parallel.
// If authOkFunc is not nil, implementations must call it as early as possible
// when authentication succeeded, before the whole file content is transferred
type transferImplementation interface {
	// WorkerStarting is called when a worker goroutine starts to process jobs
	// Implementations can run some startup logic here & return some context if needed
	WorkerStarting(workerNum int) (interface{}, error)
	// WorkerEnding is called when a worker goroutine is shutting down
	// Implementations can clean up per-worker resources here, context is as returned from WorkerStarting
	WorkerEnding(workerNum int, ctx interface{})
	// DoTransfer performs a single transfer within a worker. ctx is any context returned from WorkerStarting
	DoTransfer(ctx interface{}, t *Transfer, cb ProgressCallback, authOkFunc func()) error
}

func newAdapterBase(name string, dir Direction, ti transferImplementation) *adapterBase {
	return &adapterBase{
		name:         name,
		direction:    dir,
		transferImpl: ti,

		jobWait: new(sync.WaitGroup),
	}
}

func (a *adapterBase) Name() string {
	return a.name
}

func (a *adapterBase) Direction() Direction {
	return a.direction
}

func (a *adapterBase) Begin(maxConcurrency int, cb ProgressCallback) error {
	a.cb = cb
	a.jobChan = make(chan *job, 100)

	tracerx.Printf("xfer: adapter %q Begin() with %d workers", a.Name(), maxConcurrency)

	a.workerWait.Add(maxConcurrency)
	a.authWait.Add(1)
	for i := 0; i < maxConcurrency; i++ {
		ctx, err := a.transferImpl.WorkerStarting(i)
		if err != nil {
			return err
		}
		go a.worker(i, ctx)
	}
	tracerx.Printf("xfer: adapter %q started", a.Name())
	return nil
}

type job struct {
	T *Transfer

	results chan<- TransferResult
	wg      *sync.WaitGroup
}

func (j *job) Done(err error) {
	j.results <- TransferResult{j.T, err}
	j.wg.Done()
}

func (a *adapterBase) Add(transfers ...*Transfer) <-chan TransferResult {
	results := make(chan TransferResult, len(transfers))

	a.jobWait.Add(len(transfers))

	go func() {
		defer close(results)

		for _, t := range transfers {
			// BUG(taylor): End() is race-y here, and can close
			// jobChan before we want it to
			a.jobChan <- &job{t, results, a.jobWait}
		}

		a.jobWait.Wait()
	}()

	return results
}

func (a *adapterBase) End() {
	tracerx.Printf("xfer: adapter %q End()", a.Name())

	a.jobWait.Wait()
	close(a.jobChan)

	// wait for all transfers to complete
	a.workerWait.Wait()

	tracerx.Printf("xfer: adapter %q stopped", a.Name())
}

// worker function, many of these run per adapter
func (a *adapterBase) worker(workerNum int, ctx interface{}) {<commit_msg>tq: remove BUG comment that has since been fixed<commit_after>package tq

import (
	"fmt"
	"sync"
	"time"

	"github.com/git-lfs/git-lfs/errors"
	"github.com/rubyist/tracerx"
)

const (
	// objectExpirationToTransfer is the duration we expect to have passed
	// from the time that the object's expires_at property is checked to
	// when the transfer is executed.
	objectExpirationToTransfer = 5 * time.Second
)

// adapterBase implements the common functionality for core adapters which
// process transfers with N workers handling an oid each, and which wait for
// authentication to succeed on one worker before proceeding
type adapterBase struct {
	name         string
	direction    Direction
	transferImpl transferImplementation
	jobChan      chan *job
	cb           ProgressCallback
	// WaitGroup to sync the completion of all workers
	workerWait sync.WaitGroup
	// WaitGroup to sync the completion of all in-flight jobs
	jobWait *sync.WaitGroup
	// WaitGroup to serialise the first transfer response to perform login if needed
	authWait sync.WaitGroup
}

// transferImplementation must be implemented to provide the actual upload/download
// implementation for all core transfer approaches that use adapterBase for
// convenience. This function will be called on multiple goroutines so it
// must be either stateless or thread safe. However it will never be called
// for the same oid in parallel.
// If authOkFunc is not nil, implementations must call it as early as possible
// when authentication succeeded, before the whole file content is transferred
type transferImplementation interface {
	// WorkerStarting is called when a worker goroutine starts to process jobs
	// Implementations can run some startup logic here & return some context if needed
	WorkerStarting(workerNum int) (interface{}, error)
	// WorkerEnding is called when a worker goroutine is shutting down
	// Implementations can clean up per-worker resources here, context is as returned from WorkerStarting
	WorkerEnding(workerNum int, ctx interface{})
	// DoTransfer performs a single transfer within a worker. ctx is any context returned from WorkerStarting
	DoTransfer(ctx interface{}, t *Transfer, cb ProgressCallback, authOkFunc func()) error
}

func newAdapterBase(name string, dir Direction, ti transferImplementation) *adapterBase {
	return &adapterBase{
		name:         name,
		direction:    dir,
		transferImpl: ti,

		jobWait: new(sync.WaitGroup),
	}
}

func (a *adapterBase) Name() string {
	return a.name
}

func (a *adapterBase) Direction() Direction {
	return a.direction
}

func (a *adapterBase) Begin(maxConcurrency int, cb ProgressCallback) error {
	a.cb = cb
	a.jobChan = make(chan *job, 100)

	tracerx.Printf("xfer: adapter %q Begin() with %d workers", a.Name(), maxConcurrency)

	a.workerWait.Add(maxConcurrency)
	a.authWait.Add(1)
	for i := 0; i < maxConcurrency; i++ {
		ctx, err := a.transferImpl.WorkerStarting(i)
		if err != nil {
			return err
		}
		go a.worker(i, ctx)
	}
	tracerx.Printf("xfer: adapter %q started", a.Name())
	return nil
}

type job struct {
	T *Transfer

	results chan<- TransferResult
	wg      *sync.WaitGroup
}

func (j *job) Done(err error) {
	j.results <- TransferResult{j.T, err}
	j.wg.Done()
}

func (a *adapterBase) Add(transfers ...*Transfer) <-chan TransferResult {
	results := make(chan TransferResult, len(transfers))

	a.jobWait.Add(len(transfers))

	go func() {
		defer close(results)

		for _, t := range transfers {
			a.jobChan <- &job{t, results, a.jobWait}
		}

		a.jobWait.Wait()
	}()

	return results
}

func (a *adapterBase) End() {
	tracerx.Printf("xfer: adapter %q End()", a.Name())

	a.jobWait.Wait()
	close(a.jobChan)

	// wait for all transfers to complete
	a.workerWait.Wait()

	tracerx.Printf("xfer: adapter %q stopped", a.Name())
}

// worker function, many of these run per adapter
func (a *adapterBase) worker(workerNum int, ctx interface{}) {
{'fraction_non_alphanumeric': 0.09326028793476876, 'fraction_numerical': 0.0015288571792585043, 'mean_word_length': 3.9872935196950445, 'pattern_counts': {'":': 0, '<': 13, '<?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': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1954127', 'n_tokens_mistral': 2440, 'n_tokens_neox': 2315, 'n_words': 890}