---

yaml
---
r: 11883
b: refs/heads/master
c: b278d67
h: refs/heads/master
i:
  11881: f7266ba
  11879: 4918e0e
v: v3

Author: brson

[refs]

@@ -1,5 +1,5 @@
 ---
-refs/heads/master: 337d860a8777a761267daaad9b561787b10e7c87
+refs/heads/master: b278d675a231fdfe825c72e499d59e8a3d07ffaa
 refs/heads/snap-stage1: e33de59e47c5076a89eadeb38f4934f58a3618a6
 refs/heads/snap-stage3: 4a81779abd786ff22d71434c6d9a5917ea4cdfff
 refs/heads/try: 2898dcc5d97da9427ac367542382b6239d9c0bbf

trunk/src/rt/rust_builtin.cpp

@@ -513,27 +513,20 @@ 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();
 
     LOG(task, comm, "chan_id_send task: 0x%" PRIxPTR
         " 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);
-        if(port) {
-            port->send(sptr);
-            port->deref();
-            sent = true;
-        } else {
-            LOG(task, comm, "didn't get the port");
-        }
-        target_task->deref();
+    rust_port *port = task->kernel->get_port_by_id(target_port_id);
+    if(port) {
+        port->send(sptr);
+        port->deref();
+        sent = true;
     } else {
-        LOG(task, comm, "didn't get the task");
+        LOG(task, comm, "didn't get the port");
     }
     return (uintptr_t)sent;
 }

trunk/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) {

trunk/src/rt/rust_kernel.h

@@ -30,6 +30,12 @@ class rust_kernel {
     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 @@ class rust_kernel {
     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);
 };
 

trunk/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--;

trunk/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,49 +470,19 @@ 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);
-            if(target_port) {
-                task_notification msg;
-                msg.id = id;
-                msg.result = !success ? tr_failure : tr_success;
-
-                target_port->send(&msg);
-                target_port->deref();
-            }
-            target_task->deref();
+        rust_port *target_port =
+            kernel->get_port_by_id(notify_chan.port);
+        if(target_port) {
+            task_notification msg;
+            msg.id = id;
+            msg.result = !success ? tr_failure : tr_success;
+
+            target_port->send(&msg);
+            target_port->deref();
         }
     }
 }

trunk/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 @@ rust_task : public kernel_owned<rust_task>, rust_cond
 
     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; }