transport: remove per-connection Tx/Rx stats; proto errors vs eof

* transport pkg:
  - remove Stats, RxStats, and GetStats() - simplify
  - collapse handler interface into single struct
  - split IsEOF() into IsOkEOF/IsAnyEOF for more precise semantics
  - distinguish benign shutdown (n=0 + EOF) from protocol errors
  - add unified _loghdr() helper (both Tx and Rx)
  - amend x-moss Rx draining
  - prep for protocol error callbacks (ref 020943, disabled for now)

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

Author: alex-aizman

ais/prxclu.go

@@ -2096,7 +2096,7 @@ func (p *proxy) rmNodeFinal(msg *apc.ActMsg, si *meta.Snode, ctx *smapModifier)
 		emsg := fmt.Sprintf("%s: (%s %s) final: %v - proceeding anyway...", p, msg, sname, err)
 		switch msg.Action {
 		case apc.ActShutdownNode, apc.ActDecommissionNode: // expecting EOF
-			if !cos.IsEOF(err) {
+			if !cos.IsAnyEOF(err) {
 				nlog.Errorln(emsg)
 			}
 		case apc.ActRmNodeUnsafe:

api/cluster.go

@@ -543,7 +543,7 @@ func DecommissionCluster(bp BaseParams, rmUserData bool) error {
 	}
 	err := reqParams.DoRequest()
 	FreeRp(reqParams)
-	if err != nil && cos.IsEOF(err) {
+	if err != nil && cos.IsAnyEOF(err) {
 		err = nil
 	}
 	return err

cmn/cos/err.go

@@ -263,7 +263,7 @@ func IsUnreachable(err error, status int) bool {
 		errors.Is(err, context.DeadlineExceeded) ||
 		status == http.StatusRequestTimeout ||
 		status == http.StatusServiceUnavailable ||
-		IsEOF(err) ||
+		IsAnyEOF(err) ||
 		status == http.StatusBadGateway
 }
 

cmn/cos/ioutils.go

@@ -50,11 +50,16 @@ func ReadAll(r io.Reader) ([]byte, error) {
 
 // including "unexpecting EOF" to accommodate unsized streaming and
 // early termination of the other side (prior to sending the first byte)
-func IsEOF(err error) bool {
+func IsAnyEOF(err error) bool {
 	return err == io.EOF || err == io.ErrUnexpectedEOF ||
 		errors.Is(err, io.ErrUnexpectedEOF) || errors.Is(err, io.EOF)
 }
 
+// graceful
+func IsOkEOF(err error) bool {
+	return err == io.EOF || errors.Is(err, io.EOF)
+}
+
 // ExpandPath replaces common abbreviations in file path (eg. `~` with absolute
 // path to the current user home directory) and cleans the path.
 func ExpandPath(path string) string {

ec/manager.go

@@ -193,7 +193,7 @@ func (mgr *Manager) recvRequest(hdr *transport.ObjHdr, objReader io.Reader, err
 	// the body must be drained to avoid errors
 	if hdr.ObjAttrs.Size != 0 {
 		n, err := io.Copy(io.Discard, objReader)
-		if err != nil && !cos.IsEOF(err) {
+		if err != nil && !cos.IsOkEOF(err) {
 			nlog.Errorf("failed to read request body: %v", err)
 			return err
 		}

transport/api.go

@@ -17,7 +17,6 @@ import (
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
 	"github.com/NVIDIA/aistore/core"
-	"github.com/NVIDIA/aistore/hk"
 	"github.com/NVIDIA/aistore/memsys"
 )
 
@@ -115,12 +114,6 @@ type (
 	RecvObj func(hdr *ObjHdr, objReader io.Reader, err error) error
 )
 
-// receive-side session stats indexed by session ID (see recv.go for "uid")
-// optional, currently tests only
-type (
-	RxStats map[uint64]*Stats
-)
-
 // shared data mover (SDM)
 type (
 	Receiver interface {
@@ -204,16 +197,8 @@ func (s *Stream) Fin() {
 // receive-side API //
 //////////////////////
 
-func Handle(trname string, rxObj RecvObj, withStats ...bool) error {
-	var h handler
-	if len(withStats) > 0 && withStats[0] {
-		hkName := ObjURLPath(trname)
-		hex := &hdlExtra{hdl: hdl{trname: trname, rxObj: rxObj}, hkName: hkName}
-		hk.Reg(hkName+hk.NameSuffix, hex.cleanup, sessionIsOld)
-		h = hex
-	} else {
-		h = &hdl{trname: trname, rxObj: rxObj}
-	}
+func Handle(trname string, rxObj RecvObj) error {
+	h := &handler{trname: trname, rxObj: rxObj}
 	return oput(trname, h)
 }
 
@@ -231,17 +216,3 @@ func _urlPath(endp, trname string) string {
 	}
 	return cos.JoinW0(apc.Version, endp, trname)
 }
-
-func GetRxStats() (netstats map[string]RxStats) {
-	netstats = make(map[string]RxStats)
-	for i, hmap := range hmaps {
-		hmtxs[i].Lock()
-		for trname, h := range hmap {
-			if s := h.getStats(); s != nil {
-				netstats[trname] = s
-			}
-		}
-		hmtxs[i].Unlock()
-	}
-	return
-}

transport/base.go

@@ -26,7 +26,6 @@ import (
 	"github.com/NVIDIA/aistore/cmn/cos"
 	"github.com/NVIDIA/aistore/cmn/debug"
 	"github.com/NVIDIA/aistore/cmn/nlog"
-	"github.com/NVIDIA/aistore/core"
 )
 
 // stream TCP/HTTP session: inactive <=> active transitions
@@ -88,9 +87,9 @@ type (
 			reason string
 			mu     sync.Mutex
 			done   atomic.Bool
+			yelp   atomic.Bool
 		}
-		stats Stats // stream stats (send side - compare with rxStats)
-		time  struct {
+		time struct {
 			idleTeardown time.Duration // idle timeout
 			inSend       atomic.Bool   // true upon Send() or Read() - info for Collector to delay cleanup
 			ticks        int           // num 1s ticks until idle timeout
@@ -138,6 +137,7 @@ func newBase(client Client, dstURL, dstID string, extra *Extra) (s *streamBase)
 		buf, _ := g.mm.AllocSize(int64(extra.SizePDU))
 		s.pdu = newSendPDU(buf)
 	}
+	// idle time
 	if extra.IdleTeardown > 0 {
 		s.time.idleTeardown = extra.IdleTeardown
 	} else {
@@ -146,35 +146,14 @@ func newBase(client Client, dstURL, dstID string, extra *Extra) (s *streamBase)
 	debug.Assert(s.time.idleTeardown >= dfltTick, s.time.idleTeardown, " vs ", dfltTick)
 	s.time.ticks = int(s.time.idleTeardown / dfltTick)
 
-	s._loghdr(sid, dstID, extra)
-
+	s.loghdr = _loghdr(s.trname, sid, dstID, true, extra.Compressed())
 	s.maxhdr, _ = g.mm.AllocSize(_sizeHdr(extra.Config, int64(extra.MaxHdrSize)))
 
-	s.sessST.Store(inactive) // initiate HTTP session upon the first arrival
+	// fsm: initiate HTTP session upon the first arrival
+	s.sessST.Store(inactive)
 	return s
 }
 
-func (s *streamBase) _loghdr(sid, dstID string, extra *Extra) {
-	var (
-		sb strings.Builder
-		l  = 2 + len(s.trname) + len(sid) + 32 + len(dstID)
-	)
-	sb.Grow(l)
-
-	sb.WriteString("s-")
-	sb.WriteString(s.trname)
-	sb.WriteString(sid)
-	sb.WriteByte('[')
-	sb.WriteString(core.T.SID()) // (consider adding back session ID, as in: [node-ID/110])
-
-	extra.Lid(&sb) // + compressed
-
-	sb.WriteString("]=>")
-	sb.WriteString(dstID)
-
-	s.loghdr = sb.String() // looks like: s-<trname><sid>[<sender-id>]=><dest-id>
-}
-
 // (used on the receive side as well)
 func _sizeHdr(config *cmn.Config, size int64) int64 {
 	if size != 0 {
@@ -193,7 +172,9 @@ func (s *streamBase) startSend(streamable fmt.Stringer) (err error) {
 		// slow path
 		reason, errT := s.TermInfo()
 		err = cmn.NewErrStreamTerminated(s.String(), errT, reason, "dropping "+streamable.String())
-		nlog.Errorln(err)
+		if s.term.yelp.CAS(false, true) { // only once
+			nlog.Errorln(err)
+		}
 		return
 	}
 
@@ -230,15 +211,6 @@ func (s *streamBase) TermInfo() (reason string, err error) {
 	return
 }
 
-func (s *streamBase) GetStats() (stats Stats) {
-	// byte-num transfer stats
-	stats.Num.Store(s.stats.Num.Load())
-	stats.Offset.Store(s.stats.Offset.Load())
-	stats.Size.Store(s.stats.Size.Load())
-	stats.CompressedSize.Store(s.stats.CompressedSize.Load())
-	return
-}
-
 func (s *streamBase) isNextReq() (reason string) {
 	for {
 		select {
@@ -267,7 +239,7 @@ func (s *streamBase) isNextReq() (reason string) {
 func (s *streamBase) deactivate() (n int, err error) {
 	err = io.EOF
 	if cmn.Rom.V(5, cos.ModTransport) {
-		nlog.Infoln(s.String(), "connection teardown: [", s.numCur, s.stats.Num.Load(), "]")
+		nlog.Infoln(s.String(), "connection teardown: [", s.numCur, "]")
 	}
 	return
 }
@@ -384,14 +356,6 @@ func (extra *Extra) Compressed() bool {
 	return extra.Compression != "" && extra.Compression != apc.CompressNever
 }
 
-func (extra *Extra) Lid(sb *strings.Builder) {
-	if extra.Compressed() {
-		sb.WriteByte('[')
-		sb.WriteString(extra.Config.Transport.LZ4BlockMaxSize.String())
-		sb.WriteByte(']')
-	}
-}
-
 //
 // misc
 //
@@ -406,6 +370,25 @@ func dryrun() (dryrun bool) {
 	return
 }
 
+func _loghdr(trname, from, to string, transmit, compressed bool) string {
+	var (
+		sb strings.Builder
+		l  = len(trname) + len(from) + len(to) + 16
+	)
+	sb.Grow(l)
+
+	sb.WriteString(trname)
+	sb.WriteByte('[')
+	if compressed {
+		sb.WriteString("(z)")
+	}
+	sb.WriteString(from)
+	sb.WriteString(cos.Ternary(transmit, "=>", "<="))
+	sb.WriteString(to)
+	sb.WriteByte(']')
+	return sb.String()
+}
+
 //////////
 // rtry //
 //////////

transport/bundle/bundle.go

@@ -51,7 +51,6 @@ type (
 		multiplier   int // optionally: multiple streams per destination (round-robin)
 		manualResync bool
 	}
-	Stats map[string]*transport.Stats // by DaemonID
 
 	Args struct {
 		Extra        *transport.Extra // additional parameters
@@ -103,7 +102,7 @@ func New(cl transport.Client, args Args) (sb *Streams) {
 	sb.Resync()
 	sb.smaplock.Unlock()
 
-	sb._lid()
+	sb.lid = sb._lid()
 	nlog.Infoln("open", sb.lid)
 
 	// for auto-resync, register this stream-bundle as Smap listener
@@ -114,10 +113,14 @@ func New(cl transport.Client, args Args) (sb *Streams) {
 	return sb
 }
 
-func (sb *Streams) _lid() {
+func (sb *Streams) String() string { return sb.lid }
+
+// (compare w/ transport._loghdr)
+func (sb *Streams) _lid() string {
 	var s strings.Builder
+	s.Grow(20 + len(sb.trname))
 
-	s.WriteString("sb-[")
+	s.WriteString(cos.Ternary(sb.extra.Compressed(), "sb(z)-[", "sb-["))
 	s.WriteString(core.T.SID())
 	if sb.network != cmn.NetIntraData {
 		s.WriteByte('-')
@@ -128,11 +131,9 @@ func (sb *Streams) _lid() {
 	s.WriteByte('-')
 	s.WriteString(sb.trname)
 
-	sb.extra.Lid(&s)
-
 	s.WriteByte(']')
 
-	sb.lid = s.String() // approx. "sb[%s-%s-%s...]"
+	return s.String()
 }
 
 // Close closes all contained streams and unregisters the bundle from Smap listeners;
@@ -232,7 +233,6 @@ func _doCmpl(obj *transport.Obj, roc cos.ReadOpenCloser, err error) {
 	}
 }
 
-func (sb *Streams) String() string   { return sb.lid }
 func (sb *Streams) Smap() *meta.Smap { return sb.smap }
 
 // keep streams to => (clustered nodes as per rxNodeType) in sync at all times
@@ -247,17 +247,6 @@ func (sb *Streams) ListenSmapChanged() {
 	sb.smaplock.Unlock()
 }
 
-func (sb *Streams) GetStats() Stats {
-	streams := sb.get()
-	stats := make(Stats, len(streams))
-	for id, robin := range streams {
-		s := robin.stsdest[0]
-		tstat := s.GetStats()
-		stats[id] = &tstat
-	}
-	return stats
-}
-
 //
 // private methods
 //

transport/handlers.go

@@ -23,7 +23,7 @@ const (
 	numOld = 32
 )
 
-type hmap map[string]handler
+type hmap map[string]*handler
 
 type (
 	errTrname struct {
@@ -52,7 +52,7 @@ func _idx(trname string) byte {
 	return byte(hash & mskHmaps)
 }
 
-func oget(trname string) (h handler, err error) {
+func oget(trname string) (h *handler, err error) {
 	i := _idx(trname)
 	hmtxs[i].Lock()
 	hmap := hmaps[i]
@@ -77,7 +77,7 @@ func _lookup(trname string) error {
 	return &errUnknownTrname{errTrname{trname}}
 }
 
-func oput(trname string, h handler) (err error) {
+func oput(trname string, h *handler) (err error) {
 	i := _idx(trname)
 	hmtxs[i].Lock()
 	hmap := hmaps[i]
@@ -95,7 +95,7 @@ func odel(trname string) (err error) {
 	i := _idx(trname)
 	hmtxs[i].Lock()
 	hmap := hmaps[i]
-	h, ok := hmap[trname]
+	_, ok := hmap[trname]
 	if !ok {
 		hmtxs[i].Unlock()
 		return &errAlreadyRemovedTrname{errTrname{trname}}
@@ -104,8 +104,6 @@ func odel(trname string) (err error) {
 	delete(hmap, trname)
 	hmtxs[i].Unlock()
 
-	h.unreg()
-
 	oldMtx.Lock()
 	old[oldIdx] = trname
 	oldIdx++