auto merge of #7109 : bblum/rust/rwlocks, r=brson

r? @brson

links to issues: #7065 the race that's fixed; #7066 the perf improvement I added. There are also some minor cleanup commits here.

To measure the performance improvement from replacing the exclusive with an atomic uint, I edited the ```msgsend-ring-rw-arcs``` bench test to do a ```write_downgrade``` instead of just a ```write```, so that it stressed the code paths that accessed ```read_count```. (At first I was still using ```write``` and saw no performance difference whatsoever, whoooops.)

The bench test measures how long it takes to send 1,000,000 messages by using rwarcs to emulate pipes. I also measured the performance difference imposed by the fix to the ```access_lock``` race (which involves taking an extra semaphore in the ```cond.wait()``` path). The net result is that fixing the race imposes a 4% to 5% slowdown, but doing the atomic uint optimization gives a 6% to 8% speedup.

Note that this speedup will be most visible in read- or downgrade-heavy workloads. If an RWARC's only users are writers, the optimization doesn't matter. All the same, I think this more than justifies the extra complexity I mentioned in #7066.

The raw numbers are:
```
with xadd read count
        before write_cond fix
                4.18 to 4.26 us/message
        with write_cond fix
                4.35 to 4.39 us/message
                
with exclusive read count
        before write_cond fix
                4.41 to 4.47 us/message
        with write_cond fix
                4.65 to 4.76 us/message
```

src/libextra/arc.rs

@@ -281,7 +281,6 @@ struct RWARCInner<T> { lock: RWlock, failed: bool, data: T }
 #[mutable]
 struct RWARC<T> {
     x: UnsafeAtomicRcBox<RWARCInner<T>>,
-    cant_nest: ()
 }
 
 /// Create a reader/writer ARC with the supplied data.
@@ -299,15 +298,14 @@ pub fn rw_arc_with_condvars<T:Const + Owned>(
     let data =
         RWARCInner { lock: rwlock_with_condvars(num_condvars),
                      failed: false, data: user_data };
-    RWARC { x: UnsafeAtomicRcBox::new(data), cant_nest: () }
+    RWARC { x: UnsafeAtomicRcBox::new(data), }
 }
 
 impl<T:Const + Owned> RWARC<T> {
     /// Duplicate a rwlock-protected ARC, as arc::clone.
     pub fn clone(&self) -> RWARC<T> {
         RWARC {
             x: self.x.clone(),
-            cant_nest: (),
         }
     }
 
@@ -382,12 +380,12 @@ impl<T:Const + Owned> RWARC<T> {
      * # Example
      *
      * ~~~ {.rust}
-     * do arc.write_downgrade |write_mode| {
-     *     do (&write_mode).write_cond |state, condvar| {
+     * do arc.write_downgrade |mut write_token| {
+     *     do write_token.write_cond |state, condvar| {
      *         ... exclusive access with mutable state ...
      *     }
-     *     let read_mode = arc.downgrade(write_mode);
-     *     do (&read_mode).read |state| {
+     *     let read_token = arc.downgrade(write_token);
+     *     do read_token.read |state| {
      *         ... shared access with immutable state ...
      *     }
      * }
@@ -815,4 +813,66 @@ mod tests {
 
         wp2.recv(); // complete handshake with writer
     }
+    #[cfg(test)]
+    fn test_rw_write_cond_downgrade_read_race_helper() {
+        // Tests that when a downgrader hands off the "reader cloud" lock
+        // because of a contending reader, a writer can't race to get it
+        // instead, which would result in readers_and_writers. This tests
+        // the sync module rather than this one, but it's here because an
+        // rwarc gives us extra shared state to help check for the race.
+        // If you want to see this test fail, go to sync.rs and replace the
+        // line in RWlock::write_cond() that looks like:
+        //     "blk(&Condvar { order: opt_lock, ..*cond })"
+        // with just "blk(cond)".
+        let x = ~RWARC(true);
+        let (wp, wc) = comm::stream();
+
+        // writer task
+        let xw = (*x).clone();
+        do task::spawn {
+            do xw.write_cond |state, c| {
+                wc.send(()); // tell downgrader it's ok to go
+                c.wait();
+                // The core of the test is here: the condvar reacquire path
+                // must involve order_lock, so that it cannot race with a reader
+                // trying to receive the "reader cloud lock hand-off".
+                *state = false;
+            }
+        }
+
+        wp.recv(); // wait for writer to get in
+
+        do x.write_downgrade |mut write_mode| {
+            do write_mode.write_cond |state, c| {
+                assert!(*state);
+                // make writer contend in the cond-reacquire path
+                c.signal();
+            }
+            // make a reader task to trigger the "reader cloud lock" handoff
+            let xr = (*x).clone();
+            let (rp, rc) = comm::stream();
+            do task::spawn {
+                rc.send(());
+                do xr.read |_state| { }
+            }
+            rp.recv(); // wait for reader task to exist
+
+            let read_mode = x.downgrade(write_mode);
+            do read_mode.read |state| {
+                // if writer mistakenly got in, make sure it mutates state
+                // before we assert on it
+                for 5.times { task::yield(); }
+                // make sure writer didn't get in.
+                assert!(*state);
+            }
+        }
+    }
+    #[test]
+    fn test_rw_write_cond_downgrade_read_race() {
+        // Ideally the above test case would have yield statements in it that
+        // helped to expose the race nearly 100% of the time... but adding
+        // yields in the intuitively-right locations made it even less likely,
+        // and I wasn't sure why :( . This is a mediocre "next best" option.
+        for 8.times { test_rw_write_cond_downgrade_read_race_helper() }
+    }
 }