remove shared-DM from keep-alive piggy-back; expire locally

* part-one commit of the two-part sequence to fix:
  - #316
* remove SDM from fast-keepalive (=> primary) and stream-toggle (by primary)
* rewrite the logic to track local SDM activity
  - in particular,use both `last` time _and_ in-flight ref-count
----
* separately, node-state alerts: classify keep-alive errors as a _warning_

Signed-off-by: Alex Aizman <alex.aizman@gmail.com>

Author: alex-aizman

ais/htrun.go

@@ -2385,7 +2385,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, dmActive bool) (string /*pid*/, http.Header, error) {
+func (h *htrun) fastKalive(smap *smapX, timeout time.Duration, ecActive bool) (string /*pid*/, http.Header, error) {
 	if nlog.Stopping() {
 		return "", http.Header{}, h.errStopping()
 	}
@@ -2399,15 +2399,12 @@ func (h *htrun) fastKalive(smap *smapX, timeout time.Duration, ecActive, dmActiv
 		cargs.req = cmn.HreqArgs{Method: http.MethodPost, Base: primaryURL, Path: apc.URLPathCluKalive.Join(h.SID())}
 		cargs.timeout = timeout
 	}
-	if ecActive || dmActive {
+	if ecActive {
 		// (target => primary)
 		hdr := make(http.Header, 1)
 		if ecActive {
 			hdr.Set(apc.HdrActiveEC, "true")
 		}
-		if dmActive {
-			hdr.Set(apc.HdrActiveDM, "true")
-		}
 		cargs.req.Header = hdr
 	}
 

ais/kalive.go

@@ -21,7 +21,6 @@ import (
 	"github.com/NVIDIA/aistore/ec"
 	"github.com/NVIDIA/aistore/stats"
 	"github.com/NVIDIA/aistore/sys"
-	"github.com/NVIDIA/aistore/transport/bundle"
 )
 
 const (
@@ -140,7 +139,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())
 		return pid, 0, err
 	}
 	return t.slowKalive(smap, tkr.t, timeout)
@@ -198,18 +197,12 @@ func (pkr *palive) sendKalive(smap *smapX, timeout time.Duration, now int64, fas
 	debug.Assert(!smap.isPrimary(pkr.p.si))
 
 	if fast {
-		last, dmActive := pkr.p.dm.nonpResetActive()
-		pid, hdr, err := pkr.p.fastKalive(smap, timeout, false, dmActive /*shared streams*/)
+		pid, hdr, err := pkr.p.fastKalive(smap, timeout, false)
 		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)
-			}
-		} else if dmActive {
-			pkr.p.dm.nonpUndo(last) // (unlikely)
 		}
 		return pid, 0, err
 	}

ais/ml.go

@@ -16,7 +16,6 @@ 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"
@@ -217,11 +216,6 @@ func (p *proxy) httpmlget(w http.ResponseWriter, r *http.Request) {
 		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: redirect user's GET => DT
 	r.URL.Path = path
 	started := time.Now()

ais/proxy.go

@@ -57,7 +57,6 @@ type (
 		lstca      lstca
 		rproxy     reverseProxy
 		ec         ecToggle
-		dm         dmToggle
 
 		htrun // common w/ target
 
@@ -128,7 +127,6 @@ func (p *proxy) init(config *cmn.Config) {
 
 	// shared streams
 	p.ec.init()
-	p.dm.init()
 }
 
 func initPID(config *cmn.Config) string {

ais/prxclu.go

@@ -674,11 +674,9 @@ func (p *proxy) fastKaliveRsp(w http.ResponseWriter, r *http.Request, smap *smap
 				if si.IsTarget() {
 					// (target kalive => primary)
 					p.ec.recvKalive(p, r.Header, now, p.ec.timeout())
-					p.dm.recvKalive(p, r.Header, now, p.dm.timeout())
 				} else {
 					// (primary kalive response => non-primary)
 					p.ec.respKalive(w.Header(), now, p.ec.timeout())
-					p.dm.respKalive(w.Header(), now, p.dm.timeout())
 				}
 				return
 			}

ais/prxec.go

@@ -24,9 +24,7 @@ func (p *proxy) ecHandler(w http.ResponseWriter, r *http.Request) {
 
 // +gen:endpoint POST /v1/ec/open-ec-streams
 // +gen:endpoint POST /v1/ec/close-ec-streams
-// +gen:endpoint POST /v1/ec/open-shared-dm
-// +gen:endpoint POST /v1/ec/close-shared-dm
-// Enable or disable erasure coding and shared data management
+// Enable or disable erasure coding
 func (p *proxy) httpecpost(w http.ResponseWriter, r *http.Request) {
 	apiItems, err := p.parseURL(w, r, apc.URLPathEC.L, 1, false)
 	if err != nil {
@@ -38,13 +36,6 @@ func (p *proxy) httpecpost(w http.ResponseWriter, r *http.Request) {
 		p.ec.setActive(mono.NanoTime())
 	case apc.ActCloseEC:
 		p.ec.setActive(0)
-
-	// TODO: refactor as post-toggle-shared-streams
-	case apc.ActOpenSDM:
-		p.dm.setActive(mono.NanoTime())
-	case apc.ActCloseSDM:
-		p.dm.setActive(0)
-
 	default:
 		p.writeErr(w, r, errActEc(action))
 	}

ais/streams_toggle.go

@@ -40,9 +40,6 @@ type (
 	ecToggle struct {
 		streamsToggle
 	}
-	dmToggle struct {
-		streamsToggle
-	}
 )
 
 func (f *ecToggle) init() {
@@ -53,41 +50,12 @@ func (f *ecToggle) init() {
 
 func (*ecToggle) timeout() time.Duration { return cmn.Rom.EcStreams() }
 
-func (f *dmToggle) init() {
-	f.hdrActive = apc.HdrActiveDM
-	f.actOn = apc.ActOpenSDM
-	f.actOff = apc.ActCloseSDM
-}
-
-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(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) {
 	if _, ok := hdr[f.hdrActive]; ok {

ais/tgtec.go

@@ -20,7 +20,6 @@ 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"
 )
 
@@ -75,9 +74,8 @@ func (t *target) sendECMetafile(w http.ResponseWriter, r *http.Request, bck *met
 
 func (t *target) httpecpost(w http.ResponseWriter, r *http.Request) {
 	const (
-		hknameEC  = apc.ActCloseEC + hk.NameSuffix
-		hknameSDM = apc.ActCloseSDM + hk.NameSuffix
-		postpone  = time.Minute
+		hknameEC = apc.ActCloseEC + hk.NameSuffix
+		postpone = time.Minute
 	)
 	items, err := t.parseURL(w, r, apc.URLPathEC.L, 1, false)
 	if err != nil {
@@ -136,8 +134,6 @@ func (t *target) httpecpost(w http.ResponseWriter, r *http.Request) {
 			core.FreeLOM(lom)
 		}
 
-	// TODO [minor]: consider reusing ais/streams_toggle or otherwise reducing copy/paste
-
 	case apc.ActOpenEC:
 		hk.UnregIf(hknameEC, closeEc) // just in case, a no-op most of the time
 		ec.ECM.OpenStreams(false /*with refc*/)
@@ -152,23 +148,6 @@ func (t *target) httpecpost(w http.ResponseWriter, r *http.Request) {
 		nlog.Infoln(t.String(), "hk-postpone", action)
 		hk.Reg(hknameEC, closeEc, postpone)
 
-	case apc.ActOpenSDM:
-		// NOTE:
-		// Unlike EC, SDM is opened on-demand by the GetBatch/x-moss 3-phase protocol.
-		// This action exists only to cancel a pending idle-close (HK) and to allow
-		// the primary to track cluster-wide SDM activity time.
-		hk.UnregIf(hknameSDM, closeSDM)
-	case apc.ActCloseSDM:
-		if !t.ensureIntraControl(w, r, true /* from primary */) {
-			return
-		}
-		if bundle.SDM.IsActive() {
-			t.writeErr(w, r, _errOff(bundle.SDMName))
-			return
-		}
-		nlog.Infoln(t.String(), "hk-postpone", action)
-		hk.Reg(hknameSDM, closeSDM, postpone)
-
 	default:
 		t.writeErr(w, r, errActEc(action))
 	}
@@ -185,15 +164,6 @@ func closeEc(int64) time.Duration {
 	return hk.UnregInterval
 }
 
-func closeSDM(int64) time.Duration {
-	if bundle.SDM.IsActive() {
-		nlog.Warningln("hk-cb:", _errOff(bundle.SDMName))
-	} else if err := bundle.SDM.Close(); err != nil {
-		nlog.Errorln(err)
-	}
-	return hk.UnregInterval
-}
-
 func errActEc(act string) error {
 	return fmt.Errorf(fmtErrInvaldAction, act, []string{apc.ActOpenEC, apc.ActCloseEC})
 }

api/apc/actmsg.go

@@ -156,11 +156,6 @@ const (
 	NodeDecommission = "decommission"
 )
 
-const (
-	ActOpenSDM  = "open-shared-dm"
-	ActCloseSDM = "close-shared-dm"
-)
-
 const (
 	ActOpenEC    = "open-ec-streams"
 	ActCloseEC   = "close-ec-streams"

api/apc/headers.go

@@ -139,7 +139,6 @@ const (
 
 	// shared streams
 	HdrActiveEC = aisPrefix + "Ec"
-	HdrActiveDM = aisPrefix + "Dm"
 
 	// (ais/psetforce; advanced use)
 	HdrReadyToJoinClu = aisPrefix + "Ready-Join-Clu"