Commit a4323ef4 authored by Joseph Noir's avatar Joseph Noir

Exchange header includes from STL to CAF wrappers

The wrappers allow compatibility with RIOT.
parent 8bce1e20
...@@ -21,7 +21,6 @@ ...@@ -21,7 +21,6 @@
#define CAF_ABSTRACT_ACTOR_HPP #define CAF_ABSTRACT_ACTOR_HPP
#include <set> #include <set>
#include <mutex>
#include <atomic> #include <atomic>
#include <memory> #include <memory>
#include <string> #include <string>
...@@ -29,15 +28,16 @@ ...@@ -29,15 +28,16 @@
#include <cstdint> #include <cstdint>
#include <exception> #include <exception>
#include <type_traits> #include <type_traits>
#include <condition_variable>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/mutex.hpp"
#include "caf/attachable.hpp" #include "caf/attachable.hpp"
#include "caf/message_id.hpp" #include "caf/message_id.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
#include "caf/intrusive_ptr.hpp" #include "caf/intrusive_ptr.hpp"
#include "caf/abstract_channel.hpp" #include "caf/abstract_channel.hpp"
#include "caf/condition_variable.hpp"
#include "caf/detail/type_traits.hpp" #include "caf/detail/type_traits.hpp"
#include "caf/detail/functor_attachable.hpp" #include "caf/detail/functor_attachable.hpp"
...@@ -301,12 +301,11 @@ class abstract_actor : public abstract_channel { ...@@ -301,12 +301,11 @@ class abstract_actor : public abstract_channel {
// initially set to exit_reason::not_exited // initially set to exit_reason::not_exited
std::atomic<uint32_t> m_exit_reason; std::atomic<uint32_t> m_exit_reason;
// guards access to m_exit_reason, m_attachables, m_links, // guards access to m_exit_reason, m_attachables, and m_links
// and enqueue operations if actor is thread-mapped mutable mutex m_mtx;
mutable std::mutex m_mtx;
// only used in blocking and thread-mapped actors // only used in blocking and thread-mapped actors
mutable std::condition_variable m_cv; mutable condition_variable m_cv;
// attached functors that are executed on cleanup (monitors, links, etc) // attached functors that are executed on cleanup (monitors, links, etc)
attachable_ptr m_attachables_head; attachable_ptr m_attachables_head;
......
...@@ -21,13 +21,13 @@ ...@@ -21,13 +21,13 @@
#define CAF_DETAIL_ACTOR_REGISTRY_HPP #define CAF_DETAIL_ACTOR_REGISTRY_HPP
#include <map> #include <map>
#include <mutex>
#include <thread>
#include <atomic> #include <atomic>
#include <cstdint> #include <cstdint>
#include <condition_variable>
#include "caf/mutex.hpp"
#include "caf/thread.hpp"
#include "caf/abstract_actor.hpp" #include "caf/abstract_actor.hpp"
#include "caf/condition_variable.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
#include "caf/detail/singleton_mixin.hpp" #include "caf/detail/singleton_mixin.hpp"
...@@ -83,8 +83,8 @@ class actor_registry : public singleton_mixin<actor_registry> { ...@@ -83,8 +83,8 @@ class actor_registry : public singleton_mixin<actor_registry> {
std::atomic<size_t> m_running; std::atomic<size_t> m_running;
std::atomic<actor_id> m_ids; std::atomic<actor_id> m_ids;
std::mutex m_running_mtx; mutex m_running_mtx;
std::condition_variable m_running_cv; condition_variable m_running_cv;
mutable detail::shared_spinlock m_instances_mtx; mutable detail::shared_spinlock m_instances_mtx;
entries m_entries; entries m_entries;
......
...@@ -21,9 +21,9 @@ ...@@ -21,9 +21,9 @@
#define CAF_DETAIL_GROUP_MANAGER_HPP #define CAF_DETAIL_GROUP_MANAGER_HPP
#include <map> #include <map>
#include <mutex>
#include <thread>
#include "caf/mutex.hpp"
#include "caf/thread.hpp"
#include "caf/abstract_group.hpp" #include "caf/abstract_group.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
...@@ -66,7 +66,7 @@ class group_manager { ...@@ -66,7 +66,7 @@ class group_manager {
group_manager(); group_manager();
modules_map m_mmap; modules_map m_mmap;
std::mutex m_mmap_mtx; mutex m_mmap_mtx;
}; };
} // namespace detail } // namespace detail
......
/******************************************************************************
* ____ _ _____ *
* / ___| / \ | ___| C++ *
* | | / _ \ | |_ Actor *
* | |___ / ___ \| _| Framework *
* \____/_/ \_|_| *
* *
* Copyright (C) 2011 - 2014 *
* Dominik Charousset <dominik.charousset (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. *
******************************************************************************/
#ifndef CAF_PRODUCER_CONSUMER_LIST_HPP
#define CAF_PRODUCER_CONSUMER_LIST_HPP
#include "caf/config.hpp"
#define CAF_CACHE_LINE_SIZE 64
#include <chrono>
#include <atomic>
#include <cassert>
#include "caf/thread.hpp"
namespace caf {
namespace detail {
/**
* A producer-consumer list.
* For implementation details see http://drdobbs.com/cpp/211601363.
*/
template <class T>
class producer_consumer_list {
public:
using value_type = T;
using size_type = size_t;
using difference_type = ptrdiff_t;
using reference = value_type&;
using const_reference = const value_type&;
using pointer = value_type*;
using const_pointer = const value_type*;
class node {
public:
pointer value;
std::atomic<node*> next;
node(pointer val) : value(val), next(nullptr) {}
private:
static constexpr size_type payload_size =
sizeof(pointer) + sizeof(std::atomic<node*>);
static constexpr size_type cline_size = CAF_CACHE_LINE_SIZE;
static constexpr size_type pad_size =
(cline_size * ((payload_size / cline_size) + 1)) - payload_size;
// avoid false sharing
char pad[pad_size];
};
private:
static_assert(sizeof(node*) < CAF_CACHE_LINE_SIZE,
"sizeof(node*) >= CAF_CACHE_LINE_SIZE");
// for one consumer at a time
std::atomic<node*> m_first;
char m_pad1[CAF_CACHE_LINE_SIZE - sizeof(node*)];
// for one producers at a time
std::atomic<node*> m_last;
char m_pad2[CAF_CACHE_LINE_SIZE - sizeof(node*)];
// shared among producers
std::atomic<bool> m_consumer_lock;
std::atomic<bool> m_producer_lock;
public:
producer_consumer_list() {
auto ptr = new node(nullptr);
m_first = ptr;
m_last = ptr;
m_consumer_lock = false;
m_producer_lock = false;
}
~producer_consumer_list() {
while (m_first) {
node* tmp = m_first;
m_first = tmp->next.load();
delete tmp;
}
}
inline void push_back(pointer value) {
assert(value != nullptr);
node* tmp = new node(value);
// acquire exclusivity
while (m_producer_lock.exchange(true)) {
this_thread::yield();
}
// publish & swing last forward
m_last.load()->next = tmp;
m_last = tmp;
// release exclusivity
m_producer_lock = false;
}
// returns nullptr on failure
pointer try_pop() {
pointer result = nullptr;
while (m_consumer_lock.exchange(true)) {
this_thread::yield();
}
// only one consumer allowed
node* first = m_first;
node* next = m_first.load()->next;
if (next) {
// queue is not empty
result = next->value; // take it out of the node
next->value = nullptr;
// swing first forward
m_first = next;
// release exclusivity
m_consumer_lock = false;
// delete old dummy
// first->value = nullptr;
delete first;
return result;
} else {
// release exclusivity
m_consumer_lock = false;
return nullptr;
}
}
bool empty() const {
// atomically compares first and last pointer without locks
return m_first == m_last;
}
};
} // namespace detail
} // namespace caf
#endif // CAF_PRODUCER_CONSUMER_LIST_HPP
...@@ -89,11 +89,13 @@ class single_reader_queue { ...@@ -89,11 +89,13 @@ class single_reader_queue {
if (!e) { if (!e) {
// if tail is nullptr, the queue has been closed // if tail is nullptr, the queue has been closed
m_delete(new_element); m_delete(new_element);
printf("enqueue return enqueue_result::queue_closed\n");
return enqueue_result::queue_closed; return enqueue_result::queue_closed;
} }
// a dummy is never part of a non-empty list // a dummy is never part of a non-empty list
new_element->next = is_dummy(e) ? nullptr : e; new_element->next = is_dummy(e) ? nullptr : e;
if (m_stack.compare_exchange_strong(e, new_element)) { if (m_stack.compare_exchange_strong(e, new_element)) {
printf("enqueue return enqueue_result::unblocked_reader or enqueue_result::success\n");
return (e == reader_blocked_dummy()) ? enqueue_result::unblocked_reader return (e == reader_blocked_dummy()) ? enqueue_result::unblocked_reader
: enqueue_result::success; : enqueue_result::success;
} }
...@@ -218,70 +220,27 @@ class single_reader_queue { ...@@ -218,70 +220,27 @@ class single_reader_queue {
* support for synchronized access * * support for synchronized access *
**************************************************************************/ **************************************************************************/
#ifdef __RIOTBUILD_FLAG
bool synchronized_enqueue(pthread_mutex_t& mtx, pthread_cond_t& cv, pointer new_element) {
//template <class Mutex, class CondVar>
//bool synchronized_enqueue(Mutex& mtx, CondVar& cv, pointer new_element) {
switch (enqueue(new_element)) {
case enqueue_result::unblocked_reader: {
pthread_mutex_lock(&mtx);
pthread_cond_signal(&cv);
pthread_mutex_unlock(&mtx);
return true;
}
}
}
void synchronized_await(pthread_mutex_t& mtx, pthread_cond_t& cv) {
//template <class Mutex, class CondVar>
//void synchronized_await(Mutex& mtx, CondVar& cv) {
CAF_REQUIRE(!closed());
if (!can_fetch_more() && try_block()) { // todo what does the try_block?
pthread_mutex_lock(&mtx);
while (blocked()) {
pthread_cond_wait(&cv,&mtx);
}
pthread_mutex_unlock(&mtx);
}
}
bool synchronized_await(pthread_mutex_t& mtx, pthread_cond_t& cv,const struct timespec& timeout) {
//template <class Mutex, class CondVar, class TimePoint>
//bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) {
CAF_REQUIRE(!closed());
if (!can_fetch_more() && try_block()) {
pthread_mutex_lock(&mtx);
while (blocked()) {
auto rc = pthread_cond_timedwait(&cv,&mtx,&timeout);
if (rc == ETIMEDOUT) {
//if (cv.wait_until(guard, timeout) == std::cv_status::timeout) {
// if we're unable to set the queue from blocked to empty,
// than there's a new element in the list
pthread_mutex_unlock(&mtx);
return !try_unblock();
}
}
}
pthread_mutex_unlock(&mtx);
return true;
}
#else
template <class Mutex, class CondVar> template <class Mutex, class CondVar>
bool synchronized_enqueue(Mutex& mtx, CondVar& cv, pointer new_element) { bool synchronized_enqueue(Mutex& mtx, CondVar& cv, pointer new_element) {
switch (enqueue(new_element)) { switch (enqueue(new_element)) {
case enqueue_result::unblocked_reader: { case enqueue_result::unblocked_reader: {
std::unique_lock<Mutex> guard(mtx); printf("synchronized_enqueue -> unblocked_reader\n");
unique_lock<Mutex> guard(mtx);
printf("synchronized_enqueue -> unblocked_reader -> created guard\n");
cv.notify_one(); cv.notify_one();
printf("synchronized_enqueue -> unblocked_reader -> notified one\n");
return true; return true;
} }
case enqueue_result::success: case enqueue_result::success: {
printf("synchronized_enqueue -> success\n");
// enqueued message to a running actor's mailbox // enqueued message to a running actor's mailbox
return true; return true;
case enqueue_result::queue_closed: }
case enqueue_result::queue_closed: {
printf("synchronized_enqueue -> queue_closed\n");
// actor no longer alive // actor no longer alive
return false; return false;
}
} }
// should be unreachable // should be unreachable
CAF_CRITICAL("invalid result of enqueue()"); CAF_CRITICAL("invalid result of enqueue()");
...@@ -291,7 +250,7 @@ class single_reader_queue { ...@@ -291,7 +250,7 @@ class single_reader_queue {
void synchronized_await(Mutex& mtx, CondVar& cv) { void synchronized_await(Mutex& mtx, CondVar& cv) {
CAF_ASSERT(!closed()); CAF_ASSERT(!closed());
if (!can_fetch_more() && try_block()) { if (!can_fetch_more() && try_block()) {
std::unique_lock<Mutex> guard(mtx); unique_lock<Mutex> guard(mtx);
while (blocked()) { while (blocked()) {
cv.wait(guard); cv.wait(guard);
} }
...@@ -302,9 +261,9 @@ class single_reader_queue { ...@@ -302,9 +261,9 @@ class single_reader_queue {
bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) { bool synchronized_await(Mutex& mtx, CondVar& cv, const TimePoint& timeout) {
CAF_ASSERT(!closed()); CAF_ASSERT(!closed());
if (!can_fetch_more() && try_block()) { if (!can_fetch_more() && try_block()) {
std::unique_lock<Mutex> guard(mtx); unique_lock<Mutex> guard(mtx);
while (blocked()) { while (blocked()) {
if (cv.wait_until(guard, timeout) == std::cv_status::timeout) { if (cv.wait_until(guard, timeout) == cv_status::timeout) {
// if we're unable to set the queue from blocked to empty, // if we're unable to set the queue from blocked to empty,
// than there's a new element in the list // than there's a new element in the list
return !try_unblock(); return !try_unblock();
...@@ -313,7 +272,6 @@ class single_reader_queue { ...@@ -313,7 +272,6 @@ class single_reader_queue {
} }
return true; return true;
} }
#endif
private: private:
// exposed to "outside" access // exposed to "outside" access
......
...@@ -20,11 +20,11 @@ ...@@ -20,11 +20,11 @@
#ifndef CAF_DETAIL_SINGLETONS_HPP #ifndef CAF_DETAIL_SINGLETONS_HPP
#define CAF_DETAIL_SINGLETONS_HPP #define CAF_DETAIL_SINGLETONS_HPP
#include <mutex>
#include <atomic> #include <atomic>
#include <cstddef> // size_t #include <cstddef> // size_t
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/mutex.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/detail/cas_weak.hpp" #include "caf/detail/cas_weak.hpp"
...@@ -85,7 +85,7 @@ class singletons { ...@@ -85,7 +85,7 @@ class singletons {
static void stop_singletons(); static void stop_singletons();
private: private:
static std::mutex& get_plugin_mutex(); static mutex& get_mutex();
static std::atomic<abstract_singleton*>& get_plugin_singleton(size_t id); static std::atomic<abstract_singleton*>& get_plugin_singleton(size_t id);
...@@ -94,7 +94,7 @@ class singletons { ...@@ -94,7 +94,7 @@ class singletons {
static T* lazy_get(std::atomic<T*>& ptr, std::mutex& mtx, Factory f) { static T* lazy_get(std::atomic<T*>& ptr, std::mutex& mtx, Factory f) {
auto result = ptr.load(std::memory_order_acquire); auto result = ptr.load(std::memory_order_acquire);
if (result == nullptr) { if (result == nullptr) {
std::lock_guard<std::mutex> guard(mtx); lock_guard<mutex> guard(get_mutex());
result = ptr.load(std::memory_order_relaxed); result = ptr.load(std::memory_order_relaxed);
if (result == nullptr) { if (result == nullptr) {
result = f(); result = f();
......
...@@ -20,13 +20,10 @@ ...@@ -20,13 +20,10 @@
#ifndef CAF_DETAIL_LOCKS_HPP #ifndef CAF_DETAIL_LOCKS_HPP
#define CAF_DETAIL_LOCKS_HPP #define CAF_DETAIL_LOCKS_HPP
#include <mutex> #include "caf/mutex.hpp"
namespace caf { namespace caf {
template <class Lockable>
using unique_lock = std::unique_lock<Lockable>;
template <class SharedLockable> template <class SharedLockable>
class shared_lock { class shared_lock {
public: public:
......
...@@ -22,10 +22,10 @@ ...@@ -22,10 +22,10 @@
#include <deque> #include <deque>
#include <chrono> #include <chrono>
#include <thread>
#include <random> #include <random>
#include <cstddef> #include <cstddef>
#include "caf/thread.hpp"
#include "caf/resumable.hpp" #include "caf/resumable.hpp"
#include "caf/detail/double_ended_queue.hpp" #include "caf/detail/double_ended_queue.hpp"
...@@ -163,7 +163,7 @@ class work_stealing { ...@@ -163,7 +163,7 @@ class work_stealing {
return job; return job;
} }
} }
std::this_thread::sleep_for(strat.sleep_duration); this_thread::sleep_for(strat.sleep_duration);
} }
} }
// unreachable, because the last strategy loops // unreachable, because the last strategy loops
......
...@@ -26,4 +26,5 @@ ...@@ -26,4 +26,5 @@
#include "caf/scheduler/detached_threads.hpp" #include "caf/scheduler/detached_threads.hpp"
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#endif // CAF_SCHEDULER_HPP #endif // CAF_SCHEDULER_HPP
...@@ -20,11 +20,11 @@ ...@@ -20,11 +20,11 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include <map> #include <map>
#include <mutex>
#include <atomic> #include <atomic>
#include <stdexcept> #include <stdexcept>
#include "caf/atom.hpp" #include "caf/atom.hpp"
#include "caf/mutex.hpp"
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/actor_addr.hpp" #include "caf/actor_addr.hpp"
...@@ -41,7 +41,7 @@ ...@@ -41,7 +41,7 @@
namespace caf { namespace caf {
namespace { namespace {
using guard_type = std::unique_lock<std::mutex>; using guard_type = unique_lock<mutex>;
} // namespace <anonymous> } // namespace <anonymous>
// m_exit_reason is guaranteed to be set to 0, i.e., exit_reason::not_exited, // m_exit_reason is guaranteed to be set to 0, i.e., exit_reason::not_exited,
......
...@@ -19,20 +19,20 @@ ...@@ -19,20 +19,20 @@
#include "caf/scheduler/abstract_coordinator.hpp" #include "caf/scheduler/abstract_coordinator.hpp"
#include <thread>
#include <atomic> #include <atomic>
#include <chrono> #include <chrono>
#include <iostream> #include <iostream>
#include <condition_variable>
#include "caf/on.hpp" #include "caf/on.hpp"
#include "caf/send.hpp" #include "caf/send.hpp"
#include "caf/spawn.hpp" #include "caf/spawn.hpp"
#include "caf/thread.hpp"
#include "caf/anything.hpp" #include "caf/anything.hpp"
#include "caf/to_string.hpp" #include "caf/to_string.hpp"
#include "caf/local_actor.hpp" #include "caf/local_actor.hpp"
#include "caf/scoped_actor.hpp" #include "caf/scoped_actor.hpp"
#include "caf/system_messages.hpp" #include "caf/system_messages.hpp"
#include "caf/condition_variable.hpp"
#include "caf/scheduler/coordinator.hpp" #include "caf/scheduler/coordinator.hpp"
......
...@@ -25,14 +25,14 @@ namespace caf { ...@@ -25,14 +25,14 @@ namespace caf {
void actor_companion::disconnect(std::uint32_t rsn) { void actor_companion::disconnect(std::uint32_t rsn) {
enqueue_handler tmp; enqueue_handler tmp;
{ // lifetime scope of guard { // lifetime scope of guard
std::lock_guard<lock_type> guard(m_lock); lock_guard<lock_type> guard(m_lock);
m_on_enqueue.swap(tmp); m_on_enqueue.swap(tmp);
} }
cleanup(rsn); cleanup(rsn);
} }
void actor_companion::on_enqueue(enqueue_handler handler) { void actor_companion::on_enqueue(enqueue_handler handler) {
std::lock_guard<lock_type> guard(m_lock); lock_guard<lock_type> guard(m_lock);
m_on_enqueue = std::move(handler); m_on_enqueue = std::move(handler);
} }
......
...@@ -56,7 +56,7 @@ actor_proxy_ptr actor_proxy::anchor::get() { ...@@ -56,7 +56,7 @@ actor_proxy_ptr actor_proxy::anchor::get() {
} }
bool actor_proxy::anchor::try_expire() noexcept { bool actor_proxy::anchor::try_expire() noexcept {
std::lock_guard<detail::shared_spinlock> guard{m_lock}; lock_guard<detail::shared_spinlock> guard{m_lock};
// double-check reference count // double-check reference count
if (m_ptr.load()->get_reference_count() == 0) { if (m_ptr.load()->get_reference_count() == 0) {
m_ptr = nullptr; m_ptr = nullptr;
......
...@@ -19,11 +19,11 @@ ...@@ -19,11 +19,11 @@
#include "caf/detail/actor_registry.hpp" #include "caf/detail/actor_registry.hpp"
#include <mutex>
#include <limits> #include <limits>
#include <stdexcept> #include <stdexcept>
#include "caf/locks.hpp" #include "caf/locks.hpp"
#include "caf/mutex.hpp"
#include "caf/attachable.hpp" #include "caf/attachable.hpp"
#include "caf/exit_reason.hpp" #include "caf/exit_reason.hpp"
...@@ -108,7 +108,7 @@ size_t actor_registry::running() const { ...@@ -108,7 +108,7 @@ size_t actor_registry::running() const {
void actor_registry::dec_running() { void actor_registry::dec_running() {
size_t new_val = --m_running; size_t new_val = --m_running;
if (new_val <= 1) { if (new_val <= 1) {
std::unique_lock<std::mutex> guard(m_running_mtx); unique_lock<mutex> guard(m_running_mtx);
m_running_cv.notify_all(); m_running_cv.notify_all();
} }
CAF_LOG_DEBUG(CAF_ARG(new_val)); CAF_LOG_DEBUG(CAF_ARG(new_val));
...@@ -117,7 +117,7 @@ void actor_registry::dec_running() { ...@@ -117,7 +117,7 @@ void actor_registry::dec_running() {
void actor_registry::await_running_count_equal(size_t expected) { void actor_registry::await_running_count_equal(size_t expected) {
CAF_ASSERT(expected == 0 || expected == 1); CAF_ASSERT(expected == 0 || expected == 1);
CAF_LOG_TRACE(CAF_ARG(expected)); CAF_LOG_TRACE(CAF_ARG(expected));
std::unique_lock<std::mutex> guard{m_running_mtx}; unique_lock<mutex> guard{m_running_mtx};
while (m_running != expected) { while (m_running != expected) {
CAF_LOG_DEBUG("count = " << m_running.load()); CAF_LOG_DEBUG("count = " << m_running.load());
m_running_cv.wait(guard); m_running_cv.wait(guard);
......
...@@ -18,20 +18,20 @@ ...@@ -18,20 +18,20 @@
******************************************************************************/ ******************************************************************************/
#include <set> #include <set>
#include <mutex>
#include <sstream> #include <sstream>
#include <stdexcept> #include <stdexcept>
#include <condition_variable>
#include "caf/locks.hpp" #include "caf/locks.hpp"
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/group.hpp" #include "caf/group.hpp"
#include "caf/to_string.hpp" #include "caf/mutex.hpp"
#include "caf/message.hpp" #include "caf/message.hpp"
#include "caf/to_string.hpp"
#include "caf/serializer.hpp" #include "caf/serializer.hpp"
#include "caf/deserializer.hpp" #include "caf/deserializer.hpp"
#include "caf/event_based_actor.hpp" #include "caf/event_based_actor.hpp"
#include "caf/condition_variable.hpp"
#include "caf/detail/group_manager.hpp" #include "caf/detail/group_manager.hpp"
...@@ -470,7 +470,7 @@ void group_manager::add_module(std::unique_ptr<abstract_group::module> mptr) { ...@@ -470,7 +470,7 @@ void group_manager::add_module(std::unique_ptr<abstract_group::module> mptr) {
} }
auto& mname = mptr->name(); auto& mname = mptr->name();
{ // lifetime scope of guard { // lifetime scope of guard
std::lock_guard<std::mutex> guard(m_mmap_mtx); lock_guard<mutex> guard(m_mmap_mtx);
if (m_mmap.insert(std::make_pair(mname, std::move(mptr))).second) { if (m_mmap.insert(std::make_pair(mname, std::move(mptr))).second) {
return; // success; don't throw return; // success; don't throw
} }
...@@ -482,7 +482,7 @@ void group_manager::add_module(std::unique_ptr<abstract_group::module> mptr) { ...@@ -482,7 +482,7 @@ void group_manager::add_module(std::unique_ptr<abstract_group::module> mptr) {
} }
abstract_group::module* group_manager::get_module(const std::string& mname) { abstract_group::module* group_manager::get_module(const std::string& mname) {
std::lock_guard<std::mutex> guard(m_mmap_mtx); lock_guard<mutex> guard(m_mmap_mtx);
auto i = m_mmap.find(mname); auto i = m_mmap.find(mname);
return (i != m_mmap.end()) ? i->second.get() : nullptr; return (i != m_mmap.end()) ? i->second.get() : nullptr;
} }
......
...@@ -71,7 +71,7 @@ void local_actor::join(const group& what) { ...@@ -71,7 +71,7 @@ void local_actor::join(const group& what) {
return; return;
} }
abstract_group::subscription_token tk{what.ptr()}; abstract_group::subscription_token tk{what.ptr()};
std::unique_lock<std::mutex> guard{m_mtx}; unique_lock<mutex> guard{m_mtx};
if (detach_impl(tk, m_attachables_head, true, true) == 0) { if (detach_impl(tk, m_attachables_head, true, true) == 0) {
auto ptr = what->subscribe(address()); auto ptr = what->subscribe(address());
if (ptr) { if (ptr) {
...@@ -98,7 +98,7 @@ std::vector<group> local_actor::joined_groups() const { ...@@ -98,7 +98,7 @@ std::vector<group> local_actor::joined_groups() const {
std::vector<group> result; std::vector<group> result;
result.reserve(20); result.reserve(20);
attachable::token stk{attachable::token::subscription, nullptr}; attachable::token stk{attachable::token::subscription, nullptr};
std::unique_lock<std::mutex> guard{m_mtx}; unique_lock<mutex> guard{m_mtx};
for (attachable* i = m_attachables_head.get(); i != 0; i = i->next.get()) { for (attachable* i = m_attachables_head.get(); i != 0; i = i->next.get()) {
if (i->matches(stk)) { if (i->matches(stk)) {
auto ptr = static_cast<abstract_group::subscription*>(i); auto ptr = static_cast<abstract_group::subscription*>(i);
......
...@@ -18,12 +18,10 @@ ...@@ -18,12 +18,10 @@
******************************************************************************/ ******************************************************************************/
#include <ctime> #include <ctime>
#include <thread>
#include <cstring> #include <cstring>
#include <fstream> #include <fstream>
#include <algorithm> #include <algorithm>
#include <pthread.h> #include <pthread.h>
#include <condition_variable>
#ifndef CAF_WINDOWS #ifndef CAF_WINDOWS
#include <unistd.h> #include <unistd.h>
...@@ -40,7 +38,9 @@ ...@@ -40,7 +38,9 @@
#include "caf/string_algorithms.hpp" #include "caf/string_algorithms.hpp"
#include "caf/all.hpp" #include "caf/all.hpp"
#include "caf/thread.hpp"
#include "caf/actor_proxy.hpp" #include "caf/actor_proxy.hpp"
#include "caf/condition_variable.hpp"
#include "caf/detail/logging.hpp" #include "caf/detail/logging.hpp"
#include "caf/detail/single_reader_queue.hpp" #include "caf/detail/single_reader_queue.hpp"
...@@ -78,7 +78,7 @@ class logging_impl : public logging { ...@@ -78,7 +78,7 @@ class logging_impl : public logging {
public: public:
void initialize() override { void initialize() override {
const char* log_level_table[] = {"ERROR", "WARN", "INFO", "DEBUG", "TRACE"}; const char* log_level_table[] = {"ERROR", "WARN", "INFO", "DEBUG", "TRACE"};
m_thread = std::thread([this] { (*this)(); }); m_thread = thread([this] { (*this)(); });
std::string msg = "ENTRY log level = "; std::string msg = "ENTRY log level = ";
msg += log_level_table[global_log_level]; msg += log_level_table[global_log_level];
log("TRACE", "logging", "run", __FILE__, __LINE__, msg); log("TRACE", "logging", "run", __FILE__, __LINE__, msg);
...@@ -154,7 +154,7 @@ class logging_impl : public logging { ...@@ -154,7 +154,7 @@ class logging_impl : public logging {
} }
std::ostringstream line; std::ostringstream line;
line << time(0) << " " << level << " " line << time(0) << " " << level << " "
<< "actor" << get_aid() << " " << std::this_thread::get_id() << " " << "actor" << get_aid() << " " << this_thread::get_id() << " "
<< class_name << " " << function_name << " " << file_name << ":" << class_name << " " << function_name << " " << file_name << ":"
<< line_num << " " << msg << std::endl; << line_num << " " << msg << std::endl;
m_queue.synchronized_enqueue(m_queue_mtx, m_queue_cv, m_queue.synchronized_enqueue(m_queue_mtx, m_queue_cv,
...@@ -162,9 +162,9 @@ class logging_impl : public logging { ...@@ -162,9 +162,9 @@ class logging_impl : public logging {
} }
private: private:
std::thread m_thread; thread m_thread;
std::mutex m_queue_mtx; mutex m_queue_mtx;
std::condition_variable m_queue_cv; condition_variable m_queue_cv;
detail::single_reader_queue<log_event> m_queue; detail::single_reader_queue<log_event> m_queue;
}; };
......
...@@ -20,7 +20,8 @@ ...@@ -20,7 +20,8 @@
#include "caf/config.hpp" #include "caf/config.hpp"
#include <limits> #include <limits>
#include <thread>
#include "caf/thread.hpp"
#include "caf/detail/shared_spinlock.hpp" #include "caf/detail/shared_spinlock.hpp"
...@@ -84,7 +85,7 @@ void shared_spinlock::lock_shared() { ...@@ -84,7 +85,7 @@ void shared_spinlock::lock_shared() {
long v = m_flag.load(); long v = m_flag.load();
for (;;) { for (;;) {
if (v < 0) { if (v < 0) {
// std::this_thread::yield(); // this_thread::yield();
v = m_flag.load(); v = m_flag.load();
} else if (cas_weak(&m_flag, &v, v + 1)) { } else if (cas_weak(&m_flag, &v, v + 1)) {
return; return;
......
...@@ -23,9 +23,9 @@ ...@@ -23,9 +23,9 @@
#include <map> #include <map>
#include <vector> #include <vector>
#include <memory> #include <memory>
#include <thread>
#include "caf/fwd.hpp" #include "caf/fwd.hpp"
#include "caf/thread.hpp"
#include "caf/node_id.hpp" #include "caf/node_id.hpp"
#include "caf/actor_namespace.hpp" #include "caf/actor_namespace.hpp"
#include "caf/detail/singletons.hpp" #include "caf/detail/singletons.hpp"
...@@ -139,7 +139,7 @@ class middleman : public detail::abstract_singleton { ...@@ -139,7 +139,7 @@ class middleman : public detail::abstract_singleton {
// prevents backend from shutting down unless explicitly requested // prevents backend from shutting down unless explicitly requested
network::multiplexer::supervisor_ptr m_backend_supervisor; network::multiplexer::supervisor_ptr m_backend_supervisor;
// runs the backend // runs the backend
std::thread m_thread; thread m_thread;
// keeps track of "singleton-like" brokers // keeps track of "singleton-like" brokers
std::map<atom_value, broker_ptr> m_named_brokers; std::map<atom_value, broker_ptr> m_named_brokers;
// keeps track of anonymous brokers // keeps track of anonymous brokers
......
...@@ -20,14 +20,13 @@ ...@@ -20,14 +20,13 @@
#ifndef CAF_IO_NETWORK_DEFAULT_MULTIPLEXER_HPP #ifndef CAF_IO_NETWORK_DEFAULT_MULTIPLEXER_HPP
#define CAF_IO_NETWORK_DEFAULT_MULTIPLEXER_HPP #define CAF_IO_NETWORK_DEFAULT_MULTIPLEXER_HPP
#include <thread>
#include <vector> #include <vector>
#include <string> #include <string>
#include <cstdint> #include <cstdint>
#include "caf/config.hpp" #include "caf/config.hpp"
#include "caf/extend.hpp" #include "caf/extend.hpp"
#include "caf/thread.hpp"
#include "caf/exception.hpp" #include "caf/exception.hpp"
#include "caf/ref_counted.hpp" #include "caf/ref_counted.hpp"
......
...@@ -344,7 +344,7 @@ void middleman::initialize() { ...@@ -344,7 +344,7 @@ void middleman::initialize() {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
m_backend = network::multiplexer::make(); m_backend = network::multiplexer::make();
m_backend_supervisor = m_backend->make_supervisor(); m_backend_supervisor = m_backend->make_supervisor();
m_thread = std::thread([this] { m_thread = thread([this] {
CAF_LOG_TRACE(""); CAF_LOG_TRACE("");
m_backend->run(); m_backend->run();
}); });
......
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