Status Update: 2022-06-08

[phil-opp]

In this discussion, we describe our recent progress and the main topics that we would like to discuss in today's meeting. We use a separate post for each topic to keep the discussion focused. Feel free to create new posts for topics that are not mentioned yet.

[phil-opp]

Coordinator and Runtime Implementation

• Initial implementation of API crates, coordinator and runtime ready at Initial implementation of API crates, coordinator and runtime #19.
• dora-coordinator executable to visualize and run dataflows specified in YAML files. It supports two types of nodes: dora runtime nodes and custom nodes.

dora-coordinator run pylot.yaml
dora-coordinator visualize pylot.yaml

• Custom nodes are independent executables created by users. They link against a dora-node-api library crate to communicate with other nodes. All other logic (e.g. timeouts, input rules, etc) is implemented in custom code.

Custom Node exemple

use dora_node_api::{self, config::DataId, DoraNode};

#[tokio::main]
async fn main() -> eyre::Result<()> {
    let node = DoraNode::init_from_env().await?;

    let mut inputs = node.inputs().await?;
    let output = DataId::from("number".to_owned());
    let duration = Duration::from_secs(3);

    loop {
        let input = match timeout(duration, inputs.next()).await {
            Ok(Some(input)) => input,
            Ok(None) => break,
            Err(_) => bail!("timeout while waiting for input"),
        };

        match input.id.as_str() {
            "timestamp" => {
                let random: u64 = random();
                node.send_output(&output, &random.to_le_bytes()).await?;
            }
            other => eprintln!("Ignoring unexpected input `{other}`"),
        }
    }
    Ok(())
}

• Dora runtime nodes, on the other hand, run a predefined dora-runtime executable, which runs a set of light-weight operators. Operators are cooperative units of execution that follow a specific template. Right now, the PR only supports operators defined as shared libraries, but we also plan to add operators based on Python objects and WASM modules in the future. In addition to basic input and output handling, operators will be able to use a large set of advanced functionality, which will be provided by the dora-runtime.

Operator exemple

use dora_operator_api::{register_operator, DoraOperator, DoraOutputSender};

register_operator!(ExampleOperator);

#[derive(Debug, Default)]
struct ExampleOperator {
    time: Option<String>,
}

impl DoraOperator for ExampleOperator {
    fn on_input(
        &mut self,
        id: &str,
        data: &[u8],
        output_sender: &mut DoraOutputSender,
    ) -> Result<(), ()> {
        match id {
            "time" => {
                let parsed = std::str::from_utf8(data).map_err(|_| ())?;
                self.time = Some(parsed.to_owned());
            }
            other => eprintln!("ignoring unexpected input {other}"),
        }
        Ok(())
    }
}

• To make the creation of shared library operators easier, we provide a register_operator macro. Operator types just need to implement a DoraOperator trait and invoke the macro to autogenerate all the low-level binding code that the dora-runtime requires.

[phil-opp]

Communication Layer

• We have made an agnostic PubSub provider in order to switch protocol from the YAML Graph

#[async_trait]
pub trait CommunicationLayer {
    async fn subscribe<'a>(
        &'a self,
        topic: &str,
    ) -> Result<Pin<Box<dyn futures::Stream<Item = Vec<u8>> + 'a>>, BoxError>;

    async fn publish(&self, topic: &str, data: &[u8]) -> Result<(), BoxError>;

    fn publish_sync(&self, topic: &str, data: &[u8]) -> Result<(), BoxError>;

    async fn close(self: Box<Self>) -> Result<(), BoxError>;
}

Example YAML description:

communication:
  zenoh:
    prefix: /foo

• We have made a survey of other PubSub protocol client in Rust.

	package	Async Stream	Sync Iter	Maintained	Ref
mqtt	paho-mqtt	✔️	✔️	✔️	https://github.com/eclipse/paho.mqtt.rust
rabbitMQ	lapin	✔️	✔️	✔️	https://github.com/amqp-rs/lapin
redis	redis-rs	✔️	✔️	✔️	https://github.com/redis-rs/redis-rs
kafka	kafka-rust	❌	✔️	✔️	https://github.com/kafka-rust/kafka-rust
DDS	RustDDS	❌	❌	✔️	https://github.com/jhelovuo/RustDDS
zeromq	rust-zmq	❌	❌	❌	https://github.com/erickt/rust-zmq

• We have made an MVP of our communication layer trait on MQTT using paho-mqtt.

[phil-opp]

Connecting the coordinator and runtime implementation with Pylot

• We're moving the skaffold rust code for Pylot into proper custom node using the coordinator.
• We're planning on testing two version of Pylot: - one using only custom nodes, - and one using only operators.
• Pylot implementation using only custom nodes is close to finish, and is being used to monitor resource utilization and latency.
• Pylot implementation using only operator is going to help us define what distinctive feature should operator have.

[haixuanTao]

Modus Operandi of Releasing Dora?

• Code review should probably be before Open Summit.
• OpenSummit on 25 July.
• Announcement can be made by either Futurewei or us.

Open Atom First Contest on Open Source Software?

Brainstom the Scope, Compete, Organise, tools Contest.

The competition happens at the fall end of the year.

If Dora is ok in Q3, the incentive to push contribution.

Follow up action:
-> Write a white paper for a Contest under Dora.
-> Show the value of Rust, not, make people think that Dora is already mature and that people should directly build things on top of Dora.

Objectives:

• Creating communities.
• Not create over-expectancy about the ecosystem of Open Source Developers.
• Not make the competition overload the project that is very early stage.

Open Atom Contest on Algorithm?

More research-centric contest on validation of algorithm ( Software Research )