YMQ's documentation

[gxuu]

This discussion provides documentation for YMQ. It contains some architecture graphs, execution flowcharts, and API descriptions.

YMQ is one of the network backend for scaler. It provides zmq like utilities such as guarantee message delivery, automatic reconnect, binder model, and pub/sub model (named UNICAST and MULTICAST), etc. Here's the general architecture and execution flowchart of the library:

User of YMQ interacts mainly with these APIs:

• IOContext.createIOSocket that generates a handle to an IOSocket
• IOContext.removeIOSocket returns the handle back to the system
• IOContext.requestIOSocketStop sends a stop request to an IOSocket from possibly another thread.
• IOSocket.bindTo, IOSocket.connectTo that turns an IOSocket into a Server or Client like object. The first one bind to an address and the second one connect to an address.
• IOSocket.sendMessage, IOSocket.recvMessage sends and receive a YMQ Message.
• IOSocket.closeConnection closes a underlying connection with the matched identity.

All of above APIs are asynchronous and completion are notified with a callback. Be noted that some callbacks are completed very fast, even before the actual completion is completed. This is expected, and user can make subsequent call even before the previous operation is really completed. For example,

IOSocket.connectTo(callback1)
IOSocket.sendMessage(callback2)

will work as expected regardless the callback1 is being called or is completed or not. callback2 will be called, however, only after the system decides the underlying connection is established.

Sometimes, asynchronous APIs are very wordy, one might choose to use the APIs provided in simple_interface.h.

YMQ also bundled with an easy to use Python interface, @magniloquency is responsible to write this part.

[magniloquency]

As @gxuu stated, YMQ has a Python interface, implementing using Python's C API, compatible with Python 3.8+. This interface mimics YMQ's C++ interface closely such that code written using either interface should appear quite similar.

YMQ's Python interface supports both asyncio and synchronous invocations, async methods are named matching the C++ API e.g. IOSocket.method(), and sync methods are always suffixed with _sync, as in: IOSocket.method_sync().

Example

from scaler.io.ymq import ymq

context = ymq.IOContext(num_threads=6)
socket = await context.createIOSocket("identity", ymq.IOSocketType.Connector)

# sync and async methods can be mixed freely
# (although this is rarely wise, be careful not to block the async executor)
socket.connect_sync("tcp://127.0.0.1:3333")
await socket.send(ymq.Message(None, b"payload"))

YMQ's Python interface offers the following types:

• Bytes: This represents a core type used to manage memory inside YMQ's core, and offers zero-copy reads through the buffer interface. They can be constructed by passing in anything supporting the buffer interface, e.g. Bytes(b"hello")
• Message: This represents a YMQ message, it is a nullable address plus a payload, both of which are Bytes. This type is used both for sending and receiving
• IOSocketType: An enum with values identical to the YMQ core's type
• IOContext: A handle to a C++ IOContext, it can be instantiated directly and accepts the number of worker threads as a parameter, just as the C++ IOContext does. This class has only one method (with an async and sync version):
  • createIOSocket(identity: str, socket_type: IOSocketType) / createIOSocket_sync(...): This method creates an IOSocket with the given identity and socket type, managed by this io context
• IOSocket: This is a handle to a C++ IOSocket. Can only be created by calling a method on IOContext. Supports both sync and async versions of the methods, including:
  • send(Message) / send_sync(Message, timeout: float | None)
    • Similarly for recv, connect, bind, etc.
  • identity: str and socket_type: IOSocketType for accessing properties of the socket

The interface includes a few types for error handling:

• ErrorCode: This IntEnum matches up exactly with the YMQ core's error code enum
• YMQException: This is the exception thrown by YMQ methods when the core returns an error. It contains a code: ErrorCode field that can be checked to determine what kind of error occured, and a message: str field generated by the core. It is recommended that you always check the error code, some errors may be fatal to your application, while others may be ignored, for example:

try:
    msg = await socket.recv()
except YMQException as e:
    if e.code == ErrorCode.ConnectorSocketClosedByRemoteEnd:
        return # not an error for our use case, we can exit
    raise # re-raise all other errors