From 1cea84f4da4b2d839e90e4bb961ce31e23396dde Mon Sep 17 00:00:00 2001
From: Robin Sommer <robin@corelight.com>
Date: Mon, 21 Feb 2022 13:31:02 +0100
Subject: [PATCH 1/7] Push all asynchronous activity back to the main thread.

The scheduler now accepts, in addition timers, also "tasks" for
immediate execution at the next opportunity. If any of our threads
needs to do work, we now schedule that back to the scheduler as such a
task. It will then be processed as part of the main loop. This means
that thread-safety becomes much less of an issue, and hence require
less locking. We'll clean lock up in a subsequent thread.
---
 src/core/scheduler.cc | 46 +++++++++++++++++++++++++++++++++++++++++--
 src/core/scheduler.h  | 22 +++++++++++++++++++++
 src/io/console.cc     | 28 ++++++++------------------
 src/io/zeek.cc        | 38 ++++++++++++++++++-----------------
 src/main.cc           | 19 +-----------------
 zeek-agent            |  2 +-
 6 files changed, 96 insertions(+), 59 deletions(-)

diff --git a/src/core/scheduler.cc b/src/core/scheduler.cc
index dda15804..c1de8de1 100644
--- a/src/core/scheduler.cc
+++ b/src/core/scheduler.cc
@@ -7,6 +7,7 @@
 #include "util/testing.h"
 #include "util/threading.h"
 
+#include <algorithm>
 #include <queue>
 #include <unordered_map>
 #include <utility>
@@ -40,6 +41,9 @@ struct Pimpl<Scheduler>::Implementation {
     // Runs all update hooks.
     void updated();
 
+    // Executes scheduled activity up to current wall clock.
+    bool loop();
+
     SynchronizedBase* _synchronized =
         nullptr;            // scheduler's synchronizer, so that we can release it for callback execution
     timer::ID _next_id = 1; // counter for creating timer IDs
@@ -48,6 +52,7 @@ struct Pimpl<Scheduler>::Implementation {
     std::priority_queue<Timer*, std::vector<Timer*>, TimerComparator> _timers; // timers sorted by time
     std::vector<std::function<void()>> _update_callbacks;                      // registered update callbacks
     std::atomic<bool> _terminating; // true once termination has been requested; ok to access wo/ lock
+    ConditionVariable interrupt_loop;
 };
 
 timer::ID Scheduler::Implementation::schedule(timer::ID id, Time t, timer::Callback cb) {
@@ -90,11 +95,31 @@ bool Scheduler::Implementation::advance(Time now) {
     return true;
 }
 
+bool Scheduler::Implementation::loop() {
+    Interval timeout = 0s;
+
+    if ( _timers.empty() )
+        timeout = 15s; // TODO: Make max timoeut configurable
+    else
+        timeout = std::max(Interval(0s), _timers.top()->due - std::chrono::system_clock::now());
+
+    if ( timeout > 0s ) {
+        ZEEK_AGENT_DEBUG("scheduler", "sleeping with timeout={}", to_string(timeout));
+        _synchronized->unlockWhile([&]() { interrupt_loop.wait(timeout); });
+    }
+
+    advance(std::chrono::system_clock::now());
+    return _terminating;
+}
+
 void Scheduler::Implementation::updated() {
     _synchronized->unlockWhile([&]() {
         for ( const auto& cb : _update_callbacks )
             cb();
     });
+
+    ZEEK_AGENT_DEBUG("scheduler", "updated");
+    interrupt_loop.notify();
 }
 
 Scheduler::Scheduler() {
@@ -112,6 +137,17 @@ timer::ID Scheduler::schedule(Time t, timer::Callback cb) {
     return id;
 }
 
+void Scheduler::schedule(task::Callback cb) {
+    Synchronize _(this);
+    auto id = pimpl()->schedule(pimpl()->_next_id++, currentTime(), [cb = std::move(cb)](timer::ID) -> Interval {
+        cb();
+        return Interval(0);
+    });
+
+    ZEEK_AGENT_DEBUG("scheduler", "scheduled task {} for immediate execution", id);
+    pimpl()->updated();
+}
+
 void Scheduler::cancel(timer::ID id) {
     Synchronize _(this);
     ZEEK_AGENT_DEBUG("scheduler", "canceling timer {}", id);
@@ -119,6 +155,12 @@ void Scheduler::cancel(timer::ID id) {
     pimpl()->updated();
 }
 
+bool Scheduler::loop() {
+    Synchronize _(this);
+    ZEEK_AGENT_DEBUG("scheduler", "executing pending activity");
+    return pimpl()->loop();
+}
+
 void Scheduler::advance(Time t) {
     Synchronize _(this);
     ZEEK_AGENT_DEBUG("scheduler", "advancing time to t={}", to_string(t));
@@ -139,12 +181,12 @@ void Scheduler::terminate() {
 }
 
 bool Scheduler::terminating() const {
-    Synchronize _(this);
+    // Synchronize _(this);
     return pimpl()->_terminating;
 }
 
 Time Scheduler::currentTime() const {
-    Synchronize _(this);
+    // Synchronize _(this);
     return pimpl()->_now;
 }
 
diff --git a/src/core/scheduler.h b/src/core/scheduler.h
index 91c6b85a..181cd435 100644
--- a/src/core/scheduler.h
+++ b/src/core/scheduler.h
@@ -21,6 +21,11 @@ using ID = uint64_t;
 using Callback = std::function<Interval(ID)>;
 } // namespace timer
 
+namespace task {
+/** Callback for an operation queued for execution. */
+using Callback = std::function<void()>;
+} // namespace task
+
 /**
  * Manages a set of scheduled timers with associated callbacks to eventually
  * execute. The scheduler has an internal notion of time and will execute the
@@ -51,6 +56,13 @@ class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
      */
     timer::ID schedule(Time t, timer::Callback cb);
 
+    /**
+     * Enqueues a callback to execute at the next possible opportunity fro the scheduler's thread.
+
+     * @param cb callback to execute as soon as possible from the scheduler's thread.
+     */
+    void schedule(task::Callback cb);
+
     /**
      * Cancels a previously installed timer. The timer will be deleted without
      * its callback executing.
@@ -60,6 +72,16 @@ class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
      */
     void cancel(timer::ID id);
 
+    /**
+     * Executes pending activity up to current wall clock. If nothing is
+     * pending, blocks for a  little while (with new activity interrupting the
+     * block).
+     *
+     * @return true if processing is to continue; false if the scheduler has
+     * been asked to terminate
+     */
+    bool loop();
+
     /**
      * Advances to scheduler's notion of the current time. This will let all
      * timers fire that are currently scheduled for a time <= `now`. Their
diff --git a/src/io/console.cc b/src/io/console.cc
index fc6fd158..7d09e51a 100644
--- a/src/io/console.cc
+++ b/src/io/console.cc
@@ -34,9 +34,6 @@ struct Pimpl<Console>::Implementation {
     // Performance a query against the database.
     void query(const std::string& stmt, std::optional<query::SubscriptionType> subscription, bool terminate = false);
 
-    // Cancels the current query.
-    void cancelQuery();
-
     // Prints a message to the console.
     void message(const std::string& msg);
 
@@ -70,11 +67,6 @@ struct Pimpl<Console>::Implementation {
     replxx::Replxx _rx;                   // instance of the REPL
 };
 
-void Console::Implementation::cancelQuery() {
-    // Note we can come here from a signal handler.
-    _query_done.notify();
-}
-
 void Console::Implementation::execute(const std::string& cmd, bool terminate) {
     ZEEK_AGENT_DEBUG("console", "executing: {}", cmd);
 
@@ -234,20 +226,16 @@ void Console::Implementation::query(const std::string& stmt, std::optional<query
     std::unique_ptr<signal::Handler> sigint_handler;
     if ( ! terminate )
         // Temporarily install our our SIGINT handler while the query is running.
-        sigint_handler = std::make_unique<signal::Handler>(_signal_mgr, SIGINT, [this]() {
-            cancelQuery();
-            std::cout << std::endl;
-        });
+        sigint_handler = std::make_unique<signal::Handler>(_signal_mgr, SIGINT, [this]() { _query_done.notify(); });
 
     _query_done.reset();
-    if ( auto id = _db->query(query) ) {
-        _query_done.wait();
 
-        if ( *id )
-            _db->cancel(**id);
-    }
-    else
-        error(id.error());
+    _scheduler->schedule([this, query]() {
+        if ( auto id = _db->query(query); ! id )
+            error(id.error());
+    });
+
+    _query_done.wait();
 }
 
 void Console::Implementation::message(const std::string& msg) { _rx.print("%s\n", msg.c_str()); }
@@ -315,7 +303,7 @@ void Console::stop() {
 
     if ( pimpl()->_thread ) {
         ZEEK_AGENT_DEBUG("console", "stopping");
-        pimpl()->cancelQuery();
+        pimpl()->_query_done.notify();
         pimpl()->_thread->join();
     }
 }
diff --git a/src/io/zeek.cc b/src/io/zeek.cc
index 24156b0d..28131e47 100644
--- a/src/io/zeek.cc
+++ b/src/io/zeek.cc
@@ -177,24 +177,26 @@ Result<Nothing> BrokerConnection::connect(const std::string& destination) {
     _endpoint.subscribe_nosync(
         topics, []() { /* nop */ },
         [this](const broker::data_message& msg) {
-            if ( broker::get_topic(msg) == broker::topic::statuses_str ) {
-                Synchronize _(this);
-                auto x = broker::to<broker::status>(broker::get_data(msg));
-                assert(x); // NOLINT(bugprone-lambda-function-name)
-                processStatus(*x);
-            }
-            else if ( broker::get_topic(msg) == broker::topic::errors_str ) {
-                Synchronize _(this);
-                auto x = broker::to<broker::error>(broker::get_data(msg));
-                assert(x); // NOLINT(bugprone-lambda-function-name)
-                processError(*x);
-            }
-            else {
-                Synchronize _(this);
-                processEvent(msg);
-            }
-        },                                       //
-        [](const broker::error&) { /* nop */ }); //
+            _scheduler->schedule([this, msg]() { // process message on the main thread
+                if ( broker::get_topic(msg) == broker::topic::statuses_str ) {
+                    Synchronize _(this);
+                    auto x = broker::to<broker::status>(broker::get_data(msg));
+                    assert(x); // NOLINT(bugprone-lambda-function-name)
+                    processStatus(*x);
+                }
+                else if ( broker::get_topic(msg) == broker::topic::errors_str ) {
+                    Synchronize _(this);
+                    auto x = broker::to<broker::error>(broker::get_data(msg));
+                    assert(x); // NOLINT(bugprone-lambda-function-name)
+                    processError(*x);
+                }
+                else {
+                    Synchronize _(this);
+                    processEvent(msg);
+                }
+            });
+        },
+        [](const broker::error&) { /* nop */ });
 
     _subscriber = _endpoint.make_subscriber(topics);
     _status_subscriber = _endpoint.make_status_subscriber(true);
diff --git a/src/main.cc b/src/main.cc
index b658d594..ecc1f967 100644
--- a/src/main.cc
+++ b/src/main.cc
@@ -41,7 +41,6 @@ int main(int argc, char** argv) {
         if ( geteuid() != 0 && ! cfg.options().use_mock_data )
             logger()->warn("not running as root, information may be incomplete");
 
-        ConditionVariable main_loop;
         Scheduler scheduler;
         SignalManager signal_mgr({SIGINT});
         signal::Handler sigint(&signal_mgr, SIGINT, [&]() { scheduler.terminate(); });
@@ -68,29 +67,13 @@ int main(int argc, char** argv) {
 
         ZEEK_AGENT_DEBUG("main", "looping until terminated");
 
-        scheduler.registerUpdateCallback([&main_loop]() { main_loop.notify(); });
-
-        while ( ! scheduler.terminating() ) {
+        while ( ! scheduler.loop() ) {
             db.poll();
 
             if ( zeek )
                 zeek->poll();
 
-            if ( auto next_timer = scheduler.nextTimer(); next_timer == 0_time ) {
-                // TODO: make timeout configurable
-                auto t = Interval(15s);
-                ZEEK_AGENT_DEBUG("main", "completely idle, sleeping with timeout={}", to_string(t));
-                main_loop.wait(t);
-            }
-
-            else if ( auto t = next_timer - std::chrono::system_clock::now(); t > 0s ) {
-                ZEEK_AGENT_DEBUG("main", "sleeping with timeout={}", to_string(t));
-                main_loop.wait(t);
-            }
-
-            scheduler.advance(std::chrono::system_clock::now());
             db.expire();
-            main_loop.reset(); // clear any updates that were just flagged
         }
 
         return 0;
diff --git a/zeek-agent b/zeek-agent
index becd56f4..0caff56e 160000
--- a/zeek-agent
+++ b/zeek-agent
@@ -1 +1 @@
-Subproject commit becd56f4e0168eb033304c289db5f8024e874a4a
+Subproject commit 0caff56e17928fa9e6c452737db41bffac0ac030

From dac6a52963b20fce12189294412aaa5543b77402 Mon Sep 17 00:00:00 2001
From: Robin Sommer <robin@corelight.com>
Date: Mon, 21 Feb 2022 13:32:38 +0100
Subject: [PATCH 2/7] Remove locking from most classes.

They are no longer accessed from different threads.
---
 src/core/configuration.cc |  9 +--------
 src/core/configuration.h  |  4 ++--
 src/core/database.cc      | 41 +++++++++------------------------------
 src/core/database.h       |  5 ++---
 src/core/signal.cc        |  1 -
 src/core/sqlite.cc        |  4 ----
 src/core/sqlite.h         |  3 +--
 src/io/console.cc         | 38 +++++++++++++++++++++++-------------
 src/io/console.h          |  3 +--
 src/io/zeek.cc            | 16 +--------------
 src/io/zeek.h             |  3 +--
 src/main.cc               |  1 -
 zeek-agent                |  2 +-
 13 files changed, 43 insertions(+), 87 deletions(-)

diff --git a/src/core/configuration.cc b/src/core/configuration.cc
index 2375f0e2..f61c1d18 100644
--- a/src/core/configuration.cc
+++ b/src/core/configuration.cc
@@ -395,14 +395,9 @@ Configuration::~Configuration() {
     logger()->set_level(pimpl()->old_log_level);
 }
 
-const Options& Configuration::options() const {
-    Synchronize _(this);
-    return pimpl()->_options;
-}
+const Options& Configuration::options() const { return pimpl()->_options; }
 
 Result<Nothing> Configuration::initFromArgv(int argc, const char* const* argv) {
-    Synchronize _(this);
-
     std::vector<std::string> vargv;
     vargv.reserve(argc);
     for ( auto i = 0; i < argc; i++ )
@@ -413,13 +408,11 @@ Result<Nothing> Configuration::initFromArgv(int argc, const char* const* argv) {
 }
 
 Result<Nothing> Configuration::read(const filesystem::path& path) {
-    Synchronize _(this);
     ZEEK_AGENT_DEBUG("configuration", "reading file {}", path.native());
     return pimpl()->read(path);
 }
 
 Result<Nothing> Configuration::read(std::istream& in, const filesystem::path& path) {
-    Synchronize _(this);
     ZEEK_AGENT_DEBUG("configuration", "reading stream associated with file {}", path.native());
     return pimpl()->read(in, path);
 }
diff --git a/src/core/configuration.h b/src/core/configuration.h
index be0a71eb..f6560a43 100644
--- a/src/core/configuration.h
+++ b/src/core/configuration.h
@@ -3,9 +3,9 @@
 #pragma once
 
 #include "util/filesystem.h"
+#include "util/helpers.h"
 #include "util/pimpl.h"
 #include "util/result.h"
-#include "util/threading.h"
 
 #include <optional>
 #include <string>
@@ -107,7 +107,7 @@ struct Options {
  *
  * All public methods in this class are thread-safe.
  */
-class Configuration : public Pimpl<Configuration>, protected SynchronizedBase {
+class Configuration : public Pimpl<Configuration> {
 public:
     Configuration();
     ~Configuration();
diff --git a/src/core/database.cc b/src/core/database.cc
index 06386815..60a6b6e7 100644
--- a/src/core/database.cc
+++ b/src/core/database.cc
@@ -7,7 +7,6 @@
 #include "sqlite.h"
 #include "util/helpers.h"
 #include "util/testing.h"
-#include "util/threading.h"
 
 #include <algorithm>
 #include <list>
@@ -59,8 +58,6 @@ struct Pimpl<Database>::Implementation {
     // Helper to lookup scheduled query.
     std::optional<std::list<ScheduledQuery>::iterator> lookupQuery(query::ID);
 
-    SynchronizedBase* _synchronized =
-        nullptr; // database's synchronizer, so that we can grab it during callback execution
     const Configuration* _configuration = nullptr; // configuration object, as passed into constructor
     Scheduler* _scheduler = nullptr;               // scheduler as passed into constructor
     std::unique_ptr<SQLite> _sqlite;               // SQLite backend for performing queries
@@ -127,9 +124,7 @@ void Database::Implementation::expire() {
             continue;
 
         if ( (*i)->query.callback_done )
-            _synchronized->unlockWhile([&, id = id, regular_shutdown = regular_shutdown]() {
-                (*(*i)->query.callback_done)(id, ! regular_shutdown);
-            });
+            (*(*i)->query.callback_done)(id, ! regular_shutdown);
     }
 
     for ( const auto& [id, regular_shutdown] : _cancelled_queries ) {
@@ -231,15 +226,12 @@ static auto newRows(std::vector<std::vector<Value>> old, std::vector<std::vector
 }
 
 Interval Database::Implementation::timerCallback(timer::ID id) {
-    SynchronizedBase::Synchronize _(_synchronized);
-
     auto i = lookupQuery(id);
     if ( ! i || (*i)->query.cancelled )
         // already gone, or will be cleaned up shortly
         return 0s;
 
-    auto sql_result = _synchronized->unlockWhile(
-        [&]() { return _sqlite->runStatement(*(*i)->prepared_query, (*i)->previous_execution); });
+    auto sql_result = _sqlite->runStatement(*(*i)->prepared_query, (*i)->previous_execution);
 
     // re-lookup because we released the lock
     i = lookupQuery(id);
@@ -282,14 +274,12 @@ Interval Database::Implementation::timerCallback(timer::ID id) {
 #endif
 
         if ( (*i)->query.callback_result ) {
-            _synchronized->unlockWhile([&]() {
-                auto query_result = query::Result{.columns = sql_result->columns,
-                                                  .rows = std::move(rows),
-                                                  .cookie = (*i)->query.cookie,
-                                                  .initial_result = ! (*i)->previous_result.has_value()};
+            auto query_result = query::Result{.columns = sql_result->columns,
+                                              .rows = std::move(rows),
+                                              .cookie = (*i)->query.cookie,
+                                              .initial_result = ! (*i)->previous_result.has_value()};
 
-                (*(*i)->query.callback_result)(id, query_result);
-            });
+            (*(*i)->query.callback_result)(id, query_result);
 
             // repeat search in case map was modified by callback
             i = lookupQuery(id);
@@ -328,7 +318,6 @@ Table* Database::Implementation::table(const std::string& name) {
 
 Database::Database(Configuration* configuration, Scheduler* scheduler) {
     ZEEK_AGENT_DEBUG("database", "creating instance");
-    pimpl()->_synchronized = this;
     pimpl()->_configuration = configuration;
     pimpl()->_scheduler = scheduler;
     pimpl()->_sqlite = std::make_unique<SQLite>();
@@ -349,15 +338,9 @@ Time Database::currentTime() const {
     return pimpl()->_scheduler->currentTime();
 }
 
-size_t Database::numberQueries() const {
-    Synchronize _(this);
-    return pimpl()->_queries.size();
-}
+size_t Database::numberQueries() const { return pimpl()->_queries.size(); }
 
-Table* Database::table(const std::string& name) {
-    Synchronize _(this);
-    return pimpl()->table(name);
-}
+Table* Database::table(const std::string& name) { return pimpl()->table(name); }
 
 std::vector<const Table*> Database::tables() {
     std::vector<const Table*> out;
@@ -371,7 +354,6 @@ std::vector<const Table*> Database::tables() {
 
 Result<std::optional<query::ID>> Database::query(const Query& q) {
     ZEEK_AGENT_DEBUG("database", "new query: {} ", q.sql_stmt);
-    Synchronize _(this);
 
     auto id = pimpl()->query(q);
     if ( id ) {
@@ -388,26 +370,21 @@ Result<std::optional<query::ID>> Database::query(const Query& q) {
 
 void Database::cancel(query::ID id) {
     ZEEK_AGENT_DEBUG("database", "canceling query {}", id);
-    Synchronize _(this);
     return pimpl()->cancel(id, false);
 }
 
 void Database::poll() {
     ZEEK_AGENT_DEBUG("database", "polling database");
-    Synchronize _(this);
     pimpl()->poll();
     pimpl()->expire();
 }
 
 void Database::expire() {
     ZEEK_AGENT_DEBUG("database", "expiring database state");
-    Synchronize _(this);
     pimpl()->expire();
 }
 
 void Database::addTable(Table* t) {
-    Synchronize _(this);
-
     t->setDatabase(this);
 
     if ( configuration().options().use_mock_data )
diff --git a/src/core/database.h b/src/core/database.h
index 25e5a056..af7148b7 100644
--- a/src/core/database.h
+++ b/src/core/database.h
@@ -5,7 +5,6 @@
 #include "table.h"
 #include "util/pimpl.h"
 #include "util/result.h"
-#include "util/threading.h"
 
 #include <map>
 #include <memory>
@@ -124,9 +123,9 @@ class Scheduler;
  * Tables are typically registered at agent startup, and then remain available
  * for querying through corresponding SQL statements.
  *
- * All public methods in this class are thread-safe.
+ * Note that database methods are not safe against access from different threads.
  */
-class Database : public Pimpl<Database>, public SynchronizedBase {
+class Database : public Pimpl<Database> {
 public:
     /**
      * Constructor.
diff --git a/src/core/signal.cc b/src/core/signal.cc
index 3774e71b..6cd2f806 100644
--- a/src/core/signal.cc
+++ b/src/core/signal.cc
@@ -5,7 +5,6 @@
 #include "logger.h"
 #include "util/fmt.h"
 #include "util/testing.h"
-#include "util/threading.h"
 
 #include <algorithm>
 #include <csignal>
diff --git a/src/core/sqlite.cc b/src/core/sqlite.cc
index ca02b4d3..d91facaf 100644
--- a/src/core/sqlite.cc
+++ b/src/core/sqlite.cc
@@ -585,7 +585,6 @@ SQLite::~SQLite() {
 
 Result<std::unique_ptr<sqlite::PreparedStatement>> SQLite::prepareStatement(const std::string& stmt) {
     ZEEK_AGENT_DEBUG("sqlite", "preparing statement: \"{}\"", stmt);
-    Synchronize _(this);
     return pimpl()->prepareStatement(stmt);
 }
 
@@ -593,19 +592,16 @@ Result<sqlite::Result> SQLite::runStatement(const sqlite::PreparedStatement& stm
     ZEEK_AGENT_DEBUG("sqlite", "executing compiled statement: \"{}\"", ::sqlite3_sql(stmt.statement()));
     assert(stmt.statement());
 
-    Synchronize _(this);
     return pimpl()->runStatement(stmt, t);
 }
 
 Result<sqlite::Result> SQLite::runStatement(const std::string& stmt, std::optional<Time> t) {
-    Synchronize _(this);
     ZEEK_AGENT_DEBUG("sqlite", "executing statement: \"{}\"", stmt);
     return pimpl()->runStatement(stmt, t);
 }
 
 Result<Nothing> SQLite::addTable(Table* table) {
     ZEEK_AGENT_DEBUG("sqlite", "{} table {}", (pimpl()->table(table->name()) ? "replacing" : "adding"), table->name());
-    Synchronize _(this);
     return pimpl()->addTable(table);
 }
 
diff --git a/src/core/sqlite.h b/src/core/sqlite.h
index a71b0802..10f1f535 100644
--- a/src/core/sqlite.h
+++ b/src/core/sqlite.h
@@ -6,7 +6,6 @@
 #include "table.h"
 #include "util/pimpl.h"
 #include "util/result.h"
-#include "util/threading.h"
 
 #include <memory>
 #include <set>
@@ -77,7 +76,7 @@ struct Result {
  *
  * All public methods are thread-safe.
  */
-class SQLite : public Pimpl<SQLite>, SynchronizedBase {
+class SQLite : public Pimpl<SQLite> {
 public:
     SQLite();
     ~SQLite();
diff --git a/src/io/console.cc b/src/io/console.cc
index 7d09e51a..4e6c7b5b 100644
--- a/src/io/console.cc
+++ b/src/io/console.cc
@@ -2,7 +2,6 @@
 
 #include "console.h"
 
-#include "core/configuration.h"
 #include "core/database.h"
 #include "core/logger.h"
 #include "core/signal.h"
@@ -10,10 +9,12 @@
 #include "util/color.h"
 #include "util/fmt.h"
 #include "util/helpers.h"
+#include "util/threading.h"
 
 #include <algorithm>
 #include <csignal>
 #include <iostream>
+#include <map>
 #include <sstream>
 #include <string>
 #include <utility>
@@ -25,6 +26,9 @@ using namespace zeek::agent;
 
 template<>
 struct Pimpl<Console>::Implementation {
+    // One time initialization from main thread.
+    void init();
+
     // Main interactive loop running inside thread.
     void repl();
 
@@ -61,12 +65,19 @@ struct Pimpl<Console>::Implementation {
 
     std::string _scheduled_statement; // pre-scheduled statement
 
-    ConditionVariable _query_done; // flags when a query has been fully processed
+    std::map<std::string, Schema> _tables; // copy of table schema for thread-safety
 
     std::unique_ptr<std::thread> _thread; // console's thread
+    ConditionVariable _query_done;        // flags when a query has been fully processed
     replxx::Replxx _rx;                   // instance of the REPL
 };
 
+void Console::Implementation::init() {
+    for ( auto t : _db->tables() )
+        // Create a copy of the table schema while we are in the main thread.
+        _tables.emplace(t->name(), t->schema());
+}
+
 void Console::Implementation::execute(const std::string& cmd, bool terminate) {
     ZEEK_AGENT_DEBUG("console", "executing: {}", cmd);
 
@@ -118,6 +129,7 @@ void Console::Implementation::execute(const std::string& cmd, bool terminate) {
 }
 
 void Console::Implementation::repl() {
+    // Runs in its own thread.
     auto history_path = platform::dataDirectory() / "history";
     _rx.history_load(history_path.native());
 
@@ -185,21 +197,21 @@ void Console::Implementation::printTables() {
     AsciiTable out;
     out.addHeader({"Name", "Description"});
 
-    for ( const auto& t : _db->tables() )
-        out.addRow({t->name(), t->schema().summary});
+    for ( const auto& [name, schema] : _tables )
+        out.addRow({name, schema.summary});
 
     out.print(std::cout);
 }
 
 Result<Nothing> Console::Implementation::printSchema(const std::string& table) {
-    auto t = _db->table(table);
-    if ( ! t )
+    auto t = _tables.find(table);
+    if ( t == _tables.end() )
         return result::Error("no such table");
 
     AsciiTable out;
     out.addHeader({"Column", "Type", "Description"});
 
-    for ( const auto& c : t->schema().columns )
+    for ( const auto& c : t->second.columns )
         out.addRow({c.name, to_string(c.type), c.summary});
 
     out.print(std::cout);
@@ -236,6 +248,7 @@ void Console::Implementation::query(const std::string& stmt, std::optional<query
     });
 
     _query_done.wait();
+    std::cout << std::endl;
 }
 
 void Console::Implementation::message(const std::string& msg) { _rx.print("%s\n", msg.c_str()); }
@@ -282,14 +295,13 @@ Console::~Console() {
     stop();
 }
 
-void Console::scheduleStatementWithTermination(std::string stmt) {
-    Synchronize _(this);
-    pimpl()->_scheduled_statement = std::move(stmt);
-}
+void Console::scheduleStatementWithTermination(std::string stmt) { pimpl()->_scheduled_statement = std::move(stmt); }
 
 void Console::start() {
     ZEEK_AGENT_DEBUG("console", "starting");
-    Synchronize _(this);
+
+    pimpl()->init();
+
     pimpl()->_thread = std::make_unique<std::thread>([this]() {
         if ( pimpl()->_scheduled_statement.size() )
             pimpl()->execute(pimpl()->_scheduled_statement, true);
@@ -299,8 +311,6 @@ void Console::start() {
 }
 
 void Console::stop() {
-    Synchronize _(this);
-
     if ( pimpl()->_thread ) {
         ZEEK_AGENT_DEBUG("console", "stopping");
         pimpl()->_query_done.notify();
diff --git a/src/io/console.h b/src/io/console.h
index 9a12b879..e0b1523d 100644
--- a/src/io/console.h
+++ b/src/io/console.h
@@ -3,7 +3,6 @@
 #pragma once
 
 #include "util/pimpl.h"
-#include "util/threading.h"
 
 #include <memory>
 #include <string>
@@ -24,7 +23,7 @@ class SignalManager;
  *
  * All public methods are thread-safe.
  */
-class Console : public Pimpl<Console>, SynchronizedBase {
+class Console : public Pimpl<Console> {
 public:
     /**
      * Constructor.
diff --git a/src/io/zeek.cc b/src/io/zeek.cc
index 28131e47..2968991a 100644
--- a/src/io/zeek.cc
+++ b/src/io/zeek.cc
@@ -54,7 +54,7 @@ struct ZeekQuery {
 // more than one Zeek instance may be visible on this connection. The class
 // handles that correctly by tracking which Zeek instances it has seen (through
 // their instance IDs that the Zeek-side agent framework creates)
-class BrokerConnection : SynchronizedBase {
+class BrokerConnection {
 public:
     BrokerConnection(Database* db, Scheduler* scheduler, broker::configuration broker_config)
         : _db(db), _scheduler(scheduler), _endpoint(std::move(broker_config)) {}
@@ -134,8 +134,6 @@ class BrokerConnection : SynchronizedBase {
 };
 
 Result<Nothing> BrokerConnection::connect(const std::string& destination) {
-    Synchronize _(this);
-
     // Parse "host[:port]".
     std::string address;
     unsigned long port = 9999; // default Broker port
@@ -179,19 +177,16 @@ Result<Nothing> BrokerConnection::connect(const std::string& destination) {
         [this](const broker::data_message& msg) {
             _scheduler->schedule([this, msg]() { // process message on the main thread
                 if ( broker::get_topic(msg) == broker::topic::statuses_str ) {
-                    Synchronize _(this);
                     auto x = broker::to<broker::status>(broker::get_data(msg));
                     assert(x); // NOLINT(bugprone-lambda-function-name)
                     processStatus(*x);
                 }
                 else if ( broker::get_topic(msg) == broker::topic::errors_str ) {
-                    Synchronize _(this);
                     auto x = broker::to<broker::error>(broker::get_data(msg));
                     assert(x); // NOLINT(bugprone-lambda-function-name)
                     processError(*x);
                 }
                 else {
-                    Synchronize _(this);
                     processEvent(msg);
                 }
             });
@@ -214,8 +209,6 @@ Result<Nothing> BrokerConnection::connect(const std::string& destination) {
 }
 
 void BrokerConnection::disconnect() {
-    Synchronize _(this);
-
     if ( ! _destination )
         return;
 
@@ -236,8 +229,6 @@ void BrokerConnection::disconnect() {
 }
 
 void BrokerConnection::poll() {
-    Synchronize _(this);
-
     // Expire any state from Zeek instances we haven't seen in a while.
     std::vector<std::string> to_remove;
     for ( const auto& z : _zeek_instances ) {
@@ -258,12 +249,10 @@ void BrokerConnection::installQuery(ZeekQuery zquery) {
         return;
 
     zquery.query.callback_result = [this, zeek_id](query::ID /* query_id */, const query::Result& result) {
-        Synchronize _(this);
         transmitResult(zeek_id, result);
     };
 
     zquery.query.callback_done = [this, zeek_id](query::ID /* query_id */, bool /* cancelled */) {
-        Synchronize _(this);
         ZEEK_CONN_DEBUG("database done with query {}, removing", zeek_id);
         _zeek_queries.erase(zeek_id);
     };
@@ -688,19 +677,16 @@ Zeek::~Zeek() {
 
 void Zeek::start(const std::vector<std::string>& zeeks) {
     ZEEK_IO_DEBUG("starting");
-    Synchronize _(this);
     pimpl()->start(zeeks);
 }
 
 void Zeek::stop() {
     ZEEK_IO_DEBUG("stopping");
-    Synchronize _(this);
     pimpl()->stop();
 }
 
 void Zeek::poll() {
     ZEEK_IO_DEBUG("polling");
-    Synchronize _(this);
     pimpl()->poll();
 }
 
diff --git a/src/io/zeek.h b/src/io/zeek.h
index 9265bc5e..28e02b32 100644
--- a/src/io/zeek.h
+++ b/src/io/zeek.h
@@ -3,7 +3,6 @@
 #pragma once
 
 #include "util/pimpl.h"
-#include "util/threading.h"
 
 #include <memory>
 #include <string>
@@ -22,7 +21,7 @@ class Scheduler;
  *
  * All public methods are thread-safe.
  */
-class Zeek : public Pimpl<Zeek>, SynchronizedBase {
+class Zeek : public Pimpl<Zeek> {
 public:
     /**
      * Constructor.
diff --git a/src/main.cc b/src/main.cc
index ecc1f967..5b87c086 100644
--- a/src/main.cc
+++ b/src/main.cc
@@ -11,7 +11,6 @@
 #include "io/zeek.h"
 #include "util/fmt.h"
 #include "util/helpers.h"
-#include "util/threading.h"
 
 #include <csignal>
 #include <iostream>
diff --git a/zeek-agent b/zeek-agent
index 0caff56e..becd56f4 160000
--- a/zeek-agent
+++ b/zeek-agent
@@ -1 +1 @@
-Subproject commit 0caff56e17928fa9e6c452737db41bffac0ac030
+Subproject commit becd56f4e0168eb033304c289db5f8024e874a4a

From 35c9b5814e06186c4189aae39ddb324bf645b80c Mon Sep 17 00:00:00 2001
From: Robin Sommer <robin@corelight.com>
Date: Mon, 21 Feb 2022 13:31:18 +0100
Subject: [PATCH 3/7] Remove most locking from scheduler.

Only the methods accessing timer state retain a lock, so that we can
schedule timers and tasks from arbitrary threads.
---
 src/core/scheduler.cc | 119 +++++++++++++-----------------------------
 src/core/scheduler.h  |  51 ++++++++----------
 2 files changed, 58 insertions(+), 112 deletions(-)

diff --git a/src/core/scheduler.cc b/src/core/scheduler.cc
index c1de8de1..fc42b206 100644
--- a/src/core/scheduler.cc
+++ b/src/core/scheduler.cc
@@ -38,31 +38,36 @@ struct Pimpl<Scheduler>::Implementation {
     // Advances the current time, fireing all timers now expired.
     bool advance(Time now);
 
-    // Runs all update hooks.
+    // Signals an externally visible state change..
     void updated();
 
     // Executes scheduled activity up to current wall clock.
     bool loop();
 
-    SynchronizedBase* _synchronized =
-        nullptr;            // scheduler's synchronizer, so that we can release it for callback execution
-    timer::ID _next_id = 1; // counter for creating timer IDs
-    Time _now = 0_time;     // current time
+    timer::ID _next_id = 1;            // counter for creating timer IDs
+    Time _now = 0_time;                // current time
+    bool _terminating;                 // true once termination has been requested; ok to access wo/ lock
+    ConditionVariable _interrupt_loop; // notify to interrupt idnle waiting in loop()
+
+    mutable std::mutex
+        _timers_mutex; // mutex protecting access to the current set of timers (_timers_by_id and _timers)
     std::unordered_map<timer::ID, Timer> _timers_by_id;                        // maps IDs to their timers
     std::priority_queue<Timer*, std::vector<Timer*>, TimerComparator> _timers; // timers sorted by time
-    std::vector<std::function<void()>> _update_callbacks;                      // registered update callbacks
-    std::atomic<bool> _terminating; // true once termination has been requested; ok to access wo/ lock
-    ConditionVariable interrupt_loop;
 };
 
 timer::ID Scheduler::Implementation::schedule(timer::ID id, Time t, timer::Callback cb) {
+    std::scoped_lock lock(_timers_mutex);
+
     auto timer = Timer{.due = t, .id = id, .callback = std::move(cb)};
     auto x = _timers_by_id.emplace(id, std::move(timer));
     _timers.push(&x.first->second);
+    updated();
     return id;
 }
 
 void Scheduler::Implementation::cancel(timer::ID id) {
+    std::scoped_lock lock(_timers_mutex);
+
     if ( auto t = _timers_by_id.find(id); t != _timers_by_id.end() )
         // mark as canceled, expiration will eventually delete it
         t->second.canceled = true;
@@ -73,6 +78,7 @@ bool Scheduler::Implementation::advance(Time now) {
         return false;
 
     _now = now;
+    std::unique_lock lock(_timers_mutex);
 
     while ( _timers.size() && _timers.top()->due <= _now ) {
         auto t = std::move(*_timers.top()); // copy it out so that we can remove it before running the callback
@@ -83,123 +89,93 @@ bool Scheduler::Implementation::advance(Time now) {
         _timers_by_id.erase(t.id);
 
         if ( ! t.canceled ) {
-            // Release lock before running callback, so that that can access the scheduler.
-            if ( auto reschedule = _synchronized->unlockWhile([&]() { return t.callback(t.id); }); reschedule > 0s ) {
+            // Release lock before running callback
+            lock.unlock();
+
+            if ( auto reschedule = t.callback(t.id); reschedule > 0s ) {
                 auto due = _now + reschedule;
                 ZEEK_AGENT_DEBUG("scheduler", "rescheduling timer {} for t={}", t.id, to_string(due));
                 schedule(t.id, due, t.callback);
             }
+
+            lock.lock();
         }
     }
 
     return true;
 }
 
+void Scheduler::Implementation::updated() { _interrupt_loop.notify(); }
+
 bool Scheduler::Implementation::loop() {
     Interval timeout = 0s;
 
-    if ( _timers.empty() )
-        timeout = 15s; // TODO: Make max timoeut configurable
-    else
-        timeout = std::max(Interval(0s), _timers.top()->due - std::chrono::system_clock::now());
+    {
+        std::scoped_lock lock(_timers_mutex);
+
+        if ( _timers.empty() )
+            timeout = 5s; // TODO: Make max timeout configurable
+        else
+            timeout = std::max(Interval(0s), _timers.top()->due - std::chrono::system_clock::now());
+    }
 
     if ( timeout > 0s ) {
         ZEEK_AGENT_DEBUG("scheduler", "sleeping with timeout={}", to_string(timeout));
-        _synchronized->unlockWhile([&]() { interrupt_loop.wait(timeout); });
+        _interrupt_loop.wait(timeout);
     }
 
     advance(std::chrono::system_clock::now());
     return _terminating;
 }
 
-void Scheduler::Implementation::updated() {
-    _synchronized->unlockWhile([&]() {
-        for ( const auto& cb : _update_callbacks )
-            cb();
-    });
-
-    ZEEK_AGENT_DEBUG("scheduler", "updated");
-    interrupt_loop.notify();
-}
-
-Scheduler::Scheduler() {
-    ZEEK_AGENT_DEBUG("scheduler", "creating instance");
-    pimpl()->_synchronized = this;
-}
+Scheduler::Scheduler() { ZEEK_AGENT_DEBUG("scheduler", "creating instance"); }
 
 Scheduler::~Scheduler() { ZEEK_AGENT_DEBUG("scheduler", "destroying instance"); }
 
 timer::ID Scheduler::schedule(Time t, timer::Callback cb) {
-    Synchronize _(this);
     auto id = pimpl()->schedule(pimpl()->_next_id++, t, std::move(cb));
-    ZEEK_AGENT_DEBUG("scheduler", "scheduled timer {} for t={}", id, to_string(t));
-    pimpl()->updated();
+    ZEEK_AGENT_DEBUG("scheduler", "scheduling timer {} for t={}", id, to_string(t));
     return id;
 }
 
 void Scheduler::schedule(task::Callback cb) {
-    Synchronize _(this);
     auto id = pimpl()->schedule(pimpl()->_next_id++, currentTime(), [cb = std::move(cb)](timer::ID) -> Interval {
         cb();
         return Interval(0);
     });
 
-    ZEEK_AGENT_DEBUG("scheduler", "scheduled task {} for immediate execution", id);
-    pimpl()->updated();
+    ZEEK_AGENT_DEBUG("scheduler", "scheduling task {} for immediate execution", id);
 }
 
 void Scheduler::cancel(timer::ID id) {
-    Synchronize _(this);
     ZEEK_AGENT_DEBUG("scheduler", "canceling timer {}", id);
     pimpl()->cancel(id);
     pimpl()->updated();
 }
 
 bool Scheduler::loop() {
-    Synchronize _(this);
     ZEEK_AGENT_DEBUG("scheduler", "executing pending activity");
     return pimpl()->loop();
 }
 
 void Scheduler::advance(Time t) {
-    Synchronize _(this);
     ZEEK_AGENT_DEBUG("scheduler", "advancing time to t={}", to_string(t));
     if ( pimpl()->advance(t) )
         pimpl()->updated();
 }
 
-void Scheduler::registerUpdateCallback(std::function<void()> cb) {
-    Synchronize _(this);
-    pimpl()->_update_callbacks.push_back(std::move(cb));
-}
-
 void Scheduler::terminate() {
-    Synchronize _(this);
     ZEEK_AGENT_DEBUG("scheduler", "got request to terminate");
     pimpl()->_terminating = true;
     pimpl()->updated();
 }
 
-bool Scheduler::terminating() const {
-    // Synchronize _(this);
-    return pimpl()->_terminating;
-}
-
-Time Scheduler::currentTime() const {
-    // Synchronize _(this);
-    return pimpl()->_now;
-}
+bool Scheduler::terminating() const { return pimpl()->_terminating; }
 
-Time Scheduler::nextTimer() const {
-    Synchronize _(this);
-    if ( pimpl()->_timers.empty() )
-        return 0_time;
-    else
-        return pimpl()->_timers.top()->due;
-}
+Time Scheduler::currentTime() const { return pimpl()->_now; }
 
 size_t Scheduler::pendingTimers() const {
-    Synchronize _(this);
+    std::scoped_lock lock(pimpl()->_timers_mutex);
     return pimpl()->_timers_by_id.size();
 }
 
@@ -226,7 +202,6 @@ TEST_CASE("timer management") {
             return 0s;
         });
 
-        CHECK_EQ(scheduler.nextTimer(), 5_time);
         CHECK_EQ(scheduler.pendingTimers(), 2);
 
         scheduler.advance(3_time);
@@ -311,24 +286,4 @@ TEST_CASE("timer management") {
         scheduler.terminate();
         CHECK(scheduler.terminating());
     }
-
-    SUBCASE("update callback") {
-        // Checking all non-const methods here (except callback registration)
-        Scheduler scheduler;
-        uint64_t counter = 0;
-        scheduler.registerUpdateCallback([&]() { counter += 1; });
-
-        scheduler.advance(20_time);
-        scheduler.advance(40_time);
-        CHECK_EQ(counter, 2);
-
-        auto id = scheduler.schedule(50_time, [](timer::ID) { return 0s; });
-        CHECK_EQ(counter, 3);
-
-        scheduler.cancel(id);
-        CHECK_EQ(counter, 4);
-
-        scheduler.terminate();
-        CHECK_EQ(counter, 5);
-    }
 }
diff --git a/src/core/scheduler.h b/src/core/scheduler.h
index 181cd435..af9c7742 100644
--- a/src/core/scheduler.h
+++ b/src/core/scheduler.h
@@ -2,8 +2,8 @@
 
 #pragma once
 
+#include "util/helpers.h"
 #include "util/pimpl.h"
-#include "util/threading.h"
 
 #include <functional>
 
@@ -38,10 +38,9 @@ using Callback = std::function<void()>;
  * for example through a deterministic sequence of fixed steps. The latter is
  * particularly useful for unit testing.
  *
- * All public methods in this class are thread-safe. Callbacks will run from
- * the inside the thread that advances the schedule's time.
+ * Note that methods of this class aren't thread-safe unless stated othewise.
  */
-class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
+class Scheduler : public Pimpl<Scheduler> {
 public:
     Scheduler();
     ~Scheduler();
@@ -49,6 +48,9 @@ class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
     /**
      * Schedules a new timer to a specific point of time in the future.
      *
+     * This method is thread-safe and may be called from any thread. The
+     * callback will always be executed on the main thread.
+     *
      * @param t time to schedule the timer for
      * @param cb callback to execute when `t` has been reached
      * @returns a unique ID for the scheduled timer, which will remain valid
@@ -57,9 +59,13 @@ class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
     timer::ID schedule(Time t, timer::Callback cb);
 
     /**
-     * Enqueues a callback to execute at the next possible opportunity fro the scheduler's thread.
-
-     * @param cb callback to execute as soon as possible from the scheduler's thread.
+     * Enqueues a callback to execute at the next possible opportunity fro the
+     * scheduler's thread.
+     *
+     * This method is thread-safe and may be called from any thread. The
+     * callback will always be executed on the main thread.
+     *
+     * @param cb callback to execute as soon as possible
      */
     void schedule(task::Callback cb);
 
@@ -72,16 +78,6 @@ class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
      */
     void cancel(timer::ID id);
 
-    /**
-     * Executes pending activity up to current wall clock. If nothing is
-     * pending, blocks for a  little while (with new activity interrupting the
-     * block).
-     *
-     * @return true if processing is to continue; false if the scheduler has
-     * been asked to terminate
-     */
-    bool loop();
-
     /**
      * Advances to scheduler's notion of the current time. This will let all
      * timers fire that are currently scheduled for a time <= `now`. Their
@@ -102,12 +98,8 @@ class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
     Time currentTime() const;
 
     /**
-     * Returns the time that the most recent upcoming timer will fire at.
-     * Returns time zero if there's no timer currently scheduled.
+     * Returns the number of timers/tasks currently scheduled for execution.
      */
-    Time nextTimer() const;
-
-    /** Returns the number of timers currently scheduled for execution. */
     size_t pendingTimers() const;
 
     /**
@@ -122,15 +114,14 @@ class Scheduler : public Pimpl<Scheduler>, SynchronizedBase {
     bool terminating() const;
 
     /**
-     * Registers a callback to execute whenever something about the scheduler's
-     * state may have changed, including new or canceled timers, or advances in
-     * time. Callbacks will execute from inside the same thread that triggered
-     * the update. They may run in cases where there's no actual state
-     * change.
+     * Executes pending activity up to current wall clock. If nothing is
+     * pending, blocks for a  little while (with new activity interrupting the
+     * block).
      *
-     * @param callback callback to execute after any change in state
-     **/
-    void registerUpdateCallback(std::function<void()> cb);
+     * @return true if processing is to continue; false if the scheduler has
+     * been asked to terminate
+     */
+    bool loop();
 };
 
 } // namespace zeek::agent

From 894110b4956501724ae7525bce5d30af05edd481 Mon Sep 17 00:00:00 2001
From: Robin Sommer <robin@corelight.com>
Date: Mon, 21 Feb 2022 13:31:19 +0100
Subject: [PATCH 4/7] Remove helper class we no longer need.

---
 src/util/threading.cc | 14 --------------
 src/util/threading.h  | 33 ---------------------------------
 2 files changed, 47 deletions(-)

diff --git a/src/util/threading.cc b/src/util/threading.cc
index ee23e783..20a83bbf 100644
--- a/src/util/threading.cc
+++ b/src/util/threading.cc
@@ -9,20 +9,6 @@
 using namespace zeek::agent;
 
 TEST_SUITE("Threading") {
-    TEST_CASE("synchronized") {
-        SynchronizedBase sync;
-        {
-            SynchronizedBase::Synchronize _(&sync);
-            CHECK_FALSE(sync.mutex().try_lock());
-
-            sync.unlockWhile([&]() {
-                CHECK(sync.mutex().try_lock());
-                sync.mutex().unlock();
-            });
-        }
-        CHECK(sync.mutex().try_lock());
-    }
-
     TEST_CASE("condition variable") {
         ConditionVariable cv;
 
diff --git a/src/util/threading.h b/src/util/threading.h
index 744f3f17..938c676c 100644
--- a/src/util/threading.h
+++ b/src/util/threading.h
@@ -13,39 +13,6 @@ using namespace std::literals::chrono_literals;
 
 namespace zeek::agent {
 
-/**
- * Base class providing a synchronizatun helpers for methods to lock against
- * mutual execution.
- */
-class SynchronizedBase {
-public:
-    /**
-     * Class to instantuate to aquire a lock for a given `SynchronizedBase`
-     * instance
-     **/
-    struct Synchronize {
-    public:
-        Synchronize(const SynchronizedBase* s) : _lock(s->_mutex) {}
-
-    private:
-        std::scoped_lock<std::mutex> _lock;
-    };
-
-    /** Returns the internal mutex for testing purposes. */
-    auto& mutex() const { return _mutex; };
-
-    /** Executes a callable with the mutex unlocked, and then reaquired. */
-    template<typename Body>
-    auto unlockWhile(Body f) {
-        _mutex.unlock();
-        ScopeGuard _([&]() { _mutex.lock(); });
-        return f();
-    }
-
-private:
-    mutable std::mutex _mutex;
-};
-
 /**
  * Wraps a `std::condition_variable` with built-in protection against spurious
  * wakeups.

From 0a17521834265096c454db1a8854a7839f5a3251 Mon Sep 17 00:00:00 2001
From: Robin Sommer <robin@corelight.com>
Date: Mon, 21 Feb 2022 13:31:20 +0100
Subject: [PATCH 5/7] In `.schema` output, break out table parameters
 separately.

---
 src/io/console.cc | 25 +++++++++++++++++++++++--
 1 file changed, 23 insertions(+), 2 deletions(-)

diff --git a/src/io/console.cc b/src/io/console.cc
index 4e6c7b5b..ba2018da 100644
--- a/src/io/console.cc
+++ b/src/io/console.cc
@@ -211,10 +211,31 @@ Result<Nothing> Console::Implementation::printSchema(const std::string& table) {
     AsciiTable out;
     out.addHeader({"Column", "Type", "Description"});
 
-    for ( const auto& c : t->second.columns )
-        out.addRow({c.name, to_string(c.type), c.summary});
+    bool has_parameters = false;
+    for ( const auto& c : t->second.columns ) {
+        if ( ! c.is_parameter )
+            out.addRow({c.name, to_string(c.type), c.summary});
+        else
+            has_parameters = true;
+    }
 
+    std::cout << std::endl;
     out.print(std::cout);
+    std::cout << std::endl;
+
+    if ( has_parameters ) {
+        AsciiTable out;
+        out.addHeader({"Table Parameter", "Type", "Description"});
+
+        for ( const auto& c : t->second.columns ) {
+            if ( c.is_parameter )
+                out.addRow({ltrim(c.name, "_"), to_string(c.type), c.summary});
+        }
+
+        out.print(std::cout);
+        std::cout << std::endl;
+    }
+
     return Nothing();
 }
 

From d66284054afda14035737903c345f792dfd02de6 Mon Sep 17 00:00:00 2001
From: Robin Sommer <robin@corelight.com>
Date: Mon, 21 Feb 2022 13:31:21 +0100
Subject: [PATCH 6/7] Fix configure's `--sanitizer` argument.

---
 configure | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/configure b/configure
index cd75f789..3a11043b 100755
--- a/configure
+++ b/configure
@@ -99,7 +99,7 @@ while [ $# -ne 0 ]; do
         --enable-debug)                    cmake_build_type="Debug";;
         --enable-osx-universal)            cmake_osx_architectures="'arm64;x86_64'";;
         --enable-sanitizer)                cmake_use_sanitizers="address";;
-        --enable-sanitizer=*)              cmake_use_sanitizers="${optargs}";;
+        --enable-sanitizer=*)              cmake_use_sanitizers="${optarg}";;
         --enable-static)                   cmake_use_static_linking="yes";;
         --enable-werror)                   cmake_use_werror="yes";;
         --generator=*)                     cmake_generator="${optarg}";;

From 12ac2ecb13bbbedaf17f30dbe985bfb38c14e821 Mon Sep 17 00:00:00 2001
From: Robin Sommer <robin@corelight.com>
Date: Mon, 21 Feb 2022 13:32:22 +0100
Subject: [PATCH 7/7] Small test case fix reported by thread sanitizer.

---
 src/core/signal.cc | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/src/core/signal.cc b/src/core/signal.cc
index 6cd2f806..b7065563 100644
--- a/src/core/signal.cc
+++ b/src/core/signal.cc
@@ -116,8 +116,8 @@ SignalManager::~SignalManager() {
 
 TEST_SUITE("Signal manager") {
     TEST_CASE("signal callbacks") {
-        int count1 = 0;
-        int count2 = 0;
+        std::atomic<int> count1 = 0;
+        std::atomic<int> count2 = 0;
 
         // TODO: For unknown reasons, the first kill() below sometimes gets
         // lost, at least on macOS (sigwait() returns, but it doesn't set a