X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/5fb633ee021c613d99280e8958a6598602041011..61c91fe02574ad751c2c957f6ca0d2e1e21105f9:/sdk/go/crunchrunner/upload.go

diff --git a/sdk/go/crunchrunner/upload.go b/sdk/go/crunchrunner/upload.go
index 4feb1421f3..4ced0ce31b 100644
--- a/sdk/go/crunchrunner/upload.go
+++ b/sdk/go/crunchrunner/upload.go
@@ -11,6 +11,8 @@ import (
 	"log"
 	"os"
 	"path/filepath"
+	"sort"
+	"strings"
 )
 
 type Block struct {
@@ -31,10 +33,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) {
@@ -48,37 +49,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 {
@@ -126,7 +124,7 @@ 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
 	}
 
@@ -140,23 +138,22 @@ func (m *ManifestWriter) WalkFunc(path string, info os.FileInfo, err error) erro
 
 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 != "" {
@@ -169,10 +166,23 @@ func (m *ManifestWriter) Finish() error {
 func (m *ManifestWriter) ManifestText() string {
 	m.Finish()
 	var buf bytes.Buffer
-	for k, v := range m.Streams {
+
+	dirs := make([]string, len(m.Streams))
+	i := 0
+	for k := range m.Streams {
+		dirs[i] = k
+		i++
+	}
+	sort.Strings(dirs)
+
+	for _, k := range dirs {
+		v := m.Streams[k]
+
 		if k == "." {
 			buf.WriteString(".")
 		} else {
+			k = strings.Replace(k, " ", "\\040", -1)
+			k = strings.Replace(k, "\n", "", -1)
 			buf.WriteString("./" + k)
 		}
 		for _, b := range v.Blocks {
@@ -181,6 +191,8 @@ func (m *ManifestWriter) ManifestText() string {
 		}
 		for _, f := range v.Files {
 			buf.WriteString(" ")
+			f = strings.Replace(f, " ", "\\040", -1)
+			f = strings.Replace(f, "\n", "", -1)
 			buf.WriteString(f)
 		}
 		buf.WriteString("\n")