denshooter/mxtasking
1
0
Fork 0
mirror of https://github.com/mmueller41/mxtasking.git synced 2026-01-21 12:42:57 +01:00
Code Issues Packages Projects Releases Wiki Activity

Replaced pthreads with Caladan threads.

Browse Source
This commit is contained in:
Michael Mueller
2024-07-23 17:17:47 +02:00
parent 7e692c5265
commit f3debbbd70
1 changed files with 9 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
src/mx/tasking/scheduler.cpp
View File

@@ -4,9 +4,13 @@
#include <mx/synchronization/synchronization.h> #include <mx/synchronization/synchronization.h>
#include <mx/system/thread.h> #include <mx/system/thread.h>
#include <mx/system/topology.h> #include <mx/system/topology.h>
#include <thread>
#include <vector> #include <vector>
#define SHENANGO
#include <cc/thread.h>
#include <runtime/sync.h>
#include <cc/runtime.h>
using namespace mx::tasking; using namespace mx::tasking;
Scheduler::Scheduler(const mx::util::core_set &core_set, const std::uint16_t prefetch_distance, Scheduler::Scheduler(const mx::util::core_set &core_set, const std::uint16_t prefetch_distance,
@@ -41,13 +45,11 @@ Scheduler::~Scheduler() noexcept
void Scheduler::start_and_wait() void Scheduler::start_and_wait()
{ {
// Create threads for worker... // Create threads for worker...
std::vector<std::thread> worker_threads(this->_core_set.size() + std::vector<rt::Thread> worker_threads(this->_core_set.size() +
static_cast<std::uint16_t>(config::memory_reclamation() != config::None)); static_cast<std::uint16_t>(config::memory_reclamation() != config::None));
for (auto channel_id = 0U; channel_id < this->_core_set.size(); ++channel_id) for (auto channel_id = 0U; channel_id < this->_core_set.size(); ++channel_id)
{ {
worker_threads[channel_id] = std::thread([this, channel_id] { this->_worker[channel_id]->execute(); }); worker_threads[channel_id] = rt::Thread([this, channel_id] { this->_worker[channel_id]->execute(); });
system::thread::pin(worker_threads[channel_id], this->_worker[channel_id]->core_id());
} }
// ... and epoch management (if enabled). // ... and epoch management (if enabled).
@@ -55,7 +57,7 @@ void Scheduler::start_and_wait()
{ {
// In case we enable memory reclamation: Use an additional thread. // In case we enable memory reclamation: Use an additional thread.
worker_threads[this->_core_set.size()] = worker_threads[this->_core_set.size()] =
std::thread([this] { this->_epoch_manager.enter_epoch_periodically(); }); rt::Thread([this] { this->_epoch_manager.enter_epoch_periodically(); });
} }
// Turning the flag on starts all worker threads to execute tasks. // Turning the flag on starts all worker threads to execute tasks.
@@ -66,7 +68,7 @@ void Scheduler::start_and_wait()
// from somewhere in the application. // from somewhere in the application.
for (auto &worker_thread : worker_threads) for (auto &worker_thread : worker_threads)
{ {
worker_thread.join(); worker_thread.Join();
} }
if constexpr (config::memory_reclamation() != config::None) if constexpr (config::memory_reclamation() != config::None)