SObjectizer Tales – 15. Monitoring disconnections

Welcome back for a new SObjectizer adventure! In this iteration, we will detect client disconnections without attempting to send any data to them.

gRPC::ServerContext provides a function IsCancelled() that returns false if the corresponding RPC failed due to several possible reasons, including client-side cancellations. Since this function has to be called explicitly, we can set up a periodic message to perform this check at regular intervals.

This approach involves periodically calling IsCancelled() from each subscribe_client_agent. In essence, when the agent starts, it sends a periodic message to itself (say, every 5 seconds) to check if the context has been cancelled at regular intervals. Integrating this change into the current implementation is straightforward and just needs a few changes:

class subscribe_client_agent : public agent_t
{
	struct check_client_disconnection final : signal_t{};		
public:
	subscribe_client_agent(agent_context_t ctx, grpc::ServerContext& context, grpc::ServerWriter<subscribe_response>& writer, std::vector<mbox_t> channels)
	: agent_t(std::move(ctx)), m_context(context), m_writer(writer), m_channels(std::move(channels))
	{
	}

	void so_evt_start() override
	{
		m_check_disconnection_timer = send_periodic<check_client_disconnection>(so_direct_mbox(), 5s, 5s);
	}

	void so_define_agent() override
	{
		for (const auto& channel : m_channels)
		{
			// ... as before
		}
		
		so_subscribe_self().event([this](mhood_t<check_client_disconnection>) {
			if (m_context.IsCancelled())
			{
				so_deactivate_agent();
				so_deregister_agent_coop_normally();
			}
		});
	}
		
private:
// ... as before
	grpc::ServerContext& m_context;
	timer_id_t m_check_disconnection_timer;
};

A small refactoring consists in encapsulating the “deactivation logic” into a private function. Also, we are required to pass the reference to grcp::ServerContext from service_impl::subscribe to the agent. You will find these two additions pushed into the repository.

SObjectizer’s timers are designed to support a vast amount of timers efficiently (hundreds of millions or billions) so we don’t have concerns about performance issues.

The only comment on this approach regards scaling the system that might become more complex if we require the capability to regulate the number of cooperation deregistrations triggered. For instance, let’s assume we aim to restrict the maximum number of cooperations shutdown to a specific limit, such as “no more than 10 every 500 milliseconds.” In the existing structure, clients operate independently without direct communication. Consequently, coordinating to establish this throttling limit becomes challenging. To solve this problem, we might introduce another agent that performs the actual deregistration. This centralized helper could be responsible for throttling the number of deregistration in a certain time span.

Our current scenario isn’t overly complicated, so we’ll stick with the simple approach outlined in this brief article. Nevertheless, it’s essential to note that things could potentially become more complex under different circumstances.

Takeaway

In this episode we have learned:

• grpc::ServerContext provides IsCancelled() to determine if the current connection has been cancelled on the client side;
• again, periodic messages show their simplicity but also effectiveness to handle this kind of situation.

As usual, calico is updated and tagged.

What’s next?

We can finally release a new version of calico, allowing Giulia to work on her program. In the meantime, we receive an email from our colleague Gerri whom we discover is also a user of our program – in a small company like ours, news travels fast. He is raising a new issue: occasionally, when the program is requested to terminate, the actual shutdown takes too long…

What’s going on? In the next post we’ll start exploring approaches to interrupt pending work!

---

Thanks to Yauheni Akhotnikau for having reviewed this post.