2016-12-12 00:20:14 +00:00
|
|
|
package tq
|
2015-05-21 16:36:49 +00:00
|
|
|
|
|
|
|
import (
|
2016-12-11 01:04:03 +00:00
|
|
|
"sort"
|
2015-05-21 16:36:49 +00:00
|
|
|
"sync"
|
2015-05-21 17:47:52 +00:00
|
|
|
|
2016-11-15 17:01:18 +00:00
|
|
|
"github.com/git-lfs/git-lfs/errors"
|
2017-01-04 21:46:30 +00:00
|
|
|
"github.com/git-lfs/git-lfs/lfsapi"
|
2016-11-15 17:01:18 +00:00
|
|
|
"github.com/git-lfs/git-lfs/progress"
|
2016-05-23 18:02:27 +00:00
|
|
|
"github.com/rubyist/tracerx"
|
2015-05-21 16:36:49 +00:00
|
|
|
)
|
|
|
|
|
2015-07-09 19:59:25 +00:00
|
|
|
const (
|
2016-12-13 21:14:30 +00:00
|
|
|
defaultBatchSize = 100
|
2015-07-09 19:59:25 +00:00
|
|
|
)
|
|
|
|
|
2016-09-28 18:25:37 +00:00
|
|
|
type retryCounter struct {
|
|
|
|
MaxRetries int `git:"lfs.transfer.maxretries"`
|
|
|
|
|
|
|
|
// cmu guards count
|
|
|
|
cmu sync.Mutex
|
|
|
|
// count maps OIDs to number of retry attempts
|
|
|
|
count map[string]int
|
|
|
|
}
|
|
|
|
|
|
|
|
// newRetryCounter instantiates a new *retryCounter. It parses the gitconfig
|
|
|
|
// value: `lfs.transfer.maxretries`, and falls back to defaultMaxRetries if none
|
|
|
|
// was provided.
|
|
|
|
//
|
|
|
|
// If it encountered an error in Unmarshaling the *config.Configuration, it will
|
|
|
|
// be returned, otherwise nil.
|
2016-12-13 20:06:58 +00:00
|
|
|
func newRetryCounter() *retryCounter {
|
|
|
|
return &retryCounter{
|
2016-09-28 18:25:37 +00:00
|
|
|
MaxRetries: defaultMaxRetries,
|
2016-12-13 20:06:58 +00:00
|
|
|
count: make(map[string]int),
|
2016-09-28 18:25:37 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// Increment increments the number of retries for a given OID. It is safe to
|
|
|
|
// call across multiple goroutines.
|
|
|
|
func (r *retryCounter) Increment(oid string) {
|
|
|
|
r.cmu.Lock()
|
|
|
|
defer r.cmu.Unlock()
|
|
|
|
|
|
|
|
r.count[oid]++
|
|
|
|
}
|
|
|
|
|
|
|
|
// CountFor returns the current number of retries for a given OID. It is safe to
|
|
|
|
// call across multiple goroutines.
|
|
|
|
func (r *retryCounter) CountFor(oid string) int {
|
|
|
|
r.cmu.Lock()
|
|
|
|
defer r.cmu.Unlock()
|
|
|
|
|
|
|
|
return r.count[oid]
|
|
|
|
}
|
|
|
|
|
|
|
|
// CanRetry returns the current number of retries, and whether or not it exceeds
|
|
|
|
// the maximum number of retries (see: retryCounter.MaxRetries).
|
|
|
|
func (r *retryCounter) CanRetry(oid string) (int, bool) {
|
|
|
|
count := r.CountFor(oid)
|
|
|
|
return count, count < r.MaxRetries
|
|
|
|
}
|
|
|
|
|
2016-12-15 21:03:57 +00:00
|
|
|
// batch implements the sort.Interface interface and enables sorting on a slice
|
2016-12-14 17:57:34 +00:00
|
|
|
// of `*Transfer`s by object size.
|
2016-12-12 16:10:37 +00:00
|
|
|
//
|
|
|
|
// This interface is implemented here so that the largest objects can be
|
|
|
|
// processed first. Since adding a new batch is unable to occur until the
|
|
|
|
// current batch has finished processing, this enables us to reduce the risk of
|
|
|
|
// a single worker getting tied up on a large item at the end of a batch while
|
|
|
|
// all other workers are sitting idle.
|
2016-12-15 21:03:57 +00:00
|
|
|
type batch []*objectTuple
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2017-01-06 22:51:49 +00:00
|
|
|
func (b batch) ToTransfers() []*Transfer {
|
|
|
|
transfers := make([]*Transfer, 0, len(b))
|
2016-12-09 20:55:04 +00:00
|
|
|
for _, t := range b {
|
2017-01-06 22:51:49 +00:00
|
|
|
transfers = append(transfers, &Transfer{Oid: t.Oid, Size: t.Size})
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
return transfers
|
|
|
|
}
|
|
|
|
|
2016-12-15 21:03:57 +00:00
|
|
|
func (b batch) Len() int { return len(b) }
|
|
|
|
func (b batch) Less(i, j int) bool { return b[i].Size < b[j].Size }
|
|
|
|
func (b batch) Swap(i, j int) { b[i], b[j] = b[j], b[i] }
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2016-05-24 16:05:02 +00:00
|
|
|
// TransferQueue organises the wider process of uploading and downloading,
|
|
|
|
// including calling the API, passing the actual transfer request to transfer
|
2016-09-21 19:46:34 +00:00
|
|
|
// adapters, and dealing with progress, errors and retries.
|
2015-05-21 16:36:49 +00:00
|
|
|
type TransferQueue struct {
|
2016-12-12 00:28:47 +00:00
|
|
|
direction Direction
|
2017-01-04 17:11:16 +00:00
|
|
|
client *tqClient
|
|
|
|
remote string
|
2016-12-12 00:55:14 +00:00
|
|
|
adapter Adapter
|
2016-05-25 15:41:47 +00:00
|
|
|
adapterInProgress bool
|
|
|
|
adapterInitMutex sync.Mutex
|
|
|
|
dryRun bool
|
2016-12-07 03:29:18 +00:00
|
|
|
meter progress.Meter
|
2016-05-25 15:41:47 +00:00
|
|
|
errors []error
|
2016-12-14 18:05:57 +00:00
|
|
|
transfers map[string]*objectTuple
|
2016-12-09 17:23:49 +00:00
|
|
|
batchSize int
|
2016-12-09 20:55:04 +00:00
|
|
|
bufferDepth int
|
|
|
|
// Channel for processing (and buffering) incoming items
|
2016-12-14 18:05:57 +00:00
|
|
|
incoming chan *objectTuple
|
2016-12-09 20:55:04 +00:00
|
|
|
errorc chan error // Channel for processing errors
|
|
|
|
watchers []chan string
|
|
|
|
trMutex *sync.Mutex
|
|
|
|
startProgress sync.Once
|
2016-12-12 21:10:48 +00:00
|
|
|
collectorWait sync.WaitGroup
|
2016-12-09 20:55:04 +00:00
|
|
|
errorwait sync.WaitGroup
|
2016-09-21 19:46:34 +00:00
|
|
|
// wait is used to keep track of pending transfers. It is incremented
|
|
|
|
// once per unique OID on Add(), and is decremented when that transfer
|
|
|
|
// is marked as completed or failed, but not retried.
|
2016-11-18 18:21:09 +00:00
|
|
|
wait sync.WaitGroup
|
2016-12-12 00:28:47 +00:00
|
|
|
manifest *Manifest
|
2016-11-18 18:21:09 +00:00
|
|
|
rc *retryCounter
|
2015-05-21 16:36:49 +00:00
|
|
|
}
|
|
|
|
|
2016-12-14 18:05:57 +00:00
|
|
|
type objectTuple struct {
|
|
|
|
Name, Path, Oid string
|
|
|
|
Size int64
|
|
|
|
}
|
|
|
|
|
2016-12-12 00:43:08 +00:00
|
|
|
type Option func(*TransferQueue)
|
2016-12-07 03:41:35 +00:00
|
|
|
|
2016-12-12 00:43:08 +00:00
|
|
|
func DryRun(dryRun bool) Option {
|
2016-12-07 03:41:35 +00:00
|
|
|
return func(tq *TransferQueue) {
|
|
|
|
tq.dryRun = dryRun
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-12-12 00:43:08 +00:00
|
|
|
func WithProgress(m progress.Meter) Option {
|
2016-12-07 03:41:35 +00:00
|
|
|
return func(tq *TransferQueue) {
|
|
|
|
tq.meter = m
|
2016-12-07 03:29:18 +00:00
|
|
|
}
|
2016-12-07 03:41:35 +00:00
|
|
|
}
|
2016-12-07 03:29:18 +00:00
|
|
|
|
2016-12-12 00:43:08 +00:00
|
|
|
func WithBatchSize(size int) Option {
|
2016-12-09 17:23:49 +00:00
|
|
|
return func(tq *TransferQueue) { tq.batchSize = size }
|
|
|
|
}
|
|
|
|
|
2016-12-12 00:43:08 +00:00
|
|
|
func WithBufferDepth(depth int) Option {
|
2016-12-09 20:55:04 +00:00
|
|
|
return func(tq *TransferQueue) { tq.bufferDepth = depth }
|
|
|
|
}
|
|
|
|
|
2016-12-12 00:20:14 +00:00
|
|
|
// NewTransferQueue builds a TransferQueue, direction and underlying mechanism determined by adapter
|
2017-01-04 21:46:30 +00:00
|
|
|
func NewTransferQueue(dir Direction, manifest *Manifest, remote string, options ...Option) *TransferQueue {
|
2015-07-09 18:21:49 +00:00
|
|
|
q := &TransferQueue{
|
2016-12-14 17:57:34 +00:00
|
|
|
direction: dir,
|
2017-01-04 17:11:16 +00:00
|
|
|
client: &tqClient{Client: manifest.APIClient()},
|
2017-01-04 21:46:30 +00:00
|
|
|
remote: remote,
|
2016-12-14 17:57:34 +00:00
|
|
|
errorc: make(chan error),
|
2016-12-14 18:05:57 +00:00
|
|
|
transfers: make(map[string]*objectTuple),
|
2016-12-14 17:57:34 +00:00
|
|
|
trMutex: &sync.Mutex{},
|
|
|
|
manifest: manifest,
|
|
|
|
rc: newRetryCounter(),
|
2015-05-21 16:36:49 +00:00
|
|
|
}
|
2015-07-09 18:21:49 +00:00
|
|
|
|
2016-12-07 03:41:35 +00:00
|
|
|
for _, opt := range options {
|
|
|
|
opt(q)
|
|
|
|
}
|
|
|
|
|
2016-12-13 23:48:54 +00:00
|
|
|
q.rc.MaxRetries = q.manifest.maxRetries
|
2016-12-13 20:06:58 +00:00
|
|
|
|
2016-12-12 17:30:15 +00:00
|
|
|
if q.batchSize <= 0 {
|
|
|
|
q.batchSize = defaultBatchSize
|
|
|
|
}
|
|
|
|
if q.bufferDepth <= 0 {
|
|
|
|
q.bufferDepth = q.batchSize
|
|
|
|
}
|
|
|
|
|
2016-12-14 18:05:57 +00:00
|
|
|
q.incoming = make(chan *objectTuple, q.bufferDepth)
|
2016-12-09 17:23:49 +00:00
|
|
|
|
2016-12-07 03:41:35 +00:00
|
|
|
if q.meter == nil {
|
|
|
|
q.meter = progress.Noop()
|
|
|
|
}
|
|
|
|
|
2016-12-12 21:10:48 +00:00
|
|
|
q.collectorWait.Add(1)
|
2015-09-04 15:21:26 +00:00
|
|
|
q.errorwait.Add(1)
|
2015-07-09 18:21:49 +00:00
|
|
|
q.run()
|
|
|
|
|
|
|
|
return q
|
2015-05-21 16:36:49 +00:00
|
|
|
}
|
|
|
|
|
2016-12-14 17:57:34 +00:00
|
|
|
// Add adds a *Transfer to the transfer queue. It only increments the amount
|
|
|
|
// of waiting the TransferQueue has to do if the *Transfer "t" is new.
|
2016-12-14 18:05:57 +00:00
|
|
|
func (q *TransferQueue) Add(name, path, oid string, size int64) {
|
|
|
|
t := &objectTuple{
|
|
|
|
Name: name,
|
|
|
|
Path: path,
|
|
|
|
Oid: oid,
|
|
|
|
Size: size,
|
|
|
|
}
|
|
|
|
|
2016-12-12 22:10:10 +00:00
|
|
|
if isNew := q.remember(t); !isNew {
|
2016-12-14 17:57:34 +00:00
|
|
|
tracerx.Printf("already transferring %q, skipping duplicate", t.Oid)
|
2016-12-12 22:10:10 +00:00
|
|
|
return
|
|
|
|
}
|
|
|
|
|
|
|
|
q.incoming <- t
|
|
|
|
}
|
|
|
|
|
2016-12-14 17:57:34 +00:00
|
|
|
// remember remembers the *Transfer "t" if the *TransferQueue doesn't already
|
|
|
|
// know about a Transfer with the same OID.
|
2016-12-12 22:10:10 +00:00
|
|
|
//
|
|
|
|
// It returns if the value is new or not.
|
2016-12-14 18:05:57 +00:00
|
|
|
func (q *TransferQueue) remember(t *objectTuple) bool {
|
2016-04-22 20:24:50 +00:00
|
|
|
q.trMutex.Lock()
|
2016-12-12 22:10:10 +00:00
|
|
|
defer q.trMutex.Unlock()
|
|
|
|
|
2016-12-14 17:57:34 +00:00
|
|
|
if _, ok := q.transfers[t.Oid]; !ok {
|
2016-09-21 19:46:34 +00:00
|
|
|
q.wait.Add(1)
|
2016-12-14 17:57:34 +00:00
|
|
|
q.transfers[t.Oid] = t
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2016-12-12 22:10:10 +00:00
|
|
|
return true
|
2016-09-21 19:46:34 +00:00
|
|
|
}
|
2016-12-12 22:10:10 +00:00
|
|
|
return false
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// collectBatches collects batches in a loop, prioritizing failed items from the
|
|
|
|
// previous before adding new items. The process works as follows:
|
|
|
|
//
|
|
|
|
// 1. Create a new batch, of size `q.batchSize`, and containing no items
|
|
|
|
// 2. While the batch contains less items than `q.batchSize` AND the channel
|
|
|
|
// is open, read one item from the `q.incoming` channel.
|
|
|
|
// a. If the read was a channel close, go to step 4.
|
2016-12-14 17:57:34 +00:00
|
|
|
// b. If the read was a TransferTransferable item, go to step 3.
|
2016-12-09 20:55:04 +00:00
|
|
|
// 3. Append the item to the batch.
|
|
|
|
// 4. Sort the batch by descending object size, make a batch API call, send
|
2016-12-12 00:28:47 +00:00
|
|
|
// the items to the `*adapterBase`.
|
2016-12-09 20:55:04 +00:00
|
|
|
// 5. Process the worker results, incrementing and appending retries if
|
|
|
|
// possible.
|
|
|
|
// 6. If the `q.incoming` channel is open, go to step 2.
|
|
|
|
// 7. If the next batch is empty AND the `q.incoming` channel is closed,
|
|
|
|
// terminate immediately.
|
|
|
|
//
|
|
|
|
// collectBatches runs in its own goroutine.
|
|
|
|
func (q *TransferQueue) collectBatches() {
|
2016-12-12 21:10:48 +00:00
|
|
|
defer q.collectorWait.Done()
|
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
var closing bool
|
|
|
|
batch := q.makeBatch()
|
|
|
|
|
|
|
|
for {
|
2016-12-12 16:08:06 +00:00
|
|
|
for !closing && (len(batch) < q.batchSize) {
|
2016-12-09 20:55:04 +00:00
|
|
|
t, ok := <-q.incoming
|
|
|
|
if !ok {
|
|
|
|
closing = true
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
batch = append(batch, t)
|
|
|
|
}
|
|
|
|
|
2016-12-12 16:10:37 +00:00
|
|
|
// Before enqueuing the next batch, sort by descending object
|
|
|
|
// size.
|
2016-12-11 01:04:03 +00:00
|
|
|
sort.Sort(sort.Reverse(batch))
|
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
retries, err := q.enqueueAndCollectRetriesFor(batch)
|
|
|
|
if err != nil {
|
|
|
|
q.errorc <- err
|
|
|
|
}
|
|
|
|
|
|
|
|
if closing && len(retries) == 0 {
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
|
|
|
batch = retries
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// enqueueAndCollectRetriesFor makes a Batch API call and returns a "next" batch
|
|
|
|
// containing all of the objects that failed from the previous batch and had
|
|
|
|
// retries availale to them.
|
|
|
|
//
|
|
|
|
// If an error was encountered while making the API request, _all_ of the items
|
|
|
|
// from the previous batch (that have retries available to them) will be
|
|
|
|
// returned immediately, along with the error that was encountered.
|
|
|
|
//
|
|
|
|
// enqueueAndCollectRetriesFor blocks until the entire Batch "batch" has been
|
|
|
|
// processed.
|
2016-12-15 21:03:57 +00:00
|
|
|
func (q *TransferQueue) enqueueAndCollectRetriesFor(batch batch) (batch, error) {
|
2016-12-09 20:55:04 +00:00
|
|
|
next := q.makeBatch()
|
|
|
|
tracerx.Printf("tq: sending batch of size %d", len(batch))
|
|
|
|
|
2017-01-04 17:11:16 +00:00
|
|
|
bReq := &batchRequest{
|
2017-01-09 19:52:36 +00:00
|
|
|
Operation: q.direction.String(),
|
2017-01-06 22:51:49 +00:00
|
|
|
Objects: batch.ToTransfers(),
|
2017-01-04 17:11:16 +00:00
|
|
|
TransferAdapterNames: q.manifest.GetAdapterNames(q.direction),
|
|
|
|
}
|
|
|
|
|
|
|
|
bRes, _, err := q.client.Batch(q.remote, bReq)
|
2016-12-09 20:55:04 +00:00
|
|
|
if err != nil {
|
|
|
|
// If there was an error making the batch API call, mark all of
|
|
|
|
// the objects for retry, and return them along with the error
|
|
|
|
// that was encountered. If any of the objects couldn't be
|
|
|
|
// retried, they will be marked as failed.
|
|
|
|
for _, t := range batch {
|
2016-12-14 17:57:34 +00:00
|
|
|
if q.canRetryObject(t.Oid, err) {
|
|
|
|
q.rc.Increment(t.Oid)
|
2016-12-09 20:55:04 +00:00
|
|
|
|
|
|
|
next = append(next, t)
|
|
|
|
} else {
|
|
|
|
q.wait.Done()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return next, err
|
|
|
|
}
|
|
|
|
|
2017-01-04 17:11:16 +00:00
|
|
|
if len(bRes.Objects) == 0 {
|
|
|
|
return next, nil
|
|
|
|
}
|
|
|
|
|
2017-01-09 20:04:58 +00:00
|
|
|
q.useAdapter(bRes.TransferAdapterName)
|
2016-12-09 20:55:04 +00:00
|
|
|
q.startProgress.Do(q.meter.Start)
|
|
|
|
|
2017-01-04 17:11:16 +00:00
|
|
|
toTransfer := make([]*Transfer, 0, len(bRes.Objects))
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2017-01-04 17:11:16 +00:00
|
|
|
for _, o := range bRes.Objects {
|
2016-12-09 20:55:04 +00:00
|
|
|
if o.Error != nil {
|
|
|
|
q.errorc <- errors.Wrapf(o.Error, "[%v] %v", o.Oid, o.Error.Message)
|
|
|
|
q.Skip(o.Size)
|
|
|
|
q.wait.Done()
|
|
|
|
|
|
|
|
continue
|
|
|
|
}
|
|
|
|
|
2016-12-14 17:57:34 +00:00
|
|
|
q.trMutex.Lock()
|
|
|
|
t, ok := q.transfers[o.Oid]
|
|
|
|
q.trMutex.Unlock()
|
|
|
|
if !ok {
|
|
|
|
// If we couldn't find any associated
|
|
|
|
// Transfer object, then we give up on the
|
|
|
|
// transfer by telling the progress meter to
|
|
|
|
// skip the number of bytes in "o".
|
|
|
|
q.errorc <- errors.Errorf("[%v] The server returned an unknown OID.", o.Oid)
|
2016-12-12 22:33:15 +00:00
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
q.Skip(o.Size)
|
|
|
|
q.wait.Done()
|
2016-12-14 17:57:34 +00:00
|
|
|
} else {
|
2017-01-06 22:51:49 +00:00
|
|
|
tr := newTransfer(o, t.Name, t.Path)
|
2016-12-14 17:57:34 +00:00
|
|
|
|
2017-01-09 19:52:36 +00:00
|
|
|
if _, err := tr.Actions.Get(q.direction.String()); err != nil {
|
2016-12-14 17:57:34 +00:00
|
|
|
// XXX(taylor): duplication
|
2016-12-14 18:05:57 +00:00
|
|
|
if q.canRetryObject(tr.Oid, err) {
|
|
|
|
q.rc.Increment(tr.Oid)
|
|
|
|
count := q.rc.CountFor(tr.Oid)
|
2016-12-14 17:57:34 +00:00
|
|
|
|
2016-12-14 18:05:57 +00:00
|
|
|
tracerx.Printf("tq: enqueue retry #%d for %q (size: %d)", count, tr.Oid, tr.Size)
|
2016-12-14 17:57:34 +00:00
|
|
|
next = append(next, t)
|
|
|
|
} else {
|
|
|
|
if !IsActionMissingError(err) {
|
2016-12-14 18:05:57 +00:00
|
|
|
q.errorc <- errors.Errorf("[%v] %v", tr.Name, err)
|
2016-12-14 17:57:34 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
q.Skip(o.Size)
|
|
|
|
q.wait.Done()
|
|
|
|
}
|
|
|
|
|
|
|
|
} else {
|
|
|
|
q.meter.StartTransfer(t.Name)
|
2016-12-14 18:05:57 +00:00
|
|
|
toTransfer = append(toTransfer, tr)
|
2016-12-14 17:57:34 +00:00
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-01-09 20:00:59 +00:00
|
|
|
retries := q.addToAdapter(bRes.endpoint, toTransfer)
|
2016-12-09 20:55:04 +00:00
|
|
|
for t := range retries {
|
2016-12-14 17:57:34 +00:00
|
|
|
q.rc.Increment(t.Oid)
|
|
|
|
count := q.rc.CountFor(t.Oid)
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2016-12-14 17:57:34 +00:00
|
|
|
tracerx.Printf("tq: enqueue retry #%d for %q (size: %d)", count, t.Oid, t.Size)
|
2016-12-09 20:55:04 +00:00
|
|
|
|
|
|
|
next = append(next, t)
|
|
|
|
}
|
|
|
|
|
|
|
|
return next, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
// makeBatch returns a new, empty batch, with a capacity equal to the maximum
|
|
|
|
// batch size designated by the `*TransferQueue`.
|
2016-12-15 21:03:57 +00:00
|
|
|
func (q *TransferQueue) makeBatch() batch { return make(batch, 0, q.batchSize) }
|
2016-12-09 20:55:04 +00:00
|
|
|
|
|
|
|
// addToAdapter adds the given "pending" transfers to the transfer adapters and
|
2016-12-14 17:57:34 +00:00
|
|
|
// returns a channel of Transfers that are to be retried in the next batch.
|
2016-12-09 20:55:04 +00:00
|
|
|
// After all of the items in the batch have been processed, the channel is
|
|
|
|
// closed.
|
|
|
|
//
|
|
|
|
// addToAdapter returns immediately, and does not block.
|
2017-01-04 21:46:30 +00:00
|
|
|
func (q *TransferQueue) addToAdapter(e lfsapi.Endpoint, pending []*Transfer) <-chan *objectTuple {
|
2016-12-14 18:05:57 +00:00
|
|
|
retries := make(chan *objectTuple, len(pending))
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2017-01-04 21:46:30 +00:00
|
|
|
if err := q.ensureAdapterBegun(e); err != nil {
|
2016-12-09 20:55:04 +00:00
|
|
|
close(retries)
|
|
|
|
|
|
|
|
q.errorc <- err
|
|
|
|
for _, t := range pending {
|
2016-12-14 17:57:34 +00:00
|
|
|
q.Skip(t.Size)
|
2016-12-09 20:55:04 +00:00
|
|
|
q.wait.Done()
|
|
|
|
}
|
|
|
|
|
|
|
|
return retries
|
|
|
|
}
|
|
|
|
|
|
|
|
go func() {
|
|
|
|
defer close(retries)
|
|
|
|
|
2016-12-12 00:28:47 +00:00
|
|
|
var results <-chan TransferResult
|
2016-12-09 20:55:04 +00:00
|
|
|
if q.dryRun {
|
|
|
|
results = q.makeDryRunResults(pending)
|
|
|
|
} else {
|
|
|
|
results = q.adapter.Add(pending...)
|
|
|
|
}
|
|
|
|
|
|
|
|
for res := range results {
|
|
|
|
q.handleTransferResult(res, retries)
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
return retries
|
|
|
|
}
|
|
|
|
|
|
|
|
// makeDryRunResults returns a channel populated immediately with "successful"
|
|
|
|
// results for all of the given transfers in "ts".
|
2016-12-12 00:28:47 +00:00
|
|
|
func (q *TransferQueue) makeDryRunResults(ts []*Transfer) <-chan TransferResult {
|
|
|
|
results := make(chan TransferResult, len(ts))
|
2016-12-09 20:55:04 +00:00
|
|
|
for _, t := range ts {
|
2016-12-12 00:28:47 +00:00
|
|
|
results <- TransferResult{t, nil}
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
close(results)
|
|
|
|
|
|
|
|
return results
|
|
|
|
}
|
|
|
|
|
|
|
|
// handleTransferResult observes the transfer result, sending it on the retries
|
|
|
|
// channel if it was able to be retried.
|
|
|
|
func (q *TransferQueue) handleTransferResult(
|
2016-12-14 18:05:57 +00:00
|
|
|
res TransferResult, retries chan<- *objectTuple,
|
2016-12-09 20:55:04 +00:00
|
|
|
) {
|
2016-12-14 17:57:34 +00:00
|
|
|
oid := res.Transfer.Oid
|
2016-12-09 20:55:04 +00:00
|
|
|
|
|
|
|
if res.Error != nil {
|
|
|
|
// If there was an error encountered when processing the
|
|
|
|
// transfer (res.Transfer), handle the error as is appropriate:
|
|
|
|
|
|
|
|
if q.canRetryObject(oid, res.Error) {
|
|
|
|
// If the object can be retried, send it on the retries
|
|
|
|
// channel, where it will be read at the call-site and
|
|
|
|
// its retry count will be incremented.
|
|
|
|
tracerx.Printf("tq: retrying object %s", oid)
|
|
|
|
|
|
|
|
q.trMutex.Lock()
|
2016-12-14 17:57:34 +00:00
|
|
|
t, ok := q.transfers[oid]
|
2016-12-09 20:55:04 +00:00
|
|
|
q.trMutex.Unlock()
|
|
|
|
|
|
|
|
if ok {
|
|
|
|
retries <- t
|
|
|
|
} else {
|
|
|
|
q.errorc <- res.Error
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// If the error wasn't retriable, OR the object has
|
|
|
|
// exceeded its retry budget, it will be NOT be sent to
|
|
|
|
// the retry channel, and the error will be reported
|
|
|
|
// immediately.
|
|
|
|
q.errorc <- res.Error
|
|
|
|
q.wait.Done()
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// Otherwise, if the transfer was successful, notify all of the
|
|
|
|
// watchers, and mark it as finished.
|
|
|
|
for _, c := range q.watchers {
|
|
|
|
c <- oid
|
|
|
|
}
|
|
|
|
|
|
|
|
q.meter.FinishTransfer(res.Transfer.Name)
|
|
|
|
q.wait.Done()
|
|
|
|
}
|
2015-07-09 18:21:49 +00:00
|
|
|
}
|
|
|
|
|
2016-06-02 11:44:17 +00:00
|
|
|
func (q *TransferQueue) useAdapter(name string) {
|
|
|
|
q.adapterInitMutex.Lock()
|
|
|
|
defer q.adapterInitMutex.Unlock()
|
|
|
|
|
|
|
|
if q.adapter != nil {
|
|
|
|
if q.adapter.Name() == name {
|
|
|
|
// re-use, this is the normal path
|
|
|
|
return
|
|
|
|
}
|
|
|
|
// If the adapter we're using isn't the same as the one we've been
|
|
|
|
// told to use now, must wait for the current one to finish then switch
|
|
|
|
// This will probably never happen but is just in case server starts
|
|
|
|
// changing adapter support in between batches
|
|
|
|
q.finishAdapter()
|
|
|
|
}
|
2016-08-09 22:07:41 +00:00
|
|
|
q.adapter = q.manifest.NewAdapterOrDefault(name, q.direction)
|
2016-06-01 16:33:01 +00:00
|
|
|
}
|
2016-06-02 11:11:45 +00:00
|
|
|
|
2016-06-02 11:44:17 +00:00
|
|
|
func (q *TransferQueue) finishAdapter() {
|
|
|
|
if q.adapterInProgress {
|
|
|
|
q.adapter.End()
|
|
|
|
q.adapterInProgress = false
|
|
|
|
q.adapter = nil
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2016-04-06 19:06:34 +00:00
|
|
|
func (q *TransferQueue) Skip(size int64) {
|
|
|
|
q.meter.Skip(size)
|
|
|
|
}
|
|
|
|
|
2017-01-04 21:46:30 +00:00
|
|
|
func (q *TransferQueue) ensureAdapterBegun(e lfsapi.Endpoint) error {
|
2016-05-25 15:41:47 +00:00
|
|
|
q.adapterInitMutex.Lock()
|
|
|
|
defer q.adapterInitMutex.Unlock()
|
|
|
|
|
|
|
|
if q.adapterInProgress {
|
2016-07-11 10:17:10 +00:00
|
|
|
return nil
|
2016-05-25 15:41:47 +00:00
|
|
|
}
|
|
|
|
|
|
|
|
// Progress callback - receives byte updates
|
2016-05-27 11:52:18 +00:00
|
|
|
cb := func(name string, total, read int64, current int) error {
|
2017-01-09 19:52:36 +00:00
|
|
|
q.meter.TransferBytes(q.direction.String(), name, read, total, current)
|
2016-05-25 15:41:47 +00:00
|
|
|
return nil
|
|
|
|
}
|
|
|
|
|
|
|
|
tracerx.Printf("tq: starting transfer adapter %q", q.adapter.Name())
|
2017-01-04 22:18:23 +00:00
|
|
|
err := q.adapter.Begin(q.toAdapterCfg(e), cb)
|
2016-07-11 10:17:10 +00:00
|
|
|
if err != nil {
|
|
|
|
return err
|
|
|
|
}
|
2016-05-25 15:41:47 +00:00
|
|
|
q.adapterInProgress = true
|
|
|
|
|
2016-07-11 10:17:10 +00:00
|
|
|
return nil
|
2016-05-25 15:41:47 +00:00
|
|
|
}
|
|
|
|
|
2017-01-04 22:18:23 +00:00
|
|
|
func (q *TransferQueue) toAdapterCfg(e lfsapi.Endpoint) AdapterConfig {
|
|
|
|
apiClient := q.manifest.APIClient()
|
|
|
|
concurrency := q.manifest.ConcurrentTransfers()
|
2017-01-04 21:46:30 +00:00
|
|
|
if apiClient.Endpoints.AccessFor(e.Url) == lfsapi.NTLMAccess {
|
|
|
|
concurrency = 1
|
|
|
|
}
|
2017-01-04 22:18:23 +00:00
|
|
|
|
|
|
|
return &adapterConfig{
|
|
|
|
concurrentTransfers: concurrency,
|
|
|
|
apiClient: apiClient,
|
2017-01-04 22:22:31 +00:00
|
|
|
remote: q.remote,
|
2017-01-04 22:18:23 +00:00
|
|
|
}
|
2017-01-04 21:46:30 +00:00
|
|
|
}
|
|
|
|
|
2015-09-09 14:24:50 +00:00
|
|
|
// Wait waits for the queue to finish processing all transfers. Once Wait is
|
2016-12-14 17:57:34 +00:00
|
|
|
// called, Add will no longer add transfers to the queue. Any failed
|
2015-09-09 14:24:50 +00:00
|
|
|
// transfers will be automatically retried once.
|
2015-07-09 18:21:49 +00:00
|
|
|
func (q *TransferQueue) Wait() {
|
2016-12-09 20:55:04 +00:00
|
|
|
close(q.incoming)
|
2015-09-04 20:01:48 +00:00
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
q.wait.Wait()
|
2016-12-12 21:10:48 +00:00
|
|
|
q.collectorWait.Wait()
|
2015-09-09 14:24:50 +00:00
|
|
|
|
2016-06-02 11:44:17 +00:00
|
|
|
q.finishAdapter()
|
2015-07-09 18:21:49 +00:00
|
|
|
close(q.errorc)
|
2015-07-10 20:05:04 +00:00
|
|
|
|
2015-07-09 18:21:49 +00:00
|
|
|
for _, watcher := range q.watchers {
|
|
|
|
close(watcher)
|
|
|
|
}
|
2015-07-10 19:25:59 +00:00
|
|
|
|
2015-07-27 16:43:41 +00:00
|
|
|
q.meter.Finish()
|
2015-09-04 15:21:26 +00:00
|
|
|
q.errorwait.Wait()
|
2015-05-21 16:36:49 +00:00
|
|
|
}
|
|
|
|
|
2015-05-27 19:45:18 +00:00
|
|
|
// Watch returns a channel where the queue will write the OID of each transfer
|
|
|
|
// as it completes. The channel will be closed when the queue finishes processing.
|
|
|
|
func (q *TransferQueue) Watch() chan string {
|
2016-12-09 17:23:49 +00:00
|
|
|
c := make(chan string, q.batchSize)
|
2015-05-27 19:45:18 +00:00
|
|
|
q.watchers = append(q.watchers, c)
|
|
|
|
return c
|
|
|
|
}
|
|
|
|
|
2015-07-09 18:21:49 +00:00
|
|
|
// This goroutine collects errors returned from transfers
|
|
|
|
func (q *TransferQueue) errorCollector() {
|
|
|
|
for err := range q.errorc {
|
|
|
|
q.errors = append(q.errors, err)
|
2015-05-21 16:36:49 +00:00
|
|
|
}
|
2015-09-04 15:21:26 +00:00
|
|
|
q.errorwait.Done()
|
2015-07-09 18:21:49 +00:00
|
|
|
}
|
2015-05-21 16:36:49 +00:00
|
|
|
|
2015-07-10 20:05:04 +00:00
|
|
|
// run starts the transfer queue, doing individual or batch transfers depending
|
|
|
|
// on the Config.BatchTransfer() value. run will transfer files sequentially or
|
|
|
|
// concurrently depending on the Config.ConcurrentTransfers() value.
|
2015-07-09 18:21:49 +00:00
|
|
|
func (q *TransferQueue) run() {
|
2016-12-09 17:23:49 +00:00
|
|
|
tracerx.Printf("tq: running as batched queue, batch size of %d", q.batchSize)
|
2015-05-21 16:36:49 +00:00
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
go q.errorCollector()
|
|
|
|
go q.collectBatches()
|
2015-09-04 20:01:48 +00:00
|
|
|
}
|
2015-08-18 19:53:04 +00:00
|
|
|
|
2016-09-21 19:46:34 +00:00
|
|
|
// canRetry returns whether or not the given error "err" is retriable.
|
2015-09-04 20:01:48 +00:00
|
|
|
func (q *TransferQueue) canRetry(err error) bool {
|
2016-09-21 19:46:34 +00:00
|
|
|
return errors.IsRetriableError(err)
|
|
|
|
}
|
|
|
|
|
|
|
|
// canRetryObject returns whether the given error is retriable for the object
|
|
|
|
// given by "oid". If the an OID has met its retry limit, then it will not be
|
|
|
|
// able to be retried again. If so, canRetryObject returns whether or not that
|
|
|
|
// given error "err" is retriable.
|
|
|
|
func (q *TransferQueue) canRetryObject(oid string, err error) bool {
|
2016-09-28 18:25:37 +00:00
|
|
|
if count, ok := q.rc.CanRetry(oid); !ok {
|
2016-09-21 19:46:34 +00:00
|
|
|
tracerx.Printf("tq: refusing to retry %q, too many retries (%d)", oid, count)
|
2015-08-17 23:02:20 +00:00
|
|
|
return false
|
|
|
|
}
|
2015-09-04 13:42:37 +00:00
|
|
|
|
2016-09-21 19:46:34 +00:00
|
|
|
return q.canRetry(err)
|
2015-08-17 23:02:20 +00:00
|
|
|
}
|
|
|
|
|
2015-05-26 13:56:43 +00:00
|
|
|
// Errors returns any errors encountered during transfer.
|
2015-08-21 15:48:52 +00:00
|
|
|
func (q *TransferQueue) Errors() []error {
|
2015-05-21 16:36:49 +00:00
|
|
|
return q.errors
|
|
|
|
}
|