Using boost::thread to start/stop logging data (2n

Question

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


}

Answer 1

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";
};

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

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

Now

• 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)

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