Pthread writer

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
 

libs/config/config.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <chrono>
 #include <string>
 #include <unordered_map>
 
@@ -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:

libs/config/test_config.cpp

@@ -2,6 +2,7 @@
 #include <gmock/gmock.h>
 
 #include <config.h>
+#include <chrono>
 #include <optional>
 
 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)
 {

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);
     }
 };
 

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);
         }
     }
 };

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);
         }
     }
 };

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);
     }
 };
 

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);
     }
 };
 

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);
     }
 };
 

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);
     }
 };
 

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);
         }
     }
 };

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);
         }
     }
 };

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);
         }
     }
 };

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
 

libs/writer/writer_config/test_writer_config.cpp

@@ -3,6 +3,7 @@
 #include <gtest/gtest.h>
 #include <gmock/gmock.h>
 
+#include <chrono>
 #include <cstdlib>
 #include <optional>
 
@@ -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);