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.

- Avoid LockSupport class usage for a thread to be suspended and woken
up. Actually, 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

@@ -2,8 +2,8 @@
 
 import org.tarantool.protocol.ProtoUtils;
 import org.tarantool.protocol.ReadableViaSelectorChannel;
-import org.tarantool.protocol.TarantoolPacket;
 import org.tarantool.protocol.TarantoolGreeting;
+import org.tarantool.protocol.TarantoolPacket;
 
 import java.io.IOException;
 import java.nio.ByteBuffer;
@@ -22,7 +22,6 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.Condition;
-import java.util.concurrent.locks.LockSupport;
 import java.util.concurrent.locks.ReentrantLock;
 
 
@@ -66,14 +65,15 @@ public class TarantoolClientImpl extends TarantoolBase<Future<?>> implements Tar
     protected Thread reader;
     protected Thread writer;
 
+    protected final ReentrantLock connectorLock = new ReentrantLock();
+    protected final Condition reconnectRequired = connectorLock.newCondition();
+
     protected Thread connector = new Thread(new Runnable() {
         @Override
         public void run() {
             while (!Thread.currentThread().isInterrupted()) {
-                if (state.compareAndSet(StateHelper.RECONNECT, 0)) {
-                    reconnect(0, thumbstone);
-                }
-                LockSupport.park(state);
+                reconnect(0, thumbstone);
+                awaitReconnection();
             }
         }
     });
@@ -139,16 +139,13 @@ protected void reconnect(int retry, Throwable lastError) {
     protected void connect(final SocketChannel channel) throws Exception {
         try {
             TarantoolGreeting greeting = ProtoUtils.connect(channel,
-                config.username, config.password);
+                    config.username, config.password);
             this.serverVersion = greeting.getServerVersion();
         } catch (IOException e) {
-            try {
-                channel.close();
-            } catch (IOException ignored) {
-            }
-
+            closeChannel(channel);
             throw new CommunicationException("Couldn't connect to tarantool", e);
         }
+
         channel.configureBlocking(false);
         this.channel = channel;
         this.readChannel = new ReadableViaSelectorChannel(channel);
@@ -165,6 +162,15 @@ 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() {
@@ -174,8 +180,7 @@ public void run() {
                         readThread();
                     } finally {
                         state.release(StateHelper.READING);
-                        if (state.compareAndSet(0, StateHelper.RECONNECT))
-                            LockSupport.unpark(connector);
+                        trySignalForReconnection();
                     }
                 }
             }
@@ -189,8 +194,7 @@ public void run() {
                         writeThread();
                     } finally {
                         state.release(StateHelper.WRITING);
-                        if (state.compareAndSet(0, StateHelper.RECONNECT))
-                            LockSupport.unpark(connector);
+                        trySignalForReconnection();
                     }
                 }
             }
@@ -337,25 +341,21 @@ private boolean directWrite(ByteBuffer buffer) throws InterruptedException, IOEx
     }
 
     protected void readThread() {
-        try {
-            while (!Thread.currentThread().isInterrupted()) {
-                try {
-                    TarantoolPacket packet = ProtoUtils.readPacket(readChannel);
+        while (!Thread.currentThread().isInterrupted()) {
+            try {
+                TarantoolPacket packet = ProtoUtils.readPacket(readChannel);
 
-                    Map<Integer, Object> headers = packet.getHeaders();
+                Map<Integer, Object> headers = packet.getHeaders();
 
-                    Long syncId = (Long) headers.get(Key.SYNC.getId());
-                    CompletableFuture<?> future = futures.remove(syncId);
-                    stats.received++;
-                    wait.decrementAndGet();
-                    complete(packet, future);
-                } catch (Exception e) {
-                    die("Cant read answer", e);
-                    return;
-                }
+                Long syncId = (Long) headers.get(Key.SYNC.getId());
+                CompletableFuture<?> future = futures.remove(syncId);
+                stats.received++;
+                wait.decrementAndGet();
+                complete(packet, future);
+            } catch (Exception e) {
+                die("Cant read answer", e);
+                return;
             }
-        } catch (Exception e) {
-            die("Cant init thread", e);
         }
     }
 
@@ -414,7 +414,7 @@ protected void complete(TarantoolPacket packet, CompletableFuture<?> q) {
 
     protected void completeSql(CompletableFuture<?> q, TarantoolPacket pack) {
         Long rowCount = SqlProtoUtils.getSqlRowCount(pack);
-        if (rowCount!=null) {
+        if (rowCount != null) {
             ((CompletableFuture) q).complete(rowCount);
         } else {
             List<Map<String, Object>> values = SqlProtoUtils.readSqlResult(pack);
@@ -477,6 +477,40 @@ protected void stopIO() {
         closeChannel(channel);
     }
 
+    /**
+     * Blocks until a reconnection process can be carried on
+     *
+     * @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;
@@ -499,7 +533,7 @@ public TarantoolClientOps<Integer, List<?>, Object, List<?>> syncOps() {
 
     @Override
     public TarantoolClientOps<Integer, List<?>, Object, Future<List<?>>> asyncOps() {
-        return (TarantoolClientOps)this;
+        return (TarantoolClientOps) this;
     }
 
     @Override
@@ -515,7 +549,7 @@ public TarantoolClientOps<Integer, List<?>, Object, Long> fireAndForgetOps() {
 
     @Override
     public TarantoolSQLOps<Object, Long, List<Map<String, Object>>> sqlSyncOps() {
-        return new TarantoolSQLOps<Object, Long, List<Map<String,Object>>>() {
+        return new TarantoolSQLOps<Object, Long, List<Map<String, Object>>>() {
 
             @Override
             public Long update(String sql, Object... bind) {
@@ -531,7 +565,7 @@ public List<Map<String, Object>> query(String sql, Object... bind) {
 
     @Override
     public TarantoolSQLOps<Object, Future<Long>, Future<List<Map<String, Object>>>> sqlAsyncOps() {
-        return new TarantoolSQLOps<Object, Future<Long>, Future<List<Map<String,Object>>>>() {
+        return new TarantoolSQLOps<Object, Future<Long>, Future<List<Map<String, Object>>>>() {
             @Override
             public Future<Long> update(String sql, Object... bind) {
                 return (Future<Long>) exec(Code.EXECUTE, Key.SQL_TEXT, sql, Key.SQL_BIND, bind);
@@ -619,6 +653,7 @@ public TarantoolClientStats getStats() {
      * Manages state changes.
      */
     protected final class StateHelper {
+        static final int UNINITIALIZED = 0;
         static final int READING = 1;
         static final int WRITING = 2;
         static final int ALIVE = READING | WRITING;
@@ -628,7 +663,7 @@ protected final class StateHelper {
         private final AtomicInteger state;
 
         private final AtomicReference<CountDownLatch> nextAliveLatch =
-            new AtomicReference<CountDownLatch>(new CountDownLatch(1));
+                new AtomicReference<CountDownLatch>(new CountDownLatch(1));
 
         private final CountDownLatch closedLatch = new CountDownLatch(1);
 
@@ -641,7 +676,7 @@ protected int getState() {
         }
 
         protected boolean close() {
-            for (;;) {
+            for (; ; ) {
                 int st = getState();
                 if ((st & CLOSED) == CLOSED)
                     return false;
@@ -651,24 +686,29 @@ protected boolean close() {
         }
 
         protected boolean acquire(int mask) {
-            for (;;) {
-                int st = getState();
-                if ((st & CLOSED) == CLOSED)
+            for (; ; ) {
+                int currentState = getState();
+                if ((currentState & CLOSED) == CLOSED) {
                     return false;
-
-                if ((st & mask) != 0)
+                }
+                if ((currentState & RECONNECT) > mask) {
+                    return false;
+                }
+                if ((currentState & mask) != 0) {
                     throw new IllegalStateException("State is already " + mask);
-
-                if (compareAndSet(st, st | mask))
+                }
+                if (compareAndSet(currentState, currentState | mask)) {
                     return true;
+                }
             }
         }
 
         protected void release(int mask) {
-            for (;;) {
+            for (; ; ) {
                 int st = getState();
-                if (compareAndSet(st, st & ~mask))
+                if (compareAndSet(st, st & ~mask)) {
                     return;
+                }
             }
         }
 
@@ -710,7 +750,7 @@ private CountDownLatch getStateLatch(int state) {
                 CountDownLatch latch = nextAliveLatch.get();
                 /* It may happen so that an error is detected but the state is still alive.
                  Wait for the 'next' alive state in such cases. */
-                return  (getState() == ALIVE && thumbstone == null) ? null : latch;
+                return (getState() == ALIVE && thumbstone == null) ? null : latch;
             }
             return null;
         }