Remember to wake up blocked task on sender terminate.

2 files changed, 40 insertions, 1 deletions

diff --git a/src/libcore/pipes.rs b/src/libcore/pipes.rs
index 6f61b12ab40..441f323d7bf 100644
--- a/src/libcore/pipes.rs
+++ b/src/libcore/pipes.rs
@@ -156,7 +156,16 @@ fn peek<T: send>(p: recv_packet<T>) -> bool {
 fn sender_terminate<T: send>(p: *packet<T>) {
     let p = unsafe { uniquify(p) };
     alt swap_state_rel(p.header.state, terminated) {
-      empty | blocked {
+      empty {
+        // The receiver will eventually clean up.
+        unsafe { forget(p) }
+      }
+      blocked {
+        // wake up the target
+        let target = p.header.blocked_task.get();
+        rustrt::task_signal_event(target,
+                                  ptr::addr_of(p.header) as *libc::c_void);
+
         // The receiver will eventually clean up.
         unsafe { forget(p) }
       }
diff --git a/src/test/run-pass/pipe-detect-term.rs b/src/test/run-pass/pipe-detect-term.rs
new file mode 100644
index 00000000000..6a80fd46f34
--- /dev/null
+++ b/src/test/run-pass/pipe-detect-term.rs
@@ -0,0 +1,30 @@
+// Make sure that we can detect when one end of the pipe is closed.
+
+// xfail-pretty
+
+use std;
+import std::timer::sleep;
+import std::uv;
+
+import pipes::{recv};
+
+proto! oneshot {
+    waiting:send {
+        signal -> signaled
+    }
+
+    signaled:send { }
+}
+
+fn main() {
+    let iotask = uv::global_loop::get();
+    
+    let c = pipes::spawn_service(oneshot::init, |p| { 
+        alt recv(p) {
+          some(*) { fail }
+          none { }
+        }
+    });
+
+    sleep(iotask, 1000);
+}