使用boost :: thread启动/停止日志数据（第二次更新） [英] Using boost::thread to start/stop logging data (2nd update)

问题描述

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

I'm currently trying to log real-time data by using boost::thread and a check box. When I check the box, the logging thread starts. When I uncheck, the logging thread stops. The problem arises when I check/uncheck repeatedly and very fast (program crashes, some files aren't logged, etc.). How can I write a reliable thread-safe program where these problems don't occur when repeatedly and quickly checking/unchecking? I also don't want to use join() since this temporarily stops the data input coming from the main thread. In the secondary thread, I'm opening a log file, reading from a socket into a buffer, copying this into another buffer, and then writing this buffer to a log file. I'm thinking that maybe I should use mutex locks for reading/writing. If so, what specific locks should I use? Below is a code snippet:

//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()


}

推荐答案

我喜欢使用Boost Asio的异步内容

I like to use Boost Asio for async stuff

#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，并从其中异步读取数据。

The program connects to port 6767 of localhost and asynchronously reads data from it.

如果日志记录是活动的（ _logfile.is_open（）），所有接收到的数据写入 / tmp / output .log 。

If logging is active (_logfile.is_open()), all received data is written to /tmp/output.log.

现在

• /写在单独的线程上，但是所有的操作都是使用 _ioservice （参见例如 post $ c> close（））


• 用户可以使用Ctrl + C中止套接字读取循环


• 2秒，日志记录将被激活（请参阅 do_toggle_logging_cycle ）

• the reading/writing is on a separate thread, but all operations are serialized using _ioservice (see e.g. the post in close())
• the user can abort the the socket reading loop with Ctrl+C
• every 2 seconds, the logging will be (de)activated (see do_toggle_logging_cycle)

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

The main thread just sleeps for 15 seconds before canceling the program (similar to the user pressing Ctrl-C).

这篇关于使用boost :: thread启动/停止日志数据（第二次更新）的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！