2016-12-12 00:20:14 +00:00
|
|
|
package tq
|
2015-05-21 16:36:49 +00:00
|
|
|
|
|
|
|
import (
|
2018-07-31 18:24:17 +00:00
|
|
|
"fmt"
|
2017-01-11 20:59:25 +00:00
|
|
|
"os"
|
2016-12-11 01:04:03 +00:00
|
|
|
"sort"
|
2015-05-21 16:36:49 +00:00
|
|
|
"sync"
|
2018-12-26 15:30:53 +00:00
|
|
|
"time"
|
2015-05-21 17:47:52 +00:00
|
|
|
|
2016-11-15 17:01:18 +00:00
|
|
|
"github.com/git-lfs/git-lfs/errors"
|
2018-01-05 18:12:57 +00:00
|
|
|
"github.com/git-lfs/git-lfs/git"
|
2017-01-04 21:46:30 +00:00
|
|
|
"github.com/git-lfs/git-lfs/lfsapi"
|
2018-09-06 21:42:41 +00:00
|
|
|
"github.com/git-lfs/git-lfs/lfshttp"
|
2017-11-22 02:00:50 +00:00
|
|
|
"github.com/git-lfs/git-lfs/tools"
|
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-08-06 17:05:02 +00:00
|
|
|
// Concat concatenates two batches together, returning a single, clamped batch as
|
|
|
|
// "left", and the remainder of elements as "right". If the union of the
|
|
|
|
// receiver and "other" has cardinality less than "size", "right" will be
|
2018-12-26 15:30:53 +00:00
|
|
|
// returned as nil. Any object tuple that is not currently able to be retried
|
|
|
|
// (ie Retry-After response), will also go into the right batch. Also, when object(s)
|
|
|
|
// are returend that are rate-limited, return the minimum duration required to wait until
|
|
|
|
// a object is ready.
|
|
|
|
func (b batch) Concat(other batch, size int) (left, right batch, minWait time.Duration) {
|
2017-08-06 17:05:02 +00:00
|
|
|
u := batch(append(b, other...))
|
2018-12-26 15:30:53 +00:00
|
|
|
for _, ot := range u {
|
|
|
|
if time.Now().After(ot.ReadyTime) {
|
|
|
|
// The current time is past the time the object should
|
|
|
|
// be available.
|
|
|
|
left = append(left, ot)
|
|
|
|
} else {
|
|
|
|
// The time hasn't passed for the object.
|
|
|
|
right = append(right, ot)
|
|
|
|
wait := time.Until(ot.ReadyTime)
|
|
|
|
if minWait == 0 {
|
|
|
|
minWait = wait
|
|
|
|
} else if wait < minWait {
|
|
|
|
minWait = wait
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
if len(left) <= size {
|
|
|
|
// If the size of left fits the given size limit, return with no adjustments.
|
|
|
|
return left, right, minWait
|
2017-08-06 17:05:02 +00:00
|
|
|
}
|
2018-12-26 15:30:53 +00:00
|
|
|
// If left is too large, trip left up to size and append the rest to right.
|
|
|
|
right = append(right, left[size:]...)
|
|
|
|
left = left[:size]
|
|
|
|
return left, right, minWait
|
2017-08-06 17:05:02 +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 {
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
transfers = append(transfers, &Transfer{Oid: t.Oid, Size: t.Size, Missing: t.Missing})
|
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
|
|
|
|
2019-08-28 21:02:26 +00:00
|
|
|
type abortableWaitGroup struct {
|
|
|
|
wq sync.WaitGroup
|
|
|
|
counter int
|
|
|
|
mu sync.Mutex
|
|
|
|
}
|
|
|
|
|
|
|
|
func newAbortableWaitGroup() *abortableWaitGroup {
|
|
|
|
return &abortableWaitGroup{}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (q *abortableWaitGroup) Add(delta int) {
|
|
|
|
q.mu.Lock()
|
|
|
|
defer q.mu.Unlock()
|
|
|
|
|
|
|
|
q.counter += delta
|
|
|
|
q.wq.Add(delta)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (q *abortableWaitGroup) Done() {
|
|
|
|
q.mu.Lock()
|
|
|
|
defer q.mu.Unlock()
|
|
|
|
|
|
|
|
q.counter -= 1
|
|
|
|
q.wq.Done()
|
|
|
|
}
|
|
|
|
|
|
|
|
func (q *abortableWaitGroup) Abort() {
|
|
|
|
q.mu.Lock()
|
|
|
|
defer q.mu.Unlock()
|
|
|
|
|
|
|
|
q.wq.Add(-q.counter)
|
|
|
|
}
|
|
|
|
|
|
|
|
func (q *abortableWaitGroup) Wait() {
|
|
|
|
q.wq.Wait()
|
|
|
|
}
|
|
|
|
|
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
|
2018-01-05 18:12:57 +00:00
|
|
|
ref *git.Ref
|
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
|
2017-11-22 02:00:50 +00:00
|
|
|
cb tools.CopyCallback
|
2017-11-29 19:07:44 +00:00
|
|
|
meter *Meter
|
2016-05-25 15:41:47 +00:00
|
|
|
errors []error
|
2017-08-07 20:29:16 +00:00
|
|
|
transfers map[string]*objects
|
|
|
|
batchSize int
|
|
|
|
bufferDepth int
|
|
|
|
incoming chan *objectTuple // Channel for processing incoming items
|
|
|
|
errorc chan error // Channel for processing errors
|
|
|
|
watchers []chan *Transfer
|
|
|
|
trMutex *sync.Mutex
|
|
|
|
collectorWait sync.WaitGroup
|
|
|
|
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.
|
2019-08-28 21:02:26 +00:00
|
|
|
wait *abortableWaitGroup
|
2016-12-12 00:28:47 +00:00
|
|
|
manifest *Manifest
|
2016-11-18 18:21:09 +00:00
|
|
|
rc *retryCounter
|
2018-07-31 18:24:17 +00:00
|
|
|
|
|
|
|
// unsupportedContentType indicates whether the transfer queue ever saw
|
|
|
|
// an HTTP 422 response indicating that their upload destination does
|
|
|
|
// not support Content-Type detection.
|
|
|
|
unsupportedContentType bool
|
2015-05-21 16:36:49 +00:00
|
|
|
}
|
|
|
|
|
2017-08-03 20:44:53 +00:00
|
|
|
// objects holds a set of objects.
|
|
|
|
type objects struct {
|
|
|
|
completed bool
|
|
|
|
objects []*objectTuple
|
|
|
|
}
|
|
|
|
|
|
|
|
// All returns all *objectTuple's contained in the *objects set.
|
|
|
|
func (s *objects) All() []*objectTuple {
|
|
|
|
return s.objects
|
|
|
|
}
|
|
|
|
|
|
|
|
// Append returns a new *objects with the given *objectTuple(s) appended to the
|
|
|
|
// end of the known objects.
|
|
|
|
func (s *objects) Append(os ...*objectTuple) *objects {
|
|
|
|
return &objects{
|
|
|
|
completed: s.completed,
|
|
|
|
objects: append(s.objects, os...),
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// First returns the first *objectTuple in the chain of objects.
|
|
|
|
func (s *objects) First() *objectTuple {
|
|
|
|
if len(s.objects) == 0 {
|
|
|
|
return nil
|
|
|
|
}
|
|
|
|
return s.objects[0]
|
|
|
|
}
|
|
|
|
|
2016-12-14 18:05:57 +00:00
|
|
|
type objectTuple struct {
|
|
|
|
Name, Path, Oid string
|
|
|
|
Size int64
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
Missing bool
|
2018-12-26 15:30:53 +00:00
|
|
|
ReadyTime time.Time
|
2016-12-14 18:05:57 +00:00
|
|
|
}
|
|
|
|
|
2017-08-03 20:45:06 +00:00
|
|
|
func (o *objectTuple) ToTransfer() *Transfer {
|
|
|
|
return &Transfer{
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
Name: o.Name,
|
|
|
|
Path: o.Path,
|
|
|
|
Oid: o.Oid,
|
|
|
|
Size: o.Size,
|
|
|
|
Missing: o.Missing,
|
2017-08-03 20:45:06 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
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
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-11-29 19:07:44 +00:00
|
|
|
func WithProgress(m *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
|
|
|
|
2018-01-05 18:12:57 +00:00
|
|
|
func RemoteRef(ref *git.Ref) Option {
|
|
|
|
return func(tq *TransferQueue) {
|
|
|
|
tq.ref = ref
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-11-22 02:00:50 +00:00
|
|
|
func WithProgressCallback(cb tools.CopyCallback) Option {
|
2017-08-01 18:58:18 +00:00
|
|
|
return func(tq *TransferQueue) {
|
|
|
|
tq.cb = cb
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
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),
|
2017-08-03 20:44:53 +00:00
|
|
|
transfers: make(map[string]*objects),
|
2016-12-14 17:57:34 +00:00
|
|
|
trMutex: &sync.Mutex{},
|
|
|
|
manifest: manifest,
|
|
|
|
rc: newRetryCounter(),
|
2019-08-28 21:02:26 +00:00
|
|
|
wait: newAbortableWaitGroup(),
|
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
|
2017-08-24 21:51:24 +00:00
|
|
|
q.client.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
|
|
|
|
}
|
2018-01-06 02:01:50 +00:00
|
|
|
if q.meter != nil {
|
|
|
|
q.meter.Direction = q.direction
|
|
|
|
}
|
2016-12-12 17:30:15 +00:00
|
|
|
|
2016-12-14 18:05:57 +00:00
|
|
|
q.incoming = make(chan *objectTuple, q.bufferDepth)
|
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.
|
2017-08-03 20:45:06 +00:00
|
|
|
//
|
|
|
|
// If another transfer(s) with the same OID has been added to the *TransferQueue
|
|
|
|
// already, the given transfer will not be enqueued, but will be sent to any
|
|
|
|
// channel created by Watch() once the oldest transfer has completed.
|
|
|
|
//
|
|
|
|
// Only one file will be transferred to/from the Path element of the first
|
|
|
|
// transfer.
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
func (q *TransferQueue) Add(name, path, oid string, size int64, missing bool) {
|
2016-12-14 18:05:57 +00:00
|
|
|
t := &objectTuple{
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
Name: name,
|
|
|
|
Path: path,
|
|
|
|
Oid: oid,
|
|
|
|
Size: size,
|
|
|
|
Missing: missing,
|
2016-12-14 18:05:57 +00:00
|
|
|
}
|
|
|
|
|
2017-08-03 20:44:53 +00:00
|
|
|
if objs := q.remember(t); len(objs.objects) > 1 {
|
2017-08-03 20:45:06 +00:00
|
|
|
if objs.completed {
|
|
|
|
// If there is already a completed transfer chain for
|
|
|
|
// this OID, then this object is already "done", and can
|
|
|
|
// be sent through as completed to the watchers.
|
|
|
|
for _, w := range q.watchers {
|
|
|
|
w <- t.ToTransfer()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-08-03 20:44:53 +00:00
|
|
|
// If the chain is not done, there is no reason to enqueue this
|
|
|
|
// transfer into 'q.incoming'.
|
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.
|
2017-08-03 20:44:53 +00:00
|
|
|
func (q *TransferQueue) remember(t *objectTuple) objects {
|
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)
|
2017-08-03 20:44:53 +00:00
|
|
|
q.transfers[t.Oid] = &objects{
|
|
|
|
objects: []*objectTuple{t},
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2017-08-03 20:44:53 +00:00
|
|
|
return *q.transfers[t.Oid]
|
2016-09-21 19:46:34 +00:00
|
|
|
}
|
2017-08-03 20:44:53 +00:00
|
|
|
|
|
|
|
q.transfers[t.Oid] = q.transfers[t.Oid].Append(t)
|
|
|
|
|
|
|
|
return *q.transfers[t.Oid]
|
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.
|
2017-08-07 20:29:16 +00:00
|
|
|
// b. If the read was a transferable 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`.
|
2017-08-07 20:29:16 +00:00
|
|
|
// 5. In a separate goroutine, process the worker results, incrementing and
|
|
|
|
// appending retries if possible. On the main goroutine, accept new items
|
|
|
|
// into "pending".
|
|
|
|
// 6. Concat() the "next" and "pending" batches such that no more items than
|
|
|
|
// the maximum allowed per batch are in next, and the rest are in pending.
|
|
|
|
// 7. If the `q.incoming` channel is open, go to step 2.
|
|
|
|
// 8. If the next batch is empty AND the `q.incoming` channel is closed,
|
2016-12-09 20:55:04 +00:00
|
|
|
// 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
|
2017-08-06 17:03:42 +00:00
|
|
|
next := q.makeBatch()
|
2017-08-07 20:29:16 +00:00
|
|
|
pending := q.makeBatch()
|
2016-12-09 20:55:04 +00:00
|
|
|
|
|
|
|
for {
|
2017-08-06 17:03:42 +00:00
|
|
|
for !closing && (len(next) < q.batchSize) {
|
2016-12-09 20:55:04 +00:00
|
|
|
t, ok := <-q.incoming
|
|
|
|
if !ok {
|
|
|
|
closing = true
|
|
|
|
break
|
|
|
|
}
|
|
|
|
|
2017-08-06 17:03:42 +00:00
|
|
|
next = append(next, t)
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
|
2016-12-12 16:10:37 +00:00
|
|
|
// Before enqueuing the next batch, sort by descending object
|
|
|
|
// size.
|
2017-08-06 17:03:42 +00:00
|
|
|
sort.Sort(sort.Reverse(next))
|
2016-12-11 01:04:03 +00:00
|
|
|
|
2017-08-07 20:29:16 +00:00
|
|
|
done := make(chan struct{})
|
|
|
|
|
|
|
|
var retries batch
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
var err error
|
2017-08-07 20:29:16 +00:00
|
|
|
|
|
|
|
go func() {
|
2018-12-26 15:30:53 +00:00
|
|
|
defer close(done)
|
2017-08-07 20:29:16 +00:00
|
|
|
|
2018-12-26 15:30:53 +00:00
|
|
|
if len(next) == 0 {
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2017-08-07 20:29:16 +00:00
|
|
|
retries, err = q.enqueueAndCollectRetriesFor(next)
|
|
|
|
if err != nil {
|
|
|
|
q.errorc <- err
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
var collected batch
|
|
|
|
collected, closing = q.collectPendingUntil(done)
|
|
|
|
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
// If we've encountered a serious error here, abort immediately;
|
2019-08-28 21:02:26 +00:00
|
|
|
// don't process further batches. Abort the wait queue so that
|
|
|
|
// we don't deadlock waiting for objects to complete when they
|
|
|
|
// never will.
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
if err != nil {
|
2019-08-28 21:02:26 +00:00
|
|
|
q.wait.Abort()
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
break
|
|
|
|
}
|
|
|
|
|
2017-08-07 20:29:16 +00:00
|
|
|
// Ensure the next batch is filled with, in order:
|
|
|
|
//
|
|
|
|
// - retries from the previous batch,
|
|
|
|
// - new additions that were enqueued behind retries, &
|
|
|
|
// - items collected while the batch was processing.
|
2018-12-26 15:30:53 +00:00
|
|
|
var minWaitTime time.Duration
|
|
|
|
next, pending, minWaitTime = retries.Concat(append(pending, collected...), q.batchSize)
|
|
|
|
if len(next) == 0 && len(pending) != 0 {
|
|
|
|
// There are some pending that cound not be queued.
|
|
|
|
// Wait the requested time before resuming loop.
|
|
|
|
time.Sleep(minWaitTime)
|
tq: avoid a hang when Git is slow to provide us data
When we are cloning or checking out a repo, the filter-process smudge
filter sends the transfer queue data by calling the Add method. This
method inserts items into the incoming channel, which is then processed
by the collectBatches function. This function accepts items, downloads
a batch worth, and splits the remainder into the next and pending
queues, and then, if there are any items in either of these queues, goes
around again. If there are no items, it exits the download queue.
Unfortunately, this algorithm has a problem. If the filter-process
portion is slow and does not provide us enough data, we can end up with
both the next and pending queues empty, but the incoming channel still
open, waiting for the filter-process end to send more data. In such a
case, we stop downloading early and the process deadlocks since there's
nobody to read from the channel.
The solution, however, is simple: since we set the closing flag whenever
the incoming channel is closed, we can check if that flag is set and
only quit if it is. We know that if it is not, there's more data, and
the other end of the channel is just being slow (perhaps because Git is
processing other filters). Do this to avoid the deadlock and ensure
that we terminate properly in all cases.
2019-09-03 21:32:35 +00:00
|
|
|
} else if len(next) == 0 && len(pending) == 0 && closing {
|
2018-12-26 15:30:53 +00:00
|
|
|
// There are no items remaining, it is safe to break
|
2016-12-09 20:55:04 +00:00
|
|
|
break
|
|
|
|
}
|
2017-08-07 20:29:16 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// collectPendingUntil collects items from q.incoming into a "pending" batch
|
|
|
|
// until the given "done" channel is written to, or is closed.
|
|
|
|
//
|
|
|
|
// A "pending" batch is returned, along with whether or not "q.incoming" is
|
|
|
|
// closed.
|
|
|
|
func (q *TransferQueue) collectPendingUntil(done <-chan struct{}) (pending batch, closing bool) {
|
|
|
|
for {
|
|
|
|
select {
|
|
|
|
case t, ok := <-q.incoming:
|
|
|
|
if !ok {
|
|
|
|
closing = true
|
|
|
|
<-done
|
|
|
|
return
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2017-08-07 20:29:16 +00:00
|
|
|
pending = append(pending, t)
|
|
|
|
case <-done:
|
|
|
|
return
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
// 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-24 21:12:07 +00:00
|
|
|
q.meter.Pause()
|
2017-07-20 14:09:14 +00:00
|
|
|
var bRes *BatchResponse
|
|
|
|
if q.manifest.standaloneTransferAgent != "" {
|
|
|
|
// Trust the external transfer agent can do everything by itself.
|
|
|
|
objects := make([]*Transfer, 0, len(batch))
|
2016-12-09 20:55:04 +00:00
|
|
|
for _, t := range batch {
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
objects = append(objects, &Transfer{Oid: t.Oid, Size: t.Size, Path: t.Path, Missing: t.Missing})
|
2017-07-20 14:09:14 +00:00
|
|
|
}
|
|
|
|
bRes = &BatchResponse{
|
|
|
|
Objects: objects,
|
|
|
|
TransferAdapterName: q.manifest.standaloneTransferAgent,
|
|
|
|
}
|
|
|
|
} else {
|
|
|
|
// Query the Git LFS server for what transfer method to use and
|
|
|
|
// details such as URLs, authentication, etc.
|
|
|
|
var err error
|
2018-01-05 18:12:57 +00:00
|
|
|
bRes, err = Batch(q.manifest, q.direction, q.remote, q.ref, batch.ToTransfers())
|
2017-07-20 14:09:14 +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 {
|
|
|
|
if q.canRetryObject(t.Oid, err) {
|
|
|
|
q.rc.Increment(t.Oid)
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2018-12-26 15:30:53 +00:00
|
|
|
next = append(next, t)
|
|
|
|
} else if readyTime, canRetry := q.canRetryObjectLater(t.Oid, err); canRetry {
|
|
|
|
tracerx.Printf("tq: retrying object %s after %s seconds.", t.Oid, time.Until(readyTime).Seconds())
|
|
|
|
err = nil
|
|
|
|
t.ReadyTime = readyTime
|
2017-07-20 14:09:14 +00:00
|
|
|
next = append(next, t)
|
|
|
|
} else {
|
|
|
|
q.wait.Done()
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
|
2017-07-20 14:09:14 +00:00
|
|
|
return next, err
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
|
2017-01-04 17:11:16 +00:00
|
|
|
if len(bRes.Objects) == 0 {
|
|
|
|
return next, nil
|
|
|
|
}
|
|
|
|
|
Don't fail if we lack objects the server has
A Git LFS client may not have the entire history of the objects for the
repository. However, in some situations, we traverse the entire history
of a branch when pushing it, meaning that we need to process every
LFS object in the history of that branch. If the objects for the entire
history are not present, we currently fail to push.
Instead, let's mark objects we don't have on disk as missing and only
fail when we would need to upload those objects. We'll know the server
has the objects if the batch response provides no actions to take for
them when we request an upload. Pass the missing flag down through the
code, and always set it to false for non-uploads.
If for some reason we fail to properly flag a missing object, we will
still fail later on when we cannot open the file, just in a messier and
more poorly controlled way. The technique used here will attempt to
abort the batch as soon as we notice a problem, which means that in the
common case (less than 100 objects) we won't have transferred any
objects, so the user can notice the failure as soon as possible.
Update the tests to look for a string which will occur in the error
message, since we no longer produce the system error message for ENOENT.
2019-04-30 19:18:18 +00:00
|
|
|
// We check first that all of the objects we want to upload are present,
|
|
|
|
// and abort if any are missing. We'll never have any objects marked as
|
|
|
|
// missing except possibly on upload, so just skip iterating over the
|
|
|
|
// objects in that case.
|
|
|
|
if q.direction == Upload {
|
|
|
|
for _, o := range bRes.Objects {
|
|
|
|
// If the server already has the object, the list of
|
|
|
|
// actions will be empty. It's fine if the file is
|
|
|
|
// missing in that case, since we don't need to upload
|
|
|
|
// it.
|
|
|
|
if o.Missing && len(o.Actions) != 0 {
|
|
|
|
return nil, errors.Errorf("Unable to find source for object %v (try running git lfs fetch --all)", o.Oid)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-01-09 20:04:58 +00:00
|
|
|
q.useAdapter(bRes.TransferAdapterName)
|
2017-01-24 21:12:07 +00:00
|
|
|
q.meter.Start()
|
2016-12-09 20:55:04 +00:00
|
|
|
|
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()
|
2017-08-03 20:44:53 +00:00
|
|
|
objects, ok := q.transfers[o.Oid]
|
2016-12-14 17:57:34 +00:00
|
|
|
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-08-03 20:44:53 +00:00
|
|
|
// Pick t[0], since it will cover all transfers with the
|
|
|
|
// same OID.
|
|
|
|
tr := newTransfer(o, objects.First().Name, objects.First().Path)
|
2016-12-14 17:57:34 +00:00
|
|
|
|
2017-03-06 21:33:58 +00:00
|
|
|
if a, err := tr.Rel(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
|
|
|
|
2017-04-26 17:23:29 +00:00
|
|
|
tracerx.Printf("tq: enqueue retry #%d for %q (size: %d): %s", count, tr.Oid, tr.Size, err)
|
2017-08-03 20:44:53 +00:00
|
|
|
next = append(next, objects.First())
|
2016-12-14 17:57:34 +00:00
|
|
|
} else {
|
2017-03-06 19:36:21 +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()
|
|
|
|
}
|
2017-07-20 14:09:14 +00:00
|
|
|
} else if a == nil && q.manifest.standaloneTransferAgent == "" {
|
2017-03-06 19:36:21 +00:00
|
|
|
q.Skip(o.Size)
|
|
|
|
q.wait.Done()
|
2016-12-14 17:57:34 +00:00
|
|
|
} else {
|
2017-08-03 20:44:53 +00:00
|
|
|
q.meter.StartTransfer(objects.First().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.
|
2018-09-06 21:42:41 +00:00
|
|
|
func (q *TransferQueue) addToAdapter(e lfshttp.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
|
|
|
|
}
|
|
|
|
|
2017-01-11 20:59:25 +00:00
|
|
|
present, missingResults := q.partitionTransfers(pending)
|
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
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 {
|
2017-01-11 20:59:25 +00:00
|
|
|
results = q.makeDryRunResults(present)
|
2016-12-09 20:55:04 +00:00
|
|
|
} else {
|
2017-01-11 20:59:25 +00:00
|
|
|
results = q.adapter.Add(present...)
|
2016-12-09 20:55:04 +00:00
|
|
|
}
|
|
|
|
|
2017-01-11 20:59:25 +00:00
|
|
|
for _, res := range missingResults {
|
|
|
|
q.handleTransferResult(res, retries)
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
for res := range results {
|
|
|
|
q.handleTransferResult(res, retries)
|
|
|
|
}
|
|
|
|
}()
|
|
|
|
|
|
|
|
return retries
|
|
|
|
}
|
|
|
|
|
2017-01-11 20:59:25 +00:00
|
|
|
func (q *TransferQueue) partitionTransfers(transfers []*Transfer) (present []*Transfer, results []TransferResult) {
|
|
|
|
if q.direction != Upload {
|
|
|
|
return transfers, nil
|
|
|
|
}
|
|
|
|
|
|
|
|
present = make([]*Transfer, 0, len(transfers))
|
|
|
|
results = make([]TransferResult, 0, len(transfers))
|
|
|
|
|
|
|
|
for _, t := range transfers {
|
|
|
|
var err error
|
|
|
|
|
|
|
|
if t.Size < 0 {
|
|
|
|
err = errors.Errorf("Git LFS: object %q has invalid size (got: %d)", t.Oid, t.Size)
|
|
|
|
} else {
|
|
|
|
fd, serr := os.Stat(t.Path)
|
2017-03-24 20:37:09 +00:00
|
|
|
if serr != nil {
|
|
|
|
if os.IsNotExist(serr) {
|
|
|
|
err = newObjectMissingError(t.Name, t.Oid)
|
|
|
|
} else {
|
|
|
|
err = serr
|
|
|
|
}
|
|
|
|
} else if t.Size != fd.Size() {
|
|
|
|
err = newCorruptObjectError(t.Name, t.Oid)
|
2017-01-11 20:59:25 +00:00
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
if err != nil {
|
|
|
|
results = append(results, TransferResult{
|
|
|
|
Transfer: t,
|
|
|
|
Error: err,
|
|
|
|
})
|
|
|
|
} else {
|
|
|
|
present = append(present, t)
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
return
|
|
|
|
}
|
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
// 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:
|
2019-01-07 09:27:02 +00:00
|
|
|
if readyTime, canRetry := q.canRetryObjectLater(oid, res.Error); canRetry {
|
|
|
|
// If the object can't be retried now, but can be
|
|
|
|
// after a certain period of time, send it to
|
|
|
|
// the retry channel with a time when it's ready.
|
|
|
|
tracerx.Printf("tq: retrying object %s after %s seconds.", oid, time.Until(readyTime).Seconds())
|
2016-12-09 20:55:04 +00:00
|
|
|
q.trMutex.Lock()
|
2017-08-03 20:44:53 +00:00
|
|
|
objects, ok := q.transfers[oid]
|
2016-12-09 20:55:04 +00:00
|
|
|
q.trMutex.Unlock()
|
|
|
|
|
|
|
|
if ok {
|
2019-01-07 09:27:02 +00:00
|
|
|
t := objects.First()
|
|
|
|
t.ReadyTime = readyTime
|
|
|
|
retries <- t
|
2016-12-09 20:55:04 +00:00
|
|
|
} else {
|
|
|
|
q.errorc <- res.Error
|
|
|
|
}
|
2019-01-07 09:27:02 +00:00
|
|
|
} else 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: %s", oid, res.Error)
|
|
|
|
|
2018-12-26 15:30:53 +00:00
|
|
|
q.trMutex.Lock()
|
|
|
|
objects, ok := q.transfers[oid]
|
|
|
|
q.trMutex.Unlock()
|
|
|
|
|
|
|
|
if ok {
|
2019-01-07 09:27:02 +00:00
|
|
|
retries <- objects.First()
|
2018-12-26 15:30:53 +00:00
|
|
|
} else {
|
|
|
|
q.errorc <- res.Error
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
} 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
|
2018-07-31 18:24:17 +00:00
|
|
|
// immediately (unless the error is in response to a
|
|
|
|
// HTTP 422).
|
|
|
|
if errors.IsUnprocessableEntityError(res.Error) {
|
|
|
|
q.unsupportedContentType = true
|
|
|
|
} else {
|
|
|
|
q.errorc <- res.Error
|
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
q.wait.Done()
|
|
|
|
}
|
|
|
|
} else {
|
2017-08-03 20:44:53 +00:00
|
|
|
q.trMutex.Lock()
|
|
|
|
objects := q.transfers[oid]
|
|
|
|
objects.completed = true
|
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
// Otherwise, if the transfer was successful, notify all of the
|
|
|
|
// watchers, and mark it as finished.
|
2017-01-11 21:11:36 +00:00
|
|
|
for _, c := range q.watchers {
|
2017-08-03 20:44:53 +00:00
|
|
|
// Send one update for each transfer with the
|
|
|
|
// same OID.
|
|
|
|
for _, t := range objects.All() {
|
|
|
|
c <- &Transfer{
|
|
|
|
Name: t.Name,
|
|
|
|
Path: t.Path,
|
|
|
|
Oid: t.Oid,
|
|
|
|
Size: t.Size,
|
|
|
|
}
|
|
|
|
}
|
2017-01-11 21:11:36 +00:00
|
|
|
}
|
2016-12-09 20:55:04 +00:00
|
|
|
|
2017-08-03 20:44:53 +00:00
|
|
|
q.trMutex.Unlock()
|
|
|
|
|
2016-12-09 20:55:04 +00:00
|
|
|
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
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
2017-08-25 15:56:09 +00:00
|
|
|
// BatchSize returns the batch size of the receiving *TransferQueue, or, the
|
|
|
|
// number of transfers to accept before beginning work on them.
|
|
|
|
func (q *TransferQueue) BatchSize() int {
|
|
|
|
return q.batchSize
|
|
|
|
}
|
|
|
|
|
2016-04-06 19:06:34 +00:00
|
|
|
func (q *TransferQueue) Skip(size int64) {
|
|
|
|
q.meter.Skip(size)
|
|
|
|
}
|
|
|
|
|
2018-09-06 21:42:41 +00:00
|
|
|
func (q *TransferQueue) ensureAdapterBegun(e lfshttp.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)
|
2017-08-01 18:58:18 +00:00
|
|
|
if q.cb != nil {
|
2017-08-01 18:58:35 +00:00
|
|
|
// NOTE: this is the mechanism by which the logpath
|
|
|
|
// specified by GIT_LFS_PROGRESS is written to.
|
|
|
|
//
|
|
|
|
// See: lfs.downloadFile() for more.
|
2017-08-01 18:58:18 +00:00
|
|
|
q.cb(total, read, 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
|
|
|
}
|
|
|
|
|
2018-09-06 21:42:41 +00:00
|
|
|
func (q *TransferQueue) toAdapterCfg(e lfshttp.Endpoint) AdapterConfig {
|
2017-01-04 22:18:23 +00:00
|
|
|
apiClient := q.manifest.APIClient()
|
|
|
|
concurrency := q.manifest.ConcurrentTransfers()
|
2018-09-24 18:50:09 +00:00
|
|
|
access := apiClient.Endpoints.AccessFor(e.Url)
|
2018-10-02 23:35:54 +00:00
|
|
|
if access.Mode() == lfsapi.NTLMAccess {
|
2017-01-04 21:46:30 +00:00
|
|
|
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
|
|
|
}
|
|
|
|
|
2018-07-31 18:24:17 +00:00
|
|
|
var (
|
|
|
|
// contentTypeWarning is the message printed when a server returns an
|
|
|
|
// HTTP 422 at the end of a push.
|
|
|
|
contentTypeWarning = []string{
|
|
|
|
"Uploading failed due to unsupported Content-Type header(s).",
|
|
|
|
"Consider disabling Content-Type detection with:",
|
|
|
|
"",
|
|
|
|
" $ git config lfs.contenttype false",
|
|
|
|
}
|
|
|
|
)
|
|
|
|
|
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
|
|
|
|
2018-01-04 18:49:41 +00:00
|
|
|
q.meter.Flush()
|
2015-09-04 15:21:26 +00:00
|
|
|
q.errorwait.Wait()
|
2018-07-31 18:24:17 +00:00
|
|
|
|
|
|
|
if q.unsupportedContentType {
|
|
|
|
for _, line := range contentTypeWarning {
|
|
|
|
fmt.Fprintf(os.Stderr, "info: %s\n", line)
|
|
|
|
}
|
|
|
|
}
|
2015-05-21 16:36:49 +00:00
|
|
|
}
|
|
|
|
|
2017-08-02 19:17:39 +00:00
|
|
|
// Watch returns a channel where the queue will write the value of each transfer
|
2017-08-03 20:45:06 +00:00
|
|
|
// as it completes. If multiple transfers exist with the same OID, they will all
|
|
|
|
// be recorded here, even though only one actual transfer took place. The
|
|
|
|
// channel will be closed when the queue finishes processing.
|
2017-08-02 19:17:39 +00:00
|
|
|
func (q *TransferQueue) Watch() chan *Transfer {
|
|
|
|
c := make(chan *Transfer, 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
|
|
|
|
2017-02-13 21:30:59 +00:00
|
|
|
// run begins the transfer queue. It transfers files sequentially or
|
2015-07-10 20:05:04 +00:00
|
|
|
// 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)
|
|
|
|
}
|
|
|
|
|
2018-12-26 15:30:53 +00:00
|
|
|
// canRetryLater returns the number of seconds until an error can be retried and if the error
|
|
|
|
// is a delayed-retriable error.
|
|
|
|
func (q *TransferQueue) canRetryLater(err error) (time.Time, bool) {
|
|
|
|
return errors.IsRetriableLaterError(err)
|
|
|
|
}
|
|
|
|
|
2016-09-21 19:46:34 +00:00
|
|
|
// 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
|
|
|
}
|
|
|
|
|
2018-12-26 15:30:53 +00:00
|
|
|
func (q *TransferQueue) canRetryObjectLater(oid string, err error) (time.Time, bool) {
|
|
|
|
if count, ok := q.rc.CanRetry(oid); !ok {
|
|
|
|
tracerx.Printf("tq: refusing to retry %q, too many retries (%d)", oid, count)
|
|
|
|
return time.Time{}, false
|
|
|
|
}
|
|
|
|
|
|
|
|
return q.canRetryLater(err)
|
|
|
|
}
|
|
|
|
|
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
|
|
|
|
}
|