summaryrefslogtreecommitdiff
path: root/lib/spdlog/details/async_log_helper.h
blob: 49789b94be2bfedc1273022cde435155c5cef74c (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
//
// Copyright(c) 2015 Gabi Melman.
// Distributed under the MIT License (http://opensource.org/licenses/MIT)
//

// async log helper :
// Process logs asynchronously using a back thread.
//
// If the internal queue of log messages reaches its max size,
// then the client call will block until there is more room.
//

#pragma once

#include "../common.h"
#include "../details/log_msg.h"
#include "../details/mpmc_blocking_q.h"
#include "../details/os.h"
#include "../formatter.h"
#include "../sinks/sink.h"

#include <chrono>
#include <condition_variable>
#include <exception>
#include <functional>
#include <memory>
#include <string>
#include <thread>
#include <utility>
#include <vector>

namespace spdlog {
namespace details {

class async_log_helper
{
    // Async msg to move to/from the queue
    // Movable only. should never be copied
    enum class async_msg_type
    {
        log,
        flush,
        terminate
    };

    struct async_msg
    {
        level::level_enum level;
        log_clock::time_point time;
        size_t thread_id;
        std::string txt;
        async_msg_type msg_type;
        size_t msg_id;

        async_msg() = default;
        ~async_msg() = default;

        explicit async_msg(async_msg_type m_type)
            : level(level::info)
            , thread_id(0)
            , msg_type(m_type)
            , msg_id(0)
        {
        }

        async_msg(async_msg &&other) = default;
        async_msg &operator=(async_msg &&other) = default;

        // never copy or assign. should only be moved..
        async_msg(const async_msg &) = delete;
        async_msg &operator=(const async_msg &other) = delete;

        // construct from log_msg
        explicit async_msg(const details::log_msg &m)
            : level(m.level)
            , time(m.time)
            , thread_id(m.thread_id)
            , txt(m.raw.data(), m.raw.size())
            , msg_type(async_msg_type::log)
            , msg_id(m.msg_id)
        {
        }

        // copy into log_msg
        void fill_log_msg(log_msg &msg, std::string *logger_name)
        {
            msg.logger_name = logger_name;
            msg.level = level;
            msg.time = time;
            msg.thread_id = thread_id;
            msg.raw.clear();
            msg.raw << txt;
            msg.msg_id = msg_id;
        }
    };

public:
    using item_type = async_msg;
    using q_type = details::mpmc_bounded_queue<item_type>;

    using clock = std::chrono::steady_clock;

    async_log_helper(std::string logger_name, formatter_ptr formatter, std::vector<sink_ptr> sinks, size_t queue_size,
        const log_err_handler err_handler, const async_overflow_policy overflow_policy = async_overflow_policy::block_retry,
        std::function<void()> worker_warmup_cb = nullptr,
        const std::chrono::milliseconds &flush_interval_ms = std::chrono::milliseconds::zero(),
        std::function<void()> worker_teardown_cb = nullptr);

    void log(const details::log_msg &msg);

    // stop logging and join the back thread
    ~async_log_helper();

    async_log_helper(const async_log_helper &) = delete;
    async_log_helper &operator=(const async_log_helper &) = delete;

    void set_formatter(formatter_ptr msg_formatter);

    void flush();

    void set_error_handler(spdlog::log_err_handler err_handler);

private:
    std::string _logger_name;
    formatter_ptr _formatter;
    std::vector<std::shared_ptr<sinks::sink>> _sinks;

    // queue of messages to log
    q_type _q;

    log_err_handler _err_handler;

    std::chrono::time_point<log_clock> _last_flush;

    // overflow policy
    const async_overflow_policy _overflow_policy;

    // worker thread warmup callback - one can set thread priority, affinity, etc
    const std::function<void()> _worker_warmup_cb;

    // auto periodic sink flush parameter
    const std::chrono::milliseconds _flush_interval_ms;

    // worker thread teardown callback
    const std::function<void()> _worker_teardown_cb;

    std::mutex null_mutex_;
    // null_mutex null_mutex_;
    std::condition_variable_any not_empty_cv_;
    std::condition_variable_any not_full_cv_;

    // worker thread
    std::thread _worker_thread;

    void enqueue_msg(async_msg &&new_msg, async_overflow_policy policy);

    // worker thread main loop
    void worker_loop();

    // dequeue next message from the queue and process it.
    // return false if termination of the queue is required
    bool process_next_msg();

    void handle_flush_interval();

    void flush_sinks();
};
} // namespace details
} // namespace spdlog

///////////////////////////////////////////////////////////////////////////////
// async_sink class implementation
///////////////////////////////////////////////////////////////////////////////
inline spdlog::details::async_log_helper::async_log_helper(std::string logger_name, formatter_ptr formatter, std::vector<sink_ptr> sinks,
    size_t queue_size, log_err_handler err_handler, const async_overflow_policy overflow_policy, std::function<void()> worker_warmup_cb,
    const std::chrono::milliseconds &flush_interval_ms, std::function<void()> worker_teardown_cb)
    : _logger_name(std::move(logger_name))
    , _formatter(std::move(formatter))
    , _sinks(std::move(sinks))
    , _q(queue_size)
    , _err_handler(std::move(err_handler))
    , _last_flush(os::now())
    , _overflow_policy(overflow_policy)
    , _worker_warmup_cb(std::move(worker_warmup_cb))
    , _flush_interval_ms(flush_interval_ms)
    , _worker_teardown_cb(std::move(worker_teardown_cb))
{
    _worker_thread = std::thread(&async_log_helper::worker_loop, this);
}

// send to the worker thread terminate message, and join it.
inline spdlog::details::async_log_helper::~async_log_helper()
{
    try
    {
        enqueue_msg(async_msg(async_msg_type::terminate), async_overflow_policy::block_retry);
        _worker_thread.join();
    }
    catch (...) // don't crash in destructor
    {
    }
}

// try to push and block until succeeded (if the policy is not to discard when the queue is full)
inline void spdlog::details::async_log_helper::log(const details::log_msg &msg)
{
    enqueue_msg(async_msg(msg), _overflow_policy);
}

inline void spdlog::details::async_log_helper::enqueue_msg(details::async_log_helper::async_msg &&new_msg, async_overflow_policy policy)
{

    // block until succeeded pushing to the queue
    if (policy == async_overflow_policy::block_retry)
    {
        _q.enqueue(std::move(new_msg));
    }
    else
    {
        _q.enqueue_nowait(std::move(new_msg));
    }
}

// optionally wait for the queue be empty and request flush from the sinks
inline void spdlog::details::async_log_helper::flush()
{
    enqueue_msg(async_msg(async_msg_type::flush), _overflow_policy);
}

inline void spdlog::details::async_log_helper::worker_loop()
{
    if (_worker_warmup_cb)
    {
        _worker_warmup_cb();
    }
    auto active = true;
    while (active)
    {
        try
        {
            active = process_next_msg();
        }
        SPDLOG_CATCH_AND_HANDLE
    }
    if (_worker_teardown_cb)
    {
        _worker_teardown_cb();
    }
}

// process next message in the queue
// return true if this thread should still be active (while no terminate msg was received)
inline bool spdlog::details::async_log_helper::process_next_msg()
{
    async_msg incoming_async_msg;
    bool dequeued = _q.dequeue_for(incoming_async_msg, std::chrono::seconds(2));
    if (!dequeued)
    {
        handle_flush_interval();
        return true;
    }

    switch (incoming_async_msg.msg_type)
    {
    case async_msg_type::flush:
        flush_sinks();
        return true;

    case async_msg_type::terminate:
        flush_sinks();
        return false;

    default:
        log_msg incoming_log_msg;
        incoming_async_msg.fill_log_msg(incoming_log_msg, &_logger_name);
        _formatter->format(incoming_log_msg);
        for (auto &s : _sinks)
        {
            if (s->should_log(incoming_log_msg.level))
            {
                try
                {
                    s->log(incoming_log_msg);
                }
                SPDLOG_CATCH_AND_HANDLE
            }
        }
        handle_flush_interval();
        return true;
    }
    assert(false);
    return true; // should not be reached
}

inline void spdlog::details::async_log_helper::set_formatter(formatter_ptr msg_formatter)
{
    _formatter = std::move(msg_formatter);
}

inline void spdlog::details::async_log_helper::set_error_handler(spdlog::log_err_handler err_handler)
{
    _err_handler = std::move(err_handler);
}

// flush all sinks if _flush_interval_ms has expired.
inline void spdlog::details::async_log_helper::handle_flush_interval()
{
    if (_flush_interval_ms == std::chrono::milliseconds::zero())
    {
        return;
    }
    auto delta = details::os::now() - _last_flush;
    ;
    if (delta >= _flush_interval_ms)
    {
        flush_sinks();
    }
}

// flush all sinks if _flush_interval_ms has expired. only called if queue is empty
inline void spdlog::details::async_log_helper::flush_sinks()
{

    for (auto &s : _sinks)
    {
        try
        {
            s->flush();
        }
        SPDLOG_CATCH_AND_HANDLE
    }
    _last_flush = os::now();
}