core: return object reader, not file

* `cos.LomReader` Signed-off-by: Alex Aizman <[email protected]>
NVIDIA · Jun 21, 2024 · 3816bfb · 3816bfb
1 parent f882ef4
commit 3816bfb
Show file tree

Hide file tree

Showing 6 changed files with 38 additions and 28 deletions.
diff --git a/ais/tgtfcold.go b/ais/tgtfcold.go
@@ -26,17 +26,21 @@ const ftcg = "Warning: failed to cold-GET"
 
 func (goi *getOI) coldReopen(res *core.GetReaderResult) error {
 	var (
-		t, lom = goi.t, goi.lom
+		err    error
+		lmfh   cos.LomReader
+		wfh    *os.File
+		t      = goi.t
+		lom    = goi.lom
 		revert string
 	)
 	if goi.verchanged {
 		revert = fs.CSM.Gen(lom, fs.WorkfileType, fs.WorkfileColdget)
-		if err := lom.RenameMainTo(revert); err != nil {
-			nlog.Errorln("failed to rename prev. version - proceeding anyway", lom.FQN, "=>", revert)
+		if errV := lom.RenameMainTo(revert); errV != nil {
+			nlog.Errorln("failed to rename prev. version - proceeding anyway", lom.Cname(), "=>", revert)
 			revert = ""
 		}
 	}
-	lmfh, err := lom.Create()
+	wfh, err = lom.Create()
 	if err != nil {
 		cos.Close(res.R)
 		goi._cleanup(revert, nil, nil, nil, err, "(fcreate)")
@@ -48,28 +52,28 @@ func (goi *getOI) coldReopen(res *core.GetReaderResult) error {
 		written   int64
 		buf, slab = t.gmm.AllocSize(min(res.Size, memsys.DefaultBuf2Size))
 		cksumH    = cos.NewCksumHash(lom.CksumConf().Type)
-		mw        = cos.NewWriterMulti(lmfh, cksumH.H)
+		mw        = cos.NewWriterMulti(wfh, cksumH.H)
 	)
 	written, err = cos.CopyBuffer(mw, res.R, buf)
 	cos.Close(res.R)
 
 	if err != nil {
-		goi._cleanup(revert, lmfh, buf, slab, err, "(rr/wl)")
+		goi._cleanup(revert, wfh, buf, slab, err, "(rr/wl)")
 		return err
 	}
 	debug.Assertf(written == res.Size, "%s: remote-size %d != %d written", lom.Cname(), res.Size, written)
 
 	if lom.IsFeatureSet(feat.FsyncPUT) {
 		// fsync (flush)
-		if err = lmfh.Sync(); err != nil {
-			goi._cleanup(revert, lmfh, buf, slab, err, "(fsync)")
+		if err = wfh.Sync(); err != nil {
+			goi._cleanup(revert, wfh, buf, slab, err, "(fsync)")
 			return err
 		}
 	}
-	err = lmfh.Close()
-	lmfh = nil
+	err = wfh.Close()
+	wfh = nil
 	if err != nil {
-		goi._cleanup(revert, lmfh, buf, slab, err, "(fclose)")
+		goi._cleanup(revert, wfh, buf, slab, err, "(fclose)")
 		return err
 	}
 
@@ -83,12 +87,13 @@ func (goi *getOI) coldReopen(res *core.GetReaderResult) error {
 		}
 	}
 	if err = lom.PersistMain(); err != nil {
-		goi._cleanup(revert, lmfh, buf, slab, err, "(persist)")
+		goi._cleanup(revert, wfh, buf, slab, err, "(persist)")
 		return err
 	}
 
 	// reopen & transmit ---
-	if lmfh, err = lom.Open(); err != nil {
+	lmfh, err = lom.Open()
+	if err != nil {
 		goi._cleanup(revert, nil, buf, slab, err, "(seek)")
 		return err
 	}
@@ -168,7 +173,7 @@ func (goi *getOI) _fini(revert string, fullSize, txSize int64) error {
 	return nil
 }
 
-func (goi *getOI) _cleanup(revert string, lmfh *os.File, buf []byte, slab *memsys.Slab, err error, tag string) {
+func (goi *getOI) _cleanup(revert string, lmfh io.Closer, buf []byte, slab *memsys.Slab, err error, tag string) {
 	if lmfh != nil {
 		lmfh.Close()
 	}

diff --git a/ais/tgtobj.go b/ais/tgtobj.go
@@ -1683,11 +1683,12 @@ func (a *putA2I) do() (int, error) {
 
 cpap: // copy + append
 	var (
-		err       error
-		lmfh, wfh *os.File
-		workFQN   string
-		cksum     cos.CksumHashSize
-		aw        archive.Writer
+		err     error
+		wfh     *os.File
+		lmfh    cos.LomReader
+		workFQN string
+		cksum   cos.CksumHashSize
+		aw      archive.Writer
 	)
 	workFQN = fs.CSM.Gen(a.lom, fs.WorkfileType, fs.WorkfileAppendToArch)
 	wfh, err = os.OpenFile(workFQN, os.O_CREATE|os.O_WRONLY, cos.PermRWR)

diff --git a/cmn/cos/io.go b/cmn/cos/io.go
@@ -59,16 +59,21 @@ type (
 		size int64
 	}
 
+	ReadReaderAt interface {
+		io.Reader
+		io.ReaderAt
+	}
+	LomReader interface {
+		io.ReadCloser
+		io.ReaderAt
+	}
+
 	// implementations
 
 	nopReader struct {
 		size   int
 		offset int
 	}
-	ReadReaderAt interface {
-		io.Reader
-		io.ReaderAt
-	}
 	deferRCS struct {
 		ReadCloseSizer
 		cb func()

diff --git a/core/backend.go b/core/backend.go
@@ -8,7 +8,6 @@ import (
 	"context"
 	"io"
 	"net/http"
-	"os"
 
 	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
@@ -27,7 +26,7 @@ type (
 	}
 	LsoInvCtx struct {
 		Lom    *LOM
-		Lmfh   *os.File
+		Lmfh   cos.LomReader
 		Name   string
 		ID     string
 		Schema []string

diff --git a/core/lfile.go b/core/lfile.go
@@ -18,7 +18,7 @@ import (
 // open
 //
 
-func (lom *LOM) Open() (*os.File, error) {
+func (lom *LOM) Open() (cos.LomReader, error) {
 	return os.Open(lom.FQN)
 }
 

diff --git a/xact/xs/archive.go b/xact/xs/archive.go
@@ -148,7 +148,7 @@ func (r *XactArch) Begin(msg *cmn.ArchiveBckMsg, archlom *core.LOM) (err error)
 	if core.T.SID() == wi.tsi.ID() {
 		var (
 			s           string
-			lmfh        *os.File
+			lmfh        cos.LomReader
 			finfo, errX = os.Stat(wi.archlom.FQN)
 			exists      = errX == nil
 		)
@@ -431,7 +431,7 @@ func (r *XactArch) Snap() (snap *core.Snap) {
 // archwi //
 ////////////
 
-func (wi *archwi) beginAppend() (lmfh *os.File, err error) {
+func (wi *archwi) beginAppend() (lmfh cos.LomReader, err error) {
 	msg := wi.msg
 	if msg.Mime == archive.ExtTar {
 		if err = wi.openTarForAppend(); err == nil || err != archive.ErrTarIsEmpty {
-Original file line number
+Diff line change
@@ Expand Up / @@ -18,7 +18,7 @@ import ( @@
     // open
     //
-    func (lom *LOM) Open() (*os.File, error) {
+    func (lom *LOM) Open() (cos.LomReader, error) {
     	return os.Open(lom.FQN)
     }
@@ Expand Down @@