Forem: Ivan Kostruba

Modern C++ Features and Proven Concepts: Active Object, External Polymorphism and Coroutines

Ivan Kostruba — Mon, 09 Jan 2023 08:14:19 +0000

Summary

In this article, I'll show you how external polymorphism helps you write beautiful and clean programs, and talk about some basic and advanced implementation techniques. The good old Active Object concurrency pattern will be the running example. In the end, I'll show you how easy it is to implement this pattern using C++20 coroutines and how you can use them to make Active Object even more powerful by implementing true asynchronous functions in it.

Problem Statement

Imagine we have an established VoIP call between a user device and our business logic server. The user is in the voice menu and listens to prompts that the server plays for them. The user can press numbers to select options in the menu or can hang up and end the call. To support it, we must receive and parse SIP protocol messages and run the corresponding business logic. The details of SIP communications are beyond the scope of this example, we will focus only on the business logic. We must keep in mind that we must parse the messages as quickly as possible, but the business logic of getting the next prompt and playing it back to the user may be slower to process and should not interfere with the processing of the SIP communication. This is where Active Object comes in handy.

Active Object Internals

Even if the reader is not familiar with the aforementioned pattern, it’s pretty easy to grasp. Here is the simplest implementation.

class VoiceMenuHandler {
public:
    // These methods will return immediately, deferring the actual work.
    void receiveInput(const MenuInput& data);
    void receiveHangup(const HangUp& data);

private:

    std::string fetchMenuSectionPrompt(
        char digit, const std::string& callId);
    void playVoiceMenuPrompt(
        const std::string& callId, const std::string& prompt);
    // This method will do actual processing of user input by calling
    // the two functions from above.
    void processInput(const MenuInput& data);

    void cleanupCallData(const std::string& callId);
    // This method will do actual processing of hangup by calling the
    // function from above.
    void processHangup(const HangUp& data);

    // The thread where the tasks will run.
    Worker worker_;
};

The worker object here combines a thread and a mutex-protected queue. For the sake of brevity this is just a snippet, you can find the full implementation here if you are interested - GitHub link.

However, there is an important question about Worker: what exactly will we put in the task queue? As you can see, our controller has two "tasks": processInput and processHangup. They have different signatures, so they are two different types and cannot be placed into the same container. In the classical approach, we would create a virtual base object and derive two objects from this base, one for each task. This would be quite a lot of boilerplate, and for sure in modern C++ we can do better.

Starting from C++11 we have a type-erased wrapper for any kind of callable object - std::function. By the way, this is an example of external polymorphism where the polymorphic behavior is the function or functor call itself. Using that we could write something like the following:

// Declaration of the queue inside the Worker
std::queue<std::function<void()>> queue;

// Public method of the Worker to schedule a task
void Worker::addTask(std::function<void()> task) {
    // locking the mutex is omitted
    queue_.emplace(std::move(task));
}

// And then in the implementation of the VoiceMenuHandler -
void VoiceMenuHandler::receiveInput(const MenuInput& data) {
    // Capture the handler instance and the arguments in the lambda,
    // and pass it wrapped into a std::function
    worker_.addTask( [this, data]() { processInput(data); } );
}

// And the same for the Hangup signal
void VoiceMenuHandler::receiveHangup(const HangUp& data) {
    // This lambda has the same signature, hence it will fit perfectly
    // into the queue.
    worker_.addTask([this, data]() { processHangup(data); });
}

This will solve our problem, but std::function is the least efficient way to implement the solution. To support all possible types and sizes of callable objects, std::function must perform heap allocation on creation, and allocation is slow. I will show the numbers a little later, but we can do two or three times faster with a little effort.

External Polymorphism Implementation

So how to write your own type erased wrapper? Let’s start with a naive implementation without any type erasure.

template<typename Handler, typename Data>
class SimpleTask {
public:
    SimpleTask(Handler* handler, Data data)
        : handler_{ handler }, data_{ data }
    {}

    void operator()() {
     handler_->process(data_);
    }

private:
    Handler* handler_;
    Data data_;
};

This is a very simple template class that contains a pointer to a message handler instance and arguments. If we only had one kind of task, that would be enough, but our VoiceMenuHandler has two tasks, and instantiating this template for each of them will create two different types that we cannot store in the same queue. So we really need type erasure.

There are two fantastic talks regarding this topic, one by Sean Parent (youtube) and another by Sy Brand (youtube). My implementation is based on the latter but the former has a really good example of the benefits of type erasure and External Polymorphism. Here’s the code.

// First of all - here's the beauty of it. As clean and simple as
// std::function. Without deriving our objects from  any base class
// we have full polymorphic behavior when we need it. And when
// we don't need it we don't pay anything.

auto functorWrapper = TaskWrapper{ MyFunctorObject };
auto lambdaWrapper = TaskWrapper{ [](){ std::cout << "lambda!\n"; } };
std::queue<TaskWrapper> queue;
queue.emplace(std::move(functorWrapper));
queue.emplace(std::move(lambdaWrapper));
queue.front()(); // invoke MyFunctorObject.operator()
queue.pop();

// Implementation of the wrapper type.

namespace _detail {

    // Definition of self-made virtual table. It holds several
    // function pointers.
    struct vtable {
        // Main part of the logic, this will invoke the stored callable.
        void (*run)(void* ptr);
        // These are necessary for correct copy and destruction of
        // the stored objects.
        void (*destroy)(void* ptr);
        void (*clone)(void* storage, const void* ptr);
        void (*move_clone)(void* storage, void* ptr);
    };

    // Template variable from C++17. We instantiate vtable,
    // initializing the pointers with lambdas that decay to
    // function pointers because they have empty capture lists.
    template<typename Callable>
    constexpr vtable vtable_for{
        [](void* ptr) {
            // Inside each of the lambdas we restore the type info
            // using the information from the template parameters.
            static_cast<Callable*>(ptr)->operator()();
        },

        // Destructor
        [](void* ptr) {
            std::destroy_at(static_cast<Callable*>(ptr));
        },
        // Copy constructor
        [](void* storage, const void* ptr) {
        new (storage) Callable {
            *static_cast<const Callable*>(ptr)};
        },
        // Move constructor
        [](void* storage, void* ptr) {
            new (storage) Callable {
            std::move(*static_cast<Callable*>(ptr))};
        }
    };

}; // namespace _detail

class TaskWrapper {
public:
    TaskWrapper() : vtable_{ nullptr }
    {}

    // We need all the constructors and assignment operator machinery.
    TaskWrapper(const TaskWrapper& other) {
        other.vtable_->clone(&buf_, &other.buf_);
        vtable_ = other.vtable_;
    }

    TaskWrapper(TaskWrapper&& other) noexcept {
        other.vtable_->move_clone(&buf_, &other.buf_);
        vtable_ = other.vtable_;
    }

    ~TaskWrapper() {
        if (vtable_) {
            vtable_->destroy(&buf_);
        }
    }

    TaskWrapper& operator=(const TaskWrapper& other) {
        if (vtable_) {
            vtable_->destroy(&buf_);
        }
        if (other.vtable_) {
            other.vtable_->clone(&buf_, &other.buf_);
        }
        vtable_ = other.vtable_;
        return *this;
    }

    TaskWrapper& operator=(TaskWrapper&& other) noexcept {
        if (vtable_) {
            vtable_->destroy(&buf_);
        }
        if (other.vtable_) {
            other.vtable_->move_clone(&buf_, &other.buf_);
        }
        vtable_ = other.vtable_;
        return *this;
    }

    // This is where the magic happens. We create a virtual table
    // instance that 'remembers' the type information while the
    // callable is stored by placement new in the internal buffer,
    // e.g. as a type-less set of bytes.
    // Placement new essentially implements small buffer optimization.
    // Instead of allocating on the heap, we store data on the stack.
    // This is how we gain in performance.
    template<typename Callable>
    TaskWrapper(Callable c)
        : vtable_{ &_detail::vtable_for<Callable> }
    {
        static_assert(sizeof(Callable) < sizeof(buf_),
            "Wrapper buffer is too small.");
        new(&buf_) Callable{ std::move(c) };
    }

    // This is where we invoke the stored callable.
    void operator()() {
        if (vtable_) {
            vtable_->run(&buf_);
        }
    }

private:
    // We use aligned storage to ensure that the wrapped object
    // will be properly aligned.
    std::aligned_storage_t<64> buf_;
    const _detail::vtable* vtable_;
};

This implementation has one limitation, it only supports one specific callable object signature - a void return type, and an empty argument list, this is done for the sake of simplicity. A little more complex configurable implementation can be found here - GitHub link.

You may think that this is a bit too much boilerplate. But this only has to be written once and then you can use it across all your active objects! Implementation of VoiceMenuHandler is quite simple.

void VoiceMenuHandler::receiveInput(const MenuInput& data) {
    // Instead of processing the input synchronously, we schedule it
    // for later, and return immediately, not wasting the caller's time.
    worker_.addTask(
        TaskWrapper{[this, data]() { processInput(data); } });
}

void VoiceMenuHandler::receiveHangup(const HangUp& data) {
    worker_.addTask(
        TaskWrapper{[this, data]() { processHangup(data); } });
}

std::string VoiceMenuHandler::fetchMenuSectionPrompt(
    char digit,
    const std::string& callId
) {
    // Given the current state and user input, go to the next menu item.
    std::cout << "in the call [" << callId << "] menu item '"
        << digit << "' selected.\n";
    return callId + "_prompt_" + digit;
}

void VoiceMenuHandler::playVoiceMenuPrompt(
    const std::string& callId,
    const std::string& prompt
) {
    // Command media server to start playing the current menu prompt.
    std::cout << "play prompt [" << prompt << "]\n";
}

void VoiceMenuHandler::processInput(const MenuInput& data) {
    const auto prompt = fetchMenuSectionPrompt(data.digit, data.callId);
    playVoiceMenuPrompt(data.callId, prompt);
}

void VoiceMenuHandler::cleanupCallData(const std::string& callId) {
    // Free the resources that we used to serve the call.
    std::cout << "call [" << callId << "] ended.\n";
}

void VoiceMenuHandler::processHangup(const HangUp& data) {
    cleanupCallData(data.callId);
}

And this is how it can be used:

VoiceMenuHandler menuHandler;
std::thread sender([&menuHandler]() {
    menuHandler.receiveInput(MenuInput{ '2', "call_1@ip_addr" });
    menuHandler.receiveInput(MenuInput{ '1', "call_2@ip_addr" });
    menuHandler.receiveHangup(HangUp{ "call_1@ip_addr" });
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
});
sender.join();

The output will be:

in the call [call_1@ip_addr] menu item '2' selected.
play prompt [call_1@ip_addr_prompt_2]
in the call [call_2@ip_addr] menu item '1' selected.
play prompt [call_2@ip_addr_prompt_1]
call [call_1@ip_addr] ended.

Was It Worth The Effort?

Let’s see if our implementation is faster than std::function. I composed a little benchmark, in which I measured the time it took to create the wrapper, put it into a queue, retrieve it and invoke the stored callable. The computation in the callable is negligible, so the extra cost of the wrapper itself is measured.

As you can see, our original naive implementation is by far the fastest. So if a simple solution is enough for you, by all means go for it. Our wrapper adds some overhead for indirect calls, but it still runs twice as fast as std::function because it does not do any heap allocations. The last column shows the performance of coroutines which we are about to discuss.

Is Active Object Asynchronous Enough?

Let’s take a closer look at this method:

void VoiceMenuHandler::processInput(const MenuInput& data) {
    const auto prompt = fetchMenuSectionPrompt(data.digit, data.callId);
    playVoiceMenuPrompt(data.callId, prompt);
}

fetchMenuSectionPrompt sounds like a potential call to a remote service doesn’t it? Since we only have one worker thread, in case this call takes a long time, the whole queue will get stuck. It would be great to make this call asynchronous. One way to do it is to split this method into two. One will send a request, and when we receive a response, we’ll queue up another task to play the prompt. This approach will work, but the business logic will be scattered across two different functions, making it harder to read. And what if we have not two, but five or ten steps? Must we divide the logic into ten parts? Before C++20, we had no choice, but now we can implement asynchronous calls using coroutines. At first glance, coroutines can seem a bit intimidating due to the sheer number of customization points they provide, but actually implementing a coroutine is surprisingly easy. By the way, if you want to study the theory, there is a good talk by Andreas Fertig (youtube) and also the documentation at cppreference.com is really good.

First Step: Transform a Function into a Coroutine

In order to become a coroutine a function needs two things: use one of the co_await, co_yield or co_return keywords and return a handle to the coroutine (or, in case of the Microsoft compiler, a wrapper object containing the handle). Here’s the code for the return type we need.

struct CoroutineTask {
    // 'promise_type' is the mandatory element of the coroutine
    // return value type.
    struct promise_type;
    using handle_type = std::coroutine_handle<promise_type>;

    struct promise_type {
        CoroutineTask get_return_object() {
            return { handle_type::from_promise(*this) };
        }
        // We want our coroutine to be suspended on creation and
        // resume it later in the worker thread.
        std::suspend_always initial_suspend() noexcept { return {}; }
        // Don't suspend the coroutine when it reaches co_return.
        std::suspend_never final_suspend() noexcept { return {}; }
        // Our coroutine does not return any values, so its
        // promise_type defines the 'return_void' method, otherwise
        // definition of 'return_value' is necessary.
        void return_void() {}
        void unhandled_exception() {}
    };
    // This handle will be used in the worker thread to resume
    // the coroutine.
    handle_type h_;
};

You can find all the details about the lifetime of the object above at cppreference.com

When we have the right return type, only a slight modification will make our method into a coroutine. Note the return type and co_return keyword.

CoroutineTask VoiceMenuHandlerCoroutines::processInput(
    const MenuInput data
) {
    const auto prompt = fetchMenuSectionPrompt(data.digit, data.callId);
    playVoiceMenuPrompt(data.callId, prompt);
    co_return;
    // Coroutine state will be destroyed after this point.
}

// The public interface of VoiceMenuHandler has to change a little.

void VoiceMenuHandlerCoroutines::receiveInput(const MenuInput& data) {
    // Since the CoroutineTask::promise_type defines its
    // 'initial_suspend' method to return 'std::suspend_always'
    // the coroutine will be in suspended state after its creation,
    // so its body only runs when it is explicitly resumed.
    // Resume will happen in worker task.
    worker_.addTask( processInput(data).h_ );
}

void VoiceMenuHandlerCoroutines::receiveHangup(const HangUp& data) {
    // Note that we use the same queue and worker thread to run normal
    // tasks and coroutines.
    worker_.addTask(
        TaskWrapper{ [this, data]() { processHangup(data); } });
}

I want to draw your attention to the fact that the second receiveHangup method has remained unchanged, and we schedule old tasks together with new coroutines using the same task wrapper, which knows nothing about the coroutine_handle type. The coroutine handle doesn't know anything about our wrapper either, and yet they work well together. This is the power of external polymorphism! Full implementation of new VoiceMenuHandler can be found here - GitHub link.

Second Step: Implement an Async Operation

Transformation into a coroutine does not solve our problem by itself. We need to write the co_await keyword in front of the fetchMenuSectionPrompt call. But in order for this to work, fetchMenuSectionPrompt must first be modified to return an awaiter object that looks like following

// In order to be compatible with co_await a function must return an
// awaiter, that implements three special methods.
struct AwaitablePrompt {
    std::string callId;
    char digit;
    // External object that does the actual I/O
    PromptFetcher& fetcher_;
    std::string prompt_;

    // The special methods -
    bool await_ready();
    void await_suspend(std::coroutine_handle<> h);
    std::string await_resume();
};

// This will be called first when the compiler sees co_await expression.
bool AwaitablePrompt::await_ready() {
    // If PromptFetcher had a cache, we could check it here and
    // skip suspension of the coroutine by returning 'true'.
    return false;
}

// This will be called second.
void VoiceMenuHandlerAsync::AwaitablePrompt::await_suspend(
    std::coroutine_handle<> h
) {
    // When this method is called the coroutine is already suspended,
    // so we can pass its handle elsewhere.
    fetcher_.fetch(
        callId,
        digit,
        [this, h](
            const std::string& prompt,
            PromptFetcher::worker_type& worker
        ) {
        // co_await resumes execution right before 'await_resume' call
        // so it is safe to assign to prompt here because awaiter
        // lifetime lasts until after 'await_resume' return.
        prompt_ = prompt;
        // Worker to schedule the coroutine on will be provided externally.
        // Alternatively we could pass coroutine_handle to the fetcher
        // and then the fetcher would be responsible for scheduling it.
        worker.addTask(h);
        // from this point awaiter may be deleted at any moment so we
        // should not touch it (this* pointer) anymore. So assigning to
        // the prompt_ here would be potential undefined behavior.
        }
    );
}

std::string AwaitablePrompt::await_resume() {
    // This will become the result of co_await expression.
    return prompt_;
}

And this is how we need to change fetchMenuSectionPrompt, note its return type.

AwaitablePrompt VoiceMenuHandlerAsync::fetchMenuSectionPrompt(
    char digit,
    const std::string& callId
) {
    std::cout << "!Coroutine! - in call [" << callId
        << "] menu item '" << digit << "' selected.\n";
    return AwaitablePrompt{ callId, digit, fetcher_ };
}

If that looks a little confusing please check the full implementation here - GitHub link.
Finally, we can make our problematic method truly asynchronous.

CoroutineTask VoiceMenuHandlerAsync::processInput(const MenuInput data) {
    // co_await is compiled into the following sequence:
    // if(!AwaitablePrompt::await_ready()) {
    //   AwaitablePrompt::await_suspend(current_coro_handle);
    //   // depending on what await_suspend returns continuation
    //   // may be different.
    // }
    // AwaitablePrompt::await_resume()
    // whatever await_resume returns becomes the result of co_await
    const auto prompt = co_await fetchMenuSectionPrompt(
        data.digit, data.callId);
    playVoiceMenuPrompt(data.callId, prompt);
    co_return;
    // Coroutine state will be destroyed after this point.
}

So now, as soon as fetchMenuSectionPrompt is called this coroutine will be suspended and the worker thread will pick up the next item from the queue. And when we receive the response, we can schedule the processInput coroutine to be resumed on any thread! In my implementation, fetcher does it just to demonstrate the concept. See the code - GitHub link.

Testing the async version:

PromptFetcher fetcher;
VoiceMenuHandlerAsync menuHandlerAsync{ fetcher };
std::thread senderToAsync([&menuHandlerAsync, &fetcher]() {
    menuHandlerAsync.receiveInput(
        MenuInput{ '7', "call_async_9@ip_addr" });
    menuHandlerAsync.receiveInput(
        MenuInput{ '8', "call_async_8@ip_addr" });
    // "Receive" a response and play the prompt on the fetcher thread.
    fetcher.processResponse("call_async_8@ip_addr", "prompt_AAA");
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
    // "Receive" a response and play the prompt on the fetcher thread.
    fetcher.processResponse("call_async_9@ip_addr", "prompt_BBB");
    menuHandlerAsync.receiveHangup(HangUp{ "call_async_8@ip_addr" });
    std::this_thread::sleep_for(std::chrono::milliseconds(100));
});
senderToAsync.join();

Output will look like this:

!Coroutine! - in the call [call_async_9@ip_addr] menu item '7' selected.
fetch request sent callId [call_async_9@ip_addr], input = 7.
received response for [call_async_9@ip_addr]
play prompt [prompt_BBB]
!Coroutine! - in the call [call_async_8@ip_addr] menu item '8' selected.
fetch request sent callId [call_async_8@ip_addr], input = 8.

We can see that both requests were sent before we received any responses, and that the response to the second request came first, yet everything worked without any issues. So C++20’s coroutines provide really nice features and they are still faster than std::function :)

All right, that was a lot of information to digest. Thank you for your attention, and I hope you find this useful. You can find all the code from this article here - GitHub link.

Culture Clash in Code Review: Russia vs. USA

Ivan Kostruba — Mon, 17 Oct 2022 21:58:01 +0000

The title might be a bit dramatic, it was a bit of an argument between me and my colleague from the US during code review regarding the approach I took. Happens all the time, right? I was somewhat irritated because my colleague didn’t seem to hear my arguments very well. I explained how my solution is conceptually better than existing one, because it introduced a cleaner interface in a sense of “pure function” idea. But instead of attacking my concept my opponent simply pointed out that we could just use existing code here, here and there, and achieve the same result with existing utilities, resulting in a simpler code.

While his arguments made perfect sense and were completely logical, I couldn't help but keep wondering why he was thinking of these implementation details when we did not agree on the core concept! And then I stumbled upon this schema somewhere on the Internet. Note the last line, persuasion style.

The US and Russia lie on completely opposite ends of this axis! And Russia is on the “Principles First” side which means we tend to focus on philosophical ideas, theory and method first and all the implementation details are deduced from them. The US people on the other hand often focus on practical matters, and derive generic knowledge from specific use cases. Of course, individuals from both cultures may diverge from what’s generally accepted for that culture but a colleague and I fell so nicely into our respective statistical averages that it was even funny. Real “Aha!” moment for me.

In the past I had an idea that every software engineer in the world shares a similar background in (computer) science and strives for the perfection of the code we produce and thus all discussions should be automatically constructive, provided that we conduct them politely and logically consistently. I found that it’s true to some extent, and while speaking with engineers from around the world I definitely felt we have a lot in common, but there are a noticeable number of exceptions.

After researching this topic some more, I found that there is a recent book by Erin Meyer - “The Culture Map: decoding how people think, lead and get things done across cultures”. I read it and can absolutely recommend it if you want to know what all the other lines on that schema mean and how to deal with situations like the above in a productive and constructive way. Of course, something seems obvious, many people know about Japanese politeness and German punctuality, but there are also very interesting observations. Even some features of one's own culture become noticeable only when put in contrast with others. For example, I did not realize that Russian style of communication is relatively “High-context” meaning that many things are “in the air” or between the lines. Maybe that's one of the reasons why when I talk to people in Germany I often feel like I'm missing something. I thought it was because of the language barrier and I’m just an introvert (yes, I didn’t communicate with people much in Russia either), but maybe it’s also a cultural thing, and I’m looking for allegories which aren’t there at all.

Wrapping the post up, fortunately for me, nobody’s feelings were hurt during this review and even if I was a little annoyed I did not let it spill over into the conversation. Also, being both engineers, we valued logic and simplicity of code above everything else so we came to a consensus. But if I was aware of these cultural differences at the time of the review it would definitely go smoother.

Strive for simplicity: sanctions, transactions and a big refactoring

Ivan Kostruba — Thu, 22 Sep 2022 18:00:54 +0000

Failed deployment

It happened at my previous FinTech job. The new feature was critically important. Financial regulations required us to start screening third parties participating in transactions with our client against sanctions or PEP lists. If the person was on these lists, transactions with them as sender or beneficiary had to be investigated by Compliance. Before these new requirements we already had in place some mechanisms to block transactions for investigation, for example when a transaction amount exceeded a certain threshold.

So, one beautiful morning after development and testing were finished, a new release was rolled out and traffic started coming to the new build. Monitoring showed that API calls to a company that provides screening service were all successful, everything looked fine. However, within the hour the Compliance department raised an urgent issue – transactions that were blocked for investigation started to land on the clients' accounts! If left unresolved, that could lead to revocation of our license, so a rush began: accounts that received money in this manner were blocked, release was rolled back and transactions were manually adjusted. Quite a nervous and unpleasant day for everyone involved!

Let’s dig to the roots of the problem

How did it happen? We had automatic tests, code reviews, sandbox testing… let’s take a look at the code of the business logic. Disclaimer: real code is proprietary, so I will show some pseudocode that looks approximately the same but does not represent all of the real procedures. So here’s the high-level implementation of the process of initiation of a payout.

seqAnd(
  createTransaction(),
  label(“CollectData”),
  displayPage(“CollectPayoutData”),
  when(
    seqAnd(
      not(processHasAttribute(“Admin”)),
      not(productFlagIsSet(“NoOtpConfirmation”)))
  ).then(
    sendOtpSms(),
    displayPage(“EnterOtp”).onError(reset(“CollectData”))
  ),
  unloadAccount(),
  when(
    volatile(
      pepSanctionsScreening())
  ).then(
    enqueueApprovalRequest(),
    displayPage(“ComplianceCheck”)
  ),
  when(
    transactionHasStatus(“ToBeRejected”)
  ).then(
    loadAccount(),
    failure()
  ),
  initiateTransfer(),
…

If your impression was: “What the hell is going on here?”, then we are on the same page, this is how I felt all the time while working with this code.

This example shows that in order to describe the business logic a DSL was created. This DSL could control both backend operations and display of UI pages. Backend was written in C++, frontend was a web-UI with pages generated by the Bottle framework, written in Python. A business process written in this DSL was called an “Algorithm”. Every step of an “Algorithm” was an instance of a polymorphic class. Base interface of this class included methods: start() to invoke a function, stored in the instance, getNextSteps() to, who could have guessed, calculate next step (a key method for a flow control operator) and getState() that returned one of five possible states of a step: NotInvoked, Running, Success, Failure, Reset.

Evaluation of “Algorithms” result went as follows. First build a graph, representing all the steps. Then query the DB for the state of steps that have already been completed. Mark completed steps in the graph and find the next step by evaluating the flow control steps. Evaluation of the flow control statements was not quite straightforward, as the five states mentioned above had to be mapped to the binary logic. Every operator did it in a slightly different manner, but in the end somehow the next step was calculated. If that step is a background step, then a function contained inside it is invoked and considering its result the next evaluation iteration runs. If a step is a UI step, then “Algorithm” evaluation is interrupted until the result of user interaction comes through yet another entity, a “mediator”. Just to make reasoning about program state more fun there was a volatile wrapper, that forced step reevaluation on every “Algorithm” run, then there was reset wrapper that could mark some steps as NotInvoked returning to a specific label, just as the good old goto operator.

Are you saying: “This solution does not look particularly great”? Well, there's more. The system was not just your average, boring product, it was a modern and fancy distributed system, so it used a framework called Zeroc Ice. This is an RPC framework that also provides deployment configuration, service discovery, SSL encryption and more. By itself it’s rather interesting, but in our case it was used almost everywhere. Majority of the functions in the system were exposed as RPCs, which made them effectively globally visible. And they could call other RPCs and they called even more. And every call produced some side effects that were quite often used in condition expressions of “Algorithms” defined with DSL. These conditions were of course quite heavily nested. As a result, even though we could see in the monitoring interface which steps were already passed, it was almost impossible to tell what would happen next.

The error described in the beginning of the article was caused by multiple factors. The step that implemented screening was wrapped in volatile, so it was evaluated on every run of the “algorithm”. It was necessary as without that once the manual compliance control branch was taken, the other “happy” path could have never been selected even if compliance would approve the transaction. The other factor was a timer that periodically reran the “algorithm” so that it could check the relevant side effects and continue forward. When the new version with screening was invoked with the data from an already existing transaction that was blocked for compliance investigation, the sanctions screening request was sent (because it was not already done for the old transaction, right?). When the response came that screening was passed, the transaction was marked as clean, because the other checks (for amount and other attributes) that could have sent the transaction to investigation were already complete so they did not run again. So the now “clean” transaction proceeded happily to load money to the account, putting the whole company in danger. Nobody anticipated that, neither the author (me) nor reviewers, nor QA, nor automatic tests. I must say, test coverage was not the greatest, as in the abundance of long chains of RPCs, one had to mock tons of them for every test and there was never enough time for that.

I can not take it any more

I thought, and started to design a refactoring. Based on experience of work with the old “Algorithms” framework I defined the following problems:

Full-fledged flow control structures incite developers to move business logic on the DSL level.
Each step does not know anything about the others, so the only way to pass data between them is side effects, and that leads to implicit dependencies, obscure logic and overall fragility of the system.
Controlling frontend from backend bloats the code of a business process and different behavior of UI and backend steps produces two intertwined simultaneous flows inside the same process which is difficult to follow and modify. In other words the rich feature set of the DSL framework was just too much and led to overcomplication of the system. Extension and modification of the processes with their complicated logic, implicit dependencies and unpredictable side effects became too difficult and dangerous.

Another big architectural problem not related to the framework itself was direct dependency of business logic from the communication framework and DB access API.

Given these problems I set the following refactoring requirements:

Process state should be persistent, and it should be possible to interrupt and then resume the process, for example for retry or after receiving an asynchronous callback from a third party service.
New framework should not allow expressing business logic in the structure of a process. All branching should be done in “normal” code, and process structure should remain extremely simple. In fact, a linear sequence that just progresses from one step to the next was totally enough in our case.
It should be possible to pass data between steps in order to reduce reliance on side effects. Frontend should be separated from backend and interact with the latter through an API and not through some weird additional entities.
Business logic should be separated from implementation details, like an RPC mechanism or DB queries.

There are some existing utilities and frameworks that I checked. One is boost::future, that supports continuations, another is AWS Step Functions. The former does not provide capability to continue execution from an arbitrary point, the latter, despite having a nice UI for development and debugging, has the same problems that our framework had - business logic would creep into the Step Functions configuration language. Also, we did not want to be bound to a specific vendor. FSM framework from boost::statechart could be used, but it also has vastly excessive feature set compared to what I needed and some abstractions, like events, do not really fit in our use case.

Frontend separation was a challenge in itself, because we did not have any free frontend developers. In the end I had to learn React and compose the new UI myself. In fact, I did a lot of work myself because I was just a common (senior) developer, so I did not have any formal authority to implement my ideas with someone else’s hands. I had to persuade my colleagues and management that this refactoring is worth doing, and it’s hard to persuade anybody by just describing how great the system will become. People might even agree and say “yes, it would be cool to implement it”, but nobody will do anything. This is exactly how my first presentation failed.

What works much better is demonstration of an already working solution or at least a working prototype. I put in a lot of hours, including my free time, made a working MVP (fully redesigned one of the major business processes), and then it was a completely different talk. For example, it turned out that one of the backend developers also knows React and is willing to help with new UI for other processes. Simplicity of writing, testing and debugging of the new framework made a very good impression on other developers. Management liked that now it was possible to easily implement some money-saving features, and a large part of the work was already done, so risks and required time are quite clear, plus they suddenly got extra frontend devs, so they said yes and officially allocated time for that.

What did we get in the end

So instead of being part of complex class hierarchy each step became just a free function, taking a pair of arguments. First argument is a struct with input parameters. Second is a “context” or “environment” – an object that encapsulates all DB and RPC details. The return value of a step passed to the next as a first argument.

Here’s a simplified example of implementation of a single step. Its purpose is to load money back to a user’s account in case his outgoing transaction failed or was canceled by compliance.

// .h

// Each step defines an interface that business logic
// needs to interact with other modules.
// This implements the Dependency Inversion principle -
// instead of depending on other modules,
// business logic defines an interface, on which these
// modules will depend.
class PossibleRevertCtxI {
public:
  virtual bool loadAccount(int consumerId, int amount) = 0;
  virtual int transactionAmount(int transactionId) = 0;
};

std::optional<TransactionData> possibleRevert(
  ComplianceData data, PossibleRevertCtxI& ctx);

// .cpp

std::optional<TransactionData> possibleRevert(
  ComplianceData data, PossibleRevertCtxI& ctx
) {
  if (data.complianceDecision == REQUIRED) {
    // We have to wait for external input, so stop
    // the process on this step.
    return std::nullopt;
  }
  if (data.complianceDecision == REJECT) {
    ctx.loadAccount(
      data.consumerId,
      ctx.transactionAmount(data.transactionId));
    throw std::runtime_error{
      "Transaction rejected by Compliance." };
  }
  // decision is APPROVE or NOT_REQUIRED
  TransactionData result;
  result.consumerId = data.consumerId;
  result.requestId = data.requestId;
  result.transactionId = data.transactionId;
  return result;
}

A small library was created to take care of combining individual steps into chains. Change of state is dead simple. If a step returns value - chain switches to the next step. If a step returns std::nullopt, the chain does not switch, and execution stops. Naturally, if a step throws an exception chain is also interrupted. It is possible to run a whole chain in one go, or run just one step at a time, and obtain current step index and serialized parameters after each step in order to persist them. Library requires that the structures that hold parameters must be serializable, which means they must provide a std::string serialize() const method and they must be constructible from a string. Another limitation is that a step function must accept one or two arguments (either only the input data or input data and a “context”).

A business process implementation became something like this:

steps_chain::ChainWrapper payoutProcess(
  std::shared_ptr<ApiMock> api,
  std::shared_ptr<DbMock> db,
  std::shared_ptr<TimerMock> timer
) {
  // Steps are desined with the Interface Segregation
  // principle in mind, so they accept different
  // types as their 'context'. So we have to wrap
  // the steps in lambdas here, as ContextStepsChain
  // requires identical type of 'context' as a second
  // argument of all steps.
  return steps_chain::ChainWrapper {
    steps_chain::ContextStepsChain{
      [](InitialData d, std::shared_ptr<PayoutContext> c) {
        return unloadAccount(d, *c.get());
      },
      [](TransactionData d, std::shared_ptr<PayoutContext> c) {
        return sanctionsScreening(d, *c.get());
      },
      [](ComplianceData d, std::shared_ptr<PayoutContext> c) {
        return possibleRevert(d, *c.get());
      },
      [](TransactionData d, std::shared_ptr<PayoutContext> c) {
        return startTransfer(d, *c.get());
      }
    },
    std::make_shared<PayoutContext>(api, db, timer)
  };
}

And these chains are executed in the following manner:

void runProcess(
  steps_chain::ChainWrapper p,
  const std::string& requestId,
  std::shared_ptr<DbMock> db
) {
  const auto data = db->fetchProcessData(requestId);
  p.initialize(data.parameters, data.stepIdx);
  try {
    while (!p.is_finished()) {
      if (p.advance()) {
        const auto [idx, params] = p.get_current_state();
        db->setProcessData(data.requestId, idx, params);
      }
      else {
        // Log the interruption of the process.
        return;
      }
    }
  }
  catch (const std::exception& ex) {
    // Log the error and update process metadata.
    return;
  }
}

In this code chain state is initialized from the DB, but in case we receive additional data from an API or a callback it can be merged into the serialized parameters and thus passed to the next step. I used JSON as a serialization format partly because it’s easy to merge two JSON objects.

Overall, I believe, we got a pretty clear and simple system with nice separation of responsibility and isolation of business logic from details of RPC framework and other IO. Working with the new framework became way more pleasant, as my colleagues have said.

You can find detailed examples of code in my GitHub repository along with the public version of the library under MIT license.

I had to use a couple of interesting metaprogramming tricks during the development of the library along with a nice idiom of external polymorphism, but they probably deserve an article of their own.

Thank you for your attention.