Use blocking queue

example/example.cpp

@@ -28,6 +28,8 @@ int main(int, char *[])
 
     try
     {
+		async_example();
+		return 0;
         auto console = spdlog::stdout_color_st("console");
         console->info("Welcome to spdlog!");
 
@@ -82,7 +84,7 @@ int main(int, char *[])
 
         // Asynchronous logging is very fast..
         // Just call spdlog::set_async_mode(q_size) and all created loggers from now on will be asynchronous..
-        async_example();
+        //async_example();
 
         // Log user-defined types example
         user_defined_example();
@@ -107,15 +109,23 @@ int main(int, char *[])
 #include "spdlog/async.h"
 void async_example()
 {
-    auto async_file = spd::basic_logger_mt<spdlog::create_async>("async_file_logger", "logs/async_log.txt");
-    for (int i = 0; i < 100; ++i)
-    {
-        async_file->info("Async message #{}", i);
-    }
+    //auto async_file = spd::basic_logger_mt<spdlog::create_async>("async_file_logger", "logs/async_log.txt");
+	
+	for (int j = 0; j < 1; j++)
+	{
+		spdlog::init_thread_pool(1024, 10);
+		auto async_file = spd::stderr_color_mt<spdlog::create_async>("console");
+		for (int i = 0; i < 1024; ++i)
+		{
+			async_file->info("{} Async message #{}", j, i);
+		}
+		spdlog::drop_all();
+	}
+	//std::this_thread::sleep_for(std::chrono::seconds(1));
+		
 
     // you can also modify thread pool settings *before* creating the logger:
-    // spdlog::init_thread_pool(32768, 4); // queue with 32k of pre allocated items and 4 backing threads.
-    // if not called a defaults are: preallocated 8192 queue items and 1 worker thread.
+    // spdlog::init_thread_pool(32768, 4); // queue with max 32k items 4 backing threads.    
 }
 
 // syslog example (linux/osx/freebsd)

include/spdlog/details/async_logger_impl.h

@@ -8,7 +8,7 @@
 // async logger implementation
 // uses a thread pool to perform the actual logging
 
-#include "../details/thread_pool.h"
+#include "spdlog/details/thread_pool.h"
 
 #include <chrono>
 #include <memory>

include/spdlog/details/mpmc_blocking_q.h

+#pragma once
+
+//
+// Copyright(c) 2018 Gabi Melman.
+// Distributed under the MIT License (http://opensource.org/licenses/MIT)
+//
+
+// async log helper :
+// multi producer-multi consumer blocking queue
+// enqueue(..) - will block until room found to put the new message
+// enqueue_nowait(..) - will return immediatly with false if no room left in the queue
+// dequeue_for(..) - will block until the queue is not empty or timeout passed
+
+#include <condition_variable>
+#include <mutex>
+#include <queue>
+
+namespace spdlog {
+namespace details {
+
+template<typename T>
+class mpmc_bounded_queue
+{
+public:
+    using item_type = T;
+    explicit mpmc_bounded_queue(size_t max_items)
+        : max_items_(max_items)
+    {
+    }
+
+    // try to enqueue and block if no room left
+    void enqueue(T &&item)
+    {
+        {
+            std::unique_lock<std::mutex> lock(queue_mutex_);
+            pop_cv_.wait(lock, [this] { return this->q_.size() < this->max_items_; });
+            q_.push(std::move(item));
+        }
+        push_cv_.notify_one();
+    }
+
+    // try to enqueue and return immdeialty false if no room left
+    bool enqueue_nowait(T &&item)
+    {
+        {
+            std::unique_lock<std::mutex> lock(queue_mutex_);
+            if (q_.size() == this->max_items_)
+            {
+                return false;
+            }
+            q_.push(std::forward<T>(item));
+        }
+        push_cv_.notify_one();
+        return true;
+    }
+
+    // try to dequeue item. if no item found. wait upto timeout and try again
+    // Return true, if succeeded dequeue item, false otherwise
+    bool dequeue_for(T &popped_item, std::chrono::milliseconds wait_duration)
+    {
+        {
+            std::unique_lock<std::mutex> lock(queue_mutex_);
+            if (!push_cv_.wait_for(lock, wait_duration, [this] { return this->q_.size() > 0; }))
+            {
+                return false;
+            }
+
+            popped_item = std::move(q_.front());
+            q_.pop();
+        }
+        pop_cv_.notify_one();
+        return true;
+    }
+
+private:
+    size_t max_items_;
+    std::mutex queue_mutex_;
+    std::condition_variable push_cv_;
+    std::condition_variable pop_cv_;
+
+    std::queue<T> q_;
+};
+} // namespace details
+} // namespace spdlog

include/spdlog/details/mpmc_bounded_q.h

-/*
-A modified version of Bounded MPMC queue by Dmitry Vyukov.
-
-Original code from:
-http://www.1024cores.net/home/lock-free-algorithms/queues/bounded-mpmc-queue
-
-licensed by Dmitry Vyukov under the terms below:
-
-Simplified BSD license
-
-Copyright (c) 2010-2011 Dmitry Vyukov. All rights reserved.
-Redistribution and use in source and binary forms, with or without modification,
-are permitted provided that the following conditions are met:
-1. Redistributions of source code must retain the above copyright notice, this list of
-conditions and the following disclaimer.
-
-2. Redistributions in binary form must reproduce the above copyright notice, this list
-of conditions and the following disclaimer in the documentation and/or other materials
-provided with the distribution.
-
-THIS SOFTWARE IS PROVIDED BY DMITRY VYUKOV "AS IS" AND ANY EXPRESS OR IMPLIED
-WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF
-MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT
-SHALL DMITRY VYUKOV OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT,
-INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
-LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA,
-OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF
-LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE
-OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF
-ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-
-The views and conclusions contained in the software and documentation are those of the authors and
-should not be interpreted as representing official policies, either expressed or implied, of Dmitry Vyukov.
-*/
-
-/*
-The code in its current form adds the license below:
-
-Copyright(c) 2015 Gabi Melman.
-Distributed under the MIT License (http://opensource.org/licenses/MIT)
-
-*/
-
-#pragma once
-
-#include "spdlog/common.h"
-
-#include <atomic>
-#include <utility>
-
-namespace spdlog {
-namespace details {
-
-template<typename T>
-class mpmc_bounded_queue
-{
-public:
-    using item_type = T;
-
-    explicit mpmc_bounded_queue(size_t buffer_size)
-        : max_size_(buffer_size)
-        , buffer_(new cell_t[buffer_size])
-        , buffer_mask_(buffer_size - 1)
-    {
-        // queue size must be power of two
-        if (!((buffer_size >= 2) && ((buffer_size & (buffer_size - 1)) == 0)))
-        {
-            throw spdlog_ex("async logger queue size must be power of two");
-        }
-
-        for (size_t i = 0; i != buffer_size; i += 1)
-        {
-            buffer_[i].sequence_.store(i, std::memory_order_relaxed);
-        }
-        enqueue_pos_.store(0, std::memory_order_relaxed);
-        dequeue_pos_.store(0, std::memory_order_relaxed);
-    }
-
-    ~mpmc_bounded_queue()
-    {
-        delete[] buffer_;
-    }
-
-    mpmc_bounded_queue(mpmc_bounded_queue const &) = delete;
-    void operator=(mpmc_bounded_queue const &) = delete;
-
-    bool enqueue(T &&data)
-    {
-        cell_t *cell;
-        size_t pos = enqueue_pos_.load(std::memory_order_relaxed);
-        for (;;)
-        {
-            cell = &buffer_[pos & buffer_mask_];
-            size_t seq = cell->sequence_.load(std::memory_order_acquire);
-            intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos);
-            if (dif == 0)
-            {
-                if (enqueue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed))
-                {
-                    break;
-                }
-            }
-            else if (dif < 0)
-            {
-                return false;
-            }
-            else
-            {
-                pos = enqueue_pos_.load(std::memory_order_relaxed);
-            }
-        }
-        cell->data_ = std::move(data);
-        cell->sequence_.store(pos + 1, std::memory_order_release);
-        return true;
-    }
-
-    bool dequeue(T &data)
-    {
-        cell_t *cell;
-        size_t pos = dequeue_pos_.load(std::memory_order_relaxed);
-        for (;;)
-        {
-            cell = &buffer_[pos & buffer_mask_];
-            size_t seq = cell->sequence_.load(std::memory_order_acquire);
-            intptr_t dif = static_cast<intptr_t>(seq) - static_cast<intptr_t>(pos + 1);
-            if (dif == 0)
-            {
-                if (dequeue_pos_.compare_exchange_weak(pos, pos + 1, std::memory_order_relaxed))
-                {
-                    break;
-                }
-            }
-            else if (dif < 0)
-            {
-                return false;
-            }
-            else
-            {
-                pos = dequeue_pos_.load(std::memory_order_relaxed);
-            }
-        }
-        data = std::move(cell->data_);
-        cell->sequence_.store(pos + buffer_mask_ + 1, std::memory_order_release);
-        return true;
-    }
-
-    bool is_empty()
-    {
-        size_t front, front1, back;
-        // try to take a consistent snapshot of front/tail.
-        do
-        {
-            front = enqueue_pos_.load(std::memory_order_acquire);
-            back = dequeue_pos_.load(std::memory_order_acquire);
-            front1 = enqueue_pos_.load(std::memory_order_relaxed);
-        } while (front != front1);
-        return back == front;
-    }
-
-private:
-    struct cell_t
-    {
-        std::atomic<size_t> sequence_;
-        T data_;
-    };
-
-    size_t const max_size_;
-
-    static size_t const cacheline_size = 64;
-    using cacheline_pad_t = char[cacheline_size];
-
-    cacheline_pad_t pad0_;
-    cell_t *const buffer_;
-    size_t const buffer_mask_;
-    cacheline_pad_t pad1_;
-    std::atomic<size_t> enqueue_pos_;
-    cacheline_pad_t pad2_;
-    std::atomic<size_t> dequeue_pos_;
-    cacheline_pad_t pad3_;
-};
-
-} // namespace details
-} // namespace spdlog

tests/CMakeLists.txt

@@ -7,7 +7,8 @@ set(SPDLOG_UTESTS_SOURCES
     file_helper.cpp
     file_log.cpp
     test_misc.cpp
-	test_pattern_formatter
+	test_pattern_formatter.cpp
+    test_async.cpp
     includes.h
     registry.cpp
     test_macros.cpp

tests/test_async.cpp

+#include "includes.h"
+#include "test_sink.h"
+#include "spdlog/async.h"
+#include "spdlog/sinks/simple_file_sink.h"
+
+//std::unique_ptr<spdlog::async_logger> create_logger(size_t tp_queue_size, size_t tp_threads)
+//{
+//	auto tp = std::make_shared<details::thread_pool>(8192, 1);
+//	auto logger = std::make_shared<async_logger>("as", test_sink, tp, async_overflow_policy::block_retry);
+//}
+
+TEST_CASE("basic async test ", "[async]")
+{
+	using namespace spdlog;
+	auto test_sink = std::make_shared<sinks::test_sink_mt>();
+	size_t queue_size = 128;
+	size_t messages = 256;
+	{
+		auto tp = std::make_shared<details::thread_pool>(queue_size, 1);
+		auto logger = std::make_shared<async_logger>("as", test_sink, tp, async_overflow_policy::block_retry);
+		for (size_t i = 0; i < messages; i++)
+		{
+			logger->info("Hello message #{}", i);
+		}
+		logger->flush();
+	}
+	REQUIRE(test_sink->msg_counter() == messages);
+	REQUIRE(test_sink->flushed_msg_counter() == messages);
+}
+
+TEST_CASE("discard policy ", "[async]")
+{
+	using namespace spdlog;
+	auto test_sink = std::make_shared<sinks::test_sink_mt>();
+	size_t queue_size = 2;
+	size_t messages = 1024;
+	{
+		auto tp = std::make_shared<details::thread_pool>(queue_size, 1);
+		auto logger = std::make_shared<async_logger>("as", test_sink, tp, async_overflow_policy::discard_log_msg);
+		for (size_t i = 0; i < messages; i++)
+		{
+			logger->info("Hello message #{}", i);
+		}
+	}
+	
+	REQUIRE(test_sink->msg_counter() < messages);
+	REQUIRE(test_sink->flushed_msg_counter() < messages);
+}
+
+TEST_CASE("flush", "[async]")
+{
+	using namespace spdlog;
+	auto test_sink = std::make_shared<sinks::test_sink_mt>();
+	size_t queue_size = 256;
+	size_t messages = 256;
+	{
+		auto tp = std::make_shared<details::thread_pool>(queue_size, 1);
+		auto logger = std::make_shared<async_logger>("as", test_sink, tp, async_overflow_policy::block_retry);
+		for (size_t i = 0; i < messages; i++)
+		{
+			logger->info("Hello message #{}", i);
+		}
+		
+		logger->flush();
+	}
+	std::this_thread::sleep_for(std::chrono::milliseconds(250));
+	REQUIRE(test_sink->msg_counter() == messages);
+	REQUIRE(test_sink->flushed_msg_counter() == messages);	
+}
+
+TEST_CASE("multi threads", "[async]")
+{
+	using namespace spdlog;
+	auto test_sink = std::make_shared<sinks::test_sink_mt>();
+	size_t queue_size = 128;
+	size_t messages = 256;
+	size_t n_threads = 10;
+	{
+		auto tp = std::make_shared<details::thread_pool>(queue_size, 1);
+		auto logger = std::make_shared<async_logger>("as", test_sink, tp, async_overflow_policy::block_retry);
+
+		std::vector<std::thread> threads;
+		for (size_t i = 0; i < n_threads; i++)
+		{
+			threads.emplace_back([logger, messages] {
+				for (size_t j = 0; j < messages; j++)
+				{
+					logger->info("Hello message #{}", j);
+				}
+				
+			});
+		}
+		
+		for (auto &t : threads)
+		{
+			t.join();
+		}
+		logger->flush();
+
+	}
+		
+	REQUIRE(test_sink->msg_counter() == messages * n_threads);
+	REQUIRE(test_sink->flushed_msg_counter() == messages * n_threads);
+}
+
+TEST_CASE("to_file", "[async]")
+{
+	prepare_logdir();
+	size_t queue_size = 512;
+	size_t messages = 512;
+	size_t n_threads = 4;	
+	spdlog::init_thread_pool(queue_size, n_threads);
+	auto logger= spdlog::basic_logger_mt<spdlog::create_async>("as", "logs/async_test.log", true);
+	
+	std::vector<std::thread> threads;
+	for (size_t i = 0; i < n_threads; i++)
+	{
+		threads.emplace_back([logger, messages] {
+			for (size_t j = 0; j < messages; j++)
+			{
+				logger->info("Hello message #{}", j);
+			}
+		});
+	}
+
+	for (auto &t : threads)
+	{
+		t.join();
+	}	
+	logger.reset();
+	spdlog::drop("as");
+	std::this_thread::sleep_for(std::chrono::seconds(1));
+	REQUIRE(count_lines("logs/async_test.log") == messages * n_threads);
+}

tests/test_sink.h

+//
+// Copyright(c) 2018 Gabi Melman.
+// Distributed under the MIT License (http://opensource.org/licenses/MIT)
+//
+
+#pragma once
+
+#include "spdlog/details/null_mutex.h"
+#include "spdlog/sinks/base_sink.h"
+
+#include <mutex>
+
+namespace spdlog {
+namespace sinks {
+
+template<class Mutex>
+class test_sink : public base_sink<Mutex>
+{
+public:
+    size_t msg_counter()
+    {
+        return msg_counter_;
+    }
+
+    size_t flushed_msg_counter()
+    {
+        return flushed_msg_counter_;
+    }
+
+protected:
+    void _sink_it(const details::log_msg &) override
+    {
+        msg_counter_++;
+    }
+
+    void _flush() override
+    {
+        flushed_msg_counter_ += msg_counter_;
+    }
+    size_t msg_counter_{0};
+    size_t flushed_msg_counter_{0};
+};
+
+using test_sink_mt = test_sink<std::mutex>;
+using test_sink_st = test_sink<details::null_mutex>;
+
+} // namespace sinks
+} // namespace spdlog

tests/tests.vcxproj

@@ -129,6 +129,7 @@
     <ClCompile Include="errors.cpp" />
     <ClCompile Include="file_helper.cpp" />
     <ClCompile Include="file_log.cpp" />
+    <ClCompile Include="test_async.cpp" />
     <ClCompile Include="test_misc.cpp" />
     <ClCompile Include="main.cpp" />
     <ClCompile Include="registry.cpp" />

tests/tests.vcxproj.filters

@@ -42,6 +42,9 @@
     <ClCompile Include="test_misc.cpp">
       <Filter>Source Files</Filter>
     </ClCompile>
+    <ClCompile Include="test_async.cpp">
+      <Filter>Source Files</Filter>
+    </ClCompile>
   </ItemGroup>
   <ItemGroup>
     <ClInclude Include="includes.h">