diff --git a/ais/target.go b/ais/target.go index dcfff7996e..283bb50d82 100644 --- a/ais/target.go +++ b/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 } diff --git a/cmd/aisinit/config.go b/cmd/aisinit/config.go index 4165b19679..36036bcb37 100644 --- a/cmd/aisinit/config.go +++ b/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, diff --git a/cmn/config.go b/cmn/config.go index 55eae7989c..a41a627ab8 100644 --- a/cmn/config.go +++ b/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 // /////////////////// diff --git a/cmn/tests/config_test.go b/cmn/tests/config_test.go index 9c8eb2a7c9..497a134b89 100644 --- a/cmn/tests/config_test.go +++ b/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) +} diff --git a/xact/xs/blob_download.go b/xact/xs/blob_download.go index 3ff618869e..1d44b623d3 100644 --- a/xact/xs/blob_download.go +++ b/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*/)