c++ - 线程安全运算符<<

Question

我的记录器类有一个非常复杂的问题。它是一个单例模式日志记录类。创建线程仅用于从队列中取出项目并记录它们。通常一切正常，错误偶尔会作为段错误发生。在我决定在整个方法链上放置互斥锁之前，这种情况发生得更频繁。有了那个互斥锁，我不明白为什么会发生段错误。由于 operator<<，类(class)变得相当复杂用法。问题是运算符模板运行了很多次，因为使用 << 传递了很多项。 .因此，其他线程可以在这些模板调用之间插入。

一般的用法是这样的:
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 收集从单例类中移出到代理类中。

主.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;

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

此外，您可以在线程中访问 m_data 而无需任何锁定:

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

或

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

日志信息 (mp_logLine)

您尝试锁定太多数据。指向日志消息的指针一次只能由一个线程使用。但是你将它存储在所有线程访问的主记录器类中。您已经有一个记录器的代理，它对使用它的线程是私有(private)的。将您的消息存储在那里直到完成，然后将其添加到队列中。

总的来说，尽量减少要锁定的数据量。如果您重新编写代码并且唯一需要锁定的对象是您的队列，那么您的方法是正确的。