xreg: move xaction startup outside renewal lock (fix)

Renewable.Start runs while holding the registry-wide renewal mutex.

* make sure all respective xaction factories do construction only
* move initial list-objects throttling into its Run()
  while preserving the readiness wait
* potential behavior change: multi-object (TCO, arch) to fail fast
  on duplicate transport receiver registration
----
* tests: update unit test `shard_summ_test.go`
----

This was motivated by #336 (lock contention) further leading to an investigation
of the Renewable.Start and associated control sequences.

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

Author: alex-aizman

ais/tgtbck.go

@@ -501,6 +501,11 @@ func (t *target) bckSumm(w http.ResponseWriter, r *http.Request, phase string, b
 			t.writeErr(w, r, rns.Err, http.StatusInternalServerError, Silent)
 			return
 		}
+
+		xctn := rns.Entry.Get()
+		if !rns.IsRunning() {
+			xact.GoRunW(xctn)
+		}
 		w.WriteHeader(http.StatusAccepted)
 		return
 	}
@@ -546,6 +551,11 @@ func (t *target) shardSumm(w http.ResponseWriter, r *http.Request, phase string,
 			t.writeErr(w, r, rns.Err, http.StatusInternalServerError, Silent)
 			return
 		}
+
+		xctn := rns.Entry.Get()
+		if !rns.IsRunning() {
+			xact.GoRunW(xctn)
+		}
 		w.WriteHeader(http.StatusAccepted)
 		return
 	}

ais/tgttxn.go

@@ -749,6 +749,9 @@ func (t *target) tcobjs(c *txnSrv, msg *cmn.TCOMsg, disableDM bool) (xid string,
 		xctn := rns.Entry.Get()
 		xid = xctn.ID()
 
+		if !rns.IsRunning() {
+			xact.GoRunW(xctn)
+		}
 		xtco := xctn.(*xs.XactTCO)
 
 		debug.Assert(msg.TxnUUID == "" || msg.TxnUUID == c.uuid) // (ref050724)
@@ -905,6 +908,10 @@ func (t *target) createArchMultiObj(c *txnSrv) (string /*xaction uuid*/, error)
 		xctn := rns.Entry.Get()
 		xid = xctn.ID()
 
+		if !rns.IsRunning() {
+			xact.GoRunW(xctn)
+		}
+
 		xarch := xctn.(*xs.XactArch)
 		// finalize the message and begin local transaction
 		archMsg.TxnUUID = c.uuid

ec/getx.go

@@ -65,12 +65,11 @@ func (*getFactory) New(_ xreg.Args, bck *meta.Bck) xreg.Renewable {
 	return p
 }
 
+// construct apc.ActECGet xaction (EC manager runs it via _renewXact => xact.GoRunW)
 func (p *getFactory) Start() error {
 	xec := ECM.NewGetXact(p.Bck.Bucket())
 	xec.DemandBase.Init(cos.GenUUID(), p.Kind(), p.Bck, 0 /*use default*/)
 	p.xctn = xec
-
-	xact.GoRunW(xec)
 	return nil
 }
 func (*getFactory) Kind() string     { return apc.ActECGet }

ec/manager.go

@@ -240,7 +240,12 @@ func _renewXact(bck *meta.Bck, kind string) (core.Xact, error) {
 	if rns.Err != nil {
 		return nil, rns.Err
 	}
-	return rns.Entry.Get(), nil
+
+	xctn := rns.Entry.Get()
+	if !rns.IsRunning() {
+		xact.GoRunW(xctn)
+	}
+	return xctn, nil
 }
 
 // A function to process command requests from other targets

ec/putx.go

@@ -65,12 +65,11 @@ func (*putFactory) New(_ xreg.Args, bck *meta.Bck) xreg.Renewable {
 	return p
 }
 
+// construct apc.ActECPut xaction (EC manager runs it via _renewXact => xact.GoRunW)
 func (p *putFactory) Start() error {
 	xec := ECM.NewPutXact(p.Bck.Bucket())
 	xec.DemandBase.Init(cos.GenUUID(), p.Kind(), p.Bck, 0 /*use default*/)
 	p.xctn = xec
-
-	xact.GoRunW(xec)
 	return nil
 }
 

ec/respondx.go

@@ -59,12 +59,11 @@ func (p *rspFactory) WhenPrevIsRunning(xprev xreg.Renewable) (xreg.WPR, error) {
 	return xreg.WprUse, nil
 }
 
+// construct apc.ActECRespond xaction (EC manager runs it via _renewXact => xact.GoRunW)
 func (p *rspFactory) Start() error {
 	xec := ECM.NewRespondXact(p.Bck.Bucket())
 	xec.DemandBase.Init(cos.GenUUID(), p.Kind(), p.Bck, 0 /*use default*/)
 	p.xctn = xec
-
-	xact.GoRunW(xec)
 	return nil
 }
 

xact/xreg/xreg.go

@@ -56,13 +56,18 @@ const (
 
 type (
 	Renewable interface {
-		New(args Args, bck *meta.Bck) Renewable // new xaction stub that can be `Start`-ed; TODO ref: New(args, bck, kind)
-		Start() error                           // starts an xaction, will be called when entry is stored into registry
+		New(args Args, bck *meta.Bck) Renewable
 		Kind() string
 		Get() core.Xact
 		WhenPrevIsRunning(prevEntry Renewable) (action WPR, err error)
 		Bucket() *meta.Bck
 		UUID() string
+
+		// NOTE:
+		// Start constructs and initializes the xaction while holding the registry-wide renewMtx.
+		// It must not wait, sleep, perform blocking channel operations, or call xact.GoRunW.
+		// The renewal caller starts a newly constructed xaction after Renew returns.
+		Start() error
 	}
 	// used in constructions
 	Args struct {

xact/xs/archive.go

@@ -108,6 +108,8 @@ func (*archFactory) New(args xreg.Args, bck *meta.Bck) xreg.Renewable {
 	return p
 }
 
+// construct, initialize xact.Demand (and reg w/ HK), open streams
+// (the renewal caller does xact.GoRunW)
 func (p *archFactory) Start() (err error) {
 	//
 	// target-local generation of a global UUID
@@ -150,8 +152,6 @@ func (p *archFactory) Start() (err error) {
 		}
 	}
 	r.sntl.init(r, r.p.dm, r.config, r.smap, nat)
-
-	xact.GoRunW(r)
 	return nil
 }
 

xact/xs/lso.go

@@ -210,11 +210,6 @@ func (p *lsoFactory) Start() error {
 			RW: false,
 		},
 	)
-	// and maybe throttle right away
-	if r.adv.Sleep > 0 {
-		time.Sleep(r.adv.Sleep)
-	}
-
 	_ = r.CtlMsg()
 	r.vlabs = map[string]string{stats.VlabBucket: bck.Cname("")}
 	p.xctn = r
@@ -317,6 +312,9 @@ func (r *LsoXact) CtlMsg() string {
 }
 
 func (r *LsoXact) Run(wg *sync.WaitGroup) {
+	if r.adv.Sleep > 0 {
+		time.Sleep(r.adv.Sleep)
+	}
 	wg.Done()
 
 	if !r.walk.remote {

xact/xs/nsumm.go

@@ -64,11 +64,9 @@ func (*nsummFactory) New(args xreg.Args, bck *meta.Bck) xreg.Renewable {
 	return p
 }
 
+// construction only (the renewal caller does xact.GoRunW)
 func (p *nsummFactory) Start() (err error) {
 	p.xctn, err = newSumm(p)
-	if err == nil {
-		xact.GoRunW(p.xctn)
-	}
 	return
 }