Data Engine v2

[twitu]

Based on extensive discussion, we've come up with a new design for the data engine which is closely tied to the new design for the message bus (#1712). This is an overview of the current design where the message bus is responsible for two different kinds of communications.

• Message passing between components
• Data request/response between actors and data engines

Design v2

In the new design we want to move the data req/response flow out of the message bus -

• message bus does send, pub/sub
• data engine does data req/response

The new design

The main advantages of this architecture are -

• The internal state of the actor is not shared across threads in a closure thus removing the need for the locks
• It removes a layer of indirection between actors and data engines

We want to embrace the actor model. The actor has only one job and to process data responses. The data engine takes responses, maps them to the appropriate actor task and As it processes a data response, it can choose to communicate with other components via the message bus or send new data requests to the data engine. Internally the actor wraps state that it can share with other handlers, these handlers can be registered with the message bus and can be called by other components.

From the actors perspective this simplifies the control flow in the new design -

• data engine "drives" the actor
• actors send new data requests to the data engine - indirectly "driving" it

[twitu]

The data engine is a hub that facilitates communication between actors and data clients. Actors ask clients for data through requests and subscriptions to which clients reply. Clients reply with two different kinds of data:

• Historical data packaged in DataResponses
• Streams of data based on the current clock time as individual data items Data like quote ticks, trade ticks etc

pub trait DataClient {
    fn execute(&self, req: SubscriptionRequest);
    fn request(&self, req: DataRequest);
    
    /// helper methods
    fn is_subscribed(&self) -> bool
}

The data engine is single-threaded and sync so the interface to the data clients is also sync. Any async processing must be done within the clients implementation of the request/execution handling.

Note that when executing subscription requests, it is expected that the data client internally maintains the state of existing subscriptions and does not resubscribe. Typically this might be done as below. This might require Arc if the subscription is done in an async task and depends on the clients implementation. However, the function call exposed by the client to the actor/engine must be sync.

    fn subscribe_order_book_deltas(&mut self, command: SubscriptionCommand) {
        // parse command
        
        // check membership before subscribing
        if !self.subscriptions_order_book_delta.contains(&instrument_id) {
            self.client
                .subscribe_order_book_deltas(instrument_id, book_type, depth)
                .expect("Error on subscribe");
        }

        // update membership after subscribing
        self.subscriptions_order_book_delta.insert(instrument_id);
    }

These two flows can be seen as below. Note that the data response and the data from the stream is not returned to the actor immediately. The flow is completed once the request processing task is created and once the data stream is created.

Clients return data to the data engine through a separate channel/queue. The data engine calls the appropriate entry point on the actor. For this actors must have two functions/entrypoints to receive and process the two kinds of data

pub trait Actor {
    fn handle_response(resp: DataResponse);
    fn handle_data(data: Data);
}

The data engine receives the data from the clients and passes them on to the actor thus "driving" actors. Note that multiple actors might be subscribed to a data stream so the data engine will have to use the message bus pub/sub pattern to send data to all the actors.