Commit 8ecfc7db authored by Dominik Charousset's avatar Dominik Charousset

priority-aware actors

this patch adds a new spawn option for priority-aware actors
and refactors libcppa's message passing; this version has a known
issue with network distributed group messaging
parent 6e9b6578
...@@ -82,7 +82,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -82,7 +82,7 @@ class actor_facade<Ret(Args...)> : public actor {
result_mapping map_result) { result_mapping map_result) {
if (global_dims.empty()) { if (global_dims.empty()) {
auto str = "OpenCL kernel needs at least 1 global dimension."; auto str = "OpenCL kernel needs at least 1 global dimension.";
CPPA_LOGM_ERROR(detail::demangle(typeid(actor_facade)), str); CPPA_LOGM_ERROR(detail::demangle(typeid(actor_facade)).c_str(), str);
throw std::runtime_error(str); throw std::runtime_error(str);
} }
auto check_vec = [&](const dim_vec& vec, const char* name) { auto check_vec = [&](const dim_vec& vec, const char* name) {
...@@ -90,7 +90,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -90,7 +90,7 @@ class actor_facade<Ret(Args...)> : public actor {
std::ostringstream oss; std::ostringstream oss;
oss << name << " vector is not empty, but " oss << name << " vector is not empty, but "
<< "its size differs from global dimensions vector's size"; << "its size differs from global dimensions vector's size";
CPPA_LOGM_ERROR(detail::demangle<actor_facade>(), oss.str()); CPPA_LOGM_ERROR(detail::demangle<actor_facade>().c_str(), oss.str());
throw std::runtime_error(oss.str()); throw std::runtime_error(oss.str());
} }
}; };
...@@ -104,7 +104,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -104,7 +104,7 @@ class actor_facade<Ret(Args...)> : public actor {
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
std::ostringstream oss; std::ostringstream oss;
oss << "clCreateKernel: " << get_opencl_error(err); oss << "clCreateKernel: " << get_opencl_error(err);
CPPA_LOGM_ERROR(detail::demangle<actor_facade>(), oss.str()); CPPA_LOGM_ERROR(detail::demangle<actor_facade>().c_str(), oss.str());
throw std::runtime_error(oss.str()); throw std::runtime_error(oss.str());
} }
return new actor_facade<Ret (Args...)>{dispatcher, return new actor_facade<Ret (Args...)>{dispatcher,
...@@ -169,7 +169,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -169,7 +169,7 @@ class actor_facade<Ret(Args...)> : public actor {
m_local_dimensions, m_local_dimensions,
m_map_result)); m_map_result));
} }
else { CPPA_LOG_ERROR("actor_facade::enqueue() tuple_cast failed."); } else { CPPA_LOGMF(CPPA_ERROR, this, "actor_facade::enqueue() tuple_cast failed."); }
} }
typedef std::vector<mem_ptr> args_vec; typedef std::vector<mem_ptr> args_vec;
...@@ -217,7 +217,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -217,7 +217,7 @@ class actor_facade<Ret(Args...)> : public actor {
arg0.data(), arg0.data(),
&err); &err);
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOG_ERROR("clCreateBuffer: " << get_opencl_error(err)); CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err));
} }
else { else {
mem_ptr tmp; mem_ptr tmp;
...@@ -241,7 +241,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -241,7 +241,7 @@ class actor_facade<Ret(Args...)> : public actor {
nullptr, nullptr,
&err); &err);
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOG_ERROR("clCreateBuffer: " << get_opencl_error(err)); CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err));
} }
else { else {
mem_ptr tmp; mem_ptr tmp;
......
...@@ -80,14 +80,14 @@ struct command_dispatcher::worker { ...@@ -80,14 +80,14 @@ struct command_dispatcher::worker {
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
ostringstream oss; ostringstream oss;
oss << "clFlush: " << get_opencl_error(err); oss << "clFlush: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
} }
catch (exception& e) { catch (exception& e) {
ostringstream oss; ostringstream oss;
oss << "worker loop, e.what(): " << e.what(); oss << "worker loop, e.what(): " << e.what();
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
} }
...@@ -107,12 +107,12 @@ void command_dispatcher::worker_loop(command_dispatcher::worker* w) { ...@@ -107,12 +107,12 @@ void command_dispatcher::worker_loop(command_dispatcher::worker* w) {
void command_dispatcher::supervisor_loop(command_dispatcher* scheduler, void command_dispatcher::supervisor_loop(command_dispatcher* scheduler,
job_queue* jq, command_ptr m_dummy) { job_queue* jq, command_ptr m_dummy) {
CPPA_LOGF_TRACE("");
unique_ptr<command_dispatcher::worker> worker; unique_ptr<command_dispatcher::worker> worker;
worker.reset(new command_dispatcher::worker(scheduler, jq, m_dummy)); worker.reset(new command_dispatcher::worker(scheduler, jq, m_dummy));
worker->start(); worker->start();
worker->m_thread.join(); worker->m_thread.join();
worker.reset(); worker.reset();
CPPA_LOG_TRACE("supervisor done");
} }
void command_dispatcher::initialize() { void command_dispatcher::initialize() {
...@@ -128,13 +128,13 @@ void command_dispatcher::initialize() { ...@@ -128,13 +128,13 @@ void command_dispatcher::initialize() {
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
ostringstream oss; ostringstream oss;
oss << "clGetPlatformIDs: " << get_opencl_error(err); oss << "clGetPlatformIDs: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw logic_error(oss.str()); throw logic_error(oss.str());
} }
else if (number_of_platforms < 1) { else if (number_of_platforms < 1) {
ostringstream oss; ostringstream oss;
oss << "clGetPlatformIDs: 'no platforms found'."; oss << "clGetPlatformIDs: 'no platforms found'.";
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw logic_error(oss.str()); throw logic_error(oss.str());
} }
...@@ -152,7 +152,7 @@ void command_dispatcher::initialize() { ...@@ -152,7 +152,7 @@ void command_dispatcher::initialize() {
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
ostringstream oss; ostringstream oss;
oss << "clGetDeviceIDs: " << get_opencl_error(err); oss << "clGetDeviceIDs: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
vector<cl_device_id> devices(num_devices); vector<cl_device_id> devices(num_devices);
...@@ -160,7 +160,7 @@ void command_dispatcher::initialize() { ...@@ -160,7 +160,7 @@ void command_dispatcher::initialize() {
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
ostringstream oss; ostringstream oss;
oss << "clGetDeviceIDs: " << get_opencl_error(err); oss << "clGetDeviceIDs: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
...@@ -169,7 +169,7 @@ void command_dispatcher::initialize() { ...@@ -169,7 +169,7 @@ void command_dispatcher::initialize() {
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
ostringstream oss; ostringstream oss;
oss << "clCreateContext: " << get_opencl_error(err); oss << "clCreateContext: " << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
...@@ -183,7 +183,7 @@ void command_dispatcher::initialize() { ...@@ -183,7 +183,7 @@ void command_dispatcher::initialize() {
char buf[buf_size]; char buf[buf_size];
err = clGetDeviceInfo(device.get(), CL_DEVICE_NAME, buf_size, buf, &return_size); err = clGetDeviceInfo(device.get(), CL_DEVICE_NAME, buf_size, buf, &return_size);
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOG_ERROR("clGetDeviceInfo (CL_DEVICE_NAME): " << get_opencl_error(err)); CPPA_LOGMF(CPPA_ERROR, self, "clGetDeviceInfo (CL_DEVICE_NAME): " << get_opencl_error(err));
fill(buf, buf+buf_size, 0); fill(buf, buf+buf_size, 0);
} }
command_queue_ptr cmd_queue; command_queue_ptr cmd_queue;
...@@ -192,7 +192,7 @@ void command_dispatcher::initialize() { ...@@ -192,7 +192,7 @@ void command_dispatcher::initialize() {
CL_QUEUE_PROFILING_ENABLE, CL_QUEUE_PROFILING_ENABLE,
&err)); &err));
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOG_DEBUG("Could not create command queue for device " CPPA_LOGMF(CPPA_DEBUG, self, "Could not create command queue for device "
<< buf << ": " << get_opencl_error(err)); << buf << ": " << get_opencl_error(err));
} }
else { else {
...@@ -207,7 +207,7 @@ void command_dispatcher::initialize() { ...@@ -207,7 +207,7 @@ void command_dispatcher::initialize() {
oss << "clGetDeviceInfo (" << id oss << "clGetDeviceInfo (" << id
<< ":CL_DEVICE_MAX_WORK_GROUP_SIZE): " << ":CL_DEVICE_MAX_WORK_GROUP_SIZE): "
<< get_opencl_error(err); << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
cl_uint max_work_item_dimensions = 0; cl_uint max_work_item_dimensions = 0;
...@@ -221,7 +221,7 @@ void command_dispatcher::initialize() { ...@@ -221,7 +221,7 @@ void command_dispatcher::initialize() {
oss << "clGetDeviceInfo (" << id oss << "clGetDeviceInfo (" << id
<< ":CL_DEVICE_MAX_WORK_ITEM_DIMENSIONS): " << ":CL_DEVICE_MAX_WORK_ITEM_DIMENSIONS): "
<< get_opencl_error(err); << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
dim_vec max_work_items_per_dim(max_work_item_dimensions); dim_vec max_work_items_per_dim(max_work_item_dimensions);
...@@ -235,7 +235,7 @@ void command_dispatcher::initialize() { ...@@ -235,7 +235,7 @@ void command_dispatcher::initialize() {
oss << "clGetDeviceInfo (" << id oss << "clGetDeviceInfo (" << id
<< ":CL_DEVICE_MAX_WORK_ITEM_SIZES): " << ":CL_DEVICE_MAX_WORK_ITEM_SIZES): "
<< get_opencl_error(err); << get_opencl_error(err);
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
device_info dev_info{id, device_info dev_info{id,
...@@ -251,7 +251,7 @@ void command_dispatcher::initialize() { ...@@ -251,7 +251,7 @@ void command_dispatcher::initialize() {
ostringstream oss; ostringstream oss;
oss << "Could not create a command queue for " oss << "Could not create a command queue for "
<< "any of the present devices."; << "any of the present devices.";
CPPA_LOG_ERROR(oss.str()); CPPA_LOGMF(CPPA_ERROR, self, oss.str());
throw runtime_error(oss.str()); throw runtime_error(oss.str());
} }
else { else {
......
...@@ -89,11 +89,11 @@ program program::create(const char* kernel_source) { ...@@ -89,11 +89,11 @@ program program::create(const char* kernel_source) {
std::ostringstream oss; std::ostringstream oss;
oss << "clBuildProgram: " << get_opencl_error(err) oss << "clBuildProgram: " << get_opencl_error(err)
<< ", build log: " << build_log.data(); << ", build log: " << build_log.data();
CPPA_LOGM_ERROR(detail::demangle<program>(), oss.str()); CPPA_LOGM_ERROR(detail::demangle<program>().c_str(), oss.str());
throw std::runtime_error(oss.str()); throw std::runtime_error(oss.str());
} }
else { else {
# ifdef CPPA_DEBUG # ifdef CPPA_ENABLE_DEBUG
device_ptr device_used(cppa::detail::singleton_manager:: device_ptr device_used(cppa::detail::singleton_manager::
get_command_dispatcher()-> get_command_dispatcher()->
m_devices.front().dev_id); m_devices.front().dev_id);
...@@ -119,8 +119,8 @@ program program::create(const char* kernel_source) { ...@@ -119,8 +119,8 @@ program program::create(const char* kernel_source) {
build_log.data(), build_log.data(),
nullptr); nullptr);
build_log[ret_val_size] = '\0'; build_log[ret_val_size] = '\0';
CPPA_LOG_DEBUG("clBuildProgram build log: " CPPA_LOGM_DEBUG("program", "clBuildProgram build log: "
<< build_log.data()); << build_log.data());
# endif # endif
} }
return {cptr, pptr}; return {cptr, pptr};
......
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