<commit_before>//----------------------------------------------------------------------------
/// \file  multi_file_async_logger.cpp
//----------------------------------------------------------------------------
/// \brief Multi-file asynchronous logger
///
/// The logger logs data to multiple streams asynchronously
/// It is optimized for performance of the producer to ensure minimal latency.
/// The producer of log messages never blocks during submission of a message.
//----------------------------------------------------------------------------
// Copyright (c) 2012 Omnibius, LLC
// Author: <NAME> <<EMAIL>>
// Created: 2012-03-20
//----------------------------------------------------------------------------
/*
***** BEGIN LICENSE BLOCK *****

This file is part of utxx open-source project.

Copyright (C) 2012 <NAME> <<EMAIL>>

This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

***** END LICENSE BLOCK *****
*/

#ifndef _UTXX_ASYNC_FILE_LOGGER_HPP_
#define _UTXX_ASYNC_FILE_LOGGER_HPP_

#include <utxx/config.h>

#include <utxx/alloc_cached.hpp>
#include <utxx/string.hpp>
#include <utxx/synch.hpp>
#include <utxx/compiler_hints.hpp>
#include <utxx/time_val.hpp>
#include <utxx/logger.hpp>
#include <iostream>
#include <memory>
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <string>
#include <stdio.h>
#include <stdarg.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <fcntl.h>
#include <sched.h>

#ifdef PERF_STATS
#include <utxx/perf_histogram.hpp>
#endif

namespace utxx {

#if DEBUG_ASYNC_LOGGER == 2
#   include <utxx/timestamp.hpp>
#   define ASYNC_DEBUG_TRACE(x) do { printf x; fflush(stdout); } while(0)
#   define ASYNC_TRACE(x)
#elif defined(DEBUG_ASYNC_LOGGER)
#   define ASYNC_TRACE(x) do { printf x; fflush(stdout); } while(0)
#   define ASYNC_DEBUG_TRACE(x) ASYNC_TRACE(x)
#else
#   define ASYNC_TRACE(x)
#   define ASYNC_DEBUG_TRACE(x)
#endif

/// Traits of asynchronous logger
struct multi_file_async_logger_traits {
    typedef std::allocator<char>      allocator;
    typedef std::allocator<char>      fixed_size_allocator;
    typedef futex                     event_type;
    static const int commit_timeout = 2000;  // commit this number of usec
};

/// Multi-stream asynchronous message logger
template<typename traits = multi_file_async_logger_traits>
struct basic_multi_file_async_logger {
    /// Command sent to basic_multi_file_async_logger by message producers
    struct command_t;

    /// Stream information associated with a file descriptor
    /// used internally by the async logger
    class   stream_info;

    /// Internal stream identifier
    class   file_id;

    /// Custom destinations (other than regular files) can keep a pointer
    /// to their state associated with file_id structure
    class   stream_state_base {};

    /// Callback called before writing data to disk. It gives the last opportunity
    /// to rewrite the content written to disk and if necessary to reallocate the
    /// message buffer in a_msg using logger's allocate() and deallocate()
    /// functions.  The returned iovec will be used as the content written to disk.
    typedef std::function<
        iovec (const std::string& a_category, iovec& a_msg)
    >                                               msg_formatter;

    /// Callback called on writing scattered array of iovec structures to stream
    /// Implementation defaults to writev(3). On success the function number of
    /// bytes written or -1 or error.
    typedef std::function<
        int (stream_info& a_si, const char** a_categories,
             const iovec* a_data, size_t a_size)
    >                                               msg_writer;

    typedef std::function<
        int (stream_info& a_si, int a_errno,
             const std::string& a_err)>             err_handler;

    /// Callback executed when stream needs to be reconnected
    typedef std::function<int(stream_info& a_si)>   stream_reconnecter;

    typedef typename traits::event_type             event_type;
    typedef std::function<
        int (const std::string& name,
             stream_state_base* state,
             std::string& error)
    >                                               stream_opener;
    typedef synch::posix_event                      close_event_type;
    typedef std::shared_ptr<close_event_type>       close_event_type_ptr;
    typedef typename traits::allocator::template
        rebind<char>::other                         msg_allocator;

private:
    struct stream_info_eq {
        bool operator()(const stream_info* a, const stream_info* b) { return a == b; }
    };
    struct stream_info_lt {
        bool operator()(const stream_info* a, const stream_info* b) { return a < b; }
    };

    typedef typename traits::fixed_size_allocator::template
        rebind<stream_info*>::other                 ptr_allocator;
    typedef std::set<
        stream_info*, stream_info_lt, ptr_allocator
    >                                               pending_data_streams_set;


    typedef std::vector<stream_info*>               stream_info_vec;
    typedef typename traits::fixed_size_allocator::template
        rebind<command_t>::other                    cmd_allocator;

    std::mutex                                      m_mutex;
    std::condition_variable                         m_cond_var;
    std::shared_ptr<std::thread>                    m_thread;
    cmd_allocator                                   m_cmd_allocator;
    msg_allocator                                   m_msg_allocator;
    std::atomic<command_t*>                         m_head;
    std::atomic<bool>                               m_cancel;
    int                                             m_max_queue_size;
    std::atomic<long>                               m_total_msgs_processed;
    event_type                                      m_event;
    std::atomic<long>                               m_active_count;
    stream_info_vec                                 m_files;
    pending_data_streams_set                        m_pending_data_streams;
    int                                             m_last_version;
    double                                          m_reconnect_sec;
    err_handler                                     m_err_handler;

#ifdef PERF_STATS
    std::atomic<size_t>                             m_stats_enque_spins;
    std::atomic<size_t>                             m_stats_deque_spins;
#endif

    // Default output writer
    static int writev(stream_info& a_si, const char** a_categories,
                      const iovec* a_iovec, size_t a_sz);

    // Register a file or a stream with the logger
    file_id internal_register_stream(
        const std::string&  a_name,
        msg_writer          a_writer,
        stream_state_base*  a_state,
        int                 a_fd);

    bool internal_update_stream(stream_info* a_si, int a_fd);

    // Invoked by the async thread to flush messages from queue to file
    int  commit(const struct timespec* tsp = NULL);
    // Invoked by the async thread
    void run();
    // Enqueues msg to internal queue
    int  internal_enqueue(command_t* a_cmd, const stream_info* a_si);
    // Writes data to internal queue
    int  internal_write(const file_id& a_id, const std::string& a_category,
                        char* a_data, size_t a_sz, bool copied);

    void internal_close();
    void internal_close(stream_info* p, int a_errno = 0);

    command_t* allocate_message(const stream_info* a_si, const std::string& a_category,
                                const char* a_data, size_t a_size)
    {
        command_t* p = m_cmd_allocator.allocate(1);
        ASYNC_TRACE(("Allocated message (category=%s, size=%lu): %p\n",
                     a_category.c_str(), a_size, p));
        new (p) command_t(a_si, a_category, a_data, a_size);
        return p;
    }

    command_t* allocate_command(typename command_t::type_t a_tp, const stream_info* a_si) {
        command_t* p = m_cmd_allocator.allocate(1);
        ASYNC_TRACE(("Allocated command (type=%d): %p\n", a_tp, p));
        new (p) command_t(a_tp, a_si);
        return p;
    }

    void deallocate_command(command_t* a_cmd);

    // Write enqueued messages from a_si->begin() till a_end
    int do_writev_and_free(stream_info* a_si, command_t* a_end,
                           const char* a_categories[],
                           const iovec* a_wr_iov, size_t a_sz);

    bool check_range(int a_fd) const {
        return likely(a_fd >= 0 && (size_t)a_fd < m_files.size());
    }
    bool check_range(const file_id& a_id) const {
        if (!a_id) return false;
        int fd = a_id.fd();
        return check_range(fd) && m_files[fd] && a_id.version() == m_files[fd]->version;
    }
public:
    /// Create instance of this logger
    /// @param a_max_files is the max number of file descriptors
    /// @param a_reconnect_msec is the stream reconnection delay
    /// @param alloc is the message allocator to use
    explicit basic_multi_file_async_logger(
        size_t a_max_files = 1024,
        int    a_reconnect_msec = 5000,
        const msg_allocator& alloc = msg_allocator());

    ~basic_multi_file_async_logger() {
        stop();
    }

    /// Initialize and start asynchronous file writer
    /// @param filename     - name of the file
    /// @param max_msg_size - expected maximum message size (used to allocate
    ///                       the buffer on the stack to format the
    ///                       output string when calling fwrite function)
    /// @param mode         - set file access mode
    int  start();

    /// Stop asynchronous file writing thread
    void stop();

    /// Returns true if the async logger's thread is running
    bool running() { return m_thread.get(); }

    /// Start a new log file
    /// @param a_filename is the name of the output file
    /// @param a_append   if true the file is open in append mode
    /// @param a_mode     file permission mode (default 660)
    file_id open_file
    (
        const std::string& a_filename,
        bool               a_append = true,
        int                a_mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP
    );

    /// Same as open_file() but throws io_error exception on error 
    file_id open_file_or_throw
    (
        const std::string& a_filename,
        bool               a_append = true,
        int                a_mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP
    );

    /// Start a new logging stream
    ///
    /// The logger won't write any data to file but will call \a a_writer
    /// callback on every iovec to be written to stream. It's the caller's
    /// responsibility to perform the actual writing.
    /// @param a_name   is the name of the stream
    /// @param a_writer a callback executed on writing messages to stream
    /// @param a_state  custom state available to the implementation's writer
    /// @param a_fd     is an optional file descriptor associated with the stream
    file_id open_stream
    (
        const std::string&  a_name,
        msg_writer          a_writer,
        stream_state_base*  a_state = NULL,
        int                 a_fd = -1
    );

    /// Same as open_stream() but throws io_error exception on error
    file_id open_stream_or_throw
    (
        const std::string&  a_name,
        msg_writer          a_writer,
        stream_state_base*  a_state = NULL,
        int                 a_fd = -1
    );

    /// This callback will be called from within the logger's thread
    /// to format a message prior to writing it to a log file. The formatting
    /// function can modify the content of \a a_data and if needed can reallocate
    /// and rewrite \a a_data and \a a_size to point to some other memory. Use
    /// the allocate() and deallocate() functions for this purpose. You should
    /// call this function immediately after calling open_file() and before
    /// writing any messages to it.
    void set_formatter(file_id& a_id, msg_formatter a_formatter);

    /// This callback will be called from within the logger's thread
    /// to write an array of iovec structures to a log file. You should
    /// call this function immediately after calling open_file() and before
    /// writing any messages to it.
    void set_writer(file_id& a_id, msg_writer a_writer);

    /// This callback will be called from within the logger's thread
    /// to report a write error on a given file_id. You should
    /// call this function immediately after calling open_file() and before
    /// writing any messages to it.
    void set_error_handler(const err_handler& a_handler);

    /// Set the size of a batch used to write messages to file using on_write
    /// function. Valid value is between 1 and IOV_MAX.
    /// Call this function immediately after calling open_file() and before
    /// writing any messages to it.
    void set_batch_size(file_id& a_id, size_t a_size);

    /// Set a callback for reconnecting to stream
    void set_reconnect(file_id& a_id, stream_reconnecter a_reconnector);

    /// Close one log file
    /// @param a_id identifier of the file to be closed. After return the value
    ///             will be reset.
    /// @param a_immediate indicates whether to close file immediately or
    ///             first to write all pending data in the file descriptor's
    ///             queue prior to closing it
    /// @param a_notify_when_closed is the event that, if not NULL, will be
    ///             signaled when the file descriptor is eventually closed.
    /// @param a_wait_secs is the number of seconds to wait until the file is
    ///             closed. -1 means indefinite.
    int close_file(file_id& a_id, bool a_immediate = false,
                   int a_wait_secs = -1);

    /// @return last error of a given file
    int last_error(const file_id& a_id) const {
        return check_range(a_id) ? m_files[a_id.fd()]->error : -1;
    }

    template <class T>
    T* allocate() {
        T* p = reinterpret_cast<T*>(m_msg_allocator.allocate(sizeof(T)));
        ASYNC_TRACE(("+allocate<T>(%lu) -> %p\n", sizeof(T), p));
        return p;
    }

    /// Allocate a message that can be passed to the loggers' write() function
    /// as its \a a_data argument
    char* allocate(size_t a_sz) {
        char* p = m_msg_allocator.allocate(a_sz);
        ASYNC_TRACE(("+allocate(%lu) -> %p\n", a_sz, p));
        return p;
    }

    /// Deallocate a message previously allocated by the call to allocate()
    void deallocate(char* a_data, size_t a_size) {
        ASYNC_TRACE(("-Deallocating msg(%p, %lu)\n", a_data, a_size));
        m_msg_allocator.deallocate(a_data, a_size);
    }

    /// Formatted write with argumets.  The a_data must have been
    /// previously allocated using the allocate() function.
    /// The logger will implicitely own the pointer and
    /// will have the deallocation responsibility.
    int write(const file_id& a_id, const std::string& a_category, void* a_data, size_t a_sz);

    /// Write a copy of the string a_data to a file.
    int write(const file_id& a_id, const std::string& a_category, const std::string& a_msg);

    /// @return max size of the commit queue
    const int   max_queue_size()        const { return m_max_queue_size; }
    const long  total_msgs_processed()  const { return m_total_msgs_processed
                                                .load(std::memory_order_relaxed); }
    const int   open_files_count()      const { return m_active_count
                                                .load(std::memory_order_relaxed); }
    /// Signaling event that can be used to wake up the logging I/O thread
    const event_type& event()           const { return m_event; }

    /// True when the logger has unprocessed data in its queue
    bool  has_pending_data()            const { return m_head
                                                .load(std::memory_order_relaxed); }
#ifdef PERF_STATS
    size_t stats_enque_spins()           const { return m_stats_enque_spins
                                                .load(std::memory_order_relaxed); }
    size_t stats_deque_spins()           const { return m_stats_deque_spins
                                                .load(std::memory_order_relaxed); }
#endif

};

/// Default implementation of multi_file_async_logger
typedef basic_multi_file_async_logger<> multi_file_async_logger;

//-----------------------------------------------------------------------------
// Local classes
//-----------------------------------------------------------------------------

template<typename traits>
struct basic_multi_file_async_logger<traits>::
command_t {
    static const int MAX_CAT_LEN = 32;
    enum type_t {
        msg,            // Send data message
        close,          // Close stream
        destroy_stream  // Destroy stream object
    };

    const char* type_str() const {
        static const char* s_types[] = {"msg", "close", "destroy_stream"};
        return s_types[type];
    }

    union udata {
        struct {
            mutable iovec       data;
            mutable std::string category;
        } msg;
        struct {
            bool                immediate;
        } close;

        udata()  {}
        ~udata() {}
    };

    const type_t        type;
    const stream_info*  stream;
    udata               args;
    mutable command_t*  next;
    mutable command_t*  prev;

    command_t(const stream_info* a_si, const std::string& a_category,
              const char* a_data, size_t a_size)
        : command_t(msg, a_si)
    {
        set_category(a_category);
        args.msg.data.iov_base = (void*)a_data;
        args.msg.data.iov_len  = a_size;
    }

    command_t(type_t a_type, const stream_info* a_si)
        : type(a_type), stream(a_si), next(NULL), prev(NULL)
    {}

    ~command_t() {
        if (type == msg)
            args.msg.category.~basic_string();
    }

    int fd() const { return stream->fd; }

    void unlink() {
        if (prev) prev->next = next;
        if (next) next->prev = prev;
    }
private:
    void set_category(const std::string& a_category) {
        new (&args.msg.category) std::string(a_category);
    }
} __attribute__((aligned(UTXX_CL_SIZE)));

template<typename traits>
class basic_multi_file_async_logger<traits>::
file_id {
    stream_info* m_stream;
public:
    file_id() { reset(); }
    file_id(stream_info* a_si) : m_stream(a_si) {}
    bool invalid() const    { return !m_stream || m_stream->fd < 0; }
    int  version() const    { BOOST_ASSERT(m_stream); return m_stream->version; }
    int  fd()      const    { return likely(m_stream) ? m_stream->fd : -1; }

    stream_info*        stream()        { return m_stream; }
    const stream_info*  stream() const  { return m_stream; }

    void                reset()         { m_stream = NULL; }
    bool                operator==(const file_id& a) { return m_stream && a.m_stream; }

    operator            bool()   const  { return !invalid(); }
};

/// Stream information associated with a file descriptor
/// used internally by the async logger
template<typename traits>
class basic_multi_file_async_logger<traits>::
stream_info {
    basic_multi_file_async_logger<traits>*  m_logger;
    // This transient list stores commands that are to be written
    // to the stream represented by this stream_info structure
    command_t*                              m_pending_writes_head;
    command_t*                              m_pending_writes_tail;

    // Time of last reconnect attempt
    time_val                                m_last_reconnect_attempt;
    close_event_type_ptr                    on_close;      // Event to signal on close
    msg_formatter                           on_format;     // "before-write" formatter
    msg_writer                              on_write;      // Message writer functor
    stream_reconnecter                      on_reconnect;  // Stream reconnecter

    template <typename T> friend struct basic_multi_file_async_logger;

public:
    std::string          name;
    int                  fd;
    int                  error;
    std::string          error_msg;
    int                  version;       // Version number assigned when file is opened.
    size_t               max_batch_sz;  // Max number of messages to be batched
    stream_state_base*   state;

    explicit stream_info(stream_state_base* a_state = NULL);

    stream_info(basic_multi_file_async_logger<traits>* a_logger,
                const std::string& a_name, int a_fd, int a_version,
                msg_writer a_writer = &basic_multi_file_async_logger<traits>::writev,
                stream_state_base* a_state = NULL);

    ~stream_info() { reset(); }

    static iovec def_on_format(const std::string& a_category, iovec& a_msg) { return a_msg; }

    void reset(int a_errno = 0);

    stream_info* reset(const std::string& a_name, msg_writer a_writer,
                       stream_state_base* a_state, int a_fd);

    void set_error(int a_errno, const char* a_err = NULL);

    /// Push a list of commands to the internal pending queue in reverse order
    ///
    /// The commands are pushed as long as they are destined to this stream.
    /// This method is not thread-safe, it's meant for internal use.
    /// @return number of commands enqueued. Upon return \a a_cmd is updated
    ///         with the first command not belonging to this stream or NULL if no
    ///         such command is found in the list
    int push(const command_t*& a_cmd);

    /// Returns true of internal queue is empty
    bool             pending_queue_empty() const        { return !m_pending_writes_head; }

    command_t*&      pending_writes_head()              { return m_pending_writes_head; }
    const command_t* pending_writes_tail() const        { return m_pending_writes_tail; }
    void             pending_writes_head(command_t* a)  { m_pending_writes_head = a; }
    void             pending_writes_tail(command_t* a)  { m_pending_writes_tail = a; }

    const time_val& last_reconnect_attempt() const      { return m_last_reconnect_attempt; }

    /// Erase single \a item command from the internal queue
    void erase(command_t* item);

    /// Erase commands from \a first till \a end from the internal
    /// queue of pending commands
    void erase(command_t* first, const command_t* end);
};

//-----------------------------------------------------------------------------
// Implementation: stream_info
//-----------------------------------------------------------------------------

template<typename traits>
basic_multi_file_async_logger<traits>::
stream_info::stream_info(stream_state_base* a_state)
    : m_logger(NULL)
    , m_pending_writes_head(NULL), m_pending_writes_tail(NULL)
    , on_format(&stream_info::def_on_format)
    , on_write(&basic_multi_file_async_logger<traits>::writev)
    , fd(-1), error(0), version(0), max_batch_sz(IOV_MAX)
    , state(a_state)
{}

template<typename traits>
basic_multi_file_async_logger<traits>::
stream_info::stream_info(
    basic_multi_file_async_logger<traits>* a_logger,
    const std::string& a_name, int a_fd, int a_version,
    msg_writer a_writer,
    stream_state_base* a_state
)   : m_logger(a_logger)
    , m_pending_writes_head(NULL), m_pending_writes_tail(NULL)
    , on_format(&stream_info::def_on_format)
    , on_write(a_writer)
    , name(a_name), fd(a_fd), error(0)
    , version(a_version), max_batch_sz(IOV_MAX)
    , state(a_state)
{}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::reset(int a_errno) {
    ASYNC_TRACE(("Resetting stream %p (fd=%d)\n", this, fd));
    state = NULL;

    if (a_errno >= 0)
        set_error(a_errno, NULL);

    if (fd != -1) {
        (void)::close(fd);
        fd = -1;
    }

    if (on_close) {
        on_close->signal();
        on_close.reset();
    }
}

template<typename traits>
typename basic_multi_file_async_logger<traits>::stream_info*
basic_multi_file_async_logger<traits>::
stream_info::reset(const std::string& a_name, msg_writer a_writer,
                   stream_state_base* a_state, int a_fd)
{
#ifdef PERF_STATS
    m_stats_enque_spins.store(0u);
    m_stats_deque_spins.store(0u);
#endif
    name     = a_name;
    fd       = a_fd;
    error    = 0;
    state    = a_state;
    on_write = a_writer;
    return this;
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::set_error(int a_errno, const char* a_err) {
    error_msg = a_err ? a_err : (a_errno ? errno_string(a_errno) : std::string());
    error     = a_errno;
}

template<typename traits>
int basic_multi_file_async_logger<traits>::
stream_info::push(const command_t*& a_cmd) {
    int n = 0;
    // Reverse the list
    command_t* p = const_cast<command_t*>(a_cmd), *last = NULL;
    for (; p && p->stream == this; ++n) {
        p->prev = p->next;
        p->next = last;
        last    = p;
        p       = p->prev; // Former p->next

        ASYNC_TRACE(("  FD[%d]: caching cmd (tp=%s) %p (prev=%p, next=%p)\n",
                        fd, last->type_str(), last, last->prev, last->next));
    }

    if (!last)
        return 0;

    last->prev = m_pending_writes_tail;

    if (!m_pending_writes_head)
        m_pending_writes_head = last;

    if (m_pending_writes_tail)
        m_pending_writes_tail->next = last;

    m_pending_writes_tail = const_cast<command_t*>(a_cmd);

    ASYNC_TRACE(("  FD=%d cache head=%p tail=%p\n", fd,
                    m_pending_writes_head, m_pending_writes_tail));

    a_cmd = p;

    return n;
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::erase(command_t* item) {
    if (m_pending_writes_head == item) m_pending_writes_head = item->next;
    if (m_pending_writes_tail == item) m_pending_writes_tail = item->prev;
    if (item->prev) item->prev->next = item->next;
    if (item->next) item->next->prev = item->prev;
    m_logger->deallocate_command(item);
}

/// Erase commands from \a first till \a end from the internal
/// queue of pending commands
template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::erase(command_t* first, const command_t* end) {
    ASYNC_TRACE(("xxx stream_info(%p)::erase: purging items [%p .. %p) from queue\n",
                 this, first, end));
    for (command_t* p = first, *next; p != end; p = next) {
        next = p->next;
        m_logger->deallocate_command(p);
    }
    if (end) end->prev = NULL;
}

//-----------------------------------------------------------------------------
// Implementation: basic_multi_file_async_logger
//-----------------------------------------------------------------------------

template<typename traits>
basic_multi_file_async_logger<traits>::
basic_multi_file_async_logger(
    size_t a_max_files, int a_reconnect_msec, const msg_allocator& alloc)
    : m_thread(nullptr)
    , m_msg_allocator(alloc)
    , m_head(nullptr)
    , m_cancel(false)
    , m_max_queue_size(0)
    , m_total_msgs_processed(0)
    , m_event(0)
    , m_active_count(0)
    , m_files(a_max_files, nullptr)
    , m_last_version(0)
    , m_reconnect_sec((double)a_reconnect_msec / 1000)
#ifdef PERF_STATS
    , m_stats_enque_spins(0)
    , m_stats_deque_spins(0)
#endif
{}

template<typename traits>
inline int basic_multi_file_async_logger<traits>::
writev(stream_info& a_si, const char** a_categories, const iovec* a_iovec, size_t a_sz)
{
#ifdef PERF_NO_WRITEV
    return a_sz;
#else
    return a_si.fd < 0 ? 0 : ::writev(a_si.fd, a_iovec, a_sz);
#endif
}

template<typename traits>
int basic_multi_file_async_logger<traits>::
start() {
    std::unique_lock<std::mutex> lock(m_mutex);

    if (running())
        return -1;

    m_event.reset();
    m_cancel = false;

    m_thread.reset(
        new std::thread(
            std::bind(&basic_multi_file_async_logger<traits>::run, this))
    );

    m_cond_var.wait(lock);

    return 0;
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stop() {
    if (!running())
        return;

    ASYNC_TRACE((">>> Stopping async logger (head %p)\n", m_head.load()));

    std::shared_ptr<std::thread> t = m_thread;
    if (t) {
        m_cancel.store(true, std::memory_order_release);
        m_event.signal();

        t->join();
    }
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
run() {
    // Notify the caller that we are ready
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_cond_var.notify_all();
    }

    ASYNC_TRACE(("Started async logging thread (cancel=%s)\n",
        m_cancel ? "true" : "false"));

    static const timespec ts =
        {traits::commit_timeout / 1000, (traits::commit_timeout % 1000) * 1000000 };

    m_total_msgs_processed = 0;

    while (true) {
        #if defined(DEBUG_ASYNC_LOGGER) && DEBUG_ASYNC_LOGGER != 2
        int rc =
        #endif
        commit(&ts);

        ASYNC_TRACE(( "Async thread commit result: %d (head: %p, cancel=%s)\n",
            rc, m_head.load(), m_cancel ? "true" : "false" ));

        // CPU-friendly spin for 250us
        time_val deadline(rel_time(0, 250));
        while (!m_head.load(std::memory_order_relaxed)) {
            if (m_cancel.load(std::memory_order_relaxed))
                goto DONE;
            if (now_utc() > deadline)
                break;
            sched_yield();
        }
    }

DONE:
    ASYNC_TRACE(("Logger loop finished - calling close()\n"));
    internal_close();<commit_msg>Parameterize the use of sched_yield()<commit_after>//----------------------------------------------------------------------------
/// \file  multi_file_async_logger.cpp
//----------------------------------------------------------------------------
/// \brief Multi-file asynchronous logger
///
/// The logger logs data to multiple streams asynchronously
/// It is optimized for performance of the producer to ensure minimal latency.
/// The producer of log messages never blocks during submission of a message.
//----------------------------------------------------------------------------
// Copyright (c) 2012 Omnibius, LLC
// Author: <NAME> <<EMAIL>>
// Created: 2012-03-20
//----------------------------------------------------------------------------
/*
***** BEGIN LICENSE BLOCK *****

This file is part of utxx open-source project.

Copyright (C) 2012 <NAME> <<EMAIL>>

This library is free software; you can redistribute it and/or
modify it under the terms of the GNU Lesser General Public
License as published by the Free Software Foundation; either
version 2.1 of the License, or (at your option) any later version.

This library is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
Lesser General Public License for more details.

You should have received a copy of the GNU Lesser General Public
License along with this library; if not, write to the Free Software
Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

***** END LICENSE BLOCK *****
*/

#ifndef _UTXX_ASYNC_FILE_LOGGER_HPP_
#define _UTXX_ASYNC_FILE_LOGGER_HPP_

#include <utxx/config.h>

#include <utxx/alloc_cached.hpp>
#include <utxx/string.hpp>
#include <utxx/synch.hpp>
#include <utxx/compiler_hints.hpp>
#include <utxx/time_val.hpp>
#include <utxx/logger.hpp>
#include <iostream>
#include <memory>
#include <atomic>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <chrono>
#include <string>
#include <stdio.h>
#include <stdarg.h>
#include <sys/socket.h>
#include <sys/types.h>
#include <sys/uio.h>
#include <fcntl.h>
#include <sched.h>

#ifdef PERF_STATS
#include <utxx/perf_histogram.hpp>
#endif

namespace utxx {

#if DEBUG_ASYNC_LOGGER == 2
#   include <utxx/timestamp.hpp>
#   define ASYNC_DEBUG_TRACE(x) do { printf x; fflush(stdout); } while(0)
#   define ASYNC_TRACE(x)
#elif defined(DEBUG_ASYNC_LOGGER)
#   define ASYNC_TRACE(x) do { printf x; fflush(stdout); } while(0)
#   define ASYNC_DEBUG_TRACE(x) ASYNC_TRACE(x)
#else
#   define ASYNC_TRACE(x)
#   define ASYNC_DEBUG_TRACE(x)
#endif

/// Traits of asynchronous logger
struct multi_file_async_logger_traits {
    typedef std::allocator<char>      allocator;
    typedef std::allocator<char>      fixed_size_allocator;
    typedef futex                     event_type;
    static const int commit_timeout = 2000;  // commit this number of usec
};

/// Multi-stream asynchronous message logger
template<typename traits = multi_file_async_logger_traits>
struct basic_multi_file_async_logger {
    /// Command sent to basic_multi_file_async_logger by message producers
    struct command_t;

    /// Stream information associated with a file descriptor
    /// used internally by the async logger
    class   stream_info;

    /// Internal stream identifier
    class   file_id;

    /// Custom destinations (other than regular files) can keep a pointer
    /// to their state associated with file_id structure
    class   stream_state_base {};

    /// Callback called before writing data to disk. It gives the last opportunity
    /// to rewrite the content written to disk and if necessary to reallocate the
    /// message buffer in a_msg using logger's allocate() and deallocate()
    /// functions.  The returned iovec will be used as the content written to disk.
    typedef std::function<
        iovec (const std::string& a_category, iovec& a_msg)
    >                                               msg_formatter;

    /// Callback called on writing scattered array of iovec structures to stream
    /// Implementation defaults to writev(3). On success the function number of
    /// bytes written or -1 or error.
    typedef std::function<
        int (stream_info& a_si, const char** a_categories,
             const iovec* a_data, size_t a_size)
    >                                               msg_writer;

    typedef std::function<
        int (stream_info& a_si, int a_errno,
             const std::string& a_err)>             err_handler;

    /// Callback executed when stream needs to be reconnected
    typedef std::function<int(stream_info& a_si)>   stream_reconnecter;

    typedef typename traits::event_type             event_type;
    typedef std::function<
        int (const std::string& name,
             stream_state_base* state,
             std::string& error)
    >                                               stream_opener;
    typedef synch::posix_event                      close_event_type;
    typedef std::shared_ptr<close_event_type>       close_event_type_ptr;
    typedef typename traits::allocator::template
        rebind<char>::other                         msg_allocator;

private:
    struct stream_info_eq {
        bool operator()(const stream_info* a, const stream_info* b) { return a == b; }
    };
    struct stream_info_lt {
        bool operator()(const stream_info* a, const stream_info* b) { return a < b; }
    };

    typedef typename traits::fixed_size_allocator::template
        rebind<stream_info*>::other                 ptr_allocator;
    typedef std::set<
        stream_info*, stream_info_lt, ptr_allocator
    >                                               pending_data_streams_set;


    typedef std::vector<stream_info*>               stream_info_vec;
    typedef typename traits::fixed_size_allocator::template
        rebind<command_t>::other                    cmd_allocator;

    std::mutex                                      m_mutex;
    std::condition_variable                         m_cond_var;
    std::shared_ptr<std::thread>                    m_thread;
    cmd_allocator                                   m_cmd_allocator;
    msg_allocator                                   m_msg_allocator;
    std::atomic<command_t*>                         m_head;
    std::atomic<bool>                               m_cancel;
    int                                             m_max_queue_size;
    std::atomic<long>                               m_total_msgs_processed;
    event_type                                      m_event;
    std::atomic<long>                               m_active_count;
    stream_info_vec                                 m_files;
    pending_data_streams_set                        m_pending_data_streams;
    int                                             m_last_version;
    double                                          m_reconnect_sec;
    err_handler                                     m_err_handler;
    bool                                            m_use_sched_yield;
#ifdef PERF_STATS
    std::atomic<size_t>                             m_stats_enque_spins;
    std::atomic<size_t>                             m_stats_deque_spins;
#endif

    // Default output writer
    static int writev(stream_info& a_si, const char** a_categories,
                      const iovec* a_iovec, size_t a_sz);

    // Register a file or a stream with the logger
    file_id internal_register_stream(
        const std::string&  a_name,
        msg_writer          a_writer,
        stream_state_base*  a_state,
        int                 a_fd);

    bool internal_update_stream(stream_info* a_si, int a_fd);

    // Invoked by the async thread to flush messages from queue to file
    int  commit(const struct timespec* tsp = NULL);
    // Invoked by the async thread
    void run();
    // Enqueues msg to internal queue
    int  internal_enqueue(command_t* a_cmd, const stream_info* a_si);
    // Writes data to internal queue
    int  internal_write(const file_id& a_id, const std::string& a_category,
                        char* a_data, size_t a_sz, bool copied);

    void internal_close();
    void internal_close(stream_info* p, int a_errno = 0);

    command_t* allocate_message(const stream_info* a_si, const std::string& a_category,
                                const char* a_data, size_t a_size)
    {
        command_t* p = m_cmd_allocator.allocate(1);
        ASYNC_TRACE(("Allocated message (category=%s, size=%lu): %p\n",
                     a_category.c_str(), a_size, p));
        new (p) command_t(a_si, a_category, a_data, a_size);
        return p;
    }

    command_t* allocate_command(typename command_t::type_t a_tp, const stream_info* a_si) {
        command_t* p = m_cmd_allocator.allocate(1);
        ASYNC_TRACE(("Allocated command (type=%d): %p\n", a_tp, p));
        new (p) command_t(a_tp, a_si);
        return p;
    }

    void deallocate_command(command_t* a_cmd);

    // Write enqueued messages from a_si->begin() till a_end
    int do_writev_and_free(stream_info* a_si, command_t* a_end,
                           const char* a_categories[],
                           const iovec* a_wr_iov, size_t a_sz);

    bool check_range(int a_fd) const {
        return likely(a_fd >= 0 && (size_t)a_fd < m_files.size());
    }
    bool check_range(const file_id& a_id) const {
        if (!a_id) return false;
        int fd = a_id.fd();
        return check_range(fd) && m_files[fd] && a_id.version() == m_files[fd]->version;
    }
public:
    /// Create instance of this logger
    /// @param a_max_files is the max number of file descriptors
    /// @param a_reconnect_msec is the stream reconnection delay
    /// @param alloc is the message allocator to use
    explicit basic_multi_file_async_logger(
        size_t a_max_files = 1024,
        int    a_reconnect_msec = 5000,
        const msg_allocator& alloc = msg_allocator());

    ~basic_multi_file_async_logger() {
        stop();
    }

    /// Initialize and start asynchronous file writer
    /// @param filename     - name of the file
    /// @param max_msg_size - expected maximum message size (used to allocate
    ///                       the buffer on the stack to format the
    ///                       output string when calling fwrite function)
    /// @param mode         - set file access mode
    int  start();

    /// Stop asynchronous file writing thread
    void stop();

    /// Returns true if the async logger's thread is running
    bool running() { return m_thread.get(); }

    /// Start a new log file
    /// @param a_filename is the name of the output file
    /// @param a_append   if true the file is open in append mode
    /// @param a_mode     file permission mode (default 660)
    file_id open_file
    (
        const std::string& a_filename,
        bool               a_append = true,
        int                a_mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP
    );

    /// Same as open_file() but throws io_error exception on error 
    file_id open_file_or_throw
    (
        const std::string& a_filename,
        bool               a_append = true,
        int                a_mode = S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP
    );

    /// Start a new logging stream
    ///
    /// The logger won't write any data to file but will call \a a_writer
    /// callback on every iovec to be written to stream. It's the caller's
    /// responsibility to perform the actual writing.
    /// @param a_name   is the name of the stream
    /// @param a_writer a callback executed on writing messages to stream
    /// @param a_state  custom state available to the implementation's writer
    /// @param a_fd     is an optional file descriptor associated with the stream
    file_id open_stream
    (
        const std::string&  a_name,
        msg_writer          a_writer,
        stream_state_base*  a_state = NULL,
        int                 a_fd = -1
    );

    /// Same as open_stream() but throws io_error exception on error
    file_id open_stream_or_throw
    (
        const std::string&  a_name,
        msg_writer          a_writer,
        stream_state_base*  a_state = NULL,
        int                 a_fd = -1
    );

    /// Set the callback to be used for formatting output data.
    /// It is invoked from within the logger's thread to format a message
    /// prior to writing it to a log file. The formatting function can modify
    /// the content of \a a_data and if needed can reallocate and
    /// rewrite \a a_data and \a a_size to point to some other memory. Use
    /// the allocate() and deallocate() functions for this purpose. You should
    /// call this function immediately after calling open_file() and before
    /// writing any messages to it.
    void set_formatter(file_id& a_id, msg_formatter a_formatter);

    /// Set the callback to be used to write data to file.
    /// The callback is invoked from within the logger's thread.
    /// It writes an array of iovec structures to a log file. It
    /// should be called by user immediately after calling open_file() and
    /// before writing any messages to it.
    void set_writer(file_id& a_id, msg_writer a_writer);

    /// This callback will be called from within the logger's thread
    /// to report a write error on a given file_id. You should
    /// call this function immediately after calling open_file() and before
    /// writing any messages to it.
    void set_error_handler(const err_handler& a_handler);

    /// Set the size of a batch used to write messages to file using on_write
    /// function. Valid value is between 1 and IOV_MAX.
    /// Call this function immediately after calling open_file() and before
    /// writing any messages to it.
    void set_batch_size(file_id& a_id, size_t a_size);

    /// Set a callback for reconnecting to stream
    void set_reconnect(file_id& a_id, stream_reconnecter a_reconnector);

    /// Enable usage of sched_yield() instead of usleep() in the logging thread.
    /// Occasionally when running processing thread on max priority the use of
    /// sched_yield() can cause system resource starvation.
    void use_sched_yield(bool a_enable) { m_use_sched_yield = a_enable; }

    /// Close one log file
    /// @param a_id identifier of the file to be closed. After return the value
    ///             will be reset.
    /// @param a_immediate indicates whether to close file immediately or
    ///             first to write all pending data in the file descriptor's
    ///             queue prior to closing it
    /// @param a_notify_when_closed is the event that, if not NULL, will be
    ///             signaled when the file descriptor is eventually closed.
    /// @param a_wait_secs is the number of seconds to wait until the file is
    ///             closed. -1 means indefinite.
    int close_file(file_id& a_id, bool a_immediate = false,
                   int a_wait_secs = -1);

    /// @return last error of a given file
    int last_error(const file_id& a_id) const {
        return check_range(a_id) ? m_files[a_id.fd()]->error : -1;
    }

    template <class T>
    T* allocate() {
        T* p = reinterpret_cast<T*>(m_msg_allocator.allocate(sizeof(T)));
        ASYNC_TRACE(("+allocate<T>(%lu) -> %p\n", sizeof(T), p));
        return p;
    }

    /// Allocate a message that can be passed to the loggers' write() function
    /// as its \a a_data argument
    char* allocate(size_t a_sz) {
        char* p = m_msg_allocator.allocate(a_sz);
        ASYNC_TRACE(("+allocate(%lu) -> %p\n", a_sz, p));
        return p;
    }

    /// Deallocate a message previously allocated by the call to allocate()
    void deallocate(char* a_data, size_t a_size) {
        ASYNC_TRACE(("-Deallocating msg(%p, %lu)\n", a_data, a_size));
        m_msg_allocator.deallocate(a_data, a_size);
    }

    /// Formatted write with argumets.  The a_data must have been
    /// previously allocated using the allocate() function.
    /// The logger will implicitely own the pointer and
    /// will have the deallocation responsibility.
    int write(const file_id& a_id, const std::string& a_category, void* a_data, size_t a_sz);

    /// Write a copy of the string a_data to a file.
    int write(const file_id& a_id, const std::string& a_category, const std::string& a_msg);

    /// @return max size of the commit queue
    const int   max_queue_size()        const { return m_max_queue_size; }
    const long  total_msgs_processed()  const { return m_total_msgs_processed
                                                .load(std::memory_order_relaxed); }
    const int   open_files_count()      const { return m_active_count
                                                .load(std::memory_order_relaxed); }
    /// Signaling event that can be used to wake up the logging I/O thread
    const event_type& event()           const { return m_event; }

    /// True when the logger has unprocessed data in its queue
    bool  has_pending_data()            const { return m_head
                                                .load(std::memory_order_relaxed); }
#ifdef PERF_STATS
    size_t stats_enque_spins()           const { return m_stats_enque_spins
                                                .load(std::memory_order_relaxed); }
    size_t stats_deque_spins()           const { return m_stats_deque_spins
                                                .load(std::memory_order_relaxed); }
#endif

};

/// Default implementation of multi_file_async_logger
typedef basic_multi_file_async_logger<> multi_file_async_logger;

//-----------------------------------------------------------------------------
// Local classes
//-----------------------------------------------------------------------------

template<typename traits>
struct basic_multi_file_async_logger<traits>::
command_t {
    static const int MAX_CAT_LEN = 32;
    enum type_t {
        msg,            // Send data message
        close,          // Close stream
        destroy_stream  // Destroy stream object
    };

    const char* type_str() const {
        static const char* s_types[] = {"msg", "close", "destroy_stream"};
        return s_types[type];
    }

    union udata {
        struct {
            mutable iovec       data;
            mutable std::string category;
        } msg;
        struct {
            bool                immediate;
        } close;

        udata()  {}
        ~udata() {}
    };

    const type_t        type;
    const stream_info*  stream;
    udata               args;
    mutable command_t*  next;
    mutable command_t*  prev;

    command_t(const stream_info* a_si, const std::string& a_category,
              const char* a_data, size_t a_size)
        : command_t(msg, a_si)
    {
        set_category(a_category);
        args.msg.data.iov_base = (void*)a_data;
        args.msg.data.iov_len  = a_size;
    }

    command_t(type_t a_type, const stream_info* a_si)
        : type(a_type), stream(a_si), next(NULL), prev(NULL)
    {}

    ~command_t() {
        if (type == msg)
            args.msg.category.~basic_string();
    }

    int fd() const { return stream->fd; }

    void unlink() {
        if (prev) prev->next = next;
        if (next) next->prev = prev;
    }
private:
    void set_category(const std::string& a_category) {
        new (&args.msg.category) std::string(a_category);
    }
} __attribute__((aligned(UTXX_CL_SIZE)));

template<typename traits>
class basic_multi_file_async_logger<traits>::
file_id {
    stream_info* m_stream;
public:
    file_id() { reset(); }
    file_id(stream_info* a_si) : m_stream(a_si) {}
    bool invalid() const    { return !m_stream || m_stream->fd < 0; }
    int  version() const    { BOOST_ASSERT(m_stream); return m_stream->version; }
    int  fd()      const    { return likely(m_stream) ? m_stream->fd : -1; }

    stream_info*        stream()        { return m_stream; }
    const stream_info*  stream() const  { return m_stream; }

    void                reset()         { m_stream = NULL; }
    bool                operator==(const file_id& a) { return m_stream && a.m_stream; }

    operator            bool()   const  { return !invalid(); }
};

/// Stream information associated with a file descriptor
/// used internally by the async logger
template<typename traits>
class basic_multi_file_async_logger<traits>::
stream_info {
    basic_multi_file_async_logger<traits>*  m_logger;
    // This transient list stores commands that are to be written
    // to the stream represented by this stream_info structure
    command_t*                              m_pending_writes_head;
    command_t*                              m_pending_writes_tail;

    // Time of last reconnect attempt
    time_val                                m_last_reconnect_attempt;
    close_event_type_ptr                    on_close;      // Event to signal on close
    msg_formatter                           on_format;     // "before-write" formatter
    msg_writer                              on_write;      // Message writer functor
    stream_reconnecter                      on_reconnect;  // Stream reconnecter

    template <typename T> friend struct basic_multi_file_async_logger;

public:
    std::string          name;
    int                  fd;
    int                  error;
    std::string          error_msg;
    int                  version;       // Version number assigned when file is opened.
    size_t               max_batch_sz;  // Max number of messages to be batched
    stream_state_base*   state;

    explicit stream_info(stream_state_base* a_state = NULL);

    stream_info(basic_multi_file_async_logger<traits>* a_logger,
                const std::string& a_name, int a_fd, int a_version,
                msg_writer a_writer = &basic_multi_file_async_logger<traits>::writev,
                stream_state_base* a_state = NULL);

    ~stream_info() { reset(); }

    static iovec def_on_format(const std::string& a_category, iovec& a_msg) { return a_msg; }

    void reset(int a_errno = 0);

    stream_info* reset(const std::string& a_name, msg_writer a_writer,
                       stream_state_base* a_state, int a_fd);

    void set_error(int a_errno, const char* a_err = NULL);

    /// Push a list of commands to the internal pending queue in reverse order
    ///
    /// The commands are pushed as long as they are destined to this stream.
    /// This method is not thread-safe, it's meant for internal use.
    /// @return number of commands enqueued. Upon return \a a_cmd is updated
    ///         with the first command not belonging to this stream or NULL if no
    ///         such command is found in the list
    int push(const command_t*& a_cmd);

    /// Returns true of internal queue is empty
    bool             pending_queue_empty() const        { return !m_pending_writes_head; }

    command_t*&      pending_writes_head()              { return m_pending_writes_head; }
    const command_t* pending_writes_tail() const        { return m_pending_writes_tail; }
    void             pending_writes_head(command_t* a)  { m_pending_writes_head = a; }
    void             pending_writes_tail(command_t* a)  { m_pending_writes_tail = a; }

    const time_val& last_reconnect_attempt() const      { return m_last_reconnect_attempt; }

    /// Erase single \a item command from the internal queue
    void erase(command_t* item);

    /// Erase commands from \a first till \a end from the internal
    /// queue of pending commands
    void erase(command_t* first, const command_t* end);
};

//-----------------------------------------------------------------------------
// Implementation: stream_info
//-----------------------------------------------------------------------------

template<typename traits>
basic_multi_file_async_logger<traits>::
stream_info::stream_info(stream_state_base* a_state)
    : m_logger(NULL)
    , m_pending_writes_head(NULL), m_pending_writes_tail(NULL)
    , on_format(&stream_info::def_on_format)
    , on_write(&basic_multi_file_async_logger<traits>::writev)
    , fd(-1), error(0), version(0), max_batch_sz(IOV_MAX)
    , state(a_state)
{}

template<typename traits>
basic_multi_file_async_logger<traits>::
stream_info::stream_info(
    basic_multi_file_async_logger<traits>* a_logger,
    const std::string& a_name, int a_fd, int a_version,
    msg_writer a_writer,
    stream_state_base* a_state
)   : m_logger(a_logger)
    , m_pending_writes_head(NULL), m_pending_writes_tail(NULL)
    , on_format(&stream_info::def_on_format)
    , on_write(a_writer)
    , name(a_name), fd(a_fd), error(0)
    , version(a_version), max_batch_sz(IOV_MAX)
    , state(a_state)
{}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::reset(int a_errno) {
    ASYNC_TRACE(("Resetting stream %p (fd=%d)\n", this, fd));
    state = NULL;

    if (a_errno >= 0)
        set_error(a_errno, NULL);

    if (fd != -1) {
        (void)::close(fd);
        fd = -1;
    }

    if (on_close) {
        on_close->signal();
        on_close.reset();
    }
}

template<typename traits>
typename basic_multi_file_async_logger<traits>::stream_info*
basic_multi_file_async_logger<traits>::
stream_info::reset(const std::string& a_name, msg_writer a_writer,
                   stream_state_base* a_state, int a_fd)
{
#ifdef PERF_STATS
    m_stats_enque_spins.store(0u);
    m_stats_deque_spins.store(0u);
#endif
    name     = a_name;
    fd       = a_fd;
    error    = 0;
    state    = a_state;
    on_write = a_writer;
    return this;
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::set_error(int a_errno, const char* a_err) {
    error_msg = a_err ? a_err : (a_errno ? errno_string(a_errno) : std::string());
    error     = a_errno;
}

template<typename traits>
int basic_multi_file_async_logger<traits>::
stream_info::push(const command_t*& a_cmd) {
    int n = 0;
    // Reverse the list
    command_t* p = const_cast<command_t*>(a_cmd), *last = NULL;
    for (; p && p->stream == this; ++n) {
        p->prev = p->next;
        p->next = last;
        last    = p;
        p       = p->prev; // Former p->next

        ASYNC_TRACE(("  FD[%d]: caching cmd (tp=%s) %p (prev=%p, next=%p)\n",
                        fd, last->type_str(), last, last->prev, last->next));
    }

    if (!last)
        return 0;

    last->prev = m_pending_writes_tail;

    if (!m_pending_writes_head)
        m_pending_writes_head = last;

    if (m_pending_writes_tail)
        m_pending_writes_tail->next = last;

    m_pending_writes_tail = const_cast<command_t*>(a_cmd);

    ASYNC_TRACE(("  FD=%d cache head=%p tail=%p\n", fd,
                    m_pending_writes_head, m_pending_writes_tail));

    a_cmd = p;

    return n;
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::erase(command_t* item) {
    if (m_pending_writes_head == item) m_pending_writes_head = item->next;
    if (m_pending_writes_tail == item) m_pending_writes_tail = item->prev;
    if (item->prev) item->prev->next = item->next;
    if (item->next) item->next->prev = item->prev;
    m_logger->deallocate_command(item);
}

/// Erase commands from \a first till \a end from the internal
/// queue of pending commands
template<typename traits>
void basic_multi_file_async_logger<traits>::
stream_info::erase(command_t* first, const command_t* end) {
    ASYNC_TRACE(("xxx stream_info(%p)::erase: purging items [%p .. %p) from queue\n",
                 this, first, end));
    for (command_t* p = first, *next; p != end; p = next) {
        next = p->next;
        m_logger->deallocate_command(p);
    }
    if (end) end->prev = NULL;
}

//-----------------------------------------------------------------------------
// Implementation: basic_multi_file_async_logger
//-----------------------------------------------------------------------------

template<typename traits>
basic_multi_file_async_logger<traits>::
basic_multi_file_async_logger(
    size_t a_max_files, int a_reconnect_msec, const msg_allocator& alloc)
    : m_thread(nullptr)
    , m_msg_allocator(alloc)
    , m_head(nullptr)
    , m_cancel(false)
    , m_max_queue_size(0)
    , m_total_msgs_processed(0)
    , m_event(0)
    , m_active_count(0)
    , m_files(a_max_files, nullptr)
    , m_last_version(0)
    , m_reconnect_sec((double)a_reconnect_msec / 1000)
    , m_use_sched_yield(true)
#ifdef PERF_STATS
    , m_stats_enque_spins(0)
    , m_stats_deque_spins(0)
#endif
{}

template<typename traits>
inline int basic_multi_file_async_logger<traits>::
writev(stream_info& a_si, const char** a_categories, const iovec* a_iovec, size_t a_sz)
{
#ifdef PERF_NO_WRITEV
    return a_sz;
#else
    return a_si.fd < 0 ? 0 : ::writev(a_si.fd, a_iovec, a_sz);
#endif
}

template<typename traits>
int basic_multi_file_async_logger<traits>::
start() {
    std::unique_lock<std::mutex> lock(m_mutex);

    if (running())
        return -1;

    m_event.reset();
    m_cancel = false;

    m_thread.reset(
        new std::thread(
            std::bind(&basic_multi_file_async_logger<traits>::run, this))
    );

    m_cond_var.wait(lock);

    return 0;
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
stop() {
    if (!running())
        return;

    ASYNC_TRACE((">>> Stopping async logger (head %p)\n", m_head.load()));

    std::shared_ptr<std::thread> t = m_thread;
    if (t) {
        m_cancel.store(true, std::memory_order_release);
        m_event.signal();

        t->join();
    }
}

template<typename traits>
void basic_multi_file_async_logger<traits>::
run() {
    // Notify the caller that we are ready
    {
        std::unique_lock<std::mutex> lock(m_mutex);
        m_cond_var.notify_all();
    }

    ASYNC_TRACE(("Started async logging thread (cancel=%s)\n",
        m_cancel ? "true" : "false"));

    static const timespec ts =
        {traits::commit_timeout / 1000, (traits::commit_timeout % 1000) * 1000000 };

    m_total_msgs_processed = 0;

    while (true) {
        #if defined(DEBUG_ASYNC_LOGGER) && DEBUG_ASYNC_LOGGER != 2
        int rc =
        #endif
        commit(&ts);

        ASYNC_TRACE(( "Async thread commit result: %d (head: %p, cancel=%s)\n",
            rc, m_head.load(), m_cancel ? "true" : "false" ));

        // CPU-friendly spin for 250us
        time_val deadline(rel_time(0, 250));
        while (!m_head.load(std::memory_order_relaxed)) {
            if (m_cancel.load(std::memory_order_relaxed))
                goto DONE;
            if (now_utc() > deadline)
                break;
            // When running with maximum priority, occasionally excessive use of
            // sched_yield may use to system slowdown, so this option is
            // configurable:
            if (m_use_sched_yield)
                sched_yield();
            else
                usleep(50);
        }
    }

DONE:
    ASYNC_TRACE(("Logger loop finished - calling close()\n"));
    internal_close();
{'fraction_non_alphanumeric': 0.11851610273134937, 'fraction_numerical': 0.003880962087240114, 'mean_word_length': 2.181634241245136, 'pattern_counts': {'":': 0, '<': 211, '<?xml version=': 0, '>': 291, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 62, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3856565', 'n_tokens_mistral': 18617, 'n_tokens_neox': 17969, 'n_words': 5158}