feat: add multistream http download

* Add a feature to split http download into multiple streams
* Downnloads over libp2p are always done in one stream only
* Add NChunks configuration parameter
* Fix tests to handle not just open-ended range requests

Author: ischasny

node/config/def.go

@@ -213,6 +213,9 @@ func DefaultBoost() *Boost {
 				Port:           3104,
 			},
 		},
+		HttpDownload: HttpDownloadConfig{
+			NChunks: 5,
+		},
 	}
 	return cfg
 }

node/config/doc_gen.go

@@ -47,6 +47,7 @@ type Boost struct {
 	Tracing             TracingConfig
 	LocalIndexDirectory LocalIndexDirectoryConfig
 	ContractDeals       ContractDealsConfig
+	HttpDownload        HttpDownloadConfig
 
 	// Lotus configs
 	LotusDealmaking lotus_config.DealmakingConfig
@@ -421,3 +422,10 @@ type LocalIndexDirectoryConfig struct {
 type LocalIndexDirectoryLeveldbConfig struct {
 	Enabled bool
 }
+
+type HttpDownloadConfig struct {
+	// NChunks is a number of chunks to split HTTP downloads into. Each chunk is downloaded in the goroutine of its own
+	// which improves the overall download speed. NChunks is always equal to 1 for libp2p transport because libp2p server
+	// doesn't support range requests yet. NChunks must be greater than 0 and less than 16, with the default of 5.
+	NChunks int
+}

node/config/types.go

@@ -622,7 +622,7 @@ func NewStorageMarketProvider(provAddr address.Address, cfg *config.Boost) func(
 			SealingPipelineCacheTimeout: time.Duration(cfg.Dealmaking.SealingPipelineCacheTimeout),
 		}
 		dl := logs.NewDealLogger(logsDB)
-		tspt := httptransport.New(h, dl)
+		tspt := httptransport.New(h, dl, httptransport.NChunksOpt(cfg.HttpDownload.NChunks))
 		prov, err := storagemarket.NewProvider(prvCfg, sqldb, dealsDB, fundMgr, storageMgr, a, dp, provAddr, secb, commpc,
 			sps, cdm, df, logsSqlDB.db, logsDB, piecedirectory, ip, lp, &signatureVerifier{a}, dl, tspt)
 		if err != nil {

node/modules/storageminer.go

@@ -148,8 +148,8 @@ func HttpTestDisconnectingServer(t *testing.T, dir string, afterEvery int64) *ht
 	svr := httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		// process the start offset
 		offset := r.Header.Get("Range")
-		finalOffset := strings.TrimSuffix(strings.TrimPrefix(offset, "bytes="), "-")
-		start, _ := strconv.ParseInt(finalOffset, 10, 64)
+		startend := strings.Split(strings.TrimPrefix(offset, "bytes="), "-")
+		start, _ := strconv.ParseInt(startend[0], 10, 64)
 		// only send `afterEvery` bytes and then disconnect
 		end := start + afterEvery
 

testutil/httptestfileservers.go

@@ -19,6 +19,7 @@ import (
 	"github.com/jpillora/backoff"
 	p2phttp "github.com/libp2p/go-libp2p-http"
 	"github.com/libp2p/go-libp2p/core/host"
+	"golang.org/x/sync/errgroup"
 )
 
 const (
@@ -30,6 +31,13 @@ const (
 	maxBackOff           = 10 * time.Minute
 	factor               = 1.5
 	maxReconnectAttempts = 15
+
+	nChunks = 5
+	// minChunkBytes is a minimum size of a chunk. If splitting download into nChunks results into a chunk being smaller than minChunkBytes
+	// then the total number of chunks will be reduced. For example if nChunks is 5, number of remaining bytes is 100 and minChunkBytes is 50
+	// then download will be performed in 2 parallel streams.
+	// 1 Mib
+	minChunkBytes = 1048576
 )
 
 type httpError struct {
@@ -50,6 +58,15 @@ func BackOffRetryOpt(minBackoff, maxBackoff time.Duration, factor, maxReconnectA
 	}
 }
 
+func NChunksOpt(nChunks int) Option {
+	return func(h *httpTransport) {
+		if nChunks < 1 || nChunks > 16 {
+			return
+		}
+		h.nChunks = nChunks
+	}
+}
+
 type httpTransport struct {
 	libp2pHost   host.Host
 	libp2pClient *http.Client
@@ -59,6 +76,8 @@ type httpTransport struct {
 	backOffFactor        float64
 	maxReconnectAttempts float64
 
+	nChunks int
+
 	dl *logs.DealLogger
 }
 
@@ -69,6 +88,7 @@ func New(host host.Host, dealLogger *logs.DealLogger, opts ...Option) *httpTrans
 		maxBackoffWait:       maxBackOff,
 		backOffFactor:        factor,
 		maxReconnectAttempts: maxReconnectAttempts,
+		nChunks:              nChunks,
 		dl:                   dealLogger.Subsystem("http-transport"),
 	}
 	for _, o := range opts {
@@ -120,6 +140,12 @@ func (h *httpTransport) Execute(ctx context.Context, transportInfo []byte, dealI
 	}
 	h.dl.Infow(duuid, "existing file size", "file size", fileSize, "deal size", dealInfo.DealSize)
 
+	// default to a single stream for libp2p urls as libp2p server doesn't support range requests
+	nChunks := h.nChunks
+	if u.Scheme == "libp2p" {
+		nChunks = 1
+	}
+
 	// construct the transfer instance that will act as the transfer handler
 	tctx, cancel := context.WithCancel(ctx)
 	t := &transfer{
@@ -136,6 +162,7 @@ func (h *httpTransport) Execute(ctx context.Context, transportInfo []byte, dealI
 		},
 		maxReconnectAttempts: h.maxReconnectAttempts,
 		dl:                   h.dl,
+		nChunks:              nChunks,
 	}
 
 	cleanupFns := []func(){
@@ -216,16 +243,13 @@ type transfer struct {
 
 	client *http.Client
 	dl     *logs.DealLogger
+
+	nChunks int
 }
 
 func (t *transfer) execute(ctx context.Context) error {
 	duuid := t.dealInfo.DealUuid
 	for {
-		// construct request
-		req, err := http.NewRequest("GET", t.tInfo.URL, nil)
-		if err != nil {
-			return fmt.Errorf("failed to create http req: %w", err)
-		}
 
 		// get the number of bytes already received (the size of the output file)
 		st, err := os.Stat(t.dealInfo.OutputFile)
@@ -234,15 +258,6 @@ func (t *transfer) execute(ctx context.Context) error {
 		}
 		t.nBytesReceived = st.Size()
 
-		// add request headers
-		for name, val := range t.tInfo.Headers {
-			req.Header.Set(name, val)
-		}
-
-		// add range req to start reading from the last byte we have in the output file
-		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", t.nBytesReceived))
-		// init the request with the transfer context
-		req = req.WithContext(ctx)
 		// open output file in append-only mode for writing
 		of, err := os.OpenFile(t.dealInfo.OutputFile, os.O_APPEND|os.O_WRONLY, 0644)
 		if err != nil {
@@ -252,7 +267,75 @@ func (t *transfer) execute(ctx context.Context) error {
 
 		// start the http transfer
 		remaining := t.dealInfo.DealSize - t.nBytesReceived
-		reqErr := t.doHttp(ctx, req, of, remaining)
+
+		// split download into chunks
+		// each chunk should be bigger than minChunkBytes
+		nChunks := int(remaining/minChunkBytes) + 1
+		if nChunks > t.nChunks {
+			nChunks = t.nChunks
+		}
+
+		chunkLen := remaining / int64(nChunks)
+		var pch, nch chan bool
+		group := errgroup.Group{}
+
+		nBytesReceived := t.nBytesReceived
+
+		for i := 0; i < nChunks; i++ {
+			isLast := i == nChunks-1
+
+			// construct request
+			req, err := http.NewRequest("GET", t.tInfo.URL, nil)
+			if err != nil {
+				return fmt.Errorf("failed to create http req: %w", err)
+			}
+
+			// add request headers
+			for name, val := range t.tInfo.Headers {
+				req.Header.Set(name, val)
+			}
+
+			chunkStart := nBytesReceived + int64(i)*chunkLen
+			var chunkEnd int64
+			var srange string
+			if isLast {
+				chunkEnd = t.dealInfo.DealSize
+				srange = fmt.Sprintf("bytes=%d-", chunkStart)
+			} else {
+				chunkEnd = chunkStart + chunkLen
+				srange = fmt.Sprintf("bytes=%d-%d", chunkStart, chunkEnd)
+			}
+
+			// add range req to start reading from the last byte we have in the output file
+			req.Header.Set("Range", srange)
+			// init the request with the transfer context
+			req = req.WithContext(ctx)
+
+			if isLast {
+				nch = nil
+			} else {
+				nch = make(chan bool, 1)
+			}
+			cpch := pch
+			cnch := nch
+			group.Go(func() error {
+				err, success := t.doHttp(ctx, req, of, chunkEnd-chunkStart, cpch)
+				// signal to the next download goroutine to either proceed writing into the file if the current download has finished successfully
+				// or to abandon if it has not
+				if cnch != nil {
+					cnch <- (err == nil && success)
+					close(cnch)
+				}
+				return err
+			})
+
+			pch = nch
+		}
+		var reqErr *httpError
+		if err := group.Wait(); err != nil {
+			reqErr = err.(*httpError)
+		}
+
 		if reqErr == nil {
 			t.dl.Infow(duuid, "http transfer completed successfully")
 			// if there's no error, transfer was successful
@@ -326,22 +409,45 @@ func (t *transfer) execute(ctx context.Context) error {
 	return nil
 }
 
-func (t *transfer) doHttp(ctx context.Context, req *http.Request, dst io.Writer, toRead int64) *httpError {
+func (t *transfer) doHttp(ctx context.Context, req *http.Request, dst io.Writer, toRead int64, pch chan bool) (*httpError, bool) {
 	duid := t.dealInfo.DealUuid
 	t.dl.Infow(duid, "sending http request", "received", t.nBytesReceived, "remaining",
 		toRead, "range-rq", req.Header.Get("Range"))
 
 	// send http request and validate response
 	resp, err := t.client.Do(req)
 	if err != nil {
-		return &httpError{error: fmt.Errorf("failed to send  http req: %w", err)}
+		return &httpError{error: fmt.Errorf("failed to send  http req: %w", err)}, false
 	}
 	// we should either get back a 200 or a 206 -> anything else means something has gone wrong and we return an error.
 	defer resp.Body.Close() // nolint
+
+	// do not fail the whole download if one of the goroutines' range couldn't be satisfied
+	// if not enough bytes have ben donwloaded, then a length mismatch error will be raised up the stack
+	if resp.StatusCode == http.StatusRequestedRangeNotSatisfiable {
+		return nil, false
+	}
+
 	if resp.StatusCode != http.StatusOK && resp.StatusCode != http.StatusPartialContent {
 		return &httpError{
 			error: fmt.Errorf("http req failed: code: %d, status: %s", resp.StatusCode, resp.Status),
 			code:  resp.StatusCode,
+		}, false
+	}
+
+	// if previous download goroutine has failed - don't write into the file. Downloads can be performed in parallel while writing
+	// must be done strictly in sequential order
+	if pch != nil {
+		select {
+		case success := <-pch:
+			if !success {
+				return nil, false
+			}
+		case <-ctx.Done():
+			if ctx.Err() != nil {
+				return &httpError{error: ctx.Err()}, false
+			}
+			return nil, false
 		}
 	}
 
@@ -351,7 +457,7 @@ func (t *transfer) doHttp(ctx context.Context, req *http.Request, dst io.Writer,
 	for {
 		if ctx.Err() != nil {
 			t.dl.LogError(duid, "stopped reading http response: context canceled", ctx.Err())
-			return &httpError{error: ctx.Err()}
+			return &httpError{error: ctx.Err()}, false
 		}
 		nr, readErr := limitR.Read(buf)
 
@@ -362,9 +468,9 @@ func (t *transfer) doHttp(ctx context.Context, req *http.Request, dst io.Writer,
 			// if the number of read and written bytes don't match -> something has gone wrong, abort the http req.
 			if nw < 0 || nr != nw {
 				if writeErr != nil {
-					return &httpError{error: fmt.Errorf("failed to write to output file: %w", writeErr)}
+					return &httpError{error: fmt.Errorf("failed to write to output file: %w", writeErr)}, false
 				}
-				return &httpError{error: fmt.Errorf("read-write mismatch writing to the output file, read=%d, written=%d", nr, nw)}
+				return &httpError{error: fmt.Errorf("read-write mismatch writing to the output file, read=%d, written=%d", nr, nw)}, false
 			}
 
 			t.nBytesReceived = t.nBytesReceived + int64(nw)
@@ -375,10 +481,10 @@ func (t *transfer) doHttp(ctx context.Context, req *http.Request, dst io.Writer,
 		// the http stream we're reading from has sent us an EOF, nothing to do here.
 		if readErr == io.EOF {
 			t.dl.Infow(duid, "http server sent EOF", "received", t.nBytesReceived, "deal-size", t.dealInfo.DealSize)
-			return nil
+			return nil, true
 		}
 		if readErr != nil {
-			return &httpError{error: fmt.Errorf("error reading from http response stream: %w", readErr)}
+			return &httpError{error: fmt.Errorf("error reading from http response stream: %w", readErr)}, false
 		}
 	}
 }