diff --git a/CHANGES.txt b/CHANGES.txt index 87deaf9029..68de34fa14 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 c47c23bc01..e4a382d213 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 b964ddac26..5e936ace47 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 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. *

- * 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 diff --git a/src/java/org/apache/cassandra/streaming/async/NettyStreamingChannel.java b/src/java/org/apache/cassandra/streaming/async/NettyStreamingChannel.java index fb5d37a667..70b1c37b88 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 *

* 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 417a51c900..1d6dcf4acd 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() { @@ -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); - } }