multi-object copy/transform with '--sync' option

* initial implementation

Signed-off-by: Alex Aizman <[email protected]>

ais/earlystart.go

@@ -442,6 +442,8 @@ until:
 		goto until // repeat
 	}
 
+	// TODO: skip when there's a single target (benign)
+
 	// do
 	var (
 		msg    = &apc.ActMsg{Action: apc.ActRebalance, Value: metaction3}

cmd/cli/config/config.go

@@ -70,6 +70,7 @@ var (
 		// object
 		"get":      "object get",
 		"put":      "object put",
+		"rmo":      "object rm",
 		"prefetch": "object prefetch", // same as "job start prefetch"
 		// bucket
 		"ls":     "bucket ls",

xact/xs/prune.go

@@ -7,6 +7,8 @@ package xs
 
 import (
 	"context"
+	"fmt"
+	"net/http"
 	"time"
 
 	"github.com/NVIDIA/aistore/cmn"
@@ -37,7 +39,7 @@ type prune struct {
 }
 
 func (rp *prune) init(config *cmn.Config) {
-	debug.Assert(rp.bckFrom.HasVersioningMD(), rp.bckFrom.String())
+	debug.Assert(rp.bckFrom.IsAIS() || rp.bckFrom.HasVersioningMD(), rp.bckFrom.String())
 	rmopts := &mpather.JgroupOpts{
 		CTs:      []string{fs.ObjectType},
 		VisitObj: rp.do,
@@ -100,18 +102,37 @@ func (rp *prune) do(dst *core.LOM, _ []byte) error {
 
 	// skip objects already copied by rp.parent (compare w/ reb)
 	uname := cos.UnsafeB(src.Uname())
-	if rp.filter.Lookup(uname) {
+	if rp.filter != nil && rp.filter.Lookup(uname) { // TODO -- FIXME: rm filter nil check once x-tco supports prob. filtering
 		rp.filter.Delete(uname)
 		return nil
 	}
 
-	// head
-	_, errCode, err := core.T.Backend(src.Bck()).HeadObj(context.Background(), src)
-	if err == nil || !cos.IsNotExist(err, errCode) {
+	// check whether src lom exists
+	var (
+		err     error
+		errCode int
+	)
+	if src.Bck().IsAIS() {
+		smap := core.T.Sowner().Get()
+		tsi, errV := smap.HrwHash2T(src.Digest())
+		if errV != nil {
+			return fmt.Errorf("prune %s: fatal err: %w", rp.parent.Name(), errV)
+		}
+		if tsi.ID() == core.T.SID() {
+			err = src.Load(false, false)
+		} else {
+			if present := core.T.HeadObjT2T(src, tsi); !present {
+				errCode = http.StatusNotFound
+			}
+		}
+	} else {
+		_, errCode, err = core.T.Backend(src.Bck()).HeadObj(context.Background(), src)
+	}
+	if err == nil || !cos.IsNotExist(err, errCode) /*not complaining*/ {
 		return nil
 	}
 
-	// source does not exist: try to remove the destination (NOTE: best effort)
+	// source does not exist: try to remove the destination (NOTE best effort)
 	if !dst.TryLock(true) {
 		return nil
 	}

xact/xs/tcobjs.go

@@ -169,9 +169,23 @@ func (r *XactTCObjs) Run(wg *sync.WaitGroup) {
 			nat := smap.CountActiveTs()
 			wi.refc.Store(int32(nat - 1))
 
+			// run
+			var wg *sync.WaitGroup
 			if err = lrit.init(r, &msg.ListRange, r.Bck()); err == nil {
+				if msg.Sync && lrit.lrp != lrpList {
+					wg = &sync.WaitGroup{}
+					wg.Add(1)
+					go func() {
+						r.prune(lrit, smap)
+						wg.Done()
+					}()
+				}
 				err = lrit.run(wi, smap)
 			}
+			if wg != nil {
+				wg.Wait()
+			}
+
 			if r.IsAborted() || err != nil {
 				goto fin
 			}
@@ -322,6 +336,43 @@ func (wi *tcowi) do(lom *core.LOM, lrit *lriterator) {
 			wi.r.AddErr(err, 5, cos.SmoduleXs)
 		}
 	} else if cmn.Rom.FastV(5, cos.SmoduleXs) {
-		nlog.Infof("%s: tco-lr %s => %s", wi.r.Base.Name(), lom.Cname(), wi.r.args.BckTo.Cname(objNameTo))
+		nlog.Infoln(wi.r.Name()+":", lom.Cname(), "=>", wi.r.args.BckTo.Cname(objNameTo))
 	}
 }
+
+//
+// remove objects not present at the source (when synchronizing bckFrom => bckTo)
+// TODO: probabilistic filtering
+//
+
+type syncwi struct {
+	rp *prune
+}
+
+// interface guard
+var _ lrwi = (*syncwi)(nil)
+
+func (r *XactTCObjs) prune(lrit *lriterator, smap *meta.Smap) {
+	rp := prune{parent: r}
+	rp.bckFrom, rp.bckTo = r.FromTo()
+
+	// tcb use case
+	if lrit.lrp == lrpPrefix {
+		rp.prefix = lrit.pt.Prefix
+		rp.init(r.config)
+		rp.run()
+		rp.wait()
+		return
+	}
+
+	// same range iterator but different bucket
+	debug.Assert(lrit.lrp == lrpRange)
+	syncit := *lrit
+	syncit.bck = rp.bckTo
+	syncwi := &syncwi{&rp} // reusing only prune.do (and not init/run/wait)
+	syncit.run(syncwi, smap)
+}
+
+func (syncwi *syncwi) do(lom *core.LOM, _ *lriterator) {
+	syncwi.rp.do(lom, nil)
+}