//// Dis VM// File: channel.cpp// Author: arr// #include <runtime.h>#include <vm_memory.h>#include <debug.h>#include <exceptions.h> using disvm::debug::component_trace_t;using disvm::debug::log_level_t; using disvm::runtime::intrinsic_type_desc;using disvm::runtime::type_descriptor_t;using disvm::runtime::vm_alloc_t;using disvm::runtime::vm_array_t;using disvm::runtime::vm_channel_t;using disvm::runtime::vm_channel_request_t;using disvm::runtime::vm_request_mutex_t;using disvm::runtime::word_t; vm_request_mutex_t::vm_request_mutex_t() : pending_request{ false }{ } vm_channel_request_t::vm_channel_request_t(uint32_t thread_id, vm_request_mutex_t &request_mutex) : data{ nullptr } , request_mutex{ request_mutex } , thread_id{ thread_id }{ } std::shared_ptr<const type_descriptor_t> vm_channel_t::type_desc(){ return intrinsic_type_desc::type<vm_channel_t>();} vm_channel_t::vm_channel_t( std::shared_ptr<const type_descriptor_t> td, data_transfer_func_t transfer, word_t buffer_len) : vm_alloc_t(vm_channel_t::type_desc()) , _data_buffer{ nullptr } , _data_buffer_head{ 0 } , _data_buffer_count{ 0 } , _data_transfer{ transfer } , _data_type{ std::move(td) }{ if (buffer_len < 0) throw out_of_range_memory{}; if (buffer_len > 0) _data_buffer = new vm_array_t{ _data_type, buffer_len }; disvm::debug::log_msg(component_trace_t::memory, log_level_t::debug, "init: vm channel");} vm_channel_t::~vm_channel_t(){ dec_ref_count_and_free(_data_buffer); disvm::debug::log_msg(component_trace_t::memory, log_level_t::debug, "destroy: vm channel");} std::shared_ptr<const type_descriptor_t> vm_channel_t::get_data_type() const{ return _data_type;} bool vm_channel_t::send_data(vm_channel_request_t &send_request){ assert(send_request.request_handled_callback != nullptr && send_request.data != nullptr); std::lock(_data_lock, send_request.request_mutex.get().pending_request_lock); std::unique_lock<std::mutex> data_lock{ _data_lock, std::adopt_lock }; std::lock_guard<std::mutex> curr_req_lock{ send_request.request_mutex.get().pending_request_lock, std::adopt_lock }; if (!send_request.request_mutex.get().pending_request) return false; // Assert the sender cannot also be a receiver in the channel assert(std::end(_data_receivers) == std::find_if( std::begin(_data_receivers), std::end(_data_receivers), [&send_request](const vm_channel_request_t &req) { return req.thread_id == send_request.thread_id; })); auto pending_req_lock = std::unique_lock<std::mutex>{}; for (;;) { if (_data_receivers.empty()) { // Check if there is room in the channel buffer if (_data_buffer != nullptr) { if (try_send_to_buffer(send_request)) return true; } if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>()) disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: send: queue: %d", send_request.thread_id); _data_senders.emplace_back(std::move(send_request)); return false; } auto &receiver_maybe = _data_receivers.front(); // Take lock and check the pending request status pending_req_lock = std::unique_lock<std::mutex>{ receiver_maybe.request_mutex.get().pending_request_lock }; if (receiver_maybe.request_mutex.get().pending_request) { // Indicate both request mutexes as non-pending send_request.request_mutex.get().pending_request = false; receiver_maybe.request_mutex.get().pending_request = false; break; } pending_req_lock.unlock(); _data_receivers.pop_front(); } // Remove the receiver from the list auto receiver = std::move(_data_receivers.front()); _data_receivers.pop_front(); // Transfer data _data_transfer(receiver.data, send_request.data, _data_type.get()); // Release channel and pending request lock after data transfer pending_req_lock.unlock(); data_lock.unlock(); if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>()) disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: send: %d %d", send_request.thread_id, receiver.thread_id); receiver.request_handled_callback(*this); return true;} bool vm_channel_t::receive_data(vm_channel_request_t &receive_request){ assert(receive_request.request_handled_callback != nullptr && receive_request.data != nullptr); std::lock(_data_lock, receive_request.request_mutex.get().pending_request_lock); std::unique_lock<std::mutex> data_lock{ _data_lock, std::adopt_lock }; std::lock_guard<std::mutex> curr_req_lock{ receive_request.request_mutex.get().pending_request_lock, std::adopt_lock }; if (!receive_request.request_mutex.get().pending_request) return false; // Assert the receiver cannot also be a sender in the channel assert(std::end(_data_senders) == std::find_if( std::begin(_data_senders), std::end(_data_senders), [&receive_request](const vm_channel_request_t &req) { return req.thread_id == receive_request.thread_id; })); auto pending_req_lock = std::unique_lock<std::mutex>{}; for (;;) { if (_data_senders.empty()) { // Check if there is data in the channel buffer if (_data_buffer != nullptr) { if (try_receive_from_buffer(receive_request)) return true; } if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>()) disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: receive: queue: %d", receive_request.thread_id); _data_receivers.emplace_back(std::move(receive_request)); return false; } auto &sender_maybe = _data_senders.front(); // Take lock and check the pending request status pending_req_lock = std::unique_lock<std::mutex>{ sender_maybe.request_mutex.get().pending_request_lock }; if (sender_maybe.request_mutex.get().pending_request) { // Indicate both request mutexes as non-pending receive_request.request_mutex.get().pending_request = false; sender_maybe.request_mutex.get().pending_request = false; break; } pending_req_lock.unlock(); _data_senders.pop_front(); } // Remove the sender from the list auto sender = std::move(_data_senders.front()); _data_senders.pop_front(); // Check if there is data in the channel buffer. This is being // done to avoid starving threads that found a full buffer when // attempting to satisfy a request. if (_data_buffer != nullptr && try_receive_from_buffer(receive_request)) { // Data was taken from the buffer, so the current pending sender // can now use the empty slot. auto did_send = try_send_to_buffer(sender); assert(did_send && "Should be room in buffer"); } else { // Transfer data since there wasn't anything in the buffer _data_transfer(receive_request.data, sender.data, _data_type.get()); if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>()) disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: receive: %d %d", receive_request.thread_id, sender.thread_id); } // Release channel and pending request lock after data transfer pending_req_lock.unlock(); data_lock.unlock(); sender.request_handled_callback(*this); return true;} void vm_channel_t::cancel_request(uint32_t thread_id){ std::lock_guard<std::mutex> lock{ _data_lock }; // Create a predicate function based on the input auto request_predicate = [thread_id](const vm_channel_request_t &req) { return req.thread_id == thread_id; }; auto receiver_iter = std::find_if(std::begin(_data_receivers), std::end(_data_receivers), request_predicate); if (receiver_iter != std::end(_data_receivers)) { _data_receivers.erase(receiver_iter); } else { auto sender_iter = std::find_if(std::begin(_data_senders), std::end(_data_senders), request_predicate); if (sender_iter == std::end(_data_senders)) return; _data_senders.erase(sender_iter); } if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>()) disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: cancel: request: %d", thread_id);} word_t vm_channel_t::get_buffer_size() const{ if (_data_buffer == nullptr) return 0; else return _data_buffer->get_length();} bool vm_channel_t::try_send_to_buffer(vm_channel_request_t &send_request){ assert(_data_buffer != nullptr); const auto buffer_length = static_cast<uint32_t>(_data_buffer->get_length()); if (_data_buffer_count == buffer_length) return false; auto next_buffer_index = _data_buffer_head + _data_buffer_count; if (next_buffer_index >= buffer_length) next_buffer_index -= buffer_length; _data_buffer_count++; auto dest = _data_buffer->at(next_buffer_index); // Transfer data to buffer _data_transfer(dest, send_request.data, _data_type.get()); send_request.request_mutex.get().pending_request = false; if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>()) disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: send: buffer: %d", send_request.thread_id); return true;} bool vm_channel_t::try_receive_from_buffer(vm_channel_request_t &receive_request){ assert(_data_buffer != nullptr); if (_data_buffer_count == 0) return false; auto src = _data_buffer->at(_data_buffer_head); _data_buffer_head++; const auto buffer_length = _data_buffer->get_length(); if (_data_buffer_head == buffer_length) _data_buffer_head = 0; _data_buffer_count--; // Transfer data from buffer _data_transfer(receive_request.data, src, _data_type.get()); receive_request.request_mutex.get().pending_request = false; if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>()) disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: receive: buffer: %d", receive_request.thread_id); return true;}
//
// Dis VM
// File: channel.cpp
// Author: arr
//
#include <runtime.h>
#include <vm_memory.h>
#include <debug.h>
#include <exceptions.h>
using disvm::debug::component_trace_t;
using disvm::debug::log_level_t;
using disvm::runtime::intrinsic_type_desc;
using disvm::runtime::type_descriptor_t;
using disvm::runtime::vm_alloc_t;
using disvm::runtime::vm_array_t;
using disvm::runtime::vm_channel_t;
using disvm::runtime::vm_channel_request_t;
using disvm::runtime::vm_request_mutex_t;
using disvm::runtime::word_t;
vm_request_mutex_t::vm_request_mutex_t()
: pending_request{ false }
{ }
vm_channel_request_t::vm_channel_request_t(uint32_t thread_id, vm_request_mutex_t &request_mutex)
: data{ nullptr }
, request_mutex{ request_mutex }
, thread_id{ thread_id }
{ }
std::shared_ptr<const type_descriptor_t> vm_channel_t::type_desc()
{
return intrinsic_type_desc::type<vm_channel_t>();
}
vm_channel_t::vm_channel_t(
std::shared_ptr<const type_descriptor_t> td,
data_transfer_func_t transfer,
word_t buffer_len)
: vm_alloc_t(vm_channel_t::type_desc())
, _data_buffer{ nullptr }
, _data_buffer_head{ 0 }
, _data_buffer_count{ 0 }
, _data_transfer{ transfer }
, _data_type{ std::move(td) }
{
if (buffer_len < 0)
throw out_of_range_memory{};
if (buffer_len > 0)
_data_buffer = new vm_array_t{ _data_type, buffer_len };
disvm::debug::log_msg(component_trace_t::memory, log_level_t::debug, "init: vm channel");
}
vm_channel_t::~vm_channel_t()
{
dec_ref_count_and_free(_data_buffer);
disvm::debug::log_msg(component_trace_t::memory, log_level_t::debug, "destroy: vm channel");
}
std::shared_ptr<const type_descriptor_t> vm_channel_t::get_data_type() const
{
return _data_type;
}
bool vm_channel_t::send_data(vm_channel_request_t &send_request)
{
assert(send_request.request_handled_callback != nullptr && send_request.data != nullptr);
std::lock(_data_lock, send_request.request_mutex.get().pending_request_lock);
std::unique_lock<std::mutex> data_lock{ _data_lock, std::adopt_lock };
std::lock_guard<std::mutex> curr_req_lock{ send_request.request_mutex.get().pending_request_lock, std::adopt_lock };
if (!send_request.request_mutex.get().pending_request)
return false;
// Assert the sender cannot also be a receiver in the channel
assert(std::end(_data_receivers) == std::find_if(
std::begin(_data_receivers),
std::end(_data_receivers),
[&send_request](const vm_channel_request_t &req) { return req.thread_id == send_request.thread_id; }));
auto pending_req_lock = std::unique_lock<std::mutex>{};
for (;;)
{
if (_data_receivers.empty())
{
// Check if there is room in the channel buffer
if (_data_buffer != nullptr)
{
if (try_send_to_buffer(send_request))
return true;
}
if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>())
disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: send: queue: %d", send_request.thread_id);
_data_senders.emplace_back(std::move(send_request));
return false;
}
auto &receiver_maybe = _data_receivers.front();
// Take lock and check the pending request status
pending_req_lock = std::unique_lock<std::mutex>{ receiver_maybe.request_mutex.get().pending_request_lock };
if (receiver_maybe.request_mutex.get().pending_request)
{
// Indicate both request mutexes as non-pending
send_request.request_mutex.get().pending_request = false;
receiver_maybe.request_mutex.get().pending_request = false;
break;
}
pending_req_lock.unlock();
_data_receivers.pop_front();
}
// Remove the receiver from the list
auto receiver = std::move(_data_receivers.front());
_data_receivers.pop_front();
// Transfer data
_data_transfer(receiver.data, send_request.data, _data_type.get());
// Release channel and pending request lock after data transfer
pending_req_lock.unlock();
data_lock.unlock();
if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>())
disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: send: %d %d", send_request.thread_id, receiver.thread_id);
receiver.request_handled_callback(*this);
return true;
}
bool vm_channel_t::receive_data(vm_channel_request_t &receive_request)
{
assert(receive_request.request_handled_callback != nullptr && receive_request.data != nullptr);
std::lock(_data_lock, receive_request.request_mutex.get().pending_request_lock);
std::unique_lock<std::mutex> data_lock{ _data_lock, std::adopt_lock };
std::lock_guard<std::mutex> curr_req_lock{ receive_request.request_mutex.get().pending_request_lock, std::adopt_lock };
if (!receive_request.request_mutex.get().pending_request)
return false;
// Assert the receiver cannot also be a sender in the channel
assert(std::end(_data_senders) == std::find_if(
std::begin(_data_senders),
std::end(_data_senders),
[&receive_request](const vm_channel_request_t &req) { return req.thread_id == receive_request.thread_id; }));
auto pending_req_lock = std::unique_lock<std::mutex>{};
for (;;)
{
if (_data_senders.empty())
{
// Check if there is data in the channel buffer
if (_data_buffer != nullptr)
{
if (try_receive_from_buffer(receive_request))
return true;
}
if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>())
disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: receive: queue: %d", receive_request.thread_id);
_data_receivers.emplace_back(std::move(receive_request));
return false;
}
auto &sender_maybe = _data_senders.front();
// Take lock and check the pending request status
pending_req_lock = std::unique_lock<std::mutex>{ sender_maybe.request_mutex.get().pending_request_lock };
if (sender_maybe.request_mutex.get().pending_request)
{
// Indicate both request mutexes as non-pending
receive_request.request_mutex.get().pending_request = false;
sender_maybe.request_mutex.get().pending_request = false;
break;
}
pending_req_lock.unlock();
_data_senders.pop_front();
}
// Remove the sender from the list
auto sender = std::move(_data_senders.front());
_data_senders.pop_front();
// Check if there is data in the channel buffer. This is being
// done to avoid starving threads that found a full buffer when
// attempting to satisfy a request.
if (_data_buffer != nullptr && try_receive_from_buffer(receive_request))
{
// Data was taken from the buffer, so the current pending sender
// can now use the empty slot.
auto did_send = try_send_to_buffer(sender);
assert(did_send && "Should be room in buffer");
}
else
{
// Transfer data since there wasn't anything in the buffer
_data_transfer(receive_request.data, sender.data, _data_type.get());
if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>())
disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: receive: %d %d", receive_request.thread_id, sender.thread_id);
}
// Release channel and pending request lock after data transfer
pending_req_lock.unlock();
data_lock.unlock();
sender.request_handled_callback(*this);
return true;
}
void vm_channel_t::cancel_request(uint32_t thread_id)
{
std::lock_guard<std::mutex> lock{ _data_lock };
// Create a predicate function based on the input
auto request_predicate = [thread_id](const vm_channel_request_t &req) { return req.thread_id == thread_id; };
auto receiver_iter = std::find_if(std::begin(_data_receivers), std::end(_data_receivers), request_predicate);
if (receiver_iter != std::end(_data_receivers))
{
_data_receivers.erase(receiver_iter);
}
else
{
auto sender_iter = std::find_if(std::begin(_data_senders), std::end(_data_senders), request_predicate);
if (sender_iter == std::end(_data_senders))
return;
_data_senders.erase(sender_iter);
}
if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>())
disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: cancel: request: %d", thread_id);
}
word_t vm_channel_t::get_buffer_size() const
{
if (_data_buffer == nullptr)
return 0;
else
return _data_buffer->get_length();
}
bool vm_channel_t::try_send_to_buffer(vm_channel_request_t &send_request)
{
assert(_data_buffer != nullptr);
const auto buffer_length = static_cast<uint32_t>(_data_buffer->get_length());
if (_data_buffer_count == buffer_length)
return false;
auto next_buffer_index = _data_buffer_head + _data_buffer_count;
if (next_buffer_index >= buffer_length)
next_buffer_index -= buffer_length;
_data_buffer_count++;
auto dest = _data_buffer->at(next_buffer_index);
// Transfer data to buffer
_data_transfer(dest, send_request.data, _data_type.get());
send_request.request_mutex.get().pending_request = false;
if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>())
disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: send: buffer: %d", send_request.thread_id);
return true;
}
bool vm_channel_t::try_receive_from_buffer(vm_channel_request_t &receive_request)
{
assert(_data_buffer != nullptr);
if (_data_buffer_count == 0)
return false;
auto src = _data_buffer->at(_data_buffer_head);
_data_buffer_head++;
const auto buffer_length = _data_buffer->get_length();
if (_data_buffer_head == buffer_length)
_data_buffer_head = 0;
_data_buffer_count--;
// Transfer data from buffer
_data_transfer(receive_request.data, src, _data_type.get());
receive_request.request_mutex.get().pending_request = false;
if (disvm::debug::is_component_tracing_enabled<component_trace_t::channel>())
disvm::debug::log_msg(component_trace_t::channel, log_level_t::debug, "channel: receive: buffer: %d", receive_request.thread_id);
return true;
}