ZlibEncoder fails when driven by multiple threads

[runrc]

The ZlibEncoder is a downstream handler and extends OneToOneEncoder which in turn implements ChannelDownstreamHandler. All Netty downstream handlers can be driven by multiple threads.

In OneToOneEncoder.handleDownstream() (called from ZlibEncoder.handleDownstream()),

    Object encodedMessage = encode(ctx, e.getChannel(), originalMessage);
    if (originalMessage == encodedMessage) {
        ctx.sendDownstream(evt);
    } else if (encodedMessage != null) {
        write(ctx, e.getFuture(), encodedMessage, e.getRemoteAddress());
    }

The OneToOneEncoder doesn't provide any protection between the encode(...) and subsequent channel write(...). This is fine for stateless encoding, but is a problem when using stateful encoders like Zip.

After encode(...) is called to compress a message, it is possible for another thread to also call encode(...) with another message, and the subsequent channel write()'s to get out of order. This causes errors when decoding the stream of bytes in the upstream handler in ZlibDecoder, since the stream is now corrupted.

The fix is to ensure the encode() and the subsequent channel write() is strictly performed in order, even if driven by multiple threads. This can be done by synchronizing the encode() and write() to ensure no other thread can jump in and interfere with the order of what is written to the channel. This functionality is essential for ZlibEncoder to work correctly, since the ZlibDecoder expects to be fed the same data as was encoded.

I wrote an alternative ZlibEncoder (below) works correctly in a multi-threaded environment. It ensures no thread can jump in between the encode() and the subsequent channel write()

public class MyZlibEncoder extends org.jboss.netty.handler.codec.compression.ZlibEncoder {
public MyZlibEncoder(ZlibWrapper wrapper) {
super(wrapper, 6);
}

@Override
public void handleDownstream(ChannelHandlerContext ctx, ChannelEvent evt) throws Exception {
    if (evt instanceof MessageEvent) {
        synchronized (this) {
            MessageEvent e = (MessageEvent)evt;
            Object encodedMessage = encode(ctx, e.getChannel(), e.getMessage());
            Channels.write(ctx, e.getFuture(), encodedMessage, e.getRemoteAddress());
        }
    } else if (evt instanceof ChannelStateEvent) {
        super.handleDownstream(ctx, evt);
    } else {
        ctx.sendDownstream(evt);
    }
}

}

[jaens]

Vaguely related discussion in #545 - proposed a similar fix, waiting for someone with godlike Netty knowledge to respond :)

[normanmaurer]

@jaens I also suggested this on #545 but just for the record.

So this would be the idea:

diff --git a/src/main/java/org/jboss/netty/handler/codec/oneone/OneToOneEncoder.java b/src/main/java/org/jboss/netty/handler/codec/oneone/OneToOneEncoder.java
index 29c9d38..9713b4b 100644
--- a/src/main/java/org/jboss/netty/handler/codec/oneone/OneToOneEncoder.java
+++ b/src/main/java/org/jboss/netty/handler/codec/oneone/OneToOneEncoder.java
@@ -50,17 +50,30 @@ public abstract class OneToOneEncoder implements ChannelDownstreamHandler {
     }

     public void handleDownstream(
-            ChannelHandlerContext ctx, ChannelEvent evt) throws Exception {
+            final ChannelHandlerContext ctx, final ChannelEvent evt) throws Exception {
         if (!(evt instanceof MessageEvent)) {
             ctx.sendDownstream(evt);
             return;
         }

-        MessageEvent e = (MessageEvent) evt;
-        Object originalMessage = e.getMessage();
+        final MessageEvent e = (MessageEvent) evt;
+        ctx.getPipeline().execute(new Runnable() {
+            
+            public void run() {
+                try {
+                    encodeAndWrite(ctx, e);
+                } catch (Throwable cause) {
+                    fireExceptionCaughtLater(e.getChannel(), cause);
+                }
+            }
+        });
+    }
+
+    private void encodeAndWrite(final ChannelHandlerContext ctx, final MessageEvent e) throws Exception {
+        final Object originalMessage = e.getMessage();
         Object encodedMessage = encode(ctx, e.getChannel(), originalMessage);
         if (originalMessage == encodedMessage) {
-            ctx.sendDownstream(evt);
+            ctx.sendDownstream(e);
         } else if (encodedMessage != null) {
             write(ctx, e.getFuture(), encodedMessage, e.getRemoteAddress());
         }

[normanmaurer]

@jaens if we really want to use synchronized then we should synchronize on the Channel and not one this as a OneToOneEncoder may be shared by different Channels.

[runrc]

Something to note is that not all encoders require strict encode and write ordering. If the encoder and corresponding decoder relies on stream state, like ZIP, then we need to enforce strict encode and write ordering. Some encoders do not rely on stream state, but instead are message based and thus do not require strict encode and write ordering.

Perhaps creating a new OneToOneStrictEncoder which enforces the encode / write order should be implemented. The default OneToOneEncoder would remain as is. The ZipEncoder would extend OneToOneStrictEncoder.

Rather than enforcing the order of encode and write using synchronization, utilising the pipeline I/O thread to perform the encode and write is a good solution.

[normanmaurer]

Good point... Let me take care of this later.

Sent from my iPhone. Excuse any typos....

Am 28.08.2012 um 11:35 schrieb Ravi Cheema notifications@github.com:

Something to note is that not all encoders require strict encode and write ordering. If the encoder and corresponding decoder relies on stream state, like ZIP, then we need to enforce strict encode and write ordering. Some encoders do not rely on stream state, but instead are message based and thus do not require strict encode and write ordering.

Perhaps creating a new OneToOneStrictEncoder which enforces the encode / write order should be implemented. The default OneToOneEncoder would remain as is. The ZipEncoder would extend OneToOneStrictEncoder.

Rather than enforcing the order of encode and write using synchronization, utilising the pipeline I/O thread to perform the encode and write is a good solution.

—
Reply to this email directly or view it on GitHub.

[jaens]

@normanmaurer I did not originally open this issue, but anyway, my two cents... :)

Executing the rest of the downstream pipeline (which can contain arbitrary handlers) in the I/O worker thread is a pretty hefty change in semantics, with its own set of issues:

1. it's asynchronous
   1.1) originally any exceptions thrown will be immediately returned to the caller of the write - this does not
   1.2) all downstream handlers were known to be called after write() returns - this does not
   1.3) writes are executed out of order with respect to any writes from any channel contexts downstream of this handler, even in single-threaded use (this is somewhat tricky, but for some use cases writes do start from different contexts in the pipeline)
2. starvation problems - since this will execute in the singular worker thread for the channel, doing CPU heavy things there (like compression and SSL) will starve any other reads and writes (one of the reasons for OrderedMemoryAwareThreadPoolExecutor)
3. deadlocks - same caveat as mentioned in the Netty manual, can not wait synchronously for eg. a write future in any downstream handler.

[normanmaurer]

@jaens and I think you brought about some very good concerns. Thanks for that! So let us just use synchronized(channel) then...

[normanmaurer]

@jaens just pushed the change.. let me know what you think about it..

[jaens]

@normanmaurer looks clear & should work I guess

The SPDY codec fix pull req (which works for us in production) is pretty much the same, except it only synchronizes when necessary (since not all frame types actually access the shared compressor)

While there are probably use cases for doing something more complicated, at least for re-usable components inside Netty itself I guess it's best to do "non-surprising" things ie. consistent with how Netty behaves in other places :)

[normanmaurer]

Yeah I will fix the SPDY stuff later .. One fix after the other ;)

Sent from my iPhone. Excuse any typos....

Am 28.08.2012 um 13:54 schrieb Jaen notifications@github.com:

@normanmaurer looks clear & should work I guess

The SPDY codec fix pull req (which works for us in production) is pretty much the same, except it only synchronizes when necessary (since not all frame types actually access the shared compressor)

While there are probably use cases for doing something more complicated, at least for re-usable components inside Netty itself I guess it's best to do "non-surprising" things ie. consistent with how Netty behaves in other places :)

—
Reply to this email directly or view it on GitHub.