使用 boost::thread 开始/停止记录数据（第二次更新）

Question

我目前正在尝试使用 boost::thread 和一个复选框来记录实时数据。当我选中该框时，日志记录线程开始。当我取消选中时，日志记录线程停止。当我反复且非常快速地检查/取消检查（程序崩溃、某些文件未记录等）时，就会出现问题。如何编写一个可靠的线程安全程序，在反复快速检查/取消检查时不会出现这些问题？我也不想使用 join() 因为这会暂时停止来自主线程的数据输入。在辅助线程中，我打开一个日志文件，从套接字读取到缓冲区，将其复制到另一个缓冲区，然后将此缓冲区写入日志文件。我在想也许我应该使用互斥锁进行读/写。如果是这样，我应该使用什么特定的锁？下面是一段代码sn-p：

//Main thread
 if(m_loggingCheckBox->isChecked()) {

...

if(m_ThreadLogData.InitializeReadThread(socketInfo))//opens the socket. 
//If socket is opened and can be read, start thread.
 m_ThreadLogData.StartReadThread();
 else
 std::cout << "Did not initialize thread\n";
 }
 else if(!m_loggingCheckBox->isChecked())
 {

m_ThreadLogData.StopReadThread();

}

void ThreadLogData::StartReadThread()
 {
 //std::cout << "Thread started." << std::endl;
 m_stopLogThread = false;
 m_threadSendData = boost::thread(&ThreadLogData::LogData,this);
 }

void ThreadLogData::StopReadThread()
 {
 m_stopLogThread = true;
 m_ReadDataSocket.close_socket(); // close the socket

if(ofstreamLogFile.is_open())
 {
 ofstreamLogFile.flush(); //flush the log file before closing it.
 ofstreamLogFile.close(); // close the log file
 }
 m_threadSendData.interrupt(); // interrupt the thread
 //m_threadSendData.join(); // join the thread. Commented out since this
 temporarily stops data input.

}

//secondary thread
 bool ThreadLogData::LogData()
 {

unsigned short int buffer[1024];
 bool bufferflag;
 unsigned int iSizeOfBuffer = 1024;
 int iSizeOfBufferRead = 0;
 int lTimeout = 5;

if(!ofstreamLogFile.is_open())
 {
 ofstreamLogFile.open(directory_string().c_str(), ios::out);

if(!ofstreamLogFile.is_open())
 {
 return 0;
 }
 }

while(!m_stopLogThread)
 {
 try {
 int ret = m_ReadDataSocket.read_sock(&m_msgBuffer.m_buffer
 [0],iSizeOfBuffer,lTimeout,&iSizeOfBufferRead);

memcpy(&buffer[0],m_msgBuffer.m_buffer,iSizeOfBufferRead);
 bufferflag = m_Buffer.setBuffer(buffer);
 if(!bufferflag) return false;
 object = &m_Buffer;

unsigned int data = object->getData();

ofstreamLogFile << data << std::endl;

boost::this_thread::interruption_point();

} catch (boost::thread_interrupted& interruption) {
 std::cout << "ThreadLogData::LogData(): Caught Interruption thread." << std::endl;
 StopReadThread();
 } catch (...) {
 std::cout << "ThreadLogData::LogData(): Caught Something." << std::endl;
 StopReadThread();
 }

} // end while()


}

Answer 1

我喜欢将 Boost Asio 用于异步处理

#include <iostream>
#include <fstream>

#include <boost/asio.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/asio/ip/tcp.hpp>
#include <boost/bind.hpp>
#include <boost/optional.hpp>

#include <thread>

using boost::asio::ip::tcp;
namespace asio = boost::asio;

struct program
{
    asio::io_service       _ioservice;
    asio::deadline_timer   _timer;
    asio::signal_set       _signals;
    std::array<char, 1024> _buffer;
    tcp::socket            _client;
    tcp::resolver          _resolver;
    std::ofstream          _logfile;
    std::thread            _thread;

    program() 
        : _timer(_ioservice),
        _signals(_ioservice),
        _client(_ioservice),
        _resolver(_ioservice)
    {
        do_connect(_resolver.resolve({ "localhost", "6767" }));
        do_toggle_logging_cycle();

        _signals.add(SIGINT);
        _signals.async_wait([this](boost::system::error_code ec, int) { if (!ec) close(); });

        _thread = std::thread(boost::bind(&asio::io_service::run, boost::ref(_ioservice)));
    }

    ~program()
    {
        if (_thread.joinable())
            _thread.join();
    }

    void close() {
        _ioservice.post([this]() { 
            _signals.cancel();
            _timer.cancel();
            _client.close(); 
        });
    }

private:

  void do_toggle_logging_cycle(boost::system::error_code ec = {})
  {
      if (ec != boost::asio::error::operation_aborted)
      {
          if (_logfile.is_open())
          {
              _logfile.close();
              _logfile.clear();
          } else
          {
              _logfile.open("/tmp/output.log");
          }

          _timer.expires_from_now(boost::posix_time::seconds(2));
          _timer.async_wait(boost::bind(&program::do_toggle_logging_cycle, this, boost::asio::placeholders::error()));
      } else 
      {
          std::cerr << "\nDone, goobye\n";
      }
  }

  void do_connect(tcp::resolver::iterator endpoint_iterator) {

      boost::asio::async_connect(
          _client, endpoint_iterator,
          [this](boost::system::error_code ec, tcp::resolver::iterator) {
                if (!ec) do_read();
                else     close();
          });
  }

  void do_read() {
    boost::asio::async_read(
        _client, asio::buffer(_buffer.data(), _buffer.size()),
        [this](boost::system::error_code ec, std::size_t length) {
            if (!ec) {
              if (_logfile.is_open())
              {
                    _logfile.write(_buffer.data(), length);
              }
              do_read();
            } else {
                close();
            }
    });
  }

};

int main()
{
    {
        program p; // does socket reading and (optional) logging on a separate thread

        std::cout << "\nMain thread going to sleep for 15 seconds...\n";
        std::this_thread::sleep_for(std::chrono::seconds(15));

        p.close(); // if the user doesn't press ^C, let's take the initiative
        std::cout << "\nDestruction of program...\n";
    }
    std::cout << "\nMain thread ends\n";
};

程序连接到localhost的6767端口，异步读取数据。

如果日志记录处于活动状态 (_logfile.is_open())，则所有接收到的数据都将写入 /tmp/output.log。

现在

• 读/写在单独的线程上，但所有操作都使用_ioservice 序列化（参见例如close() 中的post）
• 用户可以使用 Ctrl+C 中止套接字读取循环
• 每 2 秒，日志记录将被（停用）激活（请参阅 do_toggle_logging_cycle）

主线程在取消程序之前只休眠 15 秒（类似于用户按 Ctrl-C）。