core: add shared (multi-use, multi-job) data mover ("shared streams")

alex-aizman · alex-aizman · commit f8ee6c2d4f6d · 2025-05-29T18:50:48.000-04:00
* shared-DM code complete - x-moss remains * part four, prev. commit: c43a7b4 Signed-off-by: Alex Aizman <alex.aizman@gmail.com>
diff --git a/ais/htrun.go b/ais/htrun.go
@@ -2042,7 +2042,7 @@ func (h *htrun) regTo(url string, psi *meta.Snode, tout time.Duration, htext hte
 }
 
 // (fast path: nodes => primary)
-func (h *htrun) fastKalive(smap *smapX, timeout time.Duration, ecActive bool) (string /*pid*/, http.Header, error) {
+func (h *htrun) fastKalive(smap *smapX, timeout time.Duration, ecActive, dmActive bool) (string /*pid*/, http.Header, error) {
 	if nlog.Stopping() {
 		return "", http.Header{}, h.errStopping()
 	}
@@ -2056,10 +2056,15 @@ func (h *htrun) fastKalive(smap *smapX, timeout time.Duration, ecActive bool) (s
 		cargs.req = cmn.HreqArgs{Method: http.MethodPost, Base: primaryURL, Path: apc.URLPathCluKalive.Join(h.SID())}
 		cargs.timeout = timeout
 	}
-	if ecActive {
+	if ecActive || dmActive {
 		// (target => primary)
 		hdr := make(http.Header, 1)
-		hdr.Set(apc.HdrActiveEC, "true")
+		if ecActive {
+			hdr.Set(apc.HdrActiveEC, "true")
+		}
+		if dmActive {
+			hdr.Set(apc.HdrActiveDM, "true")
+		}
 		cargs.req.Header = hdr
 	}
 
diff --git a/ais/kalive.go b/ais/kalive.go
@@ -21,6 +21,7 @@ import (
 	"github.com/NVIDIA/aistore/ec"
 	"github.com/NVIDIA/aistore/stats"
 	"github.com/NVIDIA/aistore/sys"
+	"github.com/NVIDIA/aistore/transport/bundle"
 )
 
 const (
@@ -131,17 +132,18 @@ func (tkr *talive) cluUptime(now int64) (elapsed time.Duration) {
 }
 
 func (tkr *talive) sendKalive(smap *smapX, timeout time.Duration, _ int64, fast bool) (pid string, status int, err error) {
+	t := tkr.t
 	if fast {
 		// additionally
-		interrupted, restarted := tkr.t.interruptedRestarted()
+		interrupted, restarted := t.interruptedRestarted()
 		fast = !interrupted && !restarted
 	}
 	if fast {
 		debug.Assert(ec.ECM != nil)
-		pid, _, err = tkr.t.fastKalive(smap, timeout, ec.ECM.IsActive())
+		pid, _, err = t.fastKalive(smap, timeout, ec.ECM.IsActive(), bundle.SDM.IsOpen())
 		return pid, 0, err
 	}
-	return tkr.t.slowKalive(smap, tkr.t, timeout)
+	return t.slowKalive(smap, tkr.t, timeout)
 }
 
 func (tkr *talive) do(config *cmn.Config) (stopped bool) {
@@ -196,12 +198,15 @@ 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 /*ec active*/)
+		pid, hdr, err := pkr.p.fastKalive(smap, timeout, false, false /*shared streams*/)
 		if err == nil {
 			// (shared streams; EC streams)
 			if pkr.p.ec.isActive(hdr) {
 				pkr.p.ec.setActive(now)
 			}
+			if pkr.p.dm.isActive(hdr) {
+				pkr.p.dm.setActive(now)
+			}
 		}
 		return pid, 0, err
 	}
diff --git a/ais/target.go b/ais/target.go
@@ -71,7 +71,6 @@ type (
 		fsprg    fsprungroup
 		reb      *reb.Reb
 		res      *res.Res
-		sdm      *bundle.SharedDM
 		txns     txns
 		regstate regstate
 	}
@@ -294,10 +293,9 @@ func (t *target) init(config *cmn.Config) {
 
 	t.fsprg.init(t, newVol) // subgroup of the daemon.rg rungroup
 
-	sc := transport.Init(ts)   // init transport sub-system
-	daemon.rg.add(sc)          // new stream collector
-	t.sdm = &bundle.SharedDM{} // shared streams
-	t.sdm.Init(config, apc.CompressNever)
+	sc := transport.Init(ts)                  // init transport sub-system
+	daemon.rg.add(sc)                         // new stream collector
+	bundle.InitSDM(config, apc.CompressNever) // shared streams
 
 	t.fshc = health.NewFSHC(t)
 
diff --git a/ais/tgtec.go b/ais/tgtec.go
@@ -21,11 +21,10 @@ import (
 	"github.com/NVIDIA/aistore/core/meta"
 	"github.com/NVIDIA/aistore/ec"
 	"github.com/NVIDIA/aistore/hk"
+	"github.com/NVIDIA/aistore/transport/bundle"
 	"github.com/NVIDIA/aistore/xact/xreg"
 )
 
-var errCloseStreams = errors.New("EC is currently active, cannot close streams")
-
 func (t *target) ecHandler(w http.ResponseWriter, r *http.Request) {
 	switch r.Method {
 	case http.MethodGet:
@@ -144,17 +143,24 @@ func (t *target) httpecpost(w http.ResponseWriter, r *http.Request) {
 			return
 		}
 		if ec.ECM.IsActive() {
-			t.writeErr(w, r, errCloseStreams)
+			t.writeErr(w, r, errors.New("EC is active, cannot close"))
 			return
 		}
 		nlog.Infoln(t.String(), "hk-postpone", action)
 		hk.Reg(hkname, closeEc, postpone)
 
-	// TODO -- FIXME: simplified (compare w/ above); refactor as toggle-streams-...
 	case apc.ActDmOpen:
-		t.sdm.Open()
+		if err := bundle.SDM.Open(); err != nil {
+			t.writeErr(w, r, err)
+		}
 	case apc.ActDmClose:
-		t.sdm.Close()
+		if !t.ensureIntraControl(w, r, true /* from primary */) {
+			return
+		}
+		// TODO: consider delaying via hk (see above)
+		if err := bundle.SDM.Close(); err != nil {
+			t.writeErr(w, r, err)
+		}
 
 	default:
 		t.writeErr(w, r, errActEc(action))
@@ -163,7 +169,7 @@ func (t *target) httpecpost(w http.ResponseWriter, r *http.Request) {
 
 func closeEc(int64) time.Duration {
 	if ec.ECM.IsActive() {
-		nlog.Warningln("hk-cb:", errCloseStreams)
+		nlog.Warningln("hk-cb: cannot close EC streams")
 	} else {
 		ec.ECM.CloseStreams(false /*with refc*/)
 	}
diff --git a/transport/bundle/shared_dm.go b/transport/bundle/shared_dm.go
@@ -11,84 +11,160 @@ import (
 	"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/transport"
 	"github.com/NVIDIA/aistore/xact"
 )
 
-type SharedDM struct {
-	dm     DM
-	rxcbs  map[string]transport.RecvObj
-	rxmu   sync.Mutex
-	openmu sync.Mutex
+// [TODO]
+// - Close() vs usage (when len(rxcbs) > 0); provide xctn.onFinished() => UnregRecv
+// - limitation: hdr.Opaque is exclusively reserved xaction ID
+
+type sharedDM struct {
+	dm    DM
+	rxcbs map[string]transport.RecvObj
+	ocmu  sync.Mutex
+	rxmu  sync.Mutex
 }
 
+// global
+var SDM sharedDM
+
 // called upon target startup
-func (s *SharedDM) Init(config *cmn.Config, compression string) {
+func InitSDM(config *cmn.Config, compression string) {
 	extra := Extra{Config: config, Compression: compression}
-	s.dm.init(s.trname(), s.recv, cmn.OwtNone, extra)
+	SDM.dm.init(SDM.trname(), SDM.recv, cmn.OwtNone, extra)
 }
 
-// NOTE: constant (until and unless we run multiple shared-DMs)
-func (*SharedDM) trname() string { return "shared-dm" }
+func (sdm *sharedDM) IsOpen() bool { return sdm.dm.stage.opened.Load() }
+
+// constant (until and unless we run multiple shared-DMs)
+func (*sharedDM) trname() string { return "shared-dm" }
+
+func (sdm *sharedDM) _already() {
+	nlog.WarningDepth(2, core.T.String(), sdm.trname(), "is already open")
+}
 
 // called on-demand
-func (s *SharedDM) Open() {
-	s.openmu.Lock()
-	defer s.openmu.Unlock()
-	if s.dm.stage.opened.Load() {
-		return
+func (sdm *sharedDM) Open() error {
+	if sdm.IsOpen() {
+		sdm._already()
+		return nil
 	}
-	s.rxcbs = make(map[string]transport.RecvObj, 4)
-	s.dm.Open()
-	nlog.InfoDepth(1, core.T.String(), "open", s.trname())
+
+	sdm.ocmu.Lock()
+	if sdm.IsOpen() {
+		sdm.ocmu.Unlock()
+		sdm._already()
+		return nil
+	}
+
+	sdm.rxmu.Lock()
+	sdm.rxcbs = make(map[string]transport.RecvObj, 4)
+	sdm.rxmu.Unlock()
+
+	if err := sdm.dm.RegRecv(); err != nil {
+		sdm.ocmu.Unlock()
+		nlog.ErrorDepth(1, core.T.String(), err)
+		debug.AssertNoErr(err)
+		return err
+	}
+	sdm.dm.Open()
+	sdm.ocmu.Unlock()
+
+	nlog.InfoDepth(1, core.T.String(), "open", sdm.trname())
+	return nil
 }
 
 // nothing running + 10m inactivity
-func (s *SharedDM) Close() error {
-	s.openmu.Lock()
-	defer s.openmu.Unlock()
-	if !s.dm.stage.opened.Load() {
+func (sdm *sharedDM) Close() error {
+	if !sdm.IsOpen() {
+		return nil
+	}
+	sdm.ocmu.Lock()
+	if !sdm.IsOpen() {
+		sdm.ocmu.Unlock()
 		return nil
 	}
-	if len(s.rxcbs) > 0 {
-		return fmt.Errorf("cannot close %s: %v", s.trname(), s.rxcbs) // TODO -- FIXME: cleanup
+
+	var (
+		xid string
+		l   int
+	)
+	sdm.rxmu.Lock()
+	for xid = range sdm.rxcbs {
+		break
+	}
+	l = len(sdm.rxcbs)
+
+	if l > 0 {
+		sdm.rxmu.Unlock()
+		debug.Assert(cos.IsValidUUID(xid), xid)
+		sdm.ocmu.Unlock()
+		return fmt.Errorf("cannot close %s: [%s, %d]", sdm.trname(), xid, l)
 	}
-	s.rxcbs = nil
-	s.dm.Close(nil)
-	nlog.InfoDepth(1, core.T.String(), "close", s.trname())
+	sdm.rxcbs = nil
+	sdm.rxmu.Unlock()
+
+	sdm.dm.Close(nil)
+	sdm.dm.UnregRecv()
+	sdm.ocmu.Unlock()
+
+	nlog.InfoDepth(1, core.T.String(), "close", sdm.trname())
 	return nil
 }
 
-func (s *SharedDM) RegRecv(xid string, cb transport.RecvObj) {
-	s.rxmu.Lock()
-	debug.Assert(s.rxcbs[xid] == nil)
-	s.rxcbs[xid] = cb
-	s.rxmu.Unlock()
+func (sdm *sharedDM) RegRecv(xid string, cb transport.RecvObj) {
+	sdm.ocmu.Lock()
+	sdm.rxmu.Lock()
+	if !sdm.IsOpen() {
+		sdm.rxmu.Unlock()
+		sdm.ocmu.Unlock()
+		debug.Assert(false, sdm.trname(), " ", "closed")
+		return
+	}
+	debug.Assert(sdm.rxcbs[xid] == nil)
+	sdm.rxcbs[xid] = cb
+	sdm.rxmu.Unlock()
+	sdm.ocmu.Unlock()
 }
 
-func (s *SharedDM) UnregRecv(xid string) {
-	s.rxmu.Lock()
-	delete(s.rxcbs, xid)
-	s.rxmu.Unlock()
+func (sdm *sharedDM) UnregRecv(xid string) {
+	sdm.ocmu.Lock()
+	sdm.rxmu.Lock()
+	if !sdm.IsOpen() {
+		sdm.rxmu.Unlock()
+		sdm.ocmu.Unlock()
+		debug.Assert(false, sdm.trname(), " ", "closed")
+		return
+	}
+	delete(sdm.rxcbs, xid)
+	sdm.rxmu.Unlock()
+	sdm.ocmu.Unlock()
 }
 
-// NOTE (and limitation): use hdr.Opaque exclusively for xaction IDs
-func (s *SharedDM) recv(hdr *transport.ObjHdr, r io.Reader, err error) error {
+func (sdm *sharedDM) recv(hdr *transport.ObjHdr, r io.Reader, err error) error {
 	if err != nil {
 		return err
 	}
 	xid := string(hdr.Opaque)
 	if err := xact.CheckValidUUID(xid); err != nil {
-		return fmt.Errorf("%s: %v", s.trname(), err)
+		return fmt.Errorf("%s: %v", sdm.trname(), err)
 	}
-	s.rxmu.Lock()
-	cb, ok := s.rxcbs[xid]
-	s.rxmu.Unlock()
+
+	sdm.rxmu.Lock()
+	if !sdm.IsOpen() {
+		sdm.rxmu.Unlock()
+		return fmt.Errorf("%s is closed, dropping recv [xid: %s, oname: %s]", sdm.trname(), xid, hdr.ObjName)
+	}
+	cb, ok := sdm.rxcbs[xid]
+	sdm.rxmu.Unlock()
+
 	if !ok {
-		return fmt.Errorf("%s: no registered handler for xact %q", s.trname(), xid)
+		return fmt.Errorf("%s: xid %s not found, dropping recv [oname: %s]", sdm.trname(), xid, hdr.ObjName)
 	}
 	return cb(hdr, r, nil)
 }

Original file line number	Diff line number	Diff line change
`@@ -2042,7 +2042,7 @@ func (h htrun) regTo(url string, psi meta.Snode, tout time.Duration, htext hte`
`2042`	`2042`	`}`
`2043`	`2043`
`2044`	`2044`	`// (fast path: nodes => primary)`
`2045`		`-func (h htrun) fastKalive(smap smapX, timeout time.Duration, ecActive bool) (string /pid/, http.Header, error) {`
	`2045`	`+func (h htrun) fastKalive(smap smapX, timeout time.Duration, ecActive, dmActive bool) (string /pid/, http.Header, error) {`
`2046`	`2046`	`if nlog.Stopping() {`
`2047`	`2047`	`return "", http.Header{}, h.errStopping()`
`2048`	`2048`	`}`
`@@ -2056,10 +2056,15 @@ func (h htrun) fastKalive(smap smapX, timeout time.Duration, ecActive bool) (s`
`2056`	`2056`	`cargs.req = cmn.HreqArgs{Method: http.MethodPost, Base: primaryURL, Path: apc.URLPathCluKalive.Join(h.SID())}`
`2057`	`2057`	`cargs.timeout = timeout`
`2058`	`2058`	`}`
`2059`		`- if ecActive {`
	`2059`	`+ if ecActive \|\| dmActive {`
`2060`	`2060`	`// (target => primary)`
`2061`	`2061`	`hdr := make(http.Header, 1)`
`2062`		`- hdr.Set(apc.HdrActiveEC, "true")`
	`2062`	`+ if ecActive {`
	`2063`	`+ hdr.Set(apc.HdrActiveEC, "true")`
	`2064`	`+ }`
	`2065`	`+ if dmActive {`
	`2066`	`+ hdr.Set(apc.HdrActiveDM, "true")`
	`2067`	`+ }`
`2063`	`2068`	`cargs.req.Header = hdr`
`2064`	`2069`	`}`
`2065`	`2070`
Original file line number	Diff line number	Diff line change
`@@ -21,6 +21,7 @@ import (`
`21`	`21`	`"github.com/NVIDIA/aistore/ec"`
`22`	`22`	`"github.com/NVIDIA/aistore/stats"`
`23`	`23`	`"github.com/NVIDIA/aistore/sys"`
	`24`	`+ "github.com/NVIDIA/aistore/transport/bundle"`
`24`	`25`	`)`
`25`	`26`
`26`	`27`	`const (`
`@@ -131,17 +132,18 @@ func (tkr *talive) cluUptime(now int64) (elapsed time.Duration) {`
`131`	`132`	`}`
`132`	`133`
`133`	`134`	`func (tkr talive) sendKalive(smap smapX, timeout time.Duration, _ int64, fast bool) (pid string, status int, err error) {`
	`135`	`+ t := tkr.t`
`134`	`136`	`if fast {`
`135`	`137`	`// additionally`
`136`		`- interrupted, restarted := tkr.t.interruptedRestarted()`
	`138`	`+ interrupted, restarted := t.interruptedRestarted()`
`137`	`139`	`fast = !interrupted && !restarted`
`138`	`140`	`}`
`139`	`141`	`if fast {`
`140`	`142`	`debug.Assert(ec.ECM != nil)`
`141`		`- pid, _, err = tkr.t.fastKalive(smap, timeout, ec.ECM.IsActive())`
	`143`	`+ pid, _, err = t.fastKalive(smap, timeout, ec.ECM.IsActive(), bundle.SDM.IsOpen())`
`142`	`144`	`return pid, 0, err`
`143`	`145`	`}`
`144`		`- return tkr.t.slowKalive(smap, tkr.t, timeout)`
	`146`	`+ return t.slowKalive(smap, tkr.t, timeout)`
`145`	`147`	`}`
`146`	`148`
`147`	`149`	`func (tkr talive) do(config cmn.Config) (stopped bool) {`
`@@ -196,12 +198,15 @@ func (pkr palive) sendKalive(smap smapX, timeout time.Duration, now int64, fas`
`196`	`198`	`debug.Assert(!smap.isPrimary(pkr.p.si))`
`197`	`199`
`198`	`200`	`if fast {`
`199`		`- pid, hdr, err := pkr.p.fastKalive(smap, timeout, false /ec active/)`
	`201`	`+ pid, hdr, err := pkr.p.fastKalive(smap, timeout, false, false /shared streams/)`
`200`	`202`	`if err == nil {`
`201`	`203`	`// (shared streams; EC streams)`
`202`	`204`	`if pkr.p.ec.isActive(hdr) {`
`203`	`205`	`pkr.p.ec.setActive(now)`
`204`	`206`	`}`
	`207`	`+ if pkr.p.dm.isActive(hdr) {`
	`208`	`+ pkr.p.dm.setActive(now)`
	`209`	`+ }`
`205`	`210`	`}`
`206`	`211`	`return pid, 0, err`
`207`	`212`	`}`