drew/acadia
1
0
Fork 0
Code Issues 6 Pull requests Projects Releases Packages Wiki Activity Actions

[zion] Create a message queue to be shared between the port and channel

Browse source
This commit is contained in:
Drew Galbraith 2023-06-20 15:29:32 -07:00
parent 1edd5023ce
commit fe1641ac38
7 changed files with 138 additions and 94 deletions
Download patch file Download diff file Expand all files Collapse all files

71
zion/object/port.cpp
View file

@ -4,29 +4,10 @@
Port::Port() {}
z_err_t Port::Write(const ZMessage& msg) {
if (msg.num_bytes > 0x1000) {
dbgln("Large message size unimplemented: %x", msg.num_bytes);
return Z_ERR_INVALID;
}
auto message = MakeShared<Message>();
message->num_bytes = msg.num_bytes;
message->bytes = new uint8_t[msg.num_bytes];
for (uint64_t i = 0; i < msg.num_bytes; i++) {
message->bytes[i] = static_cast<uint8_t*>(msg.data)[i];
}
for (uint64_t i = 0; i < msg.num_caps; i++) {
auto cap = gScheduler->CurrentProcess().ReleaseCapability(msg.caps[i]);
if (!cap) {
return Z_ERR_CAP_NOT_FOUND;
}
message->caps.PushBack(cap);
}
MutexHolder lock(mutex_);
pending_messages_.PushBack(message);
z_err_t Port::Write(uint64_t num_bytes, const void* bytes, uint64_t num_caps,
const z_cap_t* caps) {
MutexHolder h(mutex_);
RET_ERR(message_queue_.PushBack(num_bytes, bytes, num_caps, caps));
if (blocked_threads_.size() > 0) {
auto thread = blocked_threads_.PopFront();
thread->SetState(Thread::RUNNABLE);
@ -35,9 +16,10 @@ z_err_t Port::Write(const ZMessage& msg) {
return Z_OK;
}
z_err_t Port::Read(ZMessage& msg) {
z_err_t Port::Read(uint64_t* num_bytes, void* bytes, uint64_t* num_caps,
z_cap_t* caps) {
mutex_.Lock();
while (pending_messages_.size() < 1) {
while (message_queue_.empty()) {
blocked_threads_.PushBack(gScheduler->CurrentThread());
mutex_.Unlock();
gScheduler->Yield();
@ -46,48 +28,15 @@ z_err_t Port::Read(ZMessage& msg) {
mutex_.Unlock();
MutexHolder lock(mutex_);
auto next_msg = pending_messages_.PeekFront();
if (next_msg->num_bytes > msg.num_bytes) {
return Z_ERR_BUFF_SIZE;
}
if (next_msg->caps.size() > msg.num_caps) {
return Z_ERR_BUFF_SIZE;
}
msg.num_bytes = next_msg->num_bytes;
for (uint64_t i = 0; i < msg.num_bytes; i++) {
static_cast<uint8_t*>(msg.data)[i] = next_msg->bytes[i];
}
msg.num_caps = next_msg->caps.size();
auto& proc = gScheduler->CurrentProcess();
for (uint64_t i = 0; i < msg.num_caps; i++) {
msg.caps[i] = proc.AddExistingCapability(next_msg->caps.PopFront());
}
pending_messages_.PopFront();
return Z_OK;
return message_queue_.PopFront(num_bytes, bytes, num_caps, caps);
}
void Port::WriteKernel(uint64_t init, RefPtr<Capability> cap) {
MutexHolder h(mutex_);
auto msg = MakeShared<Message>();
msg->bytes = new uint8_t[8];
msg->num_bytes = sizeof(init);
uint8_t* data = reinterpret_cast<uint8_t*>(&init);
for (uint8_t i = 0; i < sizeof(init); i++) {
msg->bytes[i] = data[i];
}
msg->caps.PushBack(cap);
pending_messages_.PushBack(msg);
message_queue_.WriteKernel(init, cap);
}
bool Port::HasMessages() {
MutexHolder h(mutex_);
return pending_messages_.size() != 0;
return !message_queue_.empty();
}

17
zion/object/port.h
View file

@ -2,6 +2,7 @@
#include "capability/capability.h"
#include "lib/linked_list.h"
#include "lib/message_queue.h"
#include "lib/mutex.h"
#include "lib/shared_ptr.h"
#include "object/kernel_object.h"
@ -21,23 +22,17 @@ class Port : public KernelObject {
Port();
z_err_t Write(const ZMessage& msg);
z_err_t Read(ZMessage& msg);
z_err_t Write(uint64_t num_bytes, const void* bytes, uint64_t num_caps,
const z_cap_t* caps);
z_err_t Read(uint64_t* num_bytes, void* bytes, uint64_t* num_caps,
z_cap_t* caps);
void WriteKernel(uint64_t init, RefPtr<Capability> cap);
bool HasMessages();
private:
struct Message {
uint64_t num_bytes;
uint8_t* bytes;
LinkedList<RefPtr<Capability>> caps;
};
LinkedList<SharedPtr<Message>> pending_messages_;
UnboundedMessageQueue message_queue_;
LinkedList<RefPtr<Thread>> blocked_threads_;
Mutex mutex_{"Port"};