Arvados

Possible implementation

In (*CollectionFileWriter)goUpload(), currently we have

goUpload() {
  for block := range uploader {
    compute md5
    write to keep
    append signed locator to m.ManifestStream.Blocks
    (or append err to errors)
  }
}

We can add a buffered channel to CollectionWriter and pass it to (CFW)goUpload() and use it to limit the total number of blocks being written by all CFWs of a given CollectionWriter.

func (m *CollectionWriter) Open(path string) io.WriteCloser {
  ...
  m.mtx.Lock()
  defer m.mtx.Unlock()
  if m.workers == nil {
    if m.MaxWriters < 1 {
      m.MaxWriters = 1
    }
    m.workers = make(chan struct{}, m.MaxWriters)
  }
  go fw.goUpload(m.workers)
  m.Streams = append(m.Streams, fw)
  ...
}

goUpload(workers chan struct{}) {
  var blocks []string
  var mtx sync.Mutex
  for block := range uploader {
    mtx.Lock()
    blockIndex := len(blocks)
    blocks = append(blocks, "")
    mtx.Unlock()

    workers <- struct{}{}    // wait for an available worker slot
    wg.Add(1)

    go func(blockIndex int) {
      compute md5
      write to keep
      <-workers              // release worker slot

      mtx.Lock()
      m.ManifestStream.Blocks[blockIndex] = signedLocator
      (or append err to errors)
      mtx.Unlock()

      wg.Done()
    }(blockIndex)
  }
  wg.Wait()
  finish <- errors
}

Sorry for taking so long to send comments, was trying to understand the code the best I could.
One question: Are the upload workers number configurable? I'm not able to see where, it seems to be hardcoded to be 2.

I've run services/crunch-run tests locally without issues.

Are the upload workers number configurable?

Number of writers >1 will help us as this let's us use multiple threads to upload.

Thanks.