You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by kw...@apache.org on 2015/10/23 13:57:58 UTC
svn commit: r1710188 - in /qpid/java/trunk/broker-core/src:
main/java/org/apache/qpid/server/transport/SelectorThread.java
test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
Author: kwall
Date: Fri Oct 23 11:57:57 2015
New Revision: 1710188
URL: http://svn.apache.org/viewvc?rev=1710188&view=rev
Log:
QPID-6794: Fix race condition that may cause selector wakeups to be missed
Modified:
qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/SelectorThread.java
qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
Modified: qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/SelectorThread.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/SelectorThread.java?rev=1710188&r1=1710187&r2=1710188&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/SelectorThread.java (original)
+++ qpid/java/trunk/broker-core/src/main/java/org/apache/qpid/server/transport/SelectorThread.java Fri Oct 23 11:57:57 2015
@@ -220,6 +220,7 @@ class SelectorThread extends Thread
if (!_closed.get())
{
Thread.currentThread().setName("Selector-" + _scheduler.getName());
+ _inSelect.set(true);
try
{
if(_wakeups.getAndSet(0) > 0)
@@ -228,13 +229,8 @@ class SelectorThread extends Thread
}
else
{
- _inSelect.set(true);
-
_selector.select(_nextTimeout);
-
- _inSelect.set(false);
}
-
}
catch (IOException e)
{
@@ -243,6 +239,10 @@ class SelectorThread extends Thread
closeSelector();
return;
}
+ finally
+ {
+ _inSelect.set(false);
+ }
runTasks();
for (NonBlockingConnection connection : processSelectionKeys())
{
@@ -346,13 +346,20 @@ class SelectorThread extends Thread
try
{
+ if (LOGGER.isDebugEnabled())
+ {
+ LOGGER.debug("Registering selector on accepting port {} ",
+ socketChannel.socket().getLocalSocketAddress());
+ }
socketChannel.register(_selectionTasks[0].getSelector(), SelectionKey.OP_ACCEPT, nonBlockingNetworkTransport);
}
catch (IllegalStateException | ClosedChannelException e)
{
// TODO Communicate condition back to model object to make it go into the ERROR state
- LOGGER.error("Failed to register selector on accepting port", e);
- } }
+ LOGGER.error("Failed to register selector on accepting port {} ",
+ socketChannel.socket().getLocalSocketAddress(), e);
+ }
+ }
});
_selectionTasks[0].wakeup();
}
Modified: qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java
URL: http://svn.apache.org/viewvc/qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java?rev=1710188&r1=1710187&r2=1710188&view=diff
==============================================================================
--- qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java (original)
+++ qpid/java/trunk/broker-core/src/test/java/org/apache/qpid/server/transport/TCPandSSLTransportTest.java Fri Oct 23 11:57:57 2015
@@ -106,6 +106,7 @@ public class TCPandSSLTransportTest exte
final AmqpPort<?> port = mock(AmqpPort.class);
when(port.getPort()).thenReturn(0);
+ when(port.getName()).thenReturn("testAmqp");
when(port.getNetworkBufferSize()).thenReturn(64*1024);
when(port.canAcceptNewConnection(any(SocketAddress.class))).thenReturn(true);
when(port.getThreadPoolMinimum()).thenReturn(2);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org