src/libcore/private.rs - third_party/rust - Git at Google

 // NB: transitionary, de-mode-ing.
 // tjc: Re-forbid deprecated modes once a snapshot fixes the
 // function problem
 #[forbid(deprecated_pattern)];

 #[doc(hidden)];

 use compare_and_swap = rustrt::rust_compare_and_swap_ptr;
 use task::TaskBuilder;
 use task::atomically;

 extern mod rustrt {
     #[legacy_exports];
     fn rust_task_weaken(ch: rust_port_id);
     fn rust_task_unweaken(ch: rust_port_id);

     #[rust_stack]
     fn rust_atomic_increment(p: &mut libc::intptr_t)
         -> libc::intptr_t;

     #[rust_stack]
     fn rust_atomic_decrement(p: &mut libc::intptr_t)
         -> libc::intptr_t;

     #[rust_stack]
     fn rust_compare_and_swap_ptr(address: &mut libc::uintptr_t,
                                  oldval: libc::uintptr_t,
                                  newval: libc::uintptr_t) -> bool;

     fn rust_create_little_lock() -> rust_little_lock;
     fn rust_destroy_little_lock(lock: rust_little_lock);
     fn rust_lock_little_lock(lock: rust_little_lock);
     fn rust_unlock_little_lock(lock: rust_little_lock);
 }

 #[allow(non_camel_case_types)] // runtime type
 type rust_port_id = uint;

 type GlobalPtr = *libc::uintptr_t;

 /**
  * Atomically gets a channel from a pointer to a pointer-sized memory location
  * or, if no channel exists creates and installs a new channel and sets up a
  * new task to receive from it.
  */
 pub unsafe fn chan_from_global_ptr<T: Send>(
     global: GlobalPtr,
     task_fn: fn() -> task::TaskBuilder,
     f: fn~(comm::Port<T>)
 ) -> comm::Chan<T> {

     enum Msg {
         Proceed,
         Abort
     }

     log(debug,~"ENTERING chan_from_global_ptr, before is_prob_zero check");
     let is_probably_zero = *global == 0u;
     log(debug,~"after is_prob_zero check");
     if is_probably_zero {
         log(debug,~"is probably zero...");
         // There's no global channel. We must make it

         let (setup_po, setup_ch) = do task_fn().spawn_conversation
             |move f, setup_po, setup_ch| {
             let po = comm::Port::<T>();
             let ch = comm::Chan(&po);
             comm::send(setup_ch, ch);

             // Wait to hear if we are the official instance of
             // this global task
             match comm::recv::<Msg>(setup_po) {
               Proceed => f(move po),
               Abort => ()
             }
         };

         log(debug,~"before setup recv..");
         // This is the proposed global channel
         let ch = comm::recv(setup_po);
         // 0 is our sentinal value. It is not a valid channel
         assert *ch != 0;

         // Install the channel
         log(debug,~"BEFORE COMPARE AND SWAP");
         let swapped = compare_and_swap(
             cast::reinterpret_cast(&global),
             0u, cast::reinterpret_cast(&ch));
         log(debug,fmt!("AFTER .. swapped? %?", swapped));

         if swapped {
             // Success!
             comm::send(setup_ch, Proceed);
             ch
         } else {
             // Somebody else got in before we did
             comm::send(setup_ch, Abort);
             cast::reinterpret_cast(&*global)
         }
     } else {
         log(debug, ~"global != 0");
         cast::reinterpret_cast(&*global)
     }
 }

 #[test]
 pub fn test_from_global_chan1() {

     // This is unreadable, right?

     // The global channel
     let globchan = 0;
     let globchanp = ptr::addr_of(&globchan);

     // Create the global channel, attached to a new task
     let ch = unsafe {
         do chan_from_global_ptr(globchanp, task::task) |po| {
             let ch = comm::recv(po);
             comm::send(ch, true);
             let ch = comm::recv(po);
             comm::send(ch, true);
         }
     };
     // Talk to it
     let po = comm::Port();
     comm::send(ch, comm::Chan(&po));
     assert comm::recv(po) == true;

     // This one just reuses the previous channel
     let ch = unsafe {
         do chan_from_global_ptr(globchanp, task::task) |po| {
             let ch = comm::recv(po);
             comm::send(ch, false);
         }
     };

     // Talk to the original global task
     let po = comm::Port();
     comm::send(ch, comm::Chan(&po));
     assert comm::recv(po) == true;
 }

 #[test]
 pub fn test_from_global_chan2() {

     for iter::repeat(100) {
         // The global channel
         let globchan = 0;
         let globchanp = ptr::addr_of(&globchan);

         let resultpo = comm::Port();
         let resultch = comm::Chan(&resultpo);

         // Spawn a bunch of tasks that all want to compete to
         // create the global channel
         for uint::range(0, 10) |i| {
             do task::spawn {
                 let ch = unsafe {
                     do chan_from_global_ptr(
                         globchanp, task::task) |po| {

                         for uint::range(0, 10) |_j| {
                             let ch = comm::recv(po);
                             comm::send(ch, {i});
                         }
                     }
                 };
                 let po = comm::Port();
                 comm::send(ch, comm::Chan(&po));
                 // We are The winner if our version of the
                 // task was installed
                 let winner = comm::recv(po);
                 comm::send(resultch, winner == i);
             }
         }
         // There should be only one winner
         let mut winners = 0u;
         for uint::range(0u, 10u) |_i| {
             let res = comm::recv(resultpo);
             if res { winners += 1u };
         }
         assert winners == 1u;
     }
 }

 /**
  * Convert the current task to a 'weak' task temporarily
  *
  * As a weak task it will not be counted towards the runtime's set
  * of live tasks. When there are no more outstanding live (non-weak) tasks
  * the runtime will send an exit message on the provided channel.
  *
  * This function is super-unsafe. Do not use.
  *
  * # Safety notes
  *
  * * Weak tasks must either die on their own or exit upon receipt of
  *   the exit message. Failure to do so will cause the runtime to never
  *   exit
  * * Tasks must not call `weaken_task` multiple times. This will
  *   break the kernel's accounting of live tasks.
  * * Weak tasks must not be supervised. A supervised task keeps
  *   a reference to its parent, so the parent will not die.
  */
 pub unsafe fn weaken_task(f: fn(comm::Port<()>)) {
     let po = comm::Port();
     let ch = comm::Chan(&po);
     unsafe {
         rustrt::rust_task_weaken(cast::reinterpret_cast(&ch));
     }
     let _unweaken = Unweaken(ch);
     f(po);

     struct Unweaken {
       ch: comm::Chan<()>,
       drop unsafe {
         rustrt::rust_task_unweaken(cast::reinterpret_cast(&self.ch));
       }
     }

     fn Unweaken(ch: comm::Chan<()>) -> Unweaken {
         Unweaken {
             ch: ch
         }
     }
 }

 #[test]
 pub fn test_weaken_task_then_unweaken() {
     do task::try {
         unsafe {
             do weaken_task |_po| {
             }
         }
     };
 }

 #[test]
 pub fn test_weaken_task_wait() {
     do task::spawn_unlinked {
         unsafe {
             do weaken_task |po| {
                 comm::recv(po);
             }
         }
     }
 }

 #[test]
 pub fn test_weaken_task_stress() {
     // Create a bunch of weak tasks
     for iter::repeat(100u) {
         do task::spawn {
             unsafe {
                 do weaken_task |_po| {
                 }
             }
         }
         do task::spawn_unlinked {
             unsafe {
                 do weaken_task |po| {
                     // Wait for it to tell us to die
                     comm::recv(po);
                 }
             }
         }
     }
 }

 #[test]
 #[ignore(cfg(windows))]
 pub fn test_weaken_task_fail() {
     let res = do task::try {
         unsafe {
             do weaken_task |_po| {
                 fail;
             }
         }
     };
     assert result::is_err(&res);
 }

 /****************************************************************************
  * Shared state & exclusive ARC
  ****************************************************************************/

 // An unwrapper uses this protocol to communicate with the "other" task that
 // drops the last refcount on an arc. Unfortunately this can't be a proper
 // pipe protocol because the unwrapper has to access both stages at once.
 type UnwrapProto = ~mut Option<(pipes::ChanOne<()>, pipes::PortOne<bool>)>;

 struct ArcData<T> {
     mut count:     libc::intptr_t,
     mut unwrapper: libc::uintptr_t, // either a UnwrapProto or 0
     // FIXME(#3224) should be able to make this non-option to save memory, and
     // in unwrap() use "let ~ArcData { data: result, _ } = thing" to unwrap it
     mut data:      Option<T>,
 }

 struct ArcDestruct<T> {
     mut data: *libc::c_void,
     drop unsafe {
         if self.data.is_null() {
             return; // Happens when destructing an unwrapper's handle.
         }
         do task::unkillable {
             let data: ~ArcData<T> = cast::reinterpret_cast(&self.data);
             let new_count = rustrt::rust_atomic_decrement(&mut data.count);
             assert new_count >= 0;
             if new_count == 0 {
                 // Were we really last, or should we hand off to an unwrapper?
                 // It's safe to not xchg because the unwrapper will set the
                 // unwrap lock *before* dropping his/her reference. In effect,
                 // being here means we're the only *awake* task with the data.
                 if data.unwrapper != 0 {
                     let p: UnwrapProto =
                         cast::reinterpret_cast(&data.unwrapper);
                     let (message, response) = option::swap_unwrap(p);
                     // Send 'ready' and wait for a response.
                     pipes::send_one(move message, ());
                     // Unkillable wait. Message guaranteed to come.
                     if pipes::recv_one(move response) {
                         // Other task got the data.
                         cast::forget(move data);
                     } else {
                         // Other task was killed. drop glue takes over.
                     }
                 } else {
                     // drop glue takes over.
                 }
             } else {
                 cast::forget(move data);
             }
         }
     }
 }

 fn ArcDestruct<T>(data: *libc::c_void) -> ArcDestruct<T> {
     ArcDestruct {
         data: data
     }
 }

 pub unsafe fn unwrap_shared_mutable_state<T: Send>(rc: SharedMutableState<T>)
         -> T {
     struct DeathThroes<T> {
         mut ptr:      Option<~ArcData<T>>,
         mut response: Option<pipes::ChanOne<bool>>,
         drop unsafe {
             let response = option::swap_unwrap(&mut self.response);
             // In case we get killed early, we need to tell the person who
             // tried to wake us whether they should hand-off the data to us.
             if task::failing() {
                 pipes::send_one(move response, false);
                 // Either this swap_unwrap or the one below (at "Got here")
                 // ought to run.
                 cast::forget(option::swap_unwrap(&mut self.ptr));
             } else {
                 assert self.ptr.is_none();
                 pipes::send_one(move response, true);
             }
         }
     }

     do task::unkillable {
         let ptr: ~ArcData<T> = cast::reinterpret_cast(&rc.data);
         let (c1,p1) = pipes::oneshot(); // ()
         let (c2,p2) = pipes::oneshot(); // bool
         let server: UnwrapProto = ~mut Some((move c1,move p2));
         let serverp: libc::uintptr_t = cast::transmute(move server);
         // Try to put our server end in the unwrapper slot.
         if rustrt::rust_compare_and_swap_ptr(&mut ptr.unwrapper, 0, serverp) {
             // Got in. Step 0: Tell destructor not to run. We are now it.
             rc.data = ptr::null();
             // Step 1 - drop our own reference.
             let new_count = rustrt::rust_atomic_decrement(&mut ptr.count);
             assert new_count >= 0;
             if new_count == 0 {
                 // We were the last owner. Can unwrap immediately.
                 // Also we have to free the server endpoints.
                 let _server: UnwrapProto = cast::transmute(move serverp);
                 option::swap_unwrap(&mut ptr.data)
                 // drop glue takes over.
             } else {
                 // The *next* person who sees the refcount hit 0 will wake us.
                 let end_result =
                     DeathThroes { ptr: Some(move ptr),
                                   response: Some(move c2) };
                 let mut p1 = Some(move p1); // argh
                 do task::rekillable {
                     pipes::recv_one(option::swap_unwrap(&mut p1));
                 }
                 // Got here. Back in the 'unkillable' without getting killed.
                 // Recover ownership of ptr, then take the data out.
                 let ptr = option::swap_unwrap(&mut end_result.ptr);
                 option::swap_unwrap(&mut ptr.data)
                 // drop glue takes over.
             }
         } else {
             // Somebody else was trying to unwrap. Avoid guaranteed deadlock.
             cast::forget(move ptr);
             // Also we have to free the (rejected) server endpoints.
             let _server: UnwrapProto = cast::transmute(move serverp);
             fail ~"Another task is already unwrapping this ARC!";
         }
     }
 }

 /**
  * COMPLETELY UNSAFE. Used as a primitive for the safe versions in std::arc.
  *
  * Data races between tasks can result in crashes and, with sufficient
  * cleverness, arbitrary type coercion.
  */
 pub type SharedMutableState<T: Send> = ArcDestruct<T>;

 pub unsafe fn shared_mutable_state<T: Send>(data: T) ->
         SharedMutableState<T> {
     let data = ~ArcData { count: 1, unwrapper: 0, data: Some(move data) };
     unsafe {
         let ptr = cast::transmute(move data);
         ArcDestruct(ptr)
     }
 }

 #[inline(always)]
 pub unsafe fn get_shared_mutable_state<T: Send>(rc: &a/SharedMutableState<T>)
         -> &a/mut T {
     unsafe {
         let ptr: ~ArcData<T> = cast::reinterpret_cast(&(*rc).data);
         assert ptr.count > 0;
         // Cast us back into the correct region
         let r = cast::transmute_region(option::get_ref(&ptr.data));
         cast::forget(move ptr);
         return cast::transmute_mut(r);
     }
 }
 #[inline(always)]
 pub unsafe fn get_shared_immutable_state<T: Send>(
         rc: &a/SharedMutableState<T>) -> &a/T {
     unsafe {
         let ptr: ~ArcData<T> = cast::reinterpret_cast(&(*rc).data);
         assert ptr.count > 0;
         // Cast us back into the correct region
         let r = cast::transmute_region(option::get_ref(&ptr.data));
         cast::forget(move ptr);
         return r;
     }
 }

 pub unsafe fn clone_shared_mutable_state<T: Send>(rc: &SharedMutableState<T>)
         -> SharedMutableState<T> {
     unsafe {
         let ptr: ~ArcData<T> = cast::reinterpret_cast(&(*rc).data);
         let new_count = rustrt::rust_atomic_increment(&mut ptr.count);
         assert new_count >= 2;
         cast::forget(move ptr);
     }
     ArcDestruct((*rc).data)
 }

 /****************************************************************************/

 #[allow(non_camel_case_types)] // runtime type
 type rust_little_lock = *libc::c_void;

 struct LittleLock {
     l: rust_little_lock,
     drop { rustrt::rust_destroy_little_lock(self.l); }
 }

 fn LittleLock() -> LittleLock {
     LittleLock {
         l: rustrt::rust_create_little_lock()
     }
 }

 impl LittleLock {
     #[inline(always)]
     unsafe fn lock<T>(f: fn() -> T) -> T {
         struct Unlock {
             l: rust_little_lock,
             drop { rustrt::rust_unlock_little_lock(self.l); }
         }

         fn Unlock(l: rust_little_lock) -> Unlock {
             Unlock {
                 l: l
             }
         }

         do atomically {
             rustrt::rust_lock_little_lock(self.l);
             let _r = Unlock(self.l);
             f()
         }
     }
 }

 struct ExData<T: Send> { lock: LittleLock, mut failed: bool, mut data: T, }
 /**
  * An arc over mutable data that is protected by a lock. For library use only.
  */
 pub struct Exclusive<T: Send> { x: SharedMutableState<ExData<T>> }

 pub fn exclusive<T:Send >(user_data: T) -> Exclusive<T> {
     let data = ExData {
         lock: LittleLock(), mut failed: false, mut data: user_data
     };
     Exclusive { x: unsafe { shared_mutable_state(move data) } }
 }

 impl<T: Send> Exclusive<T> {
     // Duplicate an exclusive ARC, as std::arc::clone.
     fn clone() -> Exclusive<T> {
         Exclusive { x: unsafe { clone_shared_mutable_state(&self.x) } }
     }

     // Exactly like std::arc::mutex_arc,access(), but with the little_lock
     // instead of a proper mutex. Same reason for being unsafe.
     //
     // Currently, scheduling operations (i.e., yielding, receiving on a pipe,
     // accessing the provided condition variable) are prohibited while inside
     // the exclusive. Supporting that is a work in progress.
     #[inline(always)]
     unsafe fn with<U>(f: fn(x: &mut T) -> U) -> U {
         let rec = unsafe { get_shared_mutable_state(&self.x) };
         do rec.lock.lock {
             if rec.failed {
                 fail ~"Poisoned exclusive - another task failed inside!";
             }
             rec.failed = true;
             let result = f(&mut rec.data);
             rec.failed = false;
             move result
         }
     }

     #[inline(always)]
     unsafe fn with_imm<U>(f: fn(x: &T) -> U) -> U {
         do self.with |x| {
             f(cast::transmute_immut(x))
         }
     }
 }

 // FIXME(#3724) make this a by-move method on the exclusive
 pub fn unwrap_exclusive<T: Send>(arc: Exclusive<T>) -> T {
     let Exclusive { x: x } <- arc;
     let inner = unsafe { unwrap_shared_mutable_state(move x) };
     let ExData { data: data, _ } <- inner;
     move data
 }

 #[cfg(test)]
 pub mod tests {
     #[test]
     pub fn exclusive_arc() {
         let mut futures = ~[];

         let num_tasks = 10u;
         let count = 10u;

         let total = exclusive(~mut 0u);

         for uint::range(0u, num_tasks) |_i| {
             let total = total.clone();
             futures.push(future::spawn(|| {
                 for uint::range(0u, count) |_i| {
                     do total.with |count| {
                         **count += 1u;
                     }
                 }
             }));
         };

         for futures.each |f| { f.get() }

         do total.with |total| {
             assert **total == num_tasks * count
         };
     }

     #[test] #[should_fail] #[ignore(cfg(windows))]
     pub fn exclusive_poison() {
         // Tests that if one task fails inside of an exclusive, subsequent
         // accesses will also fail.
         let x = exclusive(1);
         let x2 = x.clone();
         do task::try {
             do x2.with |one| {
                 assert *one == 2;
             }
         };
         do x.with |one| {
             assert *one == 1;
         }
     }

     #[test]
     pub fn exclusive_unwrap_basic() {
         let x = exclusive(~~"hello");
         assert unwrap_exclusive(x) == ~~"hello";
     }

     #[test]
     pub fn exclusive_unwrap_contended() {
         let x = exclusive(~~"hello");
         let x2 = ~mut Some(x.clone());
         do task::spawn {
             let x2 = option::swap_unwrap(x2);
             do x2.with |_hello| { }
             task::yield();
         }
         assert unwrap_exclusive(x) == ~~"hello";

         // Now try the same thing, but with the child task blocking.
         let x = exclusive(~~"hello");
         let x2 = ~mut Some(x.clone());
         let mut res = None;
         do task::task().future_result(|+r| res = Some(r)).spawn {
             let x2 = option::swap_unwrap(x2);
             assert unwrap_exclusive(x2) == ~~"hello";
         }
         // Have to get rid of our reference before blocking.
         { let _x = move x; } // FIXME(#3161) util::ignore doesn't work here
         let res = option::swap_unwrap(&mut res);
         future::get(&res);
     }

     #[test] #[should_fail] #[ignore(cfg(windows))]
     pub fn exclusive_unwrap_conflict() {
         let x = exclusive(~~"hello");
         let x2 = ~mut Some(x.clone());
         let mut res = None;
         do task::task().future_result(|+r| res = Some(r)).spawn {
             let x2 = option::swap_unwrap(x2);
             assert unwrap_exclusive(x2) == ~~"hello";
         }
         assert unwrap_exclusive(x) == ~~"hello";
         let res = option::swap_unwrap(&mut res);
         future::get(&res);
     }

     #[test] #[ignore(cfg(windows))]
     pub fn exclusive_unwrap_deadlock() {
         // This is not guaranteed to get to the deadlock before being killed,
         // but it will show up sometimes, and if the deadlock were not there,
         // the test would nondeterministically fail.
         let result = do task::try {
             // a task that has two references to the same exclusive will
             // deadlock when it unwraps. nothing to be done about that.
             let x = exclusive(~~"hello");
             let x2 = x.clone();
             do task::spawn {
                 for 10.times { task::yield(); } // try to let the unwrapper go
                 fail; // punt it awake from its deadlock
             }
             let _z = unwrap_exclusive(x);
             do x2.with |_hello| { }
         };
         assert result.is_err();
     }
 }
	// NB: transitionary, de-mode-ing.
	// tjc: Re-forbid deprecated modes once a snapshot fixes the
	// function problem
	#[forbid(deprecated_pattern)];

	#[doc(hidden)];

	use compare_and_swap = rustrt::rust_compare_and_swap_ptr;
	use task::TaskBuilder;
	use task::atomically;

	extern mod rustrt {
	#[legacy_exports];
	fn rust_task_weaken(ch: rust_port_id);
	fn rust_task_unweaken(ch: rust_port_id);

	#[rust_stack]
	fn rust_atomic_increment(p: &mut libc::intptr_t)
	-> libc::intptr_t;

	#[rust_stack]
	fn rust_atomic_decrement(p: &mut libc::intptr_t)
	-> libc::intptr_t;

	#[rust_stack]
	fn rust_compare_and_swap_ptr(address: &mut libc::uintptr_t,
	oldval: libc::uintptr_t,
	newval: libc::uintptr_t) -> bool;

	fn rust_create_little_lock() -> rust_little_lock;
	fn rust_destroy_little_lock(lock: rust_little_lock);
	fn rust_lock_little_lock(lock: rust_little_lock);
	fn rust_unlock_little_lock(lock: rust_little_lock);
	}

	#[allow(non_camel_case_types)] // runtime type
	type rust_port_id = uint;

	type GlobalPtr = *libc::uintptr_t;

	/**
	* Atomically gets a channel from a pointer to a pointer-sized memory location
	* or, if no channel exists creates and installs a new channel and sets up a
	* new task to receive from it.
	*/
	pub unsafe fn chan_from_global_ptr<T: Send>(
	global: GlobalPtr,
	task_fn: fn() -> task::TaskBuilder,
	f: fn~(comm::Port<T>)
	) -> comm::Chan<T> {

	enum Msg {
	Proceed,
	Abort
	}

	log(debug,~"ENTERING chan_from_global_ptr, before is_prob_zero check");
	let is_probably_zero = *global == 0u;
	log(debug,~"after is_prob_zero check");
	if is_probably_zero {
	log(debug,~"is probably zero...");
	// There's no global channel. We must make it

	let (setup_po, setup_ch) = do task_fn().spawn_conversation
	\|move f, setup_po, setup_ch\| {
	let po = comm::Port::<T>();
	let ch = comm::Chan(&po);
	comm::send(setup_ch, ch);

	// Wait to hear if we are the official instance of
	// this global task
	match comm::recv::<Msg>(setup_po) {
	Proceed => f(move po),
	Abort => ()
	}
	};

	log(debug,~"before setup recv..");
	// This is the proposed global channel
	let ch = comm::recv(setup_po);
	// 0 is our sentinal value. It is not a valid channel
	assert *ch != 0;

	// Install the channel
	log(debug,~"BEFORE COMPARE AND SWAP");
	let swapped = compare_and_swap(
	cast::reinterpret_cast(&global),
	0u, cast::reinterpret_cast(&ch));
	log(debug,fmt!("AFTER .. swapped? %?", swapped));

	if swapped {
	// Success!
	comm::send(setup_ch, Proceed);
	ch
	} else {
	// Somebody else got in before we did
	comm::send(setup_ch, Abort);
	cast::reinterpret_cast(&*global)
	}
	} else {
	log(debug, ~"global != 0");
	cast::reinterpret_cast(&*global)
	}
	}

	#[test]
	pub fn test_from_global_chan1() {

	// This is unreadable, right?

	// The global channel
	let globchan = 0;
	let globchanp = ptr::addr_of(&globchan);

	// Create the global channel, attached to a new task
	let ch = unsafe {
	do chan_from_global_ptr(globchanp, task::task) \|po\| {
	let ch = comm::recv(po);
	comm::send(ch, true);
	let ch = comm::recv(po);
	comm::send(ch, true);
	}
	};
	// Talk to it
	let po = comm::Port();
	comm::send(ch, comm::Chan(&po));
	assert comm::recv(po) == true;

	// This one just reuses the previous channel
	let ch = unsafe {
	do chan_from_global_ptr(globchanp, task::task) \|po\| {
	let ch = comm::recv(po);
	comm::send(ch, false);
	}
	};

	// Talk to the original global task
	let po = comm::Port();
	comm::send(ch, comm::Chan(&po));
	assert comm::recv(po) == true;
	}

	#[test]
	pub fn test_from_global_chan2() {

	for iter::repeat(100) {
	// The global channel
	let globchan = 0;
	let globchanp = ptr::addr_of(&globchan);

	let resultpo = comm::Port();
	let resultch = comm::Chan(&resultpo);

	// Spawn a bunch of tasks that all want to compete to
	// create the global channel
	for uint::range(0, 10) \|i\| {
	do task::spawn {
	let ch = unsafe {
	do chan_from_global_ptr(
	globchanp, task::task) \|po\| {

	for uint::range(0, 10) \|_j\| {
	let ch = comm::recv(po);
	comm::send(ch, {i});
	}
	}
	};
	let po = comm::Port();
	comm::send(ch, comm::Chan(&po));
	// We are The winner if our version of the
	// task was installed
	let winner = comm::recv(po);
	comm::send(resultch, winner == i);
	}
	}
	// There should be only one winner
	let mut winners = 0u;
	for uint::range(0u, 10u) \|_i\| {
	let res = comm::recv(resultpo);
	if res { winners += 1u };
	}
	assert winners == 1u;
	}
	}

	/**
	* Convert the current task to a 'weak' task temporarily
	*
	* As a weak task it will not be counted towards the runtime's set
	* of live tasks. When there are no more outstanding live (non-weak) tasks
	* the runtime will send an exit message on the provided channel.
	*
	* This function is super-unsafe. Do not use.
	*
	* # Safety notes
	*
	* * Weak tasks must either die on their own or exit upon receipt of
	* the exit message. Failure to do so will cause the runtime to never
	* exit
	* * Tasks must not call `weaken_task` multiple times. This will
	* break the kernel's accounting of live tasks.
	* * Weak tasks must not be supervised. A supervised task keeps
	* a reference to its parent, so the parent will not die.
	*/
	pub unsafe fn weaken_task(f: fn(comm::Port<()>)) {
	let po = comm::Port();
	let ch = comm::Chan(&po);
	unsafe {
	rustrt::rust_task_weaken(cast::reinterpret_cast(&ch));
	}
	let _unweaken = Unweaken(ch);
	f(po);

	struct Unweaken {
	ch: comm::Chan<()>,
	drop unsafe {
	rustrt::rust_task_unweaken(cast::reinterpret_cast(&self.ch));
	}
	}

	fn Unweaken(ch: comm::Chan<()>) -> Unweaken {
	Unweaken {
	ch: ch
	}
	}
	}

	#[test]
	pub fn test_weaken_task_then_unweaken() {
	do task::try {
	unsafe {
	do weaken_task \|_po\| {
	}
	}
	};
	}

	#[test]
	pub fn test_weaken_task_wait() {
	do task::spawn_unlinked {
	unsafe {
	do weaken_task \|po\| {
	comm::recv(po);
	}
	}
	}
	}

	#[test]
	pub fn test_weaken_task_stress() {
	// Create a bunch of weak tasks
	for iter::repeat(100u) {
	do task::spawn {
	unsafe {
	do weaken_task \|_po\| {
	}
	}
	}
	do task::spawn_unlinked {
	unsafe {
	do weaken_task \|po\| {
	// Wait for it to tell us to die
	comm::recv(po);
	}
	}
	}
	}
	}

	#[test]
	#[ignore(cfg(windows))]
	pub fn test_weaken_task_fail() {
	let res = do task::try {
	unsafe {
	do weaken_task \|_po\| {
	fail;
	}
	}
	};
	assert result::is_err(&res);
	}

	/****************************************************************************
	* Shared state & exclusive ARC
	****************************************************************************/

	// An unwrapper uses this protocol to communicate with the "other" task that
	// drops the last refcount on an arc. Unfortunately this can't be a proper
	// pipe protocol because the unwrapper has to access both stages at once.
	type UnwrapProto = ~mut Option<(pipes::ChanOne<()>, pipes::PortOne<bool>)>;

	struct ArcData<T> {
	mut count: libc::intptr_t,
	mut unwrapper: libc::uintptr_t, // either a UnwrapProto or 0
	// FIXME(#3224) should be able to make this non-option to save memory, and
	// in unwrap() use "let ~ArcData { data: result, _ } = thing" to unwrap it
	mut data: Option<T>,
	}

	struct ArcDestruct<T> {
	mut data: *libc::c_void,
	drop unsafe {
	if self.data.is_null() {
	return; // Happens when destructing an unwrapper's handle.
	}
	do task::unkillable {
	let data: ~ArcData<T> = cast::reinterpret_cast(&self.data);
	let new_count = rustrt::rust_atomic_decrement(&mut data.count);
	assert new_count >= 0;
	if new_count == 0 {
	// Were we really last, or should we hand off to an unwrapper?
	// It's safe to not xchg because the unwrapper will set the
	// unwrap lock before dropping his/her reference. In effect,
	// being here means we're the only awake task with the data.
	if data.unwrapper != 0 {
	let p: UnwrapProto =
	cast::reinterpret_cast(&data.unwrapper);
	let (message, response) = option::swap_unwrap(p);
	// Send 'ready' and wait for a response.
	pipes::send_one(move message, ());
	// Unkillable wait. Message guaranteed to come.
	if pipes::recv_one(move response) {
	// Other task got the data.
	cast::forget(move data);
	} else {
	// Other task was killed. drop glue takes over.
	}
	} else {
	// drop glue takes over.
	}
	} else {
	cast::forget(move data);
	}
	}
	}
	}

	fn ArcDestruct<T>(data: *libc::c_void) -> ArcDestruct<T> {
	ArcDestruct {
	data: data
	}
	}

	pub unsafe fn unwrap_shared_mutable_state<T: Send>(rc: SharedMutableState<T>)
	-> T {
	struct DeathThroes<T> {
	mut ptr: Option<~ArcData<T>>,
	mut response: Option<pipes::ChanOne<bool>>,
	drop unsafe {
	let response = option::swap_unwrap(&mut self.response);
	// In case we get killed early, we need to tell the person who
	// tried to wake us whether they should hand-off the data to us.
	if task::failing() {
	pipes::send_one(move response, false);
	// Either this swap_unwrap or the one below (at "Got here")
	// ought to run.
	cast::forget(option::swap_unwrap(&mut self.ptr));
	} else {
	assert self.ptr.is_none();
	pipes::send_one(move response, true);
	}
	}
	}

	do task::unkillable {
	let ptr: ~ArcData<T> = cast::reinterpret_cast(&rc.data);
	let (c1,p1) = pipes::oneshot(); // ()
	let (c2,p2) = pipes::oneshot(); // bool
	let server: UnwrapProto = ~mut Some((move c1,move p2));
	let serverp: libc::uintptr_t = cast::transmute(move server);
	// Try to put our server end in the unwrapper slot.
	if rustrt::rust_compare_and_swap_ptr(&mut ptr.unwrapper, 0, serverp) {
	// Got in. Step 0: Tell destructor not to run. We are now it.
	rc.data = ptr::null();
	// Step 1 - drop our own reference.
	let new_count = rustrt::rust_atomic_decrement(&mut ptr.count);
	assert new_count >= 0;
	if new_count == 0 {
	// We were the last owner. Can unwrap immediately.
	// Also we have to free the server endpoints.
	let _server: UnwrapProto = cast::transmute(move serverp);
	option::swap_unwrap(&mut ptr.data)
	// drop glue takes over.
	} else {
	// The next person who sees the refcount hit 0 will wake us.
	let end_result =
	DeathThroes { ptr: Some(move ptr),
	response: Some(move c2) };
	let mut p1 = Some(move p1); // argh
	do task::rekillable {
	pipes::recv_one(option::swap_unwrap(&mut p1));
	}
	// Got here. Back in the 'unkillable' without getting killed.
	// Recover ownership of ptr, then take the data out.
	let ptr = option::swap_unwrap(&mut end_result.ptr);
	option::swap_unwrap(&mut ptr.data)
	// drop glue takes over.
	}
	} else {
	// Somebody else was trying to unwrap. Avoid guaranteed deadlock.
	cast::forget(move ptr);
	// Also we have to free the (rejected) server endpoints.
	let _server: UnwrapProto = cast::transmute(move serverp);
	fail ~"Another task is already unwrapping this ARC!";
	}
	}
	}

	/**
	* COMPLETELY UNSAFE. Used as a primitive for the safe versions in std::arc.
	*
	* Data races between tasks can result in crashes and, with sufficient
	* cleverness, arbitrary type coercion.
	*/
	pub type SharedMutableState<T: Send> = ArcDestruct<T>;

	pub unsafe fn shared_mutable_state<T: Send>(data: T) ->
	SharedMutableState<T> {
	let data = ~ArcData { count: 1, unwrapper: 0, data: Some(move data) };
	unsafe {
	let ptr = cast::transmute(move data);
	ArcDestruct(ptr)
	}
	}

	#[inline(always)]
	pub unsafe fn get_shared_mutable_state<T: Send>(rc: &a/SharedMutableState<T>)
	-> &a/mut T {
	unsafe {
	let ptr: ~ArcData<T> = cast::reinterpret_cast(&(*rc).data);
	assert ptr.count > 0;
	// Cast us back into the correct region
	let r = cast::transmute_region(option::get_ref(&ptr.data));
	cast::forget(move ptr);
	return cast::transmute_mut(r);
	}
	}
	#[inline(always)]
	pub unsafe fn get_shared_immutable_state<T: Send>(
	rc: &a/SharedMutableState<T>) -> &a/T {
	unsafe {
	let ptr: ~ArcData<T> = cast::reinterpret_cast(&(*rc).data);
	assert ptr.count > 0;
	// Cast us back into the correct region
	let r = cast::transmute_region(option::get_ref(&ptr.data));
	cast::forget(move ptr);
	return r;
	}
	}

	pub unsafe fn clone_shared_mutable_state<T: Send>(rc: &SharedMutableState<T>)
	-> SharedMutableState<T> {
	unsafe {
	let ptr: ~ArcData<T> = cast::reinterpret_cast(&(*rc).data);
	let new_count = rustrt::rust_atomic_increment(&mut ptr.count);
	assert new_count >= 2;
	cast::forget(move ptr);
	}
	ArcDestruct((*rc).data)
	}

	/****************************************************************************/

	#[allow(non_camel_case_types)] // runtime type
	type rust_little_lock = *libc::c_void;

	struct LittleLock {
	l: rust_little_lock,
	drop { rustrt::rust_destroy_little_lock(self.l); }
	}

	fn LittleLock() -> LittleLock {
	LittleLock {
	l: rustrt::rust_create_little_lock()
	}
	}

	impl LittleLock {
	#[inline(always)]
	unsafe fn lock<T>(f: fn() -> T) -> T {
	struct Unlock {
	l: rust_little_lock,
	drop { rustrt::rust_unlock_little_lock(self.l); }
	}

	fn Unlock(l: rust_little_lock) -> Unlock {
	Unlock {
	l: l
	}
	}

	do atomically {
	rustrt::rust_lock_little_lock(self.l);
	let _r = Unlock(self.l);
	f()
	}
	}
	}

	struct ExData<T: Send> { lock: LittleLock, mut failed: bool, mut data: T, }
	/**
	* An arc over mutable data that is protected by a lock. For library use only.
	*/
	pub struct Exclusive<T: Send> { x: SharedMutableState<ExData<T>> }

	pub fn exclusive<T:Send >(user_data: T) -> Exclusive<T> {
	let data = ExData {
	lock: LittleLock(), mut failed: false, mut data: user_data
	};
	Exclusive { x: unsafe { shared_mutable_state(move data) } }
	}

	impl<T: Send> Exclusive<T> {
	// Duplicate an exclusive ARC, as std::arc::clone.
	fn clone() -> Exclusive<T> {
	Exclusive { x: unsafe { clone_shared_mutable_state(&self.x) } }
	}

	// Exactly like std::arc::mutex_arc,access(), but with the little_lock
	// instead of a proper mutex. Same reason for being unsafe.
	//
	// Currently, scheduling operations (i.e., yielding, receiving on a pipe,
	// accessing the provided condition variable) are prohibited while inside
	// the exclusive. Supporting that is a work in progress.
	#[inline(always)]
	unsafe fn with<U>(f: fn(x: &mut T) -> U) -> U {
	let rec = unsafe { get_shared_mutable_state(&self.x) };
	do rec.lock.lock {
	if rec.failed {
	fail ~"Poisoned exclusive - another task failed inside!";
	}
	rec.failed = true;
	let result = f(&mut rec.data);
	rec.failed = false;
	move result
	}
	}

	#[inline(always)]
	unsafe fn with_imm<U>(f: fn(x: &T) -> U) -> U {
	do self.with \|x\| {
	f(cast::transmute_immut(x))
	}
	}
	}

	// FIXME(#3724) make this a by-move method on the exclusive
	pub fn unwrap_exclusive<T: Send>(arc: Exclusive<T>) -> T {
	let Exclusive { x: x } <- arc;
	let inner = unsafe { unwrap_shared_mutable_state(move x) };
	let ExData { data: data, _ } <- inner;
	move data
	}

	#[cfg(test)]
	pub mod tests {
	#[test]
	pub fn exclusive_arc() {
	let mut futures = ~[];

	let num_tasks = 10u;
	let count = 10u;

	let total = exclusive(~mut 0u);

	for uint::range(0u, num_tasks) \|_i\| {
	let total = total.clone();
	futures.push(future::spawn(\|\| {
	for uint::range(0u, count) \|_i\| {
	do total.with \|count\| {
	**count += 1u;
	}
	}
	}));
	};

	for futures.each \|f\| { f.get() }

	do total.with \|total\| {
	assert *total == num_tasks count
	};
	}

	#[test] #[should_fail] #[ignore(cfg(windows))]
	pub fn exclusive_poison() {
	// Tests that if one task fails inside of an exclusive, subsequent
	// accesses will also fail.
	let x = exclusive(1);
	let x2 = x.clone();
	do task::try {
	do x2.with \|one\| {
	assert *one == 2;
	}
	};
	do x.with \|one\| {
	assert *one == 1;
	}
	}

	#[test]
	pub fn exclusive_unwrap_basic() {
	let x = exclusive(~~"hello");
	assert unwrap_exclusive(x) == ~~"hello";
	}

	#[test]
	pub fn exclusive_unwrap_contended() {
	let x = exclusive(~~"hello");
	let x2 = ~mut Some(x.clone());
	do task::spawn {
	let x2 = option::swap_unwrap(x2);
	do x2.with \|_hello\| { }
	task::yield();
	}
	assert unwrap_exclusive(x) == ~~"hello";

	// Now try the same thing, but with the child task blocking.
	let x = exclusive(~~"hello");
	let x2 = ~mut Some(x.clone());
	let mut res = None;
	do task::task().future_result(\|+r\| res = Some(r)).spawn {
	let x2 = option::swap_unwrap(x2);
	assert unwrap_exclusive(x2) == ~~"hello";
	}
	// Have to get rid of our reference before blocking.
	{ let _x = move x; } // FIXME(#3161) util::ignore doesn't work here
	let res = option::swap_unwrap(&mut res);
	future::get(&res);
	}

	#[test] #[should_fail] #[ignore(cfg(windows))]
	pub fn exclusive_unwrap_conflict() {
	let x = exclusive(~~"hello");
	let x2 = ~mut Some(x.clone());
	let mut res = None;
	do task::task().future_result(\|+r\| res = Some(r)).spawn {
	let x2 = option::swap_unwrap(x2);
	assert unwrap_exclusive(x2) == ~~"hello";
	}
	assert unwrap_exclusive(x) == ~~"hello";
	let res = option::swap_unwrap(&mut res);
	future::get(&res);
	}

	#[test] #[ignore(cfg(windows))]
	pub fn exclusive_unwrap_deadlock() {
	// This is not guaranteed to get to the deadlock before being killed,
	// but it will show up sometimes, and if the deadlock were not there,
	// the test would nondeterministically fail.
	let result = do task::try {
	// a task that has two references to the same exclusive will
	// deadlock when it unwraps. nothing to be done about that.
	let x = exclusive(~~"hello");
	let x2 = x.clone();
	do task::spawn {
	for 10.times { task::yield(); } // try to let the unwrapper go
	fail; // punt it awake from its deadlock
	}
	let _z = unwrap_exclusive(x);
	do x2.with \|_hello\| { }
	};
	assert result.is_err();
	}
	}