Merge pull request #1609 from Altinity/issue_1607_transfer_throughput

Fix SFTP and GCS transfer throughput under RTT, rate limits and HTTP/1.1

Author: Slach

ChangeLog.md

@@ -7,6 +7,11 @@ BUG FIXES
 - `list` and `tables` text output is now written to stdout as a single write, so a log line on stderr can no longer land in the middle of a backup row when both streams are merged (`docker exec`, `2>&1`), the same fix `json` / `yaml` / `csv` / `tsv` formats got in v2.8.1
 - `restore` / `restore_remote` with `--restore-database-mapping` / `--restore-table-mapping` of tables on object disks (`s3`, `azure_blob_storage`, including `cache` and `encrypted` wrappers) no longer leaves the restored copy sharing object storage keys with the source table, so `DROP TABLE|DATABASE ... SYNC` of the copy no longer deletes the objects of the source (`Code: 499 ... The specified key does not exist (S3_ERROR)` on the source afterwards): the key rewrite was skipped when the disk endpoint key prefix contained the backup name (e.g. backup `regression` on the disk `https://s3/bucket/regression-test/live/`), and the metadata files were linked into `detached` / the table directory before the keys were rewritten, so an interrupted restore left the copy pointing to the source keys; the rewritten key now also carries a hash of the destination table (`<dir>_<backup_name>_<crc32(db.table)>/<object>`), so several copies of one backup restored next to each other never share objects, the source key of the copy is always derived from the backup, so the same local backup can be restored several times, and metadata files are written via temp file + rename, so a repeated restore never modifies the parts of an already restored table through their shared hardlink, fix [#1568](https://github.com/Altinity/clickhouse-backup/issues/1568)
 - `upload --resume` no longer writes an incomplete `manifest.bolt.gz`: files of parts, archives and table metadata which the resumable state marks as already processed were skipped without being recorded in the manifest, so the manifest of a resumed upload listed only the files of the last attempt, overwrote the fuller one and `download` / `restore_remote` fell back to a `Walk` (ListObjectsV2) per skipped part. Skipped files are now recorded from the local backup without any extra remote listing; with `--delete-source` (`create_remote`) the local part files are already removed, so parts skipped on resume stay out of the manifest and still use the `Walk` fallback, fix [#1569](https://github.com/Altinity/clickhouse-backup/issues/1569)
+- SFTP `upload` of compressed archives (`compression_format` other than `none`) now keeps `sftp.concurrency` writes in flight per file, previously the archive stream from the compressor has no known size, so every 32KiB write waited for the previous one (about 8MB/s per file at 4ms RTT), fix [#1607](https://github.com/Altinity/clickhouse-backup/issues/1607)
+- SFTP `download` with `download_max_bytes_per_second` now keeps concurrent reads per file, the rate limiter wrapper hid `WriteTo` of the SFTP file, so every 32KiB read waited for the previous one and the download stayed at about 75% of the limit, fix [#1607](https://github.com/Altinity/clickhouse-backup/issues/1607)
+- GCS ranged download (`gcs.ranged_download_concurrency`) with `download_max_bytes_per_second` now reaches the limit instead of about 80% of it, the limit is applied to each range request on the wire instead of the in-order consumer, which waited idle until the first chunks arrived and never made up the lost time, fix [#1607](https://github.com/Altinity/clickhouse-backup/issues/1607)
+- change default `gcs.ranged_download_min_size` from 256MiB to 64MiB, with `gcs.ranged_download_concurrency` enabled archives of 64-256MiB, typical for big parts, were downloaded by a single stream and stayed the tail of the download, fix [#1607](https://github.com/Altinity/clickhouse-backup/issues/1607)
+- GCS with `force_http: true` or `disable_http2: true` no longer opens a new TCP connection for most requests, the HTTP/1.1 transport kept only 1 (64 with `disable_http2`) idle connection, so a download with many concurrent transfers left closed connections in TIME_WAIT, exhausted the ephemeral port range and failed with `connect: cannot assign requested address`, fix [#1607](https://github.com/Altinity/clickhouse-backup/issues/1607)
 
 # v2.8.1
 NEW FEATURES

ReadMe.md

@@ -449,7 +449,7 @@ gcs:
   # GCS_RANGED_DOWNLOAD_CONCURRENCY, stream one large archive as N concurrent byte-range requests straight into decompression, without a temporary file and compatible with `download_max_bytes_per_second`,
   # one GCS read stream has a fixed ceiling, so a single big part is often the whole tail of a download, 0 or 1 disables, takes precedence over `allow_multipart_download` for archives >= `ranged_download_min_size`, see https://github.com/Altinity/clickhouse-backup/issues/1592
   ranged_download_concurrency: 0
-  ranged_download_min_size: 268435456  # GCS_RANGED_DOWNLOAD_MIN_SIZE, archives smaller than this size in bytes keep the single-stream reader, at least 2 * `ranged_download_chunk_size`
+  ranged_download_min_size: 67108864   # GCS_RANGED_DOWNLOAD_MIN_SIZE, archives smaller than this size in bytes keep the single-stream reader, at least 2 * `ranged_download_chunk_size`
   ranged_download_chunk_size: 33554432 # GCS_RANGED_DOWNLOAD_CHUNK_SIZE, bytes per range request, one chunk is one in-memory buffer
   ranged_download_max_buffers: 64      # GCS_RANGED_DOWNLOAD_MAX_BUFFERS, process-wide limit of chunk buffers in flight for all ranged readers, so extra memory is at most `ranged_download_max_buffers` * `ranged_download_chunk_size` (2GiB by default) however many archives are in flight
   # GCS_OBJECT_LABELS, allow setup metadata for each object during upload, use {macro_name} from system.macros and {backupName} for current backup name

pkg/config/config.go

@@ -1048,7 +1048,7 @@ func DefaultConfig() *Config {
 			DownloadConcurrency:    int(downloadConcurrency + 1),
 			// ranged streaming download is opt-in: ranged_download_concurrency: 8 enables it
 			RangedDownloadConcurrency: 0,
-			RangedDownloadMinSize:     256 * 1024 * 1024,
+			RangedDownloadMinSize:     64 * 1024 * 1024,
 			RangedDownloadChunkSize:   32 * 1024 * 1024,
 			RangedDownloadMaxBuffers:  64,
 		},

pkg/storage/bwlimit/bwlimit.go

@@ -167,12 +167,43 @@ func ReadCloser(ctx context.Context, reader io.ReadCloser, limiter *Limiter) io.
 	if limiter == nil {
 		return reader
 	}
-	return &rateLimitedReadCloser{
+	rc := &rateLimitedReadCloser{
 		Reader: Reader(ctx, reader, limiter),
 		closer: reader,
 	}
+	// keep WriteTo of the wrapped reader, sftp.File issues concurrent reads only from WriteTo,
+	// with Read it sends one request per round trip, see https://github.com/Altinity/clickhouse-backup/issues/1607
+	if wt, ok := reader.(io.WriterTo); ok {
+		return &rateLimitedWriterToReadCloser{rateLimitedReadCloser: rc, writerTo: wt, ctx: ctx, limiter: limiter}
+	}
+	return rc
 }
 
 func (r *rateLimitedReadCloser) Close() error {
 	return r.closer.Close()
 }
+
+type rateLimitedWriterToReadCloser struct {
+	*rateLimitedReadCloser
+	writerTo io.WriterTo
+	ctx      context.Context
+	limiter  *Limiter
+}
+
+// WriteTo paces the writes into w, the wrapped reader stops reading when its writes block
+func (r *rateLimitedWriterToReadCloser) WriteTo(w io.Writer) (int64, error) {
+	return r.writerTo.WriteTo(&rateLimitedWriter{ctx: r.ctx, writer: w, limiter: r.limiter})
+}
+
+type rateLimitedWriter struct {
+	ctx     context.Context
+	writer  io.Writer
+	limiter *Limiter
+}
+
+func (w *rateLimitedWriter) Write(p []byte) (int, error) {
+	if err := w.limiter.Wait(w.ctx, len(p)); err != nil {
+		return 0, err
+	}
+	return w.writer.Write(p)
+}

pkg/storage/bwlimit/bwlimit_test.go

@@ -98,3 +98,61 @@ func TestLimiterIsSharedAcrossReaders(t *testing.T) {
 		t.Fatalf("shared limiter finished too quickly: %s", elapsed)
 	}
 }
+
+// writerToSource records whether io.Copy went through its WriteTo, like sftp.File,
+// which issues concurrent reads only from WriteTo
+type writerToSource struct {
+	*bytes.Reader
+	writeToCalled bool
+}
+
+func (s *writerToSource) WriteTo(w io.Writer) (int64, error) {
+	s.writeToCalled = true
+	return s.Reader.WriteTo(w)
+}
+
+func (s *writerToSource) Close() error { return nil }
+
+func TestReadCloserKeepsWriterToAndLimits(t *testing.T) {
+	src := &writerToSource{Reader: bytes.NewReader(bytes.Repeat([]byte("a"), 200))}
+	wrapped := ReadCloser(context.Background(), src, New(100))
+
+	if _, ok := wrapped.(io.WriterTo); !ok {
+		t.Fatal("ReadCloser() should expose io.WriterTo when the wrapped reader implements it")
+	}
+	var dst bytes.Buffer
+	start := time.Now()
+	n, err := io.Copy(&dst, wrapped)
+	if err != nil || n != 200 {
+		t.Fatalf("io.Copy() = (%d, %v), want (200, nil)", n, err)
+	}
+	if !src.writeToCalled {
+		t.Fatal("io.Copy() should use WriteTo of the wrapped reader")
+	}
+	// 100 bytes initial burst, the other 100 bytes need one second
+	if elapsed := time.Since(start); elapsed < 900*time.Millisecond {
+		t.Fatalf("WriteTo is not rate limited, finished in %s", elapsed)
+	}
+}
+
+func TestReadCloserWithoutWriterTo(t *testing.T) {
+	reader := &closeRecorder{Reader: bytes.NewReader([]byte("abc"))}
+	// io.Reader without WriteTo, an io.CopyBuffer buffer must still be used
+	wrapped := ReadCloser(context.Background(), struct {
+		io.Reader
+		io.Closer
+	}{reader, reader}, New(1024))
+	if _, ok := wrapped.(io.WriterTo); ok {
+		t.Fatal("ReadCloser() should not expose io.WriterTo when the wrapped reader does not implement it")
+	}
+}
+
+func TestWriteToReturnsContextError(t *testing.T) {
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	src := &writerToSource{Reader: bytes.NewReader(bytes.Repeat([]byte("a"), 10))}
+	wrapped := ReadCloser(ctx, src, New(1)).(io.WriterTo)
+	if _, err := wrapped.WriteTo(io.Discard); err == nil {
+		t.Fatal("WriteTo() should return the context error")
+	}
+}

pkg/storage/gcs.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"math"
 	"net"
 	"net/http"
 	"os"
@@ -16,6 +17,7 @@ import (
 	"time"
 
 	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
+	"github.com/Altinity/clickhouse-backup/v2/pkg/storage/bwlimit"
 	pool "github.com/jolestar/go-commons-pool/v2"
 	"github.com/pkg/errors"
 
@@ -168,31 +170,7 @@ func (gcs *GCS) Connect(ctx context.Context) error {
 
 		var httpClient *http.Client
 		if gcs.Config.ForceHttp || gcs.Config.DisableHttp2 {
-			customTransport := &http.Transport{
-				WriteBufferSize: 128 * 1024,
-				Proxy:           http.ProxyFromEnvironment,
-				DialContext: (&net.Dialer{
-					Timeout:   30 * time.Second,
-					KeepAlive: 30 * time.Second,
-				}).DialContext,
-				MaxIdleConns:          1,
-				MaxIdleConnsPerHost:   1,
-				IdleConnTimeout:       90 * time.Second,
-				TLSHandshakeTimeout:   10 * time.Second,
-				ExpectContinueTimeout: 1 * time.Second,
-			}
-			if gcs.Config.DisableHttp2 {
-				// DisableHttp2 is designed for high-concurrency downloads — raise
-				// connection limits so each parallel part gets its own TCP stream.
-				customTransport.MaxIdleConns = 0
-				customTransport.MaxIdleConnsPerHost = 64
-			}
-			// must set ForceAttemptHTTP2 to false so that when a custom TLSClientConfig
-			// is provided Golang does not setup HTTP/2 transport
-			customTransport.ForceAttemptHTTP2 = false
-			customTransport.TLSClientConfig = &tls.Config{
-				NextProtos: []string{"http/1.1"},
-			}
+			customTransport := newGCSHTTP1Transport()
 			// ForceHttp downgrades the request scheme to cleartext http:// via
 			// rewriteTransport (needed only for internal caches like varnish).
 			// DisableHttp2 keeps the original https:// scheme so TLS and
@@ -484,11 +462,38 @@ func (gcs *GCS) rangedDownloadSettings() (workers int, minSize, chunk int64, ok
 	return workers, minSize, chunk, true
 }
 
+// newGCSHTTP1Transport builds the HTTP/1.1 transport for force_http and disable_http2.
+// Idle connections are not limited: every concurrent request above MaxIdleConnsPerHost closed its connection
+// after use, a busy download left them in TIME_WAIT and exhausted the ephemeral ports with
+// `connect: cannot assign requested address`. There are never more idle connections than concurrent requests,
+// which the transfer budgets already bound, and IdleConnTimeout closes them, see https://github.com/Altinity/clickhouse-backup/issues/1607
+func newGCSHTTP1Transport() *http.Transport {
+	return &http.Transport{
+		WriteBufferSize: 128 * 1024,
+		Proxy:           http.ProxyFromEnvironment,
+		DialContext: (&net.Dialer{
+			Timeout:   30 * time.Second,
+			KeepAlive: 30 * time.Second,
+		}).DialContext,
+		MaxIdleConns:          0,
+		MaxIdleConnsPerHost:   math.MaxInt32,
+		IdleConnTimeout:       90 * time.Second,
+		TLSHandshakeTimeout:   10 * time.Second,
+		ExpectContinueTimeout: 1 * time.Second,
+		// must set ForceAttemptHTTP2 to false so that when a custom TLSClientConfig
+		// is provided Golang does not setup HTTP/2 transport
+		ForceAttemptHTTP2: false,
+		TLSClientConfig: &tls.Config{
+			NextProtos: []string{"http/1.1"},
+		},
+	}
+}
+
 // GetFileRangedReader streams a large object as concurrent byte-range GETs (see rangedReader).
 // ok=false means the caller should use its normal single-stream reader (feature disabled or object too small).
 // Every range borrows its own pooled client for the duration of the request, so with HTTP/2 the ranges
 // do not end up multiplexed on one connection.
-func (gcs *GCS) GetFileRangedReader(ctx context.Context, key string, size int64) (io.ReadCloser, bool, error) {
+func (gcs *GCS) GetFileRangedReader(ctx context.Context, key string, size int64, limiter *bwlimit.Limiter) (io.ReadCloser, bool, error) {
 	workers, minSize, chunk, enabled := gcs.rangedDownloadSettings()
 	if !enabled || size < minSize || gcs.rangedBudget == nil {
 		return nil, false, nil
@@ -529,7 +534,7 @@ func (gcs *GCS) GetFileRangedReader(ctx context.Context, key string, size int64)
 			giveBack(true)
 			return nil, errors.Wrapf(err, "GCS NewRangeReader %s", absKey)
 		}
-		return &releasingReadCloser{ReadCloser: rc, onClose: func() { giveBack(false) }}, nil
+		return bwlimit.ReadCloser(ctx, &releasingReadCloser{ReadCloser: rc, onClose: func() { giveBack(false) }}, limiter), nil
 	}
 	log.Debug().Msgf("GCS ranged read %s size=%d chunk=%d workers=%d", absKey, size, chunk, workers)
 	return newRangedReader(ctx, fetch, size, chunk, workers, 2, gcs.rangedBudget), true, nil

pkg/storage/gcs_test.go

@@ -3,9 +3,15 @@ package storage
 import (
 	"encoding/base64"
 	"fmt"
+	"io"
+	"net"
 	"net/http"
+	"net/http/httptest"
 	"net/url"
+	"sync"
+	"sync/atomic"
 	"testing"
+	"time"
 
 	"github.com/Altinity/clickhouse-backup/v2/pkg/config"
 	"github.com/stretchr/testify/assert"
@@ -222,3 +228,42 @@ func (gcs *GCS) validateAndDecodeEncryptionKey() error {
 	gcs.encryptionKey = key
 	return nil
 }
+
+// TestGCSHTTP1TransportReusesConnections every concurrent request above MaxIdleConnsPerHost closes its
+// connection after use, on a busy download that used up the whole ephemeral port range with
+// `connect: cannot assign requested address`, see https://github.com/Altinity/clickhouse-backup/issues/1607
+func TestGCSHTTP1TransportReusesConnections(t *testing.T) {
+	const workers, requestsPerWorker = 16, 20
+	var newConns atomic.Int64
+	srv := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		time.Sleep(time.Millisecond)
+		_, _ = w.Write([]byte("ok"))
+	}))
+	srv.Config.ConnState = func(_ net.Conn, state http.ConnState) {
+		if state == http.StateNew {
+			newConns.Add(1)
+		}
+	}
+	srv.Start()
+	defer srv.Close()
+
+	client := &http.Client{Transport: newGCSHTTP1Transport()}
+	var wg sync.WaitGroup
+	for range workers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for range requestsPerWorker {
+				resp, err := client.Get(srv.URL)
+				if !assert.NoError(t, err) {
+					return
+				}
+				_, _ = io.Copy(io.Discard, resp.Body)
+				_ = resp.Body.Close()
+			}
+		}()
+	}
+	wg.Wait()
+	t.Logf("%d requests opened %d connections", workers*requestsPerWorker, newConns.Load())
+	assert.LessOrEqual(t, newConns.Load(), int64(2*workers), "concurrent requests should reuse idle connections")
+}

pkg/storage/general.go

@@ -516,10 +516,13 @@ func (bd *BackupDestination) DownloadCompressedStream(ctx context.Context, remot
 	var getReaderErr error
 	rangedUsed := false
 	// Ranged streaming keeps the object in memory chunks consumed in order (no temp file), so unlike the
-	// multipart path below it stays compatible with the rate limiter: the limiter paces the consumer and the
-	// prefetch window is bounded. Use it whenever the storage supports it and the object is large enough.
+	// multipart path below it stays compatible with the rate limiter: the limiter paces the range requests on
+	// the wire, not the in-order consumer. Pacing the consumer instead left it idle until the first chunks
+	// completed and the lost time was never made up (~80% of the cap), see https://github.com/Altinity/clickhouse-backup/issues/1607
+	// Use it whenever the storage supports it and the object is large enough.
+	limiter := bd.DownloadLimiter(maxSpeed)
 	if rp, ok := bd.RemoteStorage.(RangedReaderProvider); ok {
-		reader, rangedUsed, getReaderErr = rp.GetFileRangedReader(ctx, remotePath, remoteFileInfo.Size())
+		reader, rangedUsed, getReaderErr = rp.GetFileRangedReader(ctx, remotePath, remoteFileInfo.Size(), limiter)
 	}
 	if rangedUsed || getReaderErr != nil {
 		// reader (or error) already set by the ranged path
@@ -537,7 +540,9 @@ func (bd *BackupDestination) DownloadCompressedStream(ctx context.Context, remot
 		return 0, errors.Wrap(getReaderErr, "DownloadCompressedStream GetFileReader")
 	}
 	rawReader := reader
-	reader = bwlimit.ReadCloser(ctx, reader, bd.DownloadLimiter(maxSpeed))
+	if !rangedUsed {
+		reader = bwlimit.ReadCloser(ctx, reader, limiter)
+	}
 	var closeReaderOnce sync.Once
 	closeReader := func() {
 		closeReaderOnce.Do(func() {

pkg/storage/ranged_reader_test.go

@@ -14,6 +14,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/Altinity/clickhouse-backup/v2/pkg/storage/bwlimit"
+
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/semaphore"
 )
@@ -192,16 +194,25 @@ type rangedRemote struct {
 	archive []byte
 	used    atomic.Bool
 	budget  *semaphore.Weighted
+	limiter *bwlimit.Limiter
 }
 
 func (m *rangedRemote) StatFile(_ context.Context, _ string) (RemoteFile, error) {
 	return fakeRemoteFile{size: int64(len(m.archive))}, nil
 }
 
-func (m *rangedRemote) GetFileRangedReader(ctx context.Context, _ string, size int64) (io.ReadCloser, bool, error) {
+func (m *rangedRemote) GetFileRangedReader(ctx context.Context, _ string, size int64, limiter *bwlimit.Limiter) (io.ReadCloser, bool, error) {
 	m.used.Store(true)
+	m.limiter = limiter
 	f := &memFetcher{data: m.archive, failChunkAt: -1}
-	return newRangedReader(ctx, f.fetch, size, 64*1024, 4, 0, m.budget), true, nil
+	fetch := func(ctx context.Context, offset, length int64) (io.ReadCloser, error) {
+		rc, err := f.fetch(ctx, offset, length)
+		if err != nil {
+			return nil, err
+		}
+		return bwlimit.ReadCloser(ctx, rc, limiter), nil
+	}
+	return newRangedReader(ctx, fetch, size, 64*1024, 4, 0, m.budget), true, nil
 }
 
 func (m *rangedRemote) GetFileReader(_ context.Context, _ string) (io.ReadCloser, error) {
@@ -232,6 +243,7 @@ func TestDownloadCompressedStreamUsesRangedReaderWithRateLimit(t *testing.T) {
 	_, err = bd.DownloadCompressedStream(context.Background(), "shadow/db/tbl/part.tar", out, 1<<40)
 	require.NoError(t, err)
 	require.True(t, remote.used.Load(), "ranged reader was not used")
+	require.NotNil(t, remote.limiter, "the rate limit must be applied to the range requests, see https://github.com/Altinity/clickhouse-backup/issues/1607")
 	got, err := os.ReadFile(filepath.Join(out, "data.bin"))
 	require.NoError(t, err)
 	require.True(t, bytes.Equal(payload, got))

pkg/storage/sftp.go

@@ -251,7 +251,14 @@ func (sftp *SFTP) PutFileAbsolute(_ context.Context, key string, r io.ReadCloser
 			log.Warn().Msgf("can't close %s err=%v", key, err)
 		}
 	}()
-	if _, err = remoteFile.ReadFrom(r); err != nil {
+	// ReadFrom keeps several writes in flight only when it can learn the size of r,
+	// an io.Pipe from the compressor has none and gets one write per round trip, see https://github.com/Altinity/clickhouse-backup/issues/1607
+	if sftp.Config.Concurrency > 0 {
+		_, err = remoteFile.ReadFromWithConcurrency(r, sftp.Config.Concurrency)
+	} else {
+		_, err = remoteFile.ReadFrom(r)
+	}
+	if err != nil {
 		return errors.Wrap(err, "SFTP PutFileAbsolute ReadFrom")
 	}
 	return nil