Strange behavior of so_set_delivery_state

[Bobrov88]

I use So_5 in my project with tests.
Please, explain me, why.

Here my Agent:

void Agent::so_define_agent()
{
    so_subscribe_self()
        .event(&Agent::onDlStarted)
        .event(&Agent::onUlStarted)
        .event(&Agent::onDlStopped)
        .event(&Agent::onUlStopped);

// bool m_isUlStartedAllowed = true;
// bool m_isDlStartedAllowed= true;

    so_set_delivery_filter(so_direct_mbox(), [this](const UlStarted&) { return m_isUlStartedAllowed; });
    so_set_delivery_filter(so_direct_mbox(), [this](const DlStarted&) { return m_isDlStartedAllowed; });
}

void Agent::onDlStarted(so_5::mhood_t<DlStarted> msg)
{
    // some actions
    m_isDlStartedAllowed = false;
    std::cout << " !!!!!!!!!!!!!!!!!!!!!!!      onDl"; // for tests
}

void Agent::onlStarted(so_5::mhood_t<UlStarted> msg)
{
   // some actions
    m_isUlStartedAllowed = false;
    std::cout << " !!!!!!!!!!!!!!!!!!!!!!!      onUl";  //for tests
}

void Agent::onDlStopped(so_5::mhood_t<DlStopped> cmd)
{
   // some actions
    m_isDlStartedAllowed = true;
    std::cout << " !!!!!!!!!!!!!!!!!!!!!!!      onDl STOP";  // for tests
}

void Agent::onUlStopped(so_5::mhood_t<UlStopped> cmd)
{
    m_isUlStartedAllowed = true;
    std::cout << " !!!!!!!!!!!!!!!!!!!!!!!      onUl STOP";  // for tests
}

Look at the test:

    void testUlStartedAndStopped()
    {
        suppressWarnings();

// first UlStarted send - flag --> false
        m_testEnv.scenario()
            .define_step("::UlStarted #1")
            .impact<UlStarted>(*m_agent)
            .when(*m_agent & tests::reacts_to<UlStarted>());

//  send - flag --> true
        m_testEnv.scenario()
            .define_step("::UlStopped")
            .impact<UlStopped>(*m_agent)
            .when(*m_agent & tests::reacts_to<UlStopped>());

// second UlStarted send - flag --> false
        m_testEnv.scenario()
            .define_step("::UlStarted #2")
            .impact<UlStarted>(*m_agent)
            .when(*m_agent & tests::reacts_to<PhyUlStarted>());

// third UlStarted send - delivery filter must work
        m_testEnv.scenario()
            .define_step("::UlStarted #3")
            .impact<UlStarted>(*m_agent)
            .when(*m_agent & tests::reacts_to<UlStarted>());

        m_testEnv.scenario().run_for(200ms);
        m_testEnv.stop_then_join();

        auto result = m_testEnv.scenario().result();

        EXPECT_NE(tests::completed(), result);

        std::stringstream ss;
        ss << result;
        std::string scheme = ss.str();

        EXPECT_EQ(scheme.find("::UlStarted #1"), std::string::npos);
        EXPECT_EQ(scheme.find("::UlStopped"), std::string::npos);
        EXPECT_EQ(scheme.find("::UlStarted #2"), std::string::npos);
        EXPECT_NE(scheme.find("::UlStarted #3"), std::string::npos);
    }

Look at test result:

[       OK ]AgentTests.TestDlStartedAndStopped (3 ms)
[ RUN      ] AgentTests.TestUlStartedAndStopped
 !!!!!!!!!!!!!!!!!!!!!!!      onUl                           <----------------------------
 !!!!!!!!!!!!!!!!!!!!!!!      onUl STOP.                <----------------------------

    // WHERE IS THE REST?? FLAG IS TRUE. WHY IS NOT HANDLED?

./AgentTests.cpp:460: Failure
Expected equality of these values:
  scheme.find("::UlStarted #2")
    Which is: 30
  std::string::npos
    Which is: 18446744073709551615

../AgentTests.cpp:461: Failure

I used to try with drop_subscription/so_subsribe_self as analog of flag bool/true, but it doesn't work.
Did you get the point? I can explain if you don't
Thanx!

[eao197]

Hi!

First of all you have a potentional data race in your code: checking of a boolean flag in a delivery filter is not synchronized with changing of this boolean flag in an event handler. The delivery filters are called inside send operation on the thread of the sender and this may be a different thread than agent's thread for agent's event handlers. Because of that delivery filters have to be thread safe. It's not true in your case.

The strange behaviour you have encounter in the test related not to delivery filters, but to the testing environment and the way testing scenario steps are executed.

There is a small test program I've used to reproduce your case:

#include <so_5/experimental/testing/v1/all.hpp>

#include <test/3rd_party/various_helpers/time_limited_execution.hpp>
#include <test/3rd_party/various_helpers/ensure.hpp>

#include <test/3rd_party/utest_helper/helper.hpp>

namespace tests = so_5::experimental::testing::v1;

struct ul_started final : public so_5::message_t {};
struct ul_stopped final : public so_5::message_t {};

class test_agent_t final : public so_5::agent_t
{
	bool m_ul_allowed{ true };

public :

	test_agent_t( context_t ctx )
		:	so_5::agent_t( std::move(ctx) )
	{}

	void
	so_define_agent()
	{
		so_subscribe_self()
			.event( &test_agent_t::evt_ul_started )
			.event( &test_agent_t::evt_ul_stopped )
			;

		so_set_delivery_filter( so_direct_mbox(),
				[this]( const ul_started & ) {
					std::cout << "delivery_filter invoked" << std::endl;
					return this->m_ul_allowed;
				} );
	}

private :
	void
	evt_ul_started( mhood_t<ul_started> )
	{
		std::cout << "ul_started" << std::endl;
		m_ul_allowed = false;
	}

	void
	evt_ul_stopped( mhood_t<ul_stopped> )
	{
		std::cout << "ul_stopped" << std::endl;
		m_ul_allowed = true;
	}
};

UT_UNIT_TEST( issue_108 )
{
	run_with_time_limit(
		[]()
		{
			tests::testing_env_t env{
				 [](so_5::environment_params_t & params) {
					// Turn message delivery tracing on.
					params.message_delivery_tracer(
						so_5::msg_tracing::std_cout_tracer());
				 } };

			so_5::agent_t * test_agent = env.environment().introduce_coop(
					[](so_5::coop_t & coop) {
						return coop.make_agent< test_agent_t >();
					} );

			env.scenario().define_step( "ul_started #1" )
				.impact< ul_started >( *test_agent )
				.when( *test_agent & tests::reacts_to< ul_started >() );

			env.scenario().define_step( "ul_stopped #1" )
				.impact< ul_stopped >( *test_agent )
				.when( *test_agent & tests::reacts_to< ul_stopped >() );

			env.scenario().define_step( "ul_started #2" )
				.impact< ul_started >( *test_agent )
				.when( *test_agent & tests::reacts_to< ul_started >() );

			env.scenario().define_step( "ul_started #3" )
				.impact< ul_started >( *test_agent )
				.when( *test_agent & tests::reacts_to< ul_started >() );

			env.scenario().run_for( std::chrono::seconds(1) );

			UT_CHECK_EQ( tests::completed(), env.scenario().result() );
		},
		5 );
}

int
main()
{
	UT_RUN_UNIT_TEST( issue_108 )

	return 0;
}

And there is the result of the execution with message delivery tracing enabled:

delivery_filter invoked
[tid=132962448938816][mbox_id=3] deliver_message.push_to_queue [msg_type=10ul_started][delivery_mode=ordinary][envelope_ptr=0x5ce592d0a140][redirection_deep=1][msg_sink_ptr=0x5ce592d09d98][agent_ptr=0x5ce592d09970]
[tid=132962425231040][agent_ptr=0x5ce592d09970] demand_handler_on_enveloped_msg.find_handler [mbox_id=3][msg_type=10ul_started][envelope_ptr=0x5ce592d0ae30][state=<DEFAULT>][evt_handler=0x5ce592d09b60]
[tid=132962425231040][mbox_id=3] deliver_message.push_to_queue [msg_type=10ul_stopped][delivery_mode=ordinary][envelope_ptr=0x5ce592d0a120][redirection_deep=1][msg_sink_ptr=0x5ce592d09d98][agent_ptr=0x5ce592d09970]
ul_started
[tid=132962425231040][agent_ptr=0x5ce592d09970] demand_handler_on_enveloped_msg.find_handler [mbox_id=3][msg_type=10ul_stopped][envelope_ptr=0x78edac0010a0][state=<DEFAULT>][evt_handler=0x5ce592d09ba8]
delivery_filter invoked
[tid=132962425231040][mbox_id=3] deliver_message.message_rejected [msg_type=10ul_started][delivery_mode=ordinary][envelope_ptr=0x5ce592d0a2e0][redirection_deep=1][msg_sink_ptr=0x0]
ul_stopped
/home/eao197/sandboxes/sobjectizer.github/dev/test/so_5/testing/v1/issue_108/main.cpp(89): unit test error: comparison failed: 'tests::completed() == env.scenario().result()' where
  [
    tests::completed() is [completed],
    env.scenario().result() is [timed_out,{preactivated step:ul_started #2;}]
  ]
Unit test 'issue_108' failed

You could see the first invocation of the delivery fliter and then the delivery of the first ul_started instance:

delivery_filter invoked
[tid=132962448938816][mbox_id=3] deliver_message.push_to_queue [msg_type=10ul_started][delivery_mode=ordinary][envelope_ptr=0x5ce592d0a140][redirection_deep=1][msg_sink_ptr=0x5ce592d09d98][agent_ptr=0x5ce592d09970]
[tid=132962425231040][agent_ptr=0x5ce592d09970] demand_handler_on_enveloped_msg.find_handler [mbox_id=3][msg_type=10ul_started][envelope_ptr=0x5ce592d0ae30][state=<DEFAULT>][evt_handler=0x5ce592d09b60]

Then you could see sending of ul_stopped message:

[tid=132962425231040][mbox_id=3] deliver_message.push_to_queue [msg_type=10ul_stopped][delivery_mode=ordinary][envelope_ptr=0x5ce592d0a120][redirection_deep=1][msg_sink_ptr=0x5ce592d09d98][agent_ptr=0x5ce592d09970]

But it happens before the execution of the event handler for ul_started.

Then the event_handler for the ul_started executed:

ul_started

Then SO-5 searches a handler for ul_stopped, but the handler itself is not executed yet:

[tid=132962425231040][agent_ptr=0x5ce592d09970] demand_handler_on_enveloped_msg.find_handler [mbox_id=3][msg_type=10ul_stopped][envelope_ptr=0x78edac0010a0][state=<DEFAULT>][evt_handler=0x5ce592d09ba8]
delivery_filter invoked

But just in the same moment the next ul_started is sent:

delivery_filter invoked
[tid=132962425231040][mbox_id=3] deliver_message.message_rejected [msg_type=10ul_started][delivery_mode=ordinary][envelope_ptr=0x5ce592d0a2e0][redirection_deep=1][msg_sink_ptr=0x0]

And it's rejected because the event handler for ul_stopped isn't called yet.

This happens because of this reason: when SO-5 finds an event handler for ul_stopped it assumes that the current step of the scenario is activated and initiates preactivation for the next scenario step. The preactivation includes of impact section and sending ul_started message. This happens just before execution of the event handler for the previous scenario step.

[eao197]

As a workaround you could use impact that receives a lambda and send_delayed. Something like that:

			env.scenario().define_step( "ul_started #1" )
				.impact< ul_started >( *test_agent )
				.when( *test_agent & tests::reacts_to< ul_started >() );

			env.scenario().define_step( "ul_stopped #1" )
				.impact< ul_stopped >( *test_agent )
				.when( *test_agent & tests::reacts_to< ul_stopped >() );

			env.scenario().define_step( "ul_started #2" )
				.impact( [d = test_agent->so_direct_mbox()]()
					{
						so_5::send_delayed< ul_started >( d, 100ms );
					})
				.when( *test_agent & tests::reacts_to< ul_started >() );

			env.scenario().define_step( "ul_started #3" )
				.impact( [d = test_agent->so_direct_mbox()]()
					{
						so_5::send_delayed< ul_started >( d, 100ms );
					})
				.when( *test_agent & tests::ignores< ul_started >() );