Multithreading in C++

By: Rajesh P.S.

Multithreading in C++ is a programming approach that enables the concurrent execution of multiple threads within a single process. Each thread represents an independent path of execution, allowing tasks to be performed in parallel, using the capabilities of modern multi-core processors.

Threads can be created, managed, and synchronized using C++ standard library facilities, facilitating the coordination and sharing of resources among threads. Multithreading is essential for improving program efficiency and responsiveness, as it enables tasks to be executed concurrently and is particularly valuable in scenarios where tasks can be divided into smaller, independent subtasks that can run in parallel.

Threads and Thread Management

Threads and Thread Management in C++ involve creating and managing threads, controlling their execution, and managing their resources. Here are details with examples:

Creating Threads

You can create threads using the <thread> header, which provides a std::thread class. To create a thread, you typically pass a function or callable object to the std::thread constructor. When the thread is created, it starts executing that function.

#include <iostream> #include <thread> void threadFunction() { std::cout << "This is a new thread." << std::endl; } int main() { std::thread myThread(threadFunction); myThread.join(); // Wait for the thread to finish return 0; }

Starting Threads

Continue Reading...

Threads are started automatically upon creation, and they begin executing the specified function. In the example above, myThread starts running threadFunction as soon as it's created.

Joining Threads

To ensure the main thread waits for a created thread to finish its execution, you use the join() member function. This blocks the main thread until the thread being joined has completed.

std::thread myThread(threadFunction); myThread.join(); // Wait for myThread to finish

Detaching Threads

Alternatively, you can detach a thread, which means that the main thread doesn't wait for it to complete. The detached thread runs independently, and its resources are automatically released when it finishes.

std::thread myThread(threadFunction); myThread.detach(); // Detach myThread // The main thread may continue without waiting for myThread

Thread Identification and Thread-Local Storage

Each thread has a unique identifier associated with it, known as a thread ID or thread handle. C++ does not provide direct access to these IDs, but you can use the std::this_thread::get_id() function to obtain the ID of the currently executing thread.

Thread-local storage allows you to have variables that are local to each thread, ensuring that each thread has its own independent copy of a variable. You can use the thread_local keyword to declare thread-local variables.

#include <iostream> #include <thread> void threadFunction() { std::cout << "Thread ID: " << std::this_thread::get_id() << std::endl; } int main() { std::thread myThread1(threadFunction); std::thread myThread2(threadFunction); myThread1.join(); myThread2.join(); return 0; } Example (Thread-Local Storage): #include <iostream> #include <thread> thread_local int threadSpecificValue = 0; void threadFunction() { threadSpecificValue++; std::cout << "Thread-Specific Value: " << threadSpecificValue << std::endl; } int main() { std::thread myThread1(threadFunction); std::thread myThread2(threadFunction); myThread1.join(); myThread2.join(); return 0; }

Synchronization

Synchronization in C++ multithreading is crucial for managing access to shared resources and coordinating the execution of threads. It involves preventing data races, ensuring orderly execution, and managing inter-thread communication.

Mutual Exclusion (Mutexes)

Mutual exclusion is a technique that ensures that only one thread can access a shared resource at a time. In C++, you can use the std::mutex class to create a mutex. Typically, you lock the mutex before accessing the shared resource and unlock it after you're done.

#include <iostream> #include <thread> #include <mutex> std::mutex myMutex; void criticalSection() { myMutex.lock(); // Acquire the mutex // Access and modify shared resource myMutex.unlock(); // Release the mutex } int main() { std::thread thread1(criticalSection); std::thread thread2(criticalSection); thread1.join(); thread2.join(); return 0; }

To simplify mutex management, C++ provides RAII-based synchronization tools like std::lock_guard and std::unique_lock. These classes automatically lock and unlock the mutex, ensuring proper synchronization.

#include <iostream> #include <thread> #include <mutex> std::mutex myMutex; void criticalSection() { std::lock_guard<std::mutex> lock(myMutex); // Automatically locks and unlocks // Access and modify shared resource } int main() { std::thread thread1(criticalSection); std::thread thread2(criticalSection); thread1.join(); thread2.join(); return 0; }

Deadlocks and Prevention

Deadlocks occur when two or more threads are unable to proceed because each is waiting for a resource held by another. To prevent deadlocks, follow some best practices:

Lock resources in a consistent order.
Use timeouts or try-lock mechanisms to avoid infinite waits.
Limit the duration of mutex locks and minimize the use of nested locks.

Consider using higher-level abstractions, like std::lock, to lock multiple mutexes simultaneously and avoid deadlock.

Condition Variables

Condition variables allow threads to communicate and coordinate. They are often used when a thread must wait for a certain condition to be met before proceeding. You can use std::condition_variable and std::condition_variable_any to achieve this.

#include <iostream> #include <thread> #include <mutex> #include <condition_variable> std::mutex mtx; std::condition_variable cv; bool condition = false; void waitForCondition() { std::unique_lock<std::mutex> lock(mtx); cv.wait(lock, []{ return condition; }); std::cout << "Condition met. Proceeding." << std::endl; } void setCondition() { std::this_thread::sleep_for(std::chrono::seconds(2)); { std::lock_guard<std::mutex> lock(mtx); condition = true; } cv.notify_all(); } int main() { std::thread waitingThread(waitForCondition); std::thread settingThread(setCondition); waitingThread.join(); settingThread.join(); return 0; }

In this example, one thread waits for a condition to be met while another thread sets the condition, and a condition variable is used to coordinate their actions.

Data Sharing and Race Conditions

Data sharing and race conditions are critical aspects of multithreaded programming in C++. Ensuring data integrity and avoiding race conditions is essential for writing reliable concurrent code.

Identifying and Avoiding Data Races

Data races occur when two or more threads concurrently access shared data, leading to unpredictable and erroneous behavior. To identify and avoid data races:

Use mutexes or other synchronization mechanisms to protect shared data from concurrent access.
Ensure that shared resources are accessed by only one thread at a time.

#include <iostream> #include <thread> #include <mutex> int sharedValue = 0; std::mutex myMutex; void threadFunction() { for (int i = 0; i < 100000; ++i) { std::lock_guard<std::mutex> lock(myMutex); sharedValue++; } } int main() { std::thread thread1(threadFunction); std::thread thread2(threadFunction); thread1.join(); thread2.join(); std::cout << "Shared value: " << sharedValue << std::endl; return 0; }

Thread-Safe Data Structures and Techniques

To achieve thread safety, it's crucial to use thread-safe data structures or apply techniques that make your data access safe. C++ provides various thread-safe data structures, such as std::vector, std::queue, and std::map, which can be used in a multithreaded context. Additionally, you can use locks or synchronization primitives to protect non-thread-safe data structures when accessed by multiple threads.

#include <iostream> #include <thread> #include <queue> #include <mutex> std::queue<int> myQueue; std::mutex myMutex; void pushToQueue(int value) { std::lock_guard<std::mutex> lock(myMutex); myQueue.push(value); } int popFromQueue() { std::lock_guard<std::mutex> lock(myMutex); if (!myQueue.empty()) { int value = myQueue.front(); myQueue.pop(); return value; } return -1; // Indicates an empty queue } int main() { std::thread producer(pushToQueue, 42); std::thread consumer([]() { int value = popFromQueue(); std::cout << "Popped value: " << value << std::endl; }); producer.join(); consumer.join(); return 0; }

Atomic Operations and std::atomic Type

C++ provides the std::atomic type and atomic operations to perform thread-safe, lock-free operations on variables. This allows you to ensure that certain operations are performed atomically without the need for explicit locks.

#include <iostream> #include <thread> #include <atomic> std::atomic<int> sharedValue(0); void incrementValue() { sharedValue.fetch_add(1, std::memory_order_relaxed); } int main() { std::thread thread1(incrementValue); std::thread thread2(incrementValue); thread1.join(); thread2.join(); std::cout << "Shared value: " << sharedValue.load() << std::endl; return 0; }

In this example, the std::atomic type and the fetch_add method ensure that the increment operation is atomic, preventing data races without the need for explicit locks. This is especially useful for improving performance in cases where fine-grained synchronization is required.

Thread Safety and Memory Models

Thread safety and memory models are essential aspects of multithreading in C++. They involve understanding how memory is accessed and shared between threads and ensuring that data is manipulated correctly in a concurrent environment.

Memory Ordering and the C++ Memory Model

The C++ memory model defines the rules governing how memory operations are sequenced and how threads interact with shared memory. It ensures that threads can communicate and synchronize in a predictable manner. Memory ordering refers to the rules that dictate when the changes made by one thread become visible to other threads.

#include <iostream> #include <thread> #include <atomic> std::atomic<int> x(0); std::atomic<int> y(0); int r1, r2; void thread1() { x.store(1, std::memory_order_relaxed); r1 = y.load(std::memory_order_relaxed); } void thread2() { y.store(1, std::memory_order_relaxed); r2 = x.load(std::memory_order_relaxed); } int main() { std::thread t1(thread1); std::thread t2(thread2); t1.join(); t2.join(); std::cout << "r1 = " << r1 << ", r2 = " << r2 << std::endl; return 0; }

In this example, std::memory_order_relaxed is used, meaning that the compiler and hardware can optimize the memory accesses as long as the observed behavior is consistent with the program's semantics. Consequently, the order of stores and loads may not match the program order.

Memory Barriers and Fences

Memory barriers and fences are synchronization primitives that control the visibility of memory operations. They ensure that memory operations are not reordered by the compiler or hardware in ways that violate the desired synchronization.

#include <iostream> #include <thread> #include <atomic> std::atomic<int> x(0); std::atomic<int> y(0); int r1, r2; void thread1() { x.store(1, std::memory_order_relaxed); std::atomic_thread_fence(std::memory_order_acquire); r1 = y.load(std::memory_order_relaxed); } void thread2() { y.store(1, std::memory_order_relaxed); std::atomic_thread_fence(std::memory_order_acquire); r2 = x.load(std::memory_order_relaxed); } int main() { std::thread t1(thread1); std::thread t2(thread2); t1.join(); t2.join(); std::cout << "r1 = " << r1 << ", r2 = " << r2 << std::endl; return 0; }

In this example, std::atomic_thread_fence is used with std::memory_order_acquire to create a barrier that ensures memory operations before the fence are visible to other threads. This enforces the desired synchronization between threads.

Thread Pooling

Thread pooling is a technique used to efficiently manage a group of worker threads, allowing you to parallelize and execute multiple tasks concurrently. Thread pools are useful in scenarios where creating and destroying threads for each task is costly, and you want to maintain a set of reusable threads.

Implementing a Thread Pool

To create a thread pool, you typically define a fixed number of worker threads that are responsible for executing tasks. The pool manages the thread lifecycle, and you can submit tasks to it.

#include <iostream> #include <thread> #include <vector> #include <queue> #include <functional> #include <mutex> #include <condition_variable> class ThreadPool { public: ThreadPool(size_t numThreads) { for (size_t i = 0; i < numThreads; ++i) { workers.emplace_back([this] { while (true) { std::function<void()> task; { std::unique_lock<std::mutex> lock(queueMutex); condition.wait(lock, [this] { return !tasks.empty() stop; }); if (stop && tasks.empty()) { return; } task = tasks.front(); tasks.pop(); } task(); } }); } } template <class F> void enqueue(F&& f) { { std::unique_lock<std::mutex> lock(queueMutex); tasks.emplace(std::forward<F>(f)); } condition.notify_one(); } ~ThreadPool() { { std::unique_lock<std::mutex> lock(queueMutex); stop = true; } condition.notify_all(); for (std::thread& worker : workers) { worker.join(); } } private: std::vector<std::thread> workers; std::queue<std::function<void()>> tasks; std::mutex queueMutex; std::condition_variable condition; bool stop = false; }; int main() { ThreadPool pool(4); for (int i = 0; i < 8; ++i) { pool.enqueue([i] { std::cout << "Task " << i << " executed by thread " << std::this_thread::get_id() << std::endl; }); } return 0; }

In this example, a thread pool is implemented with a fixed number of worker threads that execute tasks submitted via the enqueue method. The pool ensures efficient reuse of threads.

Task Scheduling and Load Balancing

Thread pools handle task scheduling and load balancing automatically. As tasks are enqueued, worker threads pick them up and execute them. The pool ensures that tasks are distributed evenly among the available threads, promoting load balancing.

Parallelism

Parallelism in C++ involves dividing a task into smaller subtasks and executing them concurrently, typically to improve the performance of computationally intensive operations. The C++ Standard Library provides parallel algorithms and execution policies for achieving parallelism.

Parallel Algorithms (C++11/C++14)

C++11 and C++14 introduced parallel algorithms for common operations like sorting, searching, and transformation. These algorithms allow you to utilize the power of multiple threads to perform these operations efficiently.

#include <iostream> #include <vector> #include <algorithm> #include <thread> void printSquare(int x) { std::cout << x * x << " "; } int main() { std::vector<int> data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; std::for_each(std::execution::par, data.begin(), data.end(), printSquare); return 0; }

In this example, std::for_each is used with the std::execution::par execution policy to parallelize the application of the printSquare function to each element of the vector.

Execution Policies (C++17)

C++17 introduced execution policies as a standard way to specify the parallelism level for standard algorithms. These policies include std::execution::seq (sequential), std::execution::par (parallel), and std::execution::par_unseq (parallel with vectorization).

#include <iostream> #include <vector> #include <algorithm> void printSquare(int x) { std::cout << x * x << " "; } int main() { std::vector<int> data = {1, 2, 3, 4, 5, 6, 7, 8, 9, 10}; std::for_each(std::execution::par, data.begin(), data.end(), printSquare); return 0; }

In this example, std::execution::par specifies that the std::for_each algorithm should run in parallel, using multiple threads for processing the data.

Performance and Scalability

Performance and scalability are critical aspects of multithreaded applications. Optimizing code for parallel execution and considering scalability are essential for achieving the best performance in a multi-threaded environment.

Profiling and Optimizing Multithreaded Code

Profiling: Profiling tools help identify performance bottlenecks and areas of improvement in multithreaded code. Tools like gprof, Valgrind, and Intel VTune can be used to profile CPU and memory usage.

g++ -pg -o my_program my_program.cpp ./my_program gprof my_program

Optimization Techniques

To optimize multithreaded code, consider techniques like minimizing lock contention, reducing synchronization, and improving data locality. Using thread-safe data structures, employing fine-grained locking, and utilizing thread-local storage can all contribute to better performance.

#include <iostream> #include <thread> #include <mutex> int sharedValue = 0; std::mutex myMutex; void incrementValue(int iterations) { for (int i = 0; i < iterations; ++i) { std::lock_guard<std::mutex> lock(myMutex); sharedValue++; } } int main() { const int numThreads = 4; const int iterations = 1000000; std::thread threads[numThreads]; for (int i = 0; i < numThreads; ++i) { threads[i] = std::thread(incrementValue, iterations); } for (int i = 0; i < numThreads; ++i) { threads[i].join(); } std::cout << "Shared value: " << sharedValue << std::endl; return 0; }

Scalability Considerations

Load Balancing: In a scalable multithreaded application, tasks should be distributed evenly among threads to maximize CPU utilization. Load balancing techniques are essential to ensure that no threads are idle while others are overloaded.

#include <iostream> #include <thread> #include <vector> #include <numeric> void parallelSum(const std::vector<int>& data, int start, int end, int& result) { result = std::accumulate(data.begin() + start, data.begin() + end, 0); } int main() { const int numThreads = 4; std::vector<int> data(1000000, 1); std::vector<std::thread> threads(numThreads); std::vector<int> partialSums(numThreads); int totalSum = 0; int chunkSize = data.size() / numThreads; for (int i = 0; i < numThreads; ++i) { int start = i * chunkSize; int end = (i == numThreads - 1) ? data.size() : (i + 1) * chunkSize; threads[i] = std::thread(parallelSum, std::ref(data), start, end, std::ref(partialSums[i])); } for (int i = 0; i < numThreads; ++i) { threads[i].join(); totalSum += partialSums[i]; } std::cout << "Total sum: " << totalSum << std::endl; return 0; }

In this example, tasks are divided into chunks, and each thread computes a partial sum. This approach ensures load balancing and efficient CPU utilization in the presence of multiple threads.

C++ Multi-threading - Example

This program calculates the sum of elements in an array in parallel using a thread pool with load balancing and synchronization:

#include <iostream> #include <vector> #include <thread> #include <mutex> #include <condition_variable> #include <algorithm> class ThreadPool { public: ThreadPool(size_t numThreads) : stop(false) { for (size_t i = 0; i < numThreads; ++i) { workers.emplace_back([this] { while (true) { std::function<void()> task; { std::unique_lock<std::mutex> lock(queueMutex); condition.wait(lock, [this] { return stop !tasks.empty(); }); if (stop && tasks.empty()) { return; } task = tasks.front(); tasks.pop(); } task(); } }); } } template <class F> void enqueue(F&& f) { { std::unique_lock<std::mutex> lock(queueMutex); tasks.emplace(std::forward<F>(f)); } condition.notify_one(); } ~ThreadPool() { { std::unique_lock<std::mutex> lock(queueMutex); stop = true; } condition.notify_all(); for (std::thread& worker : workers) { worker.join(); } } private: std::vector<std::thread> workers; std::queue<std::function<void()>> tasks; std::mutex queueMutex; std::condition_variable condition; bool stop; }; std::mutex resultMutex; int totalSum = 0; void parallelSum(const std::vector<int>& data, int start, int end) { int localSum = std::accumulate(data.begin() + start, data.begin() + end, 0); { std::lock_guard<std::mutex> lock(resultMutex); totalSum += localSum; } } int main() { const int numThreads = 4; std::vector<int> data(1000000, 1); ThreadPool pool(numThreads); int chunkSize = data.size() / numThreads; for (int i = 0; i < numThreads; ++i) { int start = i * chunkSize; int end = (i == numThreads - 1) ? data.size() : (i + 1) * chunkSize; pool.enqueue([=, &data] { parallelSum(data, start, end); }); } // Main thread waits for all worker threads to finish // The pool's destructor will ensure all threads are joined when it goes out of scope. // Alternatively, you can manually join threads here. std::cout << "Total sum: " << totalSum << std::endl; return 0; }

//Output: Total sum: 1000000

Above program:

We define a ThreadPool class to manage worker threads.
We create a thread pool with four threads.
The parallelSum function calculates the local sum for a portion of the array and updates the global total sum.
Tasks are enqueued in the thread pool to perform parallel summation.
Load balancing is achieved by dividing the array into equal-sized chunks and assigning each chunk to a thread.
Synchronization using a mutex ensures the safety of the global totalSum.
The program prints the final total sum after parallel execution.

When you run this program, you should observe parallel execution across the threads, resulting in faster computation. The output will display the calculated total sum.

Conclusion

C++ provides extensive support for multi-threading, allowing you to utilize the power of modern multi-core processors and develop efficient, concurrent applications. However, multi-threading also introduces complexities, including race conditions and deadlocks, which require careful consideration and synchronization to ensure program correctness and stability.

Next > C++ Memory Management - Explained

Related Topics

More Related Topics.....