3
\$\begingroup\$

I'm writing an application to compare and sync folders on a disks. Scanning the contents of folders is performed in separate threads. I wrote a class to manage scan threads. The main application class inherits from this class.

The application is built and at first glance works without errors.

But I'm new to writing multithreaded code and I'm not sure I did everything right. Please look at my thread manager class. Is everything right there? Are there any problems? If so, how can they be corrected? Can it be made better?

Added on October 12th: I forgot to mention that my application is based on event loop. The main thread cannot wait the worker threads until they finish. Instead I use the callback function which puts a message to the event queue.

Header file:

#pragma once

#include <atomic>
#include <thread>
#include <mutex>


// This class runs two time-consuming file scanning jobs in separate threads.
// To implement specific scanning jobs tasks, inherit from this class.
class ThreadManager {
public:
    ThreadManager() = default;
    ThreadManager(const ThreadManager&) = delete;
    ThreadManager& operator=(const ThreadManager&) = delete;
    ~ThreadManager();

    // Start scan jobs.
    void startScan(const std::wstring& localDir, const std::wstring& sharedDir);

    // Stop scan jobs while they are running without waiting for results.
    void stopScan();

    // Scan worker functions.Implement them in the descendant class.
    virtual bool scanLocal(const std::wstring& localDir, const std::atomic<bool>& stopScan, std::string& errorMessage) = 0;
    virtual bool scanShared(const std::wstring& sharedDir, const std::atomic<bool>& stopScan, std::string& errorMessage) = 0;

    // A callback function that is called when both scanning tasks are completed or stopped.
    virtual void scanFinished(bool success, const std::string& errorMessage) = 0;

private:
    void freeThreadsResources();
    void scanLocalWrapper(const std::wstring& localDir);
    void scanSharedWrapper(const std::wstring& sharedDir);

private:
    std::atomic<bool> m_stopScan;
    std::mutex m_scanStateMutex;
    std::thread m_localScanThread;
    std::thread m_sharedScanThread;

    bool m_localRunning = false;
    bool m_sharedRunning = false;

    std::mutex m_localResultMutex;
    bool m_localSuccess = false;
    std::string m_localErrorMessage;
    std::mutex m_sharedResultMutex;
    bool m_sharedSuccess = false;
    std::string m_sharedErrorMessage;
};

CPP file:

#include "ThreadManager.h"
#include <sstream>

ThreadManager::~ThreadManager()
{
    freeThreadsResources();
}

void ThreadManager::startScan(const std::wstring& localDir, const std::wstring& sharedDir)
{
    freeThreadsResources();

    std::lock_guard lock(m_scanStateMutex);
    m_stopScan.store(false, std::memory_order_seq_cst);
    m_localRunning = true;
    m_sharedRunning = true;

    m_localScanThread = std::thread(&ThreadManager::scanLocalWrapper, this, localDir);
    m_sharedScanThread = std::thread(&ThreadManager::scanSharedWrapper, this, sharedDir);
}

void ThreadManager::stopScan()
{
    m_stopScan.store(true, std::memory_order_seq_cst);
}

void ThreadManager::freeThreadsResources()
{
    if (m_localScanThread.joinable())
        m_localScanThread.join();

    if (m_sharedScanThread.joinable())
        m_sharedScanThread.join();
}

static std::string composeErrorMessage(const std::string& localErrorMessage, const std::string& sharedErrorMessage)
{
    std::ostringstream result;
    result << localErrorMessage;
    if (!sharedErrorMessage.empty())
    {
        if (!localErrorMessage.empty())
            result << std::endl;
        result << sharedErrorMessage;
    }
    return result.str();
}

void ThreadManager::scanLocalWrapper(const std::wstring& localDir)
{
    std::lock_guard localResultLock(m_localResultMutex);
    m_localSuccess = scanLocal(localDir, m_stopScan, m_localErrorMessage);
    
    if (!m_localSuccess)
        stopScan();

    std::lock_guard scanLock(m_scanStateMutex);
    m_localRunning = false;
    if (!m_sharedRunning)
    {
        std::lock_guard sharedResultLock(m_sharedResultMutex);
        auto errorMessage = composeErrorMessage(m_localErrorMessage, m_sharedErrorMessage);
        scanFinished(m_localSuccess && m_sharedSuccess, errorMessage);
    }
}

void ThreadManager::scanSharedWrapper(const std::wstring& sharedDir)
{
    std::lock_guard sharedResultLock(m_sharedResultMutex);
    m_sharedSuccess = scanShared(sharedDir, m_stopScan, m_sharedErrorMessage);
    
    if (!m_sharedSuccess)
        stopScan();

    std::lock_guard scanLock(m_scanStateMutex);
    m_sharedRunning = false;
    if (!m_localRunning)
    {
        std::lock_guard localResultLock(m_localResultMutex);
        auto errorMessage = composeErrorMessage(m_localErrorMessage, m_sharedErrorMessage);
        scanFinished(m_localSuccess && m_sharedSuccess, errorMessage);
    }
}
\$\endgroup\$

2 Answers 2

2
\$\begingroup\$

I was in the middle of writing a review when one was accepted. I had only finished the design review, and was working on the actual code review. Rather than throw it all away, I’ll post the design review, and a few comments of what would have come up in the code review.

Design review

As I understand the design, the intention is for a user to derive this class, supply their own scanLocal(), scanShared(), and scanFinished() functions, and then use it like:

auto scanner = type_derived_from_thread_manager{};
scanner.startScan();

… correct?

Okay, before I start the review, I have to ask… C++11? Really? This is 2023! I would expect, at least, a bare minimum of C++17. That’s the version all major compilers default to. Concurrency support is barely functional in C++11. (It’s still not perfect as of C++20—executors is really holding everything up—but it’s much better than it was in C++11.)

Alright, so, I am not a fan of this interface. It’s extremely convoluted, and very hard to use… and very easy to misuse. Suppose I just want to do a scan and literally nothing else. Could I do this?

auto main() -> int
{
    auto scanner = type_derived_from_thread_manager{};
    scanner.startScan();
    std::cout << "scan done (successfully?)";
}

That’s already two lines for one operation, but more importantly… that won’t work. Because startScan() does indeed start the scan… but the interface provides no way to know when the scan finishes, except that the destructor completes. So if I wanted to print a completion message, I’d have to do:

auto main() -> int
{
    {
        auto scanner = type_derived_from_thread_manager{};
        scanner.startScan();
    }

    std::cout << "scan done (successfully?)";
}

Except what if I want to know whether the scan succeeded or not? There’s just no way to find out. Presumably that would have to be dealt with in the scanFinished() callback… but that is not where one would logically want to deal with success or failure; you deal with the success/failure of an operation in the code that called the operation… that’s why operations return error status to the caller, or unwind the stack with an exception, rather than everything returning void or being noexcept and handling errors internally.

At best you would have to do something like:

auto main() -> int
{
    auto result = bool{};

    {
        auto scanner = type_derived_from_thread_manager{&result};
        scanner.startScan();
    }

    if (result)
        std::cout << "scan succeeded";
    else
        std::cout << "scan failed";
}

… with the derived type properly setting that flag in scanFinished(). It’s a spaghetti mess of responsibilities.

Also, it’s wildly over-complicated, with no fewer than three mutexes, locks-within-locks, and, most dangerously, callbacks into unknown code while holding locks… which you should never, ever do. To put it bluntly, if it appears to be deadlock-free, that’s only because the most important parts of the code are missing. When those abstract member functions are defined, it’s quite likely there will be data race problems, and maybe even deadlocks.

I would suggest that the problems here stem from the mistake of thinking of this problem in terms of threads. It is better to think in terms of tasks.

What you have here are two tasks—the “scan local” and “scan shared” tasks—that are presumably independent, and that you would like to run concurrently, and wait for both to finish.

In other words, something more like:

// If you want to return actual result data, then replace void in the
// promise with the result data type.
auto do_scan(std::wstring dir, std::promise<void> promise)
{
    try
    {
        // do whatever you need to do here for the scan
        //
        // throw any exception on error

        promise.set_value_at_thread_exit();
    }
    catch (...)
    {
        promise.set_exception_at_thread_exit(std::current_exception());
    }
}

auto scan(std::wstring dir) -> std::future<void>
{
    auto promise = std::promise<void>{};
    auto future = promise.get_future();

    auto thread = std::thread{do_scan, std::move(dir), std::move(promise)};
    thread.detach();    // this is fine, because we are using promise with
                        // set_value_at_thread_exit()

    return future;
}

auto main() -> int
{
    using namespace std::literals;

    auto const local_dir = L"local"s
    auto const shared_dir = L"shared"s

    // Start all the scans, and collect the result futures.
    auto results = std::array<std::future<void>, 2>{
        scan(local_dir),
        scan(shared_dir)
    };

    // ... do any other work you want here...

    // Wait for all the results.
    std::for_each(results.begin(), results.end(),
        [](std::future<void>& result) { result.wait(); }
    );

    // Report results.
    std::cout << "all scans finished:\n";

    auto report = [](char const* which, std::future<void>& result)
    {
        try
        {
            result.get();
            std::cout << "  " << which << " scan succeeded\n";
        }
        catch (...)
        {
            std::cout << "  " << which << " scan failed\n";
        }
    };

    report("local",  result[0]);
    report("shared", result[1]);
}

The above assumes the local and shared scanning algorithms are identical… only the directory is different… but if they are not, it’s a small change. If you want to add cancellation, that’s also not hard.

Unless you are doing a large amount of scans, this should suffice. If you are doing a large amount of scans, then you will need a thread pool and a task queue. That’s a bit more complicated, but not by much. The key API functionality would probably look something like:

// This could create a thread pool and task queue, and pop tasks off the
// queue to run in the threads until done.
auto tasks = task_manager{};

// These return futures with the results of the tasks.
auto result_local = tasks.enqueue(scan_local, local_dir);
auto result_shared = tasks.enqueue(scan_shared, shared_dir);

// Now .wait() or .get() those results as you please.

And the point remains: you should think in terms of tasks… not threads. You’ll notice in the code above that there is not a single mutex or lock in sight, and zero chance of deadlock (unless the scan functions are not actually independent). The interface is almost impossible to misuse.

The standard library futures are still very simplistic—we are waiting on executors before improving them—but there are other libraries out there with better futures. Still, even the standard futures, limited as they are, are far easier to compose than threads.

Even if you want to keep your interface, startScan() should return a future so calling code can know when it’s done, and can query whether it succeeds or not:

auto main() -> int
{
    auto scanner = type_derived_from_thread_manager{};

    try
    {
        auto result = scanner.startScan().get();
        std::cout << "scan succeeded";
    }
    catch (std::exception const& x)
    {
        std::cout << "scan failed: " << x.what();
    }
}

The moral of the story is: don’t think in terms of threads, think in terms of tasks. You get concurrency almost transparently by returning futures from task functions. And even though standard futures are clunky, there are still far more composable than threads.

Code review

The code review wasn’t finished, but here are some things I was going to mention:

  1. There is quite a bit of repetition and manual unrolling of what should probably be loops.
    • For example, having two separate data members for two threads will make modifying to support a third thread difficult; had you used a container to hold the threads, adding support for a third thread would be trivial.
  2. There are way too many locks, and they are held for way too long, including locks within locks.
    • For example, in startScan(), the m_scanStateMutex should be unlocked as soon as m_localRunning and m_localRunning are set.
    • In scanLocalWrapper(), you lock m_localResultMutex and hold it for the whole function, including a dive into a user-supplied callback (!!!)… and then you lock m_scanStateMutex and, conditionally, m_sharedResultMutex too!
    • In scanLocalWrapper(), you lock m_localResultMutex, then m_sharedResultMutex. In scanSharedWrapper(), you lock m_sharedResultMutex, then m_localResultMutex. Normally this would be deadlock 101… but you may be saved by the fact that you lock m_scanStateMutex between, and do the dance with the two “running” flags. Note that I said may; I was in the process of reasoning this out, but never finished.
  3. There is no consideration whatsoever of exceptions, and that creates a number of problems.
    • For example, what if the second thread construction fails in startScan()?
    • If any exceptions are thrown anywhere in the “wrapper” functions—including any exceptions that escape from the user-supplied callbacks—you will not only not get any notice of a scan failure (because scanFinished() probably won’t get called), you’ll probably get a straight-up crash (via terminate()).
\$\endgroup\$
5
  • \$\begingroup\$ Thank you for the "task" pep talk, very helpful! \$\endgroup\$
    – J_H
    Oct 11, 2023 at 22:47
  • \$\begingroup\$ Thank you for your review. It is very informative and helpful. But my application is not for command. It based on event loop. So it cannot wait for end of tasks in another threads. It should start tasks threads and continue to work as usual. When worker threads complete their job they should put a message into the event loop to signal that their job completed. I cannot find any example of usage std::future for event based applications so I used bare threads. At least I can understand how to send a message from a thread but I cannot understand how to send message from a future. \$\endgroup\$ Oct 12, 2023 at 5:32
  • \$\begingroup\$ Note that the OP can change the accepted answer at any time, it's not a permanent choice. So don't let that green checkmark bother you when writing an answer :) \$\endgroup\$
    – G. Sliepen
    Oct 12, 2023 at 5:49
  • \$\begingroup\$ @AndreyEpifantsev What you’d do to use a task function with an event loop is add a continuation to the future. Basically startScan().and_then(postMessageToEventQueue). This is a very common pattern with JavaScript futures. std::future does not have .and_then() built in (yet, as of C++23), so you’d have to write it as a free function—and_then(startScan(), postMessageToEventQueue)—which makes it less readable… but it could have the same functionality. \$\endgroup\$
    – indi
    Oct 12, 2023 at 20:07
  • \$\begingroup\$ @G.Sliepen Yeah, but if someone is satisfied with the reviews they have, then—especially in cases where I have to do a ton of research or hard thinking (which is always the case with concurrency)—it’s not worth it when they might have already moved on. 🤷🏼 \$\endgroup\$
    – indi
    Oct 12, 2023 at 20:08
2
\$\begingroup\$

Separate concerns

Your class ThreadManager is doing too much. The name implies it is managing threads, but it also has knowledge of disk scanning. Sure, the actual implementations of the scan functions are provided by the main application, but you have already hardcoded in ThreadManager that there are two threads that each get a string as a parameter.

Ideally, you write a generic thread manager class that has no knowledge of what the threads themselves will be doing. It should allow the caller to provide the functions to run in threads, and have member functions to start threads and wait for them to finish.

For example:

class ThreadManager {
    using Function = std::function<bool<const std::atomic<bool>&)>;
    std::vector<Function> functions;
    std::vector<thread> threads;
    …
    void wrapper(Function& function) {
        bool success = function(m_stopScan);
    
        if (!m_localSuccess)
            stop();

        …
    }
        
public:
    …
    void registerFunction(Function function) {
        functions.push_back(function);
    }

    void start() {
        for (auto& function: functions)
             threads.push_back(&ThreadManager::wrapper, this):
    }
    …
};

And then the main application can use it like:

ThreadManager threadManager;
threadManager.register([](const std::atomic<bool>& stopFlag) {
    return scan(stopFlag, "localdirectory/");
}
threadManager.register([](const std::atomic<bool>& stopFlag) {
    return scan(stopFlag, "shareddirectory/");
}
threadManager.start();

Note how much more flexible ThreadManager now is: you can pass it any number of functions, and if you want to use threads to do something other than scanning directories, you can do that as well.

There are other ways to tackle this problem. Instead of having one thread per task, you could also separate the list of tasks from the list of threads: this is usually named a thread pool. Have a look at thread pool implementations on this site.

The C++ standard library also comes with some other threading primitives that might make what you want to do easier. For example, instead of raw threads, you could use std::async(); it can launch a thread to do a task, but also store the results from that task (including exceptions) in a thread-safe way. C++20 introduced std::jthread which standardizes on a way to request a thread to stop.

Questionable locking

While it looks like there is no possibility of deadlocks in your code, you have this dance with three mutexes and some flags in each wrapper function, and this looks very fragile. It would be much better if the wrapper functions didn't need this, and only run the scan function itself and call stopScan() if there is a failure. Then have a member function that waits for all threads to stop, and return the result of the scan, instead of having a callback function that is called from another thread. The caller can then do something like:

ThreadManager threadManager;
threadManager.startScan(…);
// do something else if necessary
…
// now wait for the results
auto result = threadManager.waitForResults();
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.