Commit 258fd659 authored by Joseph Noir's avatar Joseph Noir

data copy to GPU is now asynchronous

parent d6ddf1ff
...@@ -132,6 +132,9 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -132,6 +132,9 @@ class actor_facade<Ret(Args...)> : public actor {
private: private:
using evnt_vec = std::vector<cl_event>;
using args_vec = std::vector<mem_ptr>;
actor_facade(const program& prog, actor_facade(const program& prog,
kernel_ptr kernel, kernel_ptr kernel,
const dim_vec& global_dimensions, const dim_vec& global_dimensions,
...@@ -162,22 +165,23 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -162,22 +165,23 @@ class actor_facade<Ret(Args...)> : public actor {
auto opt = m_map_args(std::move(msg)); auto opt = m_map_args(std::move(msg));
if (opt) { if (opt) {
response_handle handle{this, sender, id.response_id()}; response_handle handle{this, sender, id.response_id()};
std::vector<mem_ptr> arguments; evnt_vec events;
std::vector<cl_event> events; args_vec arguments;
add_arguments_to_kernel<Ret>(arguments, add_arguments_to_kernel<Ret>(events,
arguments,
m_result_size, m_result_size,
get_ref<Is>(*opt)...); get_ref<Is>(*opt)...);
auto cmd = make_counted<command<actor_facade, Ret>>(handle, auto cmd = make_counted<command<actor_facade, Ret>>(handle,
this, this,
std::move(arguments),
std::move(events), std::move(events),
m_result_size); std::move(arguments),
m_result_size,
opt);
cmd->enqueue(); cmd->enqueue();
} }
else { CPPA_LOGMF(CPPA_ERROR, this, "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;
kernel_ptr m_kernel; kernel_ptr m_kernel;
program_ptr m_program; program_ptr m_program;
...@@ -190,7 +194,7 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -190,7 +194,7 @@ class actor_facade<Ret(Args...)> : public actor {
result_mapping m_map_result; result_mapping m_map_result;
size_t m_result_size; size_t m_result_size;
void add_arguments_to_kernel_rec(args_vec& arguments) { void add_arguments_to_kernel_rec(evnt_vec&, args_vec& arguments) {
cl_int err{0}; cl_int err{0};
// rotate left (output buffer to the end) // rotate left (output buffer to the end)
rotate(begin(arguments), begin(arguments) + 1, end(arguments)); rotate(begin(arguments), begin(arguments) + 1, end(arguments));
...@@ -202,30 +206,50 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -202,30 +206,50 @@ class actor_facade<Ret(Args...)> : public actor {
CPPA_LOG_ERROR_IF(err != CL_SUCCESS, CPPA_LOG_ERROR_IF(err != CL_SUCCESS,
"clSetKernelArg: " << get_opencl_error(err)); "clSetKernelArg: " << get_opencl_error(err));
} }
clFlush(m_queue.get());
} }
template<typename T0, typename... Ts> template<typename T0, typename... Ts>
void add_arguments_to_kernel_rec(args_vec& arguments, void add_arguments_to_kernel_rec(evnt_vec& events,
args_vec& arguments,
T0& arg0, Ts&... args) { T0& arg0, Ts&... args) {
cl_int err{0}; cl_int err{0};
auto buf = clCreateBuffer(m_context.get(), size_t buffer_size = sizeof(typename T0::value_type) * arg0.size();
CL_MEM_READ_ONLY | CL_MEM_COPY_HOST_PTR, auto buffer = clCreateBuffer(m_context.get(),
sizeof(typename T0::value_type) * arg0.size(), CL_MEM_READ_ONLY,
arg0.data(), buffer_size,
&err); nullptr,
&err);
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err)); CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err));
return;
} }
else { cl_event event;
mem_ptr tmp; err = clEnqueueWriteBuffer(m_queue.get(),
tmp.adopt(std::move(buf)); buffer,
arguments.push_back(tmp); CL_FALSE,
add_arguments_to_kernel_rec(arguments, args...); 0,
buffer_size,
arg0.data(),
0,
nullptr,
// events.size(),
// (events.size() == 0 ? nullptr : events.data()),
&event);
if (err != CL_SUCCESS) {
CPPA_LOGMF(CPPA_ERROR, this, "clEnqueueWriteBuffer: " << get_opencl_error(err));
return;
} }
events.push_back(std::move(event));
mem_ptr tmp;
tmp.adopt(std::move(buffer));
arguments.push_back(tmp);
add_arguments_to_kernel_rec(events, arguments, args...);
} }
template<typename R, typename... Ts> template<typename R, typename... Ts>
void add_arguments_to_kernel(args_vec& arguments, void add_arguments_to_kernel(evnt_vec& events,
args_vec& arguments,
size_t ret_size, size_t ret_size,
Ts&&... args) { Ts&&... args) {
arguments.clear(); arguments.clear();
...@@ -237,13 +261,12 @@ class actor_facade<Ret(Args...)> : public actor { ...@@ -237,13 +261,12 @@ class actor_facade<Ret(Args...)> : public actor {
&err); &err);
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err)); CPPA_LOGMF(CPPA_ERROR, this, "clCreateBuffer: " << get_opencl_error(err));
return;
} }
else { mem_ptr tmp;
mem_ptr tmp; tmp.adopt(std::move(buf));
tmp.adopt(std::move(buf)); arguments.push_back(tmp);
arguments.push_back(tmp); add_arguments_to_kernel_rec(events, arguments, std::forward<Ts>(args)...);
add_arguments_to_kernel_rec(arguments, std::forward<Ts>(args)...);
}
} }
}; };
......
...@@ -53,16 +53,18 @@ class command : public ref_counted { ...@@ -53,16 +53,18 @@ class command : public ref_counted {
command(response_handle handle, command(response_handle handle,
intrusive_ptr<T> actor_facade, intrusive_ptr<T> actor_facade,
std::vector<mem_ptr> arguments,
std::vector<cl_event> events, std::vector<cl_event> events,
size_t result_size) std::vector<mem_ptr> arguments,
size_t result_size,
any_tuple msg)
: m_result_size(result_size) : m_result_size(result_size)
, m_handle(handle) , m_handle(handle)
, m_actor_facade(actor_facade) , m_actor_facade(actor_facade)
, m_queue(actor_facade->m_queue) , m_queue(actor_facade->m_queue)
, m_arguments(std::move(arguments))
, m_events(std::move(events)) , m_events(std::move(events))
, m_result(m_result_size) { } , m_arguments(std::move(arguments))
, m_result(m_result_size)
, m_msg(msg) { }
~command() { ~command() {
cl_int err{0}; cl_int err{0};
...@@ -83,21 +85,20 @@ class command : public ref_counted { ...@@ -83,21 +85,20 @@ class command : public ref_counted {
auto data_or_nullptr = [](const dim_vec& vec) { auto data_or_nullptr = [](const dim_vec& vec) {
return vec.empty() ? nullptr : vec.data(); return vec.empty() ? nullptr : vec.data();
}; };
err = clEnqueueNDRangeKernel(m_queue.get(), err = clEnqueueNDRangeKernel(m_queue.get(),
m_actor_facade->m_kernel.get(), m_actor_facade->m_kernel.get(),
m_actor_facade->m_global_dimensions.size(), m_actor_facade->m_global_dimensions.size(),
data_or_nullptr(m_actor_facade->m_global_offsets), data_or_nullptr(m_actor_facade->m_global_offsets),
data_or_nullptr(m_actor_facade->m_global_dimensions), data_or_nullptr(m_actor_facade->m_global_dimensions),
data_or_nullptr(m_actor_facade->m_local_dimensions), data_or_nullptr(m_actor_facade->m_local_dimensions),
0, m_events.size(),
nullptr, (m_events.empty() ? nullptr : m_events.data()),
&event_k); &event_k);
m_events.push_back(event_k);
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOGMF(CPPA_ERROR, self, "clEnqueueNDRangeKernel: " CPPA_LOGMF(CPPA_ERROR, self, "clEnqueueNDRangeKernel: "
<< get_opencl_error(err)); << get_opencl_error(err));
this->deref(); // or can anything actually happen? this->deref(); // or can anything actually happen?
return;
} }
else { else {
cl_event event_r; cl_event event_r;
...@@ -110,11 +111,11 @@ class command : public ref_counted { ...@@ -110,11 +111,11 @@ class command : public ref_counted {
1, 1,
&event_k, &event_k,
&event_r); &event_r);
m_events.push_back(event_r);
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
throw std::runtime_error("clEnqueueReadBuffer: " throw std::runtime_error("clEnqueueReadBuffer: "
+ get_opencl_error(err)); + get_opencl_error(err));
this->deref(); // failed to enqueue command this->deref(); // failed to enqueue command
return;
} }
err = clSetEventCallback(event_r, err = clSetEventCallback(event_r,
CL_COMPLETE, CL_COMPLETE,
...@@ -128,12 +129,15 @@ class command : public ref_counted { ...@@ -128,12 +129,15 @@ class command : public ref_counted {
CPPA_LOGMF(CPPA_ERROR, self, "clSetEventCallback: " CPPA_LOGMF(CPPA_ERROR, self, "clSetEventCallback: "
<< get_opencl_error(err)); << get_opencl_error(err));
this->deref(); // callback is not set this->deref(); // callback is not set
return;
} }
err = clFlush(m_queue.get()); err = clFlush(m_queue.get());
if (err != CL_SUCCESS) { if (err != CL_SUCCESS) {
CPPA_LOGMF(CPPA_ERROR, self, "clFlush: " << get_opencl_error(err)); CPPA_LOGMF(CPPA_ERROR, self, "clFlush: " << get_opencl_error(err));
} }
m_events.push_back(std::move(event_k));
m_events.push_back(std::move(event_r));
} }
} }
...@@ -143,9 +147,10 @@ class command : public ref_counted { ...@@ -143,9 +147,10 @@ class command : public ref_counted {
response_handle m_handle; response_handle m_handle;
intrusive_ptr<T> m_actor_facade; intrusive_ptr<T> m_actor_facade;
command_queue_ptr m_queue; command_queue_ptr m_queue;
std::vector<mem_ptr> m_arguments;
std::vector<cl_event> m_events; std::vector<cl_event> m_events;
std::vector<mem_ptr> m_arguments;
R m_result; R m_result;
any_tuple m_msg; // required to keep the argument buffers alive (for async copy)
void handle_results () { void handle_results () {
reply_tuple_to(m_handle, m_actor_facade->m_map_result(m_result)); reply_tuple_to(m_handle, m_actor_facade->m_map_result(m_result));
......
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