Multi-threaded tree search

Question

This is my first nontrivial multi-threaded program. It recursively searches through a tree (haystack) to find the leaf where needle = true. It implements a thread pool and allows threads to exit early when another thread finds the answer.

It's worth noting that when using more than the main thread the program runs ~30 times slower, though I don't think this is due to something wrong in my code, just that threads are not really suited to this problem

I am especially interested in how well my code handles the threading aspect of the problem, thanks for having a look

#include <vector>
#include <random>
#include <mutex>
#include <atomic>
#include <condition_variable>
#include <list>
#include <thread>


class haystack_search;

class haystack {
private:
    friend haystack_search;
    static std::mt19937 rng;
    haystack * left;
    haystack * right;
    int depth;
    bool needle;
public:
    haystack(int layers, bool has_needle) {
        if(layers > 0) {
            depth = layers;
            //if this haystack has the needle, a rondom side gets the needle
            bool which = rng() & 0x01;
            left = new haystack(layers - 1, which ? has_needle : 0);
            right = new haystack(layers - 1, which ? 0 : has_needle);
            needle = 0;
        } else {
            depth = 0;
            left = nullptr;
            right = nullptr;
            needle = has_needle;
        }
    }
    haystack(const haystack&) = delete;
    haystack(haystack&& steal) {
        right = steal.right;
        left = steal.left;
        rng = std::move(steal.rng);
        needle = steal.needle;
        depth = steal.depth;

        steal.right = nullptr;
        steal.left = nullptr;
        steal.depth = 0;
    }
    ~haystack() {
        if(left != nullptr) {
            delete left;
        }
        if (right != nullptr) {
            delete right;
        }
    }
    bool get(int index) {
        if(depth > 0) {
            //go right if the correct order bit is set (highest bit at top level, lowest bit at low level)
            return (index >> (depth - 1)) & 0x01 ? right->get(index) : left->get(index);
        } else {
            return needle;
        }
    }
};

std::mt19937 haystack::rng;

class haystack_search {
private:
    struct request_info {
        haystack * hay;
        int index;        
    };
    std::atomic<bool> needle_found;
    std::atomic<int> needle_index;
    std::atomic<bool> kill_switch;

    std::vector<std::thread> pool;

    std::atomic<int> available_threads;
    std::atomic<int> max_threads;

    std::mutex recieve_mutex;
    std::mutex requests_mutex;
    std::condition_variable receive;
    std::list<request_info> requests;

    void thread_body() {
        std::unique_lock<std::mutex> lock(recieve_mutex);
        while (1) {
            receive.wait(lock, [this](){ 
                std::unique_lock<std::mutex> req_lock(requests_mutex);
                return !requests.empty(); 
            });
            if(kill_switch)
                break;
            std::unique_lock<std::mutex> req_lock(requests_mutex);
                haystack * hay = requests.front().hay;
                int index = requests.front().index;
                requests.pop_front();
            req_lock.unlock();
            lock.unlock();

            search_branch(hay, index);

            lock.lock();
            ++available_threads;
        }
    }

    int call_thread(haystack * hay, int index) {
        std::unique_lock<std::mutex> lock(recieve_mutex);
        if(available_threads > 0) {
            --available_threads;
            std::unique_lock<std::mutex> req_lock(requests_mutex);
                requests.push_back({hay,index});
            req_lock.unlock();
            receive.notify_one();
            return 1;
        } else {
            return 0;
        }        
    }

    void search_branch(haystack * hay, int index) {
        if(needle_found) {
            return;
        }

        if(hay->depth == 0) {
            if(hay->needle) {
                needle_found = true;
                needle_index = index;
            }
            return;
        }

        if(!call_thread(hay->right, (index << 1) + 1)) {
            //if right, low bit should be set
            search_branch(hay->right, (index << 1) + 1);
        }
        //if left, low bit should be cleared
        search_branch(hay->left, (index << 1));
    }
public:
    haystack_search(const haystack_search&) = delete;
    haystack_search(int max_thread_count) {
        max_threads = max_thread_count;
        available_threads = max_thread_count;
        kill_switch = false;
        for(int i = 0; i < max_thread_count; ++i) {
            pool.push_back(std::thread(thread_body,this));
        }
    }
    ~haystack_search() {
        std::unique_lock lock(recieve_mutex);
        kill_switch = true;
        std::unique_lock<std::mutex> req_lock(requests_mutex);
            requests.push_back({0,0});//dummy
        req_lock.unlock();
        lock.unlock();
        receive.notify_all();
        for(std::thread& t : pool) {
            t.join();
        }
    }
    int search(haystack * hay) {
        needle_found = false;
        search_branch(hay, 0);
        while(available_threads != max_threads) {
            //wait until all threads are completed
            std::this_thread::yield();
        }
        if(needle_found) {
            return needle_index;
        } else {
            return -1;
        }
    }
};

This is a sample main that uses the code

#include <iostream>
#include <chrono>
#include <cassert>

int main(int argc, char ** argv) {
    int thread_count = 0;
    if(argc == 2) {
        try {
            thread_count = std::stoi(argv[1]);
        } catch (const std::exception& e) {
            std::cout << "could not parse number of threads";
            return 1;
        }
    } else {
        std::cout << "invalid number of arguments (input number of threads)";
        return 1;
    }

    using clock = std::chrono::high_resolution_clock;
    clock::time_point start;
    clock::time_point end;

    std::vector<haystack*> hays;
    for(size_t i = 0; i < 100; ++i) {
        haystack* to_add = new haystack(14, 1);
        hays.push_back(to_add);
    }

    std::cout << "starting " << thread_count + 1 << "-threaded search...\n";
    haystack_search hsN(thread_count);
    start = clock::now();
    for(size_t i = 0; i < hays.size(); ++i) {
        int index = hsN.search(hays[i]);
        assert(hays[i]->get(index));
    }
    end = clock::now();
    std::cout << "finished with time " << (float)(end - start).count()*clock::duration::period::num/clock::duration::period::den << "\n";    
    return 0;
}

Answer 1

Overall Organization

It seems to me, that haystack_search is really at least three fundamentally different kinds of things, all rolled into one.

One is the logic for doing the search. That probably should be in this class. The second is a thread pool manager. That should be separate. The third is at least one (and maybe a couple) of thread-safe collections.

Thread safe collection

I would start by separating "requests" into a separate thread-safe queue (or stack--I didn't check how you use it). A really simple version might look something like this:

namespace sync {
    template <class T>
    class stack {
       std::list<T> items;
       std::mutex mtx;
    public:
        push(T const &t) { 
            std::lock_guard<T> L(mtx);
            items.push_back(t); 
        }

        bool pop(T &t) { 
            std::lock_guard<T> L(mtx);
            if (items.empty())
                return false; 
            t = items.pop(); 
            items.pop_back(); 
            return true;
        }
    };
}

Then the searching code doesn't need to mess with locking to access the queue/stack.

Side note: chances are that with some testing, you'll find that std::list is a poor choice here. For a queue you usually want std::deque. For a stack, either std::deque or std::vector usually works quite nicely.

Thread Pool

The thread pool class should normally be (mostly) agnostic to the kind of task the threads are being asked to execute. In fact, it should normally be possible (and easy) to push several different kind of tasks into the thread pool simultaneously. A thread then basically just grabs a task from its input, executes the task, and repeats (along with a few bits and pieces like exiting at the right times and such). The thread pool manager is responsible for things like starting up threads, shutting down threads, and so on.

Unlocking

Anytime I see a lock_guard/unique_lock and an explicit unlock, I start to get worried. Unlocking should normally be handled automatically via RAII. That's the whole point of using a lock_guard/unique_lock in the first place. Sometimes you add a scope just for this, so something like this:

        std::unique_lock<std::mutex> req_lock(requests_mutex);
            requests.push_back({hay,index});
        req_lock.unlock();

...becomes more like this instead:

{
    std::unique_lock<std::mutex> req_lock(requests_mutex);
    requests.push_back({hay, index});
}

This particular one is handled by the previous suggestion to create a thread safe queue/stack, but there are still some that can be cleaned up this way.

Killing Threads

In my experience, when you want to kill threads, it's often easier and cleaner to just push some "tasks" that basically tell each thread to kill itself. In most cases, individual tasks should be small enough that it's all right to let a thread just finish a task once it's started it. In most cases this improves speed quite a bit--once it starts executing a task, the thread doesn't have to do any locking, access atomic variables, etc. It just executes the task until it's done.