Repository: peak/s5cmd (go)

Issue:
's5cmd cat' sub command is not using concurrent connections
's5cmd cat' sub command is not using concurrent connections, like 's5cmd cp' does.

My use-case is downloading a 427 GB tarball from S3 and extracting it on the fly:
`time s5cmd cat s3://bucket/file.tar.zst - | pzstd -d | tar -xv -C /`

Example EC2 instance type: c5d.9xlarge with 36 CPU cores, 72 GB RAM, 900 GB local SSD

When just comparing the download part with aws cli:
```
# time aws s3 cp s3://bucket/file.tar.zst - | cat >/dev/null
real    37m56.415s
user    22m50.195s
sys     19m8.677s
(around 192 MB/s)
```

With 's5cmd cat':
```
# time s5cmd cat s3://bucket/file.tar.zst >/dev/null
Still running. Only around 85 MB/s on a single S3 connection, according to netstat.
```

With 's5cmd cp' and writing to disk (without decompression):
```
time s5cmd cp s3://bucket/file.tar.zst /file.tar.zst
real    23m58.230s
user    7m56.734s
sys     22m40.482s
(around 304 MB/s)
```

With higher concurrency and larger parts:
```
# time s5cmd cp -c 36 -p 600 s3://bucket/file.tar.zst /file.tar.zst
real    10m3.064s
user    6m53.378s
sys     41m30.392s
(around 729 MB/s)


Pull request:
cat: make it concurrent 
This PR adds a new `io.WriterAt` adapter for non-seekable writers. It uses an internal linked list to order the incoming chunks. The implementation is independent from the [download manager](https://github.com/aws/aws-sdk-go/blob/main/service/s3/s3manager/download.go) of aws-sdk-go, and because of that currently it **can not bound** the memory usage. In order to limit the memory usage, we would have had to write a custom manager other than the aws-sdk-go's implementation, which seemed unfeasible. 

The new implementation is about %25 percent faster than the older implementation for a 9.4 GB file with `partSize=50MB` and `concurrency=20` parameters, with significantly higher memory usage, on average it uses 0.9 `GB` of memory and at most 2.1 `GB` is observed. Obviously, the memory usage and performance is dependent on the `partSize`-`concurrency` configuration and the link.

Resolves #245.

Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index ff7326c..20e8e05 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -6,9 +6,10 @@
 
 #### Features
 - Added `--content-disposition` flag to `cp` command. ([#569](https://github.com/peak/s5cmd/issues/569))
-- Added `--show-fullpath` flag to `ls`. (#[596](https://github.com/peak/s5cmd/issues/596))
+- Added `--show-fullpath` flag to `ls`. ([#596](https://github.com/peak/s5cmd/issues/596))
 
 #### Improvements
+- Implemented concurrent multipart download support for `cat`. ([#245](https://github.com/peak/s5cmd/issues/245)) 
 
 #### Bugfixes
 - Fixed a bug introduced with `external sort` support in `sync` command which prevents `sync` to an empty destination with `--delete` option. ([#576](https://github.com/peak/s5cmd/issues/576))
diff --git a/command/cat.go b/command/cat.go
index 931f3f2..d16a116 100644
--- a/command/cat.go
+++ b/command/cat.go
@@ -3,12 +3,12 @@ package command
 import (
 	"context"
 	"fmt"
-	"io"
 	"os"
 
 	"github.com/urfave/cli/v2"
 
 	"github.com/peak/s5cmd/v2/log/stat"
+	"github.com/peak/s5cmd/v2/orderedwriter"
 	"github.com/peak/s5cmd/v2/storage"
 	"github.com/peak/s5cmd/v2/storage/url"
 )
@@ -44,6 +44,18 @@ func NewCatCommand() *cli.Command {
 				Name:  "version-id",
 				Usage: "use the specified version of an object",
 			},
+			&cli.IntFlag{
+				Name:    "concurrency",
+				Aliases: []string{"c"},
+				Value:   defaultCopyConcurrency,
+				Usage:   "number of concurrent parts transferred between host and remote server",
+			},
+			&cli.IntFlag{
+				Name:    "part-size",
+				Aliases: []string{"p"},
+				Value:   defaultPartSize,
+				Usage:   "size of each part transferred between host and remote server, in MiB",
+			},
 		},
 		CustomHelpTemplate: catHelpTemplate,
 		Before: func(c *cli.Context) error {
@@ -72,6 +84,8 @@ func NewCatCommand() *cli.Command {
 				fullCommand: fullCommand,
 
 				storageOpts: NewStorageOpts(c),
+				concurrency: c.Int("concurrency"),
+				partSize:    c.Int64("part-size") * megabytes,
 			}.Run(c.Context)
 		},
 	}
@@ -86,6 +100,8 @@ type Cat struct {
 	fullCommand string
 
 	storageOpts storage.Options
+	concurrency int
+	partSize    int64
 }
 
 // Run prints content of given source to standard output.
@@ -95,20 +111,17 @@ func (c Cat) Run(ctx context.Context) error {
 		printError(c.fullCommand, c.op, err)
 		return err
 	}
-
-	rc, err := client.Read(ctx, c.src)
+	_, err = client.Stat(ctx, c.src)
 	if err != nil {
 		printError(c.fullCommand, c.op, err)
 		return err
 	}
-	defer rc.Close()
-
-	_, err = io.Copy(os.Stdout, rc)
+	buf := orderedwriter.New(os.Stdout)
+	_, err = client.Get(ctx, c.src, buf, c.concurrency, c.partSize)
 	if err != nil {
 		printError(c.fullCommand, c.op, err)
 		return err
 	}
-
 	return nil
 }
 
diff --git a/orderedwriter/orderedwriter.go b/orderedwriter/orderedwriter.go
new file mode 100644
index 0000000..48336d4
--- /dev/null
+++ b/orderedwriter/orderedwriter.go
@@ -0,0 +1,113 @@
+// Package orderedwriter implements an unbounded buffer for ordering concurrent writes for
+// non-seekable writers. It keeps an internal linked list that keeps the chunks in order
+// and flushes buffered chunks when the expected offset is available.
+package orderedwriter
+
+import (
+	"container/list"
+	"io"
+	"sync"
+)
+
+type chunk struct {
+	offset int64
+	value  []byte
+}
+
+type OrderedWriterAt struct {
+	mu      *sync.Mutex
+	list    *list.List
+	w       io.Writer
+	written int64
+}
+
+func New(w io.Writer) *OrderedWriterAt {
+	return &OrderedWriterAt{
+		mu:      &sync.Mutex{},
+		list:    list.New(),
+		w:       w,
+		written: 0,
+	}
+}
+
+func (w *OrderedWriterAt) WriteAt(p []byte, offset int64) (int, error) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	// If the queue is empty and the chunk is writeable, push it without queueing.
+	if w.list.Front() == nil && w.written == offset {
+		n, err := w.w.Write(p)
+		if err != nil {
+			return n, err
+		}
+		w.written += int64(n)
+		return len(p), nil
+	}
+
+	// Copy the chunk, buffered writers can modify
+	// the slice before we consume them.
+	b := make([]byte, len(p))
+	copy(b, p)
+
+	// If there are no items in the list and we can't write
+	// directly push back and return early.
+	if w.list.Front() == nil {
+		w.list.PushBack(&chunk{
+			offset: offset,
+			value:  b,
+		})
+		return len(p), nil
+	}
+
+	// Traverse the list from the beginning and insert
+	// it to the smallest index possible. That is,
+	// compare the element's offset with the offset
+	// that you want to buffer.
+	var inserted bool
+	for e := w.list.Front(); e != nil; e = e.Next() {
+		v, _ := e.Value.(*chunk)
+		if offset < v.offset {
+			w.list.InsertBefore(&chunk{
+				offset: offset,
+				value:  b,
+			}, e)
+			inserted = true
+			break
+		}
+	}
+
+	// If the chunk haven't been inserted, put it at
+	// the end of the buffer.
+	if !inserted {
+		w.list.PushBack(&chunk{
+			offset: offset,
+			value:  b,
+		})
+	}
+
+	// If the expected offset is buffered,
+	// flush the items that you can.
+	var removeList []*list.Element
+	for e := w.list.Front(); e != nil; e = e.Next() {
+		v, _ := e.Value.(*chunk)
+		if v.offset != w.written {
+			break
+		}
+
+		n, err := w.w.Write(v.value)
+		if err != nil {
+			return n, err
+		}
+
+		removeList = append(removeList, e)
+		w.written += int64(n)
+
+	}
+
+	// Remove the items that have been written.
+	for _, e := range removeList {
+		w.list.Remove(e)
+	}
+
+	return len(p), nil
+}


Test patch:
diff --git a/e2e/cat_test.go b/e2e/cat_test.go
index 6c4c409..df29d57 100644
--- a/e2e/cat_test.go
+++ b/e2e/cat_test.go
@@ -10,13 +10,18 @@ import (
 	"gotest.tools/v3/icmd"
 )
 
+const (
+	kb int64 = 1024
+	mb       = kb * kb
+)
+
 func TestCatS3Object(t *testing.T) {
 	t.Parallel()
 
 	const (
 		filename = "file.txt"
 	)
-	contents, expected := getSequentialFileContent()
+	contents, expected := getSequentialFileContent(4 * mb)
 
 	testcases := []struct {
 		name      string
@@ -42,6 +47,31 @@ func TestCatS3Object(t *testing.T) {
 				jsonCheck(true),
 			},
 		},
+		{
+			name: "cat remote object with lower part size and higher concurrency",
+			cmd: []string{
+				"cat",
+				"-p",
+				"1",
+				"-c",
+				"2",
+			},
+			expected: expected,
+		},
+		{
+			name: "cat remote object with json flag lower part size and higher concurrency",
+			cmd: []string{
+				"--json",
+				"cat",
+				"-p",
+				"1",
+				"-c",
+				"2",
+			}, expected: expected,
+			assertOps: []assertOp{
+				jsonCheck(true),
+			},
+		},
 	}
 	for _, tc := range testcases {
 		tc := tc
@@ -86,7 +116,7 @@ func TestCatS3ObjectFail(t *testing.T) {
 				"cat",
 			},
 			expected: map[int]compareFunc{
-				0: match(`ERROR "cat s3://(.*)/prefix/file\.txt": NoSuchKey:`),
+				0: match(`ERROR "cat s3://(.*)/prefix/file\.txt":(.*) not found`),
 			},
 		},
 		{
@@ -97,7 +127,7 @@ func TestCatS3ObjectFail(t *testing.T) {
 				"cat",
 			},
 			expected: map[int]compareFunc{
-				0: match(`{"operation":"cat","command":"cat s3:\/\/(.*)\/prefix\/file\.txt","error":"NoSuchKey:`),
+				0: match(`{"operation":"cat","command":"cat s3:\/\/(.*)\/prefix\/file\.txt","error":"(.*) not found`),
 			},
 			assertOps: []assertOp{
 				jsonCheck(true),
@@ -142,7 +172,6 @@ func TestCatS3ObjectFail(t *testing.T) {
 			cmd := s5cmd(tc.cmd...)
 
 			result := icmd.RunCmd(cmd)
-
 			result.Assert(t, icmd.Expected{ExitCode: 1})
 			assertLines(t, result.Stderr(), tc.expected, tc.assertOps...)
 		})
@@ -200,16 +229,16 @@ func TestCatLocalFileFail(t *testing.T) {
 	}
 }
 
-// getSequentialFileContent creates a string with 64666688 in size (~61.670 MB)
-func getSequentialFileContent() (string, map[int]compareFunc) {
+// getSequentialFileContent creates a string with size bytes in size.
+func getSequentialFileContent(size int64) (string, map[int]compareFunc) {
 	sb := strings.Builder{}
 	expectedLines := make(map[int]compareFunc)
-
-	for i := 0; i < 50000; i++ {
+	totalBytesWritten := int64(0)
+	for i := 0; totalBytesWritten < size; i++ {
 		line := fmt.Sprintf(`{ "line": "%d", "id": "i%d", data: "some event %d" }`, i, i, i)
 		sb.WriteString(line)
 		sb.WriteString("\n")
-
+		totalBytesWritten += int64(len(line))
 		expectedLines[i] = equals(line)
 	}
 
@@ -272,7 +301,6 @@ func TestCatByVersionID(t *testing.T) {
 		cmd = s5cmd("cat", "--version-id", version,
 			fmt.Sprintf("s3://%v/%v", bucket, filename))
 		result = icmd.RunCmd(cmd)
-
 		if diff := cmp.Diff(contents[i], result.Stdout()); diff != "" {
 			t.Errorf("(-want +got):\n%v", diff)
 		}
diff --git a/orderedwriter/orderedwriter_test.go b/orderedwriter/orderedwriter_test.go
new file mode 100644
index 0000000..2c50517
--- /dev/null
+++ b/orderedwriter/orderedwriter_test.go
@@ -0,0 +1,319 @@
+package orderedwriter
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"math/rand"
+	"sync"
+	"testing"
+
+	"gotest.tools/v3/assert"
+)
+
+const (
+	testRuns = 64
+)
+
+func randomBytes(n int) []byte {
+	const alphabet = "abcdefghijklmnopqrstuvwxyz0123456789"
+	b := make([]byte, n)
+	for i := range b {
+		b[i] = alphabet[rand.Intn(len(alphabet))]
+
+	}
+	return b
+}
+
+func TestShuffleWriteWithStaticChunkSize(t *testing.T) {
+	t.Parallel()
+
+	// only for this test
+	const (
+		chunkSize = 5
+		fileSize  = 1000
+	)
+
+	for b := 0; b < testRuns; b++ {
+		b := b
+		t.Run(fmt.Sprintf("Run%d", b), func(t *testing.T) {
+			t.Parallel()
+			var result bytes.Buffer
+
+			// we know the filesize ahead so make it buffered
+			expected := randomBytes(fileSize)
+			chunks := []chunk{}
+
+			// first create chunks number of tasks
+			// Create tasks and put them in the array
+			for i := 0; i < fileSize; i += chunkSize {
+				task := chunk{
+					offset: int64(i),
+					value:  expected[i : i+chunkSize],
+				}
+				chunks = append(chunks, task)
+			}
+
+			rand.Shuffle(len(chunks), func(i, j int) {
+				chunks[i], chunks[j] = chunks[j], chunks[i]
+			})
+
+			buf := New(&result)
+
+			for _, chunk := range chunks {
+				buf.WriteAt(chunk.value, int64(chunk.offset))
+
+			}
+
+			// Ensure all chunks have been written correctly
+			assert.DeepEqual(t, result.Bytes(), expected)
+		})
+	}
+}
+
+func TestShuffleWriteWithRandomChunkSize(t *testing.T) {
+	t.Parallel()
+
+	const (
+		maxFileSize                = 1024 * 100
+		minChunkSize, maxChunkSize = 5, 1000
+	)
+
+	for b := 0; b < testRuns; b++ {
+		b := b
+		t.Run(fmt.Sprintf("TR%d", b), func(t *testing.T) {
+			t.Parallel()
+
+			var (
+				result   bytes.Buffer
+				offset   int64
+				expected []byte
+				chunks   []chunk
+			)
+
+			// generate chunks
+			for i := 0; i <= maxFileSize; {
+				chunkSize := minChunkSize + rand.Intn(maxChunkSize-minChunkSize)
+				bytechunk := randomBytes(chunkSize)
+
+				chunks = append(chunks, chunk{
+					offset: offset,
+					value:  bytechunk,
+				})
+				expected = append(expected, bytechunk...)
+				offset += int64(chunkSize)
+				i += chunkSize
+			}
+
+			rand.Shuffle(len(chunks), func(i, j int) {
+				chunks[i], chunks[j] = chunks[j], chunks[i]
+			})
+
+			buf := New(&result)
+
+			for _, chunk := range chunks {
+				buf.WriteAt(chunk.value, int64(chunk.offset))
+			}
+
+			// Ensure all chunks have been written correctly
+			assert.DeepEqual(t, result.Bytes(), expected)
+		})
+	}
+
+}
+
+func TestShuffleConcurrentWriteWithRandomChunkSize(t *testing.T) {
+	t.Parallel()
+
+	const (
+		maxFileSize                = 1024 * 100
+		minChunkSize, maxChunkSize = 5, 1000
+	)
+
+	for b := 0; b < testRuns; b++ {
+		b := b
+		t.Run(fmt.Sprintf("Run%d", b), func(t *testing.T) {
+			t.Parallel()
+
+			var (
+				result   bytes.Buffer
+				expected []byte
+				chunks   []chunk
+			)
+
+			// generate chunks
+			for i := 0; i <= maxFileSize; {
+				chunkSize := minChunkSize + rand.Intn(maxChunkSize-minChunkSize)
+				bytechunk := randomBytes(chunkSize)
+
+				chunks = append(chunks, chunk{
+					offset: int64(i),
+					value:  bytechunk,
+				})
+
+				expected = append(expected, bytechunk...)
+				i += chunkSize
+			}
+
+			rand.Shuffle(len(chunks), func(i, j int) {
+				chunks[i], chunks[j] = chunks[j], chunks[i]
+			})
+
+			buf := New(&result)
+
+			chunkch := make(chan chunk)
+
+			var wg sync.WaitGroup
+			workerCount := 5 + rand.Intn(20) // 5-25 workers
+			for i := 0; i < workerCount; i++ {
+				wg.Add(1)
+				go func(ch chan chunk) {
+					defer wg.Done()
+					for task := range ch {
+						buf.WriteAt(task.value, int64(task.offset))
+					}
+				}(chunkch)
+			}
+
+			for _, chunk := range chunks {
+				chunkch <- chunk
+			}
+
+			close(chunkch)
+			wg.Wait()
+
+			// Ensure all chunks have been written correctly
+			assert.DeepEqual(t, result.Bytes(), expected)
+		})
+	}
+}
+
+// when you use io.Copy, the slice it gives you changes in time.
+// You should copy the slice when a OrderedWriterAt.WriteAt
+// call occurs and don't count on the given slice's value.
+
+// We will simulate the aws-sdk-go's download manager algorithm.
+// See: https://github.com/aws/aws-sdk-go/blob/main/service/s3/s3manager/download.go
+
+// dlchunk represents a single chunk of data to write by the worker routine.
+// This structure also implements an io.SectionReader style interface for
+// io.WriterAt, effectively making it an io.SectionWriter (which does not
+// exist).
+// NOTE: value is added for testing, it's not present in AWS-SDK-GO
+type dlchunk struct {
+	w      io.WriterAt
+	offset int64
+	size   int64
+	cur    int64
+	value  []byte
+}
+
+// Write wraps io.WriterAt for the dlchunk, writing from the dlchunk's offset
+// position to its end (or EOF).
+//
+// If a range is specified on the dlchunk the size will be ignored when writing.
+// as the total size may not of be known ahead of time.
+func (c *dlchunk) Write(p []byte) (int, error) {
+	if c.cur >= c.size {
+		return 0, io.EOF
+	}
+
+	n, err := c.w.WriteAt(p, c.offset+c.cur)
+	c.cur += int64(n)
+
+	return n, err
+}
+
+func TestBufferWithChangingSlice(t *testing.T) {
+	t.Parallel()
+
+	const (
+		maxFileSize                = 1024 * 1024
+		testRuns                   = 1
+		minChunkSize, maxChunkSize = 5, 100
+	)
+
+	for b := 0; b < testRuns; b++ {
+		b := b
+		t.Run(fmt.Sprintf("Run%d", b), func(t *testing.T) {
+			t.Parallel()
+
+			var (
+				expectedFileSize int64
+				expected         []byte
+				chunks           []dlchunk
+			)
+
+			// generate chunks
+			for i := 0; i <= maxFileSize; {
+				chunkSize := minChunkSize + rand.Intn(maxChunkSize-minChunkSize)
+				bytechunk := randomBytes(chunkSize)
+
+				// we use the expectedFileSize to track the offset
+				chunks = append(chunks, dlchunk{
+					offset: int64(expectedFileSize),
+					size:   int64(chunkSize),
+					value:  bytechunk,
+				})
+
+				expected = append(expected, bytechunk...)
+				expectedFileSize += int64(chunkSize)
+				i += chunkSize
+			}
+
+			rand.Shuffle(len(chunks), func(i, j int) {
+				chunks[i], chunks[j] = chunks[j], chunks[i]
+			})
+
+			result := bytes.NewBuffer(make([]byte, 0, expectedFileSize))
+			buf := New(result)
+
+			// we set the writerAt after we initialize the buf, since
+			// the buf can't be initialized until the expectedFileSize
+			// is known.
+			for i := 0; i < len(chunks); i++ {
+				chunks[i].w = buf
+			}
+
+			var (
+				wg           sync.WaitGroup
+				blockingTask dlchunk
+			)
+
+			chunkch := make(chan dlchunk)
+			workerCount := 5 + rand.Intn(20) // 5-25 workers
+
+			for i := 0; i < workerCount; i++ {
+				wg.Add(1)
+				go func(ch chan dlchunk) {
+					defer wg.Done()
+					for task := range ch {
+						// the test is to block all chunks then
+						// releasing them, the first bytechunk is blocked
+						if task.offset == 0 {
+							blockingTask = task
+							continue
+						}
+
+						r := bytes.NewReader(task.value)
+						io.Copy(&task, r)
+					}
+				}(chunkch)
+			}
+
+			for _, bytechunk := range chunks {
+				chunkch <- bytechunk
+			}
+
+			close(chunkch)
+			wg.Wait()
+
+			// write the blocking bytechunk
+			r := bytes.NewReader(blockingTask.value)
+			io.Copy(&blockingTask, r)
+
+			// Ensure all chunks have been written correctly
+			assert.DeepEqual(t, result.Bytes(), expected)
+		})
+	}
+}