Bka latency watchdog for event loop

src/ds/asio/latency-watchdog-test.cpp

+/*
+    Copyright (c) 2019 digitalSTROM AG, Zurich, Switzerland
+
+    This file is part of digitalSTROM Server.
+
+    digitalSTROM Server is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    digitalSTROM Server is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with digitalSTROM Server. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "latency-watchdog.h"
+#include <ds/asio/loop.h>
+#include <ds/asio/test.h>
+#include <ds/catch/catch.h>
+#include <ds/clock-mock.h>
+#include <ds/log.h>
+#include "ds/catch/trompeloeil.hpp"
+
+static ds::LogChannel dsLogChannel("dsAsioLatencyWatchdogTest");
+
+static const char* TAGS = "[dsAsioLatencyWatchdog][dsAsio][ds]";
+
+using namespace std::literals;
+
+struct Fire {
+public:
+  MAKE_MOCK0(call, void());
+};
+
+TEST_CASE("dsAsioLatencyWatchdog", TAGS) {
+  auto clock = ds::ClockMocker{};
+  auto watchedLoop = ds::asio::TestLoop{};
+  auto watchdogLoop = ds::asio::TestLoop{};
+
+  const auto latency = 1000ms;
+  auto latencyWatchdog = ds::asio::LatencyWatchdog{watchedLoop.io, latency, watchdogLoop.io};
+  auto fire = Fire{};
+  latencyWatchdog.setFire([&] { fire.call(); });
+
+  SECTION("can be created and destroyed") {}
+
+  SECTION("does not fire when watchedIo is executing handlers") {
+    for (auto i = 0; i < 30; i++) {
+      clock.advance(latency / 5);
+      watchedLoop.postStopRun();
+      watchdogLoop.postStopRun();
+    }
+
+    SECTION("does not fire when watchedIo does not execute handlers for latency.") {
+      for (auto i = 0; i < 5; i++) {
+        clock.advance(latency / 5);
+        watchdogLoop.postStopRun();
+      }
+
+      SECTION("fires when watchedIo does not execute handlers for 2*latency.") {
+        REQUIRE_CALL(fire, call());
+        for (auto i = 0; i < 10; i++) {
+          clock.advance(latency / 5);
+          watchdogLoop.postStopRun();
+        }
+      }
+    }
+  }
+}
+
+// real assertion. Not part of teste suite
+TEST_CASE("dsAsioLatencyWatchdog assert", "[!hide]") {
+  auto loop = ds::asio::Loop{};
+
+  const auto latency = 500ms;
+  auto latencyWatchdog = ds::asio::LatencyWatchdog{loop.io(), latency};
+
+  DS_NOTICE("Event loop running.");
+  std::this_thread::sleep_for(latency * 2);
+
+  DS_NOTICE("Event loop blocked.");
+  loop.io().post([&] { std::this_thread::sleep_for(latency * 5); });
+  while (1) {
+  }
+}

src/ds/asio/latency-watchdog.cpp

+/*
+    Copyright (c) 2019 digitalSTROM AG, Zurich, Switzerland
+
+    This file is part of digitalSTROM Server.
+
+    digitalSTROM Server is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    digitalSTROM Server is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with digitalSTROM Server. If not, see <http://www.gnu.org/licenses/>.
+*/
+#include "latency-watchdog.h"
+#include <ds/asio/timer.h>
+#include <ds/log.h>
+#include <ds/stack-trace.h>
+#include <atomic>
+#include <future>
+
+#include <execinfo.h>
+#include <pthread.h>
+#include <signal.h>
+#include <stdio.h>
+#include <stdlib.h>
+#include <iostream>
+
+static ds::LogChannel dsLogChannel("dsAsioLatencyWatchdog");
+
+using namespace std::literals;
+
+namespace ds::asio {
+
+class Loop {
+public:
+  ~Loop() {
+    m_io.stop();
+    m_thread.join();
+  }
+
+  boost::asio::io_service &io() { return m_io; }
+
+private:
+  boost::asio::io_service m_io;
+  boost::asio::io_service::work m_work{m_io};
+  std::thread m_thread{[this] {
+    for (;;) {
+      try {
+        m_io.run();
+        break; // run() exited normally
+      } catch (std::exception &e) {
+        DS_ASSERT("Unhandled exception", e);
+      }
+    }
+  }};
+};
+
+static void assertSignalHandler(int) {
+  DS_FAIL_ASSERT("This thread stalled.");
+}
+
+static void assertWithStackTrace(pthread_t threadId) {
+  const auto signal = SIGUSR2;
+
+  // setup signal handler
+  struct sigaction sa {};
+  struct sigaction oldSa {};
+  sigfillset(&sa.sa_mask);
+  sa.sa_handler = assertSignalHandler;
+  sigaction(signal, &sa, &oldSa);
+
+  // invoke signal handler in target thread
+  pthread_kill(threadId, signal);
+}
+
+struct LatencyWatchdog::Impl {
+  boost::asio::io_service &m_watchedIo;
+  std::unique_ptr<Loop> m_watchogLoop;
+  boost::asio::io_service &m_watchogIo;
+  ds::Clock::duration m_latency;
+  ds::asio::Timer m_timer{m_watchogIo};
+  ds::Clock::time_point m_pingedAt{ds::Clock::now()};
+  std::atomic_bool m_ponged{true};
+  std::function<void()> m_fire{[this] { defaultFire(); }};
+  pthread_t m_watchedThreadId{};
+
+  Impl(boost::asio::io_service &watchedIo, ds::Clock::duration latency, boost::asio::io_service *watchdogIo)
+      : m_watchedIo{watchedIo},
+        m_watchogLoop{watchdogIo ? nullptr : std::make_unique<Loop>()},
+        m_watchogIo{watchdogIo ? *watchdogIo : m_watchogLoop->io()},
+        m_latency{latency} {
+    DS_INFO_ENTER(m_latency);
+    asyncPingLoop();
+  }
+
+  ~Impl() {
+    DS_INFO_ENTER("");
+  }
+
+  void asyncPingLoop() {
+    DS_INFO_ENTER(ds::Clock::now());
+    m_timer.randomlyExpiresFromNow(m_latency, [this] {
+      if (!m_ponged) {
+        DS_NOTICE("Fire", ds::Clock::now() - m_pingedAt);
+        m_fire();
+      }
+      m_ponged = false;
+      m_pingedAt = ds::Clock::now();
+      DS_INFO("Ping", m_pingedAt);
+      m_watchedIo.post([this]() {
+        DS_INFO("Pong", ds::Clock::now());
+        m_ponged = true;
+        m_watchedThreadId = pthread_self();
+      });
+      asyncPingLoop();
+    });
+  }
+
+  void defaultFire() {
+    DS_ERROR("LatencyWatchdog failed.", ds::Clock::now() - m_pingedAt, m_watchedThreadId);
+    if (m_watchedThreadId) {
+      assertWithStackTrace(m_watchedThreadId);
+      std::this_thread::sleep_for(2s);
+    }
+    // we shall not get here, the signal hander shall assert
+    DS_FAIL_ASSERT("LatencyWatchdog failed without stalled stacktrace");
+  }
+};
+
+LatencyWatchdog::LatencyWatchdog(boost::asio::io_service &watchedIo, ds::Clock::duration latency)
+    : m_impl(std::make_unique<Impl>(watchedIo, latency, nullptr)) {}
+
+LatencyWatchdog::LatencyWatchdog(boost::asio::io_service &watchedIo, ds::Clock::duration latency,
+                                 boost::asio::io_service &watchdogIo)
+    : m_impl(std::make_unique<Impl>(watchedIo, latency, &watchdogIo)) {}
+LatencyWatchdog::~LatencyWatchdog() = default;
+
+void LatencyWatchdog::setFire(std::function<void()> &&x) {
+  m_impl->m_fire = std::move(x);
+}
+
+} // namespace ds::asio

src/ds/asio/latency-watchdog.h

+/*
+    Copyright (c) 2019 digitalSTROM AG, Zurich, Switzerland
+
+    This file is part of digitalSTROM Server.
+
+    digitalSTROM Server is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    digitalSTROM Server is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with digitalSTROM Server. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#pragma once
+
+#include <ds/clock.h>
+#include <boost/asio/io_service.hpp>
+#include <functional>
+#include <memory>
+
+namespace ds::asio {
+
+/// Software watchdog that guards that io event loop is not stalled.
+class LatencyWatchdog {
+public:
+  /// Spawns internal watchog thread and regurarly pings `watchedIo` from it.
+  LatencyWatchdog(boost::asio::io_service &watchedIo, ds::Clock::duration latency);
+
+  /// Regurarly pings `watchedIo` from `watchdogIo`
+  LatencyWatchdog(boost::asio::io_service &io, ds::Clock::duration latency, boost::asio::io_service &watchdogIo);
+  ~LatencyWatchdog();
+
+  /// Set function called when watchog times out. Default function asserts
+  ///
+  /// WARNING: callback is called from watchdog thread.
+  void setFire(std::function<void()> &&);
+
+private:
+  struct Impl;
+  const std::unique_ptr<Impl> m_impl;
+};
+
+} // namespace ds::asio

src/ds/asio/loop.cpp

+/*
+    Copyright (c) 2019 digitalSTROM.org, Zurich, Switzerland
+
+    This file is part of digitalSTROM Server.
+
+    digitalSTROM Server is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    digitalSTROM Server is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with digitalSTROM Server. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include "loop.h"
+#include <ds/log.h>
+
+namespace ds::asio {
+
+Loop::Loop()
+    : m_thread{[this] {
+        for (;;) {
+          try {
+            m_io.run();
+            break; // run() exited normally
+          } catch (std::exception &e) {
+            DS_ASSERT("Unhandled exception", e);
+          }
+        }
+      }} {}
+
+Loop::~Loop() {
+  m_io.stop();
+  m_thread.join();
+}
+
+} // namespace ds::asio

src/ds/asio/loop.h

+/*
+    Copyright (c) 2019 digitalSTROM.org, Zurich, Switzerland
+
+    This file is part of digitalSTROM Server.
+
+    digitalSTROM Server is free software: you can redistribute it and/or modify
+    it under the terms of the GNU General Public License as published by
+    the Free Software Foundation, either version 3 of the License, or
+    (at your option) any later version.
+
+    digitalSTROM Server is distributed in the hope that it will be useful,
+    but WITHOUT ANY WARRANTY; without even the implied warranty of
+    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+    GNU General Public License for more details.
+
+    You should have received a copy of the GNU General Public License
+    along with digitalSTROM Server. If not, see <http://www.gnu.org/licenses/>.
+*/
+
+#include <boost/asio/io_service.hpp>
+#include <thread>
+
+namespace ds::asio {
+
+//
+// Event loop implemneted as ds::asio::io_service running in one thread.
+//
+class Loop {
+public:
+  Loop();
+  ~Loop();
+
+  boost::asio::io_service &io() { return m_io; }
+
+private:
+  boost::asio::io_service m_io;
+  boost::asio::io_service::work m_work{m_io};
+  std::thread m_thread;
+};
+
+} // namespace ds::asio