Skip to content

v1.1.0 - Message batching and other things.

Choose a tag to compare

View all tags
@Atheer-Ganayem Atheer-Ganayem released this 05 Sep 10:51
v1.1.0
78097ef

SnapWS v1.1.0

Message Batching System

This release mainly introduces a comprehensive message batching system designed to significantly improve performance in high-throughput WebSocket applications. Message batching addresses the overhead associated with sending numerous small messages by combining multiple messages into single WebSocket frames.

Overview

In typical real-time applications, sending individual messages results in substantial protocol overhead. Each WebSocket message requires frame headers, system calls, and network packets. The new batching system aggregates messages over configurable time intervals, reducing this overhead and improving overall throughput.

Implementation Details

The batching system operates through a background flusher that periodically combines queued messages and sends them as single WebSocket frames. Messages are collected in per-connection batches and automatically flushed based on time intervals or size thresholds.

Core Components:

  • batchFlusher: Manages periodic flushing across all connections
  • messageBatch: Per-connection message accumulation
  • BatchStrategy: Configurable message encoding formats

Batching Strategies

JSON Array Strategy
Messages are encoded as JSON arrays and sent as binary WebSocket messages:

upgrader.EnableJSONBatching(ctx, 50*time.Millisecond)

Output format: [message1, message2, message3, ...]

Length-Prefixed Binary Strategy
Messages are prefixed with 4-byte big-endian length headers:

upgrader.EnablePrefixBatching(ctx, 50*time.Millisecond)

Output format: [uint32_length][message1][uint32_length][message2]...

Custom Strategy
Allows implementation of application-specific batching formats:

upgrader.EnableBatching(ctx, flushInterval, customSendFunc)

API Methods

When batching is enabled, you have to use batch-specific methods, regular methods will still send messages instantly even if batching is enabled.

Upgrader Methods

  • EnableJSONBatching(ctx, flushEvery) - Configure JSON array batching
  • EnablePrefixBatching(ctx, flushEvery) - Configure binary length-prefixed batching
  • EnableBatching(ctx, flushEvery, sendFunc) - Configure custom batching

Conn methods

  • Batch(data) - Adds raw bytes to the batch (doesnt validate and doesnt marshal)
  • BatchJSON(v) - Marshals to JSON then calls Batch(data []byte). This is just a convenience method that marshals and calls Batch() under the hood.

Broadcasting methods
These methods are available for rooms and managers.
These methods respect BroadcastBackpressure field in options, check: [Backpressure Handling] after this.

  • BatchBroadcast(ctx, data, exclude) (int, error) - Adds a message to all connections batch.
  • BatchBroadcastJSON(ctx, data, exclude) (int, error) - Marshals and adds a message to all connections batch.

Configuration

Default Limits:

  • Maximum bytes per batch: 1MB
  • Default flush interval: 50ms

Note: all these defaults can be changed by you.

Backpressure Handling
The system supports three strategies when connection queues become full:

  • BackpressureDrop (default): Discard messages for slow clients
  • BackpressureClose: Terminate slow connections
  • BackpressureWait (not recommended): Block until queue space is available

Performance Characteristics

Batching provides the most benefit for applications with:

  • High message frequency
  • Small to medium message sizes
  • Multiple concurrent connections
  • Real-time requirements with acceptable latency

Backward Compatibility

The batching system is entirely additive. Existing applications continue to function without modification. Batching is opt-in and can be enabled per-upgrader instance.

Technical Notes

  • All batching operations are thread-safe
  • Context cancellation is respected throughout the batching pipeline
  • Connections are automatically cleaned up when closed
  • Memory usage is bounded by the configured batch limits
  • Graceful shutdown ensures all pending batches are flushed

Broadcasting

Broadcasting changed from spawning workers and killing them after the broadcast finishes, to having a broadcastQueue buffered channel per connection, and a go routine that listens for this channel.
This change was due to a Reddit comment i read, and then spent the next 2 days benchmarking and decided to change the broadcasting approach to this.
This change eliminates the overhead of goroutine creation/destruction for each broadcast operation, resulting in better performance and more predictable resource usage.

API Changes

  • Removed broadcastWorkers from options
  • Added BroadcastChannelsSize (default 8) in options
  • Added BroadcastBackpressure (default BackpressureDrop). Broadcasting strategies already explained above, in the batching section.

Manager

  • Changed GetConn(key KeyType) (*ManagedConn[KeyType], bool) to: Get(key KeyType) *ManagedConn[KeyType]. This change was made to ensure consistency between room methods and manager methods.
  • Removed GetAllConnsWithExclude(exclude ...KeyType)
  • Replaced GetAllConns() with GetAllConns(exclude ...KeyType)
  • Changed GetAllConnsWithExcludeAsConn(exclude ...KeyType) To GetAllConnsAsConn(exclude ...KeyType)

Additional changes

  • In NextWriter(ctx, msgType) method, there was an if check !conn.writer.closed, this mean that if the writer was taken it will error. This is unnecessary and causes unwanted errors. So this if check is removed now. The user should ensure all writers are closed and provide a ctx that times out, because now if the writer is not closed and another goroutine is trying to get the next writer, it will hang forever.
Assets 2
Loading