[UNDERTOW-2356] Use state updater to sync up state transitions

core/src/main/java/io/undertow/UndertowLogger.java

@@ -484,4 +484,12 @@ void nodeConfigCreated(URI connectionURI, String balancer, String domain, String
     @LogMessage(level = WARN)
     @Message(id = 5106, value = "Content mismatch for '%s'. Expected length '%s', but was '%s'.")
     void contentEntryMismatch(Object key, long indicatedSize, long written);
-}
+
+    @LogMessage(level = WARN)
+    @Message(id = 5107, value = "Failed to set web socket timeout.")
+    void failedToSetWSTimeout(@Cause Exception e);
+
+    @LogMessage(level = WARN)
+    @Message(id = 5108, value = "Failed to transition to '%s' state in '%s'.")
+    void failedToTransitionToState(String state, Object src);
+}

core/src/main/java/io/undertow/UndertowOptions.java

@@ -19,6 +19,9 @@
 package io.undertow;
 
 import org.xnio.Option;
+import org.xnio.Options;
+import org.xnio.channels.ReadTimeoutException;
+import org.xnio.channels.WriteTimeoutException;
 
 /**
  * @author Stuart Douglas
@@ -112,13 +115,18 @@ public class UndertowOptions {
      */
     public static final Option<Integer> MAX_COOKIES = Option.simple(UndertowOptions.class, "MAX_COOKIES", Integer.class);
 
+    /**
+     * Default value of {@link #ALLOW_ENCODED_SLASH} option.
+     */
+    public static final Boolean DEFAULT_ALLOW_ENCODED_SLASH = Boolean.FALSE;
+
     /**
      * If a request comes in with encoded / characters (i.e. %2F), will these be decoded.
      * <p>
      * This can cause security problems if a front end proxy does not perform the same decoding, and as a result
      * this is disabled by default.
      * <p>
-     * Defaults to false
+     * Defaults to {@link #DEFAULT_ALLOW_ENCODED_SLASH}
      * <p>
      * See <a href="http://cve.mitre.org/cgi-bin/cvename.cgi?name=CVE-2007-0450">CVE-2007-0450</a>
      * @deprecated - this option was interpreted improperly.
@@ -440,6 +448,18 @@ public class UndertowOptions {
      */
     public static final Option<Integer> MAX_RST_FRAMES_PER_WINDOW = Option.simple(UndertowOptions.class, "MAX_RST_STREAMS_PER_TIME_WINDOW", Integer.class);
 
+    /**
+     * Configure a read timeout for a web socket, in milliseconds. If its present it will override {@link org.xnio.Options.READ_TIMEOUT}. If the given amount of time elapses without
+     * a successful read taking place, the socket's next read will throw a {@link ReadTimeoutException}.
+     */
+    public static final Option<Integer> WEB_SOCKETS_READ_TIMEOUT = Option.simple(Options.class, "WEB_SOCKETS_READ_TIMEOUT", Integer.class);
+
+    /**
+     * Configure a write timeout for a web socket, in milliseconds. If its present it will override {@link org.xnio.Options.WRITE_TIMEOUT}. If the given amount of time elapses without
+     * a successful write taking place, the socket's next write will throw a {@link WriteTimeoutException}.
+     */
+    public static final Option<Integer> WEB_SOCKETS_WRITE_TIMEOUT = Option.simple(Options.class, "WEB_SOCKETS_WRITE_TIMEOUT", Integer.class);
+
     private UndertowOptions() {
 
     }

core/src/main/java/io/undertow/client/ajp/AjpClientConnection.java

@@ -35,6 +35,7 @@
 import java.util.List;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import io.undertow.client.ClientStatistics;
 import org.jboss.logging.Logger;
@@ -103,23 +104,30 @@ public void handleEvent(AjpClientResponseStreamSourceChannel channel) {
     private static final int CLOSE_REQ = 1 << 30;
     private static final int CLOSED = 1 << 31;
 
-    private int state;
+    @SuppressWarnings("unused")
+    private volatile int state;
+    private static final AtomicIntegerFieldUpdater<AjpClientConnection> stateUpdater = AtomicIntegerFieldUpdater.newUpdater(AjpClientConnection.class, "state");
 
     private final ChannelListener.SimpleSetter<AjpClientConnection> closeSetter = new ChannelListener.SimpleSetter<>();
     private final ClientStatistics clientStatistics;
     private final List<ChannelListener<ClientConnection>> closeListeners = new CopyOnWriteArrayList<>();
 
-    AjpClientConnection(final AjpClientChannel connection, final OptionMap options, final ByteBufferPool bufferPool, ClientStatistics clientStatistics) {
+    AjpClientConnection(final AjpClientChannel channel, final OptionMap options, final ByteBufferPool bufferPool, ClientStatistics clientStatistics) {
         this.clientStatistics = clientStatistics;
         this.options = options;
-        this.connection = connection;
+        this.connection = channel;
         this.bufferPool = bufferPool;
 
-        connection.addCloseTask(new ChannelListener<AjpClientChannel>() {
+        channel.addCloseTask(new ChannelListener<AjpClientChannel>() {
             @Override
             public void handleEvent(AjpClientChannel channel) {
                 log.debugf("connection to %s closed", getPeerAddress());
-                AjpClientConnection.this.state |= CLOSED;
+                final int oldVal = stateUpdater.getAndAccumulate(AjpClientConnection.this, CLOSED, (currentState, flag)-> currentState | flag);
+                if(anyAreSet(oldVal, CLOSED)) {
+                    //this was closed already?
+                    UndertowLogger.ROOT_LOGGER.failedToTransitionToState("CLOSED", AjpClientConnection.this);
+                    return;
+                }
                 ChannelListeners.invokeChannelListener(AjpClientConnection.this, closeSetter.get());
                 for(ChannelListener<ClientConnection> listener : closeListeners) {
                     listener.handleEvent(AjpClientConnection.this);
@@ -135,8 +143,8 @@ public void handleEvent(AjpClientChannel channel) {
                 }
             }
         });
-        connection.getReceiveSetter().set(new ClientReceiveListener());
-        connection.resumeReceives();
+        channel.getReceiveSetter().set(new ClientReceiveListener());
+        channel.resumeReceives();
     }
 
     @Override
@@ -236,7 +244,7 @@ public void sendRequest(final ClientRequest request, final ClientCallback<Client
         if (anyAreSet(state, UPGRADE_REQUESTED | UPGRADED)) {
             clientCallback.failed(UndertowClientMessages.MESSAGES.invalidConnectionState());
             return;
-        } else if (anyAreSet(state, CLOSE_REQ | CLOSED)) {
+        } else if (anyAreSet(state, CLOSED | CLOSE_REQ)) {
             clientCallback.failed(UndertowClientMessages.MESSAGES.closedConnectionState());
             return;
         }
@@ -266,13 +274,13 @@ private void initiateRequest(AjpClientExchange AjpClientExchange) {
         String connectionString = request.getRequestHeaders().getFirst(CONNECTION);
         if (connectionString != null) {
             if (CLOSE.equalToString(connectionString)) {
-                state |= CLOSE_REQ;
+                stateUpdater.getAndAccumulate(AjpClientConnection.this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
             }
         } else if (request.getProtocol() != Protocols.HTTP_1_1) {
-            state |= CLOSE_REQ;
+            stateUpdater.getAndAccumulate(AjpClientConnection.this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
         }
         if (request.getRequestHeaders().contains(UPGRADE)) {
-            state |= UPGRADE_REQUESTED;
+            stateUpdater.getAndAccumulate(AjpClientConnection.this, UPGRADE_REQUESTED, (currentState, flag)-> currentState | flag);
         }
 
         long length = 0;
@@ -327,7 +335,7 @@ public void close() throws IOException {
         if (anyAreSet(state, CLOSED)) {
             return;
         }
-        state |= CLOSED | CLOSE_REQ;
+        stateUpdater.accumulateAndGet(this, CLOSED | CLOSE_REQ, (currentState, flag)-> currentState | flag);
         connection.close();
     }
 
@@ -336,7 +344,6 @@ public void close() throws IOException {
      */
     public void requestDone() {
         currentRequest = null;
-
         if (anyAreSet(state, CLOSE_REQ)) {
             safeClose(connection);
         } else if (anyAreSet(state, UPGRADE_REQUESTED)) {
@@ -352,7 +359,7 @@ public void requestDone() {
     }
 
     public void requestClose() {
-        state |= CLOSE_REQ;
+        stateUpdater.getAndAccumulate(AjpClientConnection.this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
     }
 
 

core/src/main/java/io/undertow/client/ajp/AjpClientExchange.java

@@ -35,6 +35,7 @@
 import org.xnio.channels.StreamSourceChannel;
 
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import static org.xnio.Bits.anyAreSet;
 
@@ -58,7 +59,9 @@ class AjpClientExchange extends AbstractAttachable implements ClientExchange {
     private AjpClientResponseStreamSourceChannel responseChannel;
     private AjpClientRequestClientStreamSinkChannel requestChannel;
 
-    private int state = 0;
+    @SuppressWarnings("unused")
+    private volatile int state = 0;
+    private static final AtomicIntegerFieldUpdater<AjpClientExchange> stateUpdater = AtomicIntegerFieldUpdater.newUpdater(AjpClientExchange.class, "state");
     private static final int REQUEST_TERMINATED = 1;
     private static final int RESPONSE_TERMINATED = 1 << 1;
 
@@ -78,19 +81,19 @@ class AjpClientExchange extends AbstractAttachable implements ClientExchange {
     }
 
     void terminateRequest() {
-        state |= REQUEST_TERMINATED;
+        stateUpdater.accumulateAndGet(this, REQUEST_TERMINATED, (currentState, flag)-> currentState | flag);
         if(!clientConnection.isOpen()) {
-            state |= RESPONSE_TERMINATED;
+            stateUpdater.accumulateAndGet(this, RESPONSE_TERMINATED, (currentState, flag)-> currentState | flag);
         }
         if (anyAreSet(state, RESPONSE_TERMINATED)) {
             clientConnection.requestDone();
         }
     }
 
     void terminateResponse() {
-        state |= RESPONSE_TERMINATED;
+        stateUpdater.accumulateAndGet(this, RESPONSE_TERMINATED, (currentState, flag)-> currentState | flag);
         if(!clientConnection.isOpen()) {
-            state |= REQUEST_TERMINATED;
+            stateUpdater.accumulateAndGet(this, REQUEST_TERMINATED, (currentState, flag)-> currentState | flag);
         }
         if (anyAreSet(state, REQUEST_TERMINATED)) {
             clientConnection.requestDone();
@@ -155,7 +158,7 @@ public StreamSinkChannel getRequestChannel() {
         return new DetachableStreamSinkChannel(requestChannel) {
             @Override
             protected boolean isFinished() {
-                return anyAreSet(state, REQUEST_TERMINATED);
+                return anyAreSet(AjpClientExchange.this.state, REQUEST_TERMINATED);
             }
         };
     }
@@ -165,7 +168,7 @@ public StreamSourceChannel getResponseChannel() {
         return new DetachableStreamSourceChannel(responseChannel) {
             @Override
             protected boolean isFinished() {
-                return anyAreSet(state, RESPONSE_TERMINATED);
+                return anyAreSet(AjpClientExchange.this.state, RESPONSE_TERMINATED);
             }
         };
     }

core/src/main/java/io/undertow/client/http/HttpClientConnection.java

@@ -83,6 +83,7 @@
 import java.util.List;
 import java.util.Locale;
 import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
 import static io.undertow.client.UndertowClientMessages.MESSAGES;
 import static org.xnio.Bits.allAreClear;
@@ -132,7 +133,9 @@ public void handleEvent(StreamSourceConduit channel) {
     private static final int CLOSE_REQ = 1 << 30;
     private static final int CLOSED = 1 << 31;
 
-    private int state;
+    @SuppressWarnings("unused")
+    private volatile int state;
+    private static final AtomicIntegerFieldUpdater<HttpClientConnection> stateUpdater = AtomicIntegerFieldUpdater.newUpdater(HttpClientConnection.class, "state");
     private final ChannelListener.SimpleSetter<HttpClientConnection> closeSetter = new ChannelListener.SimpleSetter<>();
 
     private final ClientStatistics clientStatistics;
@@ -178,7 +181,13 @@ public void activity(long bytes) {
 
             public void handleEvent(StreamConnection channel) {
                 log.debugf("connection to %s closed", getPeerAddress());
-                HttpClientConnection.this.state |= CLOSED;
+                final int oldVal = stateUpdater.getAndAccumulate(HttpClientConnection.this, CLOSED, (currentState, flag)-> currentState | flag);
+                if(anyAreSet(oldVal, CLOSED)) {
+                    //this was closed already?
+                    UndertowLogger.ROOT_LOGGER.failedToTransitionToState("CLOSED", HttpClientConnection.this);
+                    //NOTE: cant do that....
+                    //return;
+                }
                 ChannelListeners.invokeChannelListener(HttpClientConnection.this, closeSetter.get());
                 try {
                     if (pooledBuffer != null) {
@@ -256,7 +265,7 @@ public boolean isOpen() {
         if(http2Delegate != null) {
             return http2Delegate.isOpen();
         }
-        return connection.isOpen() && allAreClear(state, CLOSE_REQ | CLOSED);
+        return connection.isOpen() && allAreClear(state, CLOSED | CLOSE_REQ);
     }
 
     @Override
@@ -348,10 +357,10 @@ public void sendRequest(final ClientRequest request, final ClientCallback<Client
             http2Delegate.sendRequest(request, clientCallback);
             return;
         }
-        if (anyAreSet(state, UPGRADE_REQUESTED | UPGRADED)) {
+        if (anyAreSet(state,UPGRADE_REQUESTED | UPGRADED)) {
             clientCallback.failed(UndertowClientMessages.MESSAGES.invalidConnectionState());
             return;
-        } else if (anyAreSet(state, CLOSE_REQ | CLOSED)) {
+        } else if (anyAreSet(state, CLOSED | CLOSE_REQ)) {
             clientCallback.failed(UndertowClientMessages.MESSAGES.closedConnectionState());
             return;
         }
@@ -381,19 +390,19 @@ private void initiateRequest(HttpClientExchange httpClientExchange) {
         String connectionString = request.getRequestHeaders().getFirst(Headers.CONNECTION);
         if (connectionString != null) {
             if (Headers.CLOSE.equalToString(connectionString)) {
-                state |= CLOSE_REQ;
+                stateUpdater.accumulateAndGet(HttpClientConnection.this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
             } else if (Headers.UPGRADE.equalToString(connectionString)) {
-                state |= UPGRADE_REQUESTED;
+                stateUpdater.accumulateAndGet(HttpClientConnection.this, UPGRADE_REQUESTED, (currentState, flag)-> currentState | flag);
             }
         } else if (request.getProtocol() != Protocols.HTTP_1_1) {
-            state |= CLOSE_REQ;
+            stateUpdater.accumulateAndGet(HttpClientConnection.this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
         }
         if (request.getRequestHeaders().contains(Headers.UPGRADE)) {
-            state |= UPGRADE_REQUESTED;
+            stateUpdater.accumulateAndGet(HttpClientConnection.this, UPGRADE_REQUESTED, (currentState, flag)-> currentState | flag);
         }
         if(request.getMethod().equals(Methods.CONNECT)) {
             //we treat CONNECT like upgrade requests
-            state |= UPGRADE_REQUESTED;
+            stateUpdater.accumulateAndGet(HttpClientConnection.this, UPGRADE_REQUESTED, (currentState, flag)-> currentState | flag);
         }
 
         //setup the client request conduits
@@ -476,7 +485,7 @@ public StreamConnection performUpgrade() throws IOException {
         if (allAreSet(state, UPGRADED | CLOSE_REQ | CLOSED)) {
             throw new IOException(UndertowClientMessages.MESSAGES.connectionClosed());
         }
-        state |= UPGRADED;
+        stateUpdater.accumulateAndGet(this, UPGRADED, (currentState, flag)-> currentState | flag);
         connection.getSinkChannel().setConduit(originalSinkConduit);
         connection.getSourceChannel().setConduit(pushBackStreamSourceConduit);
         return connection;
@@ -490,7 +499,7 @@ public void close() throws IOException {
         if (anyAreSet(state, CLOSED)) {
             return;
         }
-        state |= CLOSED | CLOSE_REQ;
+        stateUpdater.accumulateAndGet(this, CLOSED | CLOSE_REQ, (currentState, flag)-> currentState | flag);
         ConnectionUtils.cleanClose(connection);
     }
 
@@ -508,7 +517,7 @@ public void exchangeDone() {
         if (anyAreSet(state, CLOSE_REQ)) {
             currentRequest = null;
             pendingResponse = null;
-            this.state |= CLOSED;
+            stateUpdater.accumulateAndGet(this, CLOSED, (currentState, flag)-> currentState | flag);
             safeClose(connection);
         } else if (anyAreSet(state, UPGRADE_REQUESTED)) {
             connection.getSourceChannel().suspendReads();
@@ -630,7 +639,7 @@ public void handleEvent(StreamSourceChannel channel) {
                     if ((connectionString == null || !Headers.UPGRADE.equalToString(connectionString)) && !response.getResponseHeaders().contains(Headers.UPGRADE)) {
                         if(!currentRequest.getRequest().getMethod().equals(Methods.CONNECT) || response.getResponseCode() != 200) { //make sure it was not actually a connect request
                             //just unset the upgrade requested flag
-                            HttpClientConnection.this.state &= ~UPGRADE_REQUESTED;
+                            stateUpdater.accumulateAndGet(HttpClientConnection.this, ~UPGRADE_REQUESTED, (currentState, flag)-> currentState & flag);
                         }
                     }
                 }
@@ -647,7 +656,7 @@ public void handleEvent(StreamSourceChannel channel) {
                     close = true;
                 }
                 if(close) {
-                    HttpClientConnection.this.state |= CLOSE_REQ;
+                    stateUpdater.accumulateAndGet(HttpClientConnection.this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
                     //we are going to close, kill any queued connections
                     HttpClientExchange ex = pendingQueue.poll();
                     while (ex != null) {
@@ -673,7 +682,7 @@ public void handleEvent(StreamSourceChannel channel) {
                     currentRequest.setResponse(response);
                     if(response.getResponseCode() == StatusCodes.EXPECTATION_FAILED) {
                         if(HttpContinue.requiresContinueResponse(currentRequest.getRequest().getRequestHeaders())) {
-                            HttpClientConnection.this.state |= CLOSE_REQ;
+                            stateUpdater.accumulateAndGet(HttpClientConnection.this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
                             ConduitStreamSinkChannel sinkChannel = HttpClientConnection.this.connection.getSinkChannel();
                             sinkChannel.shutdownWrites();
                             if(!sinkChannel.flush()) {
@@ -749,7 +758,7 @@ private void prepareResponseChannel(ClientResponse response, ClientExchange exch
             connection.getSourceChannel().setConduit(new FixedLengthStreamSourceConduit(connection.getSourceChannel().getConduit(), 0, responseFinishedListener));
         } else {
             connection.getSourceChannel().setConduit(new FinishableStreamSourceConduit(connection.getSourceChannel().getConduit(), responseFinishedListener));
-            state |= CLOSE_REQ;
+            stateUpdater.accumulateAndGet(this, CLOSE_REQ, (currentState, flag)-> currentState | flag);
         }
     }