xs/xact: initial rechunk xaction implementation

Nahemah1022 · Nahemah1022 · commit 8ee98ea42519 · 2025-10-23T13:05:38.000-07:00
Signed-off-by: Tony Chen &lt;a122774007@gmail.com&gt;
diff --git a/ais/test/rechunk_test.go b/ais/test/rechunk_test.go
diff --git a/ais/tgtimpl.go b/ais/tgtimpl.go
@@ -37,7 +37,7 @@ func (t *target) Health(si *meta.Snode, timeout time.Duration, query url.Values)
 	return t.reqHealth(si, timeout, query, t.owner.smap.get(), false /*retry*/)
 }
 
-func (t *target) PutObject(lom *core.LOM, params *core.PutParams) error {
+func (t *target) PutObject(lom *core.LOM, params *core.PutParams) (err error) {
 	debug.Assert(params.WorkTag != "" && !params.Atime.IsZero())
 	workFQN := lom.GenFQN(fs.WorkCT, params.WorkTag)
 
@@ -54,11 +54,19 @@ func (t *target) PutObject(lom *core.LOM, params *core.PutParams) error {
 		poi.owt = params.OWT
 		poi.skipEC = params.SkipEC
 		poi.coldGET = params.ColdGET
+		poi.locked = params.Locked
 	}
 	if poi.owt != cmn.OwtPut {
 		poi.cksumToUse = params.Cksum
 	}
-	_, err := poi.putObject()
+
+	switch {
+	case params.ChunkSize > 0:
+		_, err = poi.chunk(params.ChunkSize)
+	default:
+		_, err = poi.putObject()
+	}
+
 	freePOI(poi)
 	debug.Func(func() {
 		if err == nil {
diff --git a/ais/tgtmpt.go b/ais/tgtmpt.go
@@ -60,6 +60,7 @@ type (
 		parts    apc.MptCompletedParts
 		isS3     bool
 		coldGET  bool
+		locked   bool // true if the LOM is already locked by the caller
 	}
 )
 
@@ -529,8 +530,8 @@ func (ups *ups) complete(args *completeArgs) (string, int, error) {
 
 	// atomically flip: persist manifest, mark chunked, persist main
 	// NOTE: coldGET implies the LOM's lock has been promoted to wlock
-	debug.Assertf(!args.coldGET || lom.IsLocked() == apc.LockWrite, "expecting wlock, have %d", lom.IsLocked())
-	if err = lom.CompleteUfest(manifest, args.coldGET); err != nil {
+	debug.Assertf(!args.locked || lom.IsLocked() == apc.LockWrite, "expecting wlock, have %d", lom.IsLocked())
+	if err = lom.CompleteUfest(manifest, args.locked); err != nil {
 		nlog.Errorf("upload %q: failed to complete %s locally: %v", uploadID, lom.Cname(), err)
 		return "", http.StatusInternalServerError, err
 	}
diff --git a/ais/tgtobj.go b/ais/tgtobj.go
@@ -68,6 +68,7 @@ type (
 		skipEC     bool          // do not erasure-encode when finalizing
 		skipVC     bool          // skip loading existing Version and skip comparing Checksums (skip VC)
 		coldGET    bool          // (one implication: proceed to write)
+		locked     bool          // true if the LOM is already locked by the caller
 		remoteErr  bool          // to exclude `putRemote` errors when counting soft IO errors
 	}
 
@@ -180,6 +181,7 @@ func (poi *putOI) chunk(chunkSize int64) (ecode int, err error) {
 		lom      = poi.lom
 		uploadID string
 	)
+	debug.Assertf(!poi.coldGET || poi.locked, "expecting locked LOM for cold-GET")
 
 	debug.Assertf(poi.size > 0, "poi.size is required in chunk, object name: %s", poi.lom.Cname())
 	if uploadID, err = poi.t.ups.start(poi.oreq, lom, poi.coldGET); err != nil {
@@ -239,12 +241,14 @@ func (poi *putOI) chunk(chunkSize int64) (ecode int, err error) {
 		parts:    completedParts,
 		isS3:     false,
 		coldGET:  poi.coldGET,
+		locked:   poi.locked,
 	})
 	return ecode, err
 }
 
 func (poi *putOI) putObject() (ecode int, err error) {
 	maxMonoSize := int64(poi.lom.Bprops().Chunks.MaxMonolithicSize)
+	// protect the bucket: if the object size exceeds the max monolithic size, MUST chunk
 	// NOTE: if `poi.size` is not set, don't trigger chunking
 	if maxMonoSize > 0 && poi.size > maxMonoSize {
 		if cmn.Rom.V(5, cos.ModAIS) {
@@ -453,8 +457,9 @@ func (poi *putOI) fini() (ecode int, err error) {
 	// locking strategies: optimistic and otherwise
 	// (see GetCold() implementation and cmn.OWT enum)
 	switch poi.owt {
-	case cmn.OwtGetTryLock, cmn.OwtGetLock, cmn.OwtGet:
+	case cmn.OwtGetTryLock, cmn.OwtGetLock, cmn.OwtGet, cmn.OwtChunks:
 		// do nothing: lom is already wlocked
+		debug.Assertf(lom.IsLocked() == apc.LockWrite, "lom %s is not write-locked", lom.Cname())
 	case cmn.OwtGetPrefetchLock:
 		if !lom.TryLock(true) {
 			nlog.Warningln(poi.loghdr(), "is busy")
@@ -879,6 +884,7 @@ func (goi *getOI) coldPut(res *core.GetReaderResult) (int, error) {
 		poi.owt = cmn.OwtGet
 		poi.cksumToUse = res.ExpCksum // expected checksum (to validate if the bucket's `validate_cold_get == true`)
 		poi.coldGET = true
+		poi.locked = true
 	}
 	code, err := poi.putObject()
 	freePOI(poi)
diff --git a/ais/tgtxact.go b/ais/tgtxact.go
@@ -282,6 +282,20 @@ func (t *target) xstart(args *xact.ArgsMsg, bck *meta.Bck, msg *apc.ActMsg) (xid
 		}
 		go t.runResilver(resargs, wg)
 		wg.Wait()
+	case apc.ActRechunk:
+		if err := xreg.LimitedCoexistence(t.si, bck, args.Kind); err != nil {
+			return "", err
+		}
+		rns := xreg.RenewBckRechunks(bck, args.ID, &xreg.RechunkArgs{
+			ObjSizeLimit: int64(bck.Props.Chunks.ObjSizeLimit),
+			ChunkSize:    int64(bck.Props.Chunks.ChunkSize),
+		})
+		if rns.Err != nil {
+			return "", rns.Err
+		}
+		xctn := rns.Entry.Get()
+		xact.GoRunW(xctn)
+		return xctn.ID(), nil
 	case apc.ActLoadLomCache:
 		rns := xreg.RenewBckLoadLomCache(args.ID, bck)
 		return xid, rns.Err
diff --git a/api/apc/actmsg.go b/api/apc/actmsg.go
@@ -39,6 +39,7 @@ const (
 
 	ActMakeNCopies = "make-n-copies"
 	ActPutCopies   = "put-copies"
+	ActRechunk     = "rechunk"
 
 	ActRebalance = "rebalance"
 	ActMoveBck   = "move-bck"
diff --git a/cmn/owt.go b/cmn/owt.go
@@ -25,6 +25,7 @@ const (
 	OwtArchive              // multi-obj arch
 	OwtTransform            // ETL
 	OwtCopy                 // copy and move objects within cluster
+	OwtChunks               // chunks
 	OwtRebalance            // NOTE: must be the last in PUT* group
 	//
 	// GET and friends
diff --git a/core/target.go b/core/target.go
@@ -26,15 +26,17 @@ import (
 // intra-cluster data path: control structures and types
 type (
 	PutParams struct {
-		Reader  io.ReadCloser
-		Cksum   *cos.Cksum // checksum to check
-		Atime   time.Time
-		Xact    Xact
-		WorkTag string // (=> work fqn)
-		Size    int64
-		OWT     cmn.OWT
-		SkipEC  bool // don't erasure-code when finalizing
-		ColdGET bool // this PUT is in fact a cold-GET
+		Reader    io.ReadCloser
+		Cksum     *cos.Cksum // checksum to check
+		Atime     time.Time
+		Xact      Xact
+		WorkTag   string // (=> work fqn)
+		Size      int64
+		ChunkSize int64 // if set, the object will be chunked with this size regardless of the bucket's chunk properties
+		OWT       cmn.OWT
+		SkipEC    bool // don't erasure-code when finalizing
+		ColdGET   bool // this PUT is in fact a cold-GET
+		Locked    bool // true if the LOM is already locked by the caller
 	}
 	PromoteParams struct {
 		Bck             *meta.Bck   // destination bucket
diff --git a/xact/api.go b/xact/api.go
@@ -145,6 +145,7 @@ var Table = map[string]Descriptor{
 
 	// single target (node)
 	apc.ActResilver: {Scope: ScopeT, Startable: true, Resilver: true},
+	apc.ActRechunk:  {Scope: ScopeB, Startable: true, RefreshCap: true, ConflictRebRes: true},
 
 	// on-demand EC and n-way replication
 	// (non-startable, triggered by PUT => erasure-coded or mirrored bucket)
diff --git a/xact/xreg/bucket.go b/xact/xreg/bucket.go
@@ -81,6 +81,10 @@ func RenewBckLoadLomCache(uuid string, bck *meta.Bck) RenewRes {
 	return RenewBucketXact(apc.ActLoadLomCache, bck, Args{UUID: uuid})
 }
 
+func RenewBckRechunks(bck *meta.Bck, uuid string, args *RechunkArgs) RenewRes {
+	return RenewBucketXact(apc.ActRechunk, bck, Args{Custom: args, UUID: uuid})
+}
+
 func RenewPutMirror(lom *core.LOM) RenewRes {
 	return RenewBucketXact(apc.ActPutCopies, lom.Bck(), Args{Custom: lom})
 }
diff --git a/xact/xreg/custom.go b/xact/xreg/custom.go
@@ -41,6 +41,11 @@ type (
 		Tag    string
 		Copies int
 	}
+	RechunkArgs struct {
+		ObjSizeLimit int64
+		ChunkSize    int64
+		Prefix       string
+	}
 	LsoArgs struct {
 		Msg *apc.LsoMsg
 		Hdr http.Header
diff --git a/xact/xs/init.go b/xact/xs/init.go
@@ -41,4 +41,6 @@ func Treg(coi COI) {
 	xreg.RegBckXact(&tcbFactory{kind: apc.ActETLBck})
 	xreg.RegBckXact(&tcoFactory{streamingF: streamingF{kind: apc.ActETLObjects}})
 	xreg.RegBckXact(&tcoFactory{streamingF: streamingF{kind: apc.ActCopyObjects}})
+
+	xreg.RegBckXact(&rechunkFactory{kind: apc.ActRechunk})
 }
diff --git a/xact/xs/rechunk.go b/xact/xs/rechunk.go
@@ -0,0 +1,198 @@
+// Package xs is a collection of eXtended actions (xactions), including multi-object
+// operations, list-objects, (cluster) rebalance and (target) resilver, ETL, and more.
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved.
+ */
+package xs
+
+import (
+	"fmt"
+	"sync"
+
+	"github.com/NVIDIA/aistore/cmn"
+	"github.com/NVIDIA/aistore/cmn/cos"
+	"github.com/NVIDIA/aistore/cmn/debug"
+	"github.com/NVIDIA/aistore/cmn/nlog"
+	"github.com/NVIDIA/aistore/core"
+	"github.com/NVIDIA/aistore/core/meta"
+	"github.com/NVIDIA/aistore/fs"
+	"github.com/NVIDIA/aistore/fs/mpather"
+	"github.com/NVIDIA/aistore/memsys"
+	"github.com/NVIDIA/aistore/xact"
+	"github.com/NVIDIA/aistore/xact/xreg"
+)
+
+type (
+	rechunkFactory struct {
+		xreg.RenewBase
+		xctn *xactRechunk
+		kind string
+	}
+	xactRechunk struct {
+		xact.BckJog
+		objSizeLimit int64
+		chunkSize    int64
+	}
+)
+
+// interface guard
+var (
+	_ core.Xact      = (*xactRechunk)(nil)
+	_ xreg.Renewable = (*rechunkFactory)(nil)
+)
+
+///////////////////
+// rechunkFactory //
+///////////////////
+
+func (p *rechunkFactory) New(args xreg.Args, bck *meta.Bck) xreg.Renewable {
+	return &rechunkFactory{RenewBase: xreg.RenewBase{Args: args, Bck: bck}, kind: p.kind}
+}
+
+func (p *rechunkFactory) Start() (err error) {
+	p.xctn, err = newxactRechunk(p)
+	return err
+}
+
+func (p *rechunkFactory) Kind() string   { return p.kind }
+func (p *rechunkFactory) Get() core.Xact { return p.xctn }
+
+func (p *rechunkFactory) WhenPrevIsRunning(prevEntry xreg.Renewable) (wpr xreg.WPR, err error) {
+	prev := prevEntry.(*rechunkFactory)
+
+	if p.UUID() == prev.UUID() {
+		return xreg.WprUse, nil
+	}
+
+	var (
+		prevArgs = prev.Args.Custom.(*xreg.RechunkArgs)
+		currArgs = p.Args.Custom.(*xreg.RechunkArgs)
+		xprev    = prevEntry.Get()
+	)
+
+	if prevArgs.ObjSizeLimit == currArgs.ObjSizeLimit && prevArgs.ChunkSize == currArgs.ChunkSize {
+		// Same configuration - ignore the request and reuse the existing xaction
+		return xreg.WprUse, cmn.NewErrXactUsePrev(xprev.String())
+	}
+
+	// Different configuration - fail with recommendation to abort the running one
+	return wpr, cmn.NewErrFailedTo(
+		core.T,
+		"start new rechunk",
+		xprev.String(),
+		fmt.Errorf("rechunk with different objsize_limit (%s vs %s) or chunk_size (%s vs %s) is already running; to override, first stop the running one %q",
+			cos.ToSizeIEC(prevArgs.ObjSizeLimit, 0),
+			cos.ToSizeIEC(currArgs.ObjSizeLimit, 0),
+			cos.ToSizeIEC(prevArgs.ChunkSize, 0),
+			cos.ToSizeIEC(currArgs.ChunkSize, 0),
+			xprev.ID()),
+	)
+}
+
+////////////////
+// xactRechunk //
+////////////////
+
+func newxactRechunk(p *rechunkFactory) (*xactRechunk, error) {
+	var (
+		args      = p.Args.Custom.(*xreg.RechunkArgs)
+		xch       = &xactRechunk{objSizeLimit: args.ObjSizeLimit, chunkSize: args.ChunkSize}
+		config    = cmn.GCO.Get()
+		slab, err = core.T.PageMM().GetSlab(memsys.MaxPageSlabSize)
+		mpopts    = &mpather.JgroupOpts{
+			CTs:      []string{fs.ObjCT},
+			VisitObj: xch.do,
+			Slab:     slab,
+			Prefix:   args.Prefix,
+			DoLoad:   mpather.Load,
+			Throttle: false,
+		}
+	)
+
+	if args.ChunkSize <= 0 {
+		return nil, cmn.NewErrFailedTo(core.T, "newxactRechunk", "chunk size is not set", nil)
+	}
+
+	debug.AssertNoErr(err)
+	mpopts.Bck.Copy(p.Bck.Bucket())
+
+	xch.BckJog.Init(p.UUID(), p.Kind(), xch.ctlMsg(), p.Bck, mpopts, config)
+
+	return xch, nil
+}
+
+// | Object Size       | Was Chunked? | Action                 |
+// |-------------------|--------------|------------------------|
+// | < objSizeLimit    |     Yes      | Restore monolithic     |
+// | < objSizeLimit    |     No       | No-op                  |
+// | >= objSizeLimit   |     Yes      | Re-chunk               |
+// | >= objSizeLimit   |     No       | Re-chunk               |
+func (xch *xactRechunk) do(lom *core.LOM, _ []byte) error {
+	var (
+		size      = lom.Lsize()
+		chunkSize = xch.chunkSize
+	)
+	if size < xch.objSizeLimit || xch.objSizeLimit == 0 {
+		if !lom.IsChunked() {
+			return nil // Do nothing: monolithic objects stay monolithic
+		}
+		chunkSize = 0 // restore chunked objects to monolithic
+	}
+
+	lom.Lock(true)
+	defer lom.Unlock(true)
+
+	lh, err := lom.Open()
+	if err != nil {
+		xch.AddErr(err, 0)
+		return err
+	}
+	defer lh.Close()
+
+	params := core.AllocPutParams()
+	{
+		// NOTE: if chunkSize > 0, will trigger the underlying `poi.chunk(chunkSize)`
+		params.ChunkSize = chunkSize
+		params.WorkTag = fs.WorkfilePut
+		params.Xact = xch
+		params.Reader = lh
+		// NOTE: if chunkSize == 0 (chunk disabled) but size > config.MaxMonolithicSize,
+		// it will trigger the underlying `poi.chunk(config.MaxMonolithicSize)`
+		params.Size = size
+		params.OWT = cmn.OwtChunks
+		params.Atime = lom.Atime()
+		params.Locked = true // see `lom.Lock(true)` above
+	}
+
+	err = core.T.PutObject(lom, params)
+	core.FreePutParams(params)
+	if err != nil {
+		nlog.Errorln(lom.Cname(), "putObject with chunk size", xch.chunkSize, "error", err)
+		xch.AddErr(err, 0)
+		return err
+	}
+
+	return nil
+}
+
+func (xch *xactRechunk) Run(wg *sync.WaitGroup) {
+	wg.Done()
+	xch.BckJog.Run()
+	errJog := xch.BckJog.Wait()
+	if errJog != nil && !xch.IsAborted() {
+		nlog.Warningln(xch.Name(), errJog)
+	}
+	xch.Finish()
+}
+
+func (xch *xactRechunk) Snap() *core.Snap {
+	snap := &core.Snap{}
+	xch.ToSnap(snap)
+
+	snap.IdleX = xch.IsIdle()
+	return snap
+}
+
+func (xch *xactRechunk) ctlMsg() string {
+	return fmt.Sprintf("objsize_limit=%s, chunk_size=%s", cos.ToSizeIEC(xch.objSizeLimit, 0), cos.ToSizeIEC(xch.chunkSize, 0))
+}

Original file line number	Diff line number	Diff line change
`@@ -60,6 +60,7 @@ type (`
`60`	`60`	`parts apc.MptCompletedParts`
`61`	`61`	`isS3 bool`
`62`	`62`	`coldGET bool`
	`63`	`+ locked bool // true if the LOM is already locked by the caller`
`63`	`64`	`}`
`64`	`65`	`)`
`65`	`66`
`@@ -529,8 +530,8 @@ func (ups ups) complete(args completeArgs) (string, int, error) {`
`529`	`530`
`530`	`531`	`// atomically flip: persist manifest, mark chunked, persist main`
`531`	`532`	`// NOTE: coldGET implies the LOM's lock has been promoted to wlock`
`532`		`- debug.Assertf(!args.coldGET \|\| lom.IsLocked() == apc.LockWrite, "expecting wlock, have %d", lom.IsLocked())`
`533`		`- if err = lom.CompleteUfest(manifest, args.coldGET); err != nil {`
	`533`	`+ debug.Assertf(!args.locked \|\| lom.IsLocked() == apc.LockWrite, "expecting wlock, have %d", lom.IsLocked())`
	`534`	`+ if err = lom.CompleteUfest(manifest, args.locked); err != nil {`
`534`	`535`	`nlog.Errorf("upload %q: failed to complete %s locally: %v", uploadID, lom.Cname(), err)`
`535`	`536`	`return "", http.StatusInternalServerError, err`
`536`	`537`	`}`
Original file line number	Diff line number	Diff line change
`@@ -81,6 +81,10 @@ func RenewBckLoadLomCache(uuid string, bck *meta.Bck) RenewRes {`
`81`	`81`	`return RenewBucketXact(apc.ActLoadLomCache, bck, Args{UUID: uuid})`
`82`	`82`	`}`
`83`	`83`
	`84`	`+func RenewBckRechunks(bck meta.Bck, uuid string, args RechunkArgs) RenewRes {`
	`85`	`+ return RenewBucketXact(apc.ActRechunk, bck, Args{Custom: args, UUID: uuid})`
	`86`	`+}`
	`87`	`+`
`84`	`88`	`func RenewPutMirror(lom *core.LOM) RenewRes {`
`85`	`89`	`return RenewBucketXact(apc.ActPutCopies, lom.Bck(), Args{Custom: lom})`
`86`	`90`	`}`