Async File Input

Question

I recently started work on my own Async file input "library" in C++. I got it done today and I decided to put it on Code Review to see how well I actually did, and where I can still improve the code.

The log++ #include is just some logging header I created for writing colored output to the console.

EDIT: Since the time I have posted this I made some huge changes to the code, so I guess I will post an updated version of it below:

AsyncTask.h

#pragma once

#include <iostream>
#include <fstream>
#include <thread>
#include <string>
#include <sstream>
#include <mutex>
#include <functional>
#include <condition_variable>
#include "logpp/log++.h"



class AsyncTask
{
public:


    AsyncTask();

    template<typename Callable, typename ...Args>
    explicit AsyncTask(Callable task, Args&&... args);
    ~AsyncTask();

    /*Manually starts the task if it is not already running*/
    template<typename Callable, typename ...Args>
    void start(Callable func, Args&&... args);

    /*Interrupts the task until resume() is called*/
    void pause();

    /*Unpauses the task*/
    void resume();

    /*Blocks calling thread until the task is done*/
    void wait();

    bool is_complete();

    /*Stops the task and resets all it's data*/
    virtual void reset();

private:

    std::thread m_exec_thread;

    bool m_running;
    bool m_paused;

    std::condition_variable m_pause_cv;
    std::mutex m_pause_mutex;

protected:
    void handle_pause_update();
    void toggle_running();
};


template<typename Callable, typename... Args>
AsyncTask::AsyncTask(Callable task, Args&&... args) : m_running(false), m_paused(false)
{
    start(task, std::forward<Args>(args)...);
}


template<typename Callable, typename ...Args>
void AsyncTask::start(Callable func, Args&&... args)
{
    /*Only start the task if it is not already running*/
    if (!m_running)
    {
        m_exec_thread = std::thread(func, std::forward<Args>(args)...);
        m_running = true;
    }
}


class AsyncReadTask : public AsyncTask
{
public:
    AsyncReadTask();
    explicit AsyncReadTask(std::string const& filename);

    /*Obtain data that was read from the file. Will reset the task*/
    std::stringstream get_data();

    void start(std::string const& filename);

    virtual void reset() override;

private:
    std::stringstream m_data;
    std::string m_file;

    void read(std::string const& file_name);
};

AsyncTask.cpp

#include "AsyncTask.h"


#pragma region AsyncTaskImpl

AsyncTask::AsyncTask()
{

}

void AsyncTask::handle_pause_update()
{
    while (m_paused) //check if we have to pause
    {
        std::unique_lock<std::mutex> lk { m_pause_mutex };
        m_pause_cv.wait(lk);
        lk.unlock();
    }
}


void AsyncTask::pause()
{
    if (m_paused) //don't double pause
        return;
    std::lock_guard<std::mutex> lk { m_pause_mutex };
    m_paused = true;
}


void AsyncTask::resume()
{
    std::lock_guard<std::mutex> lk { m_pause_mutex };
    m_paused = false;
    m_pause_cv.notify_one();
}


void AsyncTask::wait()
{
    while (m_running)
    {
        //wait
    }
}


bool AsyncTask::is_complete()
{
    return !m_running;
}


void AsyncTask::reset()
{
    wait();
    m_running = false;

}

void AsyncTask::toggle_running()
{
    m_running = !m_running;
}

AsyncTask::~AsyncTask()
{
    m_exec_thread.join();
}

#pragma endregion AsyncTaskImpl

#pragma region AsyncReadTaskImpl


std::stringstream AsyncReadTask::get_data()
{
    wait();

    auto data = std::move(m_data);
    reset();
    return std::move(data);
}


void AsyncReadTask::read(std::string const& file_name)
{
    /*---Open File---*/

    std::ifstream in(file_name);


    logpp::Console::log_assert(in.good() && in.is_open(), "Could not open file: " + file_name);


    /*---Read data---*/

    std::string line;
    while (std::getline(in, line))
    {
        line += "\n"; //getline removes the newline character, but we want to give the stringstream the exact same data as the file

        m_data << line;

        handle_pause_update();
    }


    /*---Cleanup---*/

    in.close();
    toggle_running();
}

AsyncReadTask::AsyncReadTask()
{

}

AsyncReadTask::AsyncReadTask(std::string const& filename) : AsyncTask(&AsyncReadTask::read, this, filename), m_file(filename)
{

}

void AsyncReadTask::start(std::string const& file_name)
{
    AsyncTask::start(&AsyncReadTask::read, this, file_name);
}

void AsyncReadTask::reset() 
{
    AsyncTask::reset();

    m_data.clear();
    m_file = "";
}

#pragma endregion AsyncReadTaskImpl

The full source can be found on GitHub too.

Answer 1

Just a few things that haven't been said yet:

• You don't have to #include everything at one place. For example, logging library does not need to be included in basic header. Consider including header files only where they are needed - this will speed up compilation a lot when your project grows bigger.

• You didn't provide enough details about logging library, however if you only log error when opening file and then continue with your task then it's wrong. If you do assert() then it's bad practice too because in release builds the error doesn't stop task. Error checking should be definitely implemented.

• Your string stream might not contain same data as your file - different platforms use different line terminators. If you need your stringstream to contain excactly same data as your file, consider loading file as binary.

• when specifying override you don't have to declare as virtual.

• you may use = default to specify default constructor.

• your AsyncTask() constructor does not initialize all members, this should be definitely fixed.

• in handle_pause_update there is a loop - I don't think it is necessary, simple if should do the job too.

• some of your functions can be const (typically getters, e.g. is_complete)

• as pointed out earlier, never code busy waits unless absolutely necessary. Why don't you just simply call m_exec_thread.join() in wait()?

• if you never start your task, then AsyncTask destructor calls join on the thread that has never been created and that throws an exception.

Answer 2

Since AsyncTask has a virtual function, and is used as a base class, the destructor ~AsyncTask should be virtual.

m_running and m_paused should be std::atomic<bool>. Without this, the compiler may assume that these values won't change if it doesn't see the change. This can cause the loop in AsyncTask::wait to never terminate, as the optimizer could remove the repeated checks of an apparently unchanging variable and leave an infinite loop. In addition, using std::atomic will allow you to get rid of some of the uses of a lock before accessing m_running or m_paused.

Also, AsyncTask::wait is not "friendly", as it is a very active busy loop. This should have something like a sleep(1) to avoid constantly burning up a CPU core, resulting in excessive heat and battery usage, and possibly system sluggishness. At the very least use the _mm_pause() intrinsic to let the CPU know it is in a spin-wait loop.

The m_running = false; expression in AsyncTask::reset effectively does nothing, since the call to wait will not return until that variable is false (and may cause harm, if another thread is set m_running back to true after wait has committed to returning).

In AsyncReadTask::get_data, you shouldn't use return std::move(data), since this suppresses RVO. Just use return data;.