Package org.apache.cassandra.net

Class AbstractMessageHandler

java.lang.Object
io.netty.channel.ChannelHandlerAdapter
io.netty.channel.ChannelInboundHandlerAdapter
org.apache.cassandra.net.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.

  • Field Details

    • decoder

      protected final FrameDecoder decoder

    • channel

      protected final io.netty.channel.Channel channel

    • largeThreshold

      protected final int largeThreshold

    • largeMessage

      protected AbstractMessageHandler.LargeMessage<?> largeMessage

    • queueCapacity

      protected final long queueCapacity

    • endpointReserveCapacity

      protected final ResourceLimits.Limit endpointReserveCapacity

    • endpointWaitQueue

      protected final AbstractMessageHandler.WaitQueue endpointWaitQueue

    • globalReserveCapacity

      protected final ResourceLimits.Limit globalReserveCapacity

    • globalWaitQueue

      protected final AbstractMessageHandler.WaitQueue globalWaitQueue

    • onClosed

      protected final AbstractMessageHandler.OnHandlerClosed onClosed

    • corruptFramesRecovered

      protected long corruptFramesRecovered

    • corruptFramesUnrecovered

      protected long corruptFramesUnrecovered

    • receivedCount

      protected long receivedCount

    • receivedBytes

      protected long receivedBytes

    • throttledCount

      protected long throttledCount

    • throttledNanos

      protected long throttledNanos

  • Constructor Details

  • Method Details

    • channelRead

      public void channelRead(io.netty.channel.ChannelHandlerContext ctx, 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 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:
      IOException

    • processOneContainedMessage

      protected abstract boolean processOneContainedMessage(ShareableBytes bytes, ResourceLimits.Limit endpointReserve, ResourceLimits.Limit globalReserve) throws IOException
      Throws:
      IOException

    • processFirstFrameOfLargeMessage

      protected abstract boolean processFirstFrameOfLargeMessage(FrameDecoder.IntactFrame frame, ResourceLimits.Limit endpointReserve, ResourceLimits.Limit globalReserve) throws IOException
      Throws:
      IOException

    • processSubsequentFrameOfLargeMessage

      protected boolean processSubsequentFrameOfLargeMessage(FrameDecoder.Frame frame)

    • processCorruptFrame

      protected abstract void processCorruptFrame(FrameDecoder.CorruptFrame frame) throws Crc.InvalidCrc
      Throws:
      Crc.InvalidCrc

    • fatalExceptionCaught

      protected abstract void fatalExceptionCaught(Throwable t)

    • processUpToOneMessage

      protected boolean processUpToOneMessage(ResourceLimits.Limit endpointReserve, ResourceLimits.Limit globalReserve) throws IOException
      Throws:
      IOException

    • 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.

    • acquireCapacity

      protected ResourceLimits.Outcome acquireCapacity(ResourceLimits.Limit endpointReserve, ResourceLimits.Limit globalReserve, int bytes)

    • 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 String id()