shared data mover: non-primary activation; stale xaction cleanup

* non-primary get-batch endpoint: signal SDM active before phase 2
  - via fast-kalive
* streams-toggle:
  - use MSB in streamsToggle.last to track non-primary state
  - nonpSetActive/nonpResetActive/nonpUndo for state transitions
* shared data mover:
  * track per-XID last-called timestamp
  * isActive() with respect to age
  * Close() now automatically cleans up stale entries

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

Author: alex-aizman

ais/kalive.go

@@ -131,7 +131,7 @@ func (tkr *talive) cluUptime(now int64) (elapsed time.Duration) {
 	return
 }
 
-func (tkr *talive) sendKalive(smap *smapX, timeout time.Duration, _ int64, fast bool) (pid string, status int, err error) {
+func (tkr *talive) sendKalive(smap *smapX, timeout time.Duration, now int64, fast bool) (pid string, status int, err error) {
 	t := tkr.t
 	if fast {
 		// additionally
@@ -140,7 +140,7 @@ func (tkr *talive) sendKalive(smap *smapX, timeout time.Duration, _ int64, fast
 	}
 	if fast {
 		debug.Assert(ec.ECM != nil)
-		pid, _, err = t.fastKalive(smap, timeout, ec.ECM.IsActive(), bundle.SDM.IsActive())
+		pid, _, err = t.fastKalive(smap, timeout, ec.ECM.IsActive(), bundle.SDM.IsActive(now))
 		return pid, 0, err
 	}
 	return t.slowKalive(smap, tkr.t, timeout)
@@ -198,7 +198,8 @@ func (pkr *palive) sendKalive(smap *smapX, timeout time.Duration, now int64, fas
 	debug.Assert(!smap.isPrimary(pkr.p.si))
 
 	if fast {
-		pid, hdr, err := pkr.p.fastKalive(smap, timeout, false, false /*shared streams*/)
+		last, dmActive := pkr.p.dm.nonpResetActive()
+		pid, hdr, err := pkr.p.fastKalive(smap, timeout, false, dmActive /*shared streams*/)
 		if err == nil {
 			// (shared streams; EC streams)
 			if pkr.p.ec.isActive(hdr) {
@@ -207,6 +208,8 @@ func (pkr *palive) sendKalive(smap *smapX, timeout time.Duration, now int64, fas
 			if pkr.p.dm.isActive(hdr) {
 				pkr.p.dm.setActive(now)
 			}
+		} else if dmActive {
+			pkr.p.dm.nonpUndo(last) // (unlikely)
 		}
 		return pid, 0, err
 	}

ais/ml.go

@@ -17,6 +17,7 @@ import (
 	"github.com/NVIDIA/aistore/cmn/archive"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
+	"github.com/NVIDIA/aistore/cmn/mono"
 	"github.com/NVIDIA/aistore/cmn/nlog"
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/transport/bundle"
@@ -168,6 +169,12 @@ func (p *proxy) httpmlget(w http.ResponseWriter, r *http.Request) {
 	if cmn.Rom.V(5, cos.ModAIS) {
 		nlog.Infoln(p.String(), apc.Moss, "DT", tsi.String(), "xid", xid, "wid", wid, "[", hreq.Path, hreq.Method, "]")
 	}
+
+	// toggle SDM (fast-kalive => primary)
+	if !smap.isPrimary(p.si) {
+		p.dm.nonpSetActive(mono.NanoTime())
+	}
+
 	// phase 2: async broadcast -> all except DT
 	if nat > 1 {
 		args := allocBcArgs()
@@ -385,7 +392,6 @@ func (ctx *mossCtx) phase2(w http.ResponseWriter, r *http.Request, smap *smapX,
 		xmoss.BcastAbort(err)
 		xmoss.Abort(err)
 		t.writeErr(w, r, err)
-		return
 	}
 }
 

ais/streams_toggle.go

@@ -12,6 +12,7 @@ import (
 	"github.com/NVIDIA/aistore/api/apc"
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/atomic"
+	"github.com/NVIDIA/aistore/cmn/debug"
 	"github.com/NVIDIA/aistore/cmn/mono"
 	"github.com/NVIDIA/aistore/cmn/nlog"
 	"github.com/NVIDIA/aistore/core/meta"
@@ -21,17 +22,18 @@ import (
 // auto-deactivates after a timeout;
 // negative timeout (see `tout` below) => never deactivate
 
-// [TODO] shared-DM:
-// - apc.HdrActiveDM
-// - dmToggle.on()
-// - extra checks on then target side (unregIf, etc.)
+// non-primary action
+const (
+	nonpBit  = uint64(1) << 63
+	nonpMask = ^nonpBit
+)
 
 type (
 	streamsToggle struct {
-		hdrActive string       // apc.HdrActiveEC, ...
-		actOn     string       // apc.ActOpenEC, ...
-		actOff    string       // apc.ActCloseEC, ...
-		last      atomic.Int64 // mono-time of last positive refresh
+		hdrActive string        // apc.HdrActiveEC, ...
+		actOn     string        // apc.ActOpenEC, ...
+		actOff    string        // apc.ActCloseEC, ...
+		last      atomic.Uint64 // mono-time of last positive refresh
 	}
 
 	ecToggle struct {
@@ -56,10 +58,34 @@ func (f *dmToggle) init() {
 	f.actOff = apc.ActCloseSDM
 }
 
-func (*dmToggle) timeout() time.Duration { return cmn.SharedStreamsDflt }
+func (*dmToggle) timeout() time.Duration { return cmn.SharedStreamsDflt } // see also: sharedDM.getActive
 
 func (f *streamsToggle) isActive(h http.Header) bool { _, ok := h[f.hdrActive]; return ok }
-func (f *streamsToggle) setActive(now int64)         { f.last.Store(now) }
+
+func (f *streamsToggle) setActive(now int64) {
+	f.last.Store(uint64(now))
+}
+
+// non-primary: executed user call requiring SDM
+func (f *streamsToggle) nonpSetActive(now int64) {
+	debug.Assert(now > 0, "invalid timestamp: ", now)
+	f.last.Store((uint64(now) & nonpMask) | nonpBit)
+}
+
+// non-primary: fast-keepalive => primary
+func (f *streamsToggle) nonpResetActive() (uint64, bool) {
+	last := f.last.Load()
+	if last&nonpBit == 0 {
+		return last, false
+	}
+	last = f.last.Swap(0)
+	return last, true
+}
+
+// non-primary: fast-keepalive => primary
+func (f *streamsToggle) nonpUndo(last uint64) {
+	_ = f.last.CAS(0, last)
+}
 
 // target => primary keep-alive
 func (f *streamsToggle) recvKalive(p *proxy, hdr http.Header, now int64, tout time.Duration) {
@@ -70,7 +96,7 @@ func (f *streamsToggle) recvKalive(p *proxy, hdr http.Header, now int64, tout ti
 	if tout < 0 {
 		return
 	}
-	last := f.last.Load()
+	last := int64(f.last.Load() & nonpMask)
 	if last == 0 || time.Duration(now-last) < tout {
 		return
 	}
@@ -80,7 +106,10 @@ func (f *streamsToggle) recvKalive(p *proxy, hdr http.Header, now int64, tout ti
 // primary => target keep-alive
 func (f *streamsToggle) respKalive(hdr http.Header, now int64, tout time.Duration) {
 	if tout > 0 {
-		if last := f.last.Load(); last != 0 && time.Duration(now-last) < tout {
+		v := f.last.Load()
+		last := int64(v & nonpMask)
+		debug.Assert(int64(v) >= 0, f.hdrActive, " invalid timestamp: ", v)
+		if last != 0 && time.Duration(now-last) < tout {
 			hdr.Set(f.hdrActive, "true")
 		}
 	}
@@ -96,7 +125,7 @@ func (f *streamsToggle) on(p *proxy, tout time.Duration) error {
 	}
 	var (
 		now  = mono.NanoTime()
-		last = f.last.Load()
+		last = int64(f.last.Load() & nonpMask)
 	)
 	if last != 0 && time.Duration(now-last) < cmn.SharedStreamsNack {
 		return nil
@@ -109,7 +138,8 @@ func (f *streamsToggle) on(p *proxy, tout time.Duration) error {
 }
 
 func (f *streamsToggle) off(p *proxy, last int64) {
-	if !f.last.CAS(last, 0) {
+	debug.Assert(last > 0, f.hdrActive, " invalid timestamp: ", last)
+	if !f.last.CAS(uint64(last), 0) {
 		return
 	}
 

ais/tgtec.go

@@ -15,6 +15,7 @@ import (
 	"github.com/NVIDIA/aistore/cmn"
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
+	"github.com/NVIDIA/aistore/cmn/mono"
 	"github.com/NVIDIA/aistore/cmn/nlog"
 	"github.com/NVIDIA/aistore/core"
 	"github.com/NVIDIA/aistore/core/meta"
@@ -161,7 +162,7 @@ func (t *target) httpecpost(w http.ResponseWriter, r *http.Request) {
 		if !t.ensureIntraControl(w, r, true /* from primary */) {
 			return
 		}
-		if bundle.SDM.IsActive() {
+		if bundle.SDM.IsActive(mono.NanoTime()) {
 			t.writeErr(w, r, _errOff(bundle.SDMName))
 			return
 		}
@@ -184,10 +185,10 @@ func closeEc(int64) time.Duration {
 	return hk.UnregInterval
 }
 
-func closeSDM(int64) time.Duration {
-	if bundle.SDM.IsActive() {
+func closeSDM(now int64) time.Duration {
+	if bundle.SDM.IsActive(now) {
 		nlog.Warningln("hk-cb:", _errOff(bundle.SDMName))
-	} else if err := bundle.SDM.Close(); err != nil {
+	} else if err := bundle.SDM.Close(now); err != nil {
 		nlog.Errorln(err)
 	}
 	return hk.UnregInterval

bench/tools/aisloader/params.go

@@ -46,7 +46,7 @@ type (
 		duration             DurationExt // benchmark duration
 		putSizeUpperBoundStr string      // total PUT size limit (unparsed)
 		tmpDir               string      // temp directory for file-based readers
-		readerType           string      // reader type: sg(default), file, rand, tar
+		readerType           string      // reader type: sg(default), file, rand, tar (PUT only)
 		putSizeUpperBound    int64       // total PUT size limit (bytes)
 		seed                 int64       // random seed for reproducibility
 		maxputs              uint64      // max number of PUT operations

bench/tools/aisloader/print.go

@@ -403,6 +403,15 @@ func printRunParams(p *params) {
 		}
 	}
 
+	// omit PUT-only
+	var (
+		minsize, maxsize int64
+		readerType       string
+	)
+	if p.putPct > 0 {
+		minsize, maxsize = p.minSize, p.maxSize
+		readerType = p.readerType
+	}
 	b, err := jsoniter.MarshalIndent(ppEssential{
 		URL:           p.proxyURL,
 		Bucket:        p.bck.Cname(""),
@@ -413,12 +422,12 @@ func printRunParams(p *params) {
 		UpdatePct:     p.updateExistingPct,
 		MultipartPct:  p.multipartPct,
 		GetBatchSize:  p.getBatchSize,
-		MinSize:       p.minSize,
-		MaxSize:       p.maxSize,
+		MinSize:       minsize,
+		MaxSize:       maxsize,
 		MaxPutBytes:   p.putSizeUpperBound,
 		Arch:          arch,
 		NameGetter:    ngLabel(),
-		ReaderType:    p.readerType,
+		ReaderType:    readerType,
 		Cleanup:       p.cleanUp.Val,
 	}, "", "   ")
 

bench/tools/aisloader/run.go

@@ -200,7 +200,7 @@ func Start(version, buildtime string) (err error) {
 		runParams.duration.Val = time.Duration(math.MaxInt64)
 	}
 
-	if runParams.readerType == readers.File {
+	if runParams.readerType == readers.File && runParams.putPct > 0 {
 		if err := cos.CreateDir(runParams.tmpDir + "/" + myName); err != nil {
 			return fmt.Errorf("failed to create local test directory %q, err = %s", runParams.tmpDir, err.Error())
 		}
@@ -707,7 +707,7 @@ func cleanupObjs(objs []string, wg *sync.WaitGroup) {
 		}
 	}
 
-	if runParams.readerType == readers.File {
+	if runParams.readerType == readers.File && runParams.putPct > 0 {
 		for _, obj := range objs {
 			if err := os.Remove(runParams.tmpDir + "/" + obj); err != nil {
 				fmt.Println("delete local file err:", err)

core/lom_xattr.go

@@ -87,7 +87,7 @@ const (
 	DumpLomEnvVar = "AIS_DUMP_LOM"
 )
 
-const lomDirtyMask = uint64(1 << 63)
+const lomDirtyMask = uint64(1) << 63
 
 const (
 	badLmeta = "bad lmeta"

core/lombid.go

@@ -15,7 +15,7 @@ import (
 // [63]=AisBID | [62:60]=legacy flags (3 bits) | [59:0]=serial
 
 const (
-	AisBID = uint64(1 << 63)
+	AisBID = uint64(1) << 63
 
 	bitshift   = 60
 	flagMaskV1 = math.MaxUint64 >> bitshift         // 0xf

core/meta/bck.go

@@ -121,7 +121,7 @@ func (b *Bck) String() string {
 	// add BID
 	// for the mask to clear "ais" bit and/or other high bits reserved for LOM flags, see core/lombid
 	const (
-		aisBID = uint64(1 << 63)
+		aisBID = uint64(1) << 63
 	)
 	var sb strings.Builder
 	sb.Grow(64)