2

Will encoder's encode method executing concurrently? I observe the encode method might be concurrent by different threads. The pipeline define as:

Channels.pipeline(
    idleHandler,
    new AmfDecoder<GameEvent>(GameEvent.class),
    new AmfEncoder<GameEvent>(),
    concurrencyHandler,
    new WebHandler());

Encoder:

public class AmfEncoder<T extends IAmfEvent> extends OneToOneEncoder{
private final SerializationContext serializationContext = new SerializationContext();
private final Amf3Output amfout = new Amf3Output(serializationContext);

@Override
protected Object encode(ChannelHandlerContext arg0, Channel arg1,
        Object arg2) throws Exception {
    T e = (T)arg2;
    ByteArrayOutputStream byteoutStreamSize = new ByteArrayOutputStream();
    amfout.setOutputStream(byteoutStreamSize);
    amfout.writeObject(e.getBody());
    // byteoutStreamSize has small probability become empty at here, in debug mode I can sure e.getBody() has data
    // I thought byteoutStreamSize might be empty by another thread call "amfout.flush()" or "amfout.reset()"
    amfout.flush();
    //...
    amfout.reset();
}

}

The calling of Channel.write is not only threads belong to netty's worker thread or threads in Exeutionhandler. There is a thread pool which created by my own will call Channel.write(). After I move 2 variables of amfout & serializationContext into encode() function to be local variable, the problem disappear.

Doc says ChannelPipeline is thread safe, I read netty 3.4.5 found "add", "remove"... operation is locked, but sendDownstream & sendUpstream has no lock. So if there are threads which not belong to worker thread pool or ExecutionHandler thread pool, and all of these threads call Channel.write(), concurrent problem will happen in decoder & encoder

2 Answers 2

5

The Channel pipeline is thread safe, but the problem here is that the event execution model is different for downstream events and upstream events:

  • Downstream Handlers are executed, using (multiple) user threads by default.

  • Downstream Handlers are not thread safe by default, since they can be executed by multiple user thread in any order (normally DownstreamEvents are lightweight and so their handlers do not maintain state in instance variables). Have a look at OneToOneEncoder implementations in the Netty code base. None of them maintains a state.

  • Upstream handlers are executed using a single thread by default or using multiple threads one by one (if execution handler is used).

  • Upstream Handlers are thread safe because of single threaded event execution (even though they can hold mutable state).

So, someone can mistakenly think that downstream handlers are as thread safe as upstream handlers.

As you said, the solution would be moving the instance variable to local scope if state is not required. Otherwise, make the downstream handling method thread safe.

0

I think you understand the concurrency correctly. You must either:

  1. Ensure your channel handlers are thread safe (no mutating instance variables)
  2. Use a ChannelPipelineFactory so that a new pipeline is created for each channel.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.