1
\$\begingroup\$

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();
}
\$\endgroup\$
  • \$\begingroup\$ Does mCondition.notify_one(); flush, or merely send a message to another thread? \$\endgroup\$ – Mooing Duck Sep 5 '12 at 17:16
2
\$\begingroup\$

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);
\$\endgroup\$
2
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ wait needs to re-acquire the lock when it wakes, so releasing the lock before notifying avoids unnecessary contention. It's legal in pthreads at least, although I'm not sure what std::condition_variable says about it. \$\endgroup\$ – Useless Sep 6 '12 at 11:53
  • \$\begingroup\$ @Useless: True. But I think that is a false assumption that it will cause unnecessary contention. Once the conditional thread is signaled it is free to run that does not mean it can (it still needs a processor to run on and this one is busy running a thread). It also is free to contend in acquire the lock with all the other threads (ie there is no special privileged given to the thread held in conditional variable). But the only way to validate this is to test and time. \$\endgroup\$ – Martin York Sep 6 '12 at 15:34
  • \$\begingroup\$ Yeah, I should have said that signal->unlock creates an avoidable opportunity for contention, or something; it's naturally neither guaranteed to happen nor the only point of contention. \$\endgroup\$ – Useless Sep 7 '12 at 13:15
1
\$\begingroup\$

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.

\$\endgroup\$
  • \$\begingroup\$ Good point about the redundant allocation. It occurred to me while walking home from work today :p \$\endgroup\$ – StackedCrooked Sep 5 '12 at 19:09

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.