Commit 4ab7c4af authored by Joseph Noir's avatar Joseph Noir

Update to make libcaf_core compile for riot, again

parent c9cc03be
......@@ -364,6 +364,7 @@ endif()
if(CAF_FOR_RIOT)
include_directories("${RIOT_BASE_DIR}/sys/include/.")
include_directories("${RIOT_BASE_DIR}/sys/net/include/.")
include_directories("${RIOT_BASE_DIR}/sys/cpp11-compat/include/.")
include_directories("${RIOT_BASE_DIR}/core/include/.")
include_directories("${RIOT_BASE_DIR}/core/include/arch/.")
include_directories("${RIOT_BASE_DIR}/cpu/native/include/.")
......
......@@ -84,15 +84,6 @@ set (LIBCAF_CORE_SRCS
src/uniform_type_info.cpp
src/uniform_type_info_map.cpp)
if (CAF_FOR_RIOT)
set (LIBCAF_CORE_SRCS
src/chrono.cpp
src/condition_variable.cpp
src/mutex.cpp
src/thread.cpp
${LIBCAF_CORE_SRCS})
endif()
add_custom_target(libcaf_core)
# build shared library if not compiling static only
......
......@@ -20,9 +20,6 @@
#ifndef CAF_BLOCKING_ACTOR_HPP
#define CAF_BLOCKING_ACTOR_HPP
#include <mutex>
#include <condition_variable>
#include "caf/none.hpp"
#include "caf/on.hpp"
......
......@@ -21,94 +21,45 @@
#ifndef CAF_CHRONO_HPP
#define CAF_CHRONO_HPP
#include <chrono>
#include <cstdio>
#include <algorithm>
#ifdef __RIOTBUILD_FLAG
#include "time.h"
#include "vtimer.h"
#include "riot/chrono.hpp"
namespace caf {
class duration;
namespace {
constexpr uint32_t microsec_in_secs = 1000000;
} // namespace anaonymous
class time_point {
using native_handle_type = timex_t;
public:
inline time_point() : m_handle{0,0} { }
inline time_point(uint32_t seconds, uint32_t microseconds)
: m_handle{seconds, microseconds} { adjust_overhead(); }
inline time_point(timex_t&& tp) : m_handle(tp) { }
constexpr time_point(const time_point& tp) = default;
constexpr time_point(time_point&& tp) = default;
inline native_handle_type native_handle() const { return m_handle; }
time_point& operator+=(const caf::duration& d);
template<class Rep, class Period>
inline time_point& operator+=(const std::chrono::duration<Rep,Period>& d) {
auto s = std::chrono::duration_cast<std::chrono::seconds>(d);
auto m = (std::chrono::duration_cast<std::chrono::microseconds>(d) - s);
m_handle.seconds += s.count();
m_handle.microseconds += m.count();
adjust_overhead();
return *this;
}
template<class Rep, class Period>
inline time_point operator+(const std::chrono::duration<Rep,Period>& d) {
time_point result{seconds(), microseconds()};
result += d;
return result;
}
inline uint32_t seconds() const { return m_handle.seconds; }
inline uint32_t microseconds() const { return m_handle.microseconds; }
private:
timex_t m_handle;
void inline adjust_overhead() {
while (m_handle.microseconds >= microsec_in_secs) {
m_handle.seconds += 1;
m_handle.microseconds -= microsec_in_secs;
}
}
};
inline time_point now() {
timex_t tp;
vtimer_now(&tp);
return time_point(std::move(tp));
}
inline bool operator<(const time_point& lhs, const time_point& rhs) {
return lhs.seconds() < rhs.seconds() ||
(lhs.seconds() == rhs.seconds() &&
lhs.microseconds() < rhs.microseconds());
}
inline bool operator>(const time_point& lhs, const time_point& rhs) {
return rhs < lhs;
using time_point = riot::time_point;
inline auto now() -> decltype(time_point()) {
return riot::now();
}
inline bool operator<=(const time_point& lhs, const time_point& rhs) {
return !(rhs < lhs);
}
inline bool operator>=(const time_point& lhs, const time_point& rhs) {
return !(lhs < rhs);
template <class Duration>
inline time_point& operator+=(time_point& tp, const Duration& d) {
switch (static_cast<uint32_t>(d.unit)) {
case 1:
tp += std::chrono::seconds(d.count);
break;
case 1000:
tp += std::chrono::microseconds(d.count);
break;
case 1000000:
tp += std::chrono::microseconds(1000 * d.count);
break;
default:
break;
}
return tp;
}
} // namespace caf
#else
#include <chrono>
namespace caf {
using time_point = std::chrono::high_resolution_clock::time_point;
......@@ -116,8 +67,6 @@ inline auto now() -> decltype(time_point()) {
return std::chrono::high_resolution_clock::now();
}
// todo mapping for caf
} // namespace caf
#endif
......
......@@ -24,85 +24,13 @@
#ifdef __RIOTBUILD_FLAG
#include "sched.h"
#include "vtimer.h"
#include "caf/mutex.hpp"
#include "caf/chrono.hpp"
#include "riot/condition_variable.hpp"
namespace caf {
enum class cv_status { no_timeout, timeout };
class condition_variable {
public:
using native_handle_type = priority_queue_t*;
inline condition_variable() {
m_queue.first = NULL;
}
~condition_variable();
void notify_one() noexcept;
void notify_all() noexcept;
void wait(unique_lock<mutex>& lock) noexcept;
template <class Predicate>
void wait(unique_lock<mutex>& lock, Predicate pred);
cv_status wait_until(unique_lock<mutex>& lock,
const time_point& timeout_time);
template <class Predicate>
bool wait_until(unique_lock<mutex>& lock, const time_point& timeout_time,
Predicate pred);
template <class Rep, class Period>
cv_status wait_for(unique_lock<mutex>& lock,
const std::chrono::duration<Rep, Period>& rel_time);
template <class Rep, class Period, class Predicate>
bool wait_for(unique_lock<mutex>& lock,
const std::chrono::duration<Rep, Period>& rel_time,
Predicate pred);
inline native_handle_type native_handle() { return &m_queue; }
private:
condition_variable(const condition_variable&);
condition_variable& operator=(const condition_variable&);
priority_queue_t m_queue;
};
template <class Predicate>
void condition_variable::wait(unique_lock<mutex>& lock, Predicate pred) {
while(!pred()) {
wait(lock);
}
}
template <class Predicate>
bool condition_variable::wait_until(unique_lock<mutex>& lock,
const time_point& timeout_time,
Predicate pred) {
while (!pred()) {
if (wait_until(lock, timeout_time) == cv_status::timeout) {
return pred();
}
}
return true;
}
template <class Rep, class Period>
cv_status condition_variable::wait_for(unique_lock<mutex>& lock,
const std::chrono::duration<Rep, Period>& timeout_duration) {
return wait_until(lock, now() + timeout_duration);
}
template <class Rep, class Period, class Predicate>
inline bool condition_variable::wait_for(unique_lock<mutex>& lock,
const std::chrono::duration<Rep, Period>& timeout_duration,
Predicate pred) {
return wait_until(lock, now() + timeout_duration, std::move(pred));
}
using cv_status = riot::cv_status;
using condition_variable = riot::condition_variable;
} // namespace caf
......
......@@ -22,10 +22,10 @@
#include "caf/send.hpp"
#include "caf/actor.hpp"
#include "caf/thread.hpp"
#include "caf/message.hpp"
#include "caf/string_algorithms.hpp"
#include <thread>
#include <vector>
#include <string>
......@@ -33,10 +33,10 @@ namespace caf {
namespace detail {
std::thread run_program_impl(caf::actor, const char*, std::vector<std::string>);
thread run_program_impl(caf::actor, const char*, std::vector<std::string>);
template <class... Ts>
std::thread run_program(caf::actor listener, const char* path, Ts&&... args) {
thread run_program(caf::actor listener, const char* path, Ts&&... args) {
std::vector<std::string> vec{convert_to_str(std::forward<Ts>(args))...};
return run_program_impl(listener, path, std::move(vec));
}
......
......@@ -23,165 +23,13 @@
#ifdef __RIOTBUILD_FLAG
#include "mutex.h"
#include <utility>
#include <stdexcept>
#include "riot/mutex.hpp"
namespace caf {
class mutex {
public:
using native_handle_type = mutex_t*;
inline constexpr mutex() noexcept : m_mtx{0, PRIORITY_QUEUE_INIT} { }
~mutex();
void lock();
bool try_lock() noexcept;
void unlock() noexcept;
inline native_handle_type native_handle() { return &m_mtx; }
private:
mutex(const mutex&);
mutex& operator=(const mutex&);
mutex_t m_mtx;
};
struct defer_lock_t {};
struct try_to_lock_t {};
struct adopt_lock_t {};
constexpr defer_lock_t defer_lock = defer_lock_t();
constexpr try_to_lock_t try_to_lock = try_to_lock_t();
constexpr adopt_lock_t adopt_lock = adopt_lock_t();
template<class Mutex>
class lock_guard {
public:
using mutex_type = Mutex;
inline explicit lock_guard(mutex_type& mtx) : m_mtx(mtx) { m_mtx.lock(); }
inline lock_guard(mutex_type& mtx, adopt_lock_t) : m_mtx{mtx} {}
inline ~lock_guard() { m_mtx.unlock(); }
private:
mutex_type& m_mtx;
};
template<class Mutex>
class unique_lock {
public:
using mutex_type = Mutex;
inline unique_lock() noexcept : m_mtx{nullptr}, m_owns{false} {}
inline explicit unique_lock(mutex_type& mtx) : m_mtx{&mtx}, m_owns{true} {
m_mtx->lock();
}
inline unique_lock(mutex_type& mtx, defer_lock_t) noexcept
: m_mtx{&mtx}, m_owns{false} { }
inline unique_lock(mutex_type& mtx, try_to_lock_t)
: m_mtx{&mtx}, m_owns{ mtx.try_lock() } { }
inline unique_lock(mutex_type& mtx, adopt_lock_t)
: m_mtx{&mtx}, m_owns{true} { }
inline ~unique_lock() {
if (m_owns) {
m_mtx->unlock();
}
}
inline unique_lock(unique_lock&& lock) noexcept
: m_mtx{lock.m_mtx},
m_owns{lock.m_owns} {
lock.m_mtx = nullptr;
lock.m_owns = false;
}
inline unique_lock& operator=(unique_lock&& lock) noexcept {
if (m_owns) {
m_mtx->unlock();
}
m_mtx = lock.m_mtx;
m_owns = lock.m_owns;
lock.m_mtx = nullptr;
lock.m_owns = false;
return *this;
}
void lock();
bool try_lock();
// template <class Rep, class Period>
// bool try_lock_for(const chrono::duration<Rep, Period>& timeout_duration);
// template <class Clock, class Duration>
// bool try_lock_until(const chrono::time_point<Clock,Duration>& timeout_time);
void unlock();
inline void swap(unique_lock& lock) noexcept {
std::swap(m_mtx, lock.m_mtx);
std::swap(m_owns, lock.m_owns);
}
inline mutex_type* release() noexcept {
mutex_type* mtx = m_mtx;
m_mtx = nullptr;
m_owns = false;
return mtx;
}
inline bool owns_lock() const noexcept { return m_owns; }
inline explicit operator bool () const noexcept { return m_owns; }
inline mutex_type* mutex() const noexcept { return m_mtx; }
private:
unique_lock(unique_lock const&);
unique_lock& operator=(unique_lock const&);
mutex_type* m_mtx;
bool m_owns;
};
template<class Mutex>
void unique_lock<Mutex>::lock() {
if (m_mtx == nullptr) {
throw std::runtime_error("unique_lock::lock: references null mutex");
}
if (m_owns) {
throw std::runtime_error("");
}
m_mtx->lock();
m_owns = true;
}
template<class Mutex>
bool unique_lock<Mutex>::try_lock() {
if (m_mtx == nullptr) {
throw std::runtime_error("unique_lock::try_lock: references null mutex");
}
if (m_owns) {
throw std::runtime_error("unique_lock::try_lock lock: already locked");
}
m_owns = m_mtx->try_lock();
return m_owns;
}
template<class Mutex>
void unique_lock<Mutex>::unlock() {
if (!m_owns) {
throw std::runtime_error("unique_lock::unlock: not locked");
}
m_mtx->unlock();
m_owns = false;
}
template<class Mutex>
inline void swap(unique_lock<Mutex>& lhs, unique_lock<Mutex>& rhs) noexcept {
lhs.swap(rhs);
}
using mutex = riot::mutex;
template <class Mutex> using lock_guard = riot::lock_guard<Mutex>;
template <class Mutex> using unique_lock = riot::unique_lock<Mutex>;
} // namespace caf
......
......@@ -24,6 +24,7 @@
#include <memory>
#include "caf/thread.hpp"
#include "caf/resumable.hpp"
#include "caf/condition_variable.hpp"
#include "caf/scheduler/worker.hpp"
......@@ -88,7 +89,7 @@ class coordinator : public abstract_coordinator {
resumable::resume_result resume(execution_unit* ptr, size_t) override {
CAF_LOG_DEBUG("shutdown_helper::resume => shutdown worker");
CAF_ASSERT(ptr != nullptr);
nique_lock<mutex> guard(mtx);
unique_lock<mutex> guard(mtx);
last_worker = ptr;
cv.notify_all();
return resumable::shutdown_execution_unit;
......
......@@ -29,13 +29,15 @@
#endif // CAF_MACOS
#include <cmath>
#include <mutex>
#include <chrono>
#include <vector>
#include <fstream>
#include <iomanip>
#include <unordered_map>
#include "caf/mutex.hpp"
#include "caf/chrono.hpp"
#include "caf/scheduler/coordinator.hpp"
#include "caf/policy/profiled.hpp"
#include "caf/policy/work_stealing.hpp"
......@@ -209,20 +211,20 @@ class profiled_coordinator : public coordinator<Policy> {
if (m.time - w.last_flush >= m_resolution) {
w.last_flush = m.time;
auto wallclock = m_system_start + (m.time - m_clock_start);
std::lock_guard<std::mutex> file_guard{m_file_mtx};
lock_guard<mutex> file_guard{m_file_mtx};
record(wallclock, "worker", worker, w.worker);
w.worker = {};
}
}
void remove_job(actor_id job) {
std::lock_guard<std::mutex> job_guard{m_job_mtx};
lock_guard<mutex> job_guard{m_job_mtx};
auto j = m_jobs.find(job);
if (j != m_jobs.end()) {
if (job != 0) {
auto now = clock_type::now().time_since_epoch();
auto wallclock = m_system_start + (now - m_clock_start);
std::lock_guard<std::mutex> file_guard{m_file_mtx};
lock_guard<mutex> file_guard{m_file_mtx};
record(wallclock, "actor", job, j->second);
}
m_jobs.erase(j);
......@@ -240,13 +242,13 @@ class profiled_coordinator : public coordinator<Policy> {
}
void report(actor_id const& job, measurement const& m) {
std::lock_guard<std::mutex> job_guard{m_job_mtx};
lock_guard<mutex> job_guard{m_job_mtx};
m_jobs[job] += m;
if (m.time - m_last_flush >= m_resolution) {
m_last_flush = m.time;
auto now = clock_type::now().time_since_epoch();
auto wallclock = m_system_start + (now - m_clock_start);
std::lock_guard<std::mutex> file_guard{m_file_mtx};
lock_guard<mutex> file_guard{m_file_mtx};
for (auto& j : m_jobs) {
record(wallclock, "actor", j.first, j.second);
j.second = {};
......@@ -254,8 +256,8 @@ class profiled_coordinator : public coordinator<Policy> {
}
}
std::mutex m_job_mtx;
std::mutex m_file_mtx;
mutex m_job_mtx;
mutex m_file_mtx;
std::ofstream m_file;
msec m_resolution;
std::chrono::system_clock::time_point m_system_start;
......
......@@ -23,209 +23,19 @@
#ifdef __RIOTBUILD_FLAG
#include "time.h"
#include "thread.h"
#include "kernel_internal.h"
#include <tuple>
#include <memory>
#include <utility>
#include <exception>
#include <stdexcept>
#include <functional>
#include <type_traits>
#include "caf/mutex.hpp"
#include "caf/chrono.hpp"
#include "caf/ref_counted.hpp"
#include "caf/thread_util.hpp"
#include "caf/intrusive_ptr.hpp"
#include "caf/condition_variable.hpp"
#include "caf/detail/int_list.hpp"
#include "caf/detail/apply_args.hpp"
#include "riot/thread.hpp"
namespace caf {
namespace {
constexpr kernel_pid_t thread_uninitialized = -1;
constexpr size_t stack_size = KERNEL_CONF_STACKSIZE_MAIN;
}
struct thread_data : ref_counted {
thread_data() : joining_thread{thread_uninitialized} { }
kernel_pid_t joining_thread;
char stack[stack_size];
};
class thread_id {
template <class T,class Traits>
friend std::basic_ostream<T,Traits>&
operator<<(std::basic_ostream<T,Traits>& out, thread_id id);
friend class thread;
public:
inline thread_id() noexcept : m_handle{thread_uninitialized} { }
inline thread_id(kernel_pid_t handle) : m_handle{handle} {}
inline bool operator==(thread_id other) noexcept {
return m_handle == other.m_handle;
}
inline bool operator!=(thread_id other) noexcept {
return !(m_handle == other.m_handle);
}
inline bool operator< (thread_id other) noexcept {
return m_handle < other.m_handle;
}
inline bool operator<=(thread_id other) noexcept {
return !(m_handle > other.m_handle);
}
inline bool operator> (thread_id other) noexcept {
return m_handle > other.m_handle;
}
inline bool operator>=(thread_id other) noexcept {
return !(m_handle < other.m_handle);
}
private:
kernel_pid_t m_handle;
};
template <class T,class Traits>
inline std::basic_ostream<T,Traits>&
operator<<(std::basic_ostream<T,Traits>& out, thread_id id) {
return out << id.m_handle;
}
using thread = riot::thread;
namespace this_thread {
inline thread_id get_id() noexcept { return thread_getpid(); }
inline void yield() noexcept { thread_yield(); }
void sleep_for(const std::chrono::nanoseconds& ns);
template <class Rep, class Period>
void sleep_for(const std::chrono::duration<Rep, Period>& sleep_duration) {
using namespace std::chrono;
if (sleep_duration > std::chrono::duration<Rep, Period>::zero()) {
constexpr std::chrono::duration<long double> max = nanoseconds::max();
nanoseconds ns;
if (sleep_duration < max) {
ns = duration_cast<nanoseconds>(sleep_duration);
if (ns < sleep_duration) {
++ns;
}
}
else {
ns = nanoseconds::max();
}
sleep_for(ns);
}
}
void sleep_until(const time_point& sleep_time);
} // namespace this_thread
class thread {
public:
using id = thread_id;
using native_handle_type = kernel_pid_t;
inline thread() noexcept : m_handle{thread_uninitialized} { }
template <class F, class ...Args,
class = typename std::enable_if<
!std::is_same<typename std::decay<F>::type,thread>::value
>::type
>
explicit thread(F&& f, Args&&... args);
~thread();
using riot::this_thread::get_id;
using riot::this_thread::yield;
using riot::this_thread::sleep_for;
thread(const thread&) = delete;
inline thread(thread&& t) noexcept
: m_handle{t.m_handle},
m_data{t.m_data} {
t.m_handle = thread_uninitialized;
t.m_data.reset();
}
thread& operator=(const thread&) = delete;
thread& operator=(thread&&) noexcept;
void swap(thread& t) noexcept {
std::swap(m_data, t.m_data);
std::swap(m_handle, t.m_handle);
}
inline bool joinable() const noexcept {
return m_handle != thread_uninitialized;
}
void join();
void detach();
inline id get_id() const noexcept { return m_handle; }
inline native_handle_type native_handle() noexcept { return m_handle; }
static unsigned hardware_concurrency() noexcept;
kernel_pid_t m_handle;
intrusive_ptr<thread_data> m_data;
};
void swap(thread& lhs, thread& rhs) noexcept;
template <class F>
void* thread_proxy(void* vp) {
std::unique_ptr<F> p(static_cast<F*>(vp));
intrusive_ptr<thread_data> data{std::get<0>(*p)};
data->deref(); // remove count incremented in constructor
auto indices =
detail::get_right_indices<caf::detail::tl_size<F>::value - 2>(*p);
try {
detail::apply_args(std::get<1>(*p), indices, *p);
} catch (...) {
// nop
}
if (data->joining_thread != thread_uninitialized) {
thread_wakeup(data->joining_thread);
}
// some riot cleanup code
sched_task_exit();
return nullptr;
}
template <class F, class ...Args,
class
>
thread::thread(F&& f, Args&&... args) : m_data{new thread_data()} {
using namespace std;
using func_and_args = tuple<typename decay<thread_data*>::type,
typename decay<F>::type,
typename decay<Args>::type...>;
std::unique_ptr<func_and_args> p(new func_and_args(
decay_copy(forward<thread_data*>(m_data.get())),
decay_copy(forward<F>(f)),
decay_copy(forward<Args>(args))...));
m_data->ref(); // maintain count in case obj is destroyed before thread runs
m_handle = thread_create(m_data->stack, stack_size,
PRIORITY_MAIN - 1, 0, // CREATE_WOUT_YIELD
&thread_proxy<func_and_args>,
p.get(), "caf_thread");
if (m_handle >= 0) {
p.release();
} else {
throw std::runtime_error("Failed to create thread.");
}
}
inline thread& thread::operator=(thread&& other) noexcept {
if (m_handle != thread_uninitialized) {
std::terminate();
}
m_handle = other.m_handle;
other.m_handle = thread_uninitialized;
std::swap(m_data, other.m_data);
return *this;
}
inline void swap (thread& lhs, thread& rhs) noexcept {
lhs.swap(rhs);
}
} // namespace this_thread
} // namespace caf
......
#ifndef CAF_THREAD_UTILS_HPP
#define CAF_THREAD_UTILS_HPP
#include <utility>
// decay copy (from clang type_traits)
template <class T>
inline typename std::decay<T>::type decay_copy(T&& t) {
return std::forward<T>(t);
}
#endif // CAF_THREAD_UTILS_HPP
......@@ -49,8 +49,6 @@ namespace scheduler {
namespace {
using hrc = std::chrono::high_resolution_clock;
struct delayed_msg {
actor_addr from;
channel to;
......@@ -64,7 +62,7 @@ inline void deliver(delayed_msg& dm) {
template <class Map, class... Ts>
inline void insert_dmsg(Map& storage, const duration& d, Ts&&... xs) {
auto tout = hrc::now();
auto tout = now();
tout += d;
delayed_msg dmsg{std::forward<Ts>(xs)...};
storage.insert(std::make_pair(std::move(tout), std::move(dmsg)));
......@@ -77,14 +75,14 @@ class timer_actor : public blocking_actor {
return next_message();
}
bool await_data(const hrc::time_point& tp) {
bool await_data(const time_point& tp) {
if (has_next_message()) {
return true;
}
return mailbox().synchronized_await(m_mtx, m_cv, tp);
}
mailbox_element_ptr try_dequeue(const hrc::time_point& tp) {
mailbox_element_ptr try_dequeue(const time_point& tp) {
if (await_data(tp)) {
return next_message();
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* Raphael Hiesgen <raphael.hiesgen (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENCE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include <cstdio>
#include <stdexcept>
#include "irq.h"
#include "irq_arch.h"
#include "sched.h"
#include "vtimer.h"
#include "priority_queue.h"
#include "caf/condition_variable.hpp"
using namespace std::chrono;
namespace caf {
condition_variable::~condition_variable() {
m_queue.first = NULL;
}
void condition_variable::notify_one() noexcept {
unsigned old_state = disableIRQ();
priority_queue_node_t *head = priority_queue_remove_head(&m_queue);
int other_prio = -1;
if (head != NULL) {
tcb_t *other_thread = (tcb_t *) sched_threads[head->data];
if (other_thread) {
other_prio = other_thread->priority;
sched_set_status(other_thread, STATUS_PENDING);
}
head->data = -1u;
}
restoreIRQ(old_state);
if (other_prio >= 0) {
sched_switch(other_prio);
}
}
void condition_variable::notify_all() noexcept {
unsigned old_state = disableIRQ();
int other_prio = -1;
while (true) {
priority_queue_node_t *head = priority_queue_remove_head(&m_queue);
if (head == NULL) {
break;
}
tcb_t *other_thread = (tcb_t *) sched_threads[head->data];
if (other_thread) {
auto max_prio = [](int a, int b) {
return (a < 0) ? b : ((a < b) ? a : b);
};
other_prio = max_prio(other_prio, other_thread->priority);
sched_set_status(other_thread, STATUS_PENDING);
}
head->data = -1u;
}
restoreIRQ(old_state);
if (other_prio >= 0) {
sched_switch(other_prio);
}
}
void condition_variable::wait(unique_lock<mutex>& lock) noexcept {
if (!lock.owns_lock()) {
throw std::runtime_error("condition_variable::wait: mutex not locked");
}
priority_queue_node_t n;
n.priority = sched_active_thread->priority;
n.data = sched_active_pid;
n.next = NULL;
// the signaling thread may not hold the mutex, the queue is not thread safe
unsigned old_state = disableIRQ();
priority_queue_add(&m_queue, &n);
restoreIRQ(old_state);
mutex_unlock_and_sleep(lock.mutex()->native_handle());
if (n.data != -1u) {
// on signaling n.data is set to -1u
// if it isn't set, then the wakeup is either spurious or a timer wakeup
old_state = disableIRQ();
priority_queue_remove(&m_queue, &n);
restoreIRQ(old_state);
}
mutex_lock(lock.mutex()->native_handle());
}
cv_status condition_variable::wait_until(unique_lock<mutex>& lock,
const time_point& timeout_time) {
vtimer_t timer;
vtimer_set_wakeup_timepoint(&timer, timeout_time.native_handle(),
sched_active_pid);
wait(lock);
timex_t after;
vtimer_now(&after);
vtimer_remove(&timer);
auto cmp = timex_cmp(after,timeout_time.native_handle());
return cmp < 1 ? cv_status::no_timeout : cv_status::timeout;
}
} // namespace caf
......@@ -19,9 +19,10 @@
#include "caf/scheduler/detached_threads.hpp"
#include <mutex>
#include <atomic>
#include <condition_variable>
#include "caf/mutex.hpp"
#include "caf/condition_variable.hpp"
namespace caf {
namespace scheduler {
......@@ -29,8 +30,8 @@ namespace scheduler {
namespace {
std::atomic<size_t> s_detached;
std::mutex s_detached_mtx;
std::condition_variable s_detached_cv;
mutex s_detached_mtx;
condition_variable s_detached_cv;
} // namespace <anonymous>
......@@ -42,13 +43,13 @@ void inc_detached_threads() {
void dec_detached_threads() {
size_t new_val = --s_detached;
if (new_val == 0) {
std::unique_lock<std::mutex> guard(s_detached_mtx);
unique_lock<mutex> guard(s_detached_mtx);
s_detached_cv.notify_all();
}
}
void await_detached_threads() {
std::unique_lock<std::mutex> guard{s_detached_mtx};
unique_lock<mutex> guard{s_detached_mtx};
while (s_detached != 0) {
s_detached_cv.wait(guard);
}
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* Raphael Hiesgen <raphael.hiesgen (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENCE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "caf/mutex.hpp"
namespace caf {
mutex::~mutex() {
// nop
}
void mutex::lock() {
mutex_lock(&m_mtx);
}
bool mutex::try_lock() noexcept {
return (1 == mutex_trylock(&m_mtx));
}
void mutex::unlock() noexcept {
mutex_unlock(&m_mtx);
}
} // namespace caf
......@@ -25,7 +25,7 @@ using namespace std;
using namespace caf;
using namespace caf::detail;
std::thread caf::detail::run_program_impl(actor rc, const char* cpath,
thread caf::detail::run_program_impl(actor rc, const char* cpath,
vector<string> args) {
string path = cpath;
replace_all(path, "'", "\\'");
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (at) haw-hamburg.de> *
* Raphael Hiesgen <raphael.hiesgen (at) haw-hamburg.de> *
* *
* Distributed under the terms and conditions of the BSD 3-Clause License or *
* (at your option) under the terms and conditions of the Boost Software *
* License 1.0. See accompanying files LICENSE and LICENCE_ALTERNATIVE. *
* *
* If you did not receive a copy of the license files, see *
* http://opensource.org/licenses/BSD-3-Clause and *
* http://www.boost.org/LICENSE_1_0.txt. *
******************************************************************************/
#include "vtimer.h"
#include <cerrno>
#include <system_error>
#include "caf/thread.hpp"
using namespace std;
namespace caf {
thread::~thread() {
if (joinable()) {
terminate();
}
}
void thread::join() {
if (this->get_id() == this_thread::get_id()) {
throw system_error(make_error_code(errc::resource_deadlock_would_occur),
"Joining this leads to a deadlock.");
}
if (joinable()) {
auto status = thread_getstatus(m_handle);
if (status != STATUS_NOT_FOUND && status != STATUS_STOPPED) {
m_data->joining_thread = sched_active_pid;
thread_sleep();
}
m_handle = thread_uninitialized;
} else {
throw system_error(make_error_code(errc::invalid_argument),
"Can not join an unjoinable thread.");
}
// missing: no_such_process system error
}
void thread::detach() {
if (joinable()) {
m_handle = thread_uninitialized;
} else {
throw system_error(make_error_code(errc::invalid_argument),
"Can not detach an unjoinable thread.");
}
}
unsigned thread::hardware_concurrency() noexcept {
return 1;
}
namespace this_thread {
void sleep_for(const chrono::nanoseconds& ns) {
using namespace chrono;
if (ns > nanoseconds::zero()) {
seconds s = duration_cast<seconds>(ns);
microseconds ms = duration_cast<microseconds>(ns);
timex_t reltime;
constexpr uint32_t uint32_max = numeric_limits<uint32_t>::max();
if (s.count() < uint32_max) {
reltime.seconds = static_cast<uint32_t>(s.count());
reltime.microseconds =
static_cast<uint32_t>(duration_cast<microseconds>((ms-s)).count());
} else {
reltime.seconds = uint32_max;
reltime.microseconds = uint32_max;
}
vtimer_t timer;
vtimer_set_wakeup(&timer, reltime, sched_active_pid);
thread_sleep();
vtimer_remove(&timer);
}
}
void sleep_until(const time_point& sleep_time) {
using namespace std::chrono;
mutex mtx;
condition_variable cv;
unique_lock<mutex> lk(mtx);
while(now() < sleep_time) {
cv.wait_until(lk,sleep_time);
}
}
} // namespace this_thread
} // namespace caf
......@@ -20,8 +20,8 @@
#define CAF_SUITE sync_timeout
#include "caf/test/unit_test.hpp"
#include <thread>
#include <chrono>
#include "caf/chrono.hpp"
#include "caf/thread.hpp"
#include "caf/all.hpp"
......@@ -36,7 +36,7 @@ using send_ping_atom = atom_constant<atom("send_ping")>;
behavior pong() {
return {
[=] (ping_atom) {
std::this_thread::sleep_for(std::chrono::seconds(1));
this_thread::sleep_for(chrono::seconds(1));
return pong_atom::value;
}
};
......@@ -52,7 +52,7 @@ behavior ping1(event_based_actor* self, const actor& pong_actor) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
},
after(std::chrono::milliseconds(100)) >> [=] {
after(chrono::milliseconds(100)) >> [=] {
CAF_MESSAGE("sync timeout: check");
self->quit(exit_reason::user_shutdown);
}
......@@ -72,13 +72,13 @@ behavior ping2(event_based_actor* self, const actor& pong_actor) {
CAF_TEST_ERROR("received pong atom");
self->quit(exit_reason::user_shutdown);
},
after(std::chrono::milliseconds(100)) >> [=] {
after(chrono::milliseconds(100)) >> [=] {
CAF_MESSAGE("inner timeout: check");
*received_inner = true;
}
);
},
after(std::chrono::milliseconds(100)) >> [=] {
after(chrono::milliseconds(100)) >> [=] {
CAF_CHECK_EQUAL(*received_inner, true);
self->quit(exit_reason::user_shutdown);
}
......
......@@ -20,7 +20,6 @@
#define CAF_SUITE remote_actor
#include "caf/test/unit_test.hpp"
#include <thread>
#include <string>
#include <cstring>
#include <sstream>
......@@ -28,6 +27,8 @@
#include <functional>
#include "caf/all.hpp"
#include "caf/thread.hpp"
#include "caf/io/all.hpp"
#include "caf/detail/logging.hpp"
......
......@@ -20,7 +20,6 @@
#define CAF_SUITE typed_remote_actor
#include "caf/test/unit_test.hpp"
#include <thread>
#include <string>
#include <cstring>
#include <sstream>
......@@ -28,6 +27,7 @@
#include <functional>
#include "caf/all.hpp"
#include "caf/io/all.hpp"
#include "caf/detail/run_program.hpp"
......
......@@ -23,7 +23,6 @@
#include <map>
#include <cmath>
#include <mutex>
#include <thread>
#include <vector>
#include <string>
#include <memory>
......@@ -31,6 +30,8 @@
#include <sstream>
#include <iostream>
#include "caf/mutex.hpp"
namespace caf {
class message;
namespace test {
......@@ -153,11 +154,11 @@ class logger {
template <class T>
void log(level lvl, const T& x) {
if (lvl <= m_level_console) {
std::lock_guard<std::mutex> io_guard{m_console_mtx};
lock_guard<mutex> io_guard{m_console_mtx};
m_console << x;
}
if (lvl <= m_level_file) {
std::lock_guard<std::mutex> io_guard{m_file_mtx};
lock_guard<mutex> io_guard{m_file_mtx};
m_file << x;
}
}
......@@ -174,8 +175,8 @@ class logger {
level m_level_file;
std::ostream& m_console;
std::ofstream m_file;
std::mutex m_console_mtx;
std::mutex m_file_mtx;
mutex m_console_mtx;
mutex m_file_mtx;
};
enum color_face {
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment