Merge branch '2411-check-copyright'

[arvados.git] / sdk / go / crunchrunner / upload.go

diff --git a/sdk/go/crunchrunner/upload.go b/sdk/go/crunchrunner/upload.go
index ac3f0655a7b4db94250addbe3d1fe167e79d23c9..fd24908ad6a7e9e49721c0ed33980ce096c7e34a 100644 (file)
--- a/sdk/go/crunchrunner/upload.go
+++ b/sdk/go/crunchrunner/upload.go
@@ -1,3 +1,7 @@
+// Copyright (C) The Arvados Authors. All rights reserved.
+//
+// SPDX-License-Identifier: Apache-2.0
+
 package main
 
 import (
@@ -33,10 +37,9 @@ type IKeepClient interface {
        PutHB(hash string, buf []byte) (string, int, error)
 }
 
-func (m *ManifestStreamWriter) Write(p []byte) (n int, err error) {
-       // Needed to conform to Writer interface, but not implemented
-       // because io.Copy will actually use ReadFrom instead.
-       return 0, nil
+func (m *ManifestStreamWriter) Write(p []byte) (int, error) {
+       n, err := m.ReadFrom(bytes.NewReader(p))
+       return int(n), err
 }
 
 func (m *ManifestStreamWriter) ReadFrom(r io.Reader) (n int64, err error) {
@@ -50,37 +53,34 @@ func (m *ManifestStreamWriter) ReadFrom(r io.Reader) (n int64, err error) {
                count, err = r.Read(m.Block.data[m.Block.offset:])
                total += int64(count)
                m.Block.offset += int64(count)
-               if count > 0 {
-                       if m.Block.offset == keepclient.BLOCKSIZE {
-                               m.uploader <- m.Block
-                               m.Block = nil
-                       }
+               if m.Block.offset == keepclient.BLOCKSIZE {
+                       m.uploader <- m.Block
+                       m.Block = nil
                }
        }
 
-       return total, err
+       if err == io.EOF {
+               return total, nil
+       } else {
+               return total, err
+       }
+
 }
 
 func (m *ManifestStreamWriter) goUpload() {
        var errors []error
        uploader := m.uploader
        finish := m.finish
-       for true {
-               select {
-               case block, valid := <-uploader:
-                       if !valid {
-                               finish <- errors
-                               return
-                       }
-                       hash := fmt.Sprintf("%x", md5.Sum(block.data[0:block.offset]))
-                       signedHash, _, err := m.ManifestWriter.IKeepClient.PutHB(hash, block.data[0:block.offset])
-                       if err != nil {
-                               errors = append(errors, err)
-                       } else {
-                               m.ManifestStream.Blocks = append(m.ManifestStream.Blocks, signedHash)
-                       }
+       for block := range uploader {
+               hash := fmt.Sprintf("%x", md5.Sum(block.data[0:block.offset]))
+               signedHash, _, err := m.ManifestWriter.IKeepClient.PutHB(hash, block.data[0:block.offset])
+               if err != nil {
+                       errors = append(errors, err)
+               } else {
+                       m.ManifestStream.Blocks = append(m.ManifestStream.Blocks, signedHash)
                }
        }
+       finish <- errors
 }
 
 type ManifestWriter struct {
@@ -128,37 +128,36 @@ func (m *ManifestWriter) WalkFunc(path string, info os.FileInfo, err error) erro
 
        var count int64
        count, err = io.Copy(stream, file)
-       if err != nil && err != io.EOF {
+       if err != nil {
                return err
        }
 
        stream.offset += count
 
-       stream.ManifestStream.Files = append(stream.ManifestStream.Files,
-               fmt.Sprintf("%v:%v:%v", fileStart, count, fn))
+       stream.ManifestStream.FileStreamSegments = append(stream.ManifestStream.FileStreamSegments,
+               manifest.FileStreamSegment{uint64(fileStart), uint64(count), fn})
 
        return nil
 }
 
 func (m *ManifestWriter) Finish() error {
        var errstring string
-       for _, v := range m.Streams {
-               if v.uploader != nil {
-                       if v.Block != nil {
-                               v.uploader <- v.Block
-                       }
-                       close(v.uploader)
-                       v.uploader = nil
-
-                       errors := <-v.finish
-                       close(v.finish)
-                       v.finish = nil
-
-                       if errors != nil {
-                               for _, r := range errors {
-                                       errstring = fmt.Sprintf("%v%v\n", errstring, r.Error())
-                               }
-                       }
+       for _, stream := range m.Streams {
+               if stream.uploader == nil {
+                       continue
+               }
+               if stream.Block != nil {
+                       stream.uploader <- stream.Block
+               }
+               close(stream.uploader)
+               stream.uploader = nil
+
+               errors := <-stream.finish
+               close(stream.finish)
+               stream.finish = nil
+
+               for _, r := range errors {
+                       errstring = fmt.Sprintf("%v%v\n", errstring, r.Error())
                }
        }
        if errstring != "" {
@@ -194,11 +193,11 @@ func (m *ManifestWriter) ManifestText() string {
                        buf.WriteString(" ")
                        buf.WriteString(b)
                }
-               for _, f := range v.Files {
+               for _, f := range v.FileStreamSegments {
                        buf.WriteString(" ")
-                       f = strings.Replace(f, " ", "\\040", -1)
-                       f = strings.Replace(f, "\n", "", -1)
-                       buf.WriteString(f)
+                       name := strings.Replace(f.Name, " ", "\\040", -1)
+                       name = strings.Replace(name, "\n", "", -1)
+                       buf.WriteString(fmt.Sprintf("%d:%d:%s", f.SegPos, f.SegLen, name))
                }
                buf.WriteString("\n")
        }