Performance and correctness of my threaded logger?

Question

My logger posted below makes use of condition variables in order to notify the "flusher" thread that new items are available. The notification occurs every time a message is logged.

My questions are:

• If the notification immediately follows the log action then does that mean that the queue will only contain one element at most?

• What would be the difference between:

  {
      std::unique_lock<std::mutex> lock(mMutex);
      mLogItems.push_back(std::make_pair(inLevel, std::move(inMessage)));
      mCondition.notify_one();
  }
  

  and

  {
      std::unique_lock<std::mutex> lock(mMutex);
      mLogItems.push_back(std::make_pair(inLevel, std::move(inMessage)));
  }
  mCondition.notify_one();
  

  And which is correct?

• Should I prefer an approach where flushing occurs at periodic intervals?

• If yes, then should I use a loop where I flush and sleep until a stop condition? Or should sleep() be avoided?

• Is it feasible to use a lockless approach?

Working code sample:

#include <atomic>
#include <condition_variable>
#include <deque>
#include <iostream>
#include <mutex>
#include <sstream>
#include <thread>


enum class LogLevel
{
    Debug,
    Info,
    Warning,
    Error
};


const char * ToString(LogLevel inLogLevel)
{
    switch (inLogLevel)
    {
        case LogLevel::Debug:   return "DEBUG";
        case LogLevel::Info:    return "INFO";
        case LogLevel::Warning: return "WARNING";
        case LogLevel::Error:   return "ERROR";
        default: return "UNKNOWN";
    }
}


class Logger
{
public:
    static Logger & Instance()
    {
        static Logger fInstance;
        return fInstance;
    }

    ~Logger()
    {
        setQuit();
    }

    void setQuit()
    {
        std::unique_lock<std::mutex> lock(mMutex);
        mQuit = true;
        mCondition.notify_one();
    }

    void setTreshold(LogLevel inLogLevel)
    {
        mTreshold = inLogLevel;
    }

    LogLevel getTreshold() const
    {
        return mTreshold;
    }

    bool isAllowed(LogLevel inLogLevel) const
    {
        bool result = inLogLevel >= mTreshold;
        return result;
    }

    typedef std::pair<LogLevel, std::string> LogItem;
    typedef std::deque<LogItem> LogItems;

    void log(LogLevel inLevel, std::string inMessage)
    {
        if (mQuit)
        {
            return;
        }

        {
            std::unique_lock<std::mutex> lock(mMutex);
            mLogItems.push_back(std::make_pair(inLevel, std::move(inMessage)));
            mCondition.notify_one();
        }
    }

    template<typename PrintFunction>
    void flush(const PrintFunction & inPrintFunction)
    {
        while (!mQuit)
        {
            LogItems items;
            {
                std::unique_lock<std::mutex> lock(mMutex);
                if (mLogItems.empty())
                {
                    mCondition.wait(lock); // unlocks the mutex
                }
                items.swap(mLogItems); // internal pointer swap
            }
            for (LogItems::const_iterator it = items.begin(), end = items.end(); it != end; ++it)
            {
                inPrintFunction(*it);
            }
        }
    }

private:
    std::atomic<bool> mQuit;
    LogLevel mTreshold;
    std::mutex mMutex;
    std::condition_variable mCondition;
    LogItems mLogItems;
};


struct LogHelper
{
    LogHelper(LogLevel inLevel) :
        mLevel(inLevel)
    {
    }

    ~LogHelper()
    {
        Logger::Instance().log(mLevel, mStream.str());
    }

    std::stringstream & ss()
    {
        return mStream;
    }

private:     
    LogLevel mLevel;
    std::stringstream mStream;
};


#define LOG(LEVEL) \
    if (!Logger::Instance().isAllowed(LogLevel::LEVEL)) {} else LogHelper(LogLevel::LEVEL).ss()

#define LogInfo() LOG(Info)
#define LogWarning() LOG(Warning)
#define LogError() LOG(Error)

#define LogDebug() \
    if (!Logger::Instance().isAllowed(LogLevel::Debug)) {} else LogHelper(LogLevel::Debug).ss() << __FILE__ << ":" << __LINE__ << ": "


void Thread(const std::string & inName)
{
    for (unsigned i = 0; i < 5; ++i)
    {
        LOG(Info) << inName << " " << i;
    }
}


struct print_to_ostream
{
    print_to_ostream(std::ostream & os) : os(os) {}

    void operator()(const Logger::LogItem & inLogItem) const
    {
        os << ToString(inLogItem.first) << "\t" << inLogItem.second << std::endl;
    }

    std::ostream & os;
};


int main()
{
    LogDebug() << "Start of program!";

    LogInfo() << "Test info";
    LogWarning() << "Test warning";
    LogError() << "Test error";

    // Start a flush thread
    std::thread flusher([]{ Logger::Instance().flush(print_to_ostream(std::cout)); });


    LogDebug() << "Starting threads...";
    std::thread t1(std::bind(&Thread, "t1"));
    std::thread t2(std::bind(&Thread, "t2"));
    std::thread t3(std::bind(&Thread, "t3"));
    std::thread t4(std::bind(&Thread, "t4"));
    std::thread t5(std::bind(&Thread, "t5"));


    LogDebug() << "Join threads...";
    t1.join();
    t2.join();
    t3.join();
    t4.join();
    t5.join();


    // Cleanup
    LogDebug() << "Cleanup ...";
    Logger::Instance().setQuit();
    flusher.join();
}

Answer 1

The main thing I'm noticing is that you have a "Throw a lock at the problem" approach (and eww mMember).

You really need a true concurrent queue- these can operate atomically (for any T). If you download, say, TBB, it ships with a concurrent_queue you can use. Here you can simply push on members without regard to your concurrency situation, as it is guaranteed to be safe to push- even whilst other threads are pushing, popping, or a bunch of other things. In addition, this protects you from, say, simply forgetting to lock the mutex one day. It also simplifies your flush code considerably.

LogItem l;
while(queue.try_pop(l))
    inPrintFunction(l);

Answer 2

Which is more correct?

{
    std::unique_lock<std::mutex> lock(mMutex);
    mLogItems.push_back(std::make_pair(inLevel, std::move(inMessage)));
    mCondition.notify_one();
}

I would say this one (though if the condition variable is written correctly it should not matter). But I tend to want to stay on the conservative side when coding with threads. Thus any resource that is used by multiple threads should only be accessed when you have a lock to make sure access is exclusive. In my opinion mCondition is a shared resource and thus needs to be only accessed when you hold a lock.

Should I prefer an approach where flushing occurs at periodic intervals?

That really depends a lot on the features and characteristics of your system.
Do you care if you batch up requests? If the answer is no then I would definately look into and try the period logging approach.

If yes, then should I use a loop where I flush and sleep until a stop condition? Or should sleep() be avoided?

Nothing wrong with sleep. It actually releases the processor for other threads to use (ie its not a busy wait). But most conditional variables have a timed wait that allows you to break out after a timed period.

            clock_t  start = clock();
            while (mLogItems.empty() || (clock() - start) < WAIT_PERIOD)
            {
                clock_t waitTime = WAIT_PERIOD - (clock() - start);
                if (waitTime < 0)
                {    waitTime = /* appropriate value for you conditional for
                                   wait until next event. This will only
                                   happen if you have eaited past WAIT_PERIOD
                                   and nothing has been added to mLogItems
                                */
                }
                mCondition.timedwait(lock, waitTime);
            }

Is it feasible to use a lockless approach?

Yes. Use a lock-less queue implementation.

Comments on Code

Inline:

void log(LogLevel inLevel, std::string inMessage)
{
    if (mQuit)
    {
        return;
    }

    { // I would not add this block
      // It looks slightly ugly

        std::unique_lock<std::mutex> lock(mMutex);
        mLogItems.push_back(std::make_pair(inLevel, std::move(inMessage)));
        mCondition.notify_one();

    } // Then you can delete this as well.
}

Standard Singleton pattern here:

static Logger & Instance()
{
    static Logger fInstance;
    return fInstance;
}

Of course this tightly couples your code with the logging system. To make singeltons work effectively they should always be used in conjunction with other patterns so that you don't tightly couple the code.

There are a couple of different ways to avoid the tight coupling. One of the simplest is to use a factory pattern. Then get the appropriate logger object from the factory. This allows you to use a different type of logger during testing etc.

Answer 3

You're already half-way to double-buffering here (you're swapping the whole deque), but you've missed a couple of tricks:

1. the log thread only waits if mLogItems.empty(), so the log function only needs to signal if mLogItems.empty() is true before pushing
2. you're discarding the working deque every time: your allocator could cache deallocated blocks and recycle them back to mLogItems' allocator instance for re-use

I suspect that reducing heap operations by re-cycling memory like this could dominate both the signalling and lock-free changes, but I'd rather benchmark and find out for sure.