Package org.apache.cassandra.net

Class AbstractMessageHandler

  • All Implemented Interfaces:
    io.netty.channel.ChannelHandler, io.netty.channel.ChannelInboundHandler, FrameDecoder.FrameProcessor
    Direct Known Subclasses:
    CQLMessageHandler, InboundMessageHandler
    public abstract class AbstractMessageHandler
extends io.netty.channel.ChannelInboundHandlerAdapter
implements FrameDecoder.FrameProcessor
    Core logic for handling inbound message deserialization and execution (in tandem with FrameDecoder). Handles small and large messages, corruption, flow control, dispatch of message processing to a suitable consumer. # Interaction with FrameDecoder An AbstractMessageHandler implementation sits on top of a FrameDecoder in the Netty pipeline, and is tightly coupled with it. FrameDecoder decodes inbound frames and relies on a supplied FrameDecoder.FrameProcessor to act on them. AbstractMessageHandler provides two implementations of that interface: - process(Frame) is the default, primary processor, and is expected to be implemented by subclasses - AbstractMessageHandler.UpToOneMessageFrameProcessor, supplied to the decoder when the handler is reactivated after being put in waiting mode due to lack of acquirable reserve memory capacity permits Return value of FrameDecoder.FrameProcessor.process(Frame) determines whether the decoder should keep processing frames (if true is returned) or stop until explicitly reactivated (if false is). To reactivate the decoder (once notified of available resource permits), FrameDecoder.reactivate() is invoked. # Frames AbstractMessageHandler operates on frames of messages, and there are several kinds of them: 1. FrameDecoder.IntactFrame that are contained. As names suggest, these contain one or multiple fully contained messages believed to be uncorrupted. Guaranteed to not contain an part of an incomplete message. See processFrameOfContainedMessages(ShareableBytes, Limit, Limit). 2. FrameDecoder.IntactFrame that are NOT contained. These are uncorrupted parts of a large message split over multiple parts due to their size. Can represent first or subsequent frame of a large message. See processFirstFrameOfLargeMessage(IntactFrame, Limit, Limit) and processSubsequentFrameOfLargeMessage(Frame). 3. FrameDecoder.CorruptFrame with corrupt header. These are unrecoverable, and force a connection to be dropped. 4. FrameDecoder.CorruptFrame with a valid header, but corrupt payload. These can be either contained or uncontained. - contained frames with corrupt payload can be gracefully dropped without dropping the connection - uncontained frames with corrupt payload can be gracefully dropped unless they represent the first frame of a new large message, as in that case we don't know how many bytes to skip See processCorruptFrame(CorruptFrame). Fundamental frame invariants: 1. A contained frame can only have fully-encapsulated messages - 1 to n, that don't cross frame boundaries 2. An uncontained frame can hold a part of one message only. It can NOT, say, contain end of one large message and a beginning of another one. All the bytes in an uncontained frame always belong to a single message. # Small vs large messages A single handler is equipped to process both small and large messages, potentially interleaved, but the logic differs depending on size. Small messages are deserialized in place, and then handed off to an appropriate thread pool for processing. Large messages accumulate frames until completion of a message, then hand off the untouched frames to the correct thread pool for the verb to be deserialized there and immediately processed. See AbstractMessageHandler.LargeMessage and subclasses for concrete AbstractMessageHandler implementations for details of the large-message accumulating state-machine, and InboundMessageHandler.ProcessMessage and its inheritors for the differences in execution. # Flow control (backpressure) To prevent message producers from overwhelming and bringing nodes down with more inbound messages that can be processed in a timely manner, AbstractMessageHandler provides support for implementations to provide their own flow control policy. Before we attempt to process a message fully, we first infer its size from the stream. This inference is delegated to implementations as the encoding of the message size is protocol specific. Having assertained the size of the incoming message, we then attempt to acquire the corresponding number of memory permits. If we succeed, then we move on actually process the message. If we fail, the frame decoder deactivates until sufficient permits are released for the message to be processed and the handler is activated again. Permits are released back once the message has been fully processed - the definition of which is again delegated to the concrete implementations. Every connection has an exclusive number of permits allocated to it. In addition to it, there is a per-endpoint reserve capacity and a global reserve capacity ResourceLimits.Limit, shared between all connections from the same host and all connections, respectively. So long as long as the handler stays within its exclusive limit, it doesn't need to tap into reserve capacity. If tapping into reserve capacity is necessary, but the handler fails to acquire capacity from either endpoint of global reserve (and it needs to acquire from both), the handler and its frame decoder become inactive and register with a AbstractMessageHandler.WaitQueue of the appropriate type, depending on which of the reserves couldn't be tapped into. Once enough messages have finished processing and had their permits released back to the reserves, AbstractMessageHandler.WaitQueue will reactivate the sleeping handlers and they'll resume processing frames. The reason we 'split' reserve capacity into two limits - endpoing and global - is to guarantee liveness, and prevent single endpoint's connections from taking over the whole reserve, starving other connections. One permit per byte of serialized message gets acquired. When inflated on-heap, each message will occupy more than that, necessarily, but despite wide variance, it's a good enough proxy that correlates with on-heap footprint.

    • Method Detail

      • channelRead

        public void channelRead​(io.netty.channel.ChannelHandlerContext ctx,
                        java.lang.Object msg)
        Specified by:
        channelRead in interface io.netty.channel.ChannelInboundHandler
        Overrides:
        channelRead in class io.netty.channel.ChannelInboundHandlerAdapter

      • handlerAdded

        public void handlerAdded​(io.netty.channel.ChannelHandlerContext ctx)
        Specified by:
        handlerAdded in interface io.netty.channel.ChannelHandler
        Overrides:
        handlerAdded in class io.netty.channel.ChannelHandlerAdapter

      • process

        public boolean process​(FrameDecoder.Frame frame)
                throws java.io.IOException
        Description copied from interface: FrameDecoder.FrameProcessor
        Frame processor that the frames should be handed off to.
        Specified by:
        process in interface FrameDecoder.FrameProcessor
        Returns:
        true if more frames can be taken by the processor, false if the decoder should pause until it's explicitly resumed.
        Throws:
        java.io.IOException

      • processSubsequentFrameOfLargeMessage

        protected boolean processSubsequentFrameOfLargeMessage​(FrameDecoder.Frame frame)

      • fatalExceptionCaught

        protected abstract void fatalExceptionCaught​(java.lang.Throwable t)

      • acquireCapacity

        protected boolean acquireCapacity​(ResourceLimits.Limit endpointReserve,
                                  ResourceLimits.Limit globalReserve,
                                  int bytes,
                                  long currentTimeNanos,
                                  long expiresAtNanos)
        Try to acquire permits for the inbound message. In case of failure, register with the right wait queue to be reactivated once permit capacity is regained.

      • releaseCapacity

        public void releaseCapacity​(int bytes)

      • releaseProcessedCapacity

        protected void releaseProcessedCapacity​(int size,
                                        Message.Header header)
        Invoked to release capacity for a message that has been fully, successfully processed. Normally no different from invoking releaseCapacity(int), but is necessary for the verifier to be able to delay capacity release for backpressure testing.

      • channelInactive

        public void channelInactive​(io.netty.channel.ChannelHandlerContext ctx)
        Specified by:
        channelInactive in interface io.netty.channel.ChannelInboundHandler
        Overrides:
        channelInactive in class io.netty.channel.ChannelInboundHandlerAdapter

      • id

        protected abstract java.lang.String id()