From 0ec9d911d35a32e05b7a1df64f2bac48bbecb92b Mon Sep 17 00:00:00 2001 From: Everett Badeaux Date: Thu, 5 Nov 2026 11:47:20 -0600 Subject: [PATCH 1/5] pthread writer --- README.md | 48 +- libs/config/config.h | 3 + libs/meter/meter_types/include/age_gauge.h | 4 +- libs/meter/meter_types/include/counter.h | 2 +- libs/meter/meter_types/include/dist_summary.h | 2 +- libs/meter/meter_types/include/gauge.h | 2 +- libs/meter/meter_types/include/max_gauge.h | 2 +- .../meter_types/include/monotonic_counter.h | 2 +- .../include/monotonic_counter_uint.h | 2 +- .../include/percentile_dist_summary.h | 2 +- .../meter_types/include/percentile_timer.h | 2 +- libs/meter/meter_types/include/timer.h | 2 +- libs/writer/README.md | 75 ++ libs/writer/writer_config/writer_config.cpp | 16 +- libs/writer/writer_config/writer_config.h | 18 +- libs/writer/writer_types/include/uds_writer.h | 9 +- .../writer_types/include/writer_types.h | 38 + libs/writer/writer_types/src/uds_writer.cpp | 21 +- libs/writer/writer_wrapper/CMakeLists.txt | 1 + .../writer/writer_wrapper/buffered_writer.cpp | 855 ++++++++++++++++++ libs/writer/writer_wrapper/buffered_writer.h | 37 + libs/writer/writer_wrapper/writer.cpp | 139 +-- libs/writer/writer_wrapper/writer.h | 41 +- spectator/CMakeLists.txt | 1 + spectator/registry.cpp | 9 +- 25 files changed, 1192 insertions(+), 141 deletions(-) create mode 100644 libs/writer/writer_wrapper/buffered_writer.cpp create mode 100644 libs/writer/writer_wrapper/buffered_writer.h diff --git a/README.md b/README.md index a8d6c2a..07d50e8 100644 --- a/README.md +++ b/README.md @@ -8,13 +8,47 @@ It consists of a thin client designed to send metrics through [spectatord](https ## High-Volume Publishing By default, the library sends every meter change to the spectatord sidecar immediately. This involves a blocking -`send` call and underlying system calls, and may not be the most efficient way to publish metrics in high-volume -use cases. For this purpose a simple buffering functionality is implemented, and it can be turned -on by passing a buffer size to the `WriterConfig` constructor. It is important to note that, until this buffer -fills up, the `Publisher` will not send any meters to the sidecar. Therefore, if your application doesn't emit -meters at a high rate, you should either keep the buffer very small, or do not configure a buffer size at all, -which will fall back to the "publish immediately" mode of operation. Meters still in the buffer when the process -exits normally are sent as the writer shuts down. +`send` call and underlying system calls on the thread that updates the meter, and it is only safe from one thread. +For high-volume use cases, and for recording from several threads, turn on buffering by passing a buffer size in +bytes to the `WriterConfig` constructor: + +```cpp +// 4096-byte buffers, with the default flush interval (1 second) and full-buffer policy (block) +WriterConfig writerConfig(WriterTypes::Unix, 4096); + +// Or: send what is buffered every 250 ms, and drop lines instead of waiting when a thread is out of buffer space +WriterConfig writerConfig(WriterTypes::Unix, 4096, std::chrono::milliseconds(250), FullBufferPolicy::Drop); + +auto registry = Registry(Config(writerConfig)); +``` + +With buffering, each thread that updates meters appends its lines to a buffer of its own, and one sender thread +sends the buffers to the sidecar. Updating a meter normally sends nothing, and threads do not wait for each other. + +* **Datagrams.** A thread hands its buffer to the sender when the next line would not fit, or as soon as a line + fills it exactly. Every datagram holds whole lines and is never larger than the buffer size, so choose a size that + the transport can carry. The one exception is a single line that is longer than the buffer size: it is sent as a + datagram of its own. +* **Order.** The lines of one thread are sent in the order in which that thread recorded them. Lines of different + threads are sent in no particular order. +* **Flush interval.** The sender thread collects and sends what every thread has buffered once per flush interval, + so a line never waits for its thread to fill a buffer. The default is 1 second: a line from a thread that records + rarely, or that records a burst and then goes idle, reaches the sidecar about a second later at most, at the cost + of at most one small datagram per recording thread per second. Pass a different interval as the third argument of + the `WriterConfig` constructor. It must be positive. +* **Full-buffer policy.** A thread may have up to 8 full buffers waiting for the sender, so a recording thread + holds at most 9 buffers: those 8 and the one it is filling. What a thread does when it needs more than that is the + fourth argument of the `WriterConfig` constructor: + * `FullBufferPolicy::Block` (the default): the thread waits until the sender has taken one of its buffers. No + line is lost. + * `FullBufferPolicy::Drop`: the thread drops the line that it is recording and returns at once, so it never + waits. The number of dropped lines is logged as a warning, at most once per flush interval. + * `FullBufferPolicy::Send`: the thread sends its own oldest waiting buffer and then carries on. No line is + lost, and the thread does not depend on the sender thread. This is the only case in which a thread that + updates a meter sends. +* **Exit.** A thread that exits hands what it has buffered to the sender. When the process exits normally, the + writer shuts down: it stops the sender thread and sends everything that is still buffered, including the lines of + threads that are still running. Lines recorded after the shutdown has begun are dropped. ## Local & IDE Configuration diff --git a/libs/config/config.h b/libs/config/config.h index 14b5dcd..0627460 100644 --- a/libs/config/config.h +++ b/libs/config/config.h @@ -1,5 +1,6 @@ #pragma once +#include #include #include @@ -28,6 +29,8 @@ class Config const std::string& GetWriterLocation() const noexcept { return m_writerConfig.GetLocation(); } const WriterType& GetWriterType() const noexcept { return m_writerConfig.GetType(); } const unsigned int GetWriterBufferSize() const noexcept { return m_writerConfig.GetBufferSize(); } + std::chrono::milliseconds GetWriterFlushInterval() const noexcept { return m_writerConfig.GetFlushInterval(); } + FullBufferPolicy GetWriterFullBufferPolicy() const noexcept { return m_writerConfig.GetFullBufferPolicy(); } private: diff --git a/libs/meter/meter_types/include/age_gauge.h b/libs/meter/meter_types/include/age_gauge.h index 0e02cbf..876779c 100644 --- a/libs/meter/meter_types/include/age_gauge.h +++ b/libs/meter/meter_types/include/age_gauge.h @@ -18,13 +18,13 @@ class AgeGauge final : public Meter void Now() const { auto line = this->ConstructLine(0); - Writer::GetInstance().Write(line); + Writer::Write(line); } void Set(const double& seconds) const { auto line = this->ConstructLine(seconds); - Writer::GetInstance().Write(line); + Writer::Write(line); } }; diff --git a/libs/meter/meter_types/include/counter.h b/libs/meter/meter_types/include/counter.h index a01469d..ccc97e1 100644 --- a/libs/meter/meter_types/include/counter.h +++ b/libs/meter/meter_types/include/counter.h @@ -20,7 +20,7 @@ class Counter final : public Meter if (delta > 0) { auto line = this->ConstructLine(delta); - Writer::GetInstance().Write(line); + Writer::Write(line); } } }; diff --git a/libs/meter/meter_types/include/dist_summary.h b/libs/meter/meter_types/include/dist_summary.h index d9e3801..9b13706 100644 --- a/libs/meter/meter_types/include/dist_summary.h +++ b/libs/meter/meter_types/include/dist_summary.h @@ -20,7 +20,7 @@ class DistributionSummary final : public Meter if (amount >= 0) { auto line = this->ConstructLine(amount); - Writer::GetInstance().Write(line); + Writer::Write(line); } } }; diff --git a/libs/meter/meter_types/include/gauge.h b/libs/meter/meter_types/include/gauge.h index 6251ad8..3ed75e2 100644 --- a/libs/meter/meter_types/include/gauge.h +++ b/libs/meter/meter_types/include/gauge.h @@ -24,7 +24,7 @@ class Gauge final : public Meter void Set(const double& value) const { auto line = this->ConstructLine(value); - Writer::GetInstance().Write(line); + Writer::Write(line); } }; diff --git a/libs/meter/meter_types/include/max_gauge.h b/libs/meter/meter_types/include/max_gauge.h index 8b59394..434a73a 100644 --- a/libs/meter/meter_types/include/max_gauge.h +++ b/libs/meter/meter_types/include/max_gauge.h @@ -18,7 +18,7 @@ class MaxGauge final : public Meter void Set(const double& value) const { auto line = this->ConstructLine(value); - Writer::GetInstance().Write(line); + Writer::Write(line); } }; diff --git a/libs/meter/meter_types/include/monotonic_counter.h b/libs/meter/meter_types/include/monotonic_counter.h index 4eef903..aaff6e0 100644 --- a/libs/meter/meter_types/include/monotonic_counter.h +++ b/libs/meter/meter_types/include/monotonic_counter.h @@ -18,7 +18,7 @@ class MonotonicCounter final : public Meter void Set(const double& amount) const { auto line = this->ConstructLine(amount); - Writer::GetInstance().Write(line); + Writer::Write(line); } }; diff --git a/libs/meter/meter_types/include/monotonic_counter_uint.h b/libs/meter/meter_types/include/monotonic_counter_uint.h index c22a886..1c28229 100644 --- a/libs/meter/meter_types/include/monotonic_counter_uint.h +++ b/libs/meter/meter_types/include/monotonic_counter_uint.h @@ -18,7 +18,7 @@ class MonotonicCounterUint final : public Meter void Set(const uint64_t& amount) const { auto line = this->ConstructLine(amount); - Writer::GetInstance().Write(line); + Writer::Write(line); } }; diff --git a/libs/meter/meter_types/include/percentile_dist_summary.h b/libs/meter/meter_types/include/percentile_dist_summary.h index 0c632bb..f31bbb9 100644 --- a/libs/meter/meter_types/include/percentile_dist_summary.h +++ b/libs/meter/meter_types/include/percentile_dist_summary.h @@ -24,7 +24,7 @@ class PercentileDistributionSummary final : public Meter if (amount >= 0) { auto line = this->ConstructLine(amount); - Writer::GetInstance().Write(line); + Writer::Write(line); } } }; diff --git a/libs/meter/meter_types/include/percentile_timer.h b/libs/meter/meter_types/include/percentile_timer.h index 88ab55a..35be59f 100644 --- a/libs/meter/meter_types/include/percentile_timer.h +++ b/libs/meter/meter_types/include/percentile_timer.h @@ -20,7 +20,7 @@ class PercentileTimer final : public Meter if (seconds >= 0) { auto line = this->ConstructLine(seconds); - Writer::GetInstance().Write(line); + Writer::Write(line); } } }; diff --git a/libs/meter/meter_types/include/timer.h b/libs/meter/meter_types/include/timer.h index b6731ef..becbefc 100644 --- a/libs/meter/meter_types/include/timer.h +++ b/libs/meter/meter_types/include/timer.h @@ -20,7 +20,7 @@ class Timer final : public Meter if (seconds >= 0) { auto line = this->ConstructLine(seconds); - Writer::GetInstance().Write(line); + Writer::Write(line); } } }; diff --git a/libs/writer/README.md b/libs/writer/README.md index 1967124..a213992 100644 --- a/libs/writer/README.md +++ b/libs/writer/README.md @@ -30,6 +30,7 @@ This component handles configuration of writers: - Handles environment variable overrides via `SPECTATOR_OUTPUT_LOCATION` - Error handling for invalid writer types - Provides a clean API for specifying writer type and location + - Turns on buffering and sets its flush interval and full-buffer policy (see [Buffering](#buffering)) - **Usage Example:** ```cpp @@ -43,6 +44,12 @@ This component handles configuration of writers: // Environment variable overrides any provided value // SPECTATOR_OUTPUT_LOCATION=unix:///custom/path/socket.sock WriterConfig config(WriterTypes::UDP); // Will use Unix socket instead + + // Buffering with 4096-byte buffers, the default flush interval and the default full-buffer policy + WriterConfig config(WriterTypes::Unix, 4096); + + // Buffering with a flush interval of 250 ms and the drop policy + WriterConfig config(WriterTypes::Unix, 4096, std::chrono::milliseconds(250), FullBufferPolicy::Drop); ``` ### Writer Wrapper (`writer_wrapper`) @@ -54,6 +61,74 @@ This component provides a wrapper around the different writer implementations: - Factory pattern for creating writers based on configuration - Handles serialization and transmission of metric data - Abstracts transport details from the rest of the library + - Buffers lines per thread and sends them from one sender thread when buffering is enabled + +## Buffering + +Without a buffer size, the thread that updates a meter sends the line itself, immediately. That mode is only safe +from one thread. + +With a buffer size greater than 0 in the `WriterConfig`, the wrapper buffers per thread: + +- Each thread that updates meters has its own buffer of at most the buffer size, created when the thread records + its first line. Recording a line appends it to that buffer and touches nothing that other recording threads use. +- A thread hands its buffer over when the next line would not fit, or as soon as a line fills it exactly, and + carries on with an empty buffer. It does not wait for the send. +- One sender thread takes the buffers that the threads have handed over and sends each as one datagram. A datagram + holds whole lines and is never larger than the buffer size, except for a single line that is longer than the + buffer size, which is sent as a datagram of its own. +- The lines of one thread are sent in the order in which the thread recorded them. There is no order between the + lines of different threads. + +| `WriterConfig` argument | Default | Meaning | +|-------------------------|--------------------------|---------------------------------------------------------------------| +| `bufferSize` | `0` (no buffering) | Size of each thread's buffer in bytes, and the largest datagram | +| `flushInterval` | `1000 ms` | How often the sender collects and sends what the threads buffered | +| `fullBufferPolicy` | `FullBufferPolicy::Block`| What a thread does when it is out of buffer space | + +### Flush interval + +A thread that records rarely, or that records a burst and then goes idle, would otherwise hold its lines until it +fills a buffer or exits. Once per flush interval the sender thread therefore also takes the buffers that are still +being filled, after any full buffers of the same thread, and sends them. The default of 1 second keeps such lines +about a second behind at most, and costs at most one wake-up of the sender thread and one small datagram per +recording thread per second. The interval must be positive; the `WriterConfig` constructor throws a +`std::runtime_error` otherwise. + +### Full-buffer policy + +A thread may have up to 8 full buffers waiting for the sender thread, so a recording thread holds at most 9 +buffers: those 8 and the one it is filling. The policy says what the thread does when it records a line that needs +a tenth: + +- `FullBufferPolicy::Block` (default): the thread waits until the sender thread has taken one of its buffers. No + line is lost. Only recording threads wait for the sender thread, never the other way around. +- `FullBufferPolicy::Drop`: the thread drops the line that it is recording and returns at once. Lines that are + already buffered are kept. The sender thread logs the number of dropped lines as a warning, at most once per + flush interval. +- `FullBufferPolicy::Send`: the thread sends its own oldest waiting buffer and then carries on. No line is lost, + and the thread does not depend on the sender thread. This is the only case in which a recording thread sends, + so the sender thread and the recording threads take turns on the socket. It does not send faster: there is one + socket, and if the socket stalls, the thread waits in its send. + +### Threads and process exit + +- When a thread exits, what it has buffered goes to the sender thread, and the thread's buffers are released once + they are sent. Handing them over never makes the exiting thread send or wait. The limit of 9 buffers therefore + applies to each running thread: what exited threads have handed over waits for the sender thread without a + limit, so many short-lived threads can build up a backlog while nothing can be sent. +- When the process exits normally, the writer stops the sender thread and then sends everything that is still + buffered, including the lines of threads that are still running, before it closes the socket. This happens on + the exiting thread, so the exit takes as long as those sends. +- Lines recorded after that shutdown has begun are dropped, and a thread that is waiting under the block policy is + released. + +### A receiver that is slow to read + +The sender thread sends in bursts. Over UDP, datagrams that the receiver has no room for are lost without an error. +Over a Unix domain socket, Linux makes the sender wait until the receiver has read. macOS fails the send instead +("no buffer space"); in buffered mode the wrapper then retries for up to 250 ms before it gives the datagram up and +logs the failed send. Without buffering such a send fails at once. ## Integration diff --git a/libs/writer/writer_config/writer_config.cpp b/libs/writer/writer_config/writer_config.cpp index 6f331ab..9ea3ff0 100644 --- a/libs/writer/writer_config/writer_config.cpp +++ b/libs/writer/writer_config/writer_config.cpp @@ -7,6 +7,7 @@ namespace spectator { struct WriterConfigConstants { static constexpr auto RuntimeErrorMessage = "Invalid writer type: "; + static constexpr auto FlushIntervalErrorMessage = "Invalid flush interval, it must be positive: "; }; std::pair GetWriterConfigFromString(const std::string& type) @@ -49,11 +50,22 @@ WriterConfig::WriterConfig(const std::string& type) Logger::info("WriterConfig initialized with type: {}, location: {}", WriterTypeToString(m_type), m_location); } -WriterConfig::WriterConfig(const std::string& type, const unsigned int bufferSize) +WriterConfig::WriterConfig(const std::string& type, const unsigned int bufferSize, + const std::chrono::milliseconds flushInterval, const FullBufferPolicy fullBufferPolicy) : WriterConfig(type) // Constructor delegation { + // The flush interval only matters when buffering is enabled. + if (bufferSize > 0 && flushInterval <= std::chrono::milliseconds::zero()) + { + throw std::runtime_error(WriterConfigConstants::FlushIntervalErrorMessage + + std::to_string(flushInterval.count()) + " ms"); + } + m_bufferSize = bufferSize; - Logger::info("WriterConfig buffering enabled with size: {}", m_bufferSize); + m_flushInterval = flushInterval; + m_fullBufferPolicy = fullBufferPolicy; + Logger::info("WriterConfig buffering enabled with size: {}, flush interval: {} ms, full buffer policy: {}", + m_bufferSize, m_flushInterval.count(), FullBufferPolicyToString(m_fullBufferPolicy)); } } // namespace spectator \ No newline at end of file diff --git a/libs/writer/writer_config/writer_config.h b/libs/writer/writer_config/writer_config.h index 737cfb5..ea928f2 100644 --- a/libs/writer/writer_config/writer_config.h +++ b/libs/writer/writer_config/writer_config.h @@ -2,6 +2,7 @@ #include +#include #include #include @@ -11,16 +12,29 @@ class WriterConfig { public: explicit WriterConfig(const std::string& type); - WriterConfig(const std::string& type, unsigned int bufferSize); + + // Enables buffering when bufferSize is greater than 0. Each thread that records meters then fills its own buffer + // of at most bufferSize bytes, and one sender thread sends the buffers. + // flushInterval: how often the sender thread collects and sends what the threads have buffered, so a line + // never waits longer than this for its thread to fill a buffer. Must be positive when + // buffering is enabled; std::runtime_error otherwise. + // fullBufferPolicy: what a thread does when it has no buffer space left (see FullBufferPolicy). + WriterConfig(const std::string& type, unsigned int bufferSize, + std::chrono::milliseconds flushInterval = BufferingDefaults::FlushInterval, + FullBufferPolicy fullBufferPolicy = BufferingDefaults::FullBuffer); [[nodiscard]] const WriterType& GetType() const noexcept { return m_type; } [[nodiscard]] unsigned int GetBufferSize() const noexcept { return m_bufferSize; } + [[nodiscard]] std::chrono::milliseconds GetFlushInterval() const noexcept { return m_flushInterval; } + [[nodiscard]] FullBufferPolicy GetFullBufferPolicy() const noexcept { return m_fullBufferPolicy; } [[nodiscard]] const std::string& GetLocation() const noexcept { return m_location; } private: WriterType m_type; std::string m_location; unsigned int m_bufferSize = 0; + std::chrono::milliseconds m_flushInterval = BufferingDefaults::FlushInterval; + FullBufferPolicy m_fullBufferPolicy = BufferingDefaults::FullBuffer; }; -} // namespace spectator \ No newline at end of file +} // namespace spectator diff --git a/libs/writer/writer_types/include/uds_writer.h b/libs/writer/writer_types/include/uds_writer.h index f8322e1..55c4f16 100644 --- a/libs/writer/writer_types/include/uds_writer.h +++ b/libs/writer/writer_types/include/uds_writer.h @@ -2,6 +2,7 @@ #include +#include #include #include #include @@ -11,7 +12,12 @@ namespace spectator { class UDSWriter final : public BaseWriter { public: - UDSWriter(const std::string& socketPath); + // receiverFullWait: how long Write() may wait for the receiver to make room when its queue is full, before the + // message is given up as a failed send. Linux makes the sender wait in that case; macOS fails the send with + // "no buffer space" instead, and Write() then retries for this long. With the default of zero such a send fails + // at once, like any other error. + UDSWriter(const std::string& socketPath, + std::chrono::milliseconds receiverFullWait = std::chrono::milliseconds::zero()); ~UDSWriter() override; void Write(const std::string& message) override; void Close() override; @@ -22,6 +28,7 @@ class UDSWriter final : public BaseWriter std::unique_ptr m_socket; boost::asio::local::datagram_protocol::endpoint m_endpoint; bool m_socketEstablished; + std::chrono::milliseconds m_receiverFullWait; bool CreateSocket(); bool TryToSend(const std::string& message); diff --git a/libs/writer/writer_types/include/writer_types.h b/libs/writer/writer_types/include/writer_types.h index 97fd0b2..5075272 100644 --- a/libs/writer/writer_types/include/writer_types.h +++ b/libs/writer/writer_types/include/writer_types.h @@ -4,6 +4,7 @@ #include "udp_writer.h" #include "uds_writer.h" +#include #include #include #include @@ -20,6 +21,28 @@ enum class WriterType Unix }; +// With buffering enabled, each thread fills its own buffer and may have a fixed number of full buffers waiting for +// the sender thread. This is what a thread does when it records a line that does not fit in its buffer while that +// number of full buffers is already waiting. +enum class FullBufferPolicy +{ + // Wait until the sender thread has taken one of the thread's buffers. No line is lost. + Block, + // Drop the line being recorded and return at once. Dropped lines are counted, and the count is logged. + Drop, + // Send the thread's oldest waiting buffer from the recording thread, then carry on. No line is lost. + Send +}; + +struct BufferingDefaults +{ + // The sender thread collects and sends what every thread has buffered at least this often, so a line never + // waits for its thread to fill a buffer. One second keeps a line from a quiet thread at most about a second + // behind, and costs at most one wake-up of the sender and one small datagram per recording thread per second. + static constexpr std::chrono::milliseconds FlushInterval{1000}; + static constexpr FullBufferPolicy FullBuffer = FullBufferPolicy::Block; +}; + struct WriterTypes { static constexpr auto Memory = "memory"; @@ -59,4 +82,19 @@ inline std::string WriterTypeToString(WriterType type) } } +inline std::string FullBufferPolicyToString(FullBufferPolicy policy) +{ + switch (policy) + { + case FullBufferPolicy::Block: + return "block"; + case FullBufferPolicy::Drop: + return "drop"; + case FullBufferPolicy::Send: + return "send"; + default: + return "Unknown"; + } +} + } // namespace spectator \ No newline at end of file diff --git a/libs/writer/writer_types/src/uds_writer.cpp b/libs/writer/writer_types/src/uds_writer.cpp index 706f5c4..84eb23d 100644 --- a/libs/writer/writer_types/src/uds_writer.cpp +++ b/libs/writer/writer_types/src/uds_writer.cpp @@ -2,13 +2,18 @@ #include +#include +#include +#include + namespace spectator { -UDSWriter::UDSWriter(const std::string& socketPath) +UDSWriter::UDSWriter(const std::string& socketPath, std::chrono::milliseconds receiverFullWait) : m_socketPath(socketPath), m_ioContext(std::make_unique()), m_socket(nullptr), - m_socketEstablished(false) + m_socketEstablished(false), + m_receiverFullWait(receiverFullWait) { if (false == CreateSocket()) { @@ -48,9 +53,21 @@ catch (const boost::system::system_error& ex) bool UDSWriter::TryToSend(const std::string& message) try { boost::system::error_code ec; + // The time spent waiting for a receiver whose queue is full, at most m_receiverFullWait per message. + auto waited = std::chrono::microseconds::zero(); + auto pause = std::chrono::microseconds(100); for (int i = 0; i < 3; i++) { size_t sent = m_socket->send_to(boost::asio::buffer(message), m_endpoint, 0, ec); + // The receiver's queue is full. Give it time to read, as a blocking send does on Linux, instead of counting + // this as a failed attempt. Never entered when m_receiverFullWait is zero. + while (ec == boost::asio::error::no_buffer_space && waited < m_receiverFullWait) + { + std::this_thread::sleep_for(pause); + waited += pause; + pause = std::min(pause * 2, std::chrono::microseconds(5000)); + sent = m_socket->send_to(boost::asio::buffer(message), m_endpoint, 0, ec); + } if (ec || sent < message.size()) { Logger::error("UDS Writer: Failed to send message - {}, sent {} bytes out of {}", ec.message(), sent, message.size()); diff --git a/libs/writer/writer_wrapper/CMakeLists.txt b/libs/writer/writer_wrapper/CMakeLists.txt index 4f68a4c..f652177 100644 --- a/libs/writer/writer_wrapper/CMakeLists.txt +++ b/libs/writer/writer_wrapper/CMakeLists.txt @@ -1,4 +1,5 @@ add_library(spectator-writer-wrapper + buffered_writer.cpp writer.cpp ) diff --git a/libs/writer/writer_wrapper/buffered_writer.cpp b/libs/writer/writer_wrapper/buffered_writer.cpp new file mode 100644 index 0000000..298c6cc --- /dev/null +++ b/libs/writer/writer_wrapper/buffered_writer.cpp @@ -0,0 +1,855 @@ +#include + +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +namespace spectator { + +// Buffered mode +// ============= +// +// Each thread that records has its own ThreadBuffers: the buffer it is filling, and up to kMaxWaitingBuffers full +// buffers that it has handed over and that wait for the sender. One sender thread takes the waiting buffers of all +// threads and sends each as one datagram. Recording a line normally means appending it to the calling thread's own +// buffer under that thread's own, normally uncontended, mutex: nothing that another recording thread writes is +// touched, and nothing is sent. +// +// Shared state, and who uses it: +// - g_current: which writer is in use, and the key under which each thread finds its own buffers. Read for every +// line, written only by initialization. +// - the registry of all ThreadBuffers: a thread registers on its first line, the sender releases the buffers of a +// thread that has exited once they are sent. Never used while recording a line. +// - the wake-up flags of the sender: a thread reads them when it hands a full buffer over, see SignalSender. +// - the socket writer and its lock: used by the sender thread, by a recording thread only under +// FullBufferPolicy::Send, and by the thread that shuts the writer down. +// +// Order of the lines of one thread: its buffers are sent oldest first. A buffer is only ever taken out of a thread's +// waiting list by a thread that holds the socket lock and that sends the buffer before it releases that lock, and +// the buffer being filled is only taken when the waiting list is empty. +// +// Lock order: a thread's own lock is always the last lock taken, after the socket lock or the registry lock. The +// socket lock and the registry lock are never held together, and the wake-up lock is never held together with any +// other lock. The sender waits for a recording thread only to take that thread's own lock for a moment (and, under +// FullBufferPolicy::Send, for the socket lock while a recording thread sends one buffer), so a recording thread +// that waits for the sender can always be served. +// +// Thread exit: a thread's buffers are attached to it as the value of a pthread key. The destructor of that key runs +// when the thread exits, after the destructors of the thread's thread_local objects, and closes the buffers: the +// sender sends what is in them and then releases them. From then on the thread is marked as exited, in the key and +// in t_threadExited. Should anything still record on it (the destructor of another key), that line goes straight +// to the sender, see RecordAfterThreadExit: the thread gets no buffers again, as nothing would let go of them. +// +// Process exit: the destructor of the key does not run for threads that are still running when the process exits, +// and that includes the thread that calls exit(). Their buffers stay attached and usable while thread_local and +// static destructors run. ~Writer then calls ShutdownAtExit(), which stops all buffers, joins the sender and sends +// what is left on the exiting thread. The writer itself is never destroyed, so a thread that records after that +// finds its own buffers stopped and drops the line, without touching anything that static destruction destroys. + +namespace { + +constexpr char kNewLine = '\n'; + +// A thread may have this many full buffers waiting for the sender. What it does when it needs to hand over one more +// is the FullBufferPolicy. +constexpr std::size_t kMaxWaitingBuffers = 8; + +// At least the cache line size of the platforms we build for, so that the buffers of two threads never share a line. +constexpr std::size_t kCacheLineSize = 128; + +// The flush interval is kept within these bounds so that the sender's timed wait is well defined for any input. +constexpr std::chrono::milliseconds kMinFlushInterval{1}; +constexpr std::chrono::milliseconds kMaxFlushInterval{std::chrono::hours{24}}; + +using Clock = std::chrono::steady_clock; + +// Once a thread has let go of its buffers at exit, the thread key holds one of the numbers 1 to kMaxExitedRounds +// instead of a pointer. The number counts the rounds of key destructors since then, see OnThreadExit. Every +// implementation runs at least this many rounds (_POSIX_THREAD_DESTRUCTOR_ITERATIONS). +constexpr std::uintptr_t kMaxExitedRounds = 4; + +// Whether the value of the thread key says that the calling thread has no buffers: it is null because the thread +// has not recorded yet, or it is the mark of a thread that has exited. +bool HasNoBuffers(const void* value) noexcept { return reinterpret_cast(value) <= kMaxExitedRounds; } + +// Set when the calling thread has let go of its buffers at exit. The mark in the thread key says the same, but +// neither lasts everywhere for as long as code can run on the thread, which is longest on the thread whose exit +// ends the process and then runs the static destructors. After the last round of key destructors, glibc has +// cleared the key but still has this flag, and macOS has released thread_local storage but still has the key. +constinit thread_local bool t_threadExited = false; + +class BufferedWriter; + +// The full buffers that one thread has handed over and that the sender has not taken yet, oldest first. +class WaitingBuffers +{ + public: + bool Empty() const noexcept { return m_count == 0; } + bool Full() const noexcept { return m_count == kMaxWaitingBuffers; } + + void Push(std::string&& buffer) noexcept + { + m_buffers[(m_first + m_count) % kMaxWaitingBuffers] = std::move(buffer); + ++m_count; + } + + std::string Pop() noexcept + { + std::string buffer = std::move(m_buffers[m_first]); + m_first = (m_first + 1) % kMaxWaitingBuffers; + --m_count; + return buffer; + } + + private: + std::array m_buffers; + std::size_t m_first = 0; + std::size_t m_count = 0; +}; + +// The buffers of one thread. The registry of the BufferedWriter owns them. The thread finds them through the thread +// key until it exits. +struct alignas(kCacheLineSize) ThreadBuffers : std::enable_shared_from_this +{ + ThreadBuffers(std::shared_ptr writer, std::size_t size, FullBufferPolicy fullBufferPolicy) + : owner(std::move(writer)), bufferSize(size), policy(fullBufferPolicy) + { + } + + // The writer these buffers belong to. Holding it here keeps it alive for as long as a thread can still reach it + // through its buffers. + const std::shared_ptr owner; + // Copies of the writer's configuration, so that recording a line reads nothing else of the writer. + const std::size_t bufferSize; + const FullBufferPolicy policy; + + // Guards everything below. Normally only the owning thread takes it. The sender takes it for a moment when it + // looks for buffers to send. + std::mutex mutex; + // FullBufferPolicy::Block: the owning thread waits here until the sender has taken one of its buffers. + std::condition_variable room; + // The buffer being filled. Never longer than bufferSize, unless it is a single line that is longer than that. + std::string current; + // Full buffers on their way to the sender. + WaitingBuffers waiting; + // Lines dropped under FullBufferPolicy::Drop that the sender has not reported yet. + std::uint64_t dropped = 0; + // The owning thread is waiting on `room`. + bool ownerWaiting = false; + // The writer has shut down: lines are no longer accepted. + bool stopped = false; + // The owning thread has let go of these buffers: nothing will be added, and the sender releases them once sent. + bool closed = false; +}; + +// What every recorded line reads and only initialization writes. Constant-initialized and trivially destructible, +// so it can be used at any time, including while the process runs its static destructors, and on a cache line of +// its own, so that reading it never waits for a write to something else. +struct alignas(kCacheLineSize) Current +{ + // The buffered writer in use, or nullptr when buffering is off. + std::atomic writer{nullptr}; + // The value of this key is the calling thread's ThreadBuffers, or says that it has none (see HasNoBuffers). + // Its destructor is OnThreadExit. Created by the first Start() and never deleted; valid whenever `writer` is + // not null. + pthread_key_t threadKey{}; + bool threadKeyCreated = false; +}; + +constinit Current g_current{}; + +// Owns the buffered writer that g_current.writer points to. Deliberately never destroyed: threads may still be +// recording while the process runs its static destructors, and after ~Writer they have to find a writer that has +// shut down, not a destroyed one. +std::shared_ptr& CurrentOwner() +{ + static auto* const owner = new std::shared_ptr(); + return *owner; +} + +class BufferedWriter final : public std::enable_shared_from_this +{ + public: + BufferedWriter(BaseWriter* writer, std::size_t bufferSize, std::chrono::milliseconds flushInterval, + FullBufferPolicy policy) + : m_writer(writer), + m_bufferSize(bufferSize), + m_policy(policy), + m_flushInterval(std::clamp(flushInterval, kMinFlushInterval, kMaxFlushInterval)) + { + } + + ~BufferedWriter() { Shutdown(); } + + BufferedWriter(const BufferedWriter&) = delete; + BufferedWriter& operator=(const BufferedWriter&) = delete; + + // Starts the sender thread. + void Start() { m_sender = std::thread(&BufferedWriter::SenderLoop, this); } + + // Stops accepting lines, releases the threads that wait for buffer space, joins the sender thread, sends what + // every thread still has buffered and lets go of the socket writer. Runs once; later calls return immediately. + void Shutdown(); + + // Records a line in the calling thread's own buffers. Returns false, without recording the line, if the writer + // that the buffers belong to has shut down. + static bool Record(ThreadBuffers& buffers, const std::string& line); + + // Records a line for a thread for which Record() is not possible or returned false: the thread has no buffers, + // or its buffers belong to a writer that has shut down. `value` is the value of the thread key, and `this` is + // the writer in use. + void RecordSlow(void* value, const std::string& line); + + // The calling thread lets go of its buffers: at thread exit, or because they belong to an earlier + // initialization. Their lines go to the sender, which releases the buffers once they are sent. + static void Detach(ThreadBuffers* buffers); + + private: + struct Visited + { + bool sent = false; // at least one buffer was sent + bool finished = false; // the thread has gone and everything it recorded has been sent + }; + + ThreadBuffers* Attach(); + void RecordAfterThreadExit(const std::string& line); + static void HandOver(ThreadBuffers& buffers) noexcept; + void SendOldest(ThreadBuffers& buffers, std::unique_lock& lock); + void SignalSender(); + + void SenderLoop(); + bool Pass(bool flush); + Visited Visit(ThreadBuffers& buffers, bool takePartial); + void Release(std::size_t index); + void DrainAll(); + void ReportDropped(); + void Send(const std::string& buffer); + + // The socket writer, which is not thread-safe. The Writer singleton owns it, and shuts this writer down before + // it replaces or destroys the socket writer. Guarded by m_socketMutex; null once this writer has shut down. + BaseWriter* m_writer; + const std::size_t m_bufferSize; + const FullBufferPolicy m_policy; + const Clock::duration m_flushInterval; + + // Held by whoever is sending. + std::mutex m_socketMutex; + + // Guards the registry. Taken when a thread records its first line, when a thread exits and once per pass of the + // sender; never while a line is recorded. + std::mutex m_registryMutex; + // All thread buffers, in the order in which they were registered. + std::vector> m_registry; + // Buffers were registered since the sender last copied the registry. + bool m_registryChanged = false; + // The final drain is done. From now on a thread that lets go of its buffers removes them from the registry + // itself, because nobody else will. + bool m_finished = false; + + // Used by the sender thread only (and by Shutdown once the sender has been joined). + std::vector m_visiting; // the sender's copy of the registry + std::uint64_t m_dropped = 0; // dropped lines collected from the threads and not logged yet + + // How a recording thread wakes the sender, see SignalSender. + std::atomic m_workPending{false}; + std::atomic m_senderWaiting{false}; + std::mutex m_wakeMutex; + std::condition_variable m_wakeCv; + + std::atomic m_shutdown{false}; + std::mutex m_shutdownMutex; + bool m_shutdownDone = false; + std::thread m_sender; +}; + +bool BufferedWriter::Record(ThreadBuffers& buffers, const std::string& line) +{ + const std::size_t lineSize = line.size() + 1; // with its newline + bool handedOver = false; + { + std::unique_lock lock(buffers.mutex); + for (;;) + { + if (buffers.stopped) + { + return false; + } + if (buffers.current.empty() || buffers.current.size() + lineSize <= buffers.bufferSize) + { + break; // the line goes into the current buffer + } + + // The line does not fit, so the current buffer has to be handed over first. + if (!buffers.waiting.Full()) + { + // The next buffer is allocated before the full one is handed over. Nothing after the hand-over + // can fail, so the sender is always told about it. + std::string next; + next.reserve(std::max(buffers.bufferSize, lineSize)); + HandOver(buffers); + buffers.current = std::move(next); + handedOver = true; + break; + } + + // The thread has reached its limit of waiting buffers. + if (buffers.policy == FullBufferPolicy::Drop) + { + ++buffers.dropped; + return true; + } + if (buffers.policy == FullBufferPolicy::Send) + { + buffers.owner->SendOldest(buffers, lock); + } + else // FullBufferPolicy::Block + { + buffers.ownerWaiting = true; + buffers.room.wait(lock, [&buffers] { return buffers.stopped || !buffers.waiting.Full(); }); + buffers.ownerWaiting = false; + } + // The lock was released in between, and the sender may have taken any of the buffers: look again. + } + + if (buffers.current.capacity() < buffers.current.size() + lineSize) + { + // The first line of a buffer, or a single line that is longer than the buffer size. + buffers.current.reserve(std::max(buffers.bufferSize, buffers.current.size() + lineSize)); + } + buffers.current.append(line); + buffers.current.push_back(kNewLine); + + // Hand the buffer over as soon as it is full: the line filled it exactly, or it is a single line that is + // longer than the buffer size and goes out on its own. If the thread is at its limit of waiting buffers, + // the buffer stays here until the next line needs the space or the sender collects it. + if (buffers.current.size() >= buffers.bufferSize && !buffers.waiting.Full()) + { + HandOver(buffers); + handedOver = true; + } + } + + if (handedOver) + { + buffers.owner->SignalSender(); + } + return true; +} + +void BufferedWriter::HandOver(ThreadBuffers& buffers) noexcept +{ + buffers.waiting.Push(std::move(buffers.current)); + buffers.current.clear(); // a moved-from string is valid, but not necessarily empty +} + +// FullBufferPolicy::Send: the calling thread sends its own oldest waiting buffer. Called and left with the thread's +// own lock held. This is the only place where a recording thread sends. +void BufferedWriter::SendOldest(ThreadBuffers& buffers, std::unique_lock& lock) +{ + // The socket lock comes before a thread's own lock. + lock.unlock(); + std::unique_lock socketLock(m_socketMutex); + lock.lock(); + if (buffers.stopped || !buffers.waiting.Full()) + { + return; // shut down, or the sender has made room in the meantime + } + + // Taken and sent under the socket lock, like the sender does, so the buffers of this thread stay in order. + const std::string oldest = buffers.waiting.Pop(); + lock.unlock(); + Send(oldest); + socketLock.unlock(); + lock.lock(); +} + +// Tells the sender that there is something to take: a full buffer, or the buffers of a thread that has exited. +// +// The sender must not go to sleep and miss this, and recording threads should not share a lock to say it. So the +// sender clears m_workPending before each pass over the threads, and before it sleeps it sets m_senderWaiting and +// then checks m_workPending, all while holding m_wakeMutex. A signalling thread sets m_workPending and then checks +// m_senderWaiting. With sequentially consistent operations, at least one of the two sees the other's flag: either +// the sender finds work pending and makes another pass, or this thread finds the sender waiting and wakes it up. +// Only the one thread that wakes a sleeping sender takes m_wakeMutex; while the sender is busy, signalling threads +// only read the two flags. +void BufferedWriter::SignalSender() +{ + if (!m_workPending.load()) + { + m_workPending.store(true); + } + if (m_senderWaiting.load() && m_senderWaiting.exchange(false)) + { + { + // The sender holds this lock from before it sets m_senderWaiting until it waits, so after this it is + // waiting (or already awake again) and the notification cannot be lost. + std::lock_guard lock(m_wakeMutex); + } + m_wakeCv.notify_one(); + } +} + +void BufferedWriter::RecordSlow(void* value, const std::string& line) +{ + // A thread that has let go of its buffers at exit: the key holds the mark, or else the flag is set. + if (value != nullptr ? HasNoBuffers(value) : t_threadExited) + { + RecordAfterThreadExit(line); + return; + } + + if (auto* const buffers = static_cast(value); buffers != nullptr) + { + if (buffers->owner.get() == this) + { + return; // Record() turned the line down: this writer has shut down, and the line is dropped + } + // The buffers belong to an earlier initialization. That writer has shut down and sent what was in them. + pthread_setspecific(g_current.threadKey, nullptr); + Detach(buffers); + } + + if (m_shutdown.load()) + { + return; + } + Record(*Attach(), line); +} + +// A line recorded on a thread that has already let go of its buffers at exit. It gets buffers of its own that are +// closed from the start, so only the sender sends the line, and then releases them. They are registered after the +// buffers that the thread had before, and the sender empties closed buffers completely in registration order, so +// the line still follows the thread's earlier lines. The limit of waiting buffers and the full-buffer policy do +// not apply to such a line. +void BufferedWriter::RecordAfterThreadExit(const std::string& line) +{ + auto buffers = std::make_shared(shared_from_this(), m_bufferSize, m_policy); + buffers->current.reserve(line.size() + 1); + buffers->current.append(line); + buffers->current.push_back(kNewLine); + buffers->closed = true; + { + std::lock_guard lock(m_registryMutex); + if (m_shutdown.load()) + { + return; + } + m_registry.push_back(std::move(buffers)); + m_registryChanged = true; + } + SignalSender(); +} + +// Creates and registers buffers for the calling thread, and attaches them to it. +ThreadBuffers* BufferedWriter::Attach() +{ + auto buffers = std::make_shared(shared_from_this(), m_bufferSize, m_policy); + ThreadBuffers* const attached = buffers.get(); + { + std::lock_guard lock(m_registryMutex); + // Shutdown() sets m_shutdown before it stops the registered buffers under this lock, so buffers that it + // does not find in the registry see the flag here. + attached->stopped = m_shutdown.load(); + m_registry.push_back(std::move(buffers)); + m_registryChanged = true; + } + + if (pthread_setspecific(g_current.threadKey, attached) != 0) + { + // Out of memory. The thread cannot keep the buffers, so let go of them again. + Detach(attached); + throw std::bad_alloc(); + } + return attached; +} + +void BufferedWriter::Detach(ThreadBuffers* buffers) +{ + // Both have to outlive this call: the writer, although these buffers may be the last thing that refers to it, + // and the buffers, although the sender may release them as soon as they are closed. The writer is declared + // first, so it is destroyed last. + const std::shared_ptr writer = buffers->owner; + const std::shared_ptr self = buffers->shared_from_this(); + { + std::lock_guard lock(buffers->mutex); + buffers->closed = true; + } + + bool released = false; + { + std::lock_guard lock(writer->m_registryMutex); + if (writer->m_finished) + { + // The writer has shut down and done its final drain, so nothing is left in the buffers and nobody else + // will release them. + released = std::erase(writer->m_registry, self) > 0; + } + } + if (!released) + { + writer->SignalSender(); + } +} + +void BufferedWriter::SenderLoop() +{ + Clock::time_point nextFlush = Clock::now() + m_flushInterval; + while (!m_shutdown.load()) + { + // Cleared before the pass, so that a signal that arrives during the pass is still pending after it. + m_workPending.store(false); + + const Clock::time_point now = Clock::now(); + const bool flush = now >= nextFlush; + if (flush) + { + nextFlush = now + m_flushInterval; + } + + const bool sent = Pass(flush); + if (flush) + { + ReportDropped(); + } + if (sent) + { + continue; // the pass is limited per thread, so more may be waiting + } + + std::unique_lock lock(m_wakeMutex); + m_senderWaiting.store(true); + if (!m_workPending.load() && !m_shutdown.load()) + { + m_wakeCv.wait_until(lock, nextFlush); + } + m_senderWaiting.store(false); + } +} + +// Visits every thread once and sends what it has waiting. A flush pass also takes the buffers that are still being +// filled. Returns whether anything was sent. +bool BufferedWriter::Pass(bool flush) +{ + { + std::lock_guard lock(m_registryMutex); + if (m_registryChanged) + { + m_visiting.clear(); + for (const auto& buffers : m_registry) + { + m_visiting.push_back(buffers.get()); + } + m_registryChanged = false; + } + } + + bool sent = false; + std::size_t index = 0; + while (index < m_visiting.size() && !m_shutdown.load()) + { + const Visited visited = Visit(*m_visiting[index], flush); + sent = sent || visited.sent; + if (visited.finished) + { + Release(index); + } + else + { + ++index; + } + } + return sent; +} + +// Sends the waiting buffers of one thread, oldest first, and after them the buffer that is still being filled if +// `takePartial` is set or the thread has gone. +BufferedWriter::Visited BufferedWriter::Visit(ThreadBuffers& buffers, bool takePartial) +{ + Visited visited; + // At most a full waiting list and the buffer being filled per visit, so that a thread that records without + // pause cannot keep the sender from the other threads. This is enough to empty the buffers of a thread that + // has gone. + for (std::size_t taken = 0; taken <= kMaxWaitingBuffers; ++taken) + { + std::string buffer; + bool wakeOwner = false; + bool nothingOlderLeft = false; + + // A buffer is taken and sent under the socket lock. That keeps the buffers of a thread in order when the + // thread sends one itself under FullBufferPolicy::Send. + std::lock_guard socketLock(m_socketMutex); + { + std::lock_guard lock(buffers.mutex); + m_dropped += std::exchange(buffers.dropped, 0); + if (!buffers.waiting.Empty()) + { + buffer = buffers.waiting.Pop(); + wakeOwner = buffers.ownerWaiting; + } + else + { + // The buffer being filled is taken only once nothing is waiting, so its lines cannot overtake the + // thread's full buffers. + if ((takePartial || buffers.closed) && !buffers.current.empty()) + { + buffer.swap(buffers.current); + } + nothingOlderLeft = true; + // If the thread has gone, nothing will be added either. + visited.finished = buffers.closed; + } + } + if (wakeOwner) + { + buffers.room.notify_one(); + } + if (!buffer.empty()) + { + Send(buffer); + visited.sent = true; + } + if (nothingOlderLeft) + { + break; + } + } + return visited; +} + +// Releases the buffers of a thread that has gone, once everything in them has been sent. +void BufferedWriter::Release(std::size_t index) +{ + const ThreadBuffers* const buffers = m_visiting[index]; + m_visiting.erase(m_visiting.begin() + static_cast(index)); + + std::shared_ptr released; + { + std::lock_guard lock(m_registryMutex); + const auto it = std::find_if(m_registry.begin(), m_registry.end(), + [buffers](const auto& registered) { return registered.get() == buffers; }); + if (it != m_registry.end()) + { + released = std::move(*it); + m_registry.erase(it); + } + } +} + +void BufferedWriter::ReportDropped() +{ + if (m_dropped == 0) + { + return; + } + Logger::warn("Buffered writer: dropped {} lines, because the buffers of the recording threads were full", + m_dropped); + m_dropped = 0; +} + +// The caller holds m_socketMutex. +void BufferedWriter::Send(const std::string& buffer) +{ + if (m_writer == nullptr) + { + return; // shut down + } + try + { + m_writer->Write(buffer); + } + catch (const std::exception& e) + { + Logger::error("Buffered writer: failed to send {} bytes: {}", buffer.size(), e.what()); + } +} + +void BufferedWriter::Shutdown() +{ + std::lock_guard shutdownLock(m_shutdownMutex); + if (m_shutdownDone) + { + return; + } + m_shutdownDone = true; + + // 1. No more lines. Lines recorded from here on are dropped, and threads that are waiting for buffer space + // are released. Buffers that are registered later are stopped by Attach(). + m_shutdown.store(true); + { + std::lock_guard registryLock(m_registryMutex); + for (const auto& buffers : m_registry) + { + { + std::lock_guard lock(buffers->mutex); + buffers->stopped = true; + } + buffers->room.notify_all(); + } + } + + // 2. Stop the sender thread. It finishes the buffer that it is sending. + { + // Taken so that the sender is either before its check of m_shutdown or already waiting. + std::lock_guard lock(m_wakeMutex); + } + m_wakeCv.notify_all(); + if (m_sender.joinable()) + { + m_sender.join(); + } + + // 3. Send what is left on this thread: the buffers of threads that have exited and that the sender did not get + // to, and the buffers of threads that are still running. + DrainAll(); + + // 4. Nothing is sent after this, so the owner of the socket writer may close and destroy it. A thread that + // still gets the socket lock finds its buffers stopped. + std::lock_guard socketLock(m_socketMutex); + m_writer = nullptr; +} + +// The final drain, on the thread that shuts the writer down, after the sender thread has been joined. +void BufferedWriter::DrainAll() +{ + std::vector> all; + { + std::lock_guard lock(m_registryMutex); + all = m_registry; + } + for (const auto& buffers : all) + { + // The buffers are stopped, so nothing is added to them any more. + while (Visit(*buffers, true).sent) + { + } + } + ReportDropped(); + + // Release the buffers of threads that have gone. The buffers of threads that are still running stay registered, + // and with them this writer stays alive, until those threads let go of them. + std::vector> released; + { + std::lock_guard registryLock(m_registryMutex); + for (auto it = m_registry.begin(); it != m_registry.end();) + { + bool closed = false; + { + std::lock_guard lock((*it)->mutex); + closed = (*it)->closed; + } + if (closed) + { + released.push_back(std::move(*it)); + it = m_registry.erase(it); + } + else + { + ++it; + } + } + m_finished = true; + } +} + +// The destructor of the thread key: the calling thread is exiting. It lets go of the thread's buffers, leaves the +// mark that the thread has exited in their place and sets t_threadExited. +// +// Setting the key again makes this destructor run again in the next round of key destructors, now with the mark as +// `value`. The mark is renewed each time, so that it is still there for destructors of other keys in later rounds, +// but only kMaxExitedRounds times: implementations are allowed to run rounds for as long as values are set. +void OnThreadExit(void* value) +{ + if (HasNoBuffers(value)) + { + const auto rounds = reinterpret_cast(value); + if (rounds < kMaxExitedRounds) + { + pthread_setspecific(g_current.threadKey, reinterpret_cast(rounds + 1)); + } + return; + } + + BufferedWriter::Detach(static_cast(value)); + pthread_setspecific(g_current.threadKey, reinterpret_cast(std::uintptr_t{1})); + t_threadExited = true; +} + +} // namespace + +namespace buffered_writer { + +void Start(BaseWriter* socketWriter, std::size_t bufferSize, std::chrono::milliseconds flushInterval, + FullBufferPolicy fullBufferPolicy) +{ + if (!g_current.threadKeyCreated) + { + if (const int error = pthread_key_create(&g_current.threadKey, &OnThreadExit); error != 0) + { + throw std::system_error(error, std::generic_category(), "Could not create the key for thread buffers"); + } + g_current.threadKeyCreated = true; + } + + auto buffered = std::make_shared(socketWriter, bufferSize, flushInterval, fullBufferPolicy); + buffered->Start(); + CurrentOwner() = buffered; + g_current.writer.store(buffered.get()); +} + +void Stop() +{ + // Threads that still have buffers of this writer find them stopped on their next line. They get new buffers + // if buffering is turned on again, and otherwise let go of the old ones when they exit. + g_current.writer.store(nullptr); + if (const std::shared_ptr previous = std::move(CurrentOwner()); previous != nullptr) + { + previous->Shutdown(); + } +} + +void ShutdownAtExit() +{ + if (const std::shared_ptr& buffered = CurrentOwner(); buffered != nullptr) + { + buffered->Shutdown(); + } +} + +bool Record(const std::string& line) +{ + BufferedWriter* const buffered = g_current.writer.load(std::memory_order_acquire); + if (buffered == nullptr) + { + return false; + } + + // Normal operation: only the calling thread's own buffers are touched. + void* const value = pthread_getspecific(g_current.threadKey); + if (HasNoBuffers(value) || !BufferedWriter::Record(*static_cast(value), line)) + { + // The first line of this thread, a thread that has exited, or buffers of a writer that has shut down. + buffered->RecordSlow(value, line); + } + return true; +} + +} // namespace buffered_writer + +} // namespace spectator diff --git a/libs/writer/writer_wrapper/buffered_writer.h b/libs/writer/writer_wrapper/buffered_writer.h new file mode 100644 index 0000000..3202261 --- /dev/null +++ b/libs/writer/writer_wrapper/buffered_writer.h @@ -0,0 +1,37 @@ +#pragma once + +#include + +#include +#include +#include + +namespace spectator { + +// The buffered mode of the Writer. Each thread that records appends its lines to its own buffer, and one sender +// thread sends the buffers through the socket writer. Internal to the writer wrapper: only Writer uses it. The +// design is described in buffered_writer.cpp. +namespace buffered_writer { + +// Turns buffering on and starts the sender thread. Lines are sent through `socketWriter`, which has to stay valid +// until Stop() or ShutdownAtExit() has returned. Not thread-safe: Writer::Initialize calls it while no other thread +// is recording, and only while buffering is off. +void Start(BaseWriter* socketWriter, std::size_t bufferSize, std::chrono::milliseconds flushInterval, + FullBufferPolicy fullBufferPolicy); + +// Turns buffering off again, if it is on: stops the sender thread and sends what is still buffered. Used before the +// writer is initialized again. Not thread-safe, like Start(). +void Stop(); + +// For process exit. Like Stop(), it stops the sender thread and sends what is still buffered, but buffering stays +// on: threads that are still recording keep being handled here, where their lines are now dropped, and are never +// sent on to a Writer that is being destroyed. Does nothing if buffering is off. +void ShutdownAtExit(); + +// Records a line, given without its trailing newline, if buffering is on. Returns false if buffering is off, and +// the caller then has to send the line itself. Thread-safe. +bool Record(const std::string& line); + +} // namespace buffered_writer + +} // namespace spectator diff --git a/libs/writer/writer_wrapper/writer.cpp b/libs/writer/writer_wrapper/writer.cpp index 69a7b73..a56e4b9 100644 --- a/libs/writer/writer_wrapper/writer.cpp +++ b/libs/writer/writer_wrapper/writer.cpp @@ -1,47 +1,41 @@ #include +#include #include #include + +#include +#include #include namespace spectator { static constexpr auto NEW_LINE = '\n'; -Writer::~Writer() -{ - auto& instance = GetInstance(); +// How long a buffered send to a Unix socket waits for a receiver whose queue is full (see UDSWriter). Buffered +// sends come in bursts, and where the kernel does not make the sender wait, a burst that outruns the receiver +// would otherwise be dropped at the socket. Bounded, so that a receiver that has stopped reading costs the sender +// this much per buffer rather than stopping it for good. +static constexpr std::chrono::milliseconds RECEIVER_FULL_WAIT{250}; - if (instance.bufferingEnabled) - { - { - // Set under the mutex so ThreadSend cannot miss the wakeup between checking its wait predicate and blocking. - std::lock_guard lock(instance.writeMutex); - instance.shutdown.store(true); - } - instance.cv_receiver.notify_all(); - instance.cv_sender.notify_all(); - if (instance.sendingThread.joinable()) { - instance.sendingThread.join(); - } +Writer::Writer() +{ + // The buffered writer logs from its sender thread and during shutdown, until the end of ~Writer. Constructing + // the Logger before this constructor returns makes sure that it is destroyed after the Writer. + Logger::GetLogger(); +} - // ThreadSend sends only full buffers; send the rest so it is not lost at exit. Take it out under the lock and - // send after releasing it, so threads still calling Write do not wait on the mutex during the send. - std::string remaining; - { - std::lock_guard lock(instance.writeMutex); - remaining = std::move(instance.buffer); - instance.buffer.clear(); - } - if (!remaining.empty()) - { - instance.TryToSend(remaining); - } - } +Writer::~Writer() +{ + // With buffering: stops the sender thread and sends what any thread still has buffered, including threads that + // are still running. After it, nothing uses the socket writer any more, and lines that such threads record are + // dropped. + buffered_writer::ShutdownAtExit(); this->Close(); } -void Writer::Initialize(WriterType type, const std::string& param, int port, unsigned int bufferSize) +void Writer::Initialize(WriterType type, const std::string& param, int port, unsigned int bufferSize, + std::chrono::milliseconds flushInterval, FullBufferPolicy fullBufferPolicy) { // Get the singleton instance directly auto& instance = GetInstance(); @@ -49,39 +43,40 @@ void Writer::Initialize(WriterType type, const std::string& param, int port, uns // Create the new writer based on type try { + std::unique_ptr impl; switch (type) { case WriterType::Memory: - instance.m_impl = std::make_unique(); + impl = std::make_unique(); Logger::info("WriterWrapper initialized as MemoryWriter"); break; case WriterType::UDP: - instance.m_impl = std::make_unique(param, port); + impl = std::make_unique(param, port); Logger::info("WriterWrapper initialized as UDPWriter with host: {} and port: {}", param, port); break; case WriterType::Unix: - instance.m_impl = std::make_unique(param); + // Only the buffered writer waits for a receiver whose queue is full. Without buffering the calling + // thread sends, and a send fails at once as before. + impl = std::make_unique( + param, bufferSize > 0 ? RECEIVER_FULL_WAIT : std::chrono::milliseconds::zero()); Logger::info("WriterWrapper initialized as UnixWriter with socket path: {}", param); break; default: throw std::runtime_error("Unsupported writer type"); } + // An earlier initialization may have turned buffering on. Turn it off, which sends what is still buffered + // through the old socket writer, before that one is replaced. + buffered_writer::Stop(); + + instance.m_impl = std::move(impl); instance.m_currentType = type; - + if (bufferSize > 0) { - instance.bufferingEnabled = true; - instance.bufferSize = bufferSize; - instance.buffer.reserve(bufferSize); - instance.writeImpl = &Writer::BufferedWrite; - // Create a thread with proper binding to the instance method - instance.sendingThread = std::thread(&Writer::ThreadSend, &instance); - } - else - { - // Explicitly set to non-buffered if buffer size is 0 - instance.writeImpl = &Writer::NonBufferedWrite; + buffered_writer::Start(instance.m_impl.get(), bufferSize, flushInterval, fullBufferPolicy); + Logger::info("WriterWrapper buffering enabled with size: {}, flush interval: {} ms, full buffer policy: {}", + bufferSize, flushInterval.count(), FullBufferPolicyToString(fullBufferPolicy)); } } catch (const std::exception& e) @@ -97,48 +92,6 @@ void Writer::TryToSend(const std::string& message) instance.m_impl->Write(message); } -void Writer::ThreadSend() -{ - auto& instance = GetInstance(); - std::string message{}; - while (instance.shutdown.load() == false) - { - { - std::unique_lock lock(instance.writeMutex); - instance.cv_sender.wait( - lock, [&instance] { return instance.buffer.size() >= instance.bufferSize || instance.shutdown.load(); }); - if (instance.shutdown.load() == true) - { - return; - } - message = std::move(instance.buffer); - instance.buffer = std::string(); - instance.buffer.reserve(instance.bufferSize); - } - instance.cv_receiver.notify_one(); - instance.TryToSend(message); - } -} - -void Writer::BufferedWrite(const std::string& message) -{ - auto& instance = GetInstance(); - { - std::unique_lock lock(instance.writeMutex); - // TODO: Optimize memory alloc to not exceed allocated size - instance.cv_receiver.wait( - lock, [&instance] { return instance.buffer.size() < instance.bufferSize || instance.shutdown.load(); }); - if (instance.shutdown.load()) - { - Logger::info("Write operation aborted due to shutdown signal"); - return; - } - instance.buffer.append(message); - instance.buffer.push_back(NEW_LINE); - } - instance.buffer.size() >= instance.bufferSize ? instance.cv_sender.notify_one() : instance.cv_receiver.notify_one(); -} - void Writer::NonBufferedWrite(const std::string& message) { // Since this is a non-static method, we're already operating on an instance @@ -146,8 +99,15 @@ void Writer::NonBufferedWrite(const std::string& message) this->TryToSend(message + NEW_LINE); } -void Writer::Write(const std::string& message) +void Writer::Write(const std::string& line) { + // With buffering, the line goes into the calling thread's own buffer and the sender thread sends it. + if (buffered_writer::Record(line)) + { + return; + } + + // Without buffering, the calling thread sends the line now. auto& instance = GetInstance(); if (!instance.m_impl) @@ -156,8 +116,7 @@ void Writer::Write(const std::string& message) return; } - // Call the member function using the pointer-to-member syntax - (instance.*instance.writeImpl)(message); + instance.NonBufferedWrite(line); } void Writer::Close() @@ -180,4 +139,4 @@ void Writer::Close() } } -} // namespace spectator \ No newline at end of file +} // namespace spectator diff --git a/libs/writer/writer_wrapper/writer.h b/libs/writer/writer_wrapper/writer.h index 82407ca..4517c22 100644 --- a/libs/writer/writer_wrapper/writer.h +++ b/libs/writer/writer_wrapper/writer.h @@ -3,11 +3,19 @@ #include #include +#include #include #include namespace spectator { +// Hands the lines that the meters record to the socket writer (UDPWriter, UDSWriter or MemoryWriter). +// +// Without buffering (buffer size 0), the calling thread sends each line immediately. This is only safe from one +// thread. +// +// With buffering, each thread appends its lines to its own buffer and one sender thread sends the buffers. Recording +// a line normally touches only the calling thread's own buffers. See buffered_writer.h and buffered_writer.cpp. class Writer final : public Singleton { public: @@ -29,18 +37,19 @@ class Writer final : public Singleton friend class Timer; // Private constructor - enforces singleton pattern - Writer() = default; + Writer(); - static void Initialize(WriterType type, const std::string& param = "", int port = 0, unsigned int bufferSize = 0); + // Not thread-safe: call it while no other thread is recording. It may be called again to replace the writer; a + // buffered writer from the earlier call is shut down first, which sends what it still has buffered. + static void Initialize(WriterType type, const std::string& param = "", int port = 0, unsigned int bufferSize = 0, + std::chrono::milliseconds flushInterval = BufferingDefaults::FlushInterval, + FullBufferPolicy fullBufferPolicy = BufferingDefaults::FullBuffer); - static void Write(const std::string& message); - - void BufferedWrite(const std::string& message); + // Records one protocol line, given without its trailing newline. Thread-safe when buffering is enabled. + static void Write(const std::string& line); void NonBufferedWrite(const std::string& message); - void ThreadSend(); - void TryToSend(const std::string& message); void Close(); @@ -49,23 +58,9 @@ class Writer final : public Singleton static BaseWriter* GetImpl() { return Writer::GetInstance().m_impl.get(); } static WriterType GetWriterType() { return GetInstance().m_currentType; } + // The socket writer. With buffering enabled, the buffered writer does all the sending through it. std::unique_ptr m_impl; WriterType m_currentType = WriterType::Memory; // Default type - bool bufferingEnabled = false; - unsigned int bufferSize = 0; - std::string buffer{}; - - // Function pointer for write strategy - member function pointer - using WriteFunction = void (Writer::*)(const std::string&); - WriteFunction writeImpl = &Writer::NonBufferedWrite; // Default to non-buffered - - - // multithreading writes - std::mutex writeMutex; - std::thread sendingThread; - std::condition_variable cv_receiver; - std::condition_variable cv_sender; - std::atomic shutdown{false}; }; -} // namespace spectator \ No newline at end of file +} // namespace spectator diff --git a/spectator/CMakeLists.txt b/spectator/CMakeLists.txt index 52f819d..1b3661e 100644 --- a/spectator/CMakeLists.txt +++ b/spectator/CMakeLists.txt @@ -9,6 +9,7 @@ add_library(spectator-registry ${CMAKE_CURRENT_SOURCE_DIR}/../libs/writer/writer_types/src/memory_writer.cpp ${CMAKE_CURRENT_SOURCE_DIR}/../libs/writer/writer_types/src/udp_writer.cpp ${CMAKE_CURRENT_SOURCE_DIR}/../libs/writer/writer_types/src/uds_writer.cpp + ${CMAKE_CURRENT_SOURCE_DIR}/../libs/writer/writer_wrapper/buffered_writer.cpp ${CMAKE_CURRENT_SOURCE_DIR}/../libs/writer/writer_wrapper/writer.cpp ) diff --git a/spectator/registry.cpp b/spectator/registry.cpp index 954f759..2263660 100644 --- a/spectator/registry.cpp +++ b/spectator/registry.cpp @@ -55,19 +55,22 @@ Registry::Registry(const Config& config) : m_config(config) if (config.GetWriterType() == WriterType::Memory) { Logger::info("Registry initializing Memory Writer"); - Writer::Initialize(config.GetWriterType(), "", 0, this->m_config.GetWriterBufferSize()); + Writer::Initialize(config.GetWriterType(), "", 0, this->m_config.GetWriterBufferSize(), + this->m_config.GetWriterFlushInterval(), this->m_config.GetWriterFullBufferPolicy()); } else if (config.GetWriterType() == WriterType::UDP) { auto [ip, port] = ParseUdpAddress(this->m_config.GetWriterLocation()); Logger::info("Registry initializing UDP Writer at {}:{}", ip, port); - Writer::Initialize(config.GetWriterType(), ip, port, this->m_config.GetWriterBufferSize()); + Writer::Initialize(config.GetWriterType(), ip, port, this->m_config.GetWriterBufferSize(), + this->m_config.GetWriterFlushInterval(), this->m_config.GetWriterFullBufferPolicy()); } else if (config.GetWriterType() == WriterType::Unix) { auto socketPath = ParseUnixAddress(this->m_config.GetWriterLocation()); Logger::info("Registry initializing UDS Writer at {}", socketPath); - Writer::Initialize(config.GetWriterType(), socketPath, 0, this->m_config.GetWriterBufferSize()); + Writer::Initialize(config.GetWriterType(), socketPath, 0, this->m_config.GetWriterBufferSize(), + this->m_config.GetWriterFlushInterval(), this->m_config.GetWriterFullBufferPolicy()); } } From 1bf9a4022c4a0dab35e37460a8d8068b86d67ac6 Mon Sep 17 00:00:00 2001 From: Everett Badeaux Date: Thu, 5 Nov 2026 13:10:56 -0600 Subject: [PATCH 2/5] Add tests --- libs/config/test_config.cpp | 25 + .../writer_config/test_writer_config.cpp | 58 + .../writer_types/test/test_uds_writer.cpp | 145 +- libs/writer/writer_wrapper/CMakeLists.txt | 14 +- .../writer_wrapper/test_buffered_writer.cpp | 1262 +++++++++++++++++ libs/writer/writer_wrapper/test_writer.cpp | 142 +- .../writer_wrapper/writer_test_helper.h | 10 +- spectator/test_registry.cpp | 24 + 8 files changed, 1673 insertions(+), 7 deletions(-) create mode 100644 libs/writer/writer_wrapper/test_buffered_writer.cpp diff --git a/libs/config/test_config.cpp b/libs/config/test_config.cpp index 71bfc5d..528b214 100644 --- a/libs/config/test_config.cpp +++ b/libs/config/test_config.cpp @@ -2,6 +2,7 @@ #include #include +#include #include using namespace spectator; @@ -91,6 +92,30 @@ TEST_F(ConfigTest, WriterConfigInitialization) } } +// Test that the buffering settings of the writer config are passed on +TEST_F(ConfigTest, WriterBufferingSettings) +{ + // Defaults + { + WriterConfig writerConfig(WriterTypes::Memory); + Config config(writerConfig); + + EXPECT_EQ(config.GetWriterBufferSize(), 0u); + EXPECT_EQ(config.GetWriterFlushInterval(), BufferingDefaults::FlushInterval); + EXPECT_EQ(config.GetWriterFullBufferPolicy(), BufferingDefaults::FullBuffer); + } + + // Buffer size, flush interval and full-buffer policy given + { + WriterConfig writerConfig(WriterTypes::Memory, 4096, std::chrono::milliseconds(250), FullBufferPolicy::Send); + Config config(writerConfig); + + EXPECT_EQ(config.GetWriterBufferSize(), 4096u); + EXPECT_EQ(config.GetWriterFlushInterval(), std::chrono::milliseconds(250)); + EXPECT_EQ(config.GetWriterFullBufferPolicy(), FullBufferPolicy::Send); + } +} + // Test extra tags handling TEST_F(ConfigTest, ExtraTags) { diff --git a/libs/writer/writer_config/test_writer_config.cpp b/libs/writer/writer_config/test_writer_config.cpp index efa1796..36d6901 100644 --- a/libs/writer/writer_config/test_writer_config.cpp +++ b/libs/writer/writer_config/test_writer_config.cpp @@ -3,6 +3,7 @@ #include #include +#include #include #include @@ -100,6 +101,63 @@ TEST_F(WriterConfigTest, BufferingConstructor) EXPECT_EQ(config.GetBufferSize(), 2048); } +TEST_F(WriterConfigTest, BufferingDefaults) +{ + // Without buffering + { + const WriterConfig config(WriterTypes::UDP); + EXPECT_EQ(config.GetBufferSize(), 0u); + EXPECT_EQ(config.GetFlushInterval(), BufferingDefaults::FlushInterval); + EXPECT_EQ(config.GetFullBufferPolicy(), BufferingDefaults::FullBuffer); + } + + // With a buffer size only + { + const WriterConfig config(WriterTypes::UDP, 2048); + EXPECT_EQ(config.GetFlushInterval(), std::chrono::milliseconds(1000)); + EXPECT_EQ(config.GetFullBufferPolicy(), FullBufferPolicy::Block); + } +} + +TEST_F(WriterConfigTest, FlushIntervalAndFullBufferPolicy) +{ + // Both given + { + const WriterConfig config(WriterTypes::Unix, 4096, std::chrono::milliseconds(250), FullBufferPolicy::Drop); + EXPECT_EQ(config.GetType(), WriterType::Unix); + EXPECT_EQ(config.GetBufferSize(), 4096u); + EXPECT_EQ(config.GetFlushInterval(), std::chrono::milliseconds(250)); + EXPECT_EQ(config.GetFullBufferPolicy(), FullBufferPolicy::Drop); + } + + // A flush interval alone leaves the default policy, and may be given in another unit + { + const WriterConfig config(WriterTypes::UDP, 2048, std::chrono::seconds(5)); + EXPECT_EQ(config.GetFlushInterval(), std::chrono::milliseconds(5000)); + EXPECT_EQ(config.GetFullBufferPolicy(), FullBufferPolicy::Block); + } + + // The send policy + { + const WriterConfig config(WriterTypes::UDP, 2048, std::chrono::milliseconds(100), FullBufferPolicy::Send); + EXPECT_EQ(config.GetFullBufferPolicy(), FullBufferPolicy::Send); + } + + // The names of the policies, as they appear in the log + EXPECT_EQ(FullBufferPolicyToString(FullBufferPolicy::Block), "block"); + EXPECT_EQ(FullBufferPolicyToString(FullBufferPolicy::Drop), "drop"); + EXPECT_EQ(FullBufferPolicyToString(FullBufferPolicy::Send), "send"); +} + +TEST_F(WriterConfigTest, InvalidFlushInterval) +{ + EXPECT_THROW({ WriterConfig config(WriterTypes::UDP, 2048, std::chrono::milliseconds(0)); }, std::runtime_error); + EXPECT_THROW({ WriterConfig config(WriterTypes::UDP, 2048, std::chrono::milliseconds(-1)); }, std::runtime_error); + + // Without buffering there is nothing to flush, so the interval is not checked + EXPECT_NO_THROW({ WriterConfig config(WriterTypes::UDP, 0, std::chrono::milliseconds(0)); }); +} + TEST_F(WriterConfigTest, InvalidWriterType) { EXPECT_THROW({ WriterConfig config("invalid_type"); }, std::runtime_error); diff --git a/libs/writer/writer_types/test/test_uds_writer.cpp b/libs/writer/writer_types/test/test_uds_writer.cpp index a0dddfa..2345f86 100644 --- a/libs/writer/writer_types/test/test_uds_writer.cpp +++ b/libs/writer/writer_types/test/test_uds_writer.cpp @@ -2,12 +2,27 @@ #include #include "../test_utils/uds_server/uds_server.h" +#include +#include +#include +#include +#include +#include #include #include #include +#include using namespace spectator; +#if defined(__linux__) +// Linux makes a sender wait in the kernel for as long as the queue of the receiver's socket is full. +constexpr bool kSendToFullQueueFails = false; +#else +// Elsewhere, as on macOS, such a send fails with "no buffer space". +constexpr bool kSendToFullQueueFails = true; +#endif + // Test fixture for UDS Writer tests class UDSWriterTest : public testing::Test { @@ -192,4 +207,132 @@ TEST_F(UDSWriterTest, ClientReconnectsWhenServerStartsLater) messages = get_uds_messages(); EXPECT_TRUE(messages.size() == 1); EXPECT_TRUE(messages.at(0) == test_message_after); -} \ No newline at end of file +} + +// A burst of messages can fill the queue of the receiver's socket before the server gets to read: the server looks +// for messages every 10 ms, and the queue may hold as few as ten of them (32 on macOS by default). With a +// receiverFullWait, the burst still arrives in full. Linux makes the sender wait while the queue is full. Where the +// send fails instead, as on macOS, the writer tries again for up to receiverFullWait. +TEST_F(UDSWriterTest, DeliversABurstToAReceiverThatIsSlowToRead) +{ + UDSWriter writer("/tmp/test_uds_socket", std::chrono::seconds(10)); + + constexpr size_t burst = 64; + for (size_t i = 0; i < burst; i++) + { + writer.Write("Burst message " + std::to_string(i)); + } + + // Give the server time to read them all + const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(10); + while (get_uds_messages().size() < burst && std::chrono::steady_clock::now() < deadline) + { + std::this_thread::sleep_for(std::chrono::milliseconds(10)); + } + + const auto messages = get_uds_messages(); + ASSERT_EQ(messages.size(), burst); + for (size_t i = 0; i < burst; i++) + { + EXPECT_EQ(messages.at(i), "Burst message " + std::to_string(i)); + } +} + +// The writer waits for a receiver whose queue is full, but only for receiverFullWait per message. Then it gives the +// message up, so that a receiver that has stopped reading does not stop the sender for good. +TEST(UDSWriterReceiverFullTest, GivesUpOnAReceiverThatDoesNotRead) +{ + if (!kSendToFullQueueFails) + { + GTEST_SKIP() << "On this platform a send to a full queue waits in the kernel instead of failing"; + } + + // A receiver that does not read + using DatagramProtocol = boost::asio::local::datagram_protocol; + const std::string socketPath = "/tmp/test_uds_full_socket"; + std::filesystem::remove(socketPath); + boost::asio::io_context ioContext; + DatagramProtocol::socket receiver(ioContext, DatagramProtocol::endpoint(socketPath)); + receiver.non_blocking(true); + + std::vector received; + const auto readQueue = [&receiver, &received] + { + std::array buffer{}; + for (;;) + { + boost::system::error_code ec; + const auto bytesRead = receiver.receive(boost::asio::buffer(buffer), 0, ec); + if (ec) + { + return; + } + received.emplace_back(buffer.data(), bytesRead); + } + }; + + // Write until three messages have each taken as long as the writer waits for a full queue. This runs on a thread + // of its own, so that a writer that does not give up fails the test instead of hanging it. + struct Written + { + size_t messages = 0; + std::chrono::milliseconds elapsed{0}; + }; + const auto receiverFullWait = std::chrono::milliseconds(50); + std::future writing = std::async( + std::launch::async, + [&socketPath, receiverFullWait] + { + constexpr size_t mostMessages = 10000; + UDSWriter writer(socketPath, receiverFullWait); + Written written; + int slowWrites = 0; + const auto start = std::chrono::steady_clock::now(); + while (written.messages < mostMessages && slowWrites < 3) + { + const auto before = std::chrono::steady_clock::now(); + writer.Write("Message " + std::to_string(written.messages)); + written.messages++; + if (std::chrono::steady_clock::now() - before >= receiverFullWait) + { + slowWrites++; + } + } + written.elapsed = + std::chrono::duration_cast(std::chrono::steady_clock::now() - start); + return written; + }); + + const bool gaveUp = writing.wait_for(std::chrono::seconds(30)) == std::future_status::ready; + EXPECT_TRUE(gaveUp) << "The writer is still waiting for the receiver"; + while (writing.wait_for(std::chrono::milliseconds(10)) != std::future_status::ready) + { + // The writer did not give up. Reading makes room in the queue, so that it gets to the end. + readQueue(); + } + const Written written = writing.get(); + if (gaveUp) + { + readQueue(); + } + std::filesystem::remove(socketPath); + + if (!gaveUp) + { + return; + } + if (received.size() == written.messages) + { + GTEST_SKIP() << "The queue of the receiver did not fill up with " << written.messages << " messages"; + } + + // The queue holds the messages from before it was full, in order. The writer gave the others up, each after it + // had waited for receiverFullWait. + ASSERT_LT(received.size(), written.messages); + for (size_t i = 0; i < received.size(); i++) + { + EXPECT_EQ(received.at(i), "Message " + std::to_string(i)); + } + const auto givenUp = static_cast(written.messages - received.size()); + EXPECT_GE(written.elapsed.count(), givenUp * receiverFullWait.count()); +} diff --git a/libs/writer/writer_wrapper/CMakeLists.txt b/libs/writer/writer_wrapper/CMakeLists.txt index f652177..8e6e7e3 100644 --- a/libs/writer/writer_wrapper/CMakeLists.txt +++ b/libs/writer/writer_wrapper/CMakeLists.txt @@ -27,4 +27,16 @@ target_link_libraries(writer_test spectator-meter-id uds_server_lib ) -add_test(NAME writer_test COMMAND writer_test) \ No newline at end of file +add_test(NAME writer_test COMMAND writer_test) + +add_executable(buffered_writer_test + test_buffered_writer.cpp +) + +target_link_libraries(buffered_writer_test + PRIVATE + GTest::GTest + GTest::Main + spectator-writer-wrapper +) +add_test(NAME buffered_writer_test COMMAND buffered_writer_test) \ No newline at end of file diff --git a/libs/writer/writer_wrapper/test_buffered_writer.cpp b/libs/writer/writer_wrapper/test_buffered_writer.cpp new file mode 100644 index 0000000..5484801 --- /dev/null +++ b/libs/writer/writer_wrapper/test_buffered_writer.cpp @@ -0,0 +1,1262 @@ +#include + +#include + +#include +#include +#include + +#include + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace spectator; + +// These tests use the buffered writer directly, through buffered_writer.h, with a FakeSocketWriter in place of a +// socket. The fake keeps what is sent and can hold a send back, which stalls the sender thread like a socket whose +// receiver does not read. With the sender stalled, a test brings a thread to its limit of waiting buffers step by +// step, so what happens at that limit does not depend on which thread is faster. + +namespace { + +// How long a test waits for something that has to happen before it gives up and fails. +constexpr std::chrono::seconds kTimeout{30}; + +// How long a test watches for something that must not happen. +constexpr std::chrono::milliseconds kQuietTime{100}; + +// A flush interval that is longer than any test runs, for the tests in which nothing may be sent on a timer. +constexpr std::chrono::milliseconds kNoTimedFlush = std::chrono::hours{1}; + +// The buffer size at which every Line() fills a buffer exactly, so that each is handed over as it is recorded. +constexpr std::size_t kOneLine = 8; + +// The number of full buffers that a thread may have waiting for the sender: kMaxWaitingBuffers in +// buffered_writer.cpp. +constexpr int kMaxWaiting = 8; + +using Datagrams = std::vector; + +// A line that is 8 bytes long with its newline: a letter and a six-digit number. +std::string Line(char letter, int number) { return fmt::format("{}{:06d}", letter, number); } + +// The datagram that holds just the given line. +std::string Alone(const std::string& line) { return line + '\n'; } + +// The datagrams that hold Line(letter, first) to Line(letter, last), one line each. +Datagrams EachAlone(char letter, int first, int last) +{ + Datagrams datagrams; + for (int number = first; number <= last; ++number) + { + datagrams.push_back(Alone(Line(letter, number))); + } + return datagrams; +} + +// Stands in for the socket writer. It keeps every datagram together with the thread that sent it. Its sends can be +// held back: a Write() that is held blocks, like a send to a socket whose receiver does not read. +class FakeSocketWriter final : public BaseWriter +{ + public: + struct Datagram + { + std::string data; + std::thread::id sentBy; + }; + + void Write(const std::string& message) override + { + // The real socket writers are not thread-safe, so no two calls may overlap. + const CallInProgress callInProgress(*this); + + std::unique_lock lock(m_mutex); + if (m_failNextWrite) + { + m_failNextWrite = false; + ++m_failedWrites; + throw std::runtime_error("FakeSocketWriter: this send fails"); + } + if (m_holding && m_allowed == 0) + { + m_held = message; + m_changed.notify_all(); + m_changed.wait(lock, [this] { return !m_holding || m_allowed > 0; }); + m_held.reset(); + } + if (m_holding) + { + --m_allowed; + } + m_lines += static_cast(std::count(message.begin(), message.end(), '\n')); + m_datagrams.push_back({message, std::this_thread::get_id()}); + m_changed.notify_all(); + } + + void Close() override {} + + // From now on, a Write() does not return until it is allowed to. + void HoldWrites() + { + std::lock_guard lock(m_mutex); + m_holding = true; + m_allowed = 0; + } + + // Lets `count` more calls of Write() return. + void AllowWrites(std::size_t count) + { + { + std::lock_guard lock(m_mutex); + m_allowed += count; + m_held.reset(); + } + m_changed.notify_all(); + } + + // Stops holding: every Write() returns. + void ReleaseWrites() + { + { + std::lock_guard lock(m_mutex); + m_holding = false; + m_held.reset(); + } + m_changed.notify_all(); + } + + // The next Write() throws, like a send that fails. + void FailNextWrite() + { + std::lock_guard lock(m_mutex); + m_failNextWrite = true; + } + + // Waits until a Write() is being held and returns its datagram, or nothing if none is held in time. + std::optional WaitForHeldWrite() + { + std::unique_lock lock(m_mutex); + m_changed.wait_for(lock, kTimeout, [this] { return m_held.has_value(); }); + return m_held; + } + + // Waits until at least `count` datagrams have been sent. False if that does not happen in time. + bool WaitForDatagrams(std::size_t count) + { + std::unique_lock lock(m_mutex); + return m_changed.wait_for(lock, kTimeout, [this, count] { return m_datagrams.size() >= count; }); + } + + // Waits until the datagrams that have been sent hold at least `count` lines. False if that does not happen in + // time. + bool WaitForLines(std::size_t count) + { + std::unique_lock lock(m_mutex); + return m_changed.wait_for(lock, kTimeout, [this, count] { return m_lines >= count; }); + } + + // The datagrams sent so far, in the order in which they were sent. + std::vector SentDatagrams() const + { + std::lock_guard lock(m_mutex); + return m_datagrams; + } + + // The same, without the threads that sent them. + Datagrams Sent() const + { + std::lock_guard lock(m_mutex); + Datagrams sent; + for (const Datagram& datagram : m_datagrams) + { + sent.push_back(datagram.data); + } + return sent; + } + + std::size_t FailedWrites() const + { + std::lock_guard lock(m_mutex); + return m_failedWrites; + } + + // Whether a Write() ever began while another one was in progress. + bool Overlapped() const { return m_overlapped.load(); } + + // Forgets the datagrams sent so far. + void Clear() + { + std::lock_guard lock(m_mutex); + m_datagrams.clear(); + m_lines = 0; + } + + private: + class CallInProgress + { + public: + explicit CallInProgress(FakeSocketWriter& writer) : m_writer(writer) + { + if (m_writer.m_callsInProgress.fetch_add(1) != 0) + { + m_writer.m_overlapped.store(true); + } + } + ~CallInProgress() { m_writer.m_callsInProgress.fetch_sub(1); } + + CallInProgress(const CallInProgress&) = delete; + CallInProgress& operator=(const CallInProgress&) = delete; + + private: + FakeSocketWriter& m_writer; + }; + + std::atomic m_callsInProgress{0}; + std::atomic m_overlapped{false}; + + mutable std::mutex m_mutex; + std::condition_variable m_changed; + std::vector m_datagrams; + std::size_t m_lines = 0; + bool m_holding = false; + std::size_t m_allowed = 0; // while holding: the number of calls that may still return + std::optional m_held; // the datagram of the call that is being held + bool m_failNextWrite = false; + std::size_t m_failedWrites = 0; +}; + +// A thread that records the lines it is told to and stays alive in between. A test uses it to decide what a thread +// other than its own has recorded, and when that thread exits. +class RecordingThread +{ + public: + RecordingThread() : m_thread([this] { Run(); }) {} + + ~RecordingThread() + { + Exit(); + m_gone.wait(); + } + + RecordingThread(const RecordingThread&) = delete; + RecordingThread& operator=(const RecordingThread&) = delete; + + std::thread::id Id() const { return m_id; } + + // Starts `work` on the thread and returns at once. The result is there once the thread has done the work. + std::future StartWork(std::function work) + { + std::packaged_task task(std::move(work)); + std::future result = task.get_future(); + { + std::lock_guard lock(m_mutex); + m_tasks.push_back(std::move(task)); + } + m_wake.notify_one(); + return result; + } + + // Does `work` on the thread. False if the work returns false, or does not return in time. + bool Do(std::function work) + { + std::future result = StartWork(std::move(work)); + return result.wait_for(kTimeout) == std::future_status::ready && result.get(); + } + + // Starts recording `line` on the thread and returns at once, for a line that the thread may have to wait with. + std::future StartRecording(const std::string& line) + { + return StartWork([line] { return buffered_writer::Record(line); }); + } + + // Records `line` on the thread. False if the line is not taken, or the thread is still waiting with it when + // time is up. + bool Record(const std::string& line) + { + return Do([line] { return buffered_writer::Record(line); }); + } + + // Ends the thread, which makes it let go of its buffers, and waits until it is gone. False if it is not gone in + // time. + bool Exit() + { + if (!m_gone.valid()) + { + { + std::lock_guard lock(m_mutex); + m_exit = true; + } + m_wake.notify_one(); + // Joined on a thread of its own, so that a thread that does not exit fails the test instead of hanging + // it. + m_gone = std::async(std::launch::async, [this] { m_thread.join(); }); + } + return m_gone.wait_for(kTimeout) == std::future_status::ready; + } + + private: + void Run() + { + for (;;) + { + std::packaged_task task; + { + std::unique_lock lock(m_mutex); + m_wake.wait(lock, [this] { return m_exit || !m_tasks.empty(); }); + if (m_tasks.empty()) + { + return; + } + task = std::move(m_tasks.front()); + m_tasks.pop_front(); + } + task(); + } + } + + // Used by Run(), so they come before m_thread, which starts it. + std::mutex m_mutex; + std::condition_variable m_wake; + std::deque> m_tasks; + bool m_exit = false; + + std::thread m_thread; + const std::thread::id m_id{m_thread.get_id()}; + std::future m_gone; +}; + +// Keeps what the library logs. The number of lines that a thread has dropped is only reported in the log, as +// "Buffered writer: dropped lines, ...", so this is where a test reads it. +class LogCapture final : public spdlog::sinks::base_sink +{ + public: + // Forgets what has been logged up to now. False if the logger does not respond in time. + bool StartOver() + { + const bool arrived = WaitForLogger(); + const std::lock_guard lock(mutex_); + m_messages.clear(); + return arrived; + } + + // The number of lines that the buffered writer has reported as dropped since StartOver(). + std::uint64_t DroppedLines() + { + static constexpr std::string_view kReport = "Buffered writer: dropped "; + + WaitForLogger(); + const std::lock_guard lock(mutex_); + std::uint64_t dropped = 0; + for (const std::string& message : m_messages) + { + if (message.starts_with(kReport)) + { + dropped += std::stoull(message.substr(kReport.size())); + } + } + return dropped; + } + + protected: + void sink_it_(const spdlog::details::log_msg& message) override + { + m_messages.emplace_back(message.payload.data(), message.payload.size()); + } + + void flush_() override + { + ++m_flushes; + m_flushed.notify_all(); + } + + private: + // The logger hands its messages to a thread of its own, which passes them on to the sinks. A flush goes the same + // way, so once it has got here, so has everything that was logged before it. + bool WaitForLogger() + { + std::unique_lock lock(mutex_); + const std::uint64_t flushesBefore = m_flushes; + lock.unlock(); + Logger::GetLogger()->flush(); + lock.lock(); + return m_flushed.wait_for(lock, kTimeout, [this, flushesBefore] { return m_flushes > flushesBefore; }); + } + + // Guarded by mutex_ of the base class, which is held whenever sink_it_() and flush_() are called. + std::vector m_messages; + std::uint64_t m_flushes = 0; + std::condition_variable m_flushed; +}; + +// The capture, which is added to the library's logger on the first call. +LogCapture& Log() +{ + static const std::shared_ptr capture = [] + { + auto created = std::make_shared(); + Logger::GetLogger()->sinks().push_back(created); + return created; + }(); + return *capture; +} + +// Adds the capture before the first test. Nothing has been logged at that point, so the logger's thread is not +// reading the list of sinks while it changes. +class LogCaptureEnvironment : public testing::Environment +{ + public: + void SetUp() override { Log(); } +}; + +testing::Environment* const logCaptureEnvironment = testing::AddGlobalTestEnvironment(new LogCaptureEnvironment); + +// The line with the given number that the given thread records in the tests with several threads. The lines of a +// thread differ in length. +std::string ThreadLine(int thread, int number) +{ + return fmt::format("t{}.{}.{}", thread, number, std::string(static_cast(number % 40), 'x')); +} + +// What arrived of the lines that threads recorded with ThreadLine(), and what was wrong with it. +struct Arrived +{ + std::vector lines; // per thread, the number of its lines that arrived + int total = 0; // the same for all threads together + int tooLarge = 0; // datagrams that are larger than the buffer size + int notWholeLines = 0; // datagrams that do not end with a newline, and lines that are not as they were recorded + int mixed = 0; // lines in a datagram that begins with a line of another thread + int outOfOrder = 0; // lines that arrived after a later line of their thread, or twice +}; + +Arrived Examine(const Datagrams& datagrams, std::size_t bufferSize, int threads, int linesPerThread) +{ + Arrived arrived; + arrived.lines.assign(static_cast(threads), 0); + std::vector next(static_cast(threads), 0); // per thread, the lowest number its next line has + + for (const std::string& datagram : datagrams) + { + if (datagram.size() > bufferSize) + { + ++arrived.tooLarge; + } + if (datagram.empty() || datagram.back() != '\n') + { + ++arrived.notWholeLines; + continue; + } + + int datagramThread = -1; + for (std::size_t start = 0; start < datagram.size();) + { + const std::size_t end = datagram.find('\n', start); + const std::string line = datagram.substr(start, end - start); + start = end + 1; + + int thread = -1; + int number = -1; + if (std::sscanf(line.c_str(), "t%d.%d.", &thread, &number) != 2 || thread < 0 || thread >= threads || + number < 0 || number >= linesPerThread || line != ThreadLine(thread, number)) + { + ++arrived.notWholeLines; + continue; + } + + if (datagramThread == -1) + { + datagramThread = thread; + } + else if (thread != datagramThread) + { + ++arrived.mixed; + } + + const auto index = static_cast(thread); + if (number < next[index]) + { + ++arrived.outOfOrder; + } + else + { + next[index] = number + 1; + } + ++arrived.lines[index]; + ++arrived.total; + } + } + return arrived; +} + +// Stands in for code that still records while a thread is being torn down: the destructor of a pthread key, which +// runs once in every round of key destructors for as long as it sets its key again. The buffered writer lets go of +// the thread's buffers in the first round, in the destructor of its own key, so from the second round on these lines +// are recorded on a thread that has no buffers any more. +// Round 1: one line. Depending on which of the two destructors runs first, the thread still has its buffers. +// Round 2: more lines than a thread may have waiting, and may hold at all. +// Round 4, the last round that an implementation has to run: a line that does not fill a buffer. +constexpr int kLastLateLineOfRound2 = kMaxWaiting + 4; +constexpr const char* kLineOfLastRound = "end"; + +pthread_key_t LateRecorderKey(); + +void RecordLate(void* value) +{ + int& round = *static_cast(value); + ++round; + if (round == 1) + { + buffered_writer::Record(Line('z', 1)); + } + else if (round == 2) + { + for (int number = 2; number <= kLastLateLineOfRound2; ++number) + { + buffered_writer::Record(Line('z', number)); + } + } + else if (round == 4) + { + buffered_writer::Record(kLineOfLastRound); + } + if (round < 4) + { + pthread_setspecific(LateRecorderKey(), value); // makes this destructor run again in the next round + } +} + +pthread_key_t LateRecorderKey() +{ + static const pthread_key_t key = [] + { + pthread_key_t created{}; + if (pthread_key_create(&created, &RecordLate) != 0) + { + throw std::runtime_error("Could not create the key of the late recorder"); + } + return created; + }(); + return key; +} + +class BufferedWriterTest : public testing::Test +{ + protected: + void SetUp() override { ASSERT_TRUE(Log().StartOver()); } + + void TearDown() override + { + // Whatever a test leaves behind, this lets everything finish: no send is held any longer, and shutting the + // writer down releases a thread that waits for buffer space. + m_socket.ReleaseWrites(); + m_secondSocket.ReleaseWrites(); + buffered_writer::Stop(); + for (std::thread& thread : m_threads) + { + thread.join(); + } + m_recordingThreads.clear(); + + EXPECT_FALSE(m_socket.Overlapped()) << "the socket writer was used by two threads at once"; + EXPECT_FALSE(m_secondSocket.Overlapped()) << "the socket writer was used by two threads at once"; + } + + // Turns buffering on, with m_socket as the socket writer. + void StartWriter(std::size_t bufferSize, std::chrono::milliseconds flushInterval = kNoTimedFlush, + FullBufferPolicy fullBufferPolicy = FullBufferPolicy::Block) + { + buffered_writer::Start(&m_socket, bufferSize, flushInterval, fullBufferPolicy); + } + + // A thread that records what the test tells it to. It exits when the test tells it to, or else when the test is + // over. + RecordingThread& NewThread() { return *m_recordingThreads.emplace_back(std::make_unique()); } + + // Runs `work` on a new thread. + void RunOnNewThread(std::function work) + { + { + std::lock_guard lock(m_threadsMutex); + ++m_threadsAtWork; + } + m_threads.emplace_back( + [this, work = std::move(work)] + { + work(); + { + std::lock_guard lock(m_threadsMutex); + --m_threadsAtWork; + } + m_threadsDone.notify_all(); + }); + } + + // Waits until the threads started with RunOnNewThread() have done their work and are gone. False if they are + // not done in time, which is how a test fails, instead of hanging, when its threads wait for each other. + bool WaitForThreads() + { + { + std::unique_lock lock(m_threadsMutex); + if (!m_threadsDone.wait_for(lock, kTimeout, [this] { return m_threadsAtWork == 0; })) + { + return false; + } + } + for (std::thread& thread : m_threads) + { + thread.join(); + } + m_threads.clear(); + return true; + } + + // Stalls the sender thread and brings `thread` to its limit, with one line per buffer: the sender is stuck + // sending the thread's first buffer, eight buffers are waiting, and the tenth is full but still with the thread. + // The lines are Line(letter, 1) to Line(letter, 10). The writer has to have a buffer size of kOneLine. + void FillUp(RecordingThread& thread, char letter) + { + m_socket.HoldWrites(); + ASSERT_TRUE(thread.Record(Line(letter, 1))); + // Only once the sender has taken the first buffer is it certain that the next eight stay with the thread. + ASSERT_EQ(m_socket.WaitForHeldWrite(), Alone(Line(letter, 1))); + for (int number = 2; number <= kMaxWaiting + 2; ++number) + { + ASSERT_TRUE(thread.Record(Line(letter, number))); + } + } + + FakeSocketWriter m_socket; + FakeSocketWriter m_secondSocket; + + private: + std::vector> m_recordingThreads; + + std::mutex m_threadsMutex; + std::condition_variable m_threadsDone; + int m_threadsAtWork = 0; + std::vector m_threads; +}; + +} // namespace + +// ---------- What goes into a datagram ---------- + +TEST_F(BufferedWriterTest, SendsABufferWhenTheNextLineDoesNotFit) +{ + StartWriter(16); + + EXPECT_TRUE(buffered_writer::Record("aaaaa")); // 6 bytes with its newline + EXPECT_TRUE(buffered_writer::Record("bbbbb")); // 12 bytes in the buffer + EXPECT_TRUE(buffered_writer::Record("ccccc")); // would make 18 + + // The first two lines go out without the third, and without a timed flush. + ASSERT_TRUE(m_socket.WaitForDatagrams(1)); + EXPECT_EQ(m_socket.Sent(), (Datagrams{"aaaaa\nbbbbb\n"})); + + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent(), (Datagrams{"aaaaa\nbbbbb\n", "ccccc\n"})); +} + +TEST_F(BufferedWriterTest, SendsABufferAsSoonAsALineFillsIt) +{ + StartWriter(12); + + EXPECT_TRUE(buffered_writer::Record("aaaaa")); + EXPECT_TRUE(buffered_writer::Record("bbbbb")); // 12 bytes: the buffer is full + + // The buffer goes out without another line being recorded. + ASSERT_TRUE(m_socket.WaitForDatagrams(1)); + EXPECT_EQ(m_socket.Sent(), (Datagrams{"aaaaa\nbbbbb\n"})); + + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent().size(), 1u); +} + +TEST_F(BufferedWriterTest, SendsALineLongerThanTheBufferOnItsOwn) +{ + StartWriter(8); + const std::string longLine(20, 'x'); + const std::string otherLongLine(30, 'y'); + + // Into an empty buffer: the line is sent whole, as soon as it is recorded. + EXPECT_TRUE(buffered_writer::Record(longLine)); + ASSERT_TRUE(m_socket.WaitForDatagrams(1)); + EXPECT_EQ(m_socket.Sent(), (Datagrams{Alone(longLine)})); + + // Into a buffer that holds a line: that line goes out first, and the long line in a datagram of its own. + EXPECT_TRUE(buffered_writer::Record("ab")); + EXPECT_TRUE(buffered_writer::Record(otherLongLine)); + ASSERT_TRUE(m_socket.WaitForDatagrams(3)); + EXPECT_EQ(m_socket.Sent(), (Datagrams{Alone(longLine), "ab\n", Alone(otherLongLine)})); + + EXPECT_TRUE(buffered_writer::Record("cd")); + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent(), (Datagrams{Alone(longLine), "ab\n", Alone(otherLongLine), "cd\n"})); +} + +// ---------- Turning buffering on and off, and the flush interval ---------- + +TEST_F(BufferedWriterTest, RecordReturnsFalseWhileBufferingIsOff) +{ + RecordingThread& thread = NewThread(); + + // The caller then sends the line itself. + EXPECT_FALSE(buffered_writer::Record("before")); + EXPECT_FALSE(thread.Record("before")); + + StartWriter(64); + EXPECT_TRUE(buffered_writer::Record("this.thread")); + EXPECT_TRUE(thread.Record("another.thread")); + buffered_writer::Stop(); + + // Both threads still have the buffers of the writer that was turned off. + EXPECT_FALSE(buffered_writer::Record("after")); + EXPECT_FALSE(thread.Record("after")); + + EXPECT_EQ(m_socket.Sent(), (Datagrams{"this.thread\n", "another.thread\n"})); +} + +TEST_F(BufferedWriterTest, SendsAPartlyFilledBufferAfterTheFlushInterval) +{ + StartWriter(4096, std::chrono::milliseconds(20)); + + EXPECT_TRUE(buffered_writer::Record("first")); + ASSERT_TRUE(m_socket.WaitForDatagrams(1)); + EXPECT_TRUE(buffered_writer::Record("second")); + ASSERT_TRUE(m_socket.WaitForDatagrams(2)); + + EXPECT_EQ(m_socket.Sent(), (Datagrams{"first\n", "second\n"})); +} + +TEST_F(BufferedWriterTest, StopSendsWhatEveryThreadStillHolds) +{ + StartWriter(4096); + RecordingThread& other = NewThread(); + + EXPECT_TRUE(buffered_writer::Record("from.this.thread")); + ASSERT_TRUE(other.Record("from.another.thread")); + + // Neither buffer is full and no timed flush is due, so nothing is sent. + std::this_thread::sleep_for(kQuietTime); + EXPECT_TRUE(m_socket.Sent().empty()); + + // The other thread is still running. Its line is sent all the same. + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent(), (Datagrams{"from.this.thread\n", "from.another.thread\n"})); +} + +TEST_F(BufferedWriterTest, WorksWithAFlushIntervalThatIsOutOfRange) +{ + // Zero or less: the line is still flushed. + std::size_t sent = 0; + for (const std::chrono::milliseconds interval : {std::chrono::milliseconds(0), std::chrono::milliseconds(-5)}) + { + StartWriter(4096, interval); + EXPECT_TRUE(buffered_writer::Record("line")); + ASSERT_TRUE(m_socket.WaitForDatagrams(++sent)); + buffered_writer::Stop(); + } + + // The longest interval there is: no timed flush, and not one that is due at once because the time of the next + // flush has overflowed. + StartWriter(4096, std::chrono::milliseconds::max()); + EXPECT_TRUE(buffered_writer::Record("line")); + std::this_thread::sleep_for(kQuietTime); + EXPECT_EQ(m_socket.Sent().size(), sent); + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent().size(), sent + 1); +} + +// ---------- Threads ---------- + +class BufferedWriterPolicyTest : public BufferedWriterTest, public testing::WithParamInterface +{ +}; + +// Several threads record without pause. They produce buffers faster than the one sender thread sends them, so they +// keep reaching their limit of waiting buffers, and what the policy says happens then happens all the time. +TEST_P(BufferedWriterPolicyTest, KeepsEachThreadsLinesInOrder) +{ + constexpr std::size_t kBufferSize = 64; + constexpr int kThreads = 6; + constexpr int kLinesPerThread = 20000; + // A short flush interval, so that timed flushes take partly filled buffers while the threads hand over full + // ones. + StartWriter(kBufferSize, std::chrono::milliseconds(1), GetParam()); + + for (int thread = 0; thread < kThreads; ++thread) + { + RunOnNewThread( + [thread] + { + for (int number = 0; number < kLinesPerThread; ++number) + { + buffered_writer::Record(ThreadLine(thread, number)); + } + }); + } + ASSERT_TRUE(WaitForThreads()); + buffered_writer::Stop(); + + const Arrived arrived = Examine(m_socket.Sent(), kBufferSize, kThreads, kLinesPerThread); + EXPECT_EQ(arrived.tooLarge, 0); + EXPECT_EQ(arrived.notWholeLines, 0); + EXPECT_EQ(arrived.mixed, 0); + EXPECT_EQ(arrived.outOfOrder, 0); + if (GetParam() != FullBufferPolicy::Send) + { + // A recording thread sends only under the send policy. Otherwise every datagram comes from the sender + // thread, or from this thread, which sends what is left when it turns the writer off. + std::set senders; + for (const FakeSocketWriter::Datagram& datagram : m_socket.SentDatagrams()) + { + if (datagram.sentBy != std::this_thread::get_id()) + { + senders.insert(datagram.sentBy); + } + } + EXPECT_LE(senders.size(), 1u); + } + if (GetParam() == FullBufferPolicy::Drop) + { + // Every line was either sent or counted as dropped. + EXPECT_EQ(static_cast(arrived.total) + Log().DroppedLines(), + static_cast(kThreads * kLinesPerThread)); + } + else + { + for (int thread = 0; thread < kThreads; ++thread) + { + EXPECT_EQ(arrived.lines[static_cast(thread)], kLinesPerThread) << "thread " << thread; + } + } +} + +INSTANTIATE_TEST_SUITE_P(AllPolicies, BufferedWriterPolicyTest, + testing::Values(FullBufferPolicy::Block, FullBufferPolicy::Drop, FullBufferPolicy::Send), + [](const testing::TestParamInfo& info) + { return FullBufferPolicyToString(info.param); }); + +TEST_F(BufferedWriterTest, SendsWhatAThreadHoldsWhenTheThreadExits) +{ + StartWriter(4096); + RecordingThread& thread = NewThread(); + + ASSERT_TRUE(thread.Record("from.a.thread")); + EXPECT_TRUE(m_socket.Sent().empty()); + ASSERT_TRUE(thread.Exit()); + + // The buffer is not full, no timed flush is due and the writer is still on: the line is sent because the thread + // has gone. + ASSERT_TRUE(m_socket.WaitForDatagrams(1)); + EXPECT_EQ(m_socket.Sent(), (Datagrams{"from.a.thread\n"})); +} + +TEST_F(BufferedWriterTest, AnExitingThreadNeitherSendsNorWaits) +{ + // The send policy is the one under which a recording thread may send at all. + StartWriter(kOneLine, kNoTimedFlush, FullBufferPolicy::Send); + RecordingThread& thread = NewThread(); + const std::thread::id threadId = thread.Id(); + ASSERT_NO_FATAL_FAILURE(FillUp(thread, 'a')); + + // The thread exits while nothing can be sent. It does not wait for that to change, and it sends nothing itself. + ASSERT_TRUE(thread.Exit()); + EXPECT_TRUE(m_socket.Sent().empty()); + + // The sender thread sends what the thread had buffered, in order, including the buffer that was not handed over. + m_socket.ReleaseWrites(); + ASSERT_TRUE(m_socket.WaitForDatagrams(10)); + EXPECT_EQ(m_socket.Sent(), EachAlone('a', 1, 10)); + for (const FakeSocketWriter::Datagram& datagram : m_socket.SentDatagrams()) + { + EXPECT_NE(datagram.sentBy, threadId); + } +} + +TEST_F(BufferedWriterTest, KeepsLinesRecordedDuringThreadExitInOrder) +{ + StartWriter(kOneLine, kNoTimedFlush, FullBufferPolicy::Send); + m_socket.HoldWrites(); + RecordingThread& thread = NewThread(); + const std::thread::id threadId = thread.Id(); + // Static, so that it is still there for RecordLate() if the thread outlives this test. + static int round; + round = 0; + ASSERT_TRUE(thread.Do([] { return pthread_setspecific(LateRecorderKey(), &round) == 0; })); + + // The sender is stuck sending the first line, so the next two wait with the thread. + ASSERT_TRUE(thread.Record(Line('a', 1))); + ASSERT_EQ(m_socket.WaitForHeldWrite(), Alone(Line('a', 1))); + ASSERT_TRUE(thread.Record(Line('a', 2))); + ASSERT_TRUE(thread.Record(Line('a', 3))); + + // The thread exits, and RecordLate() records more lines on its way out. However many they are, they do not make + // the thread wait, although nothing can be sent, and the thread does not send them itself. + ASSERT_TRUE(thread.Exit()); + EXPECT_EQ(round, 4); + EXPECT_TRUE(m_socket.Sent().empty()); + + // The late lines follow the lines that the thread recorded before, in the order in which they were recorded. + // They are all sent without a timed flush, including the last one, which did not fill a buffer: none is left + // behind in a buffer that nobody lets go of. + m_socket.ReleaseWrites(); + Datagrams expected = EachAlone('a', 1, 3); + const Datagrams late = EachAlone('z', 1, kLastLateLineOfRound2); + expected.insert(expected.end(), late.begin(), late.end()); + expected.push_back(Alone(kLineOfLastRound)); + ASSERT_TRUE(m_socket.WaitForDatagrams(expected.size())); + EXPECT_EQ(m_socket.Sent(), expected); + for (const FakeSocketWriter::Datagram& datagram : m_socket.SentDatagrams()) + { + EXPECT_NE(datagram.sentBy, threadId); + } +} + +TEST_F(BufferedWriterTest, LosesNothingFromShortLivedThreads) +{ + constexpr std::size_t kBufferSize = 64; + constexpr int kThreads = 300; + constexpr int kLinesPerThread = 3; + constexpr int kThreadsAtATime = 10; + StartWriter(kBufferSize); + + // A few threads at a time, so that threads come and go while the sender thread is at work. + for (int first = 0; first < kThreads; first += kThreadsAtATime) + { + for (int thread = first; thread < first + kThreadsAtATime; ++thread) + { + RunOnNewThread( + [thread] + { + for (int number = 0; number < kLinesPerThread; ++number) + { + buffered_writer::Record(ThreadLine(thread, number)); + } + }); + } + ASSERT_TRUE(WaitForThreads()); + } + + // The lines of a thread do not fill a buffer, no timed flush is due and the writer is still on: they are sent + // because their threads have gone. + ASSERT_TRUE(m_socket.WaitForLines(static_cast(kThreads * kLinesPerThread))); + const Arrived arrived = Examine(m_socket.Sent(), kBufferSize, kThreads, kLinesPerThread); + EXPECT_EQ(arrived.tooLarge, 0); + EXPECT_EQ(arrived.notWholeLines, 0); + EXPECT_EQ(arrived.mixed, 0); + EXPECT_EQ(arrived.outOfOrder, 0); + EXPECT_EQ(arrived.total, kThreads * kLinesPerThread); +} + +// Ten threads that record lines of exactly the buffer size. An earlier writer could deadlock in this case; see the +// test TenThreadsBufferSize10Messages in test_writer.cpp, which is commented out. +TEST_F(BufferedWriterTest, TenThreadsWithLinesOfExactlyTheBufferSize) +{ + constexpr std::size_t kBufferSize = 21; + constexpr int kThreads = 10; + constexpr int kLinesPerThread = 10; + StartWriter(kBufferSize); + + for (int thread = 0; thread < kThreads; ++thread) + { + RunOnNewThread( + [thread] + { + const std::string line = fmt::format("c:ctr{:06d}:1.000000", thread); // 21 bytes with its newline + for (int number = 0; number < kLinesPerThread; ++number) + { + buffered_writer::Record(line); + } + }); + } + ASSERT_TRUE(WaitForThreads()); + + // Every line fills a buffer, so each is sent as a datagram of its own, without a timed flush. + ASSERT_TRUE(m_socket.WaitForDatagrams(static_cast(kThreads * kLinesPerThread))); + std::map arrived; + for (const std::string& datagram : m_socket.Sent()) + { + EXPECT_EQ(datagram.size(), kBufferSize); + ++arrived[datagram]; + } + EXPECT_EQ(arrived.size(), static_cast(kThreads)); + for (int thread = 0; thread < kThreads; ++thread) + { + EXPECT_EQ(arrived[fmt::format("c:ctr{:06d}:1.000000\n", thread)], kLinesPerThread) << "thread " << thread; + } +} + +// ---------- What a thread does when it is out of buffer space ---------- + +TEST_F(BufferedWriterTest, BlockPolicyMakesAThreadWaitForTheSender) +{ + StartWriter(kOneLine, kNoTimedFlush, FullBufferPolicy::Block); + RecordingThread& thread = NewThread(); + ASSERT_NO_FATAL_FAILURE(FillUp(thread, 'a')); + + // The eleventh line needs a ninth waiting buffer, so the thread waits for as long as the sender is stuck. + std::future eleventh = thread.StartRecording(Line('a', 11)); + EXPECT_EQ(eleventh.wait_for(kQuietTime), std::future_status::timeout); + + // Once the sender takes a buffer, the thread carries on. No line is lost. + m_socket.ReleaseWrites(); + ASSERT_EQ(eleventh.wait_for(kTimeout), std::future_status::ready); + EXPECT_TRUE(eleventh.get()); + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent(), EachAlone('a', 1, 11)); +} + +TEST_F(BufferedWriterTest, BlockPolicyThreadIsReleasedByShutdown) +{ + StartWriter(kOneLine, kNoTimedFlush, FullBufferPolicy::Block); + RecordingThread& thread = NewThread(); + ASSERT_NO_FATAL_FAILURE(FillUp(thread, 'a')); + std::future eleventh = thread.StartRecording(Line('a', 11)); + EXPECT_EQ(eleventh.wait_for(kQuietTime), std::future_status::timeout); + + // Shutting down waits for the sender thread, whose send is still held, so it runs on a thread of its own. Until + // the send is let go below, nothing here may end the test early: the destructor of `stopped` waits for it. + std::future stopped = std::async(std::launch::async, [] { buffered_writer::Stop(); }); + + // The waiting thread is released while the sender is still stuck. Its line is taken, and dropped. + const bool released = eleventh.wait_for(kTimeout) == std::future_status::ready; + EXPECT_TRUE(released); + EXPECT_TRUE(m_socket.Sent().empty()); + + m_socket.ReleaseWrites(); + EXPECT_EQ(stopped.wait_for(kTimeout), std::future_status::ready); + if (released) + { + EXPECT_TRUE(eleventh.get()); + } + EXPECT_EQ(m_socket.Sent(), EachAlone('a', 1, 10)); +} + +TEST_F(BufferedWriterTest, DropPolicyDropsNewLinesAndKeepsBufferedOnes) +{ + StartWriter(kOneLine, kNoTimedFlush, FullBufferPolicy::Drop); + RecordingThread& thread = NewThread(); + ASSERT_NO_FATAL_FAILURE(FillUp(thread, 'a')); + + // The thread is out of buffer space and the sender is stuck: three more lines are dropped, and the thread does + // not wait. + for (int number = 11; number <= 13; ++number) + { + ASSERT_TRUE(thread.Record(Line('a', number))); + } + + // The ten lines that were buffered are sent, and the number of dropped lines is logged. + m_socket.ReleaseWrites(); + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent(), EachAlone('a', 1, 10)); + EXPECT_EQ(Log().DroppedLines(), 3u); +} + +// Under the send policy, a thread that needs a ninth waiting buffer sends its oldest one itself. With one thread and +// ten lines, that only happens if the thread has recorded all ten before the sender thread takes the first buffer, +// and a test cannot make the sender thread be late. What it can do is record right after the writer is started, when +// the new sender thread is not running yet, and start over with a new writer if the sender thread was in time after +// all. (With several threads that record without pause, as in KeepsEachThreadsLinesInOrder, threads send for +// themselves all the time. This test is about one such send: which buffer it is, and where it ends up.) +TEST_F(BufferedWriterTest, SendPolicyLetsAThreadSendItsOwnOldestBuffer) +{ + constexpr int kAttempts = 1000; + // Ten lines of one buffer each: eight buffers wait, the ninth is still with the thread, and the tenth line needs + // a ninth waiting buffer. + Datagrams lines; + for (int number = 1; number <= kMaxWaiting + 2; ++number) + { + lines.push_back(Line('a', number)); + } + const std::thread::id thisThread = std::this_thread::get_id(); + + bool sentByThisThread = false; + for (int attempt = 0; attempt < kAttempts && !sentByThisThread; ++attempt) + { + m_socket.Clear(); + StartWriter(kOneLine, kNoTimedFlush, FullBufferPolicy::Send); + for (const std::string& line : lines) + { + buffered_writer::Record(line); + } + // What this thread has sent by now, it has sent while recording. After this, it also sends what is left + // when it turns the writer off. + const std::vector sentWhileRecording = m_socket.SentDatagrams(); + buffered_writer::Stop(); + + for (const FakeSocketWriter::Datagram& datagram : sentWhileRecording) + { + if (datagram.sentBy == thisThread) + { + sentByThisThread = true; + // It is the oldest buffer, and the first datagram to be sent at all. + EXPECT_EQ(datagram.data, Alone(lines.front())); + EXPECT_EQ(sentWhileRecording.front().sentBy, thisThread); + } + } + // Whoever sent which buffer, every line arrives once, and in order. + ASSERT_EQ(m_socket.Sent(), EachAlone('a', 1, kMaxWaiting + 2)); + } + EXPECT_TRUE(sentByThisThread) << "the sender thread took a buffer first in each of " << kAttempts << " attempts"; +} + +// ---------- The sender thread, and the life of the writer ---------- + +// A thread that records without pause always has buffers waiting. The sender thread takes at most nine buffers of a +// thread before it goes on to the next thread, so that such a thread does not keep it from the others. +TEST_F(BufferedWriterTest, SenderLeavesABusyThreadAfterNineBuffers) +{ + StartWriter(kOneLine); + m_socket.HoldWrites(); + RecordingThread& other = NewThread(); + RecordingThread& busy = NewThread(); + + // The other thread records first, so the sender thread visits it first in each of its rounds. + ASSERT_TRUE(other.Record(Line('b', 0))); + ASSERT_EQ(m_socket.WaitForHeldWrite(), Alone(Line('b', 0))); + + // The busy thread fills up: eight buffers wait, and the ninth is still with it. + for (int number = 1; number <= kMaxWaiting + 1; ++number) + { + ASSERT_TRUE(busy.Record(Line('a', number))); + } + // The sender gets on to the busy thread. Then the other thread has something to send again. + m_socket.AllowWrites(1); + ASSERT_EQ(m_socket.WaitForHeldWrite(), Alone(Line('a', 1))); + ASSERT_TRUE(other.Record(Line('b', 1))); + + // Each time the sender has taken a buffer of the busy thread, that thread hands over another one, so it never + // runs out of waiting buffers. + for (int taken = 1; taken <= kMaxWaiting + 1; ++taken) + { + ASSERT_EQ(m_socket.WaitForHeldWrite(), Alone(Line('a', taken))); + ASSERT_TRUE(busy.Record(Line('a', kMaxWaiting + 1 + taken))); + m_socket.AllowWrites(1); + } + + // After nine buffers, the sender leaves the busy thread, which has eight more waiting, for the other thread. + ASSERT_EQ(m_socket.WaitForHeldWrite(), Alone(Line('b', 1))); + + m_socket.ReleaseWrites(); + buffered_writer::Stop(); + Datagrams expected{Alone(Line('b', 0))}; + for (int number = 1; number <= kMaxWaiting + 1; ++number) + { + expected.push_back(Alone(Line('a', number))); + } + expected.push_back(Alone(Line('b', 1))); + for (int number = kMaxWaiting + 2; number <= 2 * (kMaxWaiting + 1); ++number) + { + expected.push_back(Alone(Line('a', number))); + } + EXPECT_EQ(m_socket.Sent(), expected); +} + +// A thread is at its limit: the sender thread is stuck sending its first buffer, eight more are waiting, and the +// tenth is full but still with the thread. Once the sender thread can send again, it is the sender thread that +// sends the waiting buffers, oldest first. +TEST_F(BufferedWriterTest, SenderThreadDrainsAThreadThatIsAtItsLimit) +{ + StartWriter(kOneLine); + RecordingThread& thread = NewThread(); + ASSERT_NO_FATAL_FAILURE(FillUp(thread, 'a')); + EXPECT_TRUE(m_socket.Sent().empty()); + + // The buffer that was stuck and the eight that were waiting are sent. The tenth is not: the thread has not + // handed it over, no timed flush is due, and the writer is still on. + m_socket.ReleaseWrites(); + ASSERT_TRUE(m_socket.WaitForDatagrams(kMaxWaiting + 1)); + EXPECT_EQ(m_socket.Sent(), EachAlone('a', 1, kMaxWaiting + 1)); + + // The thread's next line makes it hand the tenth buffer over, and fills a buffer itself. Both are sent. + ASSERT_TRUE(thread.Record(Line('a', kMaxWaiting + 3))); + ASSERT_TRUE(m_socket.WaitForDatagrams(kMaxWaiting + 3)); + EXPECT_EQ(m_socket.Sent(), EachAlone('a', 1, kMaxWaiting + 3)); + + // All of them were sent by one thread, which is neither the thread that recorded them nor this one, which has + // not turned the writer off yet: the sender thread. + const std::vector datagrams = m_socket.SentDatagrams(); + const std::thread::id senderThread = datagrams.front().sentBy; + EXPECT_NE(senderThread, thread.Id()); + EXPECT_NE(senderThread, std::this_thread::get_id()); + for (const FakeSocketWriter::Datagram& datagram : datagrams) + { + EXPECT_EQ(datagram.sentBy, senderThread); + } +} + +TEST_F(BufferedWriterTest, SenderKeepsGoingAfterAFailedSend) +{ + StartWriter(kOneLine); + m_socket.FailNextWrite(); + + EXPECT_TRUE(buffered_writer::Record(Line('a', 1))); // the send of this one throws + EXPECT_TRUE(buffered_writer::Record(Line('a', 2))); + + ASSERT_TRUE(m_socket.WaitForDatagrams(1)); + EXPECT_EQ(m_socket.FailedWrites(), 1u); + EXPECT_EQ(m_socket.Sent(), (Datagrams{Alone(Line('a', 2))})); +} + +TEST_F(BufferedWriterTest, ThreadGetsNewBuffersAfterARestart) +{ + RecordingThread& thread = NewThread(); + + StartWriter(4096); + EXPECT_TRUE(buffered_writer::Record("this.thread.first")); + EXPECT_TRUE(thread.Record("another.thread.first")); + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent(), (Datagrams{"this.thread.first\n", "another.thread.first\n"})); + + // Both threads still have the buffers of the first writer. What they record now goes to the second one. + buffered_writer::Start(&m_secondSocket, 4096, kNoTimedFlush, FullBufferPolicy::Block); + EXPECT_TRUE(buffered_writer::Record("this.thread.second")); + EXPECT_TRUE(thread.Record("another.thread.second")); + buffered_writer::Stop(); + EXPECT_EQ(m_secondSocket.Sent(), (Datagrams{"this.thread.second\n", "another.thread.second\n"})); + EXPECT_EQ(m_socket.Sent().size(), 2u); +} + +// What the Writer does when the process exits. +TEST_F(BufferedWriterTest, ShutdownAtExitSendsWhatIsBufferedAndDropsLaterLines) +{ + StartWriter(4096); + RecordingThread& thread = NewThread(); + + EXPECT_TRUE(buffered_writer::Record("this.thread.before")); + ASSERT_TRUE(thread.Record("another.thread.before")); + buffered_writer::ShutdownAtExit(); + EXPECT_EQ(m_socket.Sent(), (Datagrams{"this.thread.before\n", "another.thread.before\n"})); + + // Buffering stays on, so the lines are taken here, and not handed back to a caller that would send them through + // a socket writer that is being destroyed. They are dropped. + RecordingThread& newThread = NewThread(); + EXPECT_TRUE(buffered_writer::Record("this.thread.after")); + EXPECT_TRUE(thread.Record("another.thread.after")); + EXPECT_TRUE(newThread.Record("new.thread.after")); + + // A thread can still exit, which lets go of buffers that nobody is going to send any more. + EXPECT_TRUE(thread.Exit()); + buffered_writer::Stop(); + EXPECT_EQ(m_socket.Sent().size(), 2u); +} diff --git a/libs/writer/writer_wrapper/test_writer.cpp b/libs/writer/writer_wrapper/test_writer.cpp index 5da2b89..9c17541 100644 --- a/libs/writer/writer_wrapper/test_writer.cpp +++ b/libs/writer/writer_wrapper/test_writer.cpp @@ -10,7 +10,11 @@ #include #include #include +#include #include +#include +#include +#include #include #include #include @@ -20,6 +24,46 @@ using namespace spectator; +// A Unix datagram socket that a test reads itself. A datagram is in the queue of the socket as soon as the send has +// returned, so a test finds what was sent without waiting for it. +class DatagramReceiver +{ + public: + explicit DatagramReceiver(const std::string& socketPath) : m_socketPath(socketPath), m_socket(m_ioContext) + { + std::filesystem::remove(m_socketPath); + m_socket.open(); + m_socket.bind(DatagramProtocol::endpoint(m_socketPath)); + m_socket.non_blocking(true); + } + + ~DatagramReceiver() + { + std::error_code ignored; + std::filesystem::remove(m_socketPath, ignored); + } + + // The next datagram in the queue, or nothing if the queue is empty. + std::optional Receive() + { + std::array buffer{}; + boost::system::error_code ec; + const auto bytesRead = m_socket.receive(boost::asio::buffer(buffer), 0, ec); + if (ec) + { + return std::nullopt; + } + return std::string(buffer.data(), bytesRead); + } + + private: + using DatagramProtocol = boost::asio::local::datagram_protocol; + + std::string m_socketPath; + boost::asio::io_context m_ioContext; + DatagramProtocol::socket m_socket; +}; + // Runs in the death test child: records one meter with a buffer far larger than the line, then exits normally, // which destroys the Writer singleton. static void RecordOneCounterAndExit(const std::string& socketPath) @@ -30,9 +74,10 @@ static void RecordOneCounterAndExit(const std::string& socketPath) std::exit(0); } -// The buffered writer sends only full buffers, so a short-lived process may never fill one. The Writer is a -// process-wide singleton, so the child process owns it here. Death test suites run before all other tests, so the -// parent has not initialized the Writer or started any threads when it forks. +// A line stays in the buffer of its thread until the buffer is full or the flush interval has passed, and a +// short-lived process may exit before either. The Writer is a process-wide singleton, so the child process owns it +// here. Death test suites run before all other tests, so the parent has not initialized the Writer or started any +// threads when it forks. TEST(WriterShutdownDeathTest, SendsBufferedLinesAtExit) { using DatagramProtocol = boost::asio::local::datagram_protocol; @@ -52,6 +97,97 @@ TEST(WriterShutdownDeathTest, SendsBufferedLinesAtExit) EXPECT_EQ(std::string(buffer.data(), bytesRead), "c:shutdown.counter:1.000000\n"); } +// Runs in the death test child: another thread records one meter and is still running when the process exits +// normally. +static void RecordOnAnotherThreadAndExit(const std::string& socketPath) +{ + // A flush interval that is longer than the test, so the line is still in the buffer of its thread at exit. + WriterTestHelper::InitializeWriter(WriterType::Unix, socketPath, 0, 4096, std::chrono::hours(1)); + std::promise recorded; + std::thread( + [&recorded] + { + Counter counter(MeterId("running.thread.counter")); + counter.Increment(); + recorded.set_value(); + for (;;) + { + std::this_thread::sleep_for(std::chrono::hours(1)); + } + }) + .detach(); + recorded.get_future().wait(); + std::exit(0); +} + +// A thread that is still running when the process exits has not handed its buffer over. The Writer sends it. +TEST(WriterShutdownDeathTest, SendsTheLinesOfRunningThreadsAtExit) +{ + const std::string socketPath = "/tmp/test_uds_shutdown_threads_socket"; + DatagramReceiver receiver(socketPath); + + EXPECT_EXIT(RecordOnAnotherThreadAndExit(socketPath), testing::ExitedWithCode(0), ""); + + EXPECT_EQ(receiver.Receive(), "c:running.thread.counter:1.000000\n"); + EXPECT_EQ(receiver.Receive(), std::nullopt); +} + +// Runs in the death test child: threads record without pause while the process exits normally. +static void ExitWhileThreadsAreRecording() +{ + WriterTestHelper::InitializeWriter(WriterType::Memory, "", 0, 64); + for (int i = 0; i < 4; i++) + { + std::thread( + [i] + { + Counter counter(MeterId(fmt::format("exit.counter.{}", i))); + for (;;) + { + counter.Increment(); + } + }) + .detach(); + } + std::this_thread::sleep_for(std::chrono::milliseconds(20)); + std::exit(0); +} + +// The Writer is destroyed while threads are recording. A thread that waits for buffer space is released, and what +// the threads record from then on is dropped: none of it may crash the process or keep it from exiting. +TEST(WriterShutdownDeathTest, ExitsCleanlyWhileThreadsAreRecording) +{ + EXPECT_EXIT(ExitWhileThreadsAreRecording(), testing::ExitedWithCode(0), ""); +} + +// The Writer can be initialized again, which replaces the socket writer. A buffered writer from the earlier +// initialization first sends what it still holds. +TEST(WriterReinitializeTest, SendsWhatThePreviousWriterBuffered) +{ + const std::string socketPath = "/tmp/test_uds_reinitialize_socket"; + DatagramReceiver receiver(socketPath); + // Longer than the test, so nothing is sent on a timer. + const auto noTimedFlush = std::chrono::hours(1); + + WriterTestHelper::InitializeWriter(WriterType::Unix, socketPath, 0, 4096, noTimedFlush); + Counter counter(MeterId("reinitialize.counter")); + counter.Increment(); + EXPECT_EQ(receiver.Receive(), std::nullopt); + + // Buffered again: the first writer sends the line before it is replaced. + WriterTestHelper::InitializeWriter(WriterType::Unix, socketPath, 0, 4096, noTimedFlush); + EXPECT_EQ(receiver.Receive(), "c:reinitialize.counter:1.000000\n"); + counter.Increment(2); + EXPECT_EQ(receiver.Receive(), std::nullopt); + + // Without buffering: the second writer sends its line, and from then on every line is sent as it is recorded. + WriterTestHelper::InitializeWriter(WriterType::Unix, socketPath); + EXPECT_EQ(receiver.Receive(), "c:reinitialize.counter:2.000000\n"); + counter.Increment(3); + EXPECT_EQ(receiver.Receive(), "c:reinitialize.counter:3.000000\n"); + EXPECT_EQ(receiver.Receive(), std::nullopt); +} + // Test fixture for UDS Writer tests class WriterWrapperUDSWriterTest : public testing::Test { diff --git a/libs/writer/writer_wrapper/writer_test_helper.h b/libs/writer/writer_wrapper/writer_test_helper.h index 972a00b..afd8e2e 100644 --- a/libs/writer/writer_wrapper/writer_test_helper.h +++ b/libs/writer/writer_wrapper/writer_test_helper.h @@ -2,6 +2,9 @@ #include +#include +#include + namespace spectator { /** @@ -14,9 +17,12 @@ class WriterTestHelper { public: // Initialize the Writer for testing purposes - static void InitializeWriter(WriterType type, const std::string& param = "", int port = 0, unsigned int bufferSize = 0) + static void InitializeWriter(WriterType type, const std::string& param = "", int port = 0, + unsigned int bufferSize = 0, + std::chrono::milliseconds flushInterval = BufferingDefaults::FlushInterval, + FullBufferPolicy fullBufferPolicy = BufferingDefaults::FullBuffer) { - Writer::Initialize(type, param, port, bufferSize); + Writer::Initialize(type, param, port, bufferSize, flushInterval, fullBufferPolicy); } // Get the Writer's implementation for testing purposes diff --git a/spectator/test_registry.cpp b/spectator/test_registry.cpp index 265a16d..ed7b61c 100644 --- a/spectator/test_registry.cpp +++ b/spectator/test_registry.cpp @@ -1,7 +1,10 @@ #include +#include #include #include +#include +#include #include #include @@ -71,6 +74,27 @@ TEST(RegistryTest, AgeGaugeWithId) ParseProtocolLine(memoryWriter->LastLine()).value().to_string()); } +// With a buffer size in the config, lines are held back until a buffer is full, the flush interval has passed or +// buffering is turned off. The flush interval comes from the config too: with the default interval, the first line +// would be sent on its own while this test sleeps. +TEST(RegistryTest, BufferedWriter) +{ + auto config = Config(WriterConfig(WriterTypes::Memory, 4096, std::chrono::hours(1))); + auto r = Registry(config); + auto c = r.CreateCounter("counter"); + auto memoryWriter = static_cast(WriterTestHelper::GetImpl()); + + c.Increment(); + std::this_thread::sleep_for(BufferingDefaults::FlushInterval + std::chrono::milliseconds(300)); + c.Increment(2); + EXPECT_TRUE(memoryWriter->IsEmpty()); + + // Turning buffering off sends what is buffered and stops the sender thread, so the memory writer can be read. + buffered_writer::Stop(); + ASSERT_EQ(memoryWriter->GetMessages().size(), 1u); + EXPECT_EQ("c:counter:1.000000\nc:counter:2.000000\n", memoryWriter->GetMessages().at(0)); +} + TEST(RegistryTest, Counter) { auto config = Config(WriterConfig(WriterTypes::Memory)); From f18877d558e7db1b7b56811326bfac168e2431da Mon Sep 17 00:00:00 2001 From: Everett Badeaux Date: Fri, 6 Nov 2026 15:10:00 -0600 Subject: [PATCH 3/5] Update Flush Policy --- README.md | 10 +++++----- libs/writer/README.md | 10 +++++----- libs/writer/writer_config/test_writer_config.cpp | 11 ++++++++++- libs/writer/writer_config/writer_config.cpp | 6 ++++-- libs/writer/writer_config/writer_config.h | 5 +++-- libs/writer/writer_types/include/writer_types.h | 10 +++++++--- libs/writer/writer_wrapper/buffered_writer.cpp | 2 ++ performance_tests/writer_performance_test/README.md | 4 ++-- performance_tests/writer_performance_test/scenario.h | 2 +- spectator/test_registry.cpp | 3 ++- 10 files changed, 41 insertions(+), 22 deletions(-) diff --git a/README.md b/README.md index 07d50e8..4d35491 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,7 @@ For high-volume use cases, and for recording from several threads, turn on buffe bytes to the `WriterConfig` constructor: ```cpp -// 4096-byte buffers, with the default flush interval (1 second) and full-buffer policy (block) +// 4096-byte buffers, with the default flush interval (3 seconds) and full-buffer policy (block) WriterConfig writerConfig(WriterTypes::Unix, 4096); // Or: send what is buffered every 250 ms, and drop lines instead of waiting when a thread is out of buffer space @@ -32,10 +32,10 @@ sends the buffers to the sidecar. Updating a meter normally sends nothing, and t * **Order.** The lines of one thread are sent in the order in which that thread recorded them. Lines of different threads are sent in no particular order. * **Flush interval.** The sender thread collects and sends what every thread has buffered once per flush interval, - so a line never waits for its thread to fill a buffer. The default is 1 second: a line from a thread that records - rarely, or that records a burst and then goes idle, reaches the sidecar about a second later at most, at the cost - of at most one small datagram per recording thread per second. Pass a different interval as the third argument of - the `WriterConfig` constructor. It must be positive. + so a line never waits for its thread to fill a buffer. The default is 3 seconds: a line from a thread that records + rarely, or that records a burst and then goes idle, reaches the sidecar about three seconds later at most, at the + cost of at most one small datagram per recording thread every three seconds. Pass a different interval as the third argument of + the `WriterConfig` constructor. It must be positive and at most 30 seconds. * **Full-buffer policy.** A thread may have up to 8 full buffers waiting for the sender, so a recording thread holds at most 9 buffers: those 8 and the one it is filling. What a thread does when it needs more than that is the fourth argument of the `WriterConfig` constructor: diff --git a/libs/writer/README.md b/libs/writer/README.md index a213992..de7e4d4 100644 --- a/libs/writer/README.md +++ b/libs/writer/README.md @@ -83,17 +83,17 @@ With a buffer size greater than 0 in the `WriterConfig`, the wrapper buffers per | `WriterConfig` argument | Default | Meaning | |-------------------------|--------------------------|---------------------------------------------------------------------| | `bufferSize` | `0` (no buffering) | Size of each thread's buffer in bytes, and the largest datagram | -| `flushInterval` | `1000 ms` | How often the sender collects and sends what the threads buffered | +| `flushInterval` | `3000 ms` | How often the sender collects and sends what the threads buffered | | `fullBufferPolicy` | `FullBufferPolicy::Block`| What a thread does when it is out of buffer space | ### Flush interval A thread that records rarely, or that records a burst and then goes idle, would otherwise hold its lines until it fills a buffer or exits. Once per flush interval the sender thread therefore also takes the buffers that are still -being filled, after any full buffers of the same thread, and sends them. The default of 1 second keeps such lines -about a second behind at most, and costs at most one wake-up of the sender thread and one small datagram per -recording thread per second. The interval must be positive; the `WriterConfig` constructor throws a -`std::runtime_error` otherwise. +being filled, after any full buffers of the same thread, and sends them. The default of 3 seconds keeps such lines +about three seconds behind at most, and costs at most one wake-up of the sender thread and one small datagram per +recording thread every three seconds. The interval must be positive and at most 30 seconds +(`BufferingDefaults::MaxFlushInterval`); the `WriterConfig` constructor throws a `std::runtime_error` otherwise. ### Full-buffer policy diff --git a/libs/writer/writer_config/test_writer_config.cpp b/libs/writer/writer_config/test_writer_config.cpp index 36d6901..e16d95d 100644 --- a/libs/writer/writer_config/test_writer_config.cpp +++ b/libs/writer/writer_config/test_writer_config.cpp @@ -114,7 +114,7 @@ TEST_F(WriterConfigTest, BufferingDefaults) // With a buffer size only { const WriterConfig config(WriterTypes::UDP, 2048); - EXPECT_EQ(config.GetFlushInterval(), std::chrono::milliseconds(1000)); + EXPECT_EQ(config.GetFlushInterval(), std::chrono::milliseconds(3000)); EXPECT_EQ(config.GetFullBufferPolicy(), FullBufferPolicy::Block); } } @@ -154,8 +154,17 @@ TEST_F(WriterConfigTest, InvalidFlushInterval) EXPECT_THROW({ WriterConfig config(WriterTypes::UDP, 2048, std::chrono::milliseconds(0)); }, std::runtime_error); EXPECT_THROW({ WriterConfig config(WriterTypes::UDP, 2048, std::chrono::milliseconds(-1)); }, std::runtime_error); + // At most 30 seconds + EXPECT_THROW({ WriterConfig config(WriterTypes::UDP, 2048, std::chrono::milliseconds(30001)); }, std::runtime_error); + EXPECT_THROW({ WriterConfig config(WriterTypes::UDP, 2048, std::chrono::hours(1)); }, std::runtime_error); + { + const WriterConfig config(WriterTypes::UDP, 2048, std::chrono::seconds(30)); + EXPECT_EQ(config.GetFlushInterval(), BufferingDefaults::MaxFlushInterval); + } + // Without buffering there is nothing to flush, so the interval is not checked EXPECT_NO_THROW({ WriterConfig config(WriterTypes::UDP, 0, std::chrono::milliseconds(0)); }); + EXPECT_NO_THROW({ WriterConfig config(WriterTypes::UDP, 0, std::chrono::hours(1)); }); } TEST_F(WriterConfigTest, InvalidWriterType) diff --git a/libs/writer/writer_config/writer_config.cpp b/libs/writer/writer_config/writer_config.cpp index 9ea3ff0..c320337 100644 --- a/libs/writer/writer_config/writer_config.cpp +++ b/libs/writer/writer_config/writer_config.cpp @@ -7,7 +7,7 @@ namespace spectator { struct WriterConfigConstants { static constexpr auto RuntimeErrorMessage = "Invalid writer type: "; - static constexpr auto FlushIntervalErrorMessage = "Invalid flush interval, it must be positive: "; + static constexpr auto FlushIntervalErrorMessage = "Invalid flush interval, it must be positive and at most "; }; std::pair GetWriterConfigFromString(const std::string& type) @@ -55,9 +55,11 @@ WriterConfig::WriterConfig(const std::string& type, const unsigned int bufferSiz : WriterConfig(type) // Constructor delegation { // The flush interval only matters when buffering is enabled. - if (bufferSize > 0 && flushInterval <= std::chrono::milliseconds::zero()) + if (bufferSize > 0 && + (flushInterval <= std::chrono::milliseconds::zero() || flushInterval > BufferingDefaults::MaxFlushInterval)) { throw std::runtime_error(WriterConfigConstants::FlushIntervalErrorMessage + + std::to_string(BufferingDefaults::MaxFlushInterval.count()) + " ms: " + std::to_string(flushInterval.count()) + " ms"); } diff --git a/libs/writer/writer_config/writer_config.h b/libs/writer/writer_config/writer_config.h index ea928f2..9917781 100644 --- a/libs/writer/writer_config/writer_config.h +++ b/libs/writer/writer_config/writer_config.h @@ -16,8 +16,9 @@ class WriterConfig // Enables buffering when bufferSize is greater than 0. Each thread that records meters then fills its own buffer // of at most bufferSize bytes, and one sender thread sends the buffers. // flushInterval: how often the sender thread collects and sends what the threads have buffered, so a line - // never waits longer than this for its thread to fill a buffer. Must be positive when - // buffering is enabled; std::runtime_error otherwise. + // never waits longer than this for its thread to fill a buffer. Must be positive and at + // most BufferingDefaults::MaxFlushInterval (30 seconds) when buffering is enabled; + // std::runtime_error otherwise. // fullBufferPolicy: what a thread does when it has no buffer space left (see FullBufferPolicy). WriterConfig(const std::string& type, unsigned int bufferSize, std::chrono::milliseconds flushInterval = BufferingDefaults::FlushInterval, diff --git a/libs/writer/writer_types/include/writer_types.h b/libs/writer/writer_types/include/writer_types.h index 5075272..12e99da 100644 --- a/libs/writer/writer_types/include/writer_types.h +++ b/libs/writer/writer_types/include/writer_types.h @@ -37,9 +37,13 @@ enum class FullBufferPolicy struct BufferingDefaults { // The sender thread collects and sends what every thread has buffered at least this often, so a line never - // waits for its thread to fill a buffer. One second keeps a line from a quiet thread at most about a second - // behind, and costs at most one wake-up of the sender and one small datagram per recording thread per second. - static constexpr std::chrono::milliseconds FlushInterval{1000}; + // waits for its thread to fill a buffer. Three seconds keep a line from a quiet thread at most about three + // seconds behind, and cost at most one wake-up of the sender and one small datagram per recording thread every + // three seconds. + static constexpr std::chrono::milliseconds FlushInterval{std::chrono::seconds{3}}; + // The longest flush interval that a WriterConfig accepts. A longer one would hold the lines of a quiet thread + // back for longer than any consumer of the metrics expects. + static constexpr std::chrono::milliseconds MaxFlushInterval{std::chrono::seconds{30}}; static constexpr FullBufferPolicy FullBuffer = FullBufferPolicy::Block; }; diff --git a/libs/writer/writer_wrapper/buffered_writer.cpp b/libs/writer/writer_wrapper/buffered_writer.cpp index 298c6cc..55da640 100644 --- a/libs/writer/writer_wrapper/buffered_writer.cpp +++ b/libs/writer/writer_wrapper/buffered_writer.cpp @@ -75,6 +75,8 @@ constexpr std::size_t kMaxWaitingBuffers = 8; constexpr std::size_t kCacheLineSize = 128; // The flush interval is kept within these bounds so that the sender's timed wait is well defined for any input. +// The limit that users see is BufferingDefaults::MaxFlushInterval, which WriterConfig enforces; these bounds are +// wider on purpose, so that tests that call Start() directly can keep the timed flush out of the way. constexpr std::chrono::milliseconds kMinFlushInterval{1}; constexpr std::chrono::milliseconds kMaxFlushInterval{std::chrono::hours{24}}; diff --git a/performance_tests/writer_performance_test/README.md b/performance_tests/writer_performance_test/README.md index d47bd16..ad5e04d 100644 --- a/performance_tests/writer_performance_test/README.md +++ b/performance_tests/writer_performance_test/README.md @@ -64,8 +64,8 @@ The counter is used in one of four ways: created once per thread, or created for these four uses a counter that was created once and has two tags. The full-buffer policies (`block`, `drop`, `send`) are compared flat out at the usual buffer size, and the flush -intervals (the default of 1000 ms, and 100 ms) under the paced load. Everything else uses the defaults, block and -1000 ms, and those rows keep the names they have on a version of the library without these settings, so a run can +intervals (the default of 3000 ms, and 100 ms) under the paced load. Everything else uses the defaults, block and +3000 ms, and those rows keep the names they have on a version of the library without these settings, so a run can still be compared with one from such a version. ## Reading the report diff --git a/performance_tests/writer_performance_test/scenario.h b/performance_tests/writer_performance_test/scenario.h index 65e6181..ed5f357 100644 --- a/performance_tests/writer_performance_test/scenario.h +++ b/performance_tests/writer_performance_test/scenario.h @@ -57,7 +57,7 @@ inline Policy PolicyFromName(const std::string& name) } // The library's default flush interval, in milliseconds. -inline constexpr int kDefaultFlushMillis = 1000; +inline constexpr int kDefaultFlushMillis = 3000; // One configuration to measure. To measure a new setting of the writer, add it here, to Key(), and to the places // that hand a scenario to the recording process (ToArguments and FromArguments below) and to the library diff --git a/spectator/test_registry.cpp b/spectator/test_registry.cpp index ed7b61c..bd69f66 100644 --- a/spectator/test_registry.cpp +++ b/spectator/test_registry.cpp @@ -79,7 +79,8 @@ TEST(RegistryTest, AgeGaugeWithId) // would be sent on its own while this test sleeps. TEST(RegistryTest, BufferedWriter) { - auto config = Config(WriterConfig(WriterTypes::Memory, 4096, std::chrono::hours(1))); + // The longest interval allowed, so that the timed flush does not interfere with this test + auto config = Config(WriterConfig(WriterTypes::Memory, 4096, BufferingDefaults::MaxFlushInterval)); auto r = Registry(config); auto c = r.CreateCounter("counter"); auto memoryWriter = static_cast(WriterTestHelper::GetImpl()); From 5ea75e25e12047da79882e2ea4267efd49943a19 Mon Sep 17 00:00:00 2001 From: Everett Badeaux Date: Tue, 10 Nov 2026 14:31:21 -0600 Subject: [PATCH 4/5] Fix thread-exit crash under ThreadSanitizer on Linux The destructor of the thread key marked an exited thread in the key and renewed the mark in every round of key destructors, so it also ran in the last round. ThreadSanitizer on Linux lets go of the thread in that round, and instrumented code that runs on the thread afterwards crashes. Mark the thread in the key only on macOS, where thread_local storage does not last through the rounds of key destructors. Elsewhere t_threadExited lasts until the thread is gone, so the key is left empty and the destructor runs once. The test that records in the last round stops one round earlier under ThreadSanitizer on Linux. --- .../writer/writer_wrapper/buffered_writer.cpp | 48 +++++++++++++------ .../writer_wrapper/test_buffered_writer.cpp | 26 ++++++++-- 2 files changed, 56 insertions(+), 18 deletions(-) diff --git a/libs/writer/writer_wrapper/buffered_writer.cpp b/libs/writer/writer_wrapper/buffered_writer.cpp index 55da640..9bf21ac 100644 --- a/libs/writer/writer_wrapper/buffered_writer.cpp +++ b/libs/writer/writer_wrapper/buffered_writer.cpp @@ -53,9 +53,10 @@ namespace spectator { // // Thread exit: a thread's buffers are attached to it as the value of a pthread key. The destructor of that key runs // when the thread exits, after the destructors of the thread's thread_local objects, and closes the buffers: the -// sender sends what is in them and then releases them. From then on the thread is marked as exited, in the key and -// in t_threadExited. Should anything still record on it (the destructor of another key), that line goes straight -// to the sender, see RecordAfterThreadExit: the thread gets no buffers again, as nothing would let go of them. +// sender sends what is in them and then releases them. From then on the thread is marked as exited: in +// t_threadExited, and in the key where that flag does not last (see kMarkExitedThreadInKey). Should anything still +// record on it (the destructor of another key), that line goes straight to the sender, see RecordAfterThreadExit: +// the thread gets no buffers again, as nothing would let go of them. // // Process exit: the destructor of the key does not run for threads that are still running when the process exits, // and that includes the thread that calls exit(). Their buffers stay attached and usable while thread_local and @@ -82,19 +83,33 @@ constexpr std::chrono::milliseconds kMaxFlushInterval{std::chrono::hours{24}}; using Clock = std::chrono::steady_clock; -// Once a thread has let go of its buffers at exit, the thread key holds one of the numbers 1 to kMaxExitedRounds -// instead of a pointer. The number counts the rounds of key destructors since then, see OnThreadExit. Every -// implementation runs at least this many rounds (_POSIX_THREAD_DESTRUCTOR_ITERATIONS). +// A thread that has let go of its buffers at exit has to be recognized for as long as code can run on it, which is +// longest on the thread whose exit ends the process and then runs the static destructors. t_threadExited says so, +// and so can a mark in the thread key, but neither lasts everywhere: +// - macOS releases thread_local storage while key destructors are still running and sets it up afresh when it is +// used again, so in the later rounds of key destructors t_threadExited reads as false again. The value of a key +// lasts there, also after the last round. So the thread is marked in the key as well. +// - Elsewhere thread_local storage lasts until the thread is gone, while glibc clears the keys after the last +// round. So t_threadExited alone marks the thread and the key is left empty, which also means that OnThreadExit +// runs only once. It must not run in the last round: ThreadSanitizer on Linux lets go of the thread in that +// round, in a key destructor that runs before this one, and crashes on instrumented code that runs on the thread +// after that. +#if defined(__APPLE__) +constexpr bool kMarkExitedThreadInKey = true; +#else +constexpr bool kMarkExitedThreadInKey = false; +#endif + +// The mark in the thread key: one of the numbers 1 to kMaxExitedRounds instead of a pointer. The number counts the +// rounds of key destructors since the thread let go of its buffers, see OnThreadExit. Every implementation runs at +// least this many rounds (_POSIX_THREAD_DESTRUCTOR_ITERATIONS). constexpr std::uintptr_t kMaxExitedRounds = 4; // Whether the value of the thread key says that the calling thread has no buffers: it is null because the thread -// has not recorded yet, or it is the mark of a thread that has exited. +// has not recorded yet or is marked as exited by t_threadExited alone, or it is the mark of a thread that has exited. bool HasNoBuffers(const void* value) noexcept { return reinterpret_cast(value) <= kMaxExitedRounds; } -// Set when the calling thread has let go of its buffers at exit. The mark in the thread key says the same, but -// neither lasts everywhere for as long as code can run on the thread, which is longest on the thread whose exit -// ends the process and then runs the static destructors. After the last round of key destructors, glibc has -// cleared the key but still has this flag, and macOS has released thread_local storage but still has the key. +// Set when the calling thread has let go of its buffers at exit. constinit thread_local bool t_threadExited = false; class BufferedWriter; @@ -770,8 +785,9 @@ void BufferedWriter::DrainAll() } } -// The destructor of the thread key: the calling thread is exiting. It lets go of the thread's buffers, leaves the -// mark that the thread has exited in their place and sets t_threadExited. +// The destructor of the thread key: the calling thread is exiting. It lets go of the thread's buffers and sets +// t_threadExited. Where the thread is also marked in the key (kMarkExitedThreadInKey), it leaves the mark in the +// place of the buffers; elsewhere the key stays empty and this destructor does not run again. // // Setting the key again makes this destructor run again in the next round of key destructors, now with the mark as // `value`. The mark is renewed each time, so that it is still there for destructors of other keys in later rounds, @@ -780,6 +796,7 @@ void OnThreadExit(void* value) { if (HasNoBuffers(value)) { + // A later round, and `value` is the mark. const auto rounds = reinterpret_cast(value); if (rounds < kMaxExitedRounds) { @@ -789,7 +806,10 @@ void OnThreadExit(void* value) } BufferedWriter::Detach(static_cast(value)); - pthread_setspecific(g_current.threadKey, reinterpret_cast(std::uintptr_t{1})); + if (kMarkExitedThreadInKey) + { + pthread_setspecific(g_current.threadKey, reinterpret_cast(std::uintptr_t{1})); + } t_threadExited = true; } diff --git a/libs/writer/writer_wrapper/test_buffered_writer.cpp b/libs/writer/writer_wrapper/test_buffered_writer.cpp index 5484801..c77e7b8 100644 --- a/libs/writer/writer_wrapper/test_buffered_writer.cpp +++ b/libs/writer/writer_wrapper/test_buffered_writer.cpp @@ -513,10 +513,28 @@ Arrived Examine(const Datagrams& datagrams, std::size_t bufferSize, int threads, // are recorded on a thread that has no buffers any more. // Round 1: one line. Depending on which of the two destructors runs first, the thread still has its buffers. // Round 2: more lines than a thread may have waiting, and may hold at all. -// Round 4, the last round that an implementation has to run: a line that does not fill a buffer. +// The last round, kLastLateRound: a line that does not fill a buffer. constexpr int kLastLateLineOfRound2 = kMaxWaiting + 4; constexpr const char* kLineOfLastRound = "end"; +// Whether this is a build with ThreadSanitizer: gcc defines the macro, clang answers __has_feature. +#if defined(__SANITIZE_THREAD__) +#define SPECTATOR_TEST_THREAD_SANITIZER 1 +#elif defined(__has_feature) +#if __has_feature(thread_sanitizer) +#define SPECTATOR_TEST_THREAD_SANITIZER 1 +#endif +#endif + +// The last round in which RecordLate() runs: round 4, the last round that an implementation has to run. Under +// ThreadSanitizer on Linux it is round 3: the sanitizer lets go of a thread in round 4, in a key destructor that runs +// before this one, and crashes on instrumented code that runs on the thread after that, as RecordLate() would. +#if defined(SPECTATOR_TEST_THREAD_SANITIZER) && !defined(__APPLE__) +constexpr int kLastLateRound = 3; +#else +constexpr int kLastLateRound = 4; +#endif + pthread_key_t LateRecorderKey(); void RecordLate(void* value) @@ -534,11 +552,11 @@ void RecordLate(void* value) buffered_writer::Record(Line('z', number)); } } - else if (round == 4) + else if (round == kLastLateRound) { buffered_writer::Record(kLineOfLastRound); } - if (round < 4) + if (round < kLastLateRound) { pthread_setspecific(LateRecorderKey(), value); // makes this destructor run again in the next round } @@ -913,7 +931,7 @@ TEST_F(BufferedWriterTest, KeepsLinesRecordedDuringThreadExitInOrder) // The thread exits, and RecordLate() records more lines on its way out. However many they are, they do not make // the thread wait, although nothing can be sent, and the thread does not send them itself. ASSERT_TRUE(thread.Exit()); - EXPECT_EQ(round, 4); + EXPECT_EQ(round, kLastLateRound); EXPECT_TRUE(m_socket.Sent().empty()); // The late lines follow the lines that the thread recorded before, in the order in which they were recorded. From 9c40196ef782e3bc4be3d1efcbdb2b5ca65675cb Mon Sep 17 00:00:00 2001 From: Everett Badeaux Date: Tue, 10 Nov 2026 14:35:15 -0600 Subject: [PATCH 5/5] Document that a process should have one Registry All registries of a process share one writer, and constructing a Registry sets that writer up again. A second Registry replaces the writer that the first one uses, and constructing one while another thread updates a meter can crash the process. --- README.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/README.md b/README.md index 4d35491..a1da1f6 100644 --- a/README.md +++ b/README.md @@ -5,6 +5,16 @@ This implements a basic [Spectator](https://github.com/Netflix/spectator) library for instrumenting CPP applications. It consists of a thin client designed to send metrics through [spectatord](https://github.com/Netflix-Skunkworks/spectatord). +## One Registry per Process + +Create one `Registry` for the whole process, at startup and before other threads update meters, and share that one +object with the code that creates meters. Do not create a second one. + +All registries of a process send through the same writer, and constructing a `Registry` sets that writer up from its +`Config`. A second `Registry` therefore does not get a writer of its own: it replaces the one that the first is using, +and from then on the meters of both go to the second one's location, with the second one's buffering. Constructing a +`Registry` while another thread is updating a meter is not safe either, and can crash the process. + ## High-Volume Publishing By default, the library sends every meter change to the spectatord sidecar immediately. This involves a blocking