Add a method for blocking threads on ports.

Additionally add the first lock class since we are becoming more
concurrent.

zion/object/port.cpp

@@ -1,5 +1,7 @@
 #include "object/port.h"
 
+#include "scheduler/scheduler.h"
+
 Port::Port() {}
 
 z_err_t Port::Write(const ZMessage& msg) {
@@ -21,16 +23,26 @@ z_err_t Port::Write(const ZMessage& msg) {
   for (uint64_t i = 0; i < msg.num_bytes; i++) {
     message.bytes[i] = msg.bytes[i];
   }
+
+  MutexHolder lock(mutex_);
   pending_messages_.PushBack(message);
+  if (blocked_threads_.size() > 0) {
+    gScheduler->Enqueue(blocked_threads_.PopFront());
+  }
   return Z_OK;
 }
 
 z_err_t Port::Read(ZMessage& msg) {
-  if (pending_messages_.size() < 1) {
-    dbgln("Implement blocking");
-    return Z_ERR_UNIMPLEMENTED;
+  mutex_.Lock();
+  while (pending_messages_.size() < 1) {
+    blocked_threads_.PushBack(gScheduler->CurrentThread());
+    mutex_.Unlock();
+    gScheduler->Yield();
+    mutex_.Lock();
   }
+  mutex_.Unlock();
 
+  MutexHolder lock(mutex_);
   Message next_msg = pending_messages_.PeekFront();
   if (next_msg.num_bytes > msg.num_bytes) {
     return Z_ERR_BUFF_SIZE;

zion/object/port.h

@@ -1,7 +1,9 @@
 #pragma once
 
 #include "lib/linked_list.h"
+#include "lib/mutex.h"
 #include "object/kernel_object.h"
+#include "object/thread.h"
 #include "usr/zcall_internal.h"
 
 class Port : public KernelObject {
@@ -19,4 +21,8 @@ class Port : public KernelObject {
   };
 
   LinkedList<Message> pending_messages_;
+
+  LinkedList<RefPtr<Thread>> blocked_threads_;
+
+  Mutex mutex_{"Port"};
 };

zion/object/process.cpp

@@ -34,12 +34,14 @@ Process::Process()
     : id_(gNextId++), vmas_(MakeRefCounted<AddressSpace>()), state_(RUNNING) {}
 
 RefPtr<Thread> Process::CreateThread() {
+  MutexHolder lock(mutex_);
   RefPtr<Thread> thread = MakeRefCounted<Thread>(*this, next_thread_id_++);
   threads_.PushBack(thread);
   return thread;
 }
 
 RefPtr<Thread> Process::GetThread(uint64_t tid) {
+  MutexHolder lock(mutex_);
   auto iter = threads_.begin();
   while (iter != threads_.end()) {
     if (iter->tid() == tid) {
@@ -52,6 +54,7 @@ RefPtr<Thread> Process::GetThread(uint64_t tid) {
 }
 
 void Process::CheckState() {
+  MutexHolder lock(mutex_);
   auto iter = threads_.begin();
   while (iter != threads_.end()) {
     if (iter->GetState() != Thread::FINISHED) {
@@ -63,6 +66,7 @@ void Process::CheckState() {
 }
 
 RefPtr<Capability> Process::ReleaseCapability(uint64_t cid) {
+  MutexHolder lock(mutex_);
   auto iter = caps_.begin();
   while (iter != caps_.end()) {
     if (*iter && iter->id() == cid) {
@@ -78,6 +82,7 @@ RefPtr<Capability> Process::ReleaseCapability(uint64_t cid) {
 }
 
 RefPtr<Capability> Process::GetCapability(uint64_t cid) {
+  MutexHolder lock(mutex_);
   auto iter = caps_.begin();
   while (iter != caps_.end()) {
     if (*iter && iter->id() == cid) {
@@ -91,11 +96,13 @@ RefPtr<Capability> Process::GetCapability(uint64_t cid) {
 }
 
 uint64_t Process::AddCapability(const RefPtr<Capability>& cap) {
+  MutexHolder lock(mutex_);
   cap->set_id(next_cap_id_++);
   caps_.PushBack(cap);
   return cap->id();
 }
 uint64_t Process::AddCapability(const RefPtr<Thread>& thread) {
+  MutexHolder lock(mutex_);
   uint64_t cap_id = next_cap_id_++;
   caps_.PushBack(
       MakeRefCounted<Capability>(thread, Capability::THREAD, cap_id, ZC_WRITE));
@@ -103,24 +110,28 @@ uint64_t Process::AddCapability(const RefPtr<Thread>& thread) {
 }
 
 uint64_t Process::AddCapability(const RefPtr<Process>& p) {
+  MutexHolder lock(mutex_);
   uint64_t cap_id = next_cap_id_++;
   caps_.PushBack(MakeRefCounted<Capability>(p, Capability::PROCESS, cap_id,
                                             ZC_WRITE | ZC_PROC_SPAWN_THREAD));
   return cap_id;
 }
 uint64_t Process::AddCapability(const RefPtr<AddressSpace>& vmas) {
+  MutexHolder lock(mutex_);
   uint64_t cap_id = next_cap_id_++;
   caps_.PushBack(MakeRefCounted<Capability>(vmas, Capability::ADDRESS_SPACE,
                                             cap_id, ZC_WRITE));
   return cap_id;
 }
 uint64_t Process::AddCapability(const RefPtr<MemoryObject>& vmmo) {
+  MutexHolder lock(mutex_);
   uint64_t cap_id = next_cap_id_++;
   caps_.PushBack(MakeRefCounted<Capability>(vmmo, Capability::MEMORY_OBJECT,
                                             cap_id, ZC_WRITE));
   return cap_id;
 }
 uint64_t Process::AddCapability(const RefPtr<Channel>& chan) {
+  MutexHolder lock(mutex_);
   uint64_t cap_id = next_cap_id_++;
   caps_.PushBack(MakeRefCounted<Capability>(chan, Capability::CHANNEL, cap_id,
                                             ZC_WRITE | ZC_READ));
@@ -128,6 +139,7 @@ uint64_t Process::AddCapability(const RefPtr<Channel>& chan) {
 }
 
 uint64_t Process::AddCapability(const RefPtr<Port>& port) {
+  MutexHolder lock(mutex_);
   uint64_t cap_id = next_cap_id_++;
   caps_.PushBack(MakeRefCounted<Capability>(port, Capability::PORT, cap_id,
                                             ZC_WRITE | ZC_READ));
@@ -135,6 +147,7 @@ uint64_t Process::AddCapability(const RefPtr<Port>& port) {
 }
 
 void Process::AddCapability(uint64_t cap_id, const RefPtr<MemoryObject>& vmmo) {
+  MutexHolder lock(mutex_);
   caps_.PushBack(MakeRefCounted<Capability>(vmmo, Capability::MEMORY_OBJECT,
                                             cap_id, ZC_WRITE));
 }

zion/object/process.h

@@ -4,6 +4,7 @@
 
 #include "capability/capability.h"
 #include "lib/linked_list.h"
+#include "lib/mutex.h"
 #include "lib/ref_ptr.h"
 #include "object/address_space.h"
 #include "object/channel.h"
@@ -51,6 +52,9 @@ class Process : public KernelObject {
   friend class MakeRefCountedFriend<Process>;
   Process();
   Process(uint64_t id) : id_(id), vmas_(AddressSpace::ForRoot()) {}
+
+  Mutex mutex_{"Process"};
+
   uint64_t id_;
   RefPtr<AddressSpace> vmas_;
   State state_;

zion/object/thread.cpp

@@ -15,7 +15,7 @@ extern "C" void jump_user_space(uint64_t rip, uint64_t rsp, uint64_t arg1,
 
 extern "C" void thread_init() {
   asm("sti");
-  gScheduler->CurrentThread().Init();
+  gScheduler->CurrentThread()->Init();
   panic("Reached end of thread.");
 }