Rework how large objects are segmented and properly obey SLO min size

This commit unifies the logic handling DLO and SLO segmentation, it also
adds more test cases repeatedly writing and seeking large objects.
The minimum chunk size is now exposed and can also be used with DLOs if
a large number of potentially small objects is not desired (with the
performance penalty of having to retrieve the old segment when writing
over it).

The minimum size for SLOs is also retrived from a /info call and will
override chunkSize and minChunkSize if it's larger.

The implementation of the ReadFrom method was dropped in the writer. The
reason for this is that the old implementation was incorrect sometimes
due to it using the number of written bytes to find out if it had
already finished. If the last segment had the exact same size as the
chunk size it would try writing one more segment, and this segment would
be empty. One way to solve this would be adding a DELETE call if a zero
sized segment was written, but this is ugly and wasteful. Another option
would be to buffer the read data in memory to find out if we're done,
but this is exactly what io.Copy() already do for us, so I could simply
drop the ReadFrom implementation.

Author: cezarsa

dlo.go

@@ -1,45 +1,33 @@
 package swift
 
 import (
-	"bytes"
-	"crypto/md5"
-	"io"
 	"os"
-	"strconv"
 )
 
 // DynamicLargeObjectCreateFile represents an open static large object
 type DynamicLargeObjectCreateFile struct {
-	LargeObjectCreateFile
+	largeObjectCreateFile
 }
 
-// Check it satisfies the interfaces
-var (
-	_ io.Writer     = &DynamicLargeObjectCreateFile{}
-	_ io.Seeker     = &DynamicLargeObjectCreateFile{}
-	_ io.Closer     = &DynamicLargeObjectCreateFile{}
-	_ io.ReaderFrom = &DynamicLargeObjectCreateFile{}
-)
-
 // DynamicLargeObjectCreateFile creates a dynamic large object
 // returning an object which satisfies io.Writer, io.Seeker, io.Closer
 // and io.ReaderFrom.  The flags are as passes to the
 // LargeObjectCreate method.
-func (c *Connection) DynamicLargeObjectCreateFile(opts *LargeObjectOpts) (*DynamicLargeObjectCreateFile, error) {
+func (c *Connection) DynamicLargeObjectCreateFile(opts *LargeObjectOpts) (LargeObjectFile, error) {
 	lo, err := c.LargeObjectCreate(opts)
 	if err != nil {
 		return nil, err
 	}
 
 	return &DynamicLargeObjectCreateFile{
-		LargeObjectCreateFile: *lo,
+		largeObjectCreateFile: *lo,
 	}, nil
 }
 
 // DynamicLargeObjectCreate creates or truncates an existing dynamic
 // large object returning a writeable object.  This sets opts.Flags to
 // an appropriate value before calling DynamicLargeObjectCreateFile
-func (c *Connection) DynamicLargeObjectCreate(opts *LargeObjectOpts) (*DynamicLargeObjectCreateFile, error) {
+func (c *Connection) DynamicLargeObjectCreate(opts *LargeObjectOpts) (LargeObjectFile, error) {
 	opts.Flags = os.O_TRUNC | os.O_CREATE
 	return c.DynamicLargeObjectCreateFile(opts)
 }
@@ -84,142 +72,6 @@ func (c *Connection) createDLOManifest(container string, objectName string, pref
 	return nil
 }
 
-// Write satisfies the io.Writer interface
-func (file *DynamicLargeObjectCreateFile) Write(buf []byte) (int, error) {
-	reader := bytes.NewReader(buf)
-	n, err := file.ReadFrom(reader)
-	return int(n), err
-}
-
-// ReadFrom statisfies the io.ReaderFrom interface
-func (file *DynamicLargeObjectCreateFile) ReadFrom(reader io.Reader) (n int64, err error) {
-	var (
-		multi         io.Reader
-		paddingReader io.Reader
-		cursor        int64
-		currentLength int64
-	)
-
-	partNumber := 1
-	chunkSize := int64(file.chunkSize)
-	readers := []io.Reader{}
-	hash := md5.New()
-
-	file.segments, err = file.conn.getAllDLOSegments(file.segmentContainer, file.prefix)
-	if err != nil {
-		return 0, err
-	}
-
-	for _, segment := range file.segments {
-		currentLength += segment.Bytes
-	}
-
-	// First, we skip the existing segments that are not modified by this call
-	for i := range file.segments {
-		if file.filePos < cursor+file.segments[i].Bytes {
-			break
-		}
-		cursor += file.segments[i].Bytes
-		partNumber++
-		hash.Write([]byte(file.segments[i].Hash))
-	}
-
-	if file.filePos-cursor > 0 && min(currentLength, file.filePos)-cursor > 0 {
-		// Offset is inside the current segment : we need to read the data from
-		// the beginning of the segment to offset, for this we must ensure that
-		// the manifest is already written.
-		err = file.Flush()
-		if err != nil {
-			return 0, err
-		}
-		headers := make(Headers)
-		headers["Range"] = "bytes=" + strconv.FormatInt(cursor, 10) + "-" + strconv.FormatInt(min(currentLength, file.filePos)-1, 10)
-		currentSegment, _, err := file.conn.ObjectOpen(file.container, file.objectName, false, headers)
-		if err != nil {
-			return 0, err
-		}
-		defer currentSegment.Close()
-		paddingReader = currentSegment
-		readers = append(readers, io.LimitReader(paddingReader, min(currentLength, file.filePos)-cursor))
-	}
-
-	if paddingReader != nil {
-		readers = append(readers, io.LimitReader(paddingReader, file.filePos-cursor))
-	}
-	readers = append(readers, io.LimitReader(reader, chunkSize-(file.filePos-cursor)))
-	multi = io.MultiReader(readers...)
-
-	writeSegment := func() (finished bool, bytesRead int64, err error) {
-		segment := getSegment(file.prefix, partNumber)
-
-		currentSegment, err := file.conn.ObjectCreate(file.segmentContainer, segment, false, "", file.contentType, nil)
-		if err != nil {
-			return false, bytesRead, err
-		}
-
-		n, err := io.Copy(currentSegment, multi)
-		if err != nil {
-			return false, bytesRead, err
-		}
-
-		if n > 0 {
-			defer currentSegment.Close()
-			bytesRead += n - max(0, file.filePos-cursor)
-		}
-
-		if n < chunkSize {
-			// We wrote all the data
-			end := currentLength
-			if partNumber-1 < len(file.segments) {
-				end = cursor + file.segments[partNumber-1].Bytes
-			}
-			if cursor+n < end {
-				// Copy the end of the chunk
-				headers := make(Headers)
-				headers["Range"] = "bytes=" + strconv.FormatInt(cursor+n, 10) + "-" + strconv.FormatInt(end-1, 10)
-				f, _, err := file.conn.ObjectOpen(file.container, file.objectName, false, headers)
-				if err != nil {
-					return false, bytesRead, err
-				}
-
-				_, copyErr := io.Copy(currentSegment, f)
-
-				if err := f.Close(); err != nil {
-					return false, bytesRead, err
-				}
-
-				if copyErr != nil {
-					return false, bytesRead, copyErr
-				}
-			}
-
-			return true, bytesRead, nil
-		}
-
-		multi = io.LimitReader(reader, chunkSize)
-		cursor += chunkSize
-		partNumber++
-
-		return false, bytesRead, nil
-	}
-
-	finished := false
-	read := int64(0)
-	bytesRead := int64(0)
-	startPos := file.filePos
-	for finished == false {
-		finished, read, err = writeSegment()
-		bytesRead += read
-		file.filePos += read
-		if err != nil {
-			return bytesRead, err
-		}
-	}
-	file.currentLength = max(startPos+bytesRead, currentLength)
-
-	return bytesRead, nil
-}
-
 // Close satisfies the io.Closer interface
 func (file *DynamicLargeObjectCreateFile) Close() error {
 	return file.Flush()