NVIDIA
diff --git a/‎ec/getjogger.go
Lines changed: 34 additions & 8 deletions b/‎ec/getjogger.go
Lines changed: 34 additions & 8 deletions
diff --git a/‎ec/multireader.go
Lines changed: 33 additions & 0 deletions b/‎ec/multireader.go
Lines changed: 33 additions & 0 deletions
@@ -202,6 +202,7 @@ func (c *getJogger) copyMissingReplicas(ctx *restoreCtx, reader cos.ReadOpenClos
 			nlog.Errorf("%s failed to send %s to %v: %v", core.T, ctx.lom, daemons, err)
 		}
 		freeObject(reader)
+		srcReader.Close()
 	}
 	src := &dataSource{
 		reader:   srcReader,
@@ -477,13 +478,21 @@ func cksumSlice(reader io.Reader, recvCksum *cos.Cksum, objName string) error {
 	return err
 }
 
+func closeReaders(objs []io.ReadCloser) {
+	for _, obj := range objs {
+		if obj != nil {
+			obj.Close()
+		}
+	}
+}
+
 // Reconstruct the main object from slices. Returns the list of reconstructed slices.
 func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 	var (
 		err       error
 		sliceCnt  = ctx.meta.Data + ctx.meta.Parity
 		sliceSize = SliceSize(ctx.meta.Size, ctx.meta.Data)
-		readers   = make([]io.Reader, sliceCnt)
+		readers   = make([]io.ReadCloser, sliceCnt)
 		writers   = make([]io.Writer, sliceCnt)
 		restored  = make([]*slice, sliceCnt)
 		cksums    = make([]*cos.CksumHash, sliceCnt)
@@ -511,7 +520,7 @@ func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 			continue
 		}
 
-		var cksmReader io.Reader
+		var cksmReader io.ReadCloser
 		if sgl, ok := sl.writer.(*memsys.SGL); ok {
 			readers[i] = memsys.NewReader(sgl)
 			cksmReader = memsys.NewReader(sgl)
@@ -536,9 +545,11 @@ func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 			}
 			readers[i] = nil
 		}
+		cksmReader.Close()
 	}
 
 	if err != nil {
+		closeReaders(readers)
 		return restored, err
 	}
 
@@ -547,10 +558,16 @@ func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 	}
 	stream, err := reedsolomon.NewStreamC(ctx.meta.Data, ctx.meta.Parity, true, true)
 	if err != nil {
+		closeReaders(readers)
 		return restored, err
 	}
 
-	if err := stream.Reconstruct(readers, writers); err != nil {
+	rebuildReaders := make([]io.Reader, len(readers))
+	for i, rdr := range readers {
+		rebuildReaders[i] = rdr
+	}
+	if err := stream.Reconstruct(rebuildReaders, writers); err != nil {
+		closeReaders(readers)
 		return restored, err
 	}
 
@@ -565,7 +582,7 @@ func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 	}
 
 	version := ""
-	srcReaders := make([]io.Reader, ctx.meta.Data)
+	srcReaders := make([]io.ReadCloser, ctx.meta.Data)
 	for i := range ctx.meta.Data {
 		if ctx.slices[i] != nil && ctx.slices[i].writer != nil {
 			if version == "" {
@@ -575,10 +592,12 @@ func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 				srcReaders[i] = memsys.NewReader(sgl)
 			} else {
 				if ctx.slices[i].workFQN == "" {
+					closeReaders(readers)
 					return restored, fmt.Errorf("invalid writer: %T", ctx.slices[i].writer)
 				}
 				srcReaders[i], err = cos.NewFileHandle(ctx.slices[i].workFQN)
 				if err != nil {
+					closeReaders(readers)
 					return restored, err
 				}
 			}
@@ -592,18 +611,22 @@ func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 		if restored[i].workFQN != "" {
 			srcReaders[i], err = cos.NewFileHandle(restored[i].workFQN)
 			if err != nil {
+				closeReaders(srcReaders)
+				closeReaders(readers)
 				return restored, err
 			}
 		} else {
 			sgl, ok := restored[i].obj.(*memsys.SGL)
 			if !ok {
+				closeReaders(srcReaders)
+				closeReaders(readers)
 				return restored, fmt.Errorf("empty slice %s[%d]", ctx.lom, i)
 			}
 			srcReaders[i] = memsys.NewReader(sgl)
 		}
 	}
 
-	src := io.MultiReader(srcReaders...)
+	src := newMultiReader(srcReaders...)
 	if cmn.Rom.FastV(4, cos.SmoduleEC) {
 		nlog.Infof("Saving main object %s to %q", ctx.lom, ctx.lom.FQN)
 	}
@@ -615,13 +638,15 @@ func (c *getJogger) restoreMainObj(ctx *restoreCtx) ([]*slice, error) {
 	mainMeta := *ctx.meta
 	mainMeta.SliceID = 0
 	args := &WriteArgs{
-		Reader:     src,
+		Reader:     src.mr,
 		MD:         mainMeta.NewPack(),
 		Cksum:      cos.NewCksum(cksumType, ""),
 		Generation: mainMeta.Generation,
 		Xact:       c.parent,
 	}
 	err = WriteReplicaAndMeta(ctx.lom, args)
+	src.Close()
+	closeReaders(readers)
 	return restored, err
 }
 
@@ -732,14 +757,15 @@ func (c *getJogger) uploadRestoredSlices(ctx *restoreCtx, slices []*slice) error
 		}
 
 		// Every slice's SGL is freed upon transfer completion
-		cb := func(daemonID string, s *slice) transport.ObjSentCB {
+		cb := func(daemonID string, s *slice, rdr cos.ReadOpenCloser) transport.ObjSentCB {
 			return func(_ *transport.ObjHdr, _ io.ReadCloser, _ any, err error) {
 				if err != nil {
 					nlog.Errorf("%s failed to send %s to %v: %v", core.T, ctx.lom, daemonID, err)
 				}
 				s.free()
+				rdr.Close()
 			}
-		}(tid, sl)
+		}(tid, sl, reader)
 		if err := c.parent.writeRemote([]string{tid}, ctx.lom, dataSrc, cb); err != nil {
 			remoteErr = err
 			nlog.Errorf("%s failed to send slice %s[%d] to %s", core.T, ctx.lom, sliceID, tid)
 
@@ -0,0 +1,33 @@
+// Package ec provides erasure coding (EC) based data protection for AIStore.
+/*
+ * Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved.
+ */
+package ec
+
+import "io"
+
+type multiReader struct {
+	mr  io.Reader       // standard io.multiReader
+	rcs []io.ReadCloser // to close
+}
+
+func newMultiReader(rcs ...io.ReadCloser) *multiReader {
+	mrWrapper := &multiReader{
+		rcs: rcs,
+	}
+	readers := make([]io.Reader, len(rcs))
+	for i, rc := range rcs {
+		readers[i] = io.Reader(rc)
+	}
+	mrWrapper.mr = io.MultiReader(readers...)
+	return mrWrapper
+}
+
+func (mr *multiReader) Read(p []byte) (int, error) { return mr.mr.Read(p) }
+
+func (mr *multiReader) Close() (err error) {
+	for _, r := range mr.rcs {
+		err = r.Close()
+	}
+	return err
+}