Race condition in TarantoolClientImpl

- Avoid a possible race between reading, writing and reconnecting
threads when a reconnection process is started.
It might have happened that the lagged thread (reading or writing) could
reset the state to RECONNECT after the reconnecting thread has already
started and set the state to 0. As a result, all next attempts to
reconnect will never happen. Now the reconnect thread holds on the state
as long as it is required.

- Avoid another possible race between reading and writing threads when
they are started during the reconnection process.
It might have happened that one of the threads crashed when it was
starting and another slightly lagged thread set up its flag. It could
have led that the reconnecting thread saw RECONNECT + R/W state instead
of pure RECONNECT. Again, this case broke down all next reconnection
attempts. Now reading and writing threads take into account whether
RECONNECT state is already set or not.

- Replace LockSupport with ReentrantLock.Condition for a thread to be
suspended and woken up. Our cluster tests and standalone demo app show
that LockSupport is not a safe memory barrier as it could be. The
reconnect thread relies on a visibility guarantee between park-unpark
invocations which, actually, sometimes doesn't work. Also, according to
java-docs LockSupport is more like an internal component to build
high-level blocking primitives. It is not recommended using this class
directly. It was replaced by ReentrantLock.Condition primitive based
on LockSupport but which has proper LockSupport usage inside.

Fixes: #142
Affects: #34, #136

Author: nicktorwald

src/main/java/org/tarantool/TarantoolClientImpl.java

@@ -62,33 +62,19 @@ public class TarantoolClientImpl extends TarantoolBase<Future<?>> implements Tar
      */
     protected TarantoolClientStats stats;
     protected StateHelper state = new StateHelper(StateHelper.RECONNECT);
-    protected Thread reader;
-    protected Thread writer;
-
-    /**
-     * The condition variable to signal a reconnection is needed from reader /
-     * writer threads and waiting for that signal from the reconnection thread.
-     *
-     * The lock variable to access this condition.
-     *
-     * XXX: Maybe it is better to move signalling / waiting support to the
-     * state helper (or fsm when we'll reimplement it in that way). Maybe
-     * additional RECONNECTION_NEEDED state is required to distinguish whether
-     * we're in the reconnection state already or just want to reconnect (don't
-     * sure).
-     *
-     * @see #awaitReconnection()
-     * @see #trySignalForReconnection()
-     */
-    protected final ReentrantLock connectorLock = new ReentrantLock();
-    protected final Condition reconnectRequired = connectorLock.newCondition();
+    protected volatile Thread reader;
+    protected volatile Thread writer;
 
     protected Thread connector = new Thread(new Runnable() {
         @Override
         public void run() {
             while (!Thread.currentThread().isInterrupted()) {
                 reconnect(0, thumbstone);
-                awaitReconnection();
+                try {
+                    state.awaitReconnection();
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                }
             }
         }
     });
@@ -177,15 +163,6 @@ protected void connect(final SocketChannel channel) throws Exception {
 
     protected void startThreads(String threadName) throws InterruptedException {
         final CountDownLatch init = new CountDownLatch(2);
-
-        if (reader != null) {
-            reader.join(config.initTimeoutMillis / 2);
-        }
-        if (writer != null) {
-            writer.join(config.initTimeoutMillis / 2);
-        }
-        state.release(StateHelper.RECONNECT);
-
         reader = new Thread(new Runnable() {
             @Override
             public void run() {
@@ -195,7 +172,16 @@ public void run() {
                         readThread();
                     } finally {
                         state.release(StateHelper.READING);
-                        trySignalForReconnection();
+                        // there're two cases when a read thread is here
+                        // 1. it's a new generation thread inside/outside
+                        //    a reconnection process (currentThread == reader)
+                        // 2. It's an old generation thread inside
+                        //    a reconnection process (currentThread != reader)
+                        // Skip the old gen. attempt to reconnect
+                        if (state.getState() == StateHelper.UNINITIALIZED
+                                && Thread.currentThread() == reader) {
+                            state.trySignalForReconnection();
+                        }
                     }
                 }
             }
@@ -209,12 +195,28 @@ public void run() {
                         writeThread();
                     } finally {
                         state.release(StateHelper.WRITING);
-                        trySignalForReconnection();
+                        // there're two cases when a write thread is here
+                        // 1. it's a new generation thread inside/outside
+                        //    a reconnection process (currentThread == writer)
+                        // 2. It's an old generation thread inside
+                        //    a reconnection process (currentThread != writer)
+                        // Skip the old gen. attempt to reconnect
+                        if (state.getState() == StateHelper.UNINITIALIZED
+                                && Thread.currentThread() == writer) {
+                            state.trySignalForReconnection();
+                        }
                     }
                 }
             }
         });
 
+        // reconnection preparation is done
+        // before reconnection state will be released
+        // reader/writer threads have been replaced by new ones
+        // it's required to be sure that old r/w threads
+        // won't affect new r/w threads.
+        state.release(StateHelper.RECONNECT);
+
         configureThreads(threadName);
         reader.start();
         writer.start();
@@ -492,40 +494,6 @@ protected void stopIO() {
         closeChannel(channel);
     }
 
-    /**
-     * Blocks until a reconnection signal will be received.
-     *
-     * @see #trySignalForReconnection()
-     */
-    private void awaitReconnection() {
-        connectorLock.lock();
-        try {
-            while (state.getState() != StateHelper.RECONNECT) {
-                reconnectRequired.await();
-            }
-        } catch (InterruptedException ignored) {
-            Thread.currentThread().interrupt();
-        } finally {
-            connectorLock.unlock();
-        }
-    }
-
-    /**
-     * Signals to the connector that reconnection process can be performed.
-     *
-     * @see #awaitReconnection()
-     */
-    private void trySignalForReconnection() {
-        if (state.compareAndSet(StateHelper.UNINITIALIZED, StateHelper.RECONNECT)) {
-            connectorLock.lock();
-            try {
-                reconnectRequired.signal();
-            } finally {
-                connectorLock.unlock();
-            }
-        }
-    }
-
     @Override
     public boolean isAlive() {
         return state.getState() == StateHelper.ALIVE && thumbstone == null;
@@ -682,6 +650,18 @@ protected final class StateHelper {
 
         private final CountDownLatch closedLatch = new CountDownLatch(1);
 
+        /**
+         * The condition variable to signal a reconnection is needed from reader /
+         * writer threads and waiting for that signal from the reconnection thread.
+         *
+         * The lock variable to access this condition.
+         *
+         * @see #awaitReconnection()
+         * @see #trySignalForReconnection()
+         */
+        protected final ReentrantLock connectorLock = new ReentrantLock();
+        protected final Condition reconnectRequired = connectorLock.newCondition();
+
         protected StateHelper(int state) {
             this.state = new AtomicInteger(state);
         }
@@ -701,11 +681,7 @@ protected boolean close() {
                 if ((st & CLOSED) == CLOSED)
                     return false;
 
-                /* Drop RECONNECT, set CLOSED.
-                 *
-                 * XXX: Should not we also drop other states? Or reimplement
-                 * this class as a finite state machine?
-                 */
+                /*  Drop RECONNECT, set CLOSED. */
                 if (compareAndSet(st, (st & ~RECONNECT) | CLOSED))
                     return true;
             }
@@ -766,10 +742,18 @@ protected boolean compareAndSet(int expect, int update) {
             return true;
         }
 
+        /**
+         * Reconnection uses another way to await state via receiving a signal
+         * instead of latches.
+         */
         protected void awaitState(int state) throws InterruptedException {
-            CountDownLatch latch = getStateLatch(state);
-            if (latch != null) {
-                latch.await();
+            if (state == RECONNECT) {
+                awaitReconnection();
+            } else {
+                CountDownLatch latch = getStateLatch(state);
+                if (latch != null) {
+                    latch.await();
+                }
             }
         }
 
@@ -793,5 +777,37 @@ private CountDownLatch getStateLatch(int state) {
             }
             return null;
         }
+
+        /**
+         * Blocks until a reconnection signal will be received.
+         *
+         * @see #trySignalForReconnection()
+         */
+        private void awaitReconnection() throws InterruptedException {
+            connectorLock.lock();
+            try {
+                while (getState() != StateHelper.RECONNECT) {
+                    reconnectRequired.await();
+                }
+            } finally {
+                connectorLock.unlock();
+            }
+        }
+
+        /**
+         * Signals to the connector that reconnection process can be performed.
+         *
+         * @see #awaitReconnection()
+         */
+        private void trySignalForReconnection() {
+            if (compareAndSet(StateHelper.UNINITIALIZED, StateHelper.RECONNECT)) {
+                connectorLock.lock();
+                try {
+                    reconnectRequired.signal();
+                } finally {
+                    connectorLock.unlock();
+                }
+            }
+        }
     }
 }