get-batch: handle stream breakages (ErrSBR); per-request polling and cleanup

* propagate ErrSBR from shared-DM recv() to x-moss
* add transport.ErrSBR timestamps and helpers
  - Since(), IsErrSBR(), AsErrSBR()
* maintain versioned ErrSBR map (r.sbrs) with per-WI (per get-batch) cache
* add checkSBR() to detect recent SBRs for the current sender
* extend waitFlushRx() loop to poll periodically and abort affected WIDs
* reuse timers in waitAnyRx to reduce allocations
  - work item cleanup: stop and drain
* simplify wakeup semantics; assembler thread polls and reacts on its own
* add bounded ErrSBR window (hardcoded 4s) and polling interval (1/10th timeout)
* minor log and comment updates (see "ref 032011" marker)

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

Author: alex-aizman

ais/ml.go

@@ -390,9 +390,11 @@ func (ctx *mossCtx) phase2(w http.ResponseWriter, r *http.Request, smap *smapX,
 		return
 	}
 	if err := xmoss.Send(ctx.req, &smap.Smap, tsi, ctx.wid, usingPrev); err != nil {
-		// NOTE: not aborting x-moss on a single wid failure
+		//
+		// not aborting x-moss on a single wid failure; silent log
+		//
 		xmoss.AddErr(fmt.Errorf("send wid=%s: %v", ctx.wid, err), 4, cos.ModXs)
-		t.writeErr(w, r, err)
+		t.writeErr(w, r, err, 0, Silent)
 	}
 }
 

transport/bundle/shared_dm.go

@@ -115,6 +115,7 @@ func (sdm *sharedDM) Open() error {
 	return nil
 }
 
+// TODO: prevent a) too-frequent reconnects and/or b) too-many-during-stream's-lifetime :TODO
 func (sdm *sharedDM) reconnect(dstID string, err error) {
 	if !sdm.isOpen() {
 		return
@@ -209,6 +210,26 @@ func (sdm *sharedDM) Bcast(obj *transport.Obj, roc cos.ReadOpenCloser) error {
 
 func (sdm *sharedDM) recv(hdr *transport.ObjHdr, r io.Reader, err error) error {
 	if err != nil {
+		//
+		// `transport.ErrSBR` => all reg-ed receivers
+		//
+		if e := transport.AsErrSBR(err); e != nil {
+			// clone to call outside rlock
+			sdm.rxmu.RLock()
+			receivers := make([]transport.Receiver, 0, len(sdm.receivers))
+			for xid := range sdm.receivers {
+				receivers = append(receivers, sdm.receivers[xid].rx)
+			}
+			sdm.rxmu.RUnlock()
+
+			for _, rx := range receivers {
+				_ = rx.RecvObj(hdr, nil, e)
+			}
+
+			clear(receivers)
+			return nil
+		}
+
 		return err
 	}
 	xid := hdr.Demux

transport/err.go

@@ -6,8 +6,10 @@ package transport
 
 import (
 	"strings"
+	"time"
 
 	"github.com/NVIDIA/aistore/cmn/debug"
+	"github.com/NVIDIA/aistore/cmn/mono"
 )
 
 // ErrSBR codes (receive-side stream breakage)
@@ -43,6 +45,7 @@ type (
 		sid    string // sender node ID
 		code   string
 		ctx    string
+		ts     int64
 	}
 )
 
@@ -100,8 +103,14 @@ func (e *ErrSBR) Error() string {
 	return sb.String()
 }
 
-func (e *ErrSBR) Unwrap() error { return e.err }
-func (e *ErrSBR) SID() string   { return e.sid }
+func (e *ErrSBR) Unwrap() error        { return e.err }
+func (e *ErrSBR) SID() string          { return e.sid }
+func (e *ErrSBR) Since() time.Duration { return mono.Since(e.ts) }
+
+func IsErrSBR(err error) bool {
+	_, ok := err.(*ErrSBR)
+	return ok
+}
 
 func AsErrSBR(err error) *ErrSBR {
 	if e, ok := err.(*ErrSBR); ok {

transport/recv.go

@@ -18,6 +18,7 @@ import (
 	"github.com/NVIDIA/aistore/cmn/atomic"
 	"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/memsys"
@@ -299,6 +300,7 @@ func (it *iterator) newErr(err error, code, ctx string) error {
 		sid:    it.sid,
 		code:   code,
 		ctx:    ctx,
+		ts:     mono.NanoTime(),
 	}
 }