Allow multipart downloading when pulling from remote

ais/target.go

@@ -991,7 +991,38 @@ func (t *target) getObject(w http.ResponseWriter, r *http.Request, dpq *dpq, bck
 			nlog.Warningln("GET", lom.Cname(), "via blob-download["+xid+"]:", err)
 			err = nil
 		}
-		return lom, err
+		if xid != "" || err != nil {
+			return lom, err
+		}
+		// xid == "" && err == nil: fall through to regular GET (object already cached
+		// or a concurrent blob download just completed)
+	// server-side multipart (blob) downloading: when enabled, route cold GETs from
+	// remote backends through the built-in blob-downloader (see docs/blob_downloader.md).
+	// Warm GETs (object already cached) fall through to the regular GET path below.
+	// Range reads and archive requests are excluded - the blob-downloader always
+	// fetches the full object.
+	case bck.IsRemote() && r.Header.Get(cos.HdrRange) == "" && !dpq.isArch() && cmn.GCO.Get().Multipart.Enabled:
+		config := cmn.GCO.Get()
+		msg := apc.BlobMsg{
+			ChunkSize:  int64(config.Multipart.PartMaxSize),
+			NumWorkers: config.Multipart.MaxThreads,
+			LatestVer:  _validateWarmGet(lom, dpq.latestVer),
+		}
+		args := &core.BlobParams{
+			RespWriter: w, // NOTE: make a blocking call
+			Lom:        lom,
+			Msg:        &msg,
+			Parent:     "GET",
+		}
+		xid, _, err := t.blobdl(args, nil /*oa*/, w.Header())
+		if err != nil && xid != "" {
+			nlog.Warningln("GET", lom.Cname(), "via blob-download["+xid+"]:", err)
+			err = nil
+		}
+		// xid == "" && err == nil: object is already cached - fall through to regular GET
+		if xid != "" || err != nil {
+			return lom, err
+		}
 	}
 
 	// GET: regular | archive | range
@@ -1982,6 +2013,12 @@ func (t *target) _blobdl(params *core.BlobParams, oa *cmn.ObjAttrs) (string, *xs
 	xid := cos.GenUUID()
 	rns := xs.RenewBlobDl(xid, params, oa)
 	if rns.Err != nil || rns.IsRunning() { // cmn.IsErrXactUsePrev(rns.Err): single blob-downloader per blob
+		// a blob download for this object is already running — wait for it to
+		// complete, then fall through to the regular GET path (object now cached)
+		if cmn.IsErrXactUsePrev(rns.Err) {
+			<-rns.Entry.Get().ChanAbort()
+			return "", nil, nil
+		}
 		return "", nil, rns.Err
 	}
 

cmd/aisinit/config.go

@@ -79,6 +79,13 @@ var (
 		Timeout: cos.Duration(time.Hour),
 	}
 
+	defaultMultipart = aiscmn.MultipartConf{
+		PartMaxSize: cos.SizeIEC(256 * cos.MiB),
+		Threshold:   cos.SizeIEC(512 * cos.MiB),
+		MaxThreads:  16,
+		Enabled:     false,
+	}
+
 	defaultEC = aiscmn.ECConf{
 		XactConf:     defaultXconf,
 		Enabled:      false,
@@ -230,6 +237,7 @@ func newDefaultConfig() *aiscmn.ClusterConfig {
 		Net:         defaultNet,
 		FSHC:        defaultFSHC,
 		Downloader:  defaultDownloader,
+		Multipart:   defaultMultipart,
 		EC:          defaultEC,
 		Chunks:      defaultChunks,
 		Keepalive:   defaultKeepalive,

cmn/config.go

@@ -132,6 +132,7 @@ type (
 		Periodic    PeriodConf      `json:"periodic" allow:"cluster"`
 		Client      ClientConf      `json:"client"`
 		Downloader  DownloaderConf  `json:"downloader"`
+		Multipart   MultipartConf   `json:"multipart"`
 		Features    feat.Flags      `json:"features,string" allow:"cluster"` // to flip assorted global defaults (see cmn/feat/feat and docs/feat*)
 		Version     int64           `json:"config_version,string"`
 		Versioning  VersionConf     `json:"versioning" allow:"cluster"`
@@ -161,6 +162,7 @@ type (
 		Auth        *AuthConfToSet        `json:"auth,omitempty"`
 		Keepalive   *KeepaliveConfToSet   `json:"keepalivetracker,omitempty"`
 		Downloader  *DownloaderConfToSet  `json:"downloader,omitempty"`
+		Multipart   *MultipartConfToSet   `json:"multipart,omitempty"`
 		Dsort       *DsortConfToSet       `json:"distributed_sort,omitempty"`
 		Transport   *TransportConfToSet   `json:"transport,omitempty"`
 		Memsys      *MemsysConfToSet      `json:"memsys,omitempty"`
@@ -864,6 +866,30 @@ type (
 		Timeout *cos.Duration `json:"timeout,omitempty"`
 	}
 
+	// MultipartConf configures server-side multipart (a.k.a. blob) downloading
+	// of large objects from remote backends. When enabled, GET requests for remote
+	// buckets are routed through the built-in blob-downloader (see apc.HdrBlobDownload),
+	// which concurrently fetches object data in chunks.
+	// The fields mirror the multipart-download tuning in the OCI backend implementation.
+	MultipartConf struct {
+		// PartMaxSize is the maximum size of a single downloaded part (a.k.a. chunk).
+		// It is passed to the blob-downloader as the chunk size.
+		PartMaxSize cos.SizeIEC `json:"part_max_size"`
+		// Threshold is the object-size threshold above which multipart downloading is used.
+		// Objects smaller than this value are downloaded via the regular (single-stream) path.
+		Threshold cos.SizeIEC `json:"threshold"`
+		// MaxThreads is the maximum number of concurrent download workers.
+		MaxThreads int `json:"max_threads"`
+		// Enabled toggles server-side multipart (blob) downloading.
+		Enabled bool `json:"enabled"`
+	}
+	MultipartConfToSet struct {
+		PartMaxSize *cos.SizeIEC `json:"part_max_size,omitempty"` // +gen:optional
+		Threshold   *cos.SizeIEC `json:"threshold,omitempty"`     // +gen:optional
+		MaxThreads  *int         `json:"max_threads,omitempty"`   // +gen:optional
+		Enabled     *bool        `json:"enabled,omitempty"`       // +gen:optional
+	}
+
 	DsortConf struct {
 		DuplicatedRecords   string `json:"duplicated_records"`
 		MissingShards       string `json:"missing_shards"` // cmn.SupportedReactions enum
@@ -1174,6 +1200,7 @@ var (
 	_ validator = (*AuthConf)(nil)
 	_ validator = (*HTTPConf)(nil)
 	_ validator = (*DownloaderConf)(nil)
+	_ validator = (*MultipartConf)(nil)
 	_ validator = (*TransportConf)(nil)
 	_ validator = (*MemsysConf)(nil)
 	_ validator = (*TCBConf)(nil)
@@ -2853,6 +2880,52 @@ func (c *DownloaderConf) Validate() error {
 	return nil
 }
 
+//////////////////
+// MultipartConf //
+//////////////////
+
+const (
+	mpdPartMaxSizeMin     = 4 * cos.KiB
+	mpdPartMaxSizeMax     = 5 * cos.GiB
+	mpdPartMaxSizeDefault = 256 * cos.MiB
+	mpdThresholdMin       = 4 * cos.KiB
+	mpdThresholdMax       = 5 * cos.GiB
+	mpdThresholdDefault   = 512 * cos.MiB
+	mpdMaxThreadsMin      = 1
+	mpdMaxThreadsMax      = 64
+	mpdMaxThreadsDefault  = 16
+)
+
+func (c *MultipartConf) Validate() error {
+	if c.PartMaxSize == 0 {
+		c.PartMaxSize = cos.SizeIEC(mpdPartMaxSizeDefault)
+	} else if c.PartMaxSize < mpdPartMaxSizeMin || c.PartMaxSize > mpdPartMaxSizeMax {
+		return fmt.Errorf("invalid multipart.part_max_size=%s (expected range [%s, %s])",
+			c.PartMaxSize, cos.IEC(mpdPartMaxSizeMin, 0), cos.IEC(mpdPartMaxSizeMax, 0))
+	}
+	if c.Threshold == 0 {
+		c.Threshold = cos.SizeIEC(mpdThresholdDefault)
+	} else if c.Threshold < mpdThresholdMin || c.Threshold > mpdThresholdMax {
+		return fmt.Errorf("invalid multipart.threshold=%s (expected range [%s, %s])",
+			c.Threshold, cos.IEC(mpdThresholdMin, 0), cos.IEC(mpdThresholdMax, 0))
+	}
+	if c.MaxThreads == 0 {
+		c.MaxThreads = mpdMaxThreadsDefault
+	} else if c.MaxThreads < mpdMaxThreadsMin || c.MaxThreads > mpdMaxThreadsMax {
+		return fmt.Errorf("invalid multipart.max_threads=%d (expected range [%d, %d])",
+			c.MaxThreads, mpdMaxThreadsMin, mpdMaxThreadsMax)
+	}
+	return nil
+}
+
+func (c *MultipartConf) String() string {
+	if !c.Enabled {
+		return confDisabled
+	}
+	return fmt.Sprintf("part_max_size=%s, threshold=%s, max_threads=%d",
+		c.PartMaxSize, c.Threshold, c.MaxThreads)
+}
+
 ///////////////////
 // RebalanceConf //
 ///////////////////

cmn/tests/config_test.go

@@ -545,3 +545,133 @@ func TestHTTPConfJSONFlat(t *testing.T) {
 	b, _ := jsoniter.Marshal(c)
 	tassert.Fatalf(t, !strings.Contains(string(b), `"tls"`), "unexpected nesting: %s", b)
 }
+
+func TestMultipartConfValidate(t *testing.T) {
+	tests := []struct {
+		name    string
+		in      cmn.MultipartConf
+		wantErr bool
+		// expected normalized values (only checked when wantErr is false)
+		wantSeg  cos.SizeIEC
+		wantThr  cos.SizeIEC
+		wantThrN int
+	}{
+		{
+			name:     "all zero backfilled with defaults",
+			in:       cmn.MultipartConf{},
+			wantSeg:  cos.SizeIEC(256 * cos.MiB),
+			wantThr:  cos.SizeIEC(512 * cos.MiB),
+			wantThrN: 16,
+		},
+		{
+			name: "enabled with explicit valid values preserved",
+			in: cmn.MultipartConf{
+				Enabled:     true,
+				PartMaxSize: cos.SizeIEC(64 * cos.MiB),
+				Threshold:   cos.SizeIEC(128 * cos.MiB),
+				MaxThreads:  8,
+			},
+			wantSeg:  cos.SizeIEC(64 * cos.MiB),
+			wantThr:  cos.SizeIEC(128 * cos.MiB),
+			wantThrN: 8,
+		},
+		{
+			name: "part_max_size below min rejected",
+			in: cmn.MultipartConf{
+				PartMaxSize: cos.SizeIEC(2 * cos.KiB),
+			},
+			wantErr: true,
+		},
+		{
+			name: "part_max_size above max rejected",
+			in: cmn.MultipartConf{
+				PartMaxSize: cos.SizeIEC(6 * cos.GiB),
+			},
+			wantErr: true,
+		},
+		{
+			name: "threshold below min rejected",
+			in: cmn.MultipartConf{
+				Threshold: cos.SizeIEC(2 * cos.KiB),
+			},
+			wantErr: true,
+		},
+		{
+			name: "threshold above max rejected",
+			in: cmn.MultipartConf{
+				Threshold: cos.SizeIEC(6 * cos.GiB),
+			},
+			wantErr: true,
+		},
+		{
+			name: "max_threads zero backfilled with default",
+			in: cmn.MultipartConf{
+				MaxThreads: 0,
+			},
+			wantSeg:  cos.SizeIEC(256 * cos.MiB),
+			wantThr:  cos.SizeIEC(512 * cos.MiB),
+			wantThrN: 16,
+		},
+		{
+			name: "max_threads above max rejected",
+			in: cmn.MultipartConf{
+				MaxThreads: 65,
+			},
+			wantErr: true,
+		},
+		{
+			name: "max_threads negative rejected",
+			in: cmn.MultipartConf{
+				MaxThreads: -1,
+			},
+			wantErr: true,
+		},
+		{
+			name: "boundary values accepted",
+			in: cmn.MultipartConf{
+				PartMaxSize: cos.SizeIEC(4 * cos.KiB),
+				Threshold:   cos.SizeIEC(5 * cos.GiB),
+				MaxThreads:  64,
+			},
+			wantSeg:  cos.SizeIEC(4 * cos.KiB),
+			wantThr:  cos.SizeIEC(5 * cos.GiB),
+			wantThrN: 64,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			c := tt.in
+			err := c.Validate()
+			if tt.wantErr {
+				tassert.Fatalf(t, err != nil, "expected error, got nil; result=%+v", c)
+				return
+			}
+			tassert.CheckFatal(t, err)
+			tassert.Fatalf(t, c.PartMaxSize == tt.wantSeg,
+				"part_max_size: want %d, got %d", tt.wantSeg, c.PartMaxSize)
+			tassert.Fatalf(t, c.Threshold == tt.wantThr,
+				"threshold: want %d, got %d", tt.wantThr, c.Threshold)
+			tassert.Fatalf(t, c.MaxThreads == tt.wantThrN,
+				"max_threads: want %d, got %d", tt.wantThrN, c.MaxThreads)
+		})
+	}
+}
+
+func TestMultipartConfString(t *testing.T) {
+	// disabled
+	disabled := cmn.MultipartConf{Enabled: false}
+	tassert.Fatalf(t, disabled.String() == "Disabled",
+		"expected Disabled, got %q", disabled.String())
+
+	// enabled
+	enabled := cmn.MultipartConf{
+		Enabled:     true,
+		PartMaxSize: cos.SizeIEC(256 * cos.MiB),
+		Threshold:   cos.SizeIEC(512 * cos.MiB),
+		MaxThreads:  16,
+	}
+	s := enabled.String()
+	tassert.Fatalf(t, strings.Contains(s, "256MiB"), "expected part_max_size in string, got %q", s)
+	tassert.Fatalf(t, strings.Contains(s, "512MiB"), "expected threshold in string, got %q", s)
+	tassert.Fatalf(t, strings.Contains(s, "max_threads=16"), "expected max_threads in string, got %q", s)
+}

xact/xs/blob_download.go

@@ -251,6 +251,11 @@ func (p *blobFactory) Start() (err error) {
 
 	r.setChunkSize()
 
+	// single-part: run inline in the request goroutine (no worker goroutines)
+	if r.fullSize <= r.chunkSize {
+		r.numWorkers = xact.NwpNone
+	}
+
 	// Generate uploadID and initialize manifest
 	r.uploadID = cos.GenUUID()
 	r.manifest, err = core.NewUfest(r.uploadID, lom, false /*must-exist*/)