Drop enqueued elements when a port is destructed. Closes #1155

7 files changed, 39 insertions, 2 deletions

diff --git a/src/lib/comm.rs b/src/lib/comm.rs
index ca303365435..6b73f91bb68 100644
--- a/src/lib/comm.rs
+++ b/src/lib/comm.rs
@@ -45,6 +45,7 @@ native "cdecl" mod rustrt {
     fn new_port(unit_sz: uint) -> *rust_port;
     fn del_port(po: *rust_port);
     fn get_port_id(po: *rust_port) -> port_id;
+    fn rust_port_size(po: *rust_port) -> ctypes::size_t;
 }
 
 native "rust-intrinsic" mod rusti {
@@ -75,7 +76,13 @@ tag chan<uniq T> {
     chan_t(task::task, port_id);
 }
 
-resource port_ptr(po: *rustrt::rust_port) {
+resource port_ptr<uniq T>(po: *rustrt::rust_port) {
+    // Drain the port so that all the still-enqueued items get dropped
+    while rustrt::rust_port_size(po) > 0u {
+        // FIXME: For some reason if we don't assign to something here
+        // we end up with invalid reads in the drop glue.
+        let t = rusti::recv::<T>(po);
+    }
     rustrt::del_port(po);
 }
 
@@ -90,7 +97,7 @@ transmitted. If a port value is copied, both copies refer to the same port.
 
 Ports may be associated with multiple <chan>s.
 */
-tag port<uniq T> { port_t(@port_ptr); }
+tag port<uniq T> { port_t(@port_ptr<T>); }
 
 /*
 Function: send
diff --git a/src/rt/rust_builtin.cpp b/src/rt/rust_builtin.cpp
index e8109ea7a04..db9704b540b 100644
--- a/src/rt/rust_builtin.cpp
+++ b/src/rt/rust_builtin.cpp
@@ -475,6 +475,11 @@ del_port(rust_port *port) {
     port->deref();
 }
 
+extern "C" CDECL size_t
+rust_port_size(rust_port *port) {
+    return port->size();
+}
+
 extern "C" CDECL rust_port_id
 get_port_id(rust_port *port) {
     return port->id;
diff --git a/src/rt/rust_port.cpp b/src/rt/rust_port.cpp
index 5daf5e40033..ff743eeeb40 100644
--- a/src/rt/rust_port.cpp
+++ b/src/rt/rust_port.cpp
@@ -48,6 +48,11 @@ bool rust_port::receive(void *dptr) {
     return false;
 }
 
+size_t rust_port::size() {
+    scoped_lock with(lock);
+    return buffer.size();
+}
+
 void rust_port::log_state() {
     LOG(task, comm,
         "port size: %d",
diff --git a/src/rt/rust_port.h b/src/rt/rust_port.h
index a80fc262d01..40d78159ba0 100644
--- a/src/rt/rust_port.h
+++ b/src/rt/rust_port.h
@@ -19,6 +19,7 @@ public:
     void log_state();
     void send(void *sptr);
     bool receive(void *dptr);
+    size_t size();
 };
 
 //
diff --git a/src/rt/rustrt.def.in b/src/rt/rustrt.def.in
index 13a39471b5d..d891fc16904 100644
--- a/src/rt/rustrt.def.in
+++ b/src/rt/rustrt.def.in
@@ -34,6 +34,7 @@ rust_get_stdout
 rust_get_stderr
 rust_str_push
 rust_list_files
+rust_port_size
 rust_process_wait
 rust_ptr_eq
 rust_run_program
diff --git a/src/test/run-pass/task-comm-chan-cleanup.rs b/src/test/run-pass/task-comm-chan-cleanup.rs
new file mode 100644
index 00000000000..a94edb98a26
--- /dev/null
+++ b/src/test/run-pass/task-comm-chan-cleanup.rs
@@ -0,0 +1,9 @@
+// xfail-win32 Requires unwinding
+use std;
+import std::comm;
+
+fn main() {
+    let p = comm::port();
+    let c = comm::chan(p);
+    comm::send(c, "coffee");
+}
\ No newline at end of file
diff --git a/src/test/run-pass/task-comm-chan-cleanup2.rs b/src/test/run-pass/task-comm-chan-cleanup2.rs
new file mode 100644
index 00000000000..db53bdfe74d
--- /dev/null
+++ b/src/test/run-pass/task-comm-chan-cleanup2.rs
@@ -0,0 +1,9 @@
+// xfail-win32 Requires unwinding
+use std;
+import std::comm;
+
+fn main() {
+    let p = comm::port();
+    let c = comm::chan(p);
+    comm::send(c, ~"coffee");
+}
\ No newline at end of file