Changelog:

Your Task

  1. Implement a thread pool, which runs tasks across a fixed number of threads. Each of these threads should take tasks from a queue and run them, waiting for new tasks whenever the queue is empty. Your thread pool will start tasks in the order they are submitted and also support waiting for a particular task submitted to thread pool to complete. Each task in the thread pool will be identified by a name.

    Your implementation must start threads once when the thread pool is created and only deallocate them when the Stop() method is called. You may not start new threads each time a task is submitted.

    Your thread pool should follow the interface defined in the header file pool.h in our skeleton code [last updated 2022-03-14]. This interface has two classes, one called Task representing tasks to run an done called ThreadPool representing the thread pool itself. You may add additional methods and member variables to these classes for your implementation.

    The Task class must have the following methods:

    • Task(), virtual ~Task() — constructor and virtual destructor

    • virtual void Run() = 0 — pure virtual (i.e. abstract) method overriden by subclasses. This will be run by threads in the thread pool when a task is run.

    The ThreadPool class must have the following methods:

    • ThreadPool(int num_threads) — constructor, which takes the number of threads to create to run queued tasks. (Most likely, this is where you will create threads to run tasks.)

    • void SubmitTask(const std::string &name, Task *task) — function to submit (enqueue) a task. The task can be identified by name in future calls to WaitForTask(). Either when the task completes or when the task is waited for, the ThreadPool must be deallocate it as with delete task.

      You may assume that for a particular ThreadPool object, SubmitTask will be called only with the name of a task that has not already been submitted and for which WaitForTask has not already been called.

      This method should return immediately after adding the task to a queue, allocating additional space for the queue if necessary. (If this allocation fails, we do not care what happens.) It should never wait for running tasks to finish and free up space to store additional pending tasks.

    • void WaitForTask(const std::string &name) — wait for a particular task, not returning until the task completes. Note that this method may be called any time after the task is submitted, including after it has already finished running.

      You may assume that WaitForTask is called exactly once per submitted task.

    • void Stop() — stop the thread pool, terminating all its threads normally. If any thread is in the middle of running a Task, this should wait for that thread to finish running the task rather than interrupting it. Before this returns, all the threads spawned by thread pool should have finished executing and it should be safe to deallocate the thread pool.

    Every method of the ThreadPool class except for the Stop() method may be called from any thread, including from one of the tasks in submitted to the thread pool. The Stop() method may be called from any thread except one that was created by the thread pool.

    Provided that tasks are waited for as they complete, your thread pool should not use more and more memory the more tasks that run. (For example, you should not store a list of the names of all finished tasks from which tasks are never removed until thread pool is stopped (even when those finished tasks were waited for before the thread pool stops).)

    Your implementation must support multiple ThreadPool objects being used at the same time, so you should not use global variables.

    In no case, may any of the methods above (or below, if you took CoA 2) or the threads spawned by the constructor to run queued tasks consume a lot of compute time while waiting for some condition (e.g. a task finishing or a new task being available) to become true. A thread that needs to wait should arrange to be put into a sleeping/waiting state until the condition is likely true. (That is, use something like a condition variable or semaphore to wait until another thread indicates that something has happened, rather than simply checking a condition in a loop.)

  2. If you have previously taken CoA 2, then as an additional requirement, you must implement the following methods:

    • bool CancelTask(const std::string &name) — stop a task with a particular name from running if it has not already started. This should return true if successful, and false if the task in question has already started running or completed.

      You may assume that this is called instead of calling WaitForTask for a task of the same name.

    • void Pause() — after any currently running tasks complete, temporarily stop the thread pool worker threads from running any tasks. This must not return until after all the worker threads are not running tasks.

    • void Resume() — assuming a previous call to Pause was made, cause the thread pool threads to resume processing tasks.

  3. Your submission should include a Makefile which produces a statically linked library libpool.a, like our skeleton code does.

  4. You can use the supplied pool-test-tsan and pool-test programs to help test your thread pool implementation. The pool-test-tsan program is built with compiler options that attempt to detect data races (using ThreadSanitizer); pool-test with compiler options that attempt to detect memory errors (using AddressSanitizer.

    You can run:

    • ./pool-test-tsan shorter or ./pool-test shorter to run two tests, each of which uses a ThreadPool with one thread, and submits two tasks to it (requires this version of pool-test.cc, which was included with the skeleton as of 14 March)
    • ./pool-test-tsan short or ./pool-test short to run a subset of these tests that use ThreadPools with at most two threads and avoids submitting very large numbers of tasks.
    • ./pool-test-tsan long or ./pool-test long to run a more thorough set of tests, including tests that start many tasks and use more than two threads

    You can also edit pool-test.cc (especially the main() functoin near the bottom) and recompile to change the set of tests that’s enabled to aid debugging.

    Note pool-test is not a complete test, primary because some things are complex to test automatically in this assignment. For example:

    • the test may not expose all the race conditions that might exist in your code (especially since the exact timing with which your code is run will vary between machines),
    • it does not determine if your code consumes a lot of compute time while waiting for a condition to become true,
    • it does not determine whether you comply with the requirement to not spawn new threads for each task,
    • it does not determine whether you comply with the requirement to not use more and more memory the more tasks that are run, and
    • it does not determine whether your implementation allows two ThreadPools to be used at the same time

    If you have previously taken CoA 2, then we have supplied a pool-test-coa2-extra.cc which contains some tests for the extra functions you must implement. To use these tests, add it in addition to the supplied pool-test.cc to the Makefile, following the pattern used for pool-test.cc to add additional rules to the Makefile. Like with supplied pool-test, these tests will not completely test the methods you need to implement.

  5. Produce a .tar.gz file of your submission like the one make submit will produce and submit to the submission site.

Hints

General advice

  1. The producer/consumer pattern we discussed in lecture is very useful for this assignment.

  2. You can use the C++ standard library’s deque or list as a queue, using push_back() to insert elements onto the queue; and front() and pop_front() to remove elements from the queue.

    Alternately, you could write your own queue with a linked list, or dynamic array.

  3. You can use something like the C++ standard library’s map (typically implemented with a balanced tree) to store a mapping between std::strings and information about particular tasks.

  4. To safely access almost any of the containers (e.g. map, deque, vector) in the C++ standard library, you must ensure you prevent a thread from adding or removing elements from the container while any other thread is reading the container.

  5. You will need to use some synchronization mechanism to manage the queue of waiting tasks and manage reporting when tasks finish. Probably this will either be mutexes and condition variables (what I used in my implementation) or semaphores.

  6. Most likely you will want (at least) one condition variable or semaphore for each task to handle waiting for tasks to complete.

Example of Usage

  1. To use the ThreadPool class you create, a user would create a subclass of Task that implements the Run() method that performs an operation they want to add to the queue of operations to do:

    class ComputeSumTask : public Task {
public:
    ComputeSumTask(int *sum_destination, int *array_to_sum, int array_size) {
        this->sum_destination = sum_destination;
        this->array_to_sum = array_to_sum;
        this->array_size = array_size;
    }

    void Run() {
        int sum = 0;
        for (int i = 0; i < this->array_size; ++i) {
            sum += this->array_to_sum[i];
        }
        *this->sum_destination = sum;
    }
        
    int *sum_destination,
    int *array_to_sum;
    int array_size; 
};

    Notice that the Task subclass can (and typically would) contain member variables. Then, submit a bunch of instances of this class for each thing they wanted to do in parallel

    int arrayA[ARRAY_SIZE], arrayB[ARRAY_SIZE];
int sum_of_A, sum_of_B;
ThreadPool pool(num_threads);
    
pool.SubmitTask("sum arrayA", new ComputeSumTask(&sum_of_A, arrayA, ARRAY_SIZE));
pool.SubmitTask("sum arrayB", new ComputeSumTask(&sum_of_B, arrayB, ARRAY_SIZE));
...

    and finally wait for the tasks to complete before stopping the thread pool:

    pool.WaitForTask("sum arrayA");
pool.WaitForTask("sum arrayB");

pool.Stop();

Some Pthreads Resources

  1. The lecture slides on pthreads, pthread_mutexes, pthread_conds, etc.

  2. Chapters 27-31 of Operating Systems: Three Easy Pieces. Notably chapter 30 has code examples for condition varibles and chapter 31 has code examples for semaphores.

  3. The official documentation at http://pubs.opengroup.org/onlinepubs/9699919799/.

  4. This tutorial from LLNL.

Using std::map

std::map is one of the C++ standard libraries key-value containers:

On using the department machines

  1. If you are using the department machines, you may need to run module load gcc before compiling to get access to a recent version of gcc/g++

Using GDB to help diagnose hangs

  1. If you experience a hang when running tests, I would suggest starting by trying to reproduce the hang using a debugger like GDB. On the department machines, you may need to use module load gdb to get access to the most recent version of GDB. Once running under GDB, you can run a test until it hangs, then use control-C to stop the program and go into the debugger. Once in GDB, a command like

    thread apply all backtrace

    will show what each thread was doing. This should, at least, give you more information about what might be causing the hang.

    If you need more information about what the threads are doing, you can sue the debugger to examine the local variables of each thread. In the thread apply all backtrace output, each thread will have a number. You can use this to switch between threads using a command like

    thread 14

    and then use up and down to change where on that thread’s call stack the debugger is working. Once you have the debugger selecting a particular thread and a particular function call, you can examine the local variables, etc. using print or info locals or similar.

    See also the section of the GDB manual about threads