Classic Transport: Add Parallel Serialization/Deserialization via Netty Pipeline Executor

Inizia leggendo PekkoProtocolTransport.scala, Endpoint.scala, TcpSupport.scala e PekkoPduProtobufCodec per tracciare i percorsi attuali in ingresso e in uscita. Esamina reference.conf per l’impostazione proposta codec-threads e determina come si inseriscono nella modifica della pipeline entrambi i livelli Protobuf, la gestione dello stato del protocollo e il contratto di AssociationHandle.write. Il lavoro è completato quando l’executor è configurabile, l’ordinamento è preservato e il comportamento esistente rimane disponibile quando la funzionalità è disabilitata.

Classic Transport: Add Parallel Serialization/Deserialization via Netty Pipeline Executor

Summary

Classic Transport serializes and deserializes messages synchronously on Actor dispatcher threads. Each association has one EndpointWriter (outbound) and one EndpointReader (inbound), all sharing a fork-join dispatcher pool (default: up to 16 threads, pekko.remote.default-remote-dispatcher). Within a single association, serialization/deserialization is strictly serial — messages are processed one at a time.

Artery solves this with configurable parallel lanes (inbound-lanes / outbound-lanes) using Akka Streams, but Classic Transport has no equivalent. This issue proposes adding parallel serialization/deserialization to Classic Transport by leveraging Netty's EventExecutorGroup for codec handlers in the pipeline — a minimal-invasive approach that avoids restructuring the Actor-based message pipeline.

Why invest in deprecated Classic Transport?

Classic Transport is deprecated (@deprecated("Classic remoting is deprecated, use Artery", "Akka 2.6.0")). However, major projects like Apache Flink still rely on it (see FLINK-33505), and migration to Artery is a large undertaking. This optimization provides incremental performance improvements for users who cannot switch to Artery immediately, and introduces concepts (codec executor, per-recipient partitioning) that ease the eventual migration.

Motivation

Current Architecture (Classic Transport)

Classic Transport is built on Netty 4 (io.netty.*) with a per-association Actor model.

Wire Format: Two-Level Protobuf Encoding

The wire format has two nested Protobuf layers:

1. Outer envelope (PekkoProtocolMessage): Wraps payload, heartbeat, associate, disassociate. Handled by decodePdu() / constructPayload() in PekkoPduProtobufCodec.
2. Inner message (AckAndEnvelopeContainer): Wraps the serialized message, recipient ActorRef, sender ActorRef, and ack. Handled by decodeMessage() / constructMessage() in PekkoPduProtobufCodec.

Both levels are parsed/encoded on Actor dispatcher threads.

Inbound Path (Deserialization)

Netty IO EventLoop thread
  → FrameDecoder (LengthFieldBasedFrameDecoder)              ← byte-level framing
  → TcpServerHandler.onMessage() [TcpSupport.scala:58]       ← copies bytes, notifies listener
    → listener.notify(InboundPayload(ByteString))
      │
      ├─ ProtocolStateActor (always in path, PekkoProtocolTransport.scala:559-584)
      │    → decodePdu(p)                                    ← outer Protobuf: PekkoProtocolMessage
      │    → match: Payload(payload) →
      │      → listener.notify(InboundPayload(payload))       ← forward inner payload
      │        │
      │        └─ EndpointReader.receive() [Endpoint.scala:1163]   ← Actor dispatcher thread
      │             → tryDecodeMessageAndAck() [Endpoint.scala:1276]
      │               → codec.decodeMessage()                ← inner Protobuf: AckAndEnvelopeContainer
      │                 → resolveActorRefWithLocalAddress()   ← ActorRef resolution
      │             → msgDispatch.dispatch() [Endpoint.scala:1178]
      │               → MessageSerializer.deserialize() [Endpoint.scala:83]  ← heaviest operation
      │               → targetActor ! payload
      │
      └─ (during handshake: ProtocolStateActor handles Associate/Disassociate directly)

Key insight: ProtocolStateActor is always in the message path, even post-handshake (in the Open state). It decodes the outer PekkoProtocolMessage envelope on its dispatcher thread, then forwards the inner Payload to EndpointReader, which decodes the inner AckAndEnvelopeContainer and deserializes the actual message — also on a dispatcher thread.

Outbound Path (Serialization)

EndpointWriter Actor thread [Endpoint.scala:651]              ← Actor dispatcher thread
  → serializeMessage() [Endpoint.scala:1065]
    → MessageSerializer.serialize()                          ← heaviest operation
  → codec.constructMessage() [Endpoint.scala:922]            ← inner Protobuf: AckAndEnvelopeContainer
  → handle.write(pdu)
    → PekkoProtocolHandle.write() [PekkoProtocolTransport.scala:240]
      → codec.constructPayload()                             ← outer Protobuf: PekkoProtocolMessage
      → wrappedHandle.write()
        → TcpAssociationHandle.write() [TcpSupport.scala:120]
          → channel.writeAndFlush(Unpooled.wrappedBuffer(...))
            → Netty IO EventLoop thread
              → FrameEncoder (LengthFieldPrepender)          ← add frame header
              → write to network

The Bottleneck

With N remote connections, Classic Transport has N EndpointWriters and N EndpointReaders, all sharing a fork-join dispatcher pool (default: parallelism-factor = 0.5, parallelism-max = 16, per reference.conf:280-289). The bottleneck manifests as:

1. Within a single association, serialization/deserialization is strictly serial — messages are processed one at a time per Actor.
2. CPU-intensive serialization blocks the Actor thread — large Protobuf messages or Java serialization stall the message queue, increasing latency for all messages on that association.
3. Dispatcher pool contention — under high load, the shared dispatcher pool becomes saturated, causing backpressure across all associations.

Artery Comparison

Artery solves this with a fundamentally different architecture based on Akka Streams:

Dimension	Classic Transport	Artery
Inbound deserialization	Single EndpointReader Actor per association	FixedSizePartitionHub partitions messages across N inbound lanes (default 4), each with its own Deserializer GraphStage running in parallel
Outbound serialization	Single EndpointWriter Actor per association	N outbound lanes (default 1) per association, each with its own Encoder GraphStage, merged via MergeHub
Message ordering	Naturally ordered (single thread)	Preserved via consistent hashing on recipient.path.uid
Serialization API	Serializer.toBinary() -> Array[Byte]	ByteBufferSerializer.toBinary(ByteBuffer) (zero-copy)
Memory management	Per-message byte array allocation	EnvelopeBufferPool with buffer recycling
Compression	None	ActorRef and manifest compression tables
Parallelism granularity	Per-association (one thread)	Per-message (partitioned by recipient)
Deprecation status	Deprecated since Akka 2.6.0	Recommended transport
Configuration	No parallelism config	inbound-lanes = 4, outbound-lanes = 1 (reference.conf:926, 935)

Key Artery code references:

• Inbound lane partitioning: ArteryTransport.scala:471-484
• Lane setup (TCP): ArteryTcpTransport.scala:405-467
• Deserializer per lane: Codecs.scala:645-717
• Outbound lane selection: Association.scala:476-505

Proposed Solution: Netty Pipeline Executor for Codec Handlers

Approach

Netty 4 supports assigning an independent EventExecutorGroup to specific pipeline handlers via pipeline.addLast(EventExecutorGroup, name, handler). Handlers bound to a custom executor run on that executor's thread pool instead of the IO EventLoop, enabling:

1. Cross-connection parallelism: Multiple connections' encode/decode operations run concurrently on the executor pool. With N connections and K codec threads, up to K connections can encode/decode concurrently.
2. Per-connection ordering preserved: Netty guarantees events for the same channel are delivered in order to the assigned executor thread.

Important limitation: Netty binds each channel to a specific thread in the EventExecutorGroup for its lifetime. This means:

• Parallelism is across connections, not within a single connection.
• A "hot" connection always uses the same executor thread, potentially creating load imbalance.
• This is fundamentally different from Artery's lane-based approach, which partitions by recipient and provides within-connection parallelism.

Proposed Pipeline Change

Before:
  SslHandler -> FlushConsolidationHandler -> FrameDecoder -> FrameEncoder -> ServerHandler

After:
  SslHandler -> FlushConsolidationHandler -> FrameDecoder
    -> [codecExecutor] PekkoMessageDecoder
    -> [codecExecutor] PekkoMessageEncoder
    -> FrameEncoder -> ServerHandler

• PekkoMessageDecoder: On the codec executor thread, decodes the outer PekkoProtocolMessage envelope (what decodePdu() does today in ProtocolStateActor) and, for Payload messages, also decodes the inner AckAndEnvelopeContainer (what decodeMessage() does today in EndpointReader). Outputs a structured object that downstream handlers can consume directly.
• PekkoMessageEncoder: On the codec executor thread, encodes the outer PekkoProtocolMessage envelope (what constructPayload() does today in PekkoProtocolHandle) before frame encoding.

Configuration

pekko.remote.classic.netty.tcp {
  # Number of threads for codec (serialization/deserialization) operations.
  # 0 = disabled (codec runs on Actor dispatcher threads, current behavior).
  # Recommended: min(CPU cores / 2, number of remote connections).
  codec-threads = 0
}

Note: The config path is netty.tcp to match the existing NettyTransportSettings hierarchy (reference.conf:513).

Key Design Considerations

1. ProtocolStateActor is always in the message path: Even post-handshake (in the Open state), ProtocolStateActor receives every InboundPayload, calls decodePdu() to parse the outer PekkoProtocolMessage, and forwards Payload to EndpointReader (PekkoProtocolTransport.scala:559-584). The proposed PekkoMessageDecoder must handle both layers of Protobuf encoding, and downstream consumers (ProtocolStateActor and EndpointReader) must be adapted to accept pre-decoded objects.

2. Two-level Protobuf decoding: The decoder must handle:

   • Outer: PekkoProtocolMessage (payload, heartbeat, associate, disassociate)
   • Inner: AckAndEnvelopeContainer (message, recipient, sender, ack) — only for Payload types
     Control messages (heartbeat, associate, disassociate) should pass through with minimal overhead.

3. Write path restructuring: EndpointWriter.writeSend() currently calls serializeMessage() + codec.constructMessage() + handle.write() synchronously. PekkoProtocolHandle.write() calls codec.constructPayload(). To offload the outer envelope encoding to the Netty executor, the AssociationHandle.write() contract may need to change — this has binary compatibility implications (MiMa checks).

4. True message serialization (MessageSerializer.serialize/deserialize): This is the heaviest operation but is tightly coupled to the Actor system's Serialization extension and Serialization.currentTransportInformation thread-local context. Moving it to the Netty pipeline requires propagating the serialization context, which is more invasive.

5. Netty per-channel thread affinity: Each channel is bound to one thread in the EventExecutorGroup for its lifetime. Load balancing depends on the number of channels, not message volume. Users with few connections but high message volume per connection will see minimal benefit from this approach alone.

6. Interaction with use-dispatcher-for-io: The existing use-dispatcher-for-io config (reference.conf:588) allows using a custom dispatcher for Netty IO. The codec-threads config should be independent — it creates a separate executor pool for codec operations, regardless of the IO dispatcher setting.

7. SSL interaction: If SSL is enabled, SslHandler is added first in the pipeline. The PekkoMessageDecoder receives decrypted bytes, which is correct. The EventExecutorGroup assignment does not interfere with SSL handshake processing (SSL handshake is handled by SslHandler on the EventLoop).

8. Backpressure: If the codec executor queue grows (e.g., burst of large messages), consider ChannelOption.WRITE_BUFFER_WATER_MARK or pausing reads when the executor queue is full.

Implementation Scope

Phase 1 (This Issue): Protobuf Codec Offloading

Offload the two-level Protobuf encoding/decoding (PekkoProtocolMessage + AckAndEnvelopeContainer) to the Netty codec executor. This removes Protobuf parsing from Actor dispatcher threads and enables cross-connection parallelism.

• Add PekkoMessageDecoder / PekkoMessageEncoder handlers to NettyTransport.newPipeline() with a configurable EventExecutorGroup
• Adapt ProtocolStateActor to accept pre-decoded PekkoPdu objects instead of raw ByteString
• Adapt EndpointReader to accept pre-decoded Message objects instead of raw ByteString
• Add configuration: pekko.remote.classic.netty.tcp.codec-threads
• Add metrics: codec time per message (via RemoteMetrics)
• Evaluate binary compatibility impact (MiMa)

Phase 2 (Future): Message Serialization Offloading

Offload MessageSerializer.serialize/deserialize to a dedicated thread pool. This is the heaviest operation and provides the largest throughput gain.

• Propagate Serialization.currentTransportInformation to the codec executor
• Ensure message ordering per recipient (similar to Artery's inboundLanePartitioner — hash on recipient.path.uid)
• Adapt DefaultMessageDispatcher to accept pre-deserialized messages from the pipeline

Expected Benefits

1. Cross-connection parallelism: With N connections and K codec threads, up to K connections can encode/decode concurrently.
2. Actor dispatcher relief: Codec work moves off the shared default-remote-dispatcher pool, reducing contention for Actor message processing.
3. Incremental migration path: Users who cannot switch to Artery immediately get a performance improvement.
4. Foundation for Phase 2: The codec executor infrastructure enables the later addition of MessageSerializer offloading.

Limitations:

• Single-connection throughput is not improved (Netty per-channel thread affinity).
• For users with few connections but high per-connection volume, Phase 2 (message serialization offloading with recipient-based partitioning) is needed.

Related Issues

• FLINK-33505: Switch away from Netty 3-based Pekko Classic Remoting — Flink still uses Classic Transport and would benefit from this optimization.
• FLINK-10718: Use IO executor in RpcService for message serialization — Flink's RPC layer serialization bottleneck, complementary to this transport-level optimization.
• FLINK-34105: Akka timeout in TPC-DS benchmarks — Real-world evidence that serialization overhead impacts cluster stability at scale (10TB TPC-DS).

References

• Classic Transport inbound path: Endpoint.scala:1163 (EndpointReader.receive), Endpoint.scala:1276 (tryDecodeMessageAndAck), Endpoint.scala:1178 (msgDispatch.dispatch), Endpoint.scala:83 (lazy val payload = MessageSerializer.deserialize)
• Classic Transport outbound path: Endpoint.scala:651 (EndpointWriter), Endpoint.scala:1065 (serializeMessage), Endpoint.scala:922 (codec.constructMessage)
• ProtocolStateActor (always in path): PekkoProtocolTransport.scala:559-584 (Open state, decodePdu on every message)
• PekkoProtocolHandle write: PekkoProtocolTransport.scala:240 (codec.constructPayload)
• Classic Transport codec (two-level Protobuf): PekkoPduCodec.scala:83-126 (decodePdu/encodePdu/decodeMessage/constructMessage)
• Netty pipeline setup: NettyTransport.scala:390-406 (newPipeline)
• Netty pipeline initializer: NettyTransport.scala:440-453 (where codec executor would be added)
• Dispatcher pool config: reference.conf:280-289 (pekko.remote.default-remote-dispatcher)
• Artery inbound lane partitioning: ArteryTransport.scala:471-484
• Artery lane setup (TCP): ArteryTcpTransport.scala:405-467
• Artery Deserializer per lane: Codecs.scala:645-717
• Artery outbound lane selection: Association.scala:476-505
• Artery lane config: reference.conf:926 (inbound-lanes = 4), reference.conf:935 (outbound-lanes = 1)