Revert "first version of changes to avoid accidental worker state update during worker recovery"

This reverts commit 8739920.

Author: rasamala83

lib/coordinator.go

@@ -831,7 +831,6 @@ func (crd *Coordinator) dispatchRequest(request *netstring.Netstring) error {
 		// donot return a stranded worker. recover inserts a good worker back to pool.
 		//
 		if err == ErrSaturationKill {
-			logger.GetLogger().Log(logger.Info, "trigger recovery as part of ErrSaturationKill")
 			go worker.Recover(workerpool, ticket, WorkerClientRecoverParam{allowSkipOciBreak: true}, &strandedCalInfo{raddr: crd.conn.RemoteAddr().String(), laddr: crd.conn.LocalAddr().String(), nameSuffix: "_SATURATION_RECOVERED"}, common.StrandedSaturationRecover)
 		} else {
 			go worker.Recover(workerpool, ticket, WorkerClientRecoverParam{allowSkipOciBreak: true}, &strandedCalInfo{raddr: crd.conn.RemoteAddr().String(), laddr: crd.conn.LocalAddr().String()})
@@ -926,7 +925,7 @@ func (crd *Coordinator) doRequest(ctx context.Context, worker *WorkerClient, req
 			// The worker is in ACPT state.
 			// It will not finish recovery because of ACPT. The worker will never get back into the pool.
 			// Just marking the state as FNSH and dispatchRequest will return the worker back to the pool.
-			worker.setState(wsFnsh, false)
+			worker.setState(wsFnsh)
 			return false, ErrReqParseFail
 		}
 		cnt := 1

lib/workerbroker.go

@@ -64,7 +64,7 @@ type WorkerBroker struct {
 	// and restart the stopped workers.
 	//
 	pidworkermap map[int32]*WorkerClient
-	lock         sync.Mutex
+	lock sync.Mutex
 
 	//
 	// loaded from cfg once and used later.
@@ -204,9 +204,7 @@ func (broker *WorkerBroker) GetWorkerPoolCfgs() (pCfgs []map[HeraWorkerType]*Wor
 
 // GetWorkerPool get the worker pool object for the type and id
 // ids holds optional paramenters.
-//
-//	ids[0] == instance id; ids[1] == shard id.
-//
+//   ids[0] == instance id; ids[1] == shard id.
 // if a particular id is not set, it defaults to 0.
 // TODO: interchange sid <--> instId since instId is not yet used
 func (broker *WorkerBroker) GetWorkerPool(wType HeraWorkerType, ids ...int) (workerbroker *WorkerPool, err error) {
@@ -318,7 +316,7 @@ func (broker *WorkerBroker) startWorkerMonitor() (err error) {
 								if logger.GetLogger().V(logger.Debug) {
 									logger.GetLogger().Log(logger.Debug, "worker (pid=", workerclient.pid, ") received signal. transits from state ", workerclient.Status, " to terminated.")
 								}
-								workerclient.setState(wsUnset, true) // Set the state to UNSET to make sure worker does not stay in FNSH state so long
+								workerclient.setState(wsUnset) // Set the state to UNSET to make sure worker does not stay in FNSH state so long
 								pool.RestartWorker(workerclient)
 							}
 						} else {
@@ -367,8 +365,8 @@ func (broker *WorkerBroker) startWorkerMonitor() (err error) {
 }
 
 /*
-resizePool calls workerpool.Resize to resize a worker pool when the dynamic configuration of
-the number of workers changed
+	resizePool calls workerpool.Resize to resize a worker pool when the dynamic configuration of
+	the number of workers changed
 */
 func (broker *WorkerBroker) resizePool(wType HeraWorkerType, maxWorkers int, shardID int) {
 	broker.poolCfgs[0][wType].maxWorkerCnt = maxWorkers
@@ -383,7 +381,7 @@ func (broker *WorkerBroker) resizePool(wType HeraWorkerType, maxWorkers int, sha
 }
 
 /*
-changeMaxWorkers is called when the dynamic config changed, it calls resizePool() for all the pools
+	changeMaxWorkers is called when the dynamic config changed, it calls resizePool() for all the pools
 */
 func (broker *WorkerBroker) changeMaxWorkers() {
 	wW := GetNumWWorkers(0)

lib/workerclient.go

@@ -25,7 +25,6 @@ import (
 	"net"
 	"os"
 	"path/filepath"
-	"runtime"
 	"strconv"
 	"strings"
 	"sync/atomic"
@@ -472,7 +471,7 @@ func (worker *WorkerClient) StartWorker() (err error) {
 		logger.GetLogger().Log(logger.Info, "Started ", workerPath, ", pid=", pid)
 	}
 	worker.pid = pid
-	worker.setState(wsInit, false)
+	worker.setState(wsInit)
 	return nil
 }
 
@@ -542,7 +541,7 @@ func (worker *WorkerClient) attachToWorker() (err error) {
 		logger.GetLogger().Log(logger.Info, "Got control message from worker (", worker.ID, ",", worker.pid, ",", worker.racID, ",", worker.dbUname, ")")
 	}
 
-	worker.setState(wsAcpt, false)
+	worker.setState(wsAcpt)
 
 	pool, err := GetWorkerBrokerInstance().GetWorkerPool(worker.Type, worker.instID, worker.shardID)
 	if err != nil {
@@ -634,11 +633,11 @@ type WorkerClientRecoverParam struct {
 func (worker *WorkerClient) Recover(p *WorkerPool, ticket string, recovParam WorkerClientRecoverParam, info *strandedCalInfo, param ...int) {
 	if atomic.CompareAndSwapInt32(&worker.isUnderRecovery, 0, 1) {
 		if logger.GetLogger().V(logger.Debug) {
-			logger.GetLogger().Log(logger.Debug, "begin recover worker Id: ", worker.ID, " process Id: ", worker.pid)
+			logger.GetLogger().Log(logger.Debug, "begin recover worker: ", worker.pid)
 		}
 	} else {
 		if logger.GetLogger().V(logger.Debug) {
-			logger.GetLogger().Log(logger.Debug, "worker already underrecovery: ", worker.ID, " process Id: ", worker.pid)
+			logger.GetLogger().Log(logger.Debug, "worker already underrecovery: ", worker.pid)
 		}
 		//
 		// defer will not be called.
@@ -666,10 +665,7 @@ func (worker *WorkerClient) Recover(p *WorkerPool, ticket string, recovParam Wor
 		return
 	}
 	priorWorkerStatus := worker.Status
-	if logger.GetLogger().V(logger.Debug) {
-		logger.GetLogger().Log(logger.Debug, fmt.Sprintf("about to recover worker Id: %d, worker process Id: %d as part of reconvery process, setting worker state to Quece", worker.ID, worker.pid))
-	}
-	worker.setState(wsQuce, true)
+	worker.setState(wsQuce)
 	killparam := common.StrandedClientClose
 	if len(param) > 0 {
 		killparam = param[0]
@@ -680,17 +676,9 @@ func (worker *WorkerClient) Recover(p *WorkerPool, ticket string, recovParam Wor
 		select {
 		case <-workerRecoverTimeout:
 			worker.thr.CanRun()
-			worker.setState(wsInit, true) // Set the worker state to INIT when we decide to Terminate the worker
-			GetStateLog().PublishStateEvent(StateEvent{eType: WorkerStateEvt, shardID: worker.shardID, wType: worker.Type, instID: worker.instID, workerID: worker.ID, newWState: worker.Status})
+			worker.setState(wsInit) // Set the worker state to INIT when we decide to Terminate the worker
 			worker.Terminate()
 			worker.callogStranded("RECYCLED", info)
-			if logger.GetLogger().V(logger.Debug) {
-				logger.GetLogger().Log(logger.Debug, fmt.Sprintf("worker Id: %d and process: %d recovered as part of workerRecoverTimeout set status to INIT", worker.ID, worker.pid))
-			}
-			err := p.RestartWorker(worker)
-			if err != nil {
-				logger.GetLogger().Log(logger.Alert, fmt.Sprintf("worker: %d failed to restart worker process", worker.ID))
-			}
 			return
 		case msg, ok := <-worker.channel():
 			if !ok {
@@ -727,10 +715,7 @@ func (worker *WorkerClient) Recover(p *WorkerPool, ticket string, recovParam Wor
 					}
 					worker.callogStranded("RECOVERED", info)
 
-					worker.setState(wsFnsh, true)
-					if logger.GetLogger().V(logger.Debug) {
-						logger.GetLogger().Log(logger.Debug, fmt.Sprintf("worker Id: %d, worker process: %d recovered as part of message from channel set status to FINSH", worker.ID, worker.pid))
-					}
+					worker.setState(wsFnsh)
 					p.ReturnWorker(worker, ticket)
 					//
 					// donot set state to ACPT since worker could already be picked up by another
@@ -910,13 +895,13 @@ func (worker *WorkerClient) doRead() {
 				logger.GetLogger().Log(logger.Verbose, "workerclient (<<< pid =", worker.pid, ",wrqId:", worker.rqId, "): EOR code:", eor, ", rqId: ", rqId, ", data:", DebugString(payload))
 			}
 			if eor == common.EORFree {
-				worker.setState(wsFnsh, false)
+				worker.setState(wsFnsh)
 				/*worker.sqlStartTimeMs = 0
 				if logger.GetLogger().V(logger.Verbose) {
 					logger.GetLogger().Log(logger.Verbose, "workerclient sqltime=", worker.sqlStartTimeMs)
 				}*/
 			} else {
-				worker.setState(wsWait, false)
+				worker.setState(wsWait)
 			}
 			if eor != common.EORMoreIncomingRequests {
 				worker.outCh <- &workerMsg{data: payload, eor: true, free: (eor == common.EORFree), inTransaction: ((eor == common.EORInTransaction) || (eor == common.EORInCursorInTransaction)), rqId: rqId}
@@ -940,7 +925,7 @@ func (worker *WorkerClient) doRead() {
 			return
 		default:
 			if ns.Cmd != common.RcStillExecuting {
-				worker.setState(wsWait, false)
+				worker.setState(wsWait)
 			}
 			if logger.GetLogger().V(logger.Verbose) {
 				logger.GetLogger().Log(logger.Verbose, "workerclient (<<< pid =", worker.pid, "): data:", DebugString(ns.Serialized), len(ns.Serialized))
@@ -956,7 +941,7 @@ func (worker *WorkerClient) doRead() {
 
 // Write sends a message to the worker
 func (worker *WorkerClient) Write(ns *netstring.Netstring, nsCount uint16) error {
-	worker.setState(wsBusy, false)
+	worker.setState(wsBusy)
 
 	worker.rqId += uint32(nsCount)
 
@@ -980,24 +965,12 @@ func (worker *WorkerClient) Write(ns *netstring.Netstring, nsCount uint16) error
 }
 
 // setState updates the worker state
-func (worker *WorkerClient) setState(status HeraWorkerStatus, callFromRecovery bool) {
+func (worker *WorkerClient) setState(status HeraWorkerStatus) {
 	if worker.Status == status {
 		return
 	}
-	if worker.isUnderRecovery == 1 && !callFromRecovery {
-		if logger.GetLogger().V(logger.Info) {
-			//If worker under recovery drinup of channel happens as part of DrainResponseChannel
-			logger.GetLogger().Log(logger.Info, "worker : ", worker.ID, " is under recovery. "+
-				"workerclient pid=", worker.pid, "not allowed changing status from", worker.Status, "to", status)
-		}
-		if logger.GetLogger().V(logger.Debug) {
-			worker.printCallStack()
-		}
-		return
-	}
 	if logger.GetLogger().V(logger.Debug) {
-		logger.GetLogger().Log(logger.Debug, "worker Id=", worker.ID, " worker pid=", worker.pid, " changing status from", worker.Status, "to", status)
-		worker.printCallStack()
+		logger.GetLogger().Log(logger.Debug, "worker pid=", worker.pid, " changing status from", worker.Status, "to", status)
 	}
 
 	// TODO: sync atomic set
@@ -1029,27 +1002,3 @@ func (worker *WorkerClient) isProcessRunning() bool {
 	}
 	return true
 }
-
-func (worker *WorkerClient) printCallStack() {
-	// Define a large enough buffer to capture the stack.
-	const depth = 64
-	pcs := make([]uintptr, depth)
-
-	// Collect the stack trace.
-	n := runtime.Callers(2, pcs) // Skip the first 2 callers (runtime and printCallStack itself).
-	frames := runtime.CallersFrames(pcs[:n])
-	indent := 0
-	// Iterate through the frames and print function names and line numbers.
-	var builder strings.Builder
-	builder.WriteString(fmt.Sprintf("worker Id= %d Process Id= %d Call Stack:", worker.ID, worker.pid))
-	for {
-		frame, more := frames.Next()
-		builder.WriteString(fmt.Sprintf("%s - %s\n", strings.Repeat("  ", indent), frame.Function))
-		builder.WriteString(fmt.Sprintf("%s  at %s:%d\n", strings.Repeat("  ", indent), frame.File, frame.Line))
-		indent++
-		if !more {
-			break
-		}
-	}
-	logger.GetLogger().Log(logger.Debug, builder.String())
-}

lib/workerpool.go

@@ -116,7 +116,7 @@ func (pool *WorkerPool) Init(wType HeraWorkerType, size int, instID int, shardID
 func (pool *WorkerPool) spawnWorker(wid int) error {
 	worker := NewWorker(wid, pool.Type, pool.InstID, pool.ShardID, pool.moduleName, pool.thr)
 
-	worker.setState(wsSchd, false)
+	worker.setState(wsSchd)
 	millis := rand.Intn(GetConfig().RandomStartMs)
 	if logger.GetLogger().V(logger.Alert) {
 		logger.GetLogger().Log(logger.Alert, wid, "randomized start ms", millis)
@@ -495,7 +495,7 @@ func (pool *WorkerPool) ReturnWorker(worker *WorkerClient, ticket string) (err e
 		worker.DrainResponseChannel(time.Microsecond * 10)
 	}
 
-	worker.setState(wsAcpt, true)
+	worker.setState(wsAcpt)
 	if (pool.desiredSize < pool.currentSize) && (worker.ID >= pool.desiredSize) {
 		go func(w *WorkerClient) {
 			if logger.GetLogger().V(logger.Info) {

lib/workerpool_test.go

@@ -76,12 +76,12 @@ func TestPoolDempotency(t *testing.T) {
 	wd := NewWorker(3, wtypeRW, 0, 0, "cloc", nil)
 	we := NewWorker(4, wtypeRW, 0, 0, "cloc", nil)
 	wf := NewWorker(5, wtypeRW, 0, 0, "cloc", nil)
-	wa.setState(wsAcpt, false)
-	wb.setState(wsAcpt, false)
-	wc.setState(wsAcpt, false)
-	wd.setState(wsAcpt, false)
-	we.setState(wsAcpt, false)
-	wf.setState(wsAcpt, false)
+	wa.setState(wsAcpt)
+	wb.setState(wsAcpt)
+	wc.setState(wsAcpt)
+	wd.setState(wsAcpt)
+	we.setState(wsAcpt)
+	wf.setState(wsAcpt)
 	pool.WorkerReady(wa)
 	pool.WorkerReady(wb)
 	pool.WorkerReady(wc)