---

yaml
---
r: 108250
b: refs/heads/dist-snap
c: 065e121
h: refs/heads/master
v: v3

Author: alexcrichton

[refs]

@@ -6,7 +6,7 @@ refs/heads/try: f64fdf524a434f0e5cd0bc91d09c144723f3c90d
 refs/tags/release-0.1: 1f5c5126e96c79d22cb7862f75304136e204f105
 refs/heads/ndm: f3868061cd7988080c30d6d5bf352a5a5fe2460b
 refs/heads/try2: 147ecfdd8221e4a4d4e090486829a06da1e0ca3c
-refs/heads/dist-snap: 411a01feb302dd64b669658562f2052decc80f19
+refs/heads/dist-snap: 065e121fc2675631e27b761d2ad5df3ee0c95976
 refs/tags/release-0.2: c870d2dffb391e14efb05aa27898f1f6333a9596
 refs/tags/release-0.3: b5f0d0f648d9a6153664837026ba1be43d3e2503
 refs/heads/try3: 9387340aab40a73e8424c48fd42f0c521a4875c0

branches/dist-snap/src/libstd/comm/select.rs

@@ -151,6 +151,11 @@ impl Select {
     /// event could either be that data is available or the corresponding
     /// channel has been closed.
     pub fn wait(&self) -> uint {
+        self.wait2(false)
+    }
+
+    /// Helper method for skipping the preflight checks during testing
+    fn wait2(&self, do_preflight_checks: bool) -> uint {
         // Note that this is currently an inefficient implementation. We in
         // theory have knowledge about all ports in the set ahead of time, so
         // this method shouldn't really have to iterate over all of them yet
@@ -175,7 +180,7 @@ impl Select {
             let mut amt = 0;
             for p in self.iter() {
                 amt += 1;
-                if (*p).packet.can_recv() {
+                if do_preflight_checks && (*p).packet.can_recv() {
                     return (*p).id;
                 }
             }
@@ -507,7 +512,7 @@ mod test {
         let (p2, c2) = Chan::<()>::new();
         let (p, c) = Chan::new();
         spawn(proc() {
-            let mut s = Select::new();
+            let s = Select::new();
             let mut h1 = s.handle(&p1);
             let mut h2 = s.handle(&p2);
             unsafe { h2.add(); }
@@ -521,4 +526,91 @@ mod test {
         c2.send(());
         p.recv();
     })
+
+    test!(fn preflight1() {
+        let (p, c) = Chan::new();
+        c.send(());
+        select!(
+            () = p.recv() => {},
+        )
+    })
+
+    test!(fn preflight2() {
+        let (p, c) = Chan::new();
+        c.send(());
+        c.send(());
+        select!(
+            () = p.recv() => {},
+        )
+    })
+
+    test!(fn preflight3() {
+        let (p, c) = Chan::new();
+        drop(c.clone());
+        c.send(());
+        select!(
+            () = p.recv() => {},
+        )
+    })
+
+    test!(fn preflight4() {
+        let (p, c) = Chan::new();
+        c.send(());
+        let s = Select::new();
+        let mut h = s.handle(&p);
+        unsafe { h.add(); }
+        assert_eq!(s.wait2(false), h.id);
+    })
+
+    test!(fn preflight5() {
+        let (p, c) = Chan::new();
+        c.send(());
+        c.send(());
+        let s = Select::new();
+        let mut h = s.handle(&p);
+        unsafe { h.add(); }
+        assert_eq!(s.wait2(false), h.id);
+    })
+
+    test!(fn preflight6() {
+        let (p, c) = Chan::new();
+        drop(c.clone());
+        c.send(());
+        let s = Select::new();
+        let mut h = s.handle(&p);
+        unsafe { h.add(); }
+        assert_eq!(s.wait2(false), h.id);
+    })
+
+    test!(fn preflight7() {
+        let (p, c) = Chan::<()>::new();
+        drop(c);
+        let s = Select::new();
+        let mut h = s.handle(&p);
+        unsafe { h.add(); }
+        assert_eq!(s.wait2(false), h.id);
+    })
+
+    test!(fn preflight8() {
+        let (p, c) = Chan::new();
+        c.send(());
+        drop(c);
+        p.recv();
+        let s = Select::new();
+        let mut h = s.handle(&p);
+        unsafe { h.add(); }
+        assert_eq!(s.wait2(false), h.id);
+    })
+
+    test!(fn preflight9() {
+        let (p, c) = Chan::new();
+        drop(c.clone());
+        c.send(());
+        drop(c);
+        p.recv();
+        let s = Select::new();
+        let mut h = s.handle(&p);
+        unsafe { h.add(); }
+        assert_eq!(s.wait2(false), h.id);
+    })
 }

branches/dist-snap/src/libstd/comm/shared.rs

@@ -398,6 +398,17 @@ impl<T: Send> Packet<T> {
         cnt == DISCONNECTED || cnt - self.steals > 0
     }
 
+    // increment the count on the channel (used for selection)
+    fn bump(&mut self, amt: int) -> int {
+        match self.cnt.fetch_add(amt, atomics::SeqCst) {
+            DISCONNECTED => {
+                self.cnt.store(DISCONNECTED, atomics::SeqCst);
+                DISCONNECTED
+            }
+            n => n
+        }
+    }
+
     // Inserts the blocked task for selection on this port, returning it back if
     // the port already has data on it.
     //
@@ -408,8 +419,8 @@ impl<T: Send> Packet<T> {
         match self.decrement(task) {
             Ok(()) => Ok(()),
             Err(task) => {
-                let prev = self.cnt.fetch_add(1, atomics::SeqCst);
-                assert!(prev >= 0);
+                let prev = self.bump(1);
+                assert!(prev == DISCONNECTED || prev >= 0);
                 return Err(task);
             }
         }
@@ -440,11 +451,10 @@ impl<T: Send> Packet<T> {
             let cnt = self.cnt.load(atomics::SeqCst);
             if cnt < 0 && cnt != DISCONNECTED {-cnt} else {0}
         };
-        let prev = self.cnt.fetch_add(steals + 1, atomics::SeqCst);
+        let prev = self.bump(steals + 1);
 
         if prev == DISCONNECTED {
             assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
-            self.cnt.store(DISCONNECTED, atomics::SeqCst);
             true
         } else {
             let cur = prev + steals + 1;

branches/dist-snap/src/libstd/comm/stream.rs

@@ -333,6 +333,17 @@ impl<T: Send> Packet<T> {
         }
     }
 
+    // increment the count on the channel (used for selection)
+    fn bump(&mut self, amt: int) -> int {
+        match self.cnt.fetch_add(amt, atomics::SeqCst) {
+            DISCONNECTED => {
+                self.cnt.store(DISCONNECTED, atomics::SeqCst);
+                DISCONNECTED
+            }
+            n => n
+        }
+    }
+
     // Attempts to start selecting on this port. Like a oneshot, this can fail
     // immediately because of an upgrade.
     pub fn start_selection(&mut self, task: BlockedTask) -> SelectionResult<T> {
@@ -351,8 +362,8 @@ impl<T: Send> Packet<T> {
                 };
                 // Undo our decrement above, and we should be guaranteed that the
                 // previous value is positive because we're not going to sleep
-                let prev = self.cnt.fetch_add(1, atomics::SeqCst);
-                assert!(prev >= 0);
+                let prev = self.bump(1);
+                assert!(prev == DISCONNECTED || prev >= 0);
                 return ret;
             }
         }
@@ -384,13 +395,12 @@ impl<T: Send> Packet<T> {
         // and in the stream case we can have at most one steal, so just assume
         // that we had one steal.
         let steals = 1;
-        let prev = self.cnt.fetch_add(steals + 1, atomics::SeqCst);
+        let prev = self.bump(steals + 1);
 
         // If we were previously disconnected, then we know for sure that there
         // is no task in to_wake, so just keep going
         let has_data = if prev == DISCONNECTED {
             assert_eq!(self.to_wake.load(atomics::SeqCst), 0);
-            self.cnt.store(DISCONNECTED, atomics::SeqCst);
             true // there is data, that data is that we're disconnected
         } else {
             let cur = prev + steals + 1;