crossbario / crossbario/autobahn-python
Write coalescing
Open
Nobody has claimed this yet.
feature
wamp
websocket
- Dominant language
- Python
- Stars
- 2.5k
- Forks
- 768
- PR merge metrics
- No merged PRs in 30d
Description
Write coalescing reduces the number of context switches (between user/kernel land) by batching up outgoing WAMP messages (actually, the serialized raw messages) before flattening and sending via kernel out on the network only periodically (eg every 25ms). This can bring big performance gains for chatty, low payload scenarios.
The following outdated code shows a prototype of write coalescing for RawSocket transports. I am archiving this here when we come back to actually implement the feature.
from __future__ import absolute_import
import binascii
import six
from twisted.python import log
from twisted.internet.protocol import Factory
from twisted.protocols.basic import Int32StringReceiver
from twisted.internet.error import ConnectionDone
from twisted.internet.task import LoopingCall
from autobahn.twisted.util import peer2str
from autobahn.wamp.exception import ProtocolError, SerializationError, TransportLost
__all__ = (
'WampRawSocketServerProtocol',
'WampRawSocketClientProtocol',
'WampRawSocketServerFactory',
'WampRawSocketClientFactory'
)
class WampRawSocketProtocol(Int32StringReceiver):
"""
Base class for Twisted-based WAMP-over-RawSocket protocols.
"""
def connectionMade(self):
if self.factory.debug:
log.msg("WampRawSocketProtocol: connection made")
# transport write-coalescing
#
if self.factory._coalesce:
_write = self.transport.write
self._buffer = []
def transport_write(data):
self._buffer.append(data)
self.transport.write = transport_write
def buffer_write():
if self._buffer:
_write(''.join(self._buffer))
self._buffer = []
self._buffer_writer = LoopingCall(buffer_write)
self._buffer_writer.start(float(self.factory._coalesce) / 1000000.)
if True or self.factory.debug:
log.msg("WampRawSocketProtocol: transport write-coalescing enabled ({0} ms period)".format(self.factory._coalesce))
else:
self._buffer_writer = None
if True or self.factory.debug:
log.msg("WampRawSocketProtocol: using default transport write-through")
# the peer we are connected to
#
try:
peer = self.transport.getPeer()
except AttributeError:
# ProcessProtocols lack getPeer()
self.peer = "?"
else:
self.peer = peer2str(peer)
# this will hold an ApplicationSession object
# once the RawSocket opening handshake has been
# completed
#
self._session = None
# Will hold the negotiated serializer once the opening handshake is complete
#
self._serializer = None
# Will be set to True once the opening handshake is complete
#
self._handshake_complete = False
# Buffer for opening handshake received bytes.
#
self._handshake_bytes = b''
# Client requested maximum length of serialized messages.
#
self._max_len_send = None
def _on_handshake_complete(self):
try:
self._session = self.factory._factory()
self._session.onOpen(self)
except Exception as e:
# Exceptions raised in onOpen are fatal ..
if self.factory.debug:
log.msg("WampRawSocketProtocol: ApplicationSession constructor / onOpen raised ({0})".format(e))
self.abort()
else:
if self.factory.debug:
log.msg("ApplicationSession started.")
def connectionLost(self, reason):
if self.factory.debug:
log.msg("WampRawSocketProtocol: connection lost: reason = '{0}'".format(reason))
try:
wasClean = isinstance(reason.value, ConnectionDone)
self._session.onClose(wasClean)
except Exception as e:
# silently ignore exceptions raised here ..
if self.factory.debug:
log.msg("WampRawSocketProtocol: ApplicationSession.onClose raised ({0})".format(e))
self._session = None
if self._buffer_writer:
self._buffer_writer.stop()
self._buffer_writer = None
def stringReceived(self, payload):
if self.factory.debug:
log.msg("WampRawSocketProtocol: RX octets: {0}".format(binascii.hexlify(payload)))
try:
for msg in self._serializer.unserialize(payload):
if self.factory.debug:
log.msg("WampRawSocketProtocol: RX WAMP message: {0}".format(msg))
self._session.onMessage(msg)
except ProtocolError as e:
log.msg(str(e))
if self.factory.debug:
log.msg("WampRawSocketProtocol: WAMP Protocol Error ({0}) - aborting connection".format(e))
self.abort()
except Exception as e:
if self.factory.debug:
log.msg("WampRawSocketProtocol: WAMP Internal Error ({0}) - aborting connection".format(e))
self.abort()
def send(self, msg):
"""
Implements :func:`autobahn.wamp.interfaces.ITransport.send`
"""
if self.isOpen():
if self.factory.debug:
log.msg("WampRawSocketProtocol: TX WAMP message: {0}".format(msg))
try:
payload, _ = self._serializer.serialize(msg)
except Exception as e:
# all exceptions raised from above should be serialization errors ..
raise SerializationError("WampRawSocketProtocol: unable to serialize WAMP application payload ({0})".format(e))
else:
self.sendString(payload)
if self.factory.debug:
log.msg("WampRawSocketProtocol: TX octets: {0}".format(binascii.hexlify(payload)))
else:
raise TransportLost()
def isOpen(self):
"""
Implements :func:`autobahn.wamp.interfaces.ITransport.isOpen`
"""
return self._session is not None
def close(self):
"""
Implements :func:`autobahn.wamp.interfaces.ITransport.close`
"""
if self.isOpen():
self.transport.loseConnection()
else:
raise TransportLost()
def abort(self):
"""
Implements :func:`autobahn.wamp.interfaces.ITransport.abort`
"""
if self.isOpen():
if hasattr(self.transport, 'abortConnection'):
# ProcessProtocol lacks abortConnection()
self.transport.abortConnection()
else:
self.transport.loseConnection()
else:
raise TransportLost()
class WampRawSocketServerProtocol(WampRawSocketProtocol):
"""
Base class for Twisted-based WAMP-over-RawSocket server protocols.
"""
def dataReceived(self, data):
if self._handshake_complete:
WampRawSocketProtocol.dataReceived(self, data)
else:
remaining = 4 - len(self._handshake_bytes)
self._handshake_bytes += data[:remaining]
if len(self._handshake_bytes) == 4:
if self.factory.debug:
log.msg("WampRawSocketProtocol: opening handshake received - {0}".format(binascii.b2a_hex(self._handshake_bytes)))
if ord(self._handshake_bytes[0]) != 0x7f:
if self.factory.debug:
log.msg("WampRawSocketProtocol: invalid magic byte (octet 1) in opening handshake: was 0x{0}, but expected 0x7f".format(binascii.b2a_hex(self._handshake_bytes[0])))
self.abort()
# peer requests us to send messages of maximum length 2**max_len_exp
#
self._max_len_send = 2 ** (9 + (ord(self._handshake_bytes[1]) >> 4))
if self.factory.debug:
log.msg("WampRawSocketProtocol: client requests us to send out most {} bytes per message".format(self._max_len_send))
# client wants to speak this serialization format
#
ser_id = ord(self._handshake_bytes[1]) & 0x0F
if ser_id in self.factory._serializers:
self._serializer = self.factory._serializers[ser_id]
if self.factory.debug:
log.msg("WampRawSocketProtocol: client wants to use serializer {}".format(ser_id))
else:
if self.factory.debug:
log.msg("WampRawSocketProtocol: opening handshake - no suitable serializer found (client requested {0}, and we have {1})".format(ser_id, self.factory._serializers.keys()))
self.abort()
# we request the peer to send message of maximum length 2**reply_max_len_exp
#
reply_max_len_exp = 24
# send out handshake reply
#
reply_octet2 = chr(((reply_max_len_exp - 9) << 4) | self._serializer.RAWSOCKET_SERIALIZER_ID)
self.transport.write(b'\x7F') # magic byte
self.transport.write(reply_octet2) # max length / serializer
self.transport.write(b'\x00\x00') # reserved octets
self._handshake_complete = True
self._on_handshake_complete()
if self.factory.debug:
log.msg("WampRawSocketProtocol: opening handshake completed", self._serializer)
# consume any remaining data received already ..
#
data = data[remaining:]
if data:
self.dataReceived(data)
class WampRawSocketClientProtocol(WampRawSocketProtocol):
"""
Base class for Twisted-based WAMP-over-RawSocket client protocols.
"""
def connectionMade(self):
WampRawSocketProtocol.connectionMade(self)
self._serializer = self.factory._serializer
# we request the peer to send message of maximum length 2**reply_max_len_exp
#
request_max_len_exp = 24
# send out handshake reply
#
request_octet2 = chr(((request_max_len_exp - 9) << 4) | self._serializer.RAWSOCKET_SERIALIZER_ID)
self.transport.write(b'\x7F') # magic byte
self.transport.write(request_octet2) # max length / serializer
self.transport.write(b'\x00\x00') # reserved octets
def dataReceived(self, data):
if self._handshake_complete:
WampRawSocketProtocol.dataReceived(self, data)
else:
remaining = 4 - len(self._handshake_bytes)
self._handshake_bytes += data[:remaining]
if len(self._handshake_bytes) == 4:
if self.factory.debug:
log.msg("WampRawSocketProtocol: opening handshake received - {0}".format(binascii.b2a_hex(self._handshake_bytes)))
if ord(self._handshake_bytes[0]) != 0x7f:
if self.factory.debug:
log.msg("WampRawSocketProtocol: invalid magic byte (octet 1) in opening handshake: was 0x{0}, but expected 0x7f".format(binascii.b2a_hex(self._handshake_bytes[0])))
self.abort()
# peer requests us to send messages of maximum length 2**max_len_exp
#
self._max_len_send = 2 ** (9 + (ord(self._handshake_bytes[1]) >> 4))
if self.factory.debug:
log.msg("WampRawSocketProtocol: server requests us to send out most {} bytes per message".format(self._max_len_send))
# client wants to speak this serialization format
#
ser_id = ord(self._handshake_bytes[1]) & 0x0F
if ser_id != self._serializer.RAWSOCKET_SERIALIZER_ID:
if self.factory.debug:
log.msg("WampRawSocketProtocol: opening handshake - no suitable serializer found (server replied {0}, and we requested {1})".format(ser_id, self._serializer.RAWSOCKET_SERIALIZER_ID))
self.abort()
self._handshake_complete = True
self._on_handshake_complete()
if self.factory.debug:
log.msg("WampRawSocketProtocol: opening handshake completed", self._serializer)
# consume any remaining data received already ..
#
data = data[remaining:]
if data:
self.dataReceived(data)
class WampRawSocketFactory(Factory):
"""
Base class for Twisted-based WAMP-over-RawSocket factories.
"""
class WampRawSocketServerFactory(WampRawSocketFactory):
"""
Base class for Twisted-based WAMP-over-RawSocket server factories.
"""
protocol = WampRawSocketServerProtocol
def __init__(self, factory, serializers=None, coalesce=10000, debug=False):
"""
:param factory: A callable that produces instances that implement
:class:`autobahn.wamp.interfaces.ITransportHandler`
:type factory: callable
:param serializers: A list of WAMP serializers to use (or None for default
serializers). Serializers must implement
:class:`autobahn.wamp.interfaces.ISerializer`.
:type serializers: list
:param coalesce: Coalesce writes within this many ms. If active (>0 ms), writes are not
directly passed to the underlying Twisted stream transport, all writes
during a period of the given ms, the written data is coalesced into a flat buffer,
and that buffer is then written to the underlying Twisted transport at the end of the
period. When a new period starts with an empty coalescing buffer. Enabling this feature
can massively increase the throughput by reducing the syscall rate and reducing the
data moving overhead - at the price of increased latencies.
:type coalesce: None or int
"""
assert(callable(factory))
assert(coalesce is None or (type(coalesce) in six.integer_types and (coalesce == 0 or (coalesce >= 500 and coalesce <= 100000))))
self._factory = factory
self._coalesce = coalesce
self.debug = debug
if serializers is None:
serializers = []
# try MsgPack WAMP serializer
try:
from autobahn.wamp.serializer import MsgPackSerializer
serializers.append(MsgPackSerializer(batched=True))
serializers.append(MsgPackSerializer())
except ImportError:
pass
# try JSON WAMP serializer
try:
from autobahn.wamp.serializer import JsonSerializer
serializers.append(JsonSerializer(batched=True))
serializers.append(JsonSerializer())
except ImportError:
pass
if not serializers:
raise Exception("could not import any WAMP serializers")
self._serializers = {}
for ser in serializers:
self._serializers[ser.RAWSOCKET_SERIALIZER_ID] = ser
class WampRawSocketClientFactory(WampRawSocketFactory):
"""
Base class for Twisted-based WAMP-over-RawSocket client factories.
"""
protocol = WampRawSocketClientProtocol
def __init__(self, factory, serializer=None, coalesce=10000, debug=False):
"""
:param factory: A callable that produces instances that implement
:class:`autobahn.wamp.interfaces.ITransportHandler`
:type factory: callable
:param serializer: The WAMP serializer to use (or None for default
serializer). Serializers must implement
:class:`autobahn.wamp.interfaces.ISerializer`.
:type serializer: obj
:param coalesce: Coalesce writes within this many ms. If active (>0 ms), writes are not
directly passed to the underlying Twisted stream transport, all writes
during a period of the given ms, the written data is coalesced into a flat buffer,
and that buffer is then written to the underlying Twisted transport at the end of the
period. When a new period starts with an empty coalescing buffer. Enabling this feature
can massively increase the throughput by reducing the syscall rate and reducing the
data moving overhead - at the price of increased latencies.
:type coalesce: None or int
"""
assert(callable(factory))
assert(coalesce is None or (type(coalesce) in six.integer_types and (coalesce == 0 or (coalesce >= 500 and coalesce <= 100000))))
self._factory = factory
self._coalesce = coalesce
self.debug = debug
if serializer is None:
# try MsgPack WAMP serializer
try:
from autobahn.wamp.serializer import MsgPackSerializer
serializer = MsgPackSerializer()
except ImportError:
pass
if serializer is None:
# try JSON WAMP serializer
try:
from autobahn.wamp.serializer import JsonSerializer
serializer = JsonSerializer()
except ImportError:
pass
if serializer is None:
raise Exception("could not import any WAMP serializer")
self._serializer = serializer
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Research direction
Start with the archived WampRawSocketProtocol prototype, especially connectionMade, sendString, and the LoopingCall-based buffer. Compare this design with the current RawSocket transport implementation, which the issue does not identify. Done would require an agreed implementation scope, tests for batching and connection shutdown, and confirmation that outgoing messages are periodically coalesced.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- networking
- Issue type
- Feature
- Difficulty
- 5/5
- Estimated time
- Over a week
- Activity status
- Stale
- Clarity
- Needs clarification
- Newbie friendliness
- 25/100