Merge pull request #93 from RachelTucker/GOSDK-28-excessive-polling

LogicalChaos · web-flow · commit 61b053642ab6 · 2019-10-18T14:31:52.000-06:00
GOSDK-28: When puts are in progress, the sdk logs are rolling every minute, 48k entries in that time
diff --git a/helpers/blobStrategy.go b/helpers/blobStrategy.go
@@ -4,8 +4,10 @@ import "time"
 
 // Strategy for how to blob objects, used both in writing and reading blob strategies
 type BlobStrategy interface {
-    // performs delay when no job chunks are available
-    delay()
+    // Determines the maximum amount to delay between calls to getAvailableJobChunk.
+    // If blobs are finishing processing, then we will query for more ready blobs earlier.
+    // The recommended duration is five minutes.
+    delay() time.Duration
 
     // determines the maximum number of go routines to be created when transferring objects to/from BP
     maxConcurrentTransfers() uint
@@ -21,8 +23,8 @@ type SimpleBlobStrategy struct {
     MaxWaitingTransfers    uint
 }
 
-func (strategy *SimpleBlobStrategy) delay() {
-    time.Sleep(strategy.Delay)
+func (strategy *SimpleBlobStrategy) delay() time.Duration {
+    return strategy.Delay
 }
 
 func (strategy *SimpleBlobStrategy) maxConcurrentTransfers() uint {
diff --git a/helpers/consumer.go b/helpers/consumer.go
@@ -12,33 +12,50 @@ type consumerImpl struct {
     queue                   *chan TransferOperation
     waitGroup               *sync.WaitGroup
     maxConcurrentOperations uint
+    blobDoneChannel         chan<- struct{}
 }
 
-func newConsumer(queue *chan TransferOperation, waitGroup *sync.WaitGroup, maxConcurrentOperations uint) Consumer {
+func newConsumer(queue *chan TransferOperation, blobDoneChannel chan<- struct{}, waitGroup *sync.WaitGroup, maxConcurrentOperations uint) Consumer {
     return &consumerImpl{
         queue:                   queue,
         waitGroup:               waitGroup,
         maxConcurrentOperations: maxConcurrentOperations,
+        blobDoneChannel:         blobDoneChannel,
     }
 }
 
-func performTransfer(operation *TransferOperation, semaphore *chan int, waitGroup *sync.WaitGroup) {
-    defer waitGroup.Done()
-    (*operation)()
+func performTransfer(operation TransferOperation, semaphore *chan int, blobDoneChannel chan<- struct{}, jobWaitGroup *sync.WaitGroup, childWaitGroup *sync.WaitGroup) {
+    defer func() {
+        // per operation that finishes, send a done message to the producer
+        blobDoneChannel <-struct {}{}
+        jobWaitGroup.Done()
+        childWaitGroup.Done()
+    }()
+    operation()
     <- *semaphore
 }
 
 func (consumer *consumerImpl) run() {
+    // Defer closing the blob done channel. This will signal to the producer that it can shut down.
+    defer func() {close(consumer.blobDoneChannel)}()
+
     // semaphore for controlling max number of transfer operations in flight per job
     semaphore := make(chan int, consumer.maxConcurrentOperations + 1)
+
+    var childWaitGroup sync.WaitGroup
     for {
         nextOp, ok := <- *consumer.queue
         if ok {
             semaphore <- 1
-            go performTransfer(&nextOp, &semaphore, consumer.waitGroup)
+            childWaitGroup.Add(1)
+            go performTransfer(nextOp, &semaphore, consumer.blobDoneChannel, consumer.waitGroup, &childWaitGroup)
         } else {
             consumer.waitGroup.Done()
-            return
+            break
         }
     }
+
+    // Wait for all child transfer operations to finish before shutting down.
+    // This is to stop the done channel from being close prematurely
+    childWaitGroup.Wait()
 }
diff --git a/helpers/consumer_test.go b/helpers/consumer_test.go
@@ -1,9 +1,9 @@
 package helpers
 
 import (
-    "testing"
-    "sync"
     "github.com/SpectraLogic/ds3_go_sdk/ds3_utils/ds3Testing"
+    "sync"
+    "testing"
 )
 
 func testTransferBuilder(t *testing.T, i int, resultCount *int, resultMux *sync.Mutex) TransferOperation {
@@ -17,6 +17,7 @@ func testTransferBuilder(t *testing.T, i int, resultCount *int, resultMux *sync.
 }
 
 func TestProducerConsumerModel(t *testing.T) {
+    const numOperations = 10
     var wg sync.WaitGroup
     wg.Add(2)
 
@@ -25,7 +26,7 @@ func TestProducerConsumerModel(t *testing.T) {
 
     var producer = func(queue *chan TransferOperation) {
         defer wg.Done()
-        for i := 0; i < 10; i++ {
+        for i := 0; i < numOperations; i++ {
             wg.Add(1)
 
             var transferOf = testTransferBuilder(t, i, &resultCount, &resultMux)
@@ -39,12 +40,24 @@ func TestProducerConsumerModel(t *testing.T) {
 
     queue := make(chan TransferOperation, 5)
 
-    consumer := newConsumer(&queue, &wg, 5)
+    // make the blob done channel larger than the number of transfer operations queued.
+    blobDoneChannel := make(chan struct{}, numOperations+1)
+
+    consumer := newConsumer(&queue, blobDoneChannel, &wg, 5)
 
     go producer(&queue)
     go consumer.run()
 
     wg.Wait()
 
-    ds3Testing.AssertInt(t, "Executed Transfer Operations", 10, resultCount)
+    ds3Testing.AssertInt(t, "Executed Transfer Operations", numOperations, resultCount)
+
+    // verify that 10 done messages were sent
+    ds3Testing.AssertInt(t, "Done signals sent", numOperations, len(blobDoneChannel))
+    for len(blobDoneChannel) > 0 {
+        _, ok := <-blobDoneChannel
+        ds3Testing.AssertBool(t, "expected channel not to be closed", true, ok)
+    }
+    _, ok := <- blobDoneChannel
+    ds3Testing.AssertBool(t, "expected channel to be closed", false, ok)
 }
diff --git a/helpers/getProducer.go b/helpers/getProducer.go
@@ -8,6 +8,7 @@ import (
     "github.com/SpectraLogic/ds3_go_sdk/sdk_log"
     "io"
     "sync"
+    "time"
 )
 
 type getProducer struct {
@@ -22,9 +23,25 @@ type getProducer struct {
     deferredBlobQueue    BlobDescriptionQueue // queue of blobs whose channels are not yet ready for transfer
     rangeFinder          ranges.BlobRangeFinder
     sdk_log.Logger
+
+    // Channel that represents blobs that have finished being process.
+    // This will be written to once a get object operation has completed regardless of error or success.
+    // This is used to notify the runner to re-check if any blobs are now ready to be retrieved.
+    blobDoneChannel <-chan struct{}
+
+    // Used to track if we are done queuing blobs
+    continueQueuingBlobs bool
 }
 
-func newGetProducer(jobMasterObjectList *ds3Models.MasterObjectList, getObjects *[]helperModels.GetObject, queue *chan TransferOperation, strategy *ReadTransferStrategy, client *ds3.Client, waitGroup *sync.WaitGroup) *getProducer {
+func newGetProducer(
+    jobMasterObjectList *ds3Models.MasterObjectList,
+    getObjects *[]helperModels.GetObject,
+    queue *chan TransferOperation,
+    strategy *ReadTransferStrategy,
+    blobDoneChannel <-chan struct{},
+    client *ds3.Client,
+    waitGroup *sync.WaitGroup) *getProducer {
+
     return &getProducer{
         JobMasterObjectList:  jobMasterObjectList,
         GetObjects:           getObjects,
@@ -37,6 +54,8 @@ func newGetProducer(jobMasterObjectList *ds3Models.MasterObjectList, getObjects
         deferredBlobQueue:    NewBlobDescriptionQueue(),
         rangeFinder:          ranges.NewBlobRangeFinder(getObjects),
         Logger:               client.Logger, //use the same logger as the client
+        blobDoneChannel:      blobDoneChannel,
+        continueQueuingBlobs: true,
     }
 }
 
@@ -217,40 +236,103 @@ func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string
 // Each transfer operation will retrieve one blob of content from the BP.
 // Once all blobs have been queued to be transferred, the producer will finish, even if all operations have not been consumed yet.
 func (producer *getProducer) run() error {
-    defer close(*producer.queue)
-
     // determine number of blobs to be processed
     var totalBlobCount int64 = producer.totalBlobCount()
     producer.Debugf("job status totalBlobs=%d processedBlobs=%d", totalBlobCount, producer.processedBlobTracker.NumberOfProcessedBlobs())
 
-    // process all chunks and make sure all blobs are queued for transfer
-    for producer.processedBlobTracker.NumberOfProcessedBlobs() < totalBlobCount || producer.deferredBlobQueue.Size() > 0 {
-        // Get the list of available chunks that the server can receive. The server may
-        // not be able to receive everything, so not all chunks will necessarily be
-        // returned
-        chunksReady := ds3Models.NewGetJobChunksReadyForClientProcessingSpectraS3Request(producer.JobMasterObjectList.JobId)
-        chunksReadyResponse, err := producer.client.GetJobChunksReadyForClientProcessingSpectraS3(chunksReady)
-        if err != nil {
-            producer.Errorf("unrecoverable error: %v", err)
-            return err
-        }
+    // initiate first set of blob transfers
+    err := producer.queueBlobsReadyForTransfer(totalBlobCount)
+    if err != nil {
+        return err
+    }
 
-        // Check to see if any chunks can be processed
-        numberOfChunks := len(chunksReadyResponse.MasterObjectList.Objects)
-        if numberOfChunks > 0 {
-            // Loop through all the chunks that are available for processing, and send
-            // the files that are contained within them.
-            for _, curChunk := range chunksReadyResponse.MasterObjectList.Objects {
-                producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
+    // wait for either a timer or for at least one blob to finish before attempting to queue more items for transfer
+    ticker := time.NewTicker(producer.strategy.BlobStrategy.delay())
+    var fatalErr error
+    for {
+        select {
+        case _, ok := <- producer.blobDoneChannel:
+            if ok {
+                // reset the timer
+                ticker.Stop()
+                ticker = time.NewTicker(producer.strategy.BlobStrategy.delay())
+
+                err = producer.queueBlobsReadyForTransfer(totalBlobCount)
+                if err != nil {
+                    // A fatal error has occurred, stop queuing blobs for processing and
+                    // close processing queue to signal consumer we won't be sending any more blobs.
+                    producer.continueQueuingBlobs = false
+                    fatalErr = err
+                    close(*producer.queue)
+                }
+            } else {
+                // The consumer closed the channel, signaling completion.
+                return fatalErr
+            }
+        case <- ticker.C:
+            err = producer.queueBlobsReadyForTransfer(totalBlobCount)
+            if err != nil {
+                // A fatal error has occurred, stop queuing blobs for processing and
+                // close processing queue to signal consumer we won't be sending any more blobs.
+                producer.continueQueuingBlobs = false
+                fatalErr = err
+                close(*producer.queue)
             }
+        }
+    }
+    return fatalErr
+}
+
+func (producer *getProducer) hasMoreToProcess(totalBlobCount int64) bool {
+    return producer.processedBlobTracker.NumberOfProcessedBlobs() < totalBlobCount || producer.deferredBlobQueue.Size() > 0
+}
+
+func (producer *getProducer) queueBlobsReadyForTransfer(totalBlobCount int64) error {
+    if !producer.continueQueuingBlobs {
+        // We've queued up all the blobs we are going to for this job.
+        return nil
+    }
+
+    // check if there is anything left to be queued
+    if !producer.hasMoreToProcess(totalBlobCount) {
+        // Everything has been queued for processing.
+        producer.continueQueuingBlobs = false
+        // close processing queue to signal consumer we won't be sending any more blobs.
+        close(*producer.queue)
+        return nil
+    }
+
+    // Attempt to transfer waiting blobs
+    producer.processWaitingBlobs(*producer.JobMasterObjectList.BucketName, producer.JobMasterObjectList.JobId)
+
+    // Check if we need to query the BP for allocated blobs, or if we already know everything is allocated.
+    if int64(producer.deferredBlobQueue.Size()) + producer.processedBlobTracker.NumberOfProcessedBlobs() >= totalBlobCount {
+        // Everything is already allocated, no need to query BP for allocated chunks
+        return nil
+    }
+
+    // Get the list of available chunks that the server can receive. The server may
+    // not be able to receive everything, so not all chunks will necessarily be
+    // returned
+    chunksReady := ds3Models.NewGetJobChunksReadyForClientProcessingSpectraS3Request(producer.JobMasterObjectList.JobId)
+    chunksReadyResponse, err := producer.client.GetJobChunksReadyForClientProcessingSpectraS3(chunksReady)
+    if err != nil {
+        producer.Errorf("unrecoverable error: %v", err)
+        return err
+    }
 
-            // Attempt to transfer waiting blobs
-            producer.processWaitingBlobs(*chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
-        } else {
-            // When no chunks are returned we need to sleep to allow for cache space to
-            // be freed.
-            producer.strategy.BlobStrategy.delay()
+    // Check to see if any chunks can be processed
+    numberOfChunks := len(chunksReadyResponse.MasterObjectList.Objects)
+    if numberOfChunks > 0 {
+        // Loop through all the chunks that are available for processing, and send
+        // the files that are contained within them.
+        for _, curChunk := range chunksReadyResponse.MasterObjectList.Objects {
+            producer.processChunk(&curChunk, *chunksReadyResponse.MasterObjectList.BucketName, chunksReadyResponse.MasterObjectList.JobId)
         }
+    } else {
+        // When no chunks are returned we need to sleep to allow for cache space to
+        // be freed.
+        producer.strategy.BlobStrategy.delay()
     }
     return nil
 }
diff --git a/helpers/getTransfernator.go b/helpers/getTransfernator.go
@@ -78,9 +78,10 @@ func (transceiver *getTransceiver) transfer() (string, error) {
     // init queue, producer and consumer
     var waitGroup sync.WaitGroup
 
+    blobDoneChannel := make(chan struct{}, 10)
     queue := newOperationQueue(transceiver.Strategy.BlobStrategy.maxWaitingTransfers(), transceiver.Client.Logger)
-    producer := newGetProducer(&bulkGetResponse.MasterObjectList, transceiver.ReadObjects, &queue, transceiver.Strategy, transceiver.Client, &waitGroup)
-    consumer := newConsumer(&queue, &waitGroup, transceiver.Strategy.BlobStrategy.maxConcurrentTransfers())
+    producer := newGetProducer(&bulkGetResponse.MasterObjectList, transceiver.ReadObjects, &queue, transceiver.Strategy, blobDoneChannel, transceiver.Client, &waitGroup)
+    consumer := newConsumer(&queue, blobDoneChannel, &waitGroup, transceiver.Strategy.BlobStrategy.maxConcurrentTransfers())
 
     // Wait for completion of producer-consumer goroutines
     var aggErr ds3Models.AggregateError
diff --git a/helpers/putProducer.go b/helpers/putProducer.go
diff --git a/helpers/putTransceiver.go b/helpers/putTransceiver.go
