<commit_before>#include "utils/async_if_test_helper.hxx"

#include <set>

#include "nmranet/NMRAnetWriteFlow.hxx"
#include "nmranet/AsyncIfCan.hxx"
#include "nmranet/AsyncAliasAllocator.hxx"
#include "os/OS.hxx"

namespace NMRAnet
{

Executor<1> g1_executor("g1_exec", 0, 2000);
Executor<1> g2_executor("g2_exec", 0, 2000);
Executor<1> g3_executor("g3_exec", 0, 2000);
Executor<1> g4_executor("g4_exec", 0, 2000);

Executor<1>* round_execs[] = {&g1_executor, &g2_executor,
                              &g3_executor, &g4_executor};

/** This class will create an AsyncIf, two virtual nodes on it, and send one
 * unaddressed global packet each. */
class TestNode
{
public:
    TestNode(NodeID node_id)
        : nodeId_(node_id),
          ifCan_(round_execs[(node_id >> 1) & 3], &can_hub0, 10, 10, 2)
    {
    }

    void start(BarrierNotifiable* done)
    {
        ifCan_.add_addressed_message_support();
        ifCan_.set_alias_allocator(new AsyncAliasAllocator(nodeId_, &ifCan_));
        {
            // Adds one alias buffer to the alias allocation flow.
            auto* b = ifCan_.alias_allocator()->alloc();
            ifCan_.alias_allocator()->send(b);
        }
        auto* b = ifCan_.global_message_write_flow()->alloc();
        b->data()->reset(If::MTI_EVENT_REPORT, nodeId_,
                         EventIdToBuffer(nodeId_));
        b->set_done(done->new_child());
        ifCan_.global_message_write_flow()->send(b);
        b = ifCan_.global_message_write_flow()->alloc();
        b->data()->reset(If::MTI_EVENT_REPORT, nodeId_ + 1,
                         EventIdToBuffer(nodeId_ + 1));
        b->set_done(done->new_child());
        ifCan_.global_message_write_flow()->send(b);
    }

    ~TestNode()
    {
        //ifCan_.alias_allocator()->TEST_finish_pending_allocation();
        Executor<1>* e = round_execs[(nodeId_ >> 1) & 3];
        while(!e->empty() || !g_executor.empty()
              || !ifCan_.alias_allocator()->is_waiting()) {
/*              !ifCan_.frame_dispatcher()->IsNotStarted() ||
              !ifCan_.dispatcher()->IsNotStarted() ||
              !can_hub0.empty() ||
              !gc_hub0.empty()) {*/
            usleep(100);
        }
    }

private:
    NodeID nodeId_;
    AsyncIfCan ifCan_;
//    AliasInfo testAlias_;
};

class Stats : public CanHubPortInterface
{
public:
    Stats()
        : timer_(&Stats::timer_callback, this, nullptr),
          numFrames_(0),
          badFrames_(0),
          cidFrames_(0),
          ridFrames_(0),
          amdFrames_(0),
          eventFrames_(0)
    {<commit_msg>Fixes flakiness in async_if_stress test. It was not waiting for the dispatcher to be empty.<commit_after>#include "utils/async_if_test_helper.hxx"

#include <set>

#include "nmranet/NMRAnetWriteFlow.hxx"
#include "nmranet/AsyncIfCan.hxx"
#include "nmranet/AsyncAliasAllocator.hxx"
#include "os/OS.hxx"

namespace NMRAnet
{

Executor<1> g1_executor("g1_exec", 0, 2000);
Executor<1> g2_executor("g2_exec", 0, 2000);
Executor<1> g3_executor("g3_exec", 0, 2000);
Executor<1> g4_executor("g4_exec", 0, 2000);

Executor<1>* round_execs[] = {&g1_executor, &g2_executor,
                              &g3_executor, &g4_executor};

/** This class will create an AsyncIf, two virtual nodes on it, and send one
 * unaddressed global packet each. */
class TestNode
{
public:
    TestNode(NodeID node_id)
        : nodeId_(node_id),
          ifCan_(round_execs[(node_id >> 1) & 3], &can_hub0, 10, 10, 2)
    {
    }

    void start(BarrierNotifiable* done)
    {
        ifCan_.add_addressed_message_support();
        ifCan_.set_alias_allocator(new AsyncAliasAllocator(nodeId_, &ifCan_));
        {
            // Adds one alias buffer to the alias allocation flow.
            auto* b = ifCan_.alias_allocator()->alloc();
            ifCan_.alias_allocator()->send(b);
        }
        auto* b = ifCan_.global_message_write_flow()->alloc();
        b->data()->reset(If::MTI_EVENT_REPORT, nodeId_,
                         EventIdToBuffer(nodeId_));
        b->set_done(done->new_child());
        ifCan_.global_message_write_flow()->send(b);
        b = ifCan_.global_message_write_flow()->alloc();
        b->data()->reset(If::MTI_EVENT_REPORT, nodeId_ + 1,
                         EventIdToBuffer(nodeId_ + 1));
        b->set_done(done->new_child());
        ifCan_.global_message_write_flow()->send(b);
    }

    ~TestNode()
    {
        //ifCan_.alias_allocator()->TEST_finish_pending_allocation();
        Executor<1>* e = round_execs[(nodeId_ >> 1) & 3];
        while(!e->empty() || !g_executor.empty()
              || !ifCan_.alias_allocator()->is_waiting()
              || !ifCan_.dispatcher()->is_waiting()
              || !ifCan_.frame_dispatcher()->is_waiting()
              ) {
/*              !ifCan_.frame_dispatcher()->IsNotStarted() ||
              !ifCan_.dispatcher()->IsNotStarted() ||
              !can_hub0.empty() ||
              !gc_hub0.empty()) {*/
            usleep(100);
        }
    }

private:
    NodeID nodeId_;
    AsyncIfCan ifCan_;
//    AliasInfo testAlias_;
};

class Stats : public CanHubPortInterface
{
public:
    Stats()
        : timer_(&Stats::timer_callback, this, nullptr),
          numFrames_(0),
          badFrames_(0),
          cidFrames_(0),
          ridFrames_(0),
          amdFrames_(0),
          eventFrames_(0)
    {
{'fraction_non_alphanumeric': 0.1434173669467787, 'fraction_numerical': 0.022782446311858075, 'mean_word_length': 2.5802139037433154, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 65, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5610151', 'n_tokens_mistral': 1884, 'n_tokens_neox': 1784, 'n_words': 298}