Describe the bug

I encountered the following bug when executing this simple video pipeline.

QObject::startTimer: Timers cannot be started from another thread

In short, we may have a problem if we try to use multi-threading OpenCV with the Qt backend. Since Qt is also a common GUI backend besides GTK, I would suggest some alternative solutions to the mandatory async Source/Sink trait in zenoh-flow in this case.

To reproduce

Please visit here.

System info

I found this issue because I use Arch on daily use. Here I summarize my observations.

• Arch Linux (latest)
  • Default package: OpenCV 4.6.0 + Qt (failed)
• Ubuntu 20.04
  • Default package: OpenCV 4.2.0 + GTK (passed)
  • Manual build: OpenCV 4.6.0 + Qt (failed)

After resolving some missing dependencies (clang-sys), I hit the following compilation error (zenoh-dependent):

   Compiling zenoh-plugin-trait v0.1.0-dev (https://github.com/eclipse-zenoh/zenoh.git?branch=master#70182ad7)
error[E0433]: failed to resolve: use of undeclared type `SharedMemoryAuthenticator`
   --> /home/enea/.cargo/git/checkouts/zenoh-cc237f2570fab813/70182ad/zenoh/src/net/transport/unicast/manager.rs:194:25
    |
194 |                 .insert(SharedMemoryAuthenticator::new().into());
    |                         ^^^^^^^^^^^^^^^^^^^^^^^^^ use of undeclared type `SharedMemoryAuthenticator`

cargo version 1.54, rustc version 1.54

This PR prepares the release of Zenoh-Flow version v0.4.0.

The remaining tasks are:

• [x] https://github.com/ZettaScaleLabs/zenoh-rpc/pull/4
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/3
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/4
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/5
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/6
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/10
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/7
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/13
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/12
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/11
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/14
• [x] https://github.com/ZettaScaleLabs/zenoh-flow/pull/19
• [ ] ~Update the README~
• [ ] https://github.com/ZettaScaleLabs/zenoh-flow-python/pull/14
• [ ] https://github.com/ZettaScaleLabs/zenoh-flow-examples/pull/2
• [ ] Add a release pipeline https://github.com/ZettaScaleLabs/zenoh-flow/pull/23 (will be done later)

Hello zenoh-flow team,

We use zenoh-flow in our project, this framework is great and meets our expectations!

In the process of project development, we have met some scenarios and can't find a suitable solution.

Feedback is here and we look forward to discussing suitable solutions with you.

1. In graph node configuration, the current state is that a key can only have one value:

configuration:
  key1: value1
  key2: value2
  key3: value3

But we need more fields to describe a key:

configuration:
  - id: key1
    foo: value
    bar: value
    baz: value
  - id: key2
    foo: value
    bar: value
    baz: value

or:

configuration:
  key1: 
    foo: value
    bar: value
    baz: value
  key2:
    foo: value
    bar: value
    baz: value

2. Development path of dynamic library are different with deployment path, we expect to change the default global graph args when runtime starting:

args:
  release_path: /opt/release/bin
  default_input_id: Input
  default_output_id: Output
  default_config_foo: foo
operators:
  - id : MyOperator
    uri: file://{{release_path}}/libmy_operator.so
    inputs:
      - id: {{default_input_id}}
        type: i32
    outputs:
      - id: {{default_output_id}}
        type: i32
    configuration:
      config_foo: {{default_config_foo}}

and launch with development path:

$ runtime --graph-file demo.yaml --runtime foo --release_path './build/bin'

More feedback will be updated after the demo test, looking forward to your reply：）

New functionalities

This PR implements the Zenoh-Flow runtime daemon.

• [x] Define resource tree for runtime data stored in zenoh
• [x] Implement a helper for CRUD operations on those resources
• [x] Define interfaces for the runtime and the CLI tool
• [x] Daemonize the runtime
• [x] Implement the CLI tool

Resource Tree

The resource tree in zenoh is represented by this picture.

Information is stored in the leaves, the root of the tree depends if we want to be a zenoh-plugin or a standalone process.

All the IDs are UUIDv4

The Registry will contain metadata for accessing the components.

Interface

The interface for the Runtime is defined as follow:

/// The interface the Runtime expose to a client
/// (eg. another runtime, the cli, the mgmt API)
#[znservice(
    timeout_s = 60,
    prefix = "/zf/runtime",
    service_uuid = "00000000-0000-0000-0000-000000000001"
)]
pub trait ZFRuntime {
    /// Sends an initiation request for the given [`FlowId`]
    /// Note the request is asynchronous, the runtime that receives the request
    /// flattens the descriptor, maps it to the infrastructure,
    /// and, creates the associated [`DataFlowRecord`].
    /// The record contains an [`Uuid`] that identifies the record.
    /// The actual instantiation process runs asynchronously in the runtime.
    async fn instantiate(&self, flow_id: String) -> ZFResult<DataFlowRecord>;

    /// Sends a teardown request for the given record identified by the [`Uuid`]
    /// Note the request is asynchronous, the runtime that receives the request will
    /// return immediately, but the teardown process will run asynchronously in the runtime.
    async fn teardown(&self, record_id: Uuid) -> ZFResult<DataFlowRecord>;

    /// Prepares the runtime to run the given record identified by the [`Uuid`].
    /// Preparing a runtime means, fetch the operators/source/sinks libraries,
    /// create the needed structures in memory, the links.
    /// Once everything is prepared the runtime should return the [`DataFlowRecord`]
    async fn prepare(&self, record_id: Uuid) -> ZFResult<DataFlowRecord>;

    /// Cleans the runtime from the remains of the given record.
    /// Cleans means unload the libraries, drop data structures and destroy links.
    async fn clean(&self, record_id: Uuid) -> ZFResult<DataFlowRecord>;

    /// Starts the sinks, connectors, and operators for the given record.
    async fn start(&self, record_id: Uuid) -> ZFResult<()>;

    /// Starts the sources for the given record.
    /// Note that this should be called only after the `start(record)` has returned
    /// successfully otherwise data may be lost.
    async fn start_sources(&self, record_id: Uuid) -> ZFResult<()>;

    /// Stops the sinks, connectors, and operators for the given record.
    /// Note that this should be called after the `stop_sources(record)` has returned
    /// successfully otherwise data may be lost.
    async fn stop(&self, record_id: Uuid) -> ZFResult<()>;

    /// Stops the sources for the given record.
    async fn stop_sources(&self, record_id: Uuid) -> ZFResult<()>;

    /// Starts the given graph node for the given instance.
    /// A graph node can be a source, a sink, a connector, or an operator.
    async fn start_node(&self, record_id: Uuid, node: String) -> ZFResult<()>;

    /// Stops the given graph node from the given instance.
    /// A graph node can be a source, a sink, a connector, or an operator.
    async fn stop_node(&self, record_id: Uuid, node: String) -> ZFResult<()>;

    /// Gets the state of the given graph node for the given instance.
    /// A graph node can be a source, a sink, a connector, or an operator.
    /// The node state represents the current state of the node:
    /// `enum ComponentState { Running, Stopped, Error(err) }`
    async fn get_node_state(&self, record_id: Uuid, node: String) -> ZFResult<ComponentState>;

    /// Sends the `message` to `node` for the given record.
    /// This is useful for sending out-of-band notification to a node.
    /// eg. in the case of deadline miss notification.
    async fn notify_node(
        &self,
        record_id: Uuid,
        node: String,
        message: ZFControlMessage,
    ) -> ZFResult<()>;

    /// Checks the compatibility for the given `operator`
    /// Compatibility is based on tags and some machine characteristics (eg. CPU architecture, OS)
    async fn check_operator_compatibility(&self, operator: ZFOperatorDescriptor) -> ZFResult<bool>;

    /// Checks the compatibility for the given `source`
    /// Compatibility is based on tags and some machine characteristics (eg. CPU architecture, OS)
    async fn check_source_compatibility(&self, source: ZFSourceDescriptor) -> ZFResult<bool>;

    /// Checks the compatibility for the given `sink`
    /// Compatibility is based on tags and some machine characteristics (eg. CPU architecture, OS)
    async fn check_sink_compatibility(&self, sink: ZFSinkDescriptor) -> ZFResult<bool>;
}

This interface will be exposed over Zenoh using zenoh-rpc

This PR changes the way extensions (e.g. the support for Python nodes) are defined and loaded by Zenoh Flow. Nowadays each extension needs to be configured in the runtime configuration file.

---
    pid_file : /var/zenoh-flow/runtime.pid
    path : /etc/zenoh-flow
    loader:
      ### this is what we mean by configured in the runtime configuration file
      extensions:
              - name: python
                file_extension: py
                source_lib: /home/ato/Workspace/zenoh-flow-python/target/release/libpy_source.so
                sink_lib: /home/ato/Workspace/zenoh-flow-python/target/release/libpy_sink.so
                operator_lib: /home/ato/Workspace/zenoh-flow-python/target/release/libpy_op.so
                config_lib_key: python-script
    zenoh :
      kind: peer
      listen: ["tcp/0.0.0.0:7997"]
      locators: []

They can instead be positioned on a directory like extension.d where each file contains the definition of one extension.

This facilitates adding extensions by just dropping a file in that directory.

.../extensions.d/
- 00-py.zfext
- 01-ws.zfext
- ....

Single extension description:

name: python
file_extension: py
source_lib: /etc/zenoh-flow/extension/python/libpy_source.so
sink_lib: /etc/zenoh-flow/extension/python/libpy_sink.so
operator_lib: /etc/zenoh-flow/extension/python/libpy_op.so
config_lib_key: python-script

In particular this PR implements:

• [x] Update the daemon configuration (and the default one)
• [x] Implement TryFrom<DaemonConfig> for the daemon
• [x] Refactor the LoaderConfig to hide the internal details.
• [x] Update .deb and .rpm generators
• [x] Load extensions at start-up

Signed-off-by: gabrik [email protected]

~This PR contains the code for the Zenoh-Flow registry and the cargo zenoh-flow subcommand.~

~- [ ] Registry daemon~ ~- Stores the metadata and the libraries for each component.~

• [x] cargo zenoh-flow
  • [x] New: creates a basic boilerplate for a Zenoh-Flow component
  • [x] Build: builds the component, generates the metadata needed for the Registry, and register it to the local registry ~- [ ] Push: pushes the component to a remote registry~ ~- [ ] List: shows the components stored on the local registry~ ~- [ ] Pull: pulls a component from a remote registry to the local one~

Now this PR contains only the code for the cargo-zenoh-flow subcommand.

Describe the bug

The job is indeed published on Zenoh on the correct key expression but no worker takes care of it.

The cause seems to be in the transformation of what we receive from Zenoh into a ZFJobStream: the resulting Future is never polled. The underlying issue remains to be determined (bad handling of Futures?).

To reproduce

Try to launch a DataFlow via zfctl.

System info

Zenoh-Flow dev/v0.4.0

Describe the feature

Dissociating both phases is a first step towards being able to "pause" a running data flow.

For now, when nodes are stopped, the only way to restart them is to call setup again. Calling setup regenerates (and clears) their internal state.

By dissociating the setup from the run, we would be able to choose to either resume (calling run & keeping the internal state) or reset (calling setup + run) the execution of a node.

Describe the bug

Consistent naming is preferable. Considering the fact that all unique ids are, in fact, UUID, having *_uuid seems the better option.

To reproduce

N/A

System info

N/A

Describe the bug

There is an inconsistency in the Output and Input types in the name of the function used to retrieve the ID.

In Input we have:

impl Input {
    pub fn id(&self) -> &PortId {
        &self.id
    }
    ...
}

While in Output we have:

impl Output {
    /// Returns the port id associated with this Output.
    ///
    /// Port ids are unique per type (i.e. Input / Output) and per node.
    pub fn port_id(&self) -> &PortId {
        &self.port_id
    }
    ...
 }   

They should be the same, so either id() or port_id()

To reproduce

N/A

System info

N/A

Describe the feature

It would be helpful to provide users the possibility to configure the Congestion Control and Reliability for the channels created by Zenoh Flow.

Currently, the behavior is to use CongestionControl::Block in ZenohSender. While this works in most of the cases, with some slow subscribers it would be better to have CongestionControl::Drop, thus avoiding filling the queues with "old data".

Summary

To measure and provide insights about Zenoh Flow’s performance, some metrics should be provided.

Intended outcome

"Metrics" (to be refined) are made available.

Steps

TBD.

Summary

To further ease the deployment of a data flow, an integration with Kubernetes is envisioned.

Intended outcome

A user can deploy a flow through Kubernetes’ command line tool (for instance: kubectl add flow data-flow.yaml).

Prerequisites

• Bindings in Go for Zenoh?

Steps

TBD.

Summary

Currently, the mapping of nodes is specified manually in the YAML file. Zenoh Flow should expose an interface that would let external "logic" perform this mapping.

Intended outcome

1. Zenoh Flow exposes an interface to perform this mapping.
2. We provide a simple example that assigns nodes to daemons randomly.

Steps

TBD.

Summary

Currently, the only URI that is supported for shared libraries (or Python script) is file://. Other schemas should be supported (for instance: http://, https://).

Intended outcome

Shared libraries or Python scripts should be fetched following other URI schemas.

Steps

• [ ] Determine a first list of schemas to support.
• [ ] Provide a working example for each schema.
• [ ] TBD.