diff --git a/README.md b/README.md index a8d6c2a..4d35491 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 (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 +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 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: + * `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/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/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..de7e4d4 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` | `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 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 + +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/test_writer_config.cpp b/libs/writer/writer_config/test_writer_config.cpp index efa1796..e16d95d 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,72 @@ 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(3000)); + 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); + + // 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) { EXPECT_THROW({ WriterConfig config("invalid_type"); }, std::runtime_error); diff --git a/libs/writer/writer_config/writer_config.cpp b/libs/writer/writer_config/writer_config.cpp index 6f331ab..c320337 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 and at most "; }; std::pair GetWriterConfigFromString(const std::string& type) @@ -49,11 +50,24 @@ 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() || flushInterval > BufferingDefaults::MaxFlushInterval)) + { + throw std::runtime_error(WriterConfigConstants::FlushIntervalErrorMessage + + std::to_string(BufferingDefaults::MaxFlushInterval.count()) + " ms: " + + 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..9917781 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,30 @@ 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 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, + 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..12e99da 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,32 @@ 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. 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; +}; + struct WriterTypes { static constexpr auto Memory = "memory"; @@ -59,4 +86,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_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 4f68a4c..8e6e7e3 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 ) @@ -26,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/buffered_writer.cpp b/libs/writer/writer_wrapper/buffered_writer.cpp new file mode 100644 index 0000000..55da640 --- /dev/null +++ b/libs/writer/writer_wrapper/buffered_writer.cpp @@ -0,0 +1,857 @@ +#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. +// 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}}; + +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/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.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/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/performance_tests/writer_performance_test/README.md b/performance_tests/writer_performance_test/README.md index 8f1107e..ad5e04d 100644 --- a/performance_tests/writer_performance_test/README.md +++ b/performance_tests/writer_performance_test/README.md @@ -1,8 +1,8 @@ # Writer performance test Measures what it costs to record meters through the writers, and how much of what was recorded arrives. It does so -for a list of configurations: the socket writer, the buffer size, the number of recording threads, the load, and the -way the meters are used. It needs nothing but this repository, because the test starts its own receiver in place of +for a list of configurations: the socket writer, the buffer size, the full-buffer policy, the flush interval, the number of +recording threads, the load, and the way the meters are used. It needs nothing but this repository, because the test starts its own receiver in place of spectatord. ## Run it @@ -27,8 +27,8 @@ There are three sizes of run: | Option | What it measures | Time | |-----------|------------------------------------------------------------------------------------------|------------------| -| (none) | Both writers with 1, 4 and 8 threads, three buffer sizes, one paced rate, the meter uses | about 2 minutes | -| `--full` | The same with 1 to 16 threads, every buffer size at every thread count, two paced rates | about 12 minutes | +| (none) | Both writers with 1, 4 and 8 threads, three buffer sizes, the three full-buffer policies, one paced rate with two flush intervals, the meter uses | about 4 minutes | +| `--full` | The same with 1 to 16 threads, every buffer size at every thread count, two paced rates | about 19 minutes | | `--smoke` | Two configurations at a low rate; fails if a recorded line does not arrive | a few seconds | Add `--list` to see the configurations of a run and how long they take, without running them. @@ -63,6 +63,11 @@ The counter is used in one of four ways: created once per thread, or created for `registry.CreateCounter(name, tags).Increment()`, each with two tags or with none. Everything but the comparison of 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 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 The report has one table for each kind of load, and one that compares the ways of using a counter. @@ -74,7 +79,7 @@ The report has one table for each kind of load, and one that compares the ways o | `spread` | How far the slowest and the fastest measurement are apart, as a share of the median. A difference between two configurations that is smaller than this is not a difference. | | `call p50`, `p99`, `p99.9`, `longest` | How long a call that records one line took. Half of the calls took no longer than `p50`, and 99% no longer than `p99`. | | `cpu/line` | Processor time of the recording process per line, on all its threads, which includes the thread that sends. | -| `arrived` | The share of the recorded lines that the receiver got. `100%` is shown only if every line arrived. | +| `arrived` | The share of the recorded lines that the receiver got. `100%` is shown only if every line arrived. Under the drop policy, the rest includes the lines that the library dropped. | | `lines/datagram` | How many lines a datagram held on average. | | `delay p50`, `delay p99`, `longest delay` | Paced load: the time from recording a line to its arrival at the receiver. | | `bytes/line` | The length of a line, which the tags change. | diff --git a/performance_tests/writer_performance_test/plan.cpp b/performance_tests/writer_performance_test/plan.cpp index 55b2703..4efa132 100644 --- a/performance_tests/writer_performance_test/plan.cpp +++ b/performance_tests/writer_performance_test/plan.cpp @@ -49,6 +49,8 @@ Selection SmokeSelection() selection.pacedBuffers = {512}; selection.rates = {2000}; selection.meterThreads = {}; + selection.policies = {}; + selection.flushMillis = {}; return selection; } @@ -136,6 +138,23 @@ Plan MakePlan(const Selection& selection) } } + // The other full-buffer policies. + for (const Policy policy : selection.policies) + { + for (const int threads : selection.threads) + { + if (usual != 0) + { + Scenario scenario; + scenario.writer = writer; + scenario.bufferSize = usual; + scenario.threads = threads; + scenario.policy = policy; + add(scenario); + } + } + } + // Paced. for (const unsigned int wanted : selection.pacedBuffers) { @@ -155,6 +174,12 @@ Plan MakePlan(const Selection& selection) scenario.load = Load::Paced; scenario.linesPerSecond = rate; add(scenario); + for (const int flushMillis : selection.flushMillis) + { + scenario.flushMillis = flushMillis; + add(scenario); + } + scenario.flushMillis = kDefaultFlushMillis; } } } @@ -209,6 +234,9 @@ std::string Describe(const Scenario& scenario) : std::string("flat out"); text += scenario.createMeter ? ", counter created per line" : ""; text += scenario.tags ? "" : ", no tags"; + text += scenario.policy == Policy::Block ? "" : ", " + PolicyName(scenario.policy) + " policy"; + text += scenario.flushMillis == kDefaultFlushMillis ? "" + : ", flush every " + std::to_string(scenario.flushMillis) + " ms"; return text; } diff --git a/performance_tests/writer_performance_test/plan.h b/performance_tests/writer_performance_test/plan.h index 01a4467..bc7fd16 100644 --- a/performance_tests/writer_performance_test/plan.h +++ b/performance_tests/writer_performance_test/plan.h @@ -34,6 +34,13 @@ struct Selection std::vector pacedBuffers{kUsualBuffer}; std::vector rates{20000}; // lines per second and thread + // The full-buffer policies other than the default, measured flat out at the usual buffer size with every + // number of threads. + std::vector policies{Policy::Drop, Policy::Send}; + + // Flush intervals other than the default, in milliseconds, measured under the paced load. + std::vector flushMillis{100}; + // The ways to use a meter, compared flat out at the usual buffer size. std::vector meterThreads{1, 4}; std::vector meterWriters{WriterKind::Unix}; diff --git a/performance_tests/writer_performance_test/recording_run.cpp b/performance_tests/writer_performance_test/recording_run.cpp index d8df189..3c89d33 100644 --- a/performance_tests/writer_performance_test/recording_run.cpp +++ b/performance_tests/writer_performance_test/recording_run.cpp @@ -203,6 +203,19 @@ void Record(const Scenario& scenario, const spectator::Registry& registry, const } } +spectator::FullBufferPolicy LibraryPolicy(Policy policy) +{ + switch (policy) + { + case Policy::Drop: + return spectator::FullBufferPolicy::Drop; + case Policy::Send: + return spectator::FullBufferPolicy::Send; + default: + return spectator::FullBufferPolicy::Block; + } +} + // How long reading the clock takes, in nanoseconds: what a timed call is measured too long by. It is the average // over many readings, because one reading can be shorter than the steps in which the clock advances. double ClockNanos() @@ -245,7 +258,10 @@ try // No tags from the environment, so that the lines are the same on every machine. const spectator::WriterConfig writerConfig = scenario.bufferSize == 0 ? spectator::WriterConfig(location) - : spectator::WriterConfig(location, scenario.bufferSize); + : spectator::WriterConfig( + location, scenario.bufferSize, + std::chrono::milliseconds(scenario.flushMillis), + LibraryPolicy(scenario.policy)); const spectator::Config config(writerConfig, {}, false); const spectator::Registry registry(config); diff --git a/performance_tests/writer_performance_test/report.cpp b/performance_tests/writer_performance_test/report.cpp index 0643f18..cbd7c55 100644 --- a/performance_tests/writer_performance_test/report.cpp +++ b/performance_tests/writer_performance_test/report.cpp @@ -280,13 +280,13 @@ void PrintReport(const Plan& plan, const Measurements& measurements, const std:: }; // Flat out. - std::vector header{"writer", "buffer", "threads", "lines/s", "per thread", "speed-up", "spread", + std::vector header{"writer", "buffer", "policy", "threads", "lines/s", "per thread", "speed-up", "spread", "call p50", "p99", "longest", "cpu/line", "arrived", "lines/datagram"}; if (compare) { header.push_back("vs baseline"); } - Table flatOut(header, 2); + Table flatOut(header, 3); for (std::size_t i = 0; i < scenarios.size(); ++i) { const Scenario& scenario = scenarios[i]; @@ -298,6 +298,7 @@ void PrintReport(const Plan& plan, const Measurements& measurements, const std:: const double single = withOneThread(scenario); std::vector row{WriterName(scenario.writer), Buffer(scenario.bufferSize), + scenario.bufferSize == 0 ? "-" : PolicyName(scenario.policy), std::to_string(scenario.threads), Rate(summary.linesPerSecond), Rate(summary.linesPerSecond / scenario.threads), @@ -323,9 +324,9 @@ void PrintReport(const Plan& plan, const Measurements& measurements, const std:: } // Paced. - Table paced({"writer", "buffer", "threads", "lines/s per thread", "call p50", "p99", "p99.9", "longest", + Table paced({"writer", "buffer", "flush", "threads", "lines/s per thread", "call p50", "p99", "p99.9", "longest", "delay p50", "delay p99", "longest delay", "arrived"}, - 2); + 3); for (std::size_t i = 0; i < scenarios.size(); ++i) { const Scenario& scenario = scenarios[i]; @@ -334,7 +335,8 @@ void PrintReport(const Plan& plan, const Measurements& measurements, const std:: { continue; } - paced.Add({WriterName(scenario.writer), Buffer(scenario.bufferSize), std::to_string(scenario.threads), + paced.Add({WriterName(scenario.writer), Buffer(scenario.bufferSize), + std::to_string(scenario.flushMillis) + " ms", std::to_string(scenario.threads), std::to_string(scenario.linesPerSecond), Time(summary.callP50), Time(summary.callP99), Time(summary.callP999), Time(summary.callMax), Time(summary.delayP50 * 1e3), Time(summary.delayP99 * 1e3), Time(summary.delayMax * 1e3), Arrived(summary)}); @@ -351,7 +353,7 @@ void PrintReport(const Plan& plan, const Measurements& measurements, const std:: std::set> compared; for (const Scenario& scenario : scenarios) { - if (scenario.load == Load::FlatOut && !UsualMeter(scenario)) + if (scenario.load == Load::FlatOut && !UsualMeter(scenario) && scenario.DefaultWriterSettings()) { compared.insert({scenario.writer, scenario.bufferSize, scenario.threads}); } @@ -377,7 +379,7 @@ void PrintReport(const Plan& plan, const Measurements& measurements, const std:: { const Scenario& scenario = scenarios[i]; const Summary& summary = summaries[i]; - if (scenario.load != Load::FlatOut || summary.measurements == 0 || + if (scenario.load != Load::FlatOut || summary.measurements == 0 || !scenario.DefaultWriterSettings() || compared.count({scenario.writer, scenario.bufferSize, scenario.threads}) == 0) { continue; @@ -414,7 +416,8 @@ void PrintReport(const Plan& plan, const Measurements& measurements, const std:: "\n", clockNanos); std::printf(" - cpu/line is the processor time of the whole recording process, on all its threads, per line.\n"); - std::printf(" - arrived is the share of the recorded lines that this test received.\n"); + std::printf(" - arrived is the share of the recorded lines that this test received. Under the drop policy, the" + "\n rest includes the lines that the library dropped.\n"); if (!paced.Empty()) { std::printf(" - A paced thread records the lines that are due once every millisecond, a few at a time.\n"); @@ -445,7 +448,7 @@ void WriteCsv(const std::string& path, const std::string& label, const Plan& pla } file << "label,key,writer,buffer,threads,load,rate_per_thread,meter,tags,repeat,ok,lines_per_second," "call_p50_ns,call_p99_ns,call_p999_ns,call_max_ns,clock_ns,cpu_ns_per_line,recorded_lines," - "received_lines,datagrams,bytes,delay_p50_us,delay_p99_us,delay_max_us\n"; + "received_lines,datagrams,bytes,delay_p50_us,delay_p99_us,delay_max_us,policy,flush_ms\n"; for (std::size_t i = 0; i < plan.scenarios.size(); ++i) { const Scenario& scenario = plan.scenarios[i]; @@ -462,7 +465,8 @@ void WriteCsv(const std::string& path, const std::string& label, const Plan& pla << Format("%.1f", measurement.cpuNanosPerLine) << ',' << measurement.recorded << ',' << measurement.received << ',' << measurement.datagrams << ',' << measurement.bytes << ',' << Format("%.0f", measurement.delayP50) << ',' << Format("%.0f", measurement.delayP99) << ',' - << Format("%.0f", measurement.delayMax) << '\n'; + << Format("%.0f", measurement.delayMax) << ',' << PolicyName(scenario.policy) << ',' + << scenario.flushMillis << '\n'; } } } diff --git a/performance_tests/writer_performance_test/scenario.h b/performance_tests/writer_performance_test/scenario.h index 14ef4cb..ed5f357 100644 --- a/performance_tests/writer_performance_test/scenario.h +++ b/performance_tests/writer_performance_test/scenario.h @@ -26,6 +26,39 @@ inline std::string WriterName(WriterKind kind) { return kind == WriterKind::Udp inline std::string LoadName(Load load) { return load == Load::FlatOut ? "flat-out" : "paced"; } +// What a recording thread does when it is out of buffer space: the library's FullBufferPolicy. +enum class Policy +{ + Block, + Drop, + Send +}; + +inline std::string PolicyName(Policy policy) +{ + return policy == Policy::Block ? "block" : policy == Policy::Drop ? "drop" : "send"; +} + +inline Policy PolicyFromName(const std::string& name) +{ + if (name == "block") + { + return Policy::Block; + } + if (name == "drop") + { + return Policy::Drop; + } + if (name == "send") + { + return Policy::Send; + } + throw std::invalid_argument("Unknown policy: " + name); +} + +// The library's default flush interval, in milliseconds. +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 // (recording_run.cpp). @@ -38,6 +71,11 @@ struct Scenario int linesPerSecond = 0; // paced load: the rate of each thread bool createMeter = false; // create the counter for every line, instead of once per thread bool tags = true; // whether the counter has tags + Policy policy = Policy::Block; // buffered only + int flushMillis = kDefaultFlushMillis; // buffered only: the flush interval + + // Whether the writer has the library's default policy and flush interval. + bool DefaultWriterSettings() const { return policy == Policy::Block && flushMillis == kDefaultFlushMillis; } // Names the scenario in the CSV file. Two runs are compared scenario by scenario through this. std::string Key() const @@ -50,6 +88,16 @@ struct Scenario } key += createMeter ? "/create" : "/reuse"; key += tags ? "/tags" : "/no-tags"; + // Only settings that differ from the defaults are named, so that the keys of the default settings are the + // same as in a run of a library version that does not have these settings. + if (policy != Policy::Block) + { + key += "/policy=" + PolicyName(policy); + } + if (flushMillis != kDefaultFlushMillis) + { + key += "/flush=" + std::to_string(flushMillis) + "ms"; + } return key; } }; @@ -100,6 +148,8 @@ inline std::vector ToArguments(const Scenario& scenario) "rate=" + std::to_string(scenario.linesPerSecond), "create=" + std::to_string(scenario.createMeter ? 1 : 0), "tags=" + std::to_string(scenario.tags ? 1 : 0), + "policy=" + PolicyName(scenario.policy), + "flush=" + std::to_string(scenario.flushMillis), }; } @@ -113,6 +163,8 @@ inline Scenario FromArguments(const Arguments& arguments) scenario.linesPerSecond = static_cast(Count(arguments, "rate")); scenario.createMeter = Count(arguments, "create") != 0; scenario.tags = Count(arguments, "tags") != 0; + scenario.policy = PolicyFromName(Text(arguments, "policy")); + scenario.flushMillis = static_cast(Count(arguments, "flush")); return scenario; } diff --git a/performance_tests/writer_performance_test/writer_performance_test.cpp b/performance_tests/writer_performance_test/writer_performance_test.cpp index a2dd254..c07d4e0 100644 --- a/performance_tests/writer_performance_test/writer_performance_test.cpp +++ b/performance_tests/writer_performance_test/writer_performance_test.cpp @@ -85,9 +85,9 @@ void PrintUsage() "Usage: writer_performance_test [options]\n" "\n" "Measures recording through the writers for a list of configurations and prints a report.\n" - "Without options it takes about two minutes.\n" + "Without options it takes about four minutes.\n" "\n" - " --full more thread counts, buffer sizes and rates: about twelve minutes\n" + " --full more thread counts, buffer sizes and rates: about nineteen minutes\n" " --smoke a few seconds: checks that the test works, and that every line arrives\n" " --list print the configurations and how long they take, and stop\n" "\n" 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()); } } diff --git a/spectator/test_registry.cpp b/spectator/test_registry.cpp index 265a16d..bd69f66 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,28 @@ 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) +{ + // 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()); + + 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));