Return worker index in WakerInterest::WorkerAvailable (#337)

Author: fakeshadow

actix-server/src/accept.rs

@@ -238,11 +238,10 @@ impl Accept {
                         match guard.pop_front() {
                             // worker notify it becomes available. we may want to recover
                             // from backpressure.
-                            Some(WakerInterest::WorkerAvailable) => {
+                            Some(WakerInterest::WorkerAvailable(idx)) => {
                                 drop(guard);
-                                // Assume all worker are avail as no worker index returned.
-                                self.avail.set_available_all(&self.handles);
                                 self.maybe_backpressure(&mut sockets, false);
+                                self.avail.set_available(idx, true);
                             }
                             // a new worker thread is made and it's handle would be added to Accept
                             Some(WakerInterest::Worker(handle)) => {

actix-server/src/builder.rs

@@ -320,7 +320,7 @@ impl ServerBuilder {
         idx: usize,
         waker: WakerQueue,
     ) -> (WorkerHandleAccept, WorkerHandleServer) {
-        let avail = WorkerAvailability::new(waker);
+        let avail = WorkerAvailability::new(idx, waker);
         let services = self.services.iter().map(|v| v.clone_factory()).collect();
 
         ServerWorker::start(idx, services, avail, self.worker_config)

actix-server/src/waker_queue.rs

@@ -72,7 +72,7 @@ impl WakerQueue {
 pub(crate) enum WakerInterest {
     /// `WorkerAvailable` is an interest from `Worker` notifying `Accept` there is a worker
     /// available and can accept new tasks.
-    WorkerAvailable,
+    WorkerAvailable(usize),
     /// `Pause`, `Resume`, `Stop` Interest are from `ServerBuilder` future. It listens to
     /// `ServerCommand` and notify `Accept` to do exactly these tasks.
     Pause,

actix-server/src/worker.rs

@@ -96,13 +96,15 @@ impl WorkerHandleServer {
 
 #[derive(Clone)]
 pub(crate) struct WorkerAvailability {
+    idx: usize,
     waker: WakerQueue,
     available: Arc<AtomicBool>,
 }
 
 impl WorkerAvailability {
-    pub fn new(waker: WakerQueue) -> Self {
+    pub fn new(idx: usize, waker: WakerQueue) -> Self {
         WorkerAvailability {
+            idx,
             waker,
             available: Arc::new(AtomicBool::new(false)),
         }
@@ -116,7 +118,7 @@ impl WorkerAvailability {
         let old = self.available.swap(val, Ordering::Release);
         // notify the accept on switched to available.
         if !old && val {
-            self.waker.wake(WakerInterest::WorkerAvailable);
+            self.waker.wake(WakerInterest::WorkerAvailable(self.idx));
         }
     }
 }