Skip to content

Replace per-iteration std::async with a persistent thread pool #332

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
saleelk wants to merge 1 commit into
base: candidate
Choose a base branch
from
+168 −63
Open

Replace per-iteration std::async with a persistent thread pool #332

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
231 changes: 168 additions & 63 deletions src/header/TransferBench.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -30,10 +30,13 @@ THE SOFTWARE.
#include <ifaddrs.h>
#include <filesystem>
#include <fstream>
#include <condition_variable>
#include <functional>
#include <future>
#include <map>
#include <memory>
#include <mutex>
#include <queue>
#include <net/if.h>
#include <netdb.h>
#include <netinet/in.h>
Expand Down Expand Up @@ -5401,6 +5404,100 @@ static bool IsConfiguredGid(union ibv_gid const& gid)
return ERR_NONE;
}

// Persistent worker-thread pool.

class ThreadPool {
public:
explicit ThreadPool(size_t numThreads) : stop_(false) {
for (size_t i = 0; i < numThreads; ++i) {
workers_.emplace_back([this] {
for (;;) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(mutex_);
cv_.wait(lock, [this] { return stop_ || !tasks_.empty(); });
if (stop_ && tasks_.empty()) return;
task = std::move(tasks_.front());
tasks_.pop();
}
task();
}
});
}
}

ThreadPool(ThreadPool const&) = delete;
ThreadPool& operator=(ThreadPool const&) = delete;

~ThreadPool() {
{
std::lock_guard<std::mutex> lock(mutex_);
stop_ = true;
}
cv_.notify_all();
for (auto& w : workers_)
w.join();
}

size_t size() const { return workers_.size(); }

std::future<ErrResult> submit(std::function<ErrResult()> fn) {
auto task = std::make_shared<std::packaged_task<ErrResult()>>(std::move(fn));
std::future<ErrResult> fut = task->get_future();
{
std::lock_guard<std::mutex> lock(mutex_);
tasks_.emplace([task] { (*task)(); });
}
cv_.notify_one();
return fut;
}

private:
std::vector<std::thread> workers_;
std::queue<std::function<void()>> tasks_;
std::mutex mutex_;
std::condition_variable cv_;
bool stop_;
};

// Pools are created once per RunTransfers call (sized from the run's executor/transfer counts)
// and cleared when it returns. Two separate pools avoid a nested-blocking deadlock: executor
// workers submit their transfers to the transfer pool and block on the results, so the two
// levels must never share threads.
static ThreadPool* g_executorPool = nullptr;
static ThreadPool* g_transferPool = nullptr;

// Runs "count" units of work. Fall to serial execution if the pool is unavailable/empty.
template <typename MakeTask>
static inline ErrResult ParallelRun(ThreadPool* pool, size_t count, MakeTask makeTask) {
Comment on lines +5467 to +5472
if (count == 0) return ERR_NONE;

bool const usePool = (pool != nullptr && pool->size() > 0 && count > 1);

std::vector<std::future<ErrResult>> futures;
if (usePool) {
futures.reserve(count - 1);
for (size_t i = 1; i < count; ++i)
futures.emplace_back(pool->submit(makeTask(i)));
}

// Unit 0 always runs inline on the calling thread.
ErrResult result = makeTask(0)();

if (usePool) {
for (auto& f : futures) {
ErrResult const r = f.get();
if (result.errType == ERR_NONE && r.errType != ERR_NONE) result = r;
}
} else {
for (size_t i = 1; i < count; ++i) {
ErrResult const r = makeTask(i)();
if (result.errType == ERR_NONE && r.errType != ERR_NONE) result = r;
}
Comment on lines +5488 to +5496
}
return result;
}

// Execute a single GPU executor
static ErrResult RunGpuExecutor(int const iteration,
ConfigOptions const& cfg,
Expand All @@ -5413,24 +5510,22 @@ static bool IsConfiguredGid(union ibv_gid const& gid)
int xccDim = exeInfo.useSubIndices ? exeInfo.numSubIndices : 1;

if (cfg.gfx.useMultiStream) {
// Launch one thread per Transfer in separate streams
vector<std::future<ErrResult>> asyncTransfers;
for (int i = 0; i < exeInfo.streams.size(); i++) {
asyncTransfers.emplace_back(std::async(std::launch::async,
ExecuteGpuTransfer,
iteration,
exeInfo.totalSubExecs,
exeInfo.subExecParamGpu,
exeInfo.streams[i],
cfg.gfx.useHipEvents ? exeInfo.startEvents[i] : NULL,
cfg.gfx.useHipEvents ? exeInfo.stopEvents[i] : NULL,
xccDim,
std::cref(cfg),
exeInfo.gfxKernelToUse,
std::ref(exeInfo.resources[i])));
}
for (auto& asyncTransfer : asyncTransfers)
ERR_CHECK(asyncTransfer.get());
// Launch one Transfer per stream
ERR_CHECK(ParallelRun(g_transferPool, exeInfo.streams.size(),
[&](size_t i) -> std::function<ErrResult()> {
return [&, i] {
return ExecuteGpuTransfer(iteration,
exeInfo.totalSubExecs,
exeInfo.subExecParamGpu,
exeInfo.streams[i],
cfg.gfx.useHipEvents ? exeInfo.startEvents[i] : NULL,
cfg.gfx.useHipEvents ? exeInfo.stopEvents[i] : NULL,
xccDim,
cfg,
exeInfo.gfxKernelToUse,
exeInfo.resources[i]);
};
}));
} else {
// Launch all Transfers in one kernel launch (avoid extra thread creation)
ExecuteGpuTransfer(iteration, exeInfo.totalSubExecs, exeInfo.subExecParamGpu, exeInfo.streams[0],
Expand Down Expand Up @@ -5592,22 +5687,19 @@ static bool IsConfiguredGid(union ibv_gid const& gid)
auto cpuStart = std::chrono::high_resolution_clock::now();
ERR_CHECK(hipSetDevice(exeIndex));

vector<std::future<ErrResult>> asyncTransfers;
for (int i = 0; i < exeInfo.resources.size(); i++) {
asyncTransfers.emplace_back(std::async(std::launch::async,
ExecuteDmaTransfer,
iteration,
exeInfo.useSubIndices,
exeIndex,
exeInfo.streams[i],
cfg.dma.useHipEvents ? exeInfo.startEvents[i] : NULL,
cfg.dma.useHipEvents ? exeInfo.stopEvents[i] : NULL,
std::cref(cfg),
std::ref(exeInfo.resources[i])));
}

for (auto& asyncTransfer : asyncTransfers)
ERR_CHECK(asyncTransfer.get());
ERR_CHECK(ParallelRun(g_transferPool, exeInfo.resources.size(),
[&](size_t i) -> std::function<ErrResult()> {
return [&, i] {
return ExecuteDmaTransfer(iteration,
exeInfo.useSubIndices,
exeIndex,
exeInfo.streams[i],
cfg.dma.useHipEvents ? exeInfo.startEvents[i] : NULL,
cfg.dma.useHipEvents ? exeInfo.stopEvents[i] : NULL,
cfg,
exeInfo.resources[i]);
};
}));

auto cpuDelta = std::chrono::high_resolution_clock::now() - cpuStart;
double deltaMsec = std::chrono::duration_cast<std::chrono::duration<double>>(cpuDelta).count() * 1000.0 / cfg.general.numSubIterations;
Expand Down Expand Up @@ -5679,21 +5771,18 @@ static bool IsConfiguredGid(union ibv_gid const& gid)
auto cpuStart = std::chrono::high_resolution_clock::now();
ERR_CHECK(hipSetDevice(exeIndex));

vector<std::future<ErrResult>> asyncTransfers;
for (int i = 0; i < exeInfo.resources.size(); i++) {
asyncTransfers.emplace_back(std::async(std::launch::async,
ExecuteBatchDmaTransfer,
iteration,
exeIndex,
exeInfo.streams[i],
cfg.dma.useHipEvents ? exeInfo.startEvents[i] : NULL,
cfg.dma.useHipEvents ? exeInfo.stopEvents[i] : NULL,
std::cref(cfg),
std::ref(exeInfo.resources[i])));
}

for (auto& asyncTransfer : asyncTransfers)
ERR_CHECK(asyncTransfer.get());
ERR_CHECK(ParallelRun(g_transferPool, exeInfo.resources.size(),
[&](size_t i) -> std::function<ErrResult()> {
return [&, i] {
return ExecuteBatchDmaTransfer(iteration,
exeIndex,
exeInfo.streams[i],
cfg.dma.useHipEvents ? exeInfo.startEvents[i] : NULL,
cfg.dma.useHipEvents ? exeInfo.stopEvents[i] : NULL,
cfg,
exeInfo.resources[i]);
};
}));

auto cpuDelta = std::chrono::high_resolution_clock::now() - cpuStart;
double deltaMsec = std::chrono::duration_cast<std::chrono::duration<double>>(cpuDelta).count() * 1000.0 / cfg.general.numSubIterations;
Expand Down Expand Up @@ -5972,6 +6061,20 @@ static bool IsConfiguredGid(union ibv_gid const& gid)
System::Get().Barrier();
}

// Create persistent worker pools once for this run (see ThreadPool). Sized so every executor
// and every transfer can run concurrently without the two levels sharing threads:
// - executor pool: one worker per local executor (unit 0 runs inline, so N-1 are farmed)
// - transfer pool: one worker per local transfer across all executors
size_t numLocalExecutors = localExecutors.size();
size_t numLocalTransfers = 0;
for (auto const& exeDevice : localExecutors)
numLocalTransfers += executorMap[exeDevice].resources.size();

ThreadPool executorPool(numLocalExecutors);
ThreadPool transferPool(numLocalTransfers);
g_executorPool = &executorPool;
g_transferPool = &transferPool;
Comment on lines +6073 to +6076

// Perform iterations
size_t numTimedIterations = 0;
double totalCpuTimeSec = 0.0;
Expand All @@ -5990,20 +6093,18 @@ static bool IsConfiguredGid(union ibv_gid const& gid)
// Start CPU timing for this iteration
auto cpuStart = std::chrono::high_resolution_clock::now();

// Execute all Transfers in parallel
std::vector<std::future<ErrResult>> asyncExecutors;
for (auto const& exeDevice : localExecutors) {
asyncExecutors.emplace_back(std::async(std::launch::async, RunExecutor,
iteration,
std::cref(cfg),
std::cref(exeDevice),
std::ref(executorMap[exeDevice])));
}

// Wait for all threads to finish
for (auto& asyncExecutor : asyncExecutors) {
ERR_APPEND(asyncExecutor.get(), errResults);
}
// Execute all executors in parallel via the persistent executor pool. Each ExeInfo& is
// resolved on the calling thread (inside makeTask) to keep std::map access single-threaded,
// matching the previous behaviour.
ERR_APPEND(ParallelRun(g_executorPool, localExecutors.size(),
[&](size_t i) -> std::function<ErrResult()> {
ExeDevice const& exeDevice = localExecutors[i];
ExeInfo& exeInfo = executorMap[exeDevice];
return [&cfg, &exeDevice, &exeInfo, iteration] {
return RunExecutor(iteration, cfg, exeDevice, exeInfo);
};
}),
errResults);

// Wait for all ranks to finish
System::Get().Barrier();
Expand All @@ -6023,6 +6124,10 @@ static bool IsConfiguredGid(union ibv_gid const& gid)
}
}

// Done dispatching; drop pool pointers before the pools go out of scope below.
g_executorPool = nullptr;
g_transferPool = nullptr;

// Pause for interactive mode - run validation and show per-transfer result before prompt
if (cfg.general.useInteractive) {
if (localRank == 0) {
Expand Down
Loading