Part 8 (Config)

Pipeline and Stage configuration in Ark is done through YAML files. The core concept is that there is a top-level “pipeline configuration” file that references per-stage configurations.

The pipeline configuration is able to “override” per-stage variables. Stage configurations are written in rbuf, so that your configurations are structured.

Let’s start by first adding configuration to a new rbuf at ${APP_ROOT}/tutorial/config.rbuf:

schema FirstStageConfig
{
    double rate_s;
    string message;
}

We will allow configuring the timer rate and a message. We need to update the CMakeLists.txt file in ${APP_ROOT}/tutorial/CMakeLists.txt to build the config. Add this line to the bottom of the file:

add_rbuf(tutorial_config config.rbuf)

Updating First Stage

Next, we’ll want to update the first stage to use configuration. Let’s revisit the ${APP_ROOT}/tutorial/first_stage/first_stage.cc file:

#include "tutorial/first_stage/first_stage.hh"
#include "tutorial/config.hh"
#include "tutorial/messages.hh"

#include "ark/pipeline/config.hh"
#include "ark/pipeline/publisher.hh"
#include "ark/pipeline/periodic_timer_config.hh"
#include "ark/pipeline/stage_interface.hh"

#include "fmt/format.h"

void FirstStage::initialize(ark::pipeline::StageInterface &interface)
{
    // Fetch configuration. This will be a FirstStageConfig struct
    // with to-be-defined values populated.
    auto stage_config = ark::pipeline::get_stage_config<FirstStageConfig>(interface);

    // Convert the rate double into a chrono object.
    auto rate = std::chrono::duration<double>(stage_config.rate_s);

    auto publisher = interface.add_publisher<MyString>("/my_strings");

    {
        ark::pipeline::PeriodicTimerConfiguration config;

        config.name = "emit_string";
        config.rate = std::chrono::duration_cast<std::chrono::nanoseconds>(rate);

        // Note the callback now captures the stage config
        config.callback = [publisher, stage_config](const auto &now) {
            auto time_s = std::chrono::duration<double>(now.time_since_epoch());

            MyString string;

            // The message we print out will now include the configured message.
            string.now = now;
            string.message = fmt::format("At {}, my message is '{}'", time_s.count(), stage_config.message);

            publisher->push(string);
        };

        interface.add_timer(config);
    }
}

We’ll also need to update the CMakeLists.txt file for the first stage, so that it can find the new message. Update ${APP_ROOT}/tutorials/first_stage/CMakeLists.txt to reference the new tutorial_config library you made:

add_library(first_stage first_stage.cc)

target_link_libraries(first_stage
    PUBLIC ark::pipeline
    PRIVATE fmt::fmt tutorial_messages tutorial_config)

We are now loading a configuration, but we should create one first.

Edit the ${APP_ROOT}/tutorial/first_stage_config.yml file:

config:
  default:
   rate_s: 0.5
   message: "Configured Message!"

Running

Next, let’s build and run the pipeline:

~/ark$ ./build/tutorial_pipeline
(2022-02-02 14:28:44.018) [error] (realtime_executor.cc:202) Exception during initialization: Exception while initializing FirstStage: Tried to request the config for 'FirstStage', but that config was not registered.
(2022-02-02 14:28:44.018) [error] (main_offboard.cc:340) Unhandled error during pipeline execution: Exception while initializing FirstStage: Tried to request the config for 'FirstStage', but that config was not registered.
(2022-02-02 14:28:44.018) [error] (tutorial_pipeline.cc:25) Exiting due to error: Exception while initializing FirstStage: Tried to request the config for 'FirstStage', but that config was not registered.

The pipeline terminated during initialization (fairly focally), indicating that the FirstStage was trying to access a configuration, but it wasn’t found.

This is because we don’t have a pipeline configuration yet.

Pipeline Config (First Stage)

We’ll next want to create a pipeline configuration.

Edit ${APP_ROOT}/tutorial/pipeline/tutorial_pipeline.yml:

package:
  configs:
    FirstStage:
      path: "tutorial/first_stage_config.yml"

This creates a pipeline configuration, registering a single stage, FirstStage, and pointing at the YAML file that you created.

We now can run the pipeline pointing at that config:

~/ark$ ./build/tutorial_pipeline
(2022-02-02 14:56:52.328) [error] (tutorial_pipeline.cc:30) Exiting due to error: Tried to set a pipeline configuration, but the active configuration already has content. This typically means that you have either tried to set the config package twice, or (more likely) you are mixing `ark::pipeline::add_stage_xxx` calls with using a config package, which is not currently supported.

Right now, you cannot mix calls to add_stage_with_namespace and a configuration package. We’ll to make a few more adjustments before we can run things.

Pipeline Config (Second Stage)

We’ll need to set the namespaces for the second stages in the pipeline.

Edit ${APP_ROOT}/tutorial/pipeline/tutorial_pipeline.yml:

package:
  configs:
    FirstStage:
      path: "tutorial/first_stage_config.yml"

    SecondStage01:
      metadata:
        stage_config:
          comms_namespace: "/second01"

    SecondStage02:
      metadata:
        stage_config:
          comms_namespace: "/second02"

This will apply stage_config metadata to both SecondStage01 and SecondStage02, assigning them an appropriate comms namespace.

Then go back and update your ${APP_ROOT}/tutorial/pipeline/tutorial_pipeline.cc file:

#include "tutorial/first_stage/first_stage.hh"
#include "tutorial/second_stage/second_stage.hh"

#include "ark/comms/stages/http_server_stage.hh"
#include "ark/debuglog/log.hh"
#include "ark/main/main_offboard.hh"

int main(int argc, const char **argv)
{
    ark::pipeline::Pipeline pipeline;

    pipeline.add_stage<FirstStage>();
    pipeline.add_stage<SecondStage>("SecondStage01");
    pipeline.add_stage<SecondStage>("SecondStage02");
    pipeline.add_stage<ark::comms::HttpServerStage>();

    try
    {
        ark::main::MainOffboardConfiguration config;
        config.pipeline_config_path = "tutorial/pipeline/tutorial_pipeline.yml";

        return ark::main::execute_realtime_pipeline(argc, argv, std::move(pipeline), config);
    }
    catch (const std::exception &exception)
    {
        LERROR("Exiting due to error: {}", exception.what());
        return 1;
    }
}

As you can see, we create two second stages, but we no longer add them with the namespaced configuration API, we rely on the config package to resolve that.

Finally, let’s build and run our pipeline, it should finally run:

~/ark$ ./build/tutorial_pipeline -c ./tutorial/pipeline/tutorial_pipeline.yml
(2022-02-02 14:32:45.882) [info] (http_server_stage.cc:874) HTTP Server is listening on "0.0.0.0:8080".
(2022-02-02 14:32:46.381) [info] (second_stage.cc:39) SecondStage01 stage received message: 'At 0.500472099, my message is 'Configured Message!''
(2022-02-02 14:32:46.381) [info] (second_stage.cc:39) SecondStage02 stage received message: 'At 0.500472099, my message is 'Configured Message!''
(2022-02-02 14:32:46.881) [info] (second_stage.cc:39) SecondStage02 stage received message: 'At 1.000472099, my message is 'Configured Message!''
(2022-02-02 14:32:46.881) [info] (second_stage.cc:39) SecondStage01 stage received message: 'At 1.000472099, my message is 'Configured Message!''
(2022-02-02 14:32:47.381) [info] (second_stage.cc:39) SecondStage02 stage received message: 'At 1.500472099, my message is 'Configured Message!''
(2022-02-02 14:32:47.381) [info] (second_stage.cc:39) SecondStage01 stage received message: 'At 1.500472099, my message is 'Configured Message!''

Great! We are now running twice as fast (once every half second, or 2Hz), and the message has changed to ‘configured message’. Feel free to play around with the config file and observe changes taking effect.

Default Pipeline Config

We don’t want to need to specify that command line argument every time, though, let’s once again update ${APP_ROOT}/tutorial/pipeline/tutorial_pipeline.cc.

Change the execute_realtime_pipeline block in your try/catch to something like this:

ark::main::MainOffboardConfiguration config;
config.pipeline_config_path = "tutorial/pipeline/tutorial_pipeline.yml";

return ark::main::execute_realtime_pipeline(argc, argv, std::move(pipeline), config);

The MainOffboardConfiguration allows you to tweak numerous settings in how your pipeline runs.

Building and running now should work without specifing the config manually:

~/ark$ ./make.sh
~/ark$ ./build/tutorial_pipeline
(2022-02-02 14:36:14.659) [info] (http_server_stage.cc:874) HTTP Server is listening on "0.0.0.0:8080".
(2022-02-02 14:36:15.158) [info] (second_stage.cc:39) SecondStage02 stage received message: 'At 0.500650819, my message is 'Configured Message!''
(2022-02-02 14:36:15.158) [info] (second_stage.cc:39) SecondStage01 stage received message: 'At 0.500650819, my message is 'Configured Message!''

In Part 9, we’ll go into logging our messages, so we can debug our pipeline offline.