一旦有一個已洗掉的問題，我寫了一個很大的答案，但是這個問題被洗掉了，作者拒絕取消洗掉它。
所以在這里發布這個問題的簡短摘要。并立即親自回答這個問題，只是為了分享我的結果。
問題是，如果我們std::bitset<65536>在內部回圈中逐位處理（通過某些公式），那么我們如何提高這種計算？
外回圈只是多次呼叫內回圈（比如說50 000次），而外回圈不能并行處理，因為每次下一次迭代都取決于上一次迭代的結果。
此程序的示例代碼：
std::bitset<65536> bits{};
uint64_t hash = 0;
for (size_t i = 0; i < 50000; i) {
// Process Bits
for (size_t j = 0; j < bits.size(); j)
bits[j] = ModifyBit(i, j, hash, bits[j]);
hash = Hash(bits, hash);
}
上面的代碼只是一種處理示例，并非真實案例。真實的情況是，很多時候我們std::bitset<65536>以某種方式處理所有位都可以獨立處理。
問題是我們如何在內部回圈中盡可能快地并行處理位。
一個重要的注意事項是修改位的公式是通用的，這意味著我們事先不知道它并且不能從中生成SIMD指令。
但我們知道的是，所有位都可以獨立處理。我們需要并行化這個處理。我們也不能并行化外回圈，因為它的每次迭代都取決于前一次迭代的結果。
另一個注意事項是 std::bitset<65536> 非常小，只有 1K 的 64 位字。因此，這意味著直接使用 std::thread 的 std::async 執行緒池將不起作用，因為每個執行緒的作業時間僅為 50-200 納秒左右，啟動和停止執行緒并將作業發送給它們的時間非常短。甚至 std::mutex 在我的 Windows 機器上也需要 75 納秒（雖然在 Linux 上是 20 納秒），所以使用 std::mutex 也是一個很大的開銷。
人們可能會假設上述ModifyBit()函式的每一位花費大約相同的時間，否則無法理解如何安排回圈的平衡并行化，只能將其切成非常多的小任務，希望較長的任務將被幾個較短的任務平衡.
uj5u.com熱心網友回復：
為您的任務實施了相當大且復雜的解決方案，但運行速度非常快。在我的 4 核（8 個硬體執行緒）筆記本電腦上，6x與單執行緒版本（您的代碼版本）相比，我的多核加速倍數。
下面解決方案的主要思想是實作非常快速的多核執行緒池來運行開銷很小的任意任務。我的實作每秒最多可以處理 1-1000 萬個任務（取決于 CPU 速度和內核數）。
異步啟動多個任務的常規方法是通過使用std::async或僅通過創建std::thread。這兩種方式都比我自己的實作慢得多。他們不能像我的實作那樣提供每秒 500 萬個任務的吞吐量。您的代碼每秒需要數百萬個任務才能以良好的速度運行。這就是為什么我從頭開始實施一切。
現在實作快速執行緒池后，我們可以將您的 64K 位集分割成更小的子集并并行處理這些子集。我將 64K bitset 分成 16 個相等的部分（見BitSize / 16代碼），您可以將其他部分設定為等于 2 的冪，但不要太多，否則執行緒池開銷會太大。通常最好將部分數量分割成硬體執行緒數量的兩倍（或內核數量的 4 倍）。
我用 C 代碼實作了幾個類。AtomicMutex類使用std::atomic_flag來實作非常快速的基于自旋鎖定方法的互斥體替換。此 AtomicMutex 用于保護提交以在執行緒池上運行的任務佇列。
RingBuffer類基于std::vector并實作簡單快速的佇列來存盤任何物件。它使用兩個指標（頭和尾）實作，指向向量。當新元素添加到佇列中時，尾指標向右推進，如果該指標到達向量的末尾，則它會回繞到第 0 個位置。同樣的方式，當元素從佇列中取出時，頭指標也向右前進并環繞。RingBuffer 用于存盤執行緒池任務。
Queueclass 是 RingBuffer 的包裝器，但具有 AtomicMutex 保護。這個自旋鎖互斥鎖用于保護從多個作業執行緒同時向/從佇列添加/獲取元素。
Pool自己實作多核任務池。它創建的作業執行緒與 CPU 硬體執行緒（核心數量的兩倍）減一一樣多。每個作業執行緒只是從佇列中輪詢新任務并立即執行它們。主執行緒將新任務添加到佇列中。Pool 還具有 Wait() 能力來等待所有當前任務完成，這種等待用作等待整個 64K 位集被處理（所有子部分都被處理）的屏障。Pool 接受任何要運行的 lambda（函式閉包）。你可以看到被分割成更小的部分的 64K 位集是通過 dopool.Emplace(lambda)和 later處理的pool.Wait()用于等待所有子部分完成。如果有任何錯誤，池作業人員的例外會被收集并報告給用戶。在執行 Wait() 池時，也在主執行緒內運行任務，以免浪費一個核心來等待任務完成。
控制臺中報告的計時由std::chrono模塊完成。
可以運行兩個版本 - 單執行緒（您的原始版本）和使用所有內核的多執行緒。通過將MultiThreaded = true模板引數傳遞給函式來完成單/多之間的切換ProcessBitset()。
在線嘗試！
#include <cstdint>
#include <atomic>
#include <vector>
#include <array>
#include <queue>
#include <functional>
#include <thread>
#include <future>
#include <exception>
#include <optional>
#include <memory>
#include <iostream>
#include <iomanip>
#include <bitset>
#include <string>
#include <chrono>
#include <algorithm>
#include <any>
#include <type_traits>
class AtomicMutex {
class LockerC;
public:
void lock() {
while (f_.test_and_set(std::memory_order_acquire))
//f_.wait(true, std::memory_order_acquire)
;
}
void unlock() {
f_.clear(std::memory_order_release);
//f_.notify_all();
}
LockerC Locker() { return LockerC(*this); }
private:
class LockerC {
public:
LockerC() = delete;
LockerC(AtomicMutex & mux) : pmux_(&mux) { mux.lock(); }
LockerC(LockerC const & other) = delete;
LockerC(LockerC && other) : pmux_(other.pmux_) { other.pmux_ = nullptr; }
~LockerC() { if (pmux_) pmux_->unlock(); }
LockerC & operator = (LockerC const & other) = delete;
LockerC & operator = (LockerC && other) = delete;
private:
AtomicMutex * pmux_ = nullptr;
};
std::atomic_flag f_ = ATOMIC_FLAG_INIT;
};
template <typename T>
class RingBuffer {
public:
RingBuffer() : buf_(1 << 8), last_(buf_.size() - 1) {}
T & front() { return buf_[first_]; }
T const & front() const { return buf_[first_]; }
T & back() { return buf_[last_]; }
T const & back() const { return buf_[last_]; }
size_t size() const { return size_; }
bool empty() const { return size_ == 0; }
template <typename ... Args>
void emplace(Args && ... args) {
while (size_ >= buf_.size()) {
std::rotate(&buf_[0], &buf_[first_], &buf_[buf_.size()]);
first_ = 0;
last_ = buf_.size() - 1;
buf_.resize(buf_.size() * 2);
}
size_;
last_;
if (last_ >= buf_.size())
last_ = 0;
buf_[last_] = T(std::forward<Args>(args)...);
}
void pop() {
if (size_ == 0)
return;
--size_;
first_;
if (first_ >= buf_.size())
first_ = 0;
}
private:
std::vector<T> buf_;
size_t first_ = 0, last_ = 0, size_ = 0;
};
template <typename T>
class Queue {
public:
size_t Size() const { return q_.size(); }
bool Empty() const { return q_.size() == 0; }
template <typename ... Args>
void Emplace(Args && ... args) {
auto lock = m_.Locker();
q_.emplace(std::forward<Args>(args)...);
}
T Pop(std::function<void()> const & on_empty = []{},
std::function<void()> const & on_full = []{}) {
while (true) {
if (q_.empty()) {
on_empty();
continue;
}
auto lock = m_.Locker();
if (q_.empty()) {
on_empty();
continue;
}
on_full();
T val = std::move(q_.front());
q_.pop();
return std::move(val);
}
}
std::optional<T> TryPop() {
auto lock = m_.Locker();
if (q_.empty())
return std::nullopt;
T val = std::move(q_.front());
q_.pop();
return std::move(val);
}
private:
AtomicMutex m_;
RingBuffer<T> q_;
};
class RunInDestr {
public:
RunInDestr(std::function<void()> const & f) : f_(f) {}
~RunInDestr() { f_(); }
private:
std::function<void()> const & f_;
};
class Pool {
public:
struct FinishExc {};
struct Worker {
std::unique_ptr<std::atomic<bool>> pdone = std::make_unique<std::atomic<bool>>(true);
std::unique_ptr<std::exception_ptr> pexc = std::make_unique<std::exception_ptr>();
std::unique_ptr<std::thread> thr;
};
Pool(size_t nthreads = size_t(-1)) {
if (nthreads == size_t(-1))
nthreads = std::thread::hardware_concurrency() - 1;
std::cout << "Pool has " << nthreads << " worker threads." << std::endl;
for (size_t i = 0; i < nthreads; i) {
workers_.emplace_back(Worker{});
workers_.back().thr = std::make_unique<std::thread>(
[&, pdone = workers_.back().pdone.get(), pexc = workers_.back().pexc.get()]{
try {
std::function<void()> f_done = [pdone]{
pdone->store(true, std::memory_order_relaxed);
}, f_empty = [this]{
CheckFinish();
}, f_full = [pdone]{
pdone->store(false, std::memory_order_relaxed);
};
while (true) {
RunInDestr set_done(f_done);
tasks_.Pop(f_empty, f_full)();
}
} catch (...) {
exc_.store(true, std::memory_order_relaxed);
*pexc = std::current_exception();
}
});
}
}
~Pool() {
Wait();
Finish();
}
void CheckExc() {
if (!exc_.load(std::memory_order_relaxed))
return;
Finish();
throw std::runtime_error("Pool: Exception occured!");
}
void Finish() {
finish_ = true;
for (auto & w: workers_)
try {
w.thr->join();
if (*w.pexc)
std::rethrow_exception(*w.pexc);
} catch (FinishExc const &) {}
workers_.clear();
}
template <typename ... Args>
void Emplace(Args && ... args) {
CheckExc();
tasks_.Emplace(std::forward<Args>(args)...);
}
void Wait() {
while (true) {
auto task = tasks_.TryPop();
if (!task)
break;
(*task)();
}
while (true) {
bool done = true;
for (auto & w: workers_)
if (!w.pdone->load(std::memory_order_relaxed)) {
done = false;
break;
}
if (done)
break;
}
CheckExc();
}
private:
void CheckFinish() {
if (finish_)
throw FinishExc{};
}
Queue<std::function<void()>> tasks_;
std::vector<Worker> workers_;
bool finish_ = false;
std::atomic<bool> exc_ = false;
};
template <bool MultiThreaded = true, size_t BitSize>
void ProcessBitset(Pool & pool, std::bitset<BitSize> & bset,
std::string const & businessLogicCriteria) {
static size_t constexpr block = BitSize / 16;
for (int j = 0; j < BitSize; j = block) {
auto task = [&bset, j]{
int const hi = std::min(j block, BitSize);
for (int i = j; i < hi; i) {
if (i % 2 == 0)
bset[i] = 0;
else
bset[i] = 1;
}
};
if constexpr(MultiThreaded)
pool.Emplace(std::move(task));
else
task();
}
if constexpr(MultiThreaded)
pool.Wait();
}
static auto const gtb = std::chrono::high_resolution_clock::now();
double Time() {
return std::chrono::duration_cast<std::chrono::duration<double>>(
std::chrono::high_resolution_clock::now() - gtb).count();
}
void Compute() {
Pool pool;
std::bitset<65536> bset;
std::string businessLogicCriteria;
int const hi = 50000;
for (int j = 0; j < hi; j) {
if ((j & 0x1FFF) == 0 || j 1 >= hi)
std::cout << j / 1000 << "K (" << std::fixed << std::setprecision(3) << Time() << " sec), " << std::flush;
ProcessBitset(pool, bset, businessLogicCriteria);
businessLogicCriteria = "...";
}
}
void TimeMeasure() {
size_t constexpr A = 1 << 16, B = 1 << 5;
{
Pool pool;
auto const tb = Time();
int64_t volatile x = 0;
for (size_t i = 0; i < A; i) {
for (size_t j = 0; j < B; j)
pool.Emplace([&]{ x = x 1; });
pool.Wait();
}
std::cout << "AtomicPool time " << std::fixed << std::setprecision(3) << (Time() - tb)
<< " sec, speed " << A * B / (Time() - tb) / 1000.0 << " empty K-tasks/sec, "
<< 1'000'000 / (A * B / (Time() - tb)) << " sec/M-task, no-collisions "
<< std::setprecision(7) << double(x) / (A * B) << std::endl;
}
{
auto const tb = Time();
//size_t const nthr = std::thread::hardware_concurrency();
size_t constexpr C = A / 8;
std::vector<std::future<void>> asyncs;
int64_t volatile x = 0;
for (size_t i = 0; i < C; i) {
asyncs.clear();
for (size_t j = 0; j < B; j)
asyncs.emplace_back(std::async(std::launch::async, [&]{ x = x 1; }));
asyncs.clear();
}
std::cout << "AsyncPool time " << std::fixed << std::setprecision(3) << (Time() - tb)
<< " sec, speed " << C * B / (Time() - tb) / 1000.0 << " empty K-tasks/sec, "
<< 1'000'000 / (C * B / (Time() - tb)) << " sec/M-task, no-collisions "
<< std::setprecision(7) << double(x) / (C * B) << std::endl;
}
}
int main() {
try {
TimeMeasure();
Compute();
return 0;
} catch (std::exception const & ex) {
std::cout << "Exception: " << ex.what() << std::endl;
return -1;
} catch (...) {
std::cout << "Unknown Exception!" << std::endl;
return -1;
}
}
4 核（8 個硬體執行緒）的輸出：
Pool has 7 worker threads.
AtomicPool time 0.903 sec, speed 2321.831 empty K-tasks/sec, 0.431 sec/M-task, no-collisions 0.9999967
AsyncPool time 0.982 sec, speed 266.789 empty K-tasks/sec, 3.750 sec/M-task, no-collisions 0.9999123
Pool has 7 worker threads.
0K (0.074 sec), 8K (0.670 sec), 16K (1.257 sec), 24K (1.852 sec), 32K (2.435 sec), 40K (2.984 sec), 49K (3.650 sec), 49K (3.711 sec),
下面的比較是單執行緒版本的時間，即6x時間慢：
0K (0.125 sec), 8K (3.786 sec), 16K (7.754 sec), 24K (11.202 sec), 32K (14.662 sec), 40K (18.056 sec), 49K (21.470 sec), 49K (21.841 sec),
uj5u.com熱心網友回復：
你有這個要并行化的內部回圈：
for (size_t j = 0; j < bits.size(); j)
bits[j] = ModifyBit(i, j, hash, bits[j]);
所以一個好主意是將它分成塊，并讓多個執行緒并行處理每個塊。std::atomic<int>您可以使用遞增的計數器輕松地將塊提交給作業人員，以識別要處理的塊。您還可以確保執行緒在一個回圈后全部停止作業，然后再使用以下命令開始下一個回圈std::barrier：
std::bitset<65536> bits{};
std::thread pool[8]; // Change size accordingly
std::atomic<int> task_number{0};
constexpr std::size_t tasks_per_loop = 32; // Arbitrarily chosen
constexpr std::size_t block_size = (bits.size() tasks_per_loop-1) / tasks_per_loop;
// (only written to by one thread by the barrier, so not atomic)
uint64_t hash = 0;
int i = 0;
std::barrier barrier(std::size(pool), [&]() {
task_number = 0;
i;
hash = Hash(bits, hash);
});
for (std::thread& t : pool) {
t = std::thread([&]{
while (i < 50000) {
for (int t; (t = task_number ) < tasks_per_loop;) {
int block_start = t * block_size;
int block_end = std::min(block_start block_size, bits.size());
for (int j = block_start; j < block_end; j) {
bits[j] = ModifyBit(i, j, hash, bits[j]);
}
}
// Wait for other threads to finish and hash
// to be calculated before starting next loop
barrier.arrive_and_wait();
}
});
}
for (std::thread& t : pool) t.join();
（用 OpenMP 并行化 for 回圈的看似簡單的方法在#pragma omp parallel for一些測驗中似乎更慢，可能是因為任務太小了）
這是針對您運行類似代碼的實作：https ://godbolt.org/z/en76Kv4nn
std::function<void()>在我的機器上，以 100 萬次迭代運行幾次，使用我的方法需要 28 到 32 秒，而使用一般執行緒池方法需要 44 到 50 秒（因為它不能執行任意任務，所以這不太通用）。
轉載請註明出處，本文鏈接：https://www.uj5u.com/yidong/474476.html