transport: revise retry logic; add stream reconnect (major)

* add `_shouldRetry()` to only retry on write timeouts
* complete failed sends immediately before retry check
* wire `TermedCB()` via shared data mover's reconnect callback
* transport/recv errors will now wrap `io.ErrUnexpectedEOF`
  (something that libraries may in fact recognize)

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

Author: alex-aizman

transport/api.go

@@ -67,7 +67,8 @@ type (
 	SentCB func(*ObjHdr, io.ReadCloser, any, error)
 
 	// scope: stream
-	TermedCB func(error)
+	// flow: connection dead => terminate => TermedCB => [reconnect() => fresh stream to same peer]
+	TermedCB func(dstID string, err error)
 
 	// usage and scope:
 	// - entire stream's lifetime (all Send() calls)

transport/base.go

@@ -6,15 +6,18 @@
 package transport
 
 import (
+	"errors"
 	"fmt"
 	"io"
+	"net"
 	"net/url"
 	"os"
 	"path"
 	"runtime"
 	"strconv"
 	"strings"
 	"sync"
+	"syscall"
 	"time"
 
 	"github.com/NVIDIA/aistore/api/apc"
@@ -290,13 +293,21 @@ func (s *streamBase) sendLoop(config *cmn.Config, dryrun bool) {
 					retry = nil
 				}
 			} else {
-				retriable := cos.IsRetriableConnErr(err)
-				if retriable && retry == nil {
+				// the current send failed - complete right away
+				s.streamer.errCmpl(err)
+
+				if !_shouldRetry(err) {
+					if cmn.Rom.V(4, cos.ModTransport) {
+						nlog.Errorln(s.String(), "not retriable:", err)
+					}
+					reason = reasonError
+					break
+				}
+				if retry == nil {
 					retry = newRtry(config, s.String())
 				}
-				if !retriable || retry.timeout(err) {
+				if retry.timeout(err) {
 					reason = reasonError
-					s.streamer.errCmpl(err)
 					break
 				}
 
@@ -338,7 +349,7 @@ func (s *streamBase) sendLoop(config *cmn.Config, dryrun bool) {
 	// notify parent if defined
 	if reason != reasonStopped {
 		if s.parent != nil && s.parent.TermedCB != nil {
-			s.parent.TermedCB(errExt)
+			s.parent.TermedCB(s.dstID, errExt)
 		}
 	}
 
@@ -350,6 +361,15 @@ func (s *streamBase) sendLoop(config *cmn.Config, dryrun bool) {
 	}
 }
 
+// only for timeouts on *in-flight writes*
+func _shouldRetry(err error) bool {
+	var nerr net.Error
+	if errors.As(err, &nerr) && nerr.Timeout() {
+		return true
+	}
+	return errors.Is(err, syscall.ETIMEDOUT)
+}
+
 func (s *streamBase) yelp(err error) {
 	nlog.WarningDepth(1, "Error:", s.String(), "[", err, "]")
 }

transport/bundle/bundle.go

@@ -440,10 +440,8 @@ func mdiff(oldMaps, newMaps []meta.NodeMap) (added, removed meta.NodeMap) {
 	return
 }
 
-// replace streams to a single peer (dstID) in the bundle:
-// - optional parentOverride
-// - otherwise, use sb.extra.Parent as-is
-func (sb *Streams) ReopenPeerStream(dstID string, parentOverride *transport.Parent) error {
+// renew stream (or streams) to a given peer
+func (sb *Streams) ReopenPeerStream(dstID string) error {
 	// 1) validate
 	old := sb.get()
 	orobin, ok := old[dstID]
@@ -467,13 +465,10 @@ func (sb *Streams) ReopenPeerStream(dstID string, parentOverride *transport.Pare
 	}
 	dstURL := si.URL(sb.network) + transport.ObjURLPath(sb.trname)
 
-	// 2) build new `robin` with same multiplier
+	// 2) build new `robin` (same multiplier; consider setting nrobin.i)
 	nrobin := &robin{stsdest: make(stsdest, len(orobin.stsdest))}
 	for k := range nrobin.stsdest {
-		extra := sb.extra // copy struct by value
-		if parentOverride != nil {
-			extra.Parent = parentOverride
-		}
+		extra := sb.extra // by value
 		ns := transport.NewObjStream(sb.client, dstURL, dstID, &extra)
 		nrobin.stsdest[k] = ns
 	}
@@ -483,10 +478,10 @@ func (sb *Streams) ReopenPeerStream(dstID string, parentOverride *transport.Pare
 	}
 	nbundle[dstID] = nrobin
 
-	// 3) set new CoW instance immediately
+	// 3) switch over
 	sb.streams.Store(&nbundle)
 
-	// 4) stop old streams (may take a few millis)
+	// 4) stop old streams async
 	for _, os := range orobin.stsdest {
 		if !os.IsTerminated() {
 			os.Stop() // via stopCh

transport/bundle/shared_dm.go

@@ -103,6 +103,9 @@ func (sdm *sharedDM) Open() error {
 		debug.AssertNoErr(err)
 		return err
 	}
+	sdm.dm.parent = &transport.Parent{
+		TermedCB: sdm.reconnect,
+	}
 	sdm.dm.Open()
 	sdm.ocmu.Unlock()
 
@@ -112,6 +115,18 @@ func (sdm *sharedDM) Open() error {
 	return nil
 }
 
+func (sdm *sharedDM) reconnect(dstID string, err error) {
+	if !sdm.isOpen() {
+		return
+	}
+	if e := sdm.dm.data.streams.ReopenPeerStream(dstID); e != nil {
+		err = fmt.Errorf("%v => %v", err, e)
+		nlog.Errorf("%s: failed to reconnect to %s: %v", sdm.trname(), dstID, err)
+		sdm.Close(err)
+	}
+	nlog.Warningln(core.T.String(), "reconnect", sdm.trname(), "=>", dstID)
+}
+
 func (sdm *sharedDM) housekeep(now int64) time.Duration {
 	if !sdm.isOpen() {
 		return hk.UnregInterval
@@ -125,15 +140,20 @@ func (sdm *sharedDM) housekeep(now int64) time.Duration {
 }
 
 // nothing running + cmn.SharedStreamsDflt (10m) inactivity
-func (sdm *sharedDM) Close() error {
+func (sdm *sharedDM) Close(err ...error) error {
 	sdm.ocmu.Lock()
 	sdm.rxmu.Lock()
 
-	if xid := sdm.getActive(); xid != "" {
-		sdm.rxmu.Unlock()
-		sdm.ocmu.Unlock()
-		debug.Assert(cos.IsValidUUID(xid), xid)
-		return fmt.Errorf("cannot close %s: xid %s is still active (num: %d)", sdm.trname(), xid, len(sdm.receivers))
+	xid := sdm.getActive()
+	if xid != "" {
+		msg := fmt.Sprintf("xid %s is still active (num: %d)", xid, len(sdm.receivers))
+		if len(err) == 0 {
+			sdm.rxmu.Unlock()
+			sdm.ocmu.Unlock()
+			debug.Assert(cos.IsValidUUID(xid), xid)
+			return fmt.Errorf("cannot close %s: %s", sdm.trname(), msg)
+		}
+		nlog.Errorln(sdm.trname(), "closing despite", msg, "[", err[0], "]")
 	}
 
 	sdm.dm.Close(nil)

transport/recv.go

@@ -6,6 +6,7 @@
 package transport
 
 import (
+	"errors"
 	"fmt"
 	"io"
 	"math"
@@ -318,10 +319,10 @@ func eofOK(err error) error {
 func (it *iterator) nextProtoHdr(loghdr string) (int, uint64, error) {
 	n, err := it.Read(it.hbuf[:sizeProtoHdr])
 	if n < sizeProtoHdr {
-		if err == nil {
-			err = fmt.Errorf("sbr3 %s: failed to receive proto hdr (n=%d)", loghdr, n)
+		if err == nil || errors.Is(err, io.EOF) {
+			err = io.ErrUnexpectedEOF
 		}
-		return 0, 0, err
+		return 0, 0, fmt.Errorf("sbr3 %s: failed to receive proto hdr (n=%d): %w", loghdr, n, err)
 	}
 	// extract and validate hlen
 	return extProtoHdr(it.hbuf, loghdr)
@@ -349,7 +350,10 @@ func (it *iterator) nextObj(loghdr string, hlen int) (*objReader, error) {
 			}
 		}
 		if n < hlen {
-			return nil, fmt.Errorf("sbr4 %s: failed to receive obj hdr (%d < %d)", loghdr, n, hlen)
+			if err == nil || errors.Is(err, io.EOF) {
+				err = io.ErrUnexpectedEOF
+			}
+			return nil, fmt.Errorf("sbr4 %s: failed to receive obj hdr (%d < %d): %w", loghdr, n, hlen, err)
 		}
 	}
 	hdr := ExtObjHeader(it.hbuf, hlen)
@@ -371,23 +375,34 @@ func (obj *objReader) Read(b []byte) (n int, err error) {
 		return obj.readPDU(b)
 	}
 	debug.Assert(obj.Size() >= 0)
+
 	rem := obj.Size() - obj.off
-	if rem < int64(len(b)) {
+	if rem < int64(len(b)) && rem >= 0 {
 		b = b[:int(rem)]
 	}
+
 	n, err = obj.body.Read(b)
 	obj.off += int64(n) // NOTE: `GORACE` complaining here can be safely ignored
+
 	switch err {
 	case nil:
 		if obj.off >= obj.Size() {
+			// ok (w/ EOF to the caller)
 			err = io.EOF
 		}
 	case io.EOF:
+		// premature EOF: expected size not reached
 		if obj.off != obj.Size() {
-			err = fmt.Errorf("sbr6 %s: premature eof %d != %s, err %w", obj.loghdr, obj.off, obj, err)
+			err = fmt.Errorf("sbr6 %s: premature eof %d != %s: %w", obj.loghdr, obj.off, obj, io.ErrUnexpectedEOF)
 		}
 	default:
-		err = fmt.Errorf("sbr7 %s: off %d, obj %s, err %w", obj.loghdr, obj.off, obj, err)
+		// we haven't consumed the header-specified size
+		if obj.off < obj.Size() {
+			err = fmt.Errorf("sbr7 %s: off %d, obj %s, cause: %v: %w", obj.loghdr, obj.off, obj, err, io.ErrUnexpectedEOF)
+		} else {
+			// read the full payload; just annotate
+			err = fmt.Errorf("sbr7 %s: off %d, obj %s, err %w", obj.loghdr, obj.off, obj, err)
+		}
 	}
 	return n, err
 }
@@ -433,7 +448,7 @@ func (obj *objReader) readPDU(b []byte) (n int, err error) {
 			if obj.IsUnsized() {
 				obj.hdr.ObjAttrs.Size = obj.off
 			} else if obj.Size() != obj.off {
-				err = fmt.Errorf("sbr9 %s: off %d != %s", obj.loghdr, obj.off, obj)
+				err = fmt.Errorf("sbr9 %s: off %d != %s: %w", obj.loghdr, obj.off, obj, io.ErrUnexpectedEOF)
 				nlog.Warningln(err)
 			}
 		} else {

xact/xs/moss.go

@@ -698,6 +698,7 @@ func (r *XactMoss) RecvObj(hdr *transport.ObjHdr, reader io.Reader, err error) e
 		nlog.Errorln(r.Name(), core.T.String(), "RecvObj:", err)
 		r.BcastAbort(err)
 		r.Abort(err)
+		return err
 	}
 	if hdr.ObjAttrs.Size > 0 {
 		r.InObjsAdd(1, hdr.ObjAttrs.Size)