线程安全operator<<

Question

我的伐木课有一个相当复杂的问题。它是一个单例模式日志类。创建线程仅用于从队列中取出项并记录它们。通常情况下，一切都很好，错误有时会以分段错误的形式出现。在我决定在整个方法链上加一个互斥之前，这种情况发生得更频繁了。有了这个互斥，我不明白为什么会出现分割错误。由于operator<<的使用，类变得相当复杂。它的问题是操作符模板被运行了很多次，因为使用<<传递了很多项。因此，其他线程可以在templete调用之间插入。

一般用法如下：1. instance method is called (creating or pointing to the instance pointer (singleton). mutex is locked at this moment. 2. any called methods are called, for example operator<< template. 3. finishing method is called, placing log in the queue and unlocking mutex.

我已经编辑了代码，并尝试将fata聚集从singleton类转移到代理类。

C.主要：

#include <iostream>
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>

#include "CLogger.h"
#include "CTestClass.h"

using namespace std;

int main()
{
    CLogger::instance().log.startLog("/home/lukasz/Pulpit/test.txt", true);
    CLogger::instance().log.setLogLevel(CLogger::ElogLevel::eDebug);

    CLogger::instance().printf("Print test");
    CLogger::instance().printf("Print test");
    CLogger::instance().printf("Print test");
    CLogger::instance() << "Stream test" ;
    CLogger::instance() << "Stream test" ;
    CLogger::instance() << "Stream test" ;


    //CTestClass test1(1);
    //CTestClass test2(2);
    //CTestClass test3(3);

    sleep(3);
    CLogger::instance().log.stopLog();

    return 0;
}

CLogger.h：

#ifndef CLOGGER_H_
#define CLOGGER_H_

#include <iostream>
#include <deque>
#include <string>
#include <mutex>
#include <condition_variable>
#include <pthread.h>
#include <ostream>
#include <fstream>
#include <sstream>
#include <ctime>
#include <iomanip>
#include <sys/time.h>
#include <stdarg.h>
#include <assert.h>

#include "CTeebuf.h"

using namespace std;

class CLoggerProxy;

/*!
 * \brief Singleton class used for logging
 */
class CLogger
{
public:
    /*!
     * \brief Describes the log level of called \ref CLogger object.
     */
    enum class ElogLevel { eNone = 0, eError, eWarning, eInfo, eDebug };

    /*!
     * Structure describing a single log item:
     */
    struct logline_t
    {
        string logString; /*!< String line to be saved to a file (and printed to cout). */
        ElogLevel logLevel; /*!< The \ref ElogLevel of this line. */
        timeval currentTime; /*!< time stamp of current log line */
    };

    static CLogger* internalInstance(ElogLevel lLevel = ElogLevel::eDebug);
    static CLoggerProxy instance(ElogLevel lLevel = ElogLevel::eDebug);

    bool startLog(string fileName, bool verbose);
    void setLogLevel(ElogLevel ll);
    void stopLog();
    void finaliseLine(logline_t* log);

protected:
    virtual void threadLoop();

private:
    CLogger() {};                               // Private so that it can  not be called
    CLogger(CLogger const&) {};                 // copy constructor is private
    CLogger& operator= (CLogger const&) {};     // assignment operator is private

    /*!< Global static pointer used to ensure a single instance of the class */
    static CLogger* mp_instance;
    bool m_logStarted;
    ElogLevel m_userDefinedLogLevel;
    ofstream m_logFileStream;
    bool m_verbose;
    bool m_finishLog;
    timeval m_initialTime;

    static void * threadHelper(void* handler)
    {
        ((CLogger*)handler)->threadLoop();
        return NULL;
    }

    deque<logline_t*> m_data;
    mutex m_mutex2;
    condition_variable m_cv;
    pthread_t   m_thread;

    logline_t pop_front();
    void push_back(logline_t* s);
};

/*!
 * RAII class used for its destructor, to add a log item to the queue
 */
class CLoggerProxy
{
public:
    CLogger &log;
    CLoggerProxy(CLogger &logger) : log(logger)
    {
        mp_logLine = new CLogger::logline_t;
        gettimeofday(&mp_logLine->currentTime, NULL);
    }

    ~CLoggerProxy() { log.finaliseLine(mp_logLine); }

    void printf(const char* text, ...);

    /*!
     * Takes the data from the stream and adds it to the current string.
     * @param t stream item
     * @return \ref object address
     */
    template <typename T>
    CLoggerProxy& operator<< (const T &t)
    {
        ostringstream stream;
        stream << t;
        mp_logLine->logString = (stream.str() + " ");
        return *this;
    }

private:
    CLogger::logline_t* mp_logLine;

};

#endif /* CLOGGER_H_ */

CLogger.cpp：

#include "CLogger.h"

using namespace std;

CLogger* CLogger::mp_instance = NULL;

/*!
 *  This function is called to create an instance of the class.
 *  Calling the constructor publicly is not allowed. The constructor
 *  is private and is only called by this Instance function.
 *  @param lLevel  Log level for current object
 */
CLogger* CLogger::internalInstance(ElogLevel lLevel)
{
    // Only allow one instance of class to be generated.
    if (!mp_instance)
    {
        mp_instance = new CLogger;
        assert(mp_instance);
    }

    return mp_instance;
}

/*!
 * This method is called in order to use the methods
 * within the objects.
 * @param lLevel  Log level for current object
 */
CLoggerProxy CLogger::instance(ElogLevel lLevel)
{
    return CLoggerProxy(*internalInstance(lLevel));
}

/*!
 * \brief Starts the logging system.
 *
 * This method creates and opens the log file,
 * then opens it and creates the threadloop for messages deque.
 * @param fileName desired log file path,
 * @param verbose when set true, logging will also be printed to standard output.
 */
bool CLogger::startLog(string fileName, bool verbose)
{
    if(remove(fileName.c_str()) != 0)
        perror( "Error deleting file" );

    m_logFileStream.open(fileName.c_str(), ios::out | ios::app);
    if (!m_logFileStream.is_open())
    {
        cout << "Could not open log file " << fileName << endl;
        return false;
    }

    m_finishLog = false;
    m_verbose = verbose;
    m_logStarted = true;
    gettimeofday(&m_initialTime, NULL);

    return (pthread_create(&(m_thread), NULL, threadHelper, this) == 0);
}

/*!
 * \brief puts a \ref logline_t object at the end of the queue
 * @param s object to be added to queue
 */
void CLogger::push_back(logline_t* s)
{
    unique_lock<mutex> ul(m_mutex2);
    m_data.emplace_back(move(s));
    m_cv.notify_all();
}

/*!
 * \brief takes a \ref logline_t object from the beggining of the queue
 * @return first \ref logline_t object
 */
CLogger::logline_t CLogger::pop_front()
{
    unique_lock<mutex> ul(m_mutex2);
    m_cv.wait(ul, [this]() { return !m_data.empty(); });

    logline_t retVal = move(*m_data.front());

    assert(m_data.front());
    delete m_data.front();
    m_data.front() = NULL;

    m_data.pop_front();

    return retVal;
}

/*!
 * \brief Sets the log level for the whole \ref CLogger object.
 * If \ref m_logLine is equal or higher than set level, log
 * is going to be printed.
 * @param lLevel desired user define log level.
 */
void CLogger::setLogLevel(ElogLevel lLevel)
{
    m_userDefinedLogLevel = lLevel;
}

/*!
 * \brief Stops the logging system.
 * Last final logline is being added and then the logging thread
 * is being closed.
 */
void CLogger::stopLog()
{
    m_finishLog = true;
    //instance(ElogLevel::eNone).log << "CLogger Stop";

    //pthread_join(m_thread, NULL);
}

/*!
 * This function should be run in the \ref CLoggerProxy destructor.
 * is pushes the gathered stream to the queue.
 */
void CLogger::finaliseLine(logline_t* log)
{
    if (log->logString.size() > 0)
        push_back(log);
    else
        delete log;
}

/*!
 * \brief Adds text log to the string in the printf c way.
 * Works faster than operator<< and its more atomic.
 * @param text pointer to a character string.
 * @param ... argptr parameters
 */
void CLoggerProxy::printf(const char* text, ...)
{
    va_list argptr;
    va_start(argptr, text);

    char* output = NULL;
    vasprintf(&output, text, argptr);

    mp_logLine->logString = output;
    va_end(argptr);
}

/*!
 * The loop running in a separate thread. It take items of the
 * log deque object (if there are any) and saves them to a file.
 */
void CLogger::threadLoop()
{
    logline_t logline;
    const string logLevelsStrings[] = {"eNone", "eError", "eWarning", "eInfo", "eDebug" };

    COteestream tee;
    tee.add(m_logFileStream);

    if (m_verbose)
        tee.add(cout);

    struct sched_param param;
    param.__sched_priority = 0;

    if(!sched_setscheduler(0, SCHED_IDLE, &param))
        instance().printf("Clogger scheduler policy set to %d", sched_getscheduler(0));

    int secs = 0;
    int h = 0;
    int m = 0;
    int s = 0;

    do
    {
        logline = pop_front(); // waits here for new lines

        secs = logline.currentTime.tv_sec - m_initialTime.tv_sec;
        h = secs / 3600;
        m = ( secs % 3600 ) / 60;
        s = ( secs % 3600 ) % 60;

        tee     << "["
                << setw(2) << setfill('0') << h
                << ":"
                << setw(2) << setfill('0') << m
                << ":"
                << setw(2) << setfill('0') << s
                << "."
                << setw(6) << setfill('0') << logline.currentTime.tv_usec
                << "]"
                << "["
                << setw(2) << setfill('0') << m_data.size()
                << "]"
                << "["
                << logLevelsStrings[(int)logline.logLevel]
                << "] "
                << logline.logString << "\n" << flush;
    }
    //while(!(m_finishLog && m_data.empty()));
    while(1);

    m_logFileStream.close();
}

Answer 1

您的代码有几个问题。

辛格尔顿

// Only allow one instance of class to be generated.
if (!mp_instance)
{
    mp_instance = new CLogger;
    assert(mp_instance);
}

这是个典型的问题。它可能同时被不同的线程调用，而且它不是线程安全的。最后可能会有几个单独的实例。

队列(m_data)

记录器的客户端将他们的消息放入这个队列中(显然是由m_mutext保护的)。

m_data.emplace_back(move(s));
m_cv.notify_all();

记录器线程在自己的线程中删除消息(由m_mutex2保护)。

unique_lock<mutex> ul(m_mutex2);
m_cv.wait(ul, [this]() { return !m_data.empty(); });

logline_t retVal = move(*m_data.front());

assert(m_data.front());
delete m_data.front();
m_data.front() = NULL;

m_data.pop_front();

return retVal;

这里的问题是，您使用两个不同的互斥来同步对同一个对象的访问。这行不通。

此外，您还可以访问线程中的m_data，而不需要任何锁定：

            << setw(2) << setfill('0') << m_data.size()

或

while(!(m_finishLog && m_data.empty()));

日志消息(mp_logLine)

你试图锁定太多的数据。指向日志消息的指针一次由单个线程使用。但是，您可以将其存储在主记录器类中，所有线程都可以访问这个类。您已经为您的记录器提供了一个代理，它对使用它的线程是私有的。将消息存储在那里，直到消息完成为止，然后将其添加到队列中。

通常情况下，尽量减少要锁定的数据量。如果您重新工作您的代码，并且唯一需要锁定的对象是您的队列，那么您就走上了正确的道路。

Answer 2

这看起来一点也不安全。

mp_logLine->logString += (stream.str() + " ");

这看起来是在登录到实例的所有线程之间共享的。从您的代码中不清楚+=是否是线程安全的。

另一点是，当您将一个项目推回您的队列时，您不会锁定互斥锁。如果两个线程同时做这件事，他们可能会搞砸这件事。不能保证您可以并行执行push_back和pop_front，所以在它周围设置一个互斥对象。

当您获得一个内部实例时，您会锁定mp_instance->m_mutex，但它看起来并不是要解锁它。

您可以从并行线程中使用bool m_logStarted，这也会引入争用条件和不一致。

以上任何一项都会导致分割错误。

正确的多线程是非常困难的。调试它更加困难。尝试将多线程组件卸载到您已经知道的库中，并在单个线程上下文中添加内容。在这种情况下，它意味着对日志的每个调用使用一个单独的类实例，然后将该项推送到由某个库实现并保证线程安全的生产者-使用者队列中。还有很多其他方法可以做到这一点。