CASSANDRA-13630 support large internode messages with netty #253

jasobrown · 2018-08-21T20:46:38Z

No description provided.

dineshjoshi · 2018-09-06T18:35:15Z

src/java/org/apache/cassandra/net/async/MessageOutHandler.java

+            // if we allocated too much buffer for this message, we'll log here.
+            // if we allocated to little buffer space, we would have hit an exception when trying to write more bytes to it
+            if (out.isWritable())
+                errorLogger.error("{} reported message size {}, actual message size {}, msg {}",


How likely are we to cause log spam?

The errorLogger is an instance of NoSpamLogger, so we can adjust the log intervals. I'm not sure if making it an assert is better (as that would throw and Exception, and the connection would get killed). As it's mostly a developer-level logging thing, I could just remove it, as well. wdyt?

It's fine. We can leave it in.

dineshjoshi · 2018-09-06T18:36:31Z

src/java/org/apache/cassandra/net/async/RebufferingByteBufDataInputPlus.java

+    /**
+     * Default to a very large value.
+     */
+    private static final long DEFAULT_REBUFFER_BLOCK_IN_MILLIS = TimeUnit.DAYS.toMillis(2);


Its risky to default to a very large value. Why do we want to default to such a large value?

The corresponding pre-4.0 code just does blocking IO on a socket, with no timeouts. Thus it blocks forever. 2 days seems shorter than forever, but I'm game to change this to any time-bounded value.

I think it doesn't make sense to wait for more than 5 minutes to rebuffer. Given buffer sizes are on the order of a few megabytes at most, 5 minutes is an eternity. It's best to keep it short so failures are exposed sooner than later.

I've lowered it to 3 minutes.

dineshjoshi · 2018-09-06T18:40:05Z

src/java/org/apache/cassandra/net/async/RebufferingByteBufDataInputPlus.java

@@ -183,6 +195,11 @@ public int available() throws EOFException
        return availableBytes;
    }

+    public boolean isEmpty() throws EOFException
+    {
+        return available() == 0;


The method available() has a side effect of channelConfig.setAutoRead(true) when the availableBytes falls below the lowWatermark. Are you sure that is ok?

Nice find. I added that side effect to
StreamingInboundHandler to help deserialization. Clearly, I could move that into a separate method, and have StreamingInboundHandler invoke that after the available() call. wdyt?

So, looking at this again, I needed to figure out why available() does the autoRead check. For posterity, here's the reasoning behind it:

There a case where we enqueue a buffer, and disable autoRead. Then read all of the buffer exactly (and no more bytes are available to read in the instance). In an effort to not block indefinitely (so I could check if StreamingInboundHandler.close is true), i called available() (and did the autoRead check) instead of attempt to read a value from the inputPlus (which would call reBuffer() and block either indefinitely or throw an exception if I added a timeout). I didn't want the indefinite block nor an exception to be thrown, so I bolted the autoRead check onto available().

I think it still makes sense to move this into a separate method call.

dineshjoshi · 2018-09-06T18:42:00Z

src/java/org/apache/cassandra/streaming/async/NettyStreamingMessageSender.java

+        {
+            try
+            {
+                session.onError(t).get(5, TimeUnit.MINUTES);


Why do we have a 5 minute timeout? We should pull this out as a constant.

dineshjoshi · 2018-09-06T18:53:50Z

src/java/org/apache/cassandra/net/MessageIn.java

+        }
+
+        @SuppressWarnings("resource")
+        public void process(ByteBuf in) throws IOException


This method is a bit complicated to understand. Could you refactor it? For exampleRedisDecoder provides a nice example of this - https://github.com/netty/netty/blob/4.1/codec-redis/src/main/java/io/netty/handler/codec/redis/RedisDecoder.java

done. I also pulled the main loop logic into the base class.

…ing on the queue in RebufferingByteBufDataInputPlus; fix MIHTest

…; added tests

…o the base class, and moved logic for each case statement into submethods rather than directly in-line with the loop

dineshjoshi · 2018-09-07T21:09:56Z

src/java/org/apache/cassandra/net/MessageIn.java

+            {
+                String key = DataInputStream.readUTF(in);
+                ParameterType parameterType = ParameterType.byName.get(key);
+                int valueLength = in.readInt();


Unused variable.

dineshjoshi · 2018-09-07T21:10:26Z

src/java/org/apache/cassandra/net/MessageIn.java

+            {
+                String key = DataInputStream.readUTF(inputTracker);
+                ParameterType parameterType = ParameterType.byName.get(key);
+                long valueLength = VIntCoding.readUnsignedVInt(inputTracker);


Unused variable.

dineshjoshi · 2018-09-07T22:08:17Z

src/java/org/apache/cassandra/net/async/ByteBufDataOutputStreamPlus.java

@@ -196,10 +220,12 @@ protected void doFlush(int count) throws IOException
            int byteCount = buffer.position();
            currentBuf.writerIndex(byteCount);

-            if (!Uninterruptibles.tryAcquireUninterruptibly(channelRateLimiter, byteCount, 2, TimeUnit.MINUTES))
+            if (!Uninterruptibles.tryAcquireUninterruptibly(channelRateLimiter, byteCount, rateLimiterBlockTime, rateLimiterBlockTimeUnit))


This is going up from 2 minutes to potentially 5 minutes. I think this should be ok but just making sure.

Good catch. ftr, the only place where this went from 2 up to 5 minutes was in NettyStreamingMessageSender.FIleSendTask where I defaulted to 5 minutes. I've switched that call site to default to 2 minutes.

dineshjoshi · 2018-09-07T22:29:44Z

test/unit/org/apache/cassandra/net/MessageInProcessorPre40Test.java

+
+    private void buildParamBufPre40(int valueLength) throws IOException
+    {
+        buf = Unpooled.buffer(1024, 1024); // 1k should be enough for everybody!


dineshjoshi · 2018-09-07T22:40:23Z

src/java/org/apache/cassandra/net/async/MessageInHandler.java

     */
-    @SuppressWarnings("resource")
-    public void handleDecode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception
+    class NonblockingBufferHandler implements BufferHandler


This can be made into a static inner class.

dineshjoshi · 2018-09-07T22:56:28Z

src/java/org/apache/cassandra/net/MessageOut.java

+    {
+        for (int ii = 0; ii < parameters.size(); ii += PARAMETER_TUPLE_SIZE)
+        {
+            if (((ParameterType)parameters.get(ii + PARAMETER_TUPLE_TYPE_OFFSET)).equals(type))


Don't need the typecast to ParameterType

dineshjoshi · 2018-09-07T22:59:22Z

src/java/org/apache/cassandra/net/MessageOut.java

+    {
+        try
+        {
+            UUID sessionId =  (UUID)getParameter(ParameterType.TRACE_SESSION);


Nit: Spacing

dineshjoshi · 2018-09-07T23:04:11Z

src/java/org/apache/cassandra/net/MessageOut.java

+        try
+        {
+            UUID sessionId =  (UUID)getParameter(ParameterType.TRACE_SESSION);
+            if (sessionId != null)


I assume sessionId != null means that tracing is enabled? Otherwise we should explicitly check whether tracing is enabled.

Yes. Further, this method is basically taken verbatim from 3.11's OutboundTcpConnection.writeConnected() method.

dineshjoshi · 2018-09-10T01:37:13Z

test/unit/org/apache/cassandra/net/async/ChannelWriterTest.java

+        if (type == ChannelWriterType.COALESCING)
+            coalescingStrategy = CoalescingStrategies.newCoalescingStrategy(CoalescingStrategies.Strategy.FIXED.name(), COALESCE_WINDOW_MS, null, "test");
+        else
+            coalescingStrategy = Optional.empty();


This is beyond the scope of this PR, however, we should refactor OutboundConnectionParams to not store an Optional.

Wait for CASSANDRA-14503 :)

smiklosovic · 2022-03-16T11:53:47Z

@jasobrown can be this PR closed?

dineshjoshi requested changes Sep 7, 2018

View reviewed changes

jasobrown added 8 commits September 7, 2018 12:48

Support sanely sending large messages with internode messaging via netty

49864e1

clean up error handling logic in MIH; allow custom timeouts for block…

9367c6e

…ing on the queue in RebufferingByteBufDataInputPlus; fix MIHTest

add RebufferingByteBufDataInputPlus timeout test

08f3607

fix ChannelWriterTest to wait for asynchronous blokcing implemenations

d8b5183

point to jasobrown dtest repo

38383a8

trivial lambda change

02f561a

moved autoRead check out of available method and into it's own method…

2578f0f

…; added tests

refactored MessageInProcessor.process to move the main loop logic int…

d52e012

…o the base class, and moved logic for each case statement into submethods rather than directly in-line with the loop

jasobrown force-pushed the 13630 branch from 8296bf9 to d52e012 Compare September 7, 2018 21:36

jasobrown added 2 commits September 7, 2018 15:41

changed a few timeout values

158b374

fix eclipse-warning

ef7a48e

dineshjoshi requested changes Sep 7, 2018

View reviewed changes

address Dinesh review comments

0252a8c

jasobrown changed the title ~~13630~~ CASSANDRA-13630 Sep 9, 2018

dineshjoshi reviewed Sep 10, 2018

View reviewed changes

belliottsmith force-pushed the trunk branch from 822ea57 to 16ef9ac Compare February 8, 2019 17:46

iamaleksey force-pushed the trunk branch from 532c033 to 0d48503 Compare June 12, 2019 15:27

belliottsmith force-pushed the trunk branch from d0be441 to 7dc5b70 Compare August 20, 2019 15:11

smiklosovic changed the title ~~CASSANDRA-13630~~ CASSANDRA-13630 support large internode messages with netty Mar 16, 2022

belliottsmith force-pushed the trunk branch 2 times, most recently from df3eb40 to 54e39a9 Compare July 23, 2025 11:19

CASSANDRA-13630 support large internode messages with netty #253

Are you sure you want to change the base?

CASSANDRA-13630 support large internode messages with netty #253

Uh oh!

Conversation

jasobrown commented Aug 21, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

dineshjoshi Sep 7, 2018 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

smiklosovic commented Mar 16, 2022

Uh oh!

Uh oh!

dineshjoshi Sep 7, 2018 •

edited

Loading