2016-06-03 15:32:46 +00:00
|
|
|
package transfer
|
|
|
|
|
|
|
|
import (
|
2016-07-05 14:25:45 +00:00
|
|
|
"fmt"
|
2016-06-03 15:32:46 +00:00
|
|
|
"sync"
|
2016-07-05 14:25:45 +00:00
|
|
|
"time"
|
2016-06-03 15:32:46 +00:00
|
|
|
|
2016-08-18 20:20:33 +00:00
|
|
|
"github.com/github/git-lfs/errors"
|
2016-06-03 15:32:46 +00:00
|
|
|
"github.com/rubyist/tracerx"
|
|
|
|
)
|
|
|
|
|
2016-07-06 14:18:11 +00:00
|
|
|
const (
|
|
|
|
// objectExpirationGracePeriod is the grace period applied to objects
|
|
|
|
// when checking whether or not they have expired.
|
|
|
|
objectExpirationGracePeriod = 5 * time.Second
|
|
|
|
)
|
|
|
|
|
2016-06-03 15:32:46 +00:00
|
|
|
// 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 *Transfer
|
|
|
|
cb TransferProgressCallback
|
|
|
|
outChan chan TransferResult
|
|
|
|
// WaitGroup to sync the completion of all workers
|
|
|
|
workerWait 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 {
|
2016-07-08 15:30:46 +00:00
|
|
|
// 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
|
2016-07-08 15:31:55 +00:00
|
|
|
// Implementations can clean up per-worker resources here, context is as returned from WorkerStarting
|
2016-07-08 15:30:46 +00:00
|
|
|
WorkerEnding(workerNum int, ctx interface{})
|
2016-07-08 15:31:55 +00:00
|
|
|
// DoTransfer performs a single transfer within a worker. ctx is any context returned from WorkerStarting
|
2016-07-08 15:30:46 +00:00
|
|
|
DoTransfer(ctx interface{}, t *Transfer, cb TransferProgressCallback, authOkFunc func()) error
|
2016-06-03 15:32:46 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
func newAdapterBase(name string, dir Direction, ti transferImplementation) *adapterBase {
|
|
|
|
return &adapterBase{name: name, direction: dir, transferImpl: ti}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (a *adapterBase) Name() string {
|
|
|
|
return a.name
|
|
|
|
}
|
|
|
|
|
|
|
|
func (a *adapterBase) Direction() Direction {
|
|
|
|
return a.direction
|
|
|
|
}
|
|
|
|
|
|
|
|
func (a *adapterBase) Begin(maxConcurrency int, cb TransferProgressCallback, completion chan TransferResult) error {
|
|
|
|
a.cb = cb
|
|
|
|
a.outChan = completion
|
|
|
|
a.jobChan = make(chan *Transfer, 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++ {
|
2016-07-08 15:30:46 +00:00
|
|
|
ctx, err := a.transferImpl.WorkerStarting(i)
|
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
|
|
|
go a.worker(i, ctx)
|
2016-06-03 15:32:46 +00:00
|
|
|
}
|
|
|
|
tracerx.Printf("xfer: adapter %q started", a.Name())
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
func (a *adapterBase) Add(t *Transfer) {
|
|
|
|
tracerx.Printf("xfer: adapter %q Add() for %q", a.Name(), t.Object.Oid)
|
|
|
|
a.jobChan <- t
|
|
|
|
}
|
|
|
|
|
|
|
|
func (a *adapterBase) End() {
|
|
|
|
tracerx.Printf("xfer: adapter %q End()", a.Name())
|
|
|
|
close(a.jobChan)
|
|
|
|
// wait for all transfers to complete
|
|
|
|
a.workerWait.Wait()
|
|
|
|
if a.outChan != nil {
|
|
|
|
close(a.outChan)
|
|
|
|
}
|
|
|
|
tracerx.Printf("xfer: adapter %q stopped", a.Name())
|
|
|
|
}
|
|
|
|
|
|
|
|
// worker function, many of these run per adapter
|
2016-07-08 15:30:46 +00:00
|
|
|
func (a *adapterBase) worker(workerNum int, ctx interface{}) {
|
2016-06-03 15:32:46 +00:00
|
|
|
|
|
|
|
tracerx.Printf("xfer: adapter %q worker %d starting", a.Name(), workerNum)
|
|
|
|
waitForAuth := workerNum > 0
|
|
|
|
signalAuthOnResponse := workerNum == 0
|
|
|
|
|
|
|
|
// First worker is the only one allowed to start immediately
|
|
|
|
// The rest wait until successful response from 1st worker to
|
|
|
|
// make sure only 1 login prompt is presented if necessary
|
|
|
|
// Deliberately outside jobChan processing so we know worker 0 will process 1st item
|
|
|
|
if waitForAuth {
|
|
|
|
tracerx.Printf("xfer: adapter %q worker %d waiting for Auth", a.Name(), workerNum)
|
|
|
|
a.authWait.Wait()
|
|
|
|
tracerx.Printf("xfer: adapter %q worker %d auth signal received", a.Name(), workerNum)
|
|
|
|
}
|
|
|
|
|
|
|
|
for t := range a.jobChan {
|
|
|
|
var authCallback func()
|
|
|
|
if signalAuthOnResponse {
|
|
|
|
authCallback = func() {
|
|
|
|
a.authWait.Done()
|
2016-06-07 11:54:14 +00:00
|
|
|
signalAuthOnResponse = false
|
2016-06-03 15:32:46 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
tracerx.Printf("xfer: adapter %q worker %d processing job for %q", a.Name(), workerNum, t.Object.Oid)
|
2016-07-05 14:25:45 +00:00
|
|
|
|
2016-06-03 15:32:46 +00:00
|
|
|
// Actual transfer happens here
|
2016-07-05 14:25:45 +00:00
|
|
|
var err error
|
2016-07-06 14:18:11 +00:00
|
|
|
if t.Object.IsExpired(time.Now().Add(objectExpirationGracePeriod)) {
|
2016-07-05 14:25:45 +00:00
|
|
|
tracerx.Printf("xfer: adapter %q worker %d found job for %q expired, retrying...", a.Name(), workerNum, t.Object.Oid)
|
2016-08-18 20:20:33 +00:00
|
|
|
err = errors.NewRetriableError(fmt.Errorf("lfs/transfer: object %q has expired", t.Object.Oid))
|
2016-07-13 18:58:55 +00:00
|
|
|
} else if t.Object.Size < 0 {
|
|
|
|
tracerx.Printf("xfer: adapter %q worker %d found invalid size for %q (got: %d), retrying...", a.Name(), workerNum, t.Object.Oid, t.Object.Size)
|
|
|
|
err = fmt.Errorf("Git LFS: object %q has invalid size (got: %d)", t.Object.Oid, t.Object.Size)
|
2016-07-05 14:25:45 +00:00
|
|
|
} else {
|
2016-07-08 15:30:46 +00:00
|
|
|
err = a.transferImpl.DoTransfer(ctx, t, a.cb, authCallback)
|
2016-07-05 14:25:45 +00:00
|
|
|
}
|
2016-06-03 15:32:46 +00:00
|
|
|
|
|
|
|
if a.outChan != nil {
|
|
|
|
res := TransferResult{t, err}
|
|
|
|
a.outChan <- res
|
|
|
|
}
|
|
|
|
|
|
|
|
tracerx.Printf("xfer: adapter %q worker %d finished job for %q", a.Name(), workerNum, t.Object.Oid)
|
|
|
|
}
|
|
|
|
// This will only happen if no jobs were submitted; just wake up all workers to finish
|
|
|
|
if signalAuthOnResponse {
|
|
|
|
a.authWait.Done()
|
|
|
|
}
|
|
|
|
tracerx.Printf("xfer: adapter %q worker %d stopping", a.Name(), workerNum)
|
2016-07-08 15:30:46 +00:00
|
|
|
a.transferImpl.WorkerEnding(workerNum, ctx)
|
2016-06-03 15:32:46 +00:00
|
|
|
a.workerWait.Done()
|
|
|
|
}
|
2016-06-08 16:03:05 +00:00
|
|
|
|
|
|
|
func advanceCallbackProgress(cb TransferProgressCallback, t *Transfer, numBytes int64) {
|
|
|
|
if cb != nil {
|
|
|
|
// Must split into max int sizes since read count is int
|
|
|
|
const maxInt = int(^uint(0) >> 1)
|
|
|
|
for read := int64(0); read < numBytes; {
|
|
|
|
remainder := numBytes - read
|
|
|
|
if remainder > int64(maxInt) {
|
|
|
|
read += int64(maxInt)
|
|
|
|
cb(t.Name, t.Object.Size, read, maxInt)
|
|
|
|
} else {
|
|
|
|
read += remainder
|
|
|
|
cb(t.Name, t.Object.Size, read, int(remainder))
|
|
|
|
}
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|