Repository: netty/netty
Issue: Blocking in SingleThreadEventExecutor#startThread
Labels: improvement

Currently `io.netty.util.concurrent.SingleThreadEventExecutor#startThread` called every time when `SingleThreadEventExecutor#execute` is called. 

This can cause unnecessary blocking because startThread contains synchronized block which called every time (performance penalty are small anyway, about 0.2% for my application).


Issue comments:
Let me check... maybe I can do something smart with an `AtomicReference`

@trustin I wonder why we even lazy start the Thread and add the task to the priority queue. Why not just do this in the constructor right away ?

A user could create relatively large executor group and then only register smaller number of channels, then we will be wasting threads.

@trustin true... not sure how this would happen in reality though ... Let me just use some AtomicInteger with compareAndSet and spinning. This should work.

@valodzka should be fixed now by #2310 . Please re-test and let me know.

Tested with 4.0.18.Final-20140314.085205-40, issue has been solved

@valodzka awesome... thanks for the feedback and like I said. Keep them coming ;)


Pull request review comments:
`case ST_NOT_STARTED:` could also just fall-through.

Maybe like this?

``` java
for (;;) {
    int oldState = state.get();
    if (oldState >= ST_SHUTTING_DOWN || state.compareAndSet(oldState, ST_SHUTTING_DOWN)) {
        break;
    }
}
```

Could we also use `PlatformDependent.newAtomicIntegerFieldUpdater()` if possible?


Patch that resolved the issue:
diff --git a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
index 37bf9808151..209e3beba03 100644
--- a/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
+++ b/common/src/main/java/io/netty/util/concurrent/SingleThreadEventExecutor.java
@@ -33,6 +33,7 @@
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * Abstract base class for {@link EventExecutor}'s that execute all its submitted tasks in a single thread.
@@ -61,13 +62,12 @@ public void run() {
     final Queue<ScheduledFutureTask<?>> delayedTaskQueue = new PriorityQueue<ScheduledFutureTask<?>>();
 
     private final Thread thread;
-    private final Object stateLock = new Object();
     private final Semaphore threadLock = new Semaphore(0);
     private final Set<Runnable> shutdownHooks = new LinkedHashSet<Runnable>();
     private final boolean addTaskWakesUp;
 
     private long lastExecutionTime;
-    private volatile int state = ST_NOT_STARTED;
+    private final AtomicInteger state = new AtomicInteger(ST_NOT_STARTED);
     private volatile long gracefulShutdownQuietPeriod;
     private volatile long gracefulShutdownTimeout;
     private long gracefulShutdownStartTime;
@@ -103,8 +103,9 @@ public void run() {
                 } catch (Throwable t) {
                     logger.warn("Unexpected exception from an event executor: ", t);
                 } finally {
-                    if (state < ST_SHUTTING_DOWN) {
-                        state = ST_SHUTTING_DOWN;
+                    // TODO: Maybe compareAndSet ?
+                    if (state.get() < ST_SHUTTING_DOWN) {
+                        state.set(ST_SHUTTING_DOWN);
                     }
 
                     // Check if confirmShutdown() was called at the end of the loop.
@@ -126,9 +127,7 @@ public void run() {
                         try {
                             cleanup();
                         } finally {
-                            synchronized (stateLock) {
-                                state = ST_TERMINATED;
-                            }
+                            state.set(ST_TERMINATED);
                             threadLock.release();
                             if (!taskQueue.isEmpty()) {
                                 logger.warn(
@@ -414,7 +413,7 @@ protected void cleanup() {
     }
 
     protected void wakeup(boolean inEventLoop) {
-        if (!inEventLoop || state == ST_SHUTTING_DOWN) {
+        if (!inEventLoop || state.get() == ST_SHUTTING_DOWN) {
             taskQueue.add(WAKEUP_TASK);
         }
     }
@@ -498,32 +497,39 @@ public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit uni
         }
 
         boolean inEventLoop = inEventLoop();
-        boolean wakeup = true;
-
-        synchronized (stateLock) {
+        boolean wakeup;
+        int oldState;
+        for (;;) {
             if (isShuttingDown()) {
                 return terminationFuture();
             }
-
-            gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);
-            gracefulShutdownTimeout = unit.toNanos(timeout);
-
+            int newState;
+            wakeup = true;
+            oldState = state.get();
             if (inEventLoop) {
-                assert state == ST_STARTED;
-                state = ST_SHUTTING_DOWN;
+                newState = ST_SHUTTING_DOWN;
             } else {
-                switch (state) {
+                switch (oldState) {
                     case ST_NOT_STARTED:
-                        state = ST_SHUTTING_DOWN;
-                        thread.start();
+                        newState = ST_SHUTTING_DOWN;
                         break;
                     case ST_STARTED:
-                        state = ST_SHUTTING_DOWN;
+                        newState = ST_SHUTTING_DOWN;
                         break;
                     default:
+                        newState = oldState;
                         wakeup = false;
                 }
             }
+            if (state.compareAndSet(oldState, newState)) {
+                break;
+            }
+        }
+        gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);
+        gracefulShutdownTimeout = unit.toNanos(timeout);
+
+        if (oldState == ST_NOT_STARTED) {
+            thread.start();
         }
 
         if (wakeup) {
@@ -546,30 +552,38 @@ public void shutdown() {
         }
 
         boolean inEventLoop = inEventLoop();
-        boolean wakeup = true;
-
-        synchronized (stateLock) {
-            if (isShutdown()) {
+        boolean wakeup;
+        int oldState;
+        for (;;) {
+            if (isShuttingDown()) {
                 return;
             }
-
+            int newState;
+            wakeup = true;
+            oldState = state.get();
             if (inEventLoop) {
-                assert state == ST_STARTED || state == ST_SHUTTING_DOWN;
-                state = ST_SHUTDOWN;
+                newState = ST_SHUTDOWN;
             } else {
-                switch (state) {
-                case ST_NOT_STARTED:
-                    state = ST_SHUTDOWN;
-                    thread.start();
-                    break;
-                case ST_STARTED:
-                case ST_SHUTTING_DOWN:
-                    state = ST_SHUTDOWN;
-                    break;
-                default:
-                    wakeup = false;
+                switch (oldState) {
+                    case ST_NOT_STARTED:
+                        newState = ST_SHUTDOWN;
+                        break;
+                    case ST_STARTED:
+                    case ST_SHUTTING_DOWN:
+                        newState = ST_SHUTDOWN;
+                        break;
+                    default:
+                        newState = oldState;
+                        wakeup = false;
                 }
             }
+            if (state.compareAndSet(oldState, newState)) {
+                break;
+            }
+        }
+
+        if (oldState == ST_NOT_STARTED) {
+            thread.start();
         }
 
         if (wakeup) {
@@ -579,17 +593,17 @@ public void shutdown() {
 
     @Override
     public boolean isShuttingDown() {
-        return state >= ST_SHUTTING_DOWN;
+        return state.get() >= ST_SHUTTING_DOWN;
     }
 
     @Override
     public boolean isShutdown() {
-        return state >= ST_SHUTDOWN;
+        return state.get() >= ST_SHUTDOWN;
     }
 
     @Override
     public boolean isTerminated() {
-        return state == ST_TERMINATED;
+        return state.get() == ST_TERMINATED;
     }
 
     /**
@@ -808,9 +822,8 @@ public void run() {
     }
 
     private void startThread() {
-        synchronized (stateLock) {
-            if (state == ST_NOT_STARTED) {
-                state = ST_STARTED;
+        if (state.get() == ST_NOT_STARTED) {
+            if (state.compareAndSet(ST_NOT_STARTED, ST_STARTED)) {
                 delayedTaskQueue.add(new ScheduledFutureTask<Void>(
                         this, delayedTaskQueue, Executors.<Void>callable(new PurgeTask(), null),
                         ScheduledFutureTask.deadlineNanos(SCHEDULE_PURGE_INTERVAL), -SCHEDULE_PURGE_INTERVAL));


Test patch: