From 4151bbffe7b3ab882de5f7ac29f41c974679f087 Mon Sep 17 00:00:00 2001
From: Valery Piashchynski <piashchynski.valery@gmail.com>
Date: Wed, 14 Jul 2021 16:46:32 +0300
Subject: Fix TTL issue, added explanation comments. The worker after it
 executed the request, may overwrite the TTL state. This inconsistency leads
 to the +1 worker in the FIFO channel. In this state, the Push operation was
 blocked.

Add RR_BROADCAST_PATH.

Signed-off-by: Valery Piashchynski <piashchynski.valery@gmail.com>
---
 pkg/worker_watcher/container/vec.go  |  2 +-
 pkg/worker_watcher/interface.go      | 31 -------------------------------
 pkg/worker_watcher/worker_watcher.go | 21 ++++++++++++++++-----
 3 files changed, 17 insertions(+), 37 deletions(-)
 delete mode 100644 pkg/worker_watcher/interface.go

(limited to 'pkg/worker_watcher')

diff --git a/pkg/worker_watcher/container/vec.go b/pkg/worker_watcher/container/vec.go
index b9150c43..24b5fa6d 100644
--- a/pkg/worker_watcher/container/vec.go
+++ b/pkg/worker_watcher/container/vec.go
@@ -13,7 +13,7 @@ type Vec struct {
 	workers chan worker.BaseProcess
 }
 
-func NewVector(initialNumOfWorkers uint64) Vector {
+func NewVector(initialNumOfWorkers uint64) *Vec {
 	vec := &Vec{
 		destroy: 0,
 		workers: make(chan worker.BaseProcess, initialNumOfWorkers),
diff --git a/pkg/worker_watcher/interface.go b/pkg/worker_watcher/interface.go
deleted file mode 100644
index 29fa3640..00000000
--- a/pkg/worker_watcher/interface.go
+++ /dev/null
@@ -1,31 +0,0 @@
-package worker_watcher //nolint:stylecheck
-
-import (
-	"context"
-
-	"github.com/spiral/roadrunner/v2/pkg/worker"
-)
-
-// Watcher is an interface for the Sync workers lifecycle
-type Watcher interface {
-	// Watch used to add workers to the container
-	Watch(workers []worker.BaseProcess) error
-
-	// Get provide first free worker
-	Get(ctx context.Context) (worker.BaseProcess, error)
-
-	// Push enqueues worker back
-	Push(w worker.BaseProcess)
-
-	// Allocate - allocates new worker and put it into the WorkerWatcher
-	Allocate() error
-
-	// Destroy destroys the underlying container
-	Destroy(ctx context.Context)
-
-	// List return all container w/o removing it from internal storage
-	List() []worker.BaseProcess
-
-	// Remove will remove worker from the container
-	Remove(wb worker.BaseProcess)
-}
diff --git a/pkg/worker_watcher/worker_watcher.go b/pkg/worker_watcher/worker_watcher.go
index f82de958..b2d61d48 100755
--- a/pkg/worker_watcher/worker_watcher.go
+++ b/pkg/worker_watcher/worker_watcher.go
@@ -11,8 +11,18 @@ import (
 	"github.com/spiral/roadrunner/v2/pkg/worker_watcher/container"
 )
 
+// Vector interface represents vector container
+type Vector interface {
+	// Enqueue used to put worker to the vector
+	Enqueue(worker.BaseProcess)
+	// Dequeue used to get worker from the vector
+	Dequeue(ctx context.Context) (worker.BaseProcess, error)
+	// Destroy used to stop releasing the workers
+	Destroy()
+}
+
 // NewSyncWorkerWatcher is a constructor for the Watcher
-func NewSyncWorkerWatcher(allocator worker.Allocator, numWorkers uint64, events events.Handler) Watcher {
+func NewSyncWorkerWatcher(allocator worker.Allocator, numWorkers uint64, events events.Handler) *workerWatcher {
 	ww := &workerWatcher{
 		container:  container.NewVector(numWorkers),
 		numWorkers: numWorkers,
@@ -26,7 +36,7 @@ func NewSyncWorkerWatcher(allocator worker.Allocator, numWorkers uint64, events
 
 type workerWatcher struct {
 	sync.RWMutex
-	container container.Vector
+	container Vector
 	// used to control the Destroy stage (that all workers are in the container)
 	numWorkers uint64
 	workers    []worker.BaseProcess
@@ -150,11 +160,12 @@ func (ww *workerWatcher) Remove(wb worker.BaseProcess) {
 
 // Push O(1) operation
 func (ww *workerWatcher) Push(w worker.BaseProcess) {
-	if w.State().Value() != worker.StateReady {
+	switch w.State().Value() {
+	case worker.StateReady:
+		ww.container.Enqueue(w)
+	default:
 		_ = w.Kill()
-		return
 	}
-	ww.container.Enqueue(w)
 }
 
 // Destroy all underlying container (but let them to complete the task)
-- 
cgit v1.2.3