natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Streaming sessions longer than 3 minutes fail with timeout 

patch by Jon Meredith; reviewed by Benedict Elliott Smith, Caleb Rackliffe for CASSANDRA-17510
This commit is contained in:
Jon Meredith 2022-04-01 12:58:02 -06:00
parent ffa1e9cf5a
commit fa7185ef02
5 changed files with 99 additions and 118 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -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)

3
src/java/org/apache/cassandra/io/sstable/format/big/BigTableZeroCopyWriter.java
View File

@ -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();
}

100
src/java/org/apache/cassandra/net/AsyncStreamingInputPlus.java
View File

@ -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
do
{
next = queue.poll(rebufferTimeoutNanos, TimeUnit.NANOSECONDS);
}
catch (InterruptedException ie)
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) // the indicator that the input is closed
{
// nop
}
if (null == next)
throw new InputTimeoutException();
if (next == Unpooled.EMPTY_BUFFER) // Unpooled.EMPTY_BUFFER is 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

2
src/java/org/apache/cassandra/streaming/async/NettyStreamingChannel.java
View File

@ -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;

111
test/unit/org/apache/cassandra/net/AsyncStreamingInputPlusTest.java
View File

@ -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()
{
@ -161,6 +129,60 @@ public class AsyncStreamingInputPlusTest
Assert.assertEquals(size, inputPlus.unsafeAvailable());
}
@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
{
@ -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);
}
}