From ecaeaf7615777ba147dd2dacc4720ab38c482ac5 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Tue, 23 Jun 2009 17:55:39 +0000 Subject: [PATCH] Under heavy load and large column values, we still saw lockups in tcp connection. Here is the problem. The following code that sets the interest ops seems innocent, but it's the source of the problem. The reason is that this operation is not atomic. Another thread could sneak in between the reading of the ops and the setting of it. As a result, some wrong bits could be set. key_.interestOps(key_.interestOps() | SelectionKey.OP_READ) This is a sequence that demonstrates how we can lose the OP_READ bit forever and thus jam the read channel: 1. Thread 1: we want to write a message and in write(Message) we are about to turn on OP_WRITE because the message can't be written in one shot. 2. Thread 2: a read comes in and in read(SelectionKey), we turn off OP_READ and submit the read request to ReadWorkItem in Thread 3. 3. Thread 1: read interestOps and see OP_READ as off. 4. Thread 3: finished processing the read request and turn OP_READ on 5. Thread 1: resumes and turn on OP_WRITE. However, by doing that, we also turned off OP_READ. The read channel is thus blocked forever after this. patch by Jun Rao; reviewed by jbellis for CASSANDRA-220 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/branches/cassandra-0.3@787765 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/net/SelectionKeyHandler.java | 16 +++++++++++ .../apache/cassandra/net/TcpConnection.java | 27 ++++++++++--------- .../apache/cassandra/net/UdpConnection.java | 4 +-- .../cassandra/net/http/HttpConnection.java | 4 +-- 4 files changed, 35 insertions(+), 16 deletions(-) diff --git a/src/java/org/apache/cassandra/net/SelectionKeyHandler.java b/src/java/org/apache/cassandra/net/SelectionKeyHandler.java index 0048b7bce7..5f9605ab24 100644 --- a/src/java/org/apache/cassandra/net/SelectionKeyHandler.java +++ b/src/java/org/apache/cassandra/net/SelectionKeyHandler.java @@ -65,4 +65,20 @@ public class SelectionKeyHandler { throw new UnsupportedOperationException("write() cannot be called on " + getClass().getName() + "!"); } + + protected static void turnOnInterestOps(SelectionKey key, int ops) + { + synchronized(key) + { + key.interestOps(key.interestOps() | ops); + } + } + + protected static void turnOffInterestOps(SelectionKey key, int ops) + { + synchronized(key) + { + key.interestOps(key.interestOps() & (~ops) ); + } + } } diff --git a/src/java/org/apache/cassandra/net/TcpConnection.java b/src/java/org/apache/cassandra/net/TcpConnection.java index 99c0f9f3ec..c87127e5a6 100644 --- a/src/java/org/apache/cassandra/net/TcpConnection.java +++ b/src/java/org/apache/cassandra/net/TcpConnection.java @@ -183,7 +183,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable if (buffer.remaining() > 0) { pendingWrites_.add(buffer); - key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE); + turnOnInterestOps(key_, SelectionKey.OP_WRITE); } } } @@ -229,7 +229,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable if (buffer.remaining() > 0) { pendingWrites_.add(buffer); - key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE); + turnOnInterestOps(key_, SelectionKey.OP_WRITE); condition_.await(); } } @@ -245,7 +245,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable */ if ( bytesTransferred < limit && bytesWritten != total ) { - key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE); + turnOnInterestOps(key_, SelectionKey.OP_WRITE); condition_.await(); } } @@ -346,17 +346,20 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable // called in the selector thread public void connect(SelectionKey key) { - key.interestOps(key.interestOps() & (~SelectionKey.OP_CONNECT)); + turnOffInterestOps(key, SelectionKey.OP_CONNECT); try { if (socketChannel_.finishConnect()) { - key.interestOps(key.interestOps() | SelectionKey.OP_READ); + turnOnInterestOps(key, SelectionKey.OP_READ); - // this will flush the pending - if (!pendingWrites_.isEmpty()) + synchronized(this) { - key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE); + // this will flush the pending + if (!pendingWrites_.isEmpty()) + { + turnOnInterestOps(key_, SelectionKey.OP_WRITE); + } } resumeStreaming(); } @@ -376,7 +379,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable // called in the selector thread public void write(SelectionKey key) { - key.interestOps( key.interestOps() & ( ~SelectionKey.OP_WRITE ) ); + turnOffInterestOps(key, SelectionKey.OP_WRITE); doPendingWrites(); /* * This is executed only if we are in streaming mode. @@ -415,7 +418,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable { if (!pendingWrites_.isEmpty()) { - key_.interestOps(key_.interestOps() | SelectionKey.OP_WRITE); + turnOnInterestOps(key_, SelectionKey.OP_WRITE); } } } @@ -424,7 +427,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable // called in the selector thread public void read(SelectionKey key) { - key.interestOps( key.interestOps() & ( ~SelectionKey.OP_READ ) ); + turnOffInterestOps(key, SelectionKey.OP_READ); // publish this event onto to the TCPReadEvent Queue. MessagingService.getReadExecutor().execute(readWork_); } @@ -486,7 +489,7 @@ public class TcpConnection extends SelectionKeyHandler implements Comparable } finally { - key_.interestOps(key_.interestOps() | SelectionKey.OP_READ); + turnOnInterestOps(key_, SelectionKey.OP_READ); } } diff --git a/src/java/org/apache/cassandra/net/UdpConnection.java b/src/java/org/apache/cassandra/net/UdpConnection.java index 53fddbb88b..177b73bf05 100644 --- a/src/java/org/apache/cassandra/net/UdpConnection.java +++ b/src/java/org/apache/cassandra/net/UdpConnection.java @@ -131,7 +131,7 @@ public class UdpConnection extends SelectionKeyHandler public void read(SelectionKey key) { - key.interestOps( key.interestOps() & (~SelectionKey.OP_READ) ); + turnOffInterestOps(key, SelectionKey.OP_READ); ByteBuffer buffer = ByteBuffer.allocate(BUFFER_SIZE); try { @@ -160,7 +160,7 @@ public class UdpConnection extends SelectionKeyHandler } finally { - key.interestOps( key_.interestOps() | SelectionKey.OP_READ ); + turnOnInterestOps(key_, SelectionKey.OP_READ ); } } } diff --git a/src/java/org/apache/cassandra/net/http/HttpConnection.java b/src/java/org/apache/cassandra/net/http/HttpConnection.java index d0ce82846c..ce6ff4dc1f 100644 --- a/src/java/org/apache/cassandra/net/http/HttpConnection.java +++ b/src/java/org/apache/cassandra/net/http/HttpConnection.java @@ -136,7 +136,7 @@ public class HttpConnection extends SelectionKeyHandler implements HttpStartLine httpChannel_ = (SocketChannel)key.channel(); } /* deregister interest for read */ - key.interestOps( key.interestOps() & ( ~SelectionKey.OP_READ ) ); + turnOffInterestOps(key, SelectionKey.OP_READ); /* Add a task to process the HTTP request */ MessagingService.getReadExecutor().execute(httpReader_); } @@ -330,7 +330,7 @@ public class HttpConnection extends SelectionKeyHandler implements HttpStartLine } finally { - httpKey_.interestOps(httpKey_.interestOps() | SelectionKey.OP_READ); + turnOnInterestOps(httpKey_, SelectionKey.OP_READ); } }