rt: Remove rust_chan

Author: brson

mk/rt.mk

@@ -45,7 +45,6 @@ RUNTIME_CS_$(1) := \
               rt/rust_scheduler.cpp \
               rt/rust_task.cpp \
               rt/rust_task_list.cpp \
-              rt/rust_chan.cpp \
               rt/rust_port.cpp \
               rt/rust_upcall.cpp \
               rt/rust_log.cpp \
@@ -78,7 +77,6 @@ RUNTIME_HDR_$(1) := rt/globals.h \
                rt/rust_gc.h \
                rt/rust_internal.h \
                rt/rust_util.h \
-               rt/rust_chan.h \
                rt/rust_env.h \
                rt/rust_obstack.h \
                rt/rust_unwind.h \

src/rt/rust_builtin.cpp

@@ -463,20 +463,15 @@ new_port(size_t unit_sz) {
     rust_task *task = rust_scheduler::get_task();
     LOG(task, comm, "new_port(task=0x%" PRIxPTR " (%s), unit_sz=%d)",
         (uintptr_t) task, task->name, unit_sz);
-    // take a reference on behalf of the port
-    task->ref();
+    // port starts with refcount == 1
     return new (task->kernel, "rust_port") rust_port(task, unit_sz);
 }
 
 extern "C" CDECL void
 del_port(rust_port *port) {
     rust_task *task = rust_scheduler::get_task();
     LOG(task, comm, "del_port(0x%" PRIxPTR ")", (uintptr_t) port);
-    I(task->sched, !port->ref_count);
-    delete port;
-
-    // FIXME: this should happen in the port.
-    task->deref();
+    port->deref();
 }
 
 extern "C" CDECL rust_port_id
@@ -486,7 +481,6 @@ get_port_id(rust_port *port) {
 
 extern "C" CDECL
 void drop_port(rust_port *port) {
-    port->ref_count--;
 }
 
 extern "C" CDECL void
@@ -497,10 +491,11 @@ chan_id_send(type_desc *t, rust_task_id target_task_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);
+        target_task->deref();
         if(port) {
             port->send(sptr);
+            port->deref();
         }
-        target_task->deref();
     }
 }
 

src/rt/rust_chan.cpp

@@ -55,7 +55,6 @@ struct rust_scheduler;
 struct rust_task;
 class rust_log;
 class rust_port;
-class rust_chan;
 class rust_kernel;
 class rust_crate_cache;
 
@@ -282,7 +281,6 @@ struct type_desc {
 
 #include "circular_buffer.h"
 #include "rust_task.h"
-#include "rust_chan.h"
 #include "rust_port.h"
 #include "memory.h"
 

src/rt/rust_chan.h

@@ -1,61 +1,47 @@
 #include "rust_internal.h"
 #include "rust_port.h"
-#include "rust_chan.h"
 
 
 rust_port::rust_port(rust_task *task, size_t unit_sz)
     : ref_count(1), kernel(task->kernel), task(task),
-      unit_sz(unit_sz) {
+      unit_sz(unit_sz), buffer(kernel, unit_sz) {
 
     LOG(task, comm,
         "new rust_port(task=0x%" PRIxPTR ", unit_sz=%d) -> port=0x%"
         PRIxPTR, (uintptr_t)task, unit_sz, (uintptr_t)this);
 
+    task->ref();
     id = task->register_port(this);
-    remote_chan = new (task->kernel, "rust_chan")
-        rust_chan(task->kernel, this, unit_sz);
-    remote_chan->ref();
-    remote_chan->port = this;
 }
 
 rust_port::~rust_port() {
     LOG(task, comm, "~rust_port 0x%" PRIxPTR, (uintptr_t) this);
 
-    {
-        scoped_lock with(lock);
-        remote_chan->port = NULL;
-        remote_chan->deref();
-        remote_chan = NULL;
-    }
-
     task->release_port(id);
+    task->deref();
 }
 
 void rust_port::send(void *sptr) {
-    if (!remote_chan->is_associated()) {
-        W(kernel, remote_chan->is_associated(),
-          "rust_chan::transmit with no associated port.");
-        return;
-    }
-
+    // FIXME: Is this lock really necessary? Why do we send with the lock
+    // but not receive with the lock?
     scoped_lock with(lock);
 
-    remote_chan->buffer.enqueue(sptr);
+    buffer.enqueue(sptr);
 
-    A(kernel, !remote_chan->buffer.is_empty(),
+    A(kernel, !buffer.is_empty(),
       "rust_chan::transmit with nothing to send.");
 
     if (task->blocked_on(this)) {
         KLOG(kernel, comm, "dequeued in rendezvous_ptr");
-        remote_chan->buffer.dequeue(task->rendezvous_ptr);
+        buffer.dequeue(task->rendezvous_ptr);
         task->rendezvous_ptr = 0;
         task->wakeup(this);
     }
 }
 
 bool rust_port::receive(void *dptr) {
-    if (remote_chan->buffer.is_empty() == false) {
-        remote_chan->buffer.dequeue(dptr);
+    if (buffer.is_empty() == false) {
+        buffer.dequeue(dptr);
         LOG(task, comm, "<=== read data ===");
         return true;
     }
@@ -64,9 +50,8 @@ bool rust_port::receive(void *dptr) {
 
 void rust_port::log_state() {
     LOG(task, comm,
-        "\tchan: 0x%" PRIxPTR ", size: %d",
-        remote_chan,
-        remote_chan->buffer.size());
+        "port size: %d",
+        buffer.size());
 }
 
 //

src/rt/rust_internal.h

@@ -3,14 +3,14 @@
 
 class rust_port : public kernel_owned<rust_port>, public rust_cond {
 public:
-    RUST_REFCOUNTED(rust_port);
+    RUST_ATOMIC_REFCOUNT();
 
     rust_port_id id;
 
     rust_kernel *kernel;
     rust_task *task;
-    rust_chan *remote_chan;
     size_t unit_sz;
+    circular_buffer buffer;
 
     lock_and_signal lock;
 

src/rt/rust_port.cpp

@@ -122,21 +122,20 @@ rust_task::rust_task(rust_scheduler *sched, rust_task_list *state,
 rust_task::~rust_task()
 {
     I(sched, !sched->lock.lock_held_by_current_thread());
+    I(sched, port_table.is_empty());
     DLOG(sched, task, "~rust_task %s @0x%" PRIxPTR ", refcnt=%d",
          name, (uintptr_t)this, ref_count);
 
     if(user.notify_enabled) {
-        rust_chan *target =
-            get_chan_by_handle(&user.notify_chan);
+        rust_port *target =
+            get_port_by_chan_handle(&user.notify_chan);
         if(target) {
             task_notification msg;
             msg.id = user.id;
             msg.result = failed ? tr_failure : tr_success;
 
-            if (target->is_associated()) {
-                target->port->send(&msg);
-                target->deref();
-            }
+            target->send(&msg);
+            target->deref();
         }
     }
 
@@ -552,16 +551,18 @@ rust_port *rust_task::get_port_by_id(rust_port_id id) {
     scoped_lock with(lock);
     rust_port *port = NULL;
     port_table.get(id, &port);
+    if (port) {
+        port->ref();
+    }
     return port;
 }
 
-rust_chan *rust_task::get_chan_by_handle(chan_handle *handle) {
+rust_port *rust_task::get_port_by_chan_handle(chan_handle *handle) {
     rust_task *target_task = kernel->get_task_by_id(handle->task);
     if(target_task) {
         rust_port *port = target_task->get_port_by_id(handle->port);
         target_task->deref();
-        port->remote_chan->ref();
-        return port->remote_chan;
+        return port;
     }
     return NULL;
 }

src/rt/rust_port.h

@@ -209,7 +209,7 @@ rust_task : public kernel_owned<rust_task>, rust_cond
     // not at all safe.
     intptr_t get_ref_count() const { return ref_count; }
 
-    rust_chan *get_chan_by_handle(chan_handle *handle);
+    rust_port *get_port_by_chan_handle(chan_handle *handle);
 
     // FIXME: These functions only exist to get the tasking system off the
     // ground. We should never be migrating shared boxes between tasks.