std::vector is one of the most commonly used data structures in C++. Compared to regular arrays/std::array, it supports dynamic capacity expansion, making it more flexible. Unlike raw pointers that require manual memory management, it automatically manages memory. It is contiguous in memory, supporting O(1) time complexity for random access. It seamlessly integrates with algorithms in the C++ standard library, providing powerful functionality. These features make std::vector one of the most widely used data structures.
In multi-threaded computing scenarios, there are times when you need to store the results of multi-threaded calculations in a vector. In this case, you need to manually lock the areas where the vector is accessed, which can be cumbersome. Therefore, having a thread-safe vector would simplify many tasks. There are various implementations available, such as Intel’s TBB library and Microsoft’s PPL library, which provide implementations of concurrent_vector. However, sometimes we may not want to introduce too many third-party libraries. Let’s look at our own implementation.
First, the initial idea is quite simple: encapsulate a regular vector and lock the relevant operations. It is generally believed that vector is a structure where read operations outnumber write operations, so we can use shared_mutex as the basic lock, using shared_lock for read operations and unique_lock for write operations. A brief introduction to the difference between shared_mutex and regular mutex: a regular mutex can only be locked by one thread at a time and can be used with lock_guard or unique_lock for automatic unlocking; whereas a shared_mutex allows multiple threads to acquire the lock simultaneously through shared_lock, which is useful when read operations are more frequent than write operations, as it allows multiple threads to enter without waiting. However, when using unique_lock with shared_mutex, only one thread can enter, which is typically used during write operations. If both read and write operations are frequent, shared_mutex may have slightly lower performance due to maintaining a count, in which case switching to a regular mutex is advisable.
At this point, we can easily write the following code, essentially needing to encapsulate the read operations with std::shared_lock<std::shared_mutex> lock(mutex_); and encapsulate the write operations with std::unique_lock<std::shared_mutex> lock(mutex_); For the assignment operator, we need to lock this’s lock for writing and lock other’s lock for reading. Here we use std::lock to lock both in order to avoid the possibility of deadlock (if we lock one by one, and in some function, the locks are acquired in the opposite order, it will lead to deadlock).
#pragma once
#include <vector>
#include <shared_mutex>
#include <mutex>
#include <memory>
#include <iterator>
#include <algorithm>
#include <initializer_list>
namespace concurrency {
template<typename T, typename Allocator = std::allocator<T>>
class concurrent_vector {
public:
using value_type = T;
using allocator_type = Allocator;
using size_type = std::size_t;
using difference_type = std::ptrdiff_t;
using reference = value_type&;
using const_reference = const value_type&;
private:
mutable std::shared_mutex mutex_;
std::vector<T, Allocator> data_;
public:
// Constructor
concurrent_vector() = default;
explicit concurrent_vector(const Allocator& alloc)
: data_(alloc) {}
concurrent_vector(size_type count, const T& value, const Allocator& alloc = Allocator())
: data_(count, value, alloc) {}
explicit concurrent_vector(size_type count, const Allocator& alloc = Allocator())
: data_(count, alloc) {}
template<typename InputIt>
concurrent_vector(InputIt first, InputIt last, const Allocator& alloc = Allocator())
: data_(first, last, alloc) {}
concurrent_vector(const concurrent_vector& other) {
std::shared_lock<std::shared_mutex> lock(other.mutex_);
data_ = other.data_;
}
concurrent_vector(std::initializer_list<T> init, const Allocator& alloc = Allocator())
: data_(init, alloc) {}
// Assignment operator
concurrent_vector& operator=(const concurrent_vector& other) {
if (this != &other) {
std::unique_lock<std::shared_mutex> lock1(mutex_, std::defer_lock);
std::shared_lock<std::shared_mutex> lock2(other.mutex_, std::defer_lock);
std::lock(lock1, lock2);
data_ = other.data_;
}
return *this;
}
// Element access (Note: the returned reference may become invalid after the lock is released, the caller assumes the risk)
reference at(size_type pos) {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.at(pos);
}
const_reference at(size_type pos) const {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.at(pos);
}
reference operator[](size_type pos) {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_[pos];
}
const_reference operator[](size_type pos) const {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_[pos];
}
reference front() {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.front();
}
const_reference front() const {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.front();
}
reference back() {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.back();
}
const_reference back() const {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.back();
}
// Capacity-related (requires locking)
bool empty() const noexcept {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.empty();
}
size_type size() const noexcept {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.size();
}
size_type max_size() const noexcept {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.max_size();
}
void reserve(size_type new_cap) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.reserve(new_cap);
}
size_type capacity() const noexcept {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_.capacity();
}
void shrink_to_fit() {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.shrink_to_fit();
}
// Modification operations
void clear() noexcept {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.clear();
}
void push_back(const T& value) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.push_back(value);
}
void push_back(T&& value) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.push_back(std::move(value));
}
template<typename... Args>
reference emplace_back(Args&&... args) {
std::unique_lock<std::shared_mutex> lock(mutex_);
return data_.emplace_back(std::forward<Args>(args)...);
}
void pop_back() {
std::unique_lock<std::shared_mutex> lock(mutex_);
if (!data_.empty()) {
data_.pop_back();
}
}
void resize(size_type count) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.resize(count);
}
void resize(size_type count, const value_type& value) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.resize(count, value);
}
// Thread-safe bulk operations
template<typename InputIt>
void assign(InputIt first, InputIt last) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.assign(first, last);
}
void assign(size_type count, const T& value) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.assign(count, value);
}
void assign(std::initializer_list<T> ilist) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.assign(ilist.begin(), ilist.end());
}
// Safe snapshot (avoids iterator invalidation)
std::vector<T> snapshot() const {
std::shared_lock<std::shared_mutex> lock(mutex_);
return data_;
}
// Atomic conditional operations
bool try_pop_back(T& result) {
std::unique_lock<std::shared_mutex> lock(mutex_);
if (data_.empty()) {
return false;
}
result = std::move(data_.back());
data_.pop_back();
return true;
}
// Bulk insertion
template<typename InputIt>
void append(InputIt first, InputIt last) {
std::unique_lock<std::shared_mutex> lock(mutex_);
data_.insert(data_.end(), first, last);
}
// Thread-safe search
template<typename Predicate>
bool find_if(Predicate pred, T& result) const {
std::shared_lock<std::shared_mutex> lock(mutex_);
auto it = std::find_if(data_.begin(), data_.end(), pred);
if (it != data_.end()) {
result = *it;
return true;
}
return false;
}
// Safe copy at specified position
bool safe_at(size_type pos, T& result) const {
std::shared_lock<std::shared_mutex> lock(mutex_);
if (pos >= data_.size()) {
return false;
}
result = data_[pos];
return true;
}
// Thread-safe swap
void swap(concurrent_vector& other) noexcept {
if (this != &other) {
std::unique_lock<std::shared_mutex> lock1(mutex_, std::defer_lock);
std::unique_lock<std::shared_mutex> lock2(other.mutex_, std::defer_lock);
std::lock(lock1, lock2);
data_.swap(other.data_);
}
}
};
// Specialization of swap function
template<typename T, typename Alloc>
void swap(concurrent_vector<T, Alloc>& lhs, concurrent_vector<T, Alloc>& rhs) noexcept {
lhs.swap(rhs);
}
}