diff --git a/README.md b/README.md index 4bcac91..a8d6c2a 100644 --- a/README.md +++ b/README.md @@ -13,7 +13,8 @@ use cases. For this purpose a simple buffering functionality is implemented, and 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. +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. ## Local & IDE Configuration diff --git a/libs/writer/writer_wrapper/test_writer.cpp b/libs/writer/writer_wrapper/test_writer.cpp index f6a8f37..5da2b89 100644 --- a/libs/writer/writer_wrapper/test_writer.cpp +++ b/libs/writer/writer_wrapper/test_writer.cpp @@ -5,6 +5,11 @@ #include #include +#include +#include +#include +#include +#include #include #include #include @@ -15,6 +20,38 @@ using namespace spectator; +// 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) +{ + WriterTestHelper::InitializeWriter(WriterType::Unix, socketPath, 0, 4096); + Counter counter(MeterId("shutdown.counter")); + counter.Increment(); + 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. +TEST(WriterShutdownDeathTest, SendsBufferedLinesAtExit) +{ + using DatagramProtocol = boost::asio::local::datagram_protocol; + const std::string socketPath = "/tmp/test_uds_shutdown_socket"; + std::filesystem::remove(socketPath); + boost::asio::io_context ioContext; + DatagramProtocol::socket socket(ioContext, DatagramProtocol::endpoint(socketPath)); + socket.non_blocking(true); + + EXPECT_EXIT(RecordOneCounterAndExit(socketPath), testing::ExitedWithCode(0), ""); + + std::array buffer{}; + boost::system::error_code ec; + const auto bytesRead = socket.receive(boost::asio::buffer(buffer), 0, ec); + std::filesystem::remove(socketPath); + ASSERT_FALSE(ec) << "nothing was sent at exit: " << ec.message(); + EXPECT_EQ(std::string(buffer.data(), bytesRead), "c:shutdown.counter:1.000000\n"); +} + // 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 aac4887..69a7b73 100644 --- a/libs/writer/writer_wrapper/writer.cpp +++ b/libs/writer/writer_wrapper/writer.cpp @@ -14,12 +14,29 @@ Writer::~Writer() if (instance.bufferingEnabled) { - instance.shutdown.store(true); + { + // 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(); } + + // 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); + } } this->Close(); }