You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by jo...@apache.org on 2022/04/01 23:02:20 UTC
[cassandra] branch trunk updated: Streaming sessions longer than 3 minutes fail with timeout
This is an automated email from the ASF dual-hosted git repository.
jonmeredith pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/trunk by this push:
new fa7185e Streaming sessions longer than 3 minutes fail with timeout
fa7185e is described below
commit fa7185ef02c313ff322fcc50d2de03186b9f96a7
Author: Jon Meredith <jo...@apache.org>
AuthorDate: Fri Apr 1 12:58:02 2022 -0600
Streaming sessions longer than 3 minutes fail with timeout
patch by Jon Meredith; reviewed by Benedict Elliott Smith, Caleb Rackliffe for CASSANDRA-17510
---
CHANGES.txt | 1 +
.../sstable/format/big/BigTableZeroCopyWriter.java | 3 +-
.../cassandra/net/AsyncStreamingInputPlus.java | 100 ++++++++-----------
.../streaming/async/NettyStreamingChannel.java | 2 +-
.../cassandra/net/AsyncStreamingInputPlusTest.java | 111 +++++++++++----------
5 files changed, 99 insertions(+), 118 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 87deaf9..68de34f 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
4.1
+ * Streaming sessions longer than 3 minutes fail with timeout (CASSANDRA-17510)
* Add ability to track state in repair (CASSANDRA-15399)
* Remove unused 'parse' module (CASSANDRA-17484)
* change six functions in cqlshlib to native Python 3 (CASSANDRA-17417)
diff --git a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java
index c47c23b..e4a382d 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java
@@ -218,8 +218,7 @@ public class BigTableZeroCopyWriter extends SSTable implements SSTableMultiWrite
in.consume(writer::writeDirectlyToChannel, size);
writer.sync();
}
- // FIXME: handle ACIP exceptions properly
- catch (EOFException | AsyncStreamingInputPlus.InputTimeoutException e)
+ catch (EOFException e)
{
in.close();
}
diff --git a/src/java/org/apache/cassandra/net/AsyncStreamingInputPlus.java b/src/java/org/apache/cassandra/net/AsyncStreamingInputPlus.java
index b964dda..5e936ac 100644
--- a/src/java/org/apache/cassandra/net/AsyncStreamingInputPlus.java
+++ b/src/java/org/apache/cassandra/net/AsyncStreamingInputPlus.java
@@ -22,7 +22,6 @@ import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.util.concurrent.BlockingQueue;
-import java.util.concurrent.TimeUnit;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.primitives.Ints;
@@ -36,15 +35,11 @@ import org.apache.cassandra.streaming.StreamingDataInputPlus;
import static org.apache.cassandra.utils.concurrent.BlockingQueues.newBlockingQueue;
-// TODO: rewrite
+/*
+ * This class expects a single producer (Netty event loop) and single consumer thread (StreamingDeserializerTask).
+ */
public class AsyncStreamingInputPlus extends RebufferingInputStream implements StreamingDataInputPlus
{
- public static class InputTimeoutException extends IOException
- {
- }
-
- private static final long DEFAULT_REBUFFER_BLOCK_IN_MILLIS = TimeUnit.MINUTES.toMillis(3);
-
private final Channel channel;
/**
@@ -54,22 +49,15 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
private final BlockingQueue<ByteBuf> queue;
- private final long rebufferTimeoutNanos;
-
- private volatile boolean isClosed;
+ private boolean isProducerClosed = false;
+ private boolean isConsumerClosed = false;
public AsyncStreamingInputPlus(Channel channel)
{
- this(channel, DEFAULT_REBUFFER_BLOCK_IN_MILLIS, TimeUnit.MILLISECONDS);
- }
-
- AsyncStreamingInputPlus(Channel channel, long rebufferTimeout, TimeUnit rebufferTimeoutUnit)
- {
super(Unpooled.EMPTY_BUFFER.nioBuffer());
currentBuf = Unpooled.EMPTY_BUFFER;
queue = newBlockingQueue();
- rebufferTimeoutNanos = rebufferTimeoutUnit.toNanos(rebufferTimeout);
this.channel = channel;
channel.config().setAutoRead(false);
@@ -82,18 +70,11 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
*/
public boolean append(ByteBuf buf) throws IllegalStateException
{
- if (isClosed) return false;
+ if (isProducerClosed)
+ return false; // buf should be released in NettyStreamingChannel.channelRead
queue.add(buf);
- /*
- * it's possible for append() to race with close(), so we need to ensure
- * that the bytebuf gets released in that scenario
- */
- if (isClosed)
- while ((buf = queue.poll()) != null)
- buf.release();
-
return true;
}
@@ -102,16 +83,17 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
*
* Release open buffers and poll the {@link #queue} for more data.
* <p>
- * This is best, and more or less expected, to be invoked on a consuming thread (not the event loop)
- * becasue if we block on the queue we can't fill it on the event loop (as that's where the buffers are coming from).
+ * This is invoked on a consuming thread (not the event loop)
+ * because if we block on the queue we can't fill it on the event loop (as that's where the buffers are coming from).
*
- * @throws EOFException when no further reading from this instance should occur. Implies this instance is closed.
- * @throws InputTimeoutException when no new buffers arrive for reading before
- * the {@link #rebufferTimeoutNanos} elapses while blocking. It's then not safe to reuse this instance again.
+ * @throws ClosedChannelException when no further reading from this instance should occur. Implies this instance is closed.
*/
@Override
- protected void reBuffer() throws ClosedChannelException, InputTimeoutException
+ protected void reBuffer() throws ClosedChannelException
{
+ if (isConsumerClosed)
+ throw new ClosedChannelException();
+
if (queue.isEmpty())
channel.read();
@@ -120,20 +102,23 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
buffer = null;
ByteBuf next = null;
- try
- {
- next = queue.poll(rebufferTimeoutNanos, TimeUnit.NANOSECONDS);
- }
- catch (InterruptedException ie)
+ do
{
- // nop
- }
-
- if (null == next)
- throw new InputTimeoutException();
+ try
+ {
+ next = queue.take(); // rely on sentinel being sent to terminate this loop
+ }
+ catch (InterruptedException ie)
+ {
+ // ignore interruptions, retry and rely on being shut down by requestClosure
+ }
+ } while (next == null);
- if (next == Unpooled.EMPTY_BUFFER) // Unpooled.EMPTY_BUFFER is the indicator that the input is closed
+ if (next == Unpooled.EMPTY_BUFFER) // the indicator that the input is closed
+ {
+ isConsumerClosed = true;
throw new ClosedChannelException();
+ }
currentBuf = next;
buffer = next.nioBuffer();
@@ -186,17 +171,9 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
return Ints.checkedCast(count);
}
- // TODO:JEB add docs
- // TL;DR if there's no Bufs open anywhere here, issue a channle read to try and grab data.
- public void maybeIssueRead()
- {
- if (isEmpty())
- channel.read();
- }
-
public boolean isEmpty()
{
- return queue.isEmpty() && (buffer == null || !buffer.hasRemaining());
+ return isConsumerClosed || (queue.isEmpty() && (buffer == null || !buffer.hasRemaining()));
}
/**
@@ -207,9 +184,11 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
@Override
public void close()
{
- if (isClosed)
+ if (isConsumerClosed)
return;
+ isConsumerClosed = true;
+
if (currentBuf != null)
{
currentBuf.release();
@@ -221,19 +200,16 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
{
try
{
- ByteBuf buf = queue.poll(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
+ ByteBuf buf = queue.take();
if (buf == Unpooled.EMPTY_BUFFER)
break;
- else
- buf.release();
+ buf.release();
}
catch (InterruptedException e)
{
- //
+ // ignore and rely on requestClose having been called
}
}
-
- isClosed = true;
}
/**
@@ -243,7 +219,11 @@ public class AsyncStreamingInputPlus extends RebufferingInputStream implements S
*/
public void requestClosure()
{
- queue.add(Unpooled.EMPTY_BUFFER);
+ if (!isProducerClosed)
+ {
+ queue.add(Unpooled.EMPTY_BUFFER);
+ isProducerClosed = true;
+ }
}
// TODO: let's remove this like we did for AsyncChannelOutputPlus
diff --git a/src/java/org/apache/cassandra/streaming/async/NettyStreamingChannel.java b/src/java/org/apache/cassandra/streaming/async/NettyStreamingChannel.java
index fb5d37a..70b1c37 100644
--- a/src/java/org/apache/cassandra/streaming/async/NettyStreamingChannel.java
+++ b/src/java/org/apache/cassandra/streaming/async/NettyStreamingChannel.java
@@ -68,7 +68,7 @@ public class NettyStreamingChannel extends ChannelInboundHandlerAdapter implemen
* <p>
* For thread safety, this structure's resources are released on the consuming thread
* (via {@link AsyncStreamingInputPlus#close()},
- * but the producing side calls {@link AsyncStreamingInputPlus#requestClosure()} to notify the input that is should close.
+ * but the producing side calls {@link AsyncStreamingInputPlus#requestClosure()} to notify the input that it should close.
*/
@VisibleForTesting
final AsyncStreamingInputPlus in;
diff --git a/test/unit/org/apache/cassandra/net/AsyncStreamingInputPlusTest.java b/test/unit/org/apache/cassandra/net/AsyncStreamingInputPlusTest.java
index 417a51c..1d6dcf4 100644
--- a/test/unit/org/apache/cassandra/net/AsyncStreamingInputPlusTest.java
+++ b/test/unit/org/apache/cassandra/net/AsyncStreamingInputPlusTest.java
@@ -18,11 +18,11 @@
package org.apache.cassandra.net;
-import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.WritableByteChannel;
+import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;
import org.junit.After;
@@ -34,11 +34,9 @@ import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.embedded.EmbeddedChannel;
import org.apache.cassandra.io.util.BufferedDataOutputStreamPlus;
-import org.apache.cassandra.net.AsyncStreamingInputPlus;
-import org.apache.cassandra.net.AsyncStreamingInputPlus.InputTimeoutException;
-import static org.apache.cassandra.utils.Clock.Global.nanoTime;
import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.fail;
public class AsyncStreamingInputPlusTest
{
@@ -61,14 +59,6 @@ public class AsyncStreamingInputPlusTest
buf.release(buf.refCnt());
}
-// @Test
-// public void isOpen()
-// {
-// Assert.assertTrue(inputPlus.isOpen());
-// inputPlus.requestClosure();
-// Assert.assertFalse(inputPlus.isOpen());
-// }
-
@Test
public void append_closed()
{
@@ -106,30 +96,8 @@ public class AsyncStreamingInputPlusTest
buf.writerIndex(8);
inputPlus.append(buf);
Assert.assertEquals(16, inputPlus.unsafeAvailable());
-
-// ByteBuffer out = ByteBuffer.allocate(4);
-// int readCount = inputPlus.read(out);
-// Assert.assertEquals(4, readCount);
-// out.flip();
-// Assert.assertEquals(42, out.getInt());
-// Assert.assertEquals(12, inputPlus.unsafeAvailable());
-
-// out = ByteBuffer.allocate(8);
-// readCount = inputPlus.read(out);
-// Assert.assertEquals(8, readCount);
-// out.flip();
-// Assert.assertEquals(42, out.getLong());
-// Assert.assertEquals(4, inputPlus.unsafeAvailable());
}
-// @Test (expected = EOFException.class)
-// public void read_closed() throws IOException
-// {
-// inputPlus.requestClosure();
-// ByteBuffer buf = ByteBuffer.allocate(1);
-// inputPlus.read(buf);
-// }
-
@Test
public void available_closed()
{
@@ -162,6 +130,60 @@ public class AsyncStreamingInputPlusTest
}
@Test
+ public void rebufferAndCloseToleratesInterruption() throws InterruptedException
+ {
+ ByteBuf beforeInterrupt = channel.alloc().heapBuffer(1024);
+ beforeInterrupt.writeCharSequence("BEFORE", StandardCharsets.US_ASCII);
+ ByteBuf afterInterrupt = channel.alloc().heapBuffer(1024);
+ afterInterrupt.writeCharSequence("AFTER", StandardCharsets.US_ASCII);
+ final int totalBytes = beforeInterrupt.readableBytes() + afterInterrupt.readableBytes();
+
+ inputPlus = new AsyncStreamingInputPlus(channel);
+ Thread consumer = new Thread(() -> {
+ try
+ {
+ byte[] buffer = new byte[totalBytes];
+ Assert.assertEquals(totalBytes, inputPlus.read(buffer, 0, totalBytes));
+ }
+ catch (Throwable tr)
+ {
+ fail("Unexpected exception: " + tr);
+ }
+
+ try
+ {
+ inputPlus.readByte();
+ fail("Expected EOFException");
+ }
+ catch (ClosedChannelException ex)
+ {
+ // expected
+ }
+ catch (Throwable tr)
+ {
+ fail("Unexpected: " + tr);
+ }
+ });
+
+ try
+ {
+ consumer.start();
+ inputPlus.append(beforeInterrupt);
+ consumer.interrupt();
+ inputPlus.append(afterInterrupt);
+ inputPlus.requestClosure();
+ consumer.interrupt();
+ }
+ finally
+ {
+ consumer.join(TimeUnit.MINUTES.toMillis(1), 0);
+
+ // Check the input plus is closed by attempting to append to it
+ Assert.assertFalse(inputPlus.append(beforeInterrupt));
+ }
+ }
+
+ @Test
public void consumeUntil_SingleBuffer_Partial_HappyPath() throws IOException
{
consumeUntilTestCycle(1, 8, 0, 4);
@@ -258,25 +280,4 @@ public class AsyncStreamingInputPlusTest
isOpen = false;
}
}
-
- @Test
- public void rebufferTimeout() throws IOException
- {
- long timeoutMillis = 1000;
- inputPlus = new AsyncStreamingInputPlus(channel, timeoutMillis, TimeUnit.MILLISECONDS);
-
- long startNanos = nanoTime();
- try
- {
- inputPlus.readInt();
- Assert.fail("should not have been able to read from the queue");
- }
- catch (InputTimeoutException e)
- {
- // this is the success case, and is expected. any other exception is a failure.
- }
-
- long durationNanos = nanoTime() - startNanos;
- Assert.assertTrue(TimeUnit.MILLISECONDS.toNanos(timeoutMillis) <= durationNanos);
- }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org