Skip to content

Commit

Permalink
Format static_thread_pool.hpp
Browse files Browse the repository at this point in the history
  • Loading branch information
@ispeters
ispeters committed Oct 30, 2023
1 parent ee5bdee commit 0c2372b
Showing 1 changed file with 115 additions and 121 deletions.
236 changes: 115 additions & 121 deletions include/unifex/static_thread_pool.hpp
View file
Original file line number Diff line number Diff line change
Expand Up @@ -22,167 +22,161 @@
#include <unifex/stop_token_concepts.hpp>
#include <unifex/detail/intrusive_queue.hpp>

#include <atomic>
#include <condition_variable>
#include <mutex>
#include <thread>
#include <type_traits>
#include <vector>
#include <atomic>
#include <mutex>
#include <condition_variable>

#include <unifex/detail/prologue.hpp>

namespace unifex {
namespace _static_thread_pool {
struct task_base {
task_base* next;
void (*execute)(task_base*) noexcept;
};

template <typename Receiver>
struct _op {
class type;
};
struct task_base {
task_base* next;
void (*execute)(task_base*) noexcept;
};

template <typename Receiver>
struct _op {
class type;
};
template <typename Receiver>
using operation = typename _op<remove_cvref_t<Receiver>>::type;

class context {
template <typename Receiver>
using operation = typename _op<remove_cvref_t<Receiver>>::type;
friend struct _op;

public:
context();
context(std::uint32_t threadCount);
~context();

class context {
class scheduler {
template <typename Receiver>
friend struct _op;
public:
context();
context(std::uint32_t threadCount);
~context();

class scheduler {
template <typename Receiver>
friend struct _op;
class schedule_sender {
public:
template <
template <typename...> class Variant,
template <typename...> class Tuple>
using value_types = Variant<Tuple<>>;

template <template <typename...> class Variant>
using error_types = Variant<>;

static constexpr bool sends_done = true;

static constexpr blocking_kind blocking = blocking_kind::never;

static constexpr bool is_always_scheduler_affine = false;

private:
template <typename Receiver>
operation<Receiver> make_operation_(Receiver&& r) const {
return operation<Receiver>{pool_, (Receiver &&) r};
}
class schedule_sender {
public:
template <
template <typename...>
class Variant,
template <typename...>
class Tuple>
using value_types = Variant<Tuple<>>;

template(typename Receiver)
(requires receiver_of<Receiver>)
friend operation<Receiver>
tag_invoke(tag_t<connect>, schedule_sender s, Receiver&& r) {
return s.make_operation_((Receiver &&) r);
}
template <template <typename...> class Variant>
using error_types = Variant<>;

friend class context::scheduler;
static constexpr bool sends_done = true;

explicit schedule_sender(context& pool) noexcept
: pool_(pool) {}
static constexpr blocking_kind blocking = blocking_kind::never;

context& pool_;
};
static constexpr bool is_always_scheduler_affine = false;

schedule_sender make_sender_() const {
return schedule_sender{pool_};
private:
template <typename Receiver>
operation<Receiver> make_operation_(Receiver&& r) const {
return operation<Receiver>{pool_, (Receiver &&) r};
}

friend schedule_sender
tag_invoke(tag_t<schedule>, const scheduler& s) noexcept {
return s.make_sender_();
template(typename Receiver) //
(requires receiver_of<Receiver>) //
friend operation<Receiver> tag_invoke(
tag_t<connect>, schedule_sender s, Receiver&& r) {
return s.make_operation_((Receiver &&) r);
}

friend class context;
explicit scheduler(context& pool) noexcept
: pool_(pool) {}
friend class context::scheduler;

friend bool operator==(scheduler a, scheduler b) noexcept {
return &a.pool_ == &b.pool_;
}
friend bool operator!=(scheduler a, scheduler b) noexcept {
return &a.pool_ != &b.pool_;
}
explicit schedule_sender(context& pool) noexcept : pool_(pool) {}

context& pool_;
};

scheduler get_scheduler() noexcept { return scheduler{*this}; }
schedule_sender make_sender_() const { return schedule_sender{pool_}; }

void request_stop() noexcept;
friend schedule_sender
tag_invoke(tag_t<schedule>, const scheduler& s) noexcept {
return s.make_sender_();
}

private:
class thread_state {
public:
task_base* try_pop();
task_base* pop();
bool try_push(task_base* task);
void push(task_base* task);
void request_stop();
friend class context;
explicit scheduler(context& pool) noexcept : pool_(pool) {}

private:
std::mutex mut_;
std::condition_variable cv_;
intrusive_queue<task_base, &task_base::next> queue_;
bool stopRequested_ = false;
};
friend bool operator==(scheduler a, scheduler b) noexcept {
return &a.pool_ == &b.pool_;
}
friend bool operator!=(scheduler a, scheduler b) noexcept {
return &a.pool_ != &b.pool_;
}

void run(std::uint32_t index) noexcept;
void join() noexcept;
context& pool_;
};

void enqueue(task_base* task) noexcept;
scheduler get_scheduler() noexcept { return scheduler{*this}; }

std::uint32_t threadCount_;
std::vector<std::thread> threads_;
std::vector<thread_state> threadStates_;
std::atomic<std::uint32_t> nextThread_;
};
void request_stop() noexcept;

template <typename Receiver>
class _op<Receiver>::type : task_base {
friend context::scheduler::schedule_sender;
private:
class thread_state {
public:
task_base* try_pop();
task_base* pop();
bool try_push(task_base* task);
void push(task_base* task);
void request_stop();

context& pool_;
Receiver receiver_;

explicit type(context& pool, Receiver&& r)
: pool_(pool)
, receiver_((Receiver &&) r) {
this->execute = [](task_base* t) noexcept {
auto& op = *static_cast<type*>(t);
if constexpr (!is_stop_never_possible_v<
stop_token_type_t<Receiver>>) {
if (get_stop_token(op.receiver_).stop_requested()) {
unifex::set_done((Receiver &&) op.receiver_);
return;
}
private:
std::mutex mut_;
std::condition_variable cv_;
intrusive_queue<task_base, &task_base::next> queue_;
bool stopRequested_ = false;
};

void run(std::uint32_t index) noexcept;
void join() noexcept;

void enqueue(task_base* task) noexcept;

std::uint32_t threadCount_;
std::vector<std::thread> threads_;
std::vector<thread_state> threadStates_;
std::atomic<std::uint32_t> nextThread_;
};

template <typename Receiver>
class _op<Receiver>::type : task_base {
friend context::scheduler::schedule_sender;

context& pool_;
Receiver receiver_;

explicit type(context& pool, Receiver&& r)
: pool_(pool)
, receiver_((Receiver &&) r) {
this->execute = [](task_base* t) noexcept {
auto& op = *static_cast<type*>(t);
if constexpr (!is_stop_never_possible_v<stop_token_type_t<Receiver>>) {
if (get_stop_token(op.receiver_).stop_requested()) {
unifex::set_done((Receiver &&) op.receiver_);
return;
}
unifex::set_value((Receiver &&) op.receiver_);
};
}
}
unifex::set_value((Receiver &&) op.receiver_);
};
}

void enqueue_(task_base* op) const {
pool_.enqueue(op);
}
void enqueue_(task_base* op) const { pool_.enqueue(op); }

friend void tag_invoke(tag_t<start>, type& op) noexcept {
op.enqueue_(&op);
}
};
friend void tag_invoke(tag_t<start>, type& op) noexcept { op.enqueue_(&op); }
};

} // _static_thread_pool
} // namespace _static_thread_pool

using static_thread_pool = _static_thread_pool::context;

} // namespace unifex
} // namespace unifex

#include <unifex/detail/epilogue.hpp>

0 comments on commit 0c2372b

Please sign in to comment.