GOSDK-32: System shutdown due to too many go routines when putting lots of files to BP. Removed done channel which was causing go routines in the consumer to be left open longer than necessary.

Author: RachelTucker

ds3_integration/utils/testUtils.go

@@ -375,7 +375,7 @@ func DeleteBucketLogError(t *testing.T, client *ds3.Client, bucketName string) (
 
 // Deletes the specified bucket and returns an error if one occurs
 func DeleteBucket(client *ds3.Client, bucketName string) (error) {
-    deleteBucket, deleteErr := client.DeleteBucket(models.NewDeleteBucketRequest(bucketName))
+    deleteBucket, deleteErr := client.DeleteBucketSpectraS3(models.NewDeleteBucketSpectraS3Request(bucketName).WithForce())
     if deleteErr != nil {
         return deleteErr
     }

helpers/conditionalBool.go

@@ -0,0 +1,43 @@
+package helpers
+
+import "sync"
+
+type NotifyBlobDone interface {
+    // Waits for at least one done signal.
+    Wait()
+
+    // Sends a done signal. Multiple signals have no additional effect.
+    SignalDone()
+}
+
+
+func NewConditionalBool() *ConditionalBool {
+    conditional :=sync.NewCond(&sync.Mutex{})
+    return &ConditionalBool{
+        conditional: *conditional,
+        Done: false,
+    }
+}
+
+type ConditionalBool struct {
+    conditional sync.Cond
+    Done bool
+}
+
+func (conditionalBool *ConditionalBool) Wait() {
+    conditionalBool.conditional.L.Lock()
+    // wait for a done signal to be received
+    for !conditionalBool.Done {
+        conditionalBool.conditional.Wait()
+    }
+    // reset done notifier
+    conditionalBool.Done = false
+    conditionalBool.conditional.L.Unlock()
+}
+
+func (conditionalBool *ConditionalBool) SignalDone() {
+    conditionalBool.conditional.L.Lock()
+    conditionalBool.Done = true
+    conditionalBool.conditional.Broadcast()
+    conditionalBool.conditional.L.Unlock()
+}

helpers/consumer.go

@@ -12,50 +12,42 @@ type consumerImpl struct {
     queue                   *chan TransferOperation
     waitGroup               *sync.WaitGroup
     maxConcurrentOperations uint
-    blobDoneChannel         chan<- struct{}
+
+    // Conditional value that gets triggered when a blob has finished being transferred
+    doneNotifier NotifyBlobDone
 }
 
-func newConsumer(queue *chan TransferOperation, blobDoneChannel chan<- struct{}, waitGroup *sync.WaitGroup, maxConcurrentOperations uint) Consumer {
+func newConsumer(queue *chan TransferOperation, waitGroup *sync.WaitGroup, maxConcurrentOperations uint, doneNotifier NotifyBlobDone) Consumer {
     return &consumerImpl{
         queue:                   queue,
         waitGroup:               waitGroup,
         maxConcurrentOperations: maxConcurrentOperations,
-        blobDoneChannel:         blobDoneChannel,
+        doneNotifier:            doneNotifier,
     }
 }
 
-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()
-    }()
+func performTransfer(operation TransferOperation, semaphore *chan int, waitGroup *sync.WaitGroup, doneNotifier NotifyBlobDone) {
+    defer waitGroup.Done()
     operation()
+
+    // send done signal
+    doneNotifier.SignalDone()
+
     <- *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
-            childWaitGroup.Add(1)
-            go performTransfer(nextOp, &semaphore, consumer.blobDoneChannel, consumer.waitGroup, &childWaitGroup)
+            go performTransfer(nextOp, &semaphore, consumer.waitGroup, consumer.doneNotifier)
         } else {
             consumer.waitGroup.Done()
-            break
+            return
         }
     }
-
-    // Wait for all child transfer operations to finish before shutting down.
-    // This is to stop the done channel from being close prematurely
-    childWaitGroup.Wait()
 }

helpers/consumer_test.go

@@ -40,24 +40,15 @@ func TestProducerConsumerModel(t *testing.T) {
 
     queue := make(chan TransferOperation, 5)
 
-    // make the blob done channel larger than the number of transfer operations queued.
-    blobDoneChannel := make(chan struct{}, numOperations+1)
+    doneNotifier := NewConditionalBool()
 
-    consumer := newConsumer(&queue, blobDoneChannel, &wg, 5)
+    consumer := newConsumer(&queue, &wg, 5, doneNotifier)
 
     go producer(&queue)
     go consumer.run()
 
     wg.Wait()
 
     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)
+    ds3Testing.AssertBool(t, "received done notification", true, doneNotifier.Done)
 }

helpers/getProducer.go

@@ -24,23 +24,18 @@ type getProducer struct {
     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
+    // Conditional value that gets triggered when a blob has finished being transferred
+    doneNotifier NotifyBlobDone
 }
 
 func newGetProducer(
     jobMasterObjectList *ds3Models.MasterObjectList,
     getObjects *[]helperModels.GetObject,
     queue *chan TransferOperation,
     strategy *ReadTransferStrategy,
-    blobDoneChannel <-chan struct{},
     client *ds3.Client,
-    waitGroup *sync.WaitGroup) *getProducer {
+    waitGroup *sync.WaitGroup,
+    doneNotifier NotifyBlobDone) *getProducer {
 
     return &getProducer{
         JobMasterObjectList:  jobMasterObjectList,
@@ -54,8 +49,7 @@ func newGetProducer(
         deferredBlobQueue:    NewBlobDescriptionQueue(),
         rangeFinder:          ranges.NewBlobRangeFinder(getObjects),
         Logger:               client.Logger, //use the same logger as the client
-        blobDoneChannel:      blobDoneChannel,
-        continueQueuingBlobs: true,
+        doneNotifier:         doneNotifier,
     }
 }
 
@@ -75,15 +69,20 @@ func toReadObjectMap(getObjects *[]helperModels.GetObject) map[string]helperMode
 }
 
 // Processes all the blobs in a chunk that are ready for transfer from BP
-func (producer *getProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) {
+// Returns the number of blobs queued for process
+func (producer *getProducer) processChunk(curChunk *ds3Models.Objects, bucketName string, jobId string) int {
     producer.Debugf("begin chunk processing %s", curChunk.ChunkId)
 
+    processedCount := 0
     // transfer blobs that are ready, and queue those that are waiting for channel
     for _, curObj := range curChunk.Objects {
         producer.Debugf("queuing object in waiting to be processed %s offset=%d length=%d", *curObj.Name, curObj.Offset, curObj.Length)
         blob := helperModels.NewBlobDescription(*curObj.Name, curObj.Offset, curObj.Length)
-        producer.queueBlobForTransfer(&blob, bucketName, jobId)
+        if producer.queueBlobForTransfer(&blob, bucketName, jobId) {
+            processedCount++
+        }
     }
+    return processedCount
 }
 
 // Information required to perform a get operation of a blob with BP as data source and channelBuilder as destination
@@ -174,9 +173,10 @@ func writeRangeToDestination(channelBuilder helperModels.WriteChannelBuilder, bl
 
 // Attempts to transfer a single blob from the BP to the client. If the blob is not ready for transfer,
 // then it is added to the waiting to transfer queue
-func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) {
+// Returns whether or not the blob was queued for transfer
+func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescription, bucketName string, jobId string) bool {
     if producer.processedBlobTracker.IsProcessed(*blob) {
-        return
+        return false // already been processed
     }
 
     curReadObj := producer.readObjectMap[blob.Name()]
@@ -185,13 +185,13 @@ func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
         // a fatal error happened on a previous blob for this file, skip processing
         producer.Warningf("fatal error occurred while transferring previous blob on this file, skipping blob '%s' offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
         producer.processedBlobTracker.MarkProcessed(*blob)
-        return
+        return false // not going to process
     }
 
     if !curReadObj.ChannelBuilder.IsChannelAvailable(blob.Offset()) {
         producer.Debugf("channel is not currently available for getting blob '%s' offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
         producer.deferredBlobQueue.Push(blob)
-        return
+        return false // not ready to be processed
     }
 
     producer.Debugf("channel is available for getting blob '%s' offset=%d length=%d", blob.Name(), blob.Offset(), blob.Length())
@@ -212,11 +212,16 @@ func (producer *getProducer) queueBlobForTransfer(blob *helperModels.BlobDescrip
 
     // Mark blob as processed
     producer.processedBlobTracker.MarkProcessed(*blob)
+
+    return true
 }
 
 // Attempts to process all blobs whose channels were not available for transfer.
 // Blobs whose channels are still not available are placed back on the queue.
-func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string) {
+// Returns the number of blobs queued for processing.
+func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string) int {
+    processedCount := 0
+
     // attempt to process all blobs in waiting to be transferred
     waitingBlobs := producer.deferredBlobQueue.Size()
     for i := 0; i < waitingBlobs; i++ {
@@ -228,87 +233,54 @@ func (producer *getProducer) processWaitingBlobs(bucketName string, jobId string
             producer.Errorf("failure during blob transfer '%s' at offset %d: %s", curBlob.Name(), curBlob.Offset(), err.Error())
             break
         }
-        producer.queueBlobForTransfer(curBlob, bucketName, jobId)
+        if producer.queueBlobForTransfer(curBlob, bucketName, jobId) {
+            processedCount++
+        }
     }
+    return processedCount
 }
 
 // This initiates the production of the transfer operations which will be consumed by a consumer running in a separate go routine.
 // 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())
 
-    // initiate first set of blob transfers
-    err := producer.queueBlobsReadyForTransfer(totalBlobCount)
-    if err != nil {
-        return err
-    }
+    // process all chunks and make sure all blobs are queued for transfer
+    for producer.hasMoreToProcess(totalBlobCount) {
+        processedCount, err := producer.queueBlobsReadyForTransfer(totalBlobCount)
+        if err != nil {
+            return err
+        }
 
-    // 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)
-            }
+        // If the last operation processed blobs, then wait for something to finish
+        if processedCount > 0 {
+            producer.doneNotifier.Wait()
+        } else if producer.hasMoreToProcess(totalBlobCount) {
+            // nothing could be processed, cache is probably full, wait a bit before trying again
+            time.Sleep(producer.strategy.BlobStrategy.delay())
         }
     }
-    return fatalErr
+    return nil
 }
 
 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
-    }
-
+// Returns the number of blobs that have been queued for transfer
+func (producer *getProducer) queueBlobsReadyForTransfer(totalBlobCount int64) (int, error) {
     // Attempt to transfer waiting blobs
-    producer.processWaitingBlobs(*producer.JobMasterObjectList.BucketName, producer.JobMasterObjectList.JobId)
+    processedCount := 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
+        return processedCount, nil
     }
 
     // Get the list of available chunks that the server can receive. The server may
@@ -318,7 +290,7 @@ func (producer *getProducer) queueBlobsReadyForTransfer(totalBlobCount int64) er
     chunksReadyResponse, err := producer.client.GetJobChunksReadyForClientProcessingSpectraS3(chunksReady)
     if err != nil {
         producer.Errorf("unrecoverable error: %v", err)
-        return err
+        return processedCount, err
     }
 
     // Check to see if any chunks can be processed
@@ -327,14 +299,10 @@ func (producer *getProducer) queueBlobsReadyForTransfer(totalBlobCount int64) er
         // 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)
+            processedCount += 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
+    return processedCount, nil
 }
 
 // Determines the number of blobs to be transferred.