rt: Look up ports through a single port table

Instead of a two-level lookup, just use one big table

src/rt/rust_builtin.cpp

@@ -513,7 +513,6 @@ get_port_id(rust_port *port) {
 extern "C" CDECL uintptr_t
 chan_id_send(type_desc *t, rust_task_id target_task_id,
              rust_port_id target_port_id, void *sptr) {
-    // FIXME: make sure this is thread-safe
     bool sent = false;
     rust_task *task = rust_task_thread::get_task();
 
@@ -521,9 +520,7 @@ chan_id_send(type_desc *t, rust_task_id target_task_id,
         " port: 0x%" PRIxPTR, (uintptr_t) target_task_id,
         (uintptr_t) target_port_id);
 
-    rust_task *target_task = task->kernel->get_task_by_id(target_task_id);
-    if(target_task) {
-        rust_port *port = target_task->get_port_by_id(target_port_id);
+    rust_port *port = task->kernel->get_port_by_id(target_port_id);
     if(port) {
         port->send(sptr);
         port->deref();
@@ -531,10 +528,6 @@ chan_id_send(type_desc *t, rust_task_id target_task_id,
     } else {
         LOG(task, comm, "didn't get the port");
     }
-        target_task->deref();
-    } else {
-        LOG(task, comm, "didn't get the task");
-    }
     return (uintptr_t)sent;
 }
 

src/rt/rust_kernel.cpp

@@ -18,6 +18,7 @@ rust_kernel::rust_kernel(rust_srv *srv) :
     _log(srv, NULL),
     srv(srv),
     max_task_id(0),
+    max_port_id(0),
     rval(0),
     max_sched_id(0),
     env(srv->env)
@@ -212,6 +213,46 @@ rust_kernel::get_task_by_id(rust_task_id id) {
     return task;
 }
 
+rust_port_id
+rust_kernel::register_port(rust_port *port) {
+    uintptr_t new_live_ports;
+    rust_port_id new_port_id;
+    {
+        scoped_lock with(port_lock);
+        new_port_id = max_port_id++;
+        port_table.put(new_port_id, port);
+        new_live_ports = port_table.count();
+    }
+    K(srv, new_port_id != INTPTR_MAX, "Hit the maximum port id");
+    KLOG_("Registered port %" PRIdPTR, new_port_id);
+    KLOG_("Total outstanding ports: %d", new_live_ports);
+    return new_port_id;
+}
+
+void
+rust_kernel::release_port_id(rust_port_id id) {
+    KLOG_("Releasing port %" PRIdPTR, id);
+    uintptr_t new_live_ports;
+    {
+        scoped_lock with(port_lock);
+        port_table.remove(id);
+        new_live_ports = port_table.count();
+    }
+    KLOG_("Total outstanding ports: %d", new_live_ports);
+}
+
+rust_port *
+rust_kernel::get_port_by_id(rust_port_id id) {
+    scoped_lock with(port_lock);
+    rust_port *port = NULL;
+    // get leaves port unchanged if not found.
+    port_table.get(id, &port);
+    if(port) {
+        port->ref();
+    }
+    return port;
+}
+
 #ifdef __WIN32__
 void
 rust_kernel::win32_require(LPCTSTR fn, BOOL ok) {

src/rt/rust_kernel.h

@@ -30,6 +30,12 @@ private:
     rust_task_id max_task_id;
     hash_map<rust_task_id, rust_task *> task_table;
 
+    // Protects max_port_id and port_table
+    lock_and_signal port_lock;
+    // The next port id
+    rust_task_id max_port_id;
+    hash_map<rust_port_id, rust_port *> port_table;
+
     lock_and_signal rval_lock;
     int rval;
 
@@ -73,6 +79,10 @@ public:
     rust_task *get_task_by_id(rust_task_id id);
     void release_task_id(rust_task_id tid);
 
+    rust_port_id register_port(rust_port *port);
+    rust_port *get_port_by_id(rust_port_id id);
+    void release_port_id(rust_port_id tid);
+
     void set_exit_status(int code);
 };
 

src/rt/rust_port.cpp

@@ -11,7 +11,7 @@ rust_port::rust_port(rust_task *task, size_t unit_sz)
         PRIxPTR, (uintptr_t)task, unit_sz, (uintptr_t)this);
 
     task->ref();
-    id = task->register_port(this);
+    id = kernel->register_port(this);
 }
 
 rust_port::~rust_port() {
@@ -39,7 +39,7 @@ void rust_port::deref() {
 void rust_port::begin_detach(uintptr_t *yield) {
     *yield = false;
 
-    task->release_port(id);
+    kernel->release_port_id(id);
 
     scoped_lock with(ref_lock);
     ref_count--;

src/rt/rust_task.cpp

@@ -75,7 +75,6 @@ rust_task::rust_task(rust_task_thread *thread, rust_task_list *state,
     kernel(thread->kernel),
     name(name),
     list_index(-1),
-    next_port_id(0),
     rendezvous_ptr(0),
     local_region(&thread->srv->local_region),
     boxed(&local_region),
@@ -107,11 +106,6 @@ rust_task::rust_task(rust_task_thread *thread, rust_task_list *state,
 void
 rust_task::delete_this()
 {
-    {
-        scoped_lock with (port_lock);
-        I(thread, port_table.is_empty());
-    }
-
     DLOG(thread, task, "~rust_task %s @0x%" PRIxPTR ", refcnt=%d",
          name, (uintptr_t)this, ref_count);
 
@@ -476,40 +470,12 @@ rust_task::calloc(size_t size, const char *tag) {
     return local_region.calloc(size, tag);
 }
 
-rust_port_id rust_task::register_port(rust_port *port) {
-    I(thread, !port_lock.lock_held_by_current_thread());
-    scoped_lock with(port_lock);
-
-    rust_port_id id = next_port_id++;
-    A(thread, id != INTPTR_MAX, "Hit the maximum port id");
-    port_table.put(id, port);
-    return id;
-}
-
-void rust_task::release_port(rust_port_id id) {
-    scoped_lock with(port_lock);
-    port_table.remove(id);
-}
-
-rust_port *rust_task::get_port_by_id(rust_port_id id) {
-    I(thread, !port_lock.lock_held_by_current_thread());
-    scoped_lock with(port_lock);
-    rust_port *port = NULL;
-    port_table.get(id, &port);
-    if (port) {
-        port->ref();
-    }
-    return port;
-}
-
 void
 rust_task::notify(bool success) {
     // FIXME (1078) Do this in rust code
     if(notify_enabled) {
-        rust_task *target_task = kernel->get_task_by_id(notify_chan.task);
-        if (target_task) {
         rust_port *target_port =
-                target_task->get_port_by_id(notify_chan.port);
+            kernel->get_port_by_id(notify_chan.port);
         if(target_port) {
             task_notification msg;
             msg.id = id;
@@ -518,8 +484,6 @@ rust_task::notify(bool success) {
             target_port->send(&msg);
             target_port->deref();
         }
-            target_task->deref();
-        }
     }
 }
 

src/rt/rust_task.h

@@ -70,8 +70,6 @@ rust_task : public kernel_owned<rust_task>, rust_cond
     const char *const name;
     int32_t list_index;
 
-    rust_port_id next_port_id;
-
     // Rendezvous pointer for receiving data when blocked on a port. If we're
     // trying to read data and no data is available on any incoming channel,
     // we block on the port, and yield control to the scheduler. Since, we
@@ -100,10 +98,6 @@ rust_task : public kernel_owned<rust_task>, rust_cond
 
 private:
 
-    // Protects port_table
-    lock_and_signal port_lock;
-    hash_map<rust_port_id, rust_port *> port_table;
-
     // Protects state, cond, cond_name
     lock_and_signal state_lock;
     rust_task_list *state;
@@ -201,10 +195,6 @@ public:
 
     void *calloc(size_t size, const char *tag);
 
-    rust_port_id register_port(rust_port *port);
-    void release_port(rust_port_id id);
-    rust_port *get_port_by_id(rust_port_id id);
-
     // Use this function sparingly. Depending on the ref count is generally
     // not at all safe.
     intptr_t get_ref_count() const { return ref_count; }