Feature: DAG

[kemingy]

Now Mosec only supports the linear pipeline. DAG can be helpful in some complex scenarios.

Feel free to leave comments about the possible scenarios and design.

[kemingy]

• use case

            /-> model_1 -\
preprocess  --> model_2 --> aggregate & postprocess
            \-> model_3 -/

[kemingy]

Design

• representation of DAG
  • Python support
  • unified format (decode & encode)
• protocol support
  • first bit in data: raw data or named data
    • named data: [name_length][name][data_length][data]
    • raw data: [data_length][data]

Edge case

• multiple endpoints: one for the response, one for the data dump
• error handling

Refs

• Airflow DAG

[lkevinzc]

Perhaps no need to change protocol? Just let Rust know and register the DAG information, then it can control the workflow.

[kemingy]

How are you going to send multiple data to one worker and distinguish them?

[lkevinzc]

The data should be ordered. For example, Rust side should have the DAG information, so that after sending to multiple workers in parallel and receiving all the results, it should concatenate in the batch dimension in a defined order and follow the same protocol to send to the next stage.

The next stage Python side is aware of the number of parallel workers and their order, so that it can de-batch them into a more clear structure (like dict of list of data, dict level indicates different worker, list indicates the batch).

[kemingy]

The Task structure will be changed a lot to support this feature.

Some format for graph representation: networkx

[lkevinzc]

Refactoring

Notations: N - number of stages; M - within one stage, number of parallel execution node; k - number of processes per node (assuming every node has the same k for notational simplicity). Note that in the current linear pipeline setting, M = 1.

• Current status: Python side: Server -> N * k * Coordinator (with Worker & Protocol-client); Rust side: N * Protocol-server.
• Introduce the abstraction layer named Workflow, which is responsible to build and maintain the DAG graph.
• Target status: Python side: Server -> Workflow -- each node --> Coordinator (with Worker & Protocol-client); Rust side: still N * Protocol-server, but with {node, edge} information to distribute the tasks.

[lkevinzc]

API-wise, for each Workflow, we can still use append_worker, so that if no other specification, we just treat it as linear pipeline with the append order. But if we need to enable the DAG functionality, we call something like Workflow.dagify(dagRepresentation), where the dagRepresentation can be object or string, to reorganize the execution flow.

notes: the name dagify just randomly appeared in my mind, then I searched and found this.