rt: Make rust_port_detach less contentious

It still utterly dominates some benchmarks with busy waiting, but at least it
doesn't create lock contention while doing so.

src/rt/rust_builtin.cpp

@@ -492,7 +492,7 @@ extern "C" CDECL void
 del_port(rust_port *port) {
     rust_task *task = rust_task_thread::get_task();
     LOG(task, comm, "del_port(0x%" PRIxPTR ")", (uintptr_t) port);
-    A(task->thread, port->ref_count == 1, "Expected port ref_count == 1");
+    A(task->thread, port->get_ref_count() == 1, "Expected port ref_count == 1");
     port->deref();
 }
 
@@ -522,7 +522,6 @@ chan_id_send(type_desc *t, rust_task_id target_task_id,
         rust_port *port = target_task->get_port_by_id(target_port_id);
         if(port) {
             port->send(sptr);
-            scoped_lock with(target_task->port_lock);
             port->deref();
             sent = true;
         } else {

src/rt/rust_internal.h

@@ -104,15 +104,20 @@ static size_t const BUF_BYTES = 2048;
   void ref() { ++ref_count; } \
   void deref() { if (--ref_count == 0) { dtor; } }
 
-#define RUST_ATOMIC_REFCOUNT()                                            \
-private:                                                                  \
-   intptr_t ref_count;                                                    \
-public:                                                                   \
-   void ref() {                                                           \
-       intptr_t old = sync::increment(ref_count);                         \
-       assert(old > 0);                                                   \
-   }                                                                      \
-   void deref() { if(0 == sync::decrement(ref_count)) { delete_this(); } }
+#define RUST_ATOMIC_REFCOUNT()                                             \
+private:                                                                   \
+   intptr_t ref_count;                                                     \
+public:                                                                    \
+   void ref() {                                                            \
+       intptr_t old = sync::increment(ref_count);                          \
+       assert(old > 0);                                                    \
+   }                                                                       \
+   void deref() { if(0 == sync::decrement(ref_count)) { delete_this(); } } \
+   intptr_t get_ref_count() {                                              \
+      sync::increment(ref_count);                                          \
+      intptr_t current = sync::decrement(ref_count);                       \
+      return current;                                                      \
+   }
 
 template <typename T> struct task_owned {
     inline void *operator new(size_t size, rust_task *task, const char *tag);

src/rt/rust_port.cpp

@@ -25,8 +25,7 @@ void rust_port::detach() {
     // FIXME: Busy waiting until we're the only ref
     bool done = false;
     while (!done) {
-        scoped_lock with(task->port_lock);
-        done = ref_count == 1;
+        done = get_ref_count() == 1;
     }
 }
 

src/rt/rust_port.h

@@ -5,7 +5,7 @@
 
 class rust_port : public kernel_owned<rust_port>, public rust_cond {
 public:
-    RUST_REFCOUNTED(rust_port)
+    RUST_ATOMIC_REFCOUNT();
 
     rust_port_id id;
 
@@ -18,6 +18,8 @@ public:
 
     rust_port(rust_task *task, size_t unit_sz);
     ~rust_port();
+    void delete_this() { delete this; }
+
     void log_state();
     void send(void *sptr);
     void receive(void *dptr, uintptr_t *yield);

src/rt/rust_port_selector.cpp

@@ -69,7 +69,6 @@ void
 rust_port_selector::msg_sent_on(rust_port *port) {
     rust_task *task = port->task;
 
-    I(task->thread, !task->port_lock.lock_held_by_current_thread());
     I(task->thread, !port->lock.lock_held_by_current_thread());
     I(task->thread, !rendezvous_lock.lock_held_by_current_thread());
 

src/rt/rust_task.cpp

@@ -516,7 +516,6 @@ rust_task::notify(bool success) {
                 msg.result = !success ? tr_failure : tr_success;
 
                 target_port->send(&msg);
-                scoped_lock with(target_task->port_lock);
                 target_port->deref();
             }
             target_task->deref();

src/rt/rust_task.h

@@ -89,10 +89,6 @@ rust_task : public kernel_owned<rust_task>, rust_cond
 
     bool propagate_failure;
 
-    // Protects port_table
-    lock_and_signal port_lock;
-    hash_map<rust_port_id, rust_port *> port_table;
-
     rust_obstack dynastack;
 
     uint32_t cc_counter;
@@ -104,6 +100,10 @@ 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;