Port ID-based channels.

4 files changed, 34 insertions, 2 deletions

diff --git a/src/rt/rust_builtin.cpp b/src/rt/rust_builtin.cpp
index 4b8af7a67b6..a7d6a902231 100644
--- a/src/rt/rust_builtin.cpp
+++ b/src/rt/rust_builtin.cpp
@@ -702,6 +702,11 @@ unpin_task(rust_task *task) {
     task->unpin();
 }
 
+extern "C" CDECL rust_task_id
+get_task_id(rust_task *task) {
+    return task->id;
+}
+
 extern "C" CDECL rust_chan *
 clone_chan(rust_task *task, rust_chan *chan) {
     return chan->clone(task);
@@ -738,6 +743,11 @@ del_port(rust_task *task, rust_port *port) {
     task->deref();
 }
 
+extern "C" CDECL rust_port_id
+get_port_id(rust_task *task, rust_port *port) {
+    return port->id;
+}
+
 extern "C" CDECL rust_chan*
 new_chan(rust_task *task, rust_port *port) {
     rust_scheduler *sched = task->sched;
@@ -776,6 +786,19 @@ chan_send(rust_task *task, rust_chan *chan, void *sptr) {
 }
 
 extern "C" CDECL void
+chan_id_send(rust_task *task, type_desc *t, rust_task_id target_task_id,
+             rust_port_id target_port_id, void *sptr) {
+    // FIXME: make sure this is thread-safe
+    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->remote_chan->send(sptr);
+        }
+    }
+}
+
+extern "C" CDECL void
 port_recv(rust_task *task, uintptr_t *dptr, rust_port *port) {
     {
         scoped_lock with(port->lock);
diff --git a/src/rt/rust_port.cpp b/src/rt/rust_port.cpp
index 448babfbdee..e0d3a347b98 100644
--- a/src/rt/rust_port.cpp
+++ b/src/rt/rust_port.cpp
@@ -1,6 +1,9 @@
 #include "rust_internal.h"
 #include "rust_port.h"
 
+extern "C" CDECL rust_chan*
+new_chan(rust_task *task, rust_port *port);
+
 rust_port::rust_port(rust_task *task, size_t unit_sz)
     : ref_count(1), kernel(task->kernel), task(task),
       unit_sz(unit_sz), writers(task), chans(task) {
@@ -10,6 +13,7 @@ rust_port::rust_port(rust_task *task, size_t unit_sz)
         PRIxPTR, (uintptr_t)task, unit_sz, (uintptr_t)this);
 
     id = task->register_port(this);
+    remote_chan = new_chan(task, this);
 }
 
 rust_port::~rust_port() {
@@ -22,6 +26,9 @@ rust_port::~rust_port() {
         chan->disassociate();
     }
 
+    remote_chan->deref();
+    remote_chan = NULL;
+
     task->release_port(id);
 }
 
diff --git a/src/rt/rust_port.h b/src/rt/rust_port.h
index 9dac42244ef..8e7b215c154 100644
--- a/src/rt/rust_port.h
+++ b/src/rt/rust_port.h
@@ -5,12 +5,11 @@ class rust_port : public kernel_owned<rust_port>, public rust_cond {
 public:
     RUST_REFCOUNTED(rust_port);
 
-private:
     rust_port_id id;
 
-public:
     rust_kernel *kernel;
     rust_task *task;
+    rust_chan *remote_chan;
     size_t unit_sz;
     ptr_vec<rust_token> writers;
     ptr_vec<rust_chan> chans;
diff --git a/src/rt/rustrt.def.in b/src/rt/rustrt.def.in
index a2174e39168..09f4097b9c1 100644
--- a/src/rt/rustrt.def.in
+++ b/src/rt/rustrt.def.in
@@ -9,6 +9,7 @@ aio_serve
 aio_stop
 aio_writedata
 align_of
+chan_id_send
 chan_send
 check_claims
 clone_chan
@@ -25,6 +26,8 @@ debug_tydesc
 do_gc
 drop_chan
 drop_port
+get_port_id
+get_task_id
 get_time
 hack_allow_leaks
 ivec_copy_from_buf