src/libstd/rt/comm.rs

   1 // Copyright 2013 The Rust Project Developers. See the COPYRIGHT
   2 // file at the top-level directory of this distribution and at
   3 // http://rust-lang.org/COPYRIGHT.
   4 //
   5 // Licensed under the Apache License, Version 2.0 <LICENSE-APACHE or
   6 // http://www.apache.org/licenses/LICENSE-2.0> or the MIT license
   7 // <LICENSE-MIT or http://opensource.org/licenses/MIT>, at your
   8 // option. This file may not be copied, modified, or distributed
   9 // except according to those terms.
  10
  11 //! Ports and channels.
  12
  13 use option::*;
  14 use cast;
  15 use ops::Drop;
  16 use rt::kill::BlockedTask;
  17 use kinds::Send;
  18 use rt::sched::Scheduler;
  19 use rt::local::Local;
  20 use rt::select::{Select, SelectPort};
  21 use unstable::atomics::{AtomicUint, AtomicOption, Acquire, Relaxed, SeqCst};
  22 use unstable::sync::UnsafeAtomicRcBox;
  23 use util::Void;
  24 use comm::{GenericChan, GenericSmartChan, GenericPort, Peekable};
  25 use cell::Cell;
  26 use clone::Clone;
  27 use rt::{context, SchedulerContext};
  28 use tuple::ImmutableTuple;
  29
  30 /// A combined refcount / BlockedTask-as-uint pointer.
  31 ///
  32 /// Can be equal to the following values:
  33 ///
  34 /// * 2 - both endpoints are alive
  35 /// * 1 - either the sender or the receiver is dead, determined by context
  36 /// * <ptr> - A pointer to a blocked Task (see BlockedTask::cast_{to,from}_uint)
  37 type State = uint;
  38
  39 static STATE_BOTH: State = 2;
  40 static STATE_ONE: State = 1;
  41
  42 /// The heap-allocated structure shared between two endpoints.
  43 struct Packet<T> {
  44     state: AtomicUint,
  45     payload: Option<T>,
  46 }
  47
  48 /// A one-shot channel.
  49 pub struct ChanOne<T> {
  50     void_packet: *mut Void,
  51     suppress_finalize: bool
  52 }
  53
  54 /// A one-shot port.
  55 pub struct PortOne<T> {
  56     void_packet: *mut Void,
  57     suppress_finalize: bool
  58 }
  59
  60 pub fn oneshot<T: Send>() -> (PortOne<T>, ChanOne<T>) {
  61     let packet: ~Packet<T> = ~Packet {
  62         state: AtomicUint::new(STATE_BOTH),
  63         payload: None
  64     };
  65
  66     unsafe {
  67         let packet: *mut Void = cast::transmute(packet);
  68         let port = PortOne {
  69             void_packet: packet,
  70             suppress_finalize: false
  71         };
  72         let chan = ChanOne {
  73             void_packet: packet,
  74             suppress_finalize: false
  75         };
  76         return (port, chan);
  77     }
  78 }
  79
  80 impl<T> ChanOne<T> {
  81     #[inline]
  82     fn packet(&self) -> *mut Packet<T> {
  83         unsafe {
  84             let p: *mut ~Packet<T> = cast::transmute(&self.void_packet);
  85             let p: *mut Packet<T> = &mut **p;
  86             return p;
  87         }
  88     }
  89
  90     /// Send a message on the one-shot channel. If a receiver task is blocked
  91     /// waiting for the message, will wake it up and reschedule to it.
  92     pub fn send(self, val: T) {
  93         self.try_send(val);
  94     }
  95
  96     /// As `send`, but also returns whether or not the receiver endpoint is still open.
  97     pub fn try_send(self, val: T) -> bool {
  98         self.try_send_inner(val, true)
  99     }
 100
 101     /// Send a message without immediately rescheduling to a blocked receiver.
 102     /// This can be useful in contexts where rescheduling is forbidden, or to
 103     /// optimize for when the sender expects to still have useful work to do.
 104     pub fn send_deferred(self, val: T) {
 105         self.try_send_deferred(val);
 106     }
 107
 108     /// As `send_deferred` and `try_send` together.
 109     pub fn try_send_deferred(self, val: T) -> bool {
 110         self.try_send_inner(val, false)
 111     }
 112
 113     // 'do_resched' configures whether the scheduler immediately switches to
 114     // the receiving task, or leaves the sending task still running.
 115     fn try_send_inner(self, val: T, do_resched: bool) -> bool {
 116         rtassert!(context() != SchedulerContext);
 117
 118         let mut this = self;
 119         let mut recvr_active = true;
 120         let packet = this.packet();
 121
 122         unsafe {
 123
 124             // Install the payload
 125             assert!((*packet).payload.is_none());
 126             (*packet).payload = Some(val);
 127
 128             // Atomically swap out the old state to figure out what
 129             // the port's up to, issuing a release barrier to prevent
 130             // reordering of the payload write. This also issues an
 131             // acquire barrier that keeps the subsequent access of the
 132             // ~Task pointer from being reordered.
 133             let oldstate = (*packet).state.swap(STATE_ONE, SeqCst);
 134
 135             // Suppress the synchronizing actions in the finalizer. We're
 136             // done with the packet. NB: In case of do_resched, this *must*
 137             // happen before waking up a blocked task (or be unkillable),
 138             // because we might get a kill signal during the reschedule.
 139             this.suppress_finalize = true;
 140
 141             match oldstate {
 142                 STATE_BOTH => {
 143                     // Port is not waiting yet. Nothing to do
 144                     do Local::borrow::<Scheduler, ()> |sched| {
 145                         rtdebug!("non-rendezvous send");
 146                         sched.metrics.non_rendezvous_sends += 1;
 147                     }
 148                 }
 149                 STATE_ONE => {
 150                     do Local::borrow::<Scheduler, ()> |sched| {
 151                         rtdebug!("rendezvous send");
 152                         sched.metrics.rendezvous_sends += 1;
 153                     }
 154                     // Port has closed. Need to clean up.
 155                     let _packet: ~Packet<T> = cast::transmute(this.void_packet);
 156                     recvr_active = false;
 157                 }
 158                 task_as_state => {
 159                     // Port is blocked. Wake it up.
 160                     let recvr = BlockedTask::cast_from_uint(task_as_state);
 161                     if do_resched {
 162                         do recvr.wake().map_move |woken_task| {
 163                             Scheduler::run_task(woken_task);
 164                         };
 165                     } else {
 166                         let recvr = Cell::new(recvr);
 167                         do Local::borrow::<Scheduler, ()> |sched| {
 168                             sched.enqueue_blocked_task(recvr.take());
 169                         }
 170                     }
 171                 }
 172             }
 173         }
 174
 175         return recvr_active;
 176     }
 177 }
 178
 179 impl<T> PortOne<T> {
 180     fn packet(&self) -> *mut Packet<T> {
 181         unsafe {
 182             let p: *mut ~Packet<T> = cast::transmute(&self.void_packet);
 183             let p: *mut Packet<T> = &mut **p;
 184             return p;
 185         }
 186     }
 187
 188     /// Wait for a message on the one-shot port. Fails if the send end is closed.
 189     pub fn recv(self) -> T {
 190         match self.try_recv() {
 191             Some(val) => val,
 192             None => {
 193                 fail!("receiving on closed channel");
 194             }
 195         }
 196     }
 197
 198     /// As `recv`, but returns `None` if the send end is closed rather than failing.
 199     pub fn try_recv(self) -> Option<T> {
 200         let mut this = self;
 201
 202         // Optimistic check. If data was sent already, we don't even need to block.
 203         // No release barrier needed here; we're not handing off our task pointer yet.
 204         if !this.optimistic_check() {
 205             // No data available yet.
 206             // Switch to the scheduler to put the ~Task into the Packet state.
 207             let sched = Local::take::<Scheduler>();
 208             do sched.deschedule_running_task_and_then |sched, task| {
 209                 this.block_on(sched, task);
 210             }
 211         }
 212
 213         // Task resumes.
 214         this.recv_ready()
 215     }
 216 }
 217
 218 impl<T> Select for PortOne<T> {
 219     #[inline] #[cfg(not(test))]
 220     fn optimistic_check(&mut self) -> bool {
 221         unsafe { (*self.packet()).state.load(Acquire) == STATE_ONE }
 222     }
 223
 224     #[inline] #[cfg(test)]
 225     fn optimistic_check(&mut self) -> bool {
 226         // The optimistic check is never necessary for correctness. For testing
 227         // purposes, making it randomly return false simulates a racing sender.
 228         use rand::{Rand};
 229         let actually_check = do Local::borrow::<Scheduler, bool> |sched| {
 230             Rand::rand(&mut sched.rng)
 231         };
 232         if actually_check {
 233             unsafe { (*self.packet()).state.load(Acquire) == STATE_ONE }
 234         } else {
 235             false
 236         }
 237     }
 238
 239     fn block_on(&mut self, sched: &mut Scheduler, task: BlockedTask) -> bool {
 240         unsafe {
 241             // Atomically swap the task pointer into the Packet state, issuing
 242             // an acquire barrier to prevent reordering of the subsequent read
 243             // of the payload. Also issues a release barrier to prevent
 244             // reordering of any previous writes to the task structure.
 245             let task_as_state = task.cast_to_uint();
 246             let oldstate = (*self.packet()).state.swap(task_as_state, SeqCst);
 247             match oldstate {
 248                 STATE_BOTH => {
 249                     // Data has not been sent. Now we're blocked.
 250                     rtdebug!("non-rendezvous recv");
 251                     sched.metrics.non_rendezvous_recvs += 1;
 252                     false
 253                 }
 254                 STATE_ONE => {
 255                     // Re-record that we are the only owner of the packet.
 256                     // No barrier needed, even if the task gets reawoken
 257                     // on a different core -- this is analogous to writing a
 258                     // payload; a barrier in enqueueing the task protects it.
 259                     // NB(#8132). This *must* occur before the enqueue below.
 260                     // FIXME(#6842, #8130) This is usually only needed for the
 261                     // assertion in recv_ready, except in the case of select().
 262                     // This won't actually ever have cacheline contention, but
 263                     // maybe should be optimized out with a cfg(test) anyway?
 264                     (*self.packet()).state.store(STATE_ONE, Relaxed);
 265
 266                     rtdebug!("rendezvous recv");
 267                     sched.metrics.rendezvous_recvs += 1;
 268
 269                     // Channel is closed. Switch back and check the data.
 270                     // NB: We have to drop back into the scheduler event loop here
 271                     // instead of switching immediately back or we could end up
 272                     // triggering infinite recursion on the scheduler's stack.
 273                     let recvr = BlockedTask::cast_from_uint(task_as_state);
 274                     sched.enqueue_blocked_task(recvr);
 275                     true
 276                 }
 277                 _ => rtabort!("can't block_on; a task is already blocked")
 278             }
 279         }
 280     }
 281
 282     // This is the only select trait function that's not also used in recv.
 283     fn unblock_from(&mut self) -> bool {
 284         let packet = self.packet();
 285         unsafe {
 286             // In case the data is available, the acquire barrier here matches
 287             // the release barrier the sender used to release the payload.
 288             match (*packet).state.load(Acquire) {
 289                 // Impossible. We removed STATE_BOTH when blocking on it, and
 290                 // no self-respecting sender would put it back.
 291                 STATE_BOTH    => rtabort!("refcount already 2 in unblock_from"),
 292                 // Here, a sender already tried to wake us up. Perhaps they
 293                 // even succeeded! Data is available.
 294                 STATE_ONE     => true,
 295                 // Still registered as blocked. Need to "unblock" the pointer.
 296                 task_as_state => {
 297                     // In the window between the load and the CAS, a sender
 298                     // might take the pointer and set the refcount to ONE. If
 299                     // that happens, we shouldn't clobber that with BOTH!
 300                     // Acquire barrier again for the same reason as above.
 301                     match (*packet).state.compare_and_swap(task_as_state, STATE_BOTH,
 302                                                            Acquire) {
 303                         STATE_BOTH => rtabort!("refcount became 2 in unblock_from"),
 304                         STATE_ONE  => true, // Lost the race. Data available.
 305                         same_ptr   => {
 306                             // We successfully unblocked our task pointer.
 307                             assert!(task_as_state == same_ptr);
 308                             let handle = BlockedTask::cast_from_uint(task_as_state);
 309                             // Because we are already awake, the handle we
 310                             // gave to this port shall already be empty.
 311                             handle.assert_already_awake();
 312                             false
 313                         }
 314                     }
 315                 }
 316             }
 317         }
 318     }
 319 }
 320
 321 impl<T> SelectPort<T> for PortOne<T> {
 322     fn recv_ready(self) -> Option<T> {
 323         let mut this = self;
 324         let packet = this.packet();
 325
 326         // No further memory barrier is needed here to access the
 327         // payload. Some scenarios:
 328         //
 329         // 1) We encountered STATE_ONE above - the atomic_xchg was the acq barrier. We're fine.
 330         // 2) We encountered STATE_BOTH above and blocked. The sending task then ran us
 331         //    and ran on its thread. The sending task issued a read barrier when taking the
 332         //    pointer to the receiving task.
 333         // 3) We encountered STATE_BOTH above and blocked, but the receiving task (this task)
 334         //    is pinned to some other scheduler, so the sending task had to give us to
 335         //    a different scheduler for resuming. That send synchronized memory.
 336         unsafe {
 337             // See corresponding store() above in block_on for rationale.
 338             // FIXME(#8130) This can happen only in test builds.
 339             assert!((*packet).state.load(Relaxed) == STATE_ONE);
 340
 341             let payload = (*packet).payload.take();
 342
 343             // The sender has closed up shop. Drop the packet.
 344             let _packet: ~Packet<T> = cast::transmute(this.void_packet);
 345             // Suppress the synchronizing actions in the finalizer. We're done with the packet.
 346             this.suppress_finalize = true;
 347             return payload;
 348         }
 349     }
 350 }
 351
 352 impl<T> Peekable<T> for PortOne<T> {
 353     fn peek(&self) -> bool {
 354         unsafe {
 355             let packet: *mut Packet<T> = self.packet();
 356             let oldstate = (*packet).state.load(SeqCst);
 357             match oldstate {
 358                 STATE_BOTH => false,
 359                 STATE_ONE => (*packet).payload.is_some(),
 360                 _ => rtabort!("peeked on a blocked task")
 361             }
 362         }
 363     }
 364 }
 365
 366 #[unsafe_destructor]
 367 impl<T> Drop for ChanOne<T> {
 368     fn drop(&self) {
 369         if self.suppress_finalize { return }
 370
 371         unsafe {
 372             let this = cast::transmute_mut(self);
 373             let oldstate = (*this.packet()).state.swap(STATE_ONE, SeqCst);
 374             match oldstate {
 375                 STATE_BOTH => {
 376                     // Port still active. It will destroy the Packet.
 377                 },
 378                 STATE_ONE => {
 379                     let _packet: ~Packet<T> = cast::transmute(this.void_packet);
 380                 },
 381                 task_as_state => {
 382                     // The port is blocked waiting for a message we will never send. Wake it.
 383                     assert!((*this.packet()).payload.is_none());
 384                     let recvr = BlockedTask::cast_from_uint(task_as_state);
 385                     do recvr.wake().map_move |woken_task| {
 386                         Scheduler::run_task(woken_task);
 387                     };
 388                 }
 389             }
 390         }
 391     }
 392 }
 393
 394 #[unsafe_destructor]
 395 impl<T> Drop for PortOne<T> {
 396     fn drop(&self) {
 397         if self.suppress_finalize { return }
 398
 399         unsafe {
 400             let this = cast::transmute_mut(self);
 401             let oldstate = (*this.packet()).state.swap(STATE_ONE, SeqCst);
 402             match oldstate {
 403                 STATE_BOTH => {
 404                     // Chan still active. It will destroy the packet.
 405                 },
 406                 STATE_ONE => {
 407                     let _packet: ~Packet<T> = cast::transmute(this.void_packet);
 408                 }
 409                 task_as_state => {
 410                     // This case occurs during unwinding, when the blocked
 411                     // receiver was killed awake. The task can't still be
 412                     // blocked (we are it), but we need to free the handle.
 413                     let recvr = BlockedTask::cast_from_uint(task_as_state);
 414                     recvr.assert_already_awake();
 415                 }
 416             }
 417         }
 418     }
 419 }
 420
 421 /// Trait for non-rescheduling send operations, similar to `send_deferred` on ChanOne.
 422 pub trait SendDeferred<T> {
 423     fn send_deferred(&self, val: T);
 424     fn try_send_deferred(&self, val: T) -> bool;
 425 }
 426
 427 struct StreamPayload<T> {
 428     val: T,
 429     next: PortOne<StreamPayload<T>>
 430 }
 431
 432 type StreamChanOne<T> = ChanOne<StreamPayload<T>>;
 433 type StreamPortOne<T> = PortOne<StreamPayload<T>>;
 434
 435 /// A channel with unbounded size.
 436 pub struct Chan<T> {
 437     // FIXME #5372. Using Cell because we don't take &mut self
 438     next: Cell<StreamChanOne<T>>
 439 }
 440
 441 /// An port with unbounded size.
 442 pub struct Port<T> {
 443     // FIXME #5372. Using Cell because we don't take &mut self
 444     next: Cell<StreamPortOne<T>>
 445 }
 446
 447 pub fn stream<T: Send>() -> (Port<T>, Chan<T>) {
 448     let (pone, cone) = oneshot();
 449     let port = Port { next: Cell::new(pone) };
 450     let chan = Chan { next: Cell::new(cone) };
 451     return (port, chan);
 452 }
 453
 454 impl<T: Send> Chan<T> {
 455     fn try_send_inner(&self, val: T, do_resched: bool) -> bool {
 456         let (next_pone, next_cone) = oneshot();
 457         let cone = self.next.take();
 458         self.next.put_back(next_cone);
 459         cone.try_send_inner(StreamPayload { val: val, next: next_pone }, do_resched)
 460     }
 461 }
 462
 463 impl<T: Send> GenericChan<T> for Chan<T> {
 464     fn send(&self, val: T) {
 465         self.try_send(val);
 466     }
 467 }
 468
 469 impl<T: Send> GenericSmartChan<T> for Chan<T> {
 470     fn try_send(&self, val: T) -> bool {
 471         self.try_send_inner(val, true)
 472     }
 473 }
 474
 475 impl<T: Send> SendDeferred<T> for Chan<T> {
 476     fn send_deferred(&self, val: T) {
 477         self.try_send_deferred(val);
 478     }
 479     fn try_send_deferred(&self, val: T) -> bool {
 480         self.try_send_inner(val, false)
 481     }
 482 }
 483
 484 impl<T> GenericPort<T> for Port<T> {
 485     fn recv(&self) -> T {
 486         match self.try_recv() {
 487             Some(val) => val,
 488             None => {
 489                 fail!("receiving on closed channel");
 490             }
 491         }
 492     }
 493
 494     fn try_recv(&self) -> Option<T> {
 495         let pone = self.next.take();
 496         match pone.try_recv() {
 497             Some(StreamPayload { val, next }) => {
 498                 self.next.put_back(next);
 499                 Some(val)
 500             }
 501             None => None
 502         }
 503     }
 504 }
 505
 506 impl<T> Peekable<T> for Port<T> {
 507     fn peek(&self) -> bool {
 508         self.next.with_mut_ref(|p| p.peek())
 509     }
 510 }
 511
 512 impl<T> Select for Port<T> {
 513     #[inline]
 514     fn optimistic_check(&mut self) -> bool {
 515         do self.next.with_mut_ref |pone| { pone.optimistic_check() }
 516     }
 517
 518     #[inline]
 519     fn block_on(&mut self, sched: &mut Scheduler, task: BlockedTask) -> bool {
 520         let task = Cell::new(task);
 521         do self.next.with_mut_ref |pone| { pone.block_on(sched, task.take()) }
 522     }
 523
 524     #[inline]
 525     fn unblock_from(&mut self) -> bool {
 526         do self.next.with_mut_ref |pone| { pone.unblock_from() }
 527     }
 528 }
 529
 530 impl<T> SelectPort<(T, Port<T>)> for Port<T> {
 531     fn recv_ready(self) -> Option<(T, Port<T>)> {
 532         match self.next.take().recv_ready() {
 533             Some(StreamPayload { val, next }) => {
 534                 self.next.put_back(next);
 535                 Some((val, self))
 536             }
 537             None => None
 538         }
 539     }
 540 }
 541
 542 pub struct SharedChan<T> {
 543     // Just like Chan, but a shared AtomicOption instead of Cell
 544     priv next: UnsafeAtomicRcBox<AtomicOption<StreamChanOne<T>>>
 545 }
 546
 547 impl<T> SharedChan<T> {
 548     pub fn new(chan: Chan<T>) -> SharedChan<T> {
 549         let next = chan.next.take();
 550         let next = AtomicOption::new(~next);
 551         SharedChan { next: UnsafeAtomicRcBox::new(next) }
 552     }
 553 }
 554
 555 impl<T: Send> SharedChan<T> {
 556     fn try_send_inner(&self, val: T, do_resched: bool) -> bool {
 557         unsafe {
 558             let (next_pone, next_cone) = oneshot();
 559             let cone = (*self.next.get()).swap(~next_cone, SeqCst);
 560             cone.unwrap().try_send_inner(StreamPayload { val: val, next: next_pone },
 561                                          do_resched)
 562         }
 563     }
 564 }
 565
 566 impl<T: Send> GenericChan<T> for SharedChan<T> {
 567     fn send(&self, val: T) {
 568         self.try_send(val);
 569     }
 570 }
 571
 572 impl<T: Send> GenericSmartChan<T> for SharedChan<T> {
 573     fn try_send(&self, val: T) -> bool {
 574         self.try_send_inner(val, true)
 575     }
 576 }
 577
 578 impl<T: Send> SendDeferred<T> for SharedChan<T> {
 579     fn send_deferred(&self, val: T) {
 580         self.try_send_deferred(val);
 581     }
 582     fn try_send_deferred(&self, val: T) -> bool {
 583         self.try_send_inner(val, false)
 584     }
 585 }
 586
 587 impl<T> Clone for SharedChan<T> {
 588     fn clone(&self) -> SharedChan<T> {
 589         SharedChan {
 590             next: self.next.clone()
 591         }
 592     }
 593 }
 594
 595 pub struct SharedPort<T> {
 596     // The next port on which we will receive the next port on which we will receive T
 597     priv next_link: UnsafeAtomicRcBox<AtomicOption<PortOne<StreamPortOne<T>>>>
 598 }
 599
 600 impl<T> SharedPort<T> {
 601     pub fn new(port: Port<T>) -> SharedPort<T> {
 602         // Put the data port into a new link pipe
 603         let next_data_port = port.next.take();
 604         let (next_link_port, next_link_chan) = oneshot();
 605         next_link_chan.send(next_data_port);
 606         let next_link = AtomicOption::new(~next_link_port);
 607         SharedPort { next_link: UnsafeAtomicRcBox::new(next_link) }
 608     }
 609 }
 610
 611 impl<T: Send> GenericPort<T> for SharedPort<T> {
 612     fn recv(&self) -> T {
 613         match self.try_recv() {
 614             Some(val) => val,
 615             None => {
 616                 fail!("receiving on a closed channel");
 617             }
 618         }
 619     }
 620
 621     fn try_recv(&self) -> Option<T> {
 622         unsafe {
 623             let (next_link_port, next_link_chan) = oneshot();
 624             let link_port = (*self.next_link.get()).swap(~next_link_port, SeqCst);
 625             let link_port = link_port.unwrap();
 626             let data_port = link_port.recv();
 627             let (next_data_port, res) = match data_port.try_recv() {
 628                 Some(StreamPayload { val, next }) => {
 629                     (next, Some(val))
 630                 }
 631                 None => {
 632                     let (next_data_port, _) = oneshot();
 633                     (next_data_port, None)
 634                 }
 635             };
 636             next_link_chan.send(next_data_port);
 637             return res;
 638         }
 639     }
 640 }
 641
 642 impl<T> Clone for SharedPort<T> {
 643     fn clone(&self) -> SharedPort<T> {
 644         SharedPort {
 645             next_link: self.next_link.clone()
 646         }
 647     }
 648 }
 649
 650 // XXX: Need better name
 651 type MegaPipe<T> = (SharedPort<T>, SharedChan<T>);
 652
 653 pub fn megapipe<T: Send>() -> MegaPipe<T> {
 654     let (port, chan) = stream();
 655     (SharedPort::new(port), SharedChan::new(chan))
 656 }
 657
 658 impl<T: Send> GenericChan<T> for MegaPipe<T> {
 659     fn send(&self, val: T) {
 660         self.second_ref().send(val)
 661     }
 662 }
 663
 664 impl<T: Send> GenericSmartChan<T> for MegaPipe<T> {
 665     fn try_send(&self, val: T) -> bool {
 666         self.second_ref().try_send(val)
 667     }
 668 }
 669
 670 impl<T: Send> GenericPort<T> for MegaPipe<T> {
 671     fn recv(&self) -> T {
 672         self.first_ref().recv()
 673     }
 674
 675     fn try_recv(&self) -> Option<T> {
 676         self.first_ref().try_recv()
 677     }
 678 }
 679
 680 impl<T: Send> SendDeferred<T> for MegaPipe<T> {
 681     fn send_deferred(&self, val: T) {
 682         self.second_ref().send_deferred(val)
 683     }
 684     fn try_send_deferred(&self, val: T) -> bool {
 685         self.second_ref().try_send_deferred(val)
 686     }
 687 }
 688
 689 #[cfg(test)]
 690 mod test {
 691     use super::*;
 692     use option::*;
 693     use rt::test::*;
 694     use cell::Cell;
 695     use iter::Times;
 696
 697     #[test]
 698     fn oneshot_single_thread_close_port_first() {
 699         // Simple test of closing without sending
 700         do run_in_newsched_task {
 701             let (port, _chan) = oneshot::<int>();
 702             { let _p = port; }
 703         }
 704     }
 705
 706     #[test]
 707     fn oneshot_single_thread_close_chan_first() {
 708         // Simple test of closing without sending
 709         do run_in_newsched_task {
 710             let (_port, chan) = oneshot::<int>();
 711             { let _c = chan; }
 712         }
 713     }
 714
 715     #[test]
 716     fn oneshot_single_thread_send_port_close() {
 717         // Testing that the sender cleans up the payload if receiver is closed
 718         do run_in_newsched_task {
 719             let (port, chan) = oneshot::<~int>();
 720             { let _p = port; }
 721             chan.send(~0);
 722         }
 723     }
 724
 725     #[test]
 726     fn oneshot_single_thread_recv_chan_close() {
 727         // Receiving on a closed chan will fail
 728         do run_in_newsched_task {
 729             let res = do spawntask_try {
 730                 let (port, chan) = oneshot::<~int>();
 731                 { let _c = chan; }
 732                 port.recv();
 733             };
 734             // What is our res?
 735             rtdebug!("res is: %?", res.is_err());
 736             assert!(res.is_err());
 737         }
 738     }
 739
 740     #[test]
 741     fn oneshot_single_thread_send_then_recv() {
 742         do run_in_newsched_task {
 743             let (port, chan) = oneshot::<~int>();
 744             chan.send(~10);
 745             assert!(port.recv() == ~10);
 746         }
 747     }
 748
 749     #[test]
 750     fn oneshot_single_thread_try_send_open() {
 751         do run_in_newsched_task {
 752             let (port, chan) = oneshot::<int>();
 753             assert!(chan.try_send(10));
 754             assert!(port.recv() == 10);
 755         }
 756     }
 757
 758     #[test]
 759     fn oneshot_single_thread_try_send_closed() {
 760         do run_in_newsched_task {
 761             let (port, chan) = oneshot::<int>();
 762             { let _p = port; }
 763             assert!(!chan.try_send(10));
 764         }
 765     }
 766
 767     #[test]
 768     fn oneshot_single_thread_try_recv_open() {
 769         do run_in_newsched_task {
 770             let (port, chan) = oneshot::<int>();
 771             chan.send(10);
 772             assert!(port.try_recv() == Some(10));
 773         }
 774     }
 775
 776     #[test]
 777     fn oneshot_single_thread_try_recv_closed() {
 778         do run_in_newsched_task {
 779             let (port, chan) = oneshot::<int>();
 780             { let _c = chan; }
 781             assert!(port.try_recv() == None);
 782         }
 783     }
 784
 785     #[test]
 786     fn oneshot_single_thread_peek_data() {
 787         do run_in_newsched_task {
 788             let (port, chan) = oneshot::<int>();
 789             assert!(!port.peek());
 790             chan.send(10);
 791             assert!(port.peek());
 792         }
 793     }
 794
 795     #[test]
 796     fn oneshot_single_thread_peek_close() {
 797         do run_in_newsched_task {
 798             let (port, chan) = oneshot::<int>();
 799             { let _c = chan; }
 800             assert!(!port.peek());
 801             assert!(!port.peek());
 802         }
 803     }
 804
 805     #[test]
 806     fn oneshot_single_thread_peek_open() {
 807         do run_in_newsched_task {
 808             let (port, _) = oneshot::<int>();
 809             assert!(!port.peek());
 810         }
 811     }
 812
 813     #[test]
 814     fn oneshot_multi_task_recv_then_send() {
 815         do run_in_newsched_task {
 816             let (port, chan) = oneshot::<~int>();
 817             let port_cell = Cell::new(port);
 818             do spawntask {
 819                 assert!(port_cell.take().recv() == ~10);
 820             }
 821
 822             chan.send(~10);
 823         }
 824     }
 825
 826     #[test]
 827     fn oneshot_multi_task_recv_then_close() {
 828         do run_in_newsched_task {
 829             let (port, chan) = oneshot::<~int>();
 830             let port_cell = Cell::new(port);
 831             let chan_cell = Cell::new(chan);
 832             do spawntask_later {
 833                 let _cell = chan_cell.take();
 834             }
 835             let res = do spawntask_try {
 836                 assert!(port_cell.take().recv() == ~10);
 837             };
 838             assert!(res.is_err());
 839         }
 840     }
 841
 842     #[test]
 843     fn oneshot_multi_thread_close_stress() {
 844         do stress_factor().times {
 845             do run_in_newsched_task {
 846                 let (port, chan) = oneshot::<int>();
 847                 let port_cell = Cell::new(port);
 848                 let thread = do spawntask_thread {
 849                     let _p = port_cell.take();
 850                 };
 851                 let _chan = chan;
 852                 thread.join();
 853             }
 854         }
 855     }
 856
 857     #[test]
 858     fn oneshot_multi_thread_send_close_stress() {
 859         do stress_factor().times {
 860             do run_in_newsched_task {
 861                 let (port, chan) = oneshot::<int>();
 862                 let chan_cell = Cell::new(chan);
 863                 let port_cell = Cell::new(port);
 864                 let thread1 = do spawntask_thread {
 865                     let _p = port_cell.take();
 866                 };
 867                 let thread2 = do spawntask_thread {
 868                     let c = chan_cell.take();
 869                     c.send(1);
 870                 };
 871                 thread1.join();
 872                 thread2.join();
 873             }
 874         }
 875     }
 876
 877     #[test]
 878     fn oneshot_multi_thread_recv_close_stress() {
 879         do stress_factor().times {
 880             do run_in_newsched_task {
 881                 let (port, chan) = oneshot::<int>();
 882                 let chan_cell = Cell::new(chan);
 883                 let port_cell = Cell::new(port);
 884                 let thread1 = do spawntask_thread {
 885                     let port_cell = Cell::new(port_cell.take());
 886                     let res = do spawntask_try {
 887                         port_cell.take().recv();
 888                     };
 889                     assert!(res.is_err());
 890                 };
 891                 let thread2 = do spawntask_thread {
 892                     let chan_cell = Cell::new(chan_cell.take());
 893                     do spawntask {
 894                         chan_cell.take();
 895                     }
 896                 };
 897                 thread1.join();
 898                 thread2.join();
 899             }
 900         }
 901     }
 902
 903     #[test]
 904     fn oneshot_multi_thread_send_recv_stress() {
 905         do stress_factor().times {
 906             do run_in_newsched_task {
 907                 let (port, chan) = oneshot::<~int>();
 908                 let chan_cell = Cell::new(chan);
 909                 let port_cell = Cell::new(port);
 910                 let thread1 = do spawntask_thread {
 911                     chan_cell.take().send(~10);
 912                 };
 913                 let thread2 = do spawntask_thread {
 914                     assert!(port_cell.take().recv() == ~10);
 915                 };
 916                 thread1.join();
 917                 thread2.join();
 918             }
 919         }
 920     }
 921
 922     #[test]
 923     fn stream_send_recv_stress() {
 924         do stress_factor().times {
 925             do run_in_mt_newsched_task {
 926                 let (port, chan) = stream::<~int>();
 927
 928                 send(chan, 0);
 929                 recv(port, 0);
 930
 931                 fn send(chan: Chan<~int>, i: int) {
 932                     if i == 10 { return }
 933
 934                     let chan_cell = Cell::new(chan);
 935                     do spawntask_random {
 936                         let chan = chan_cell.take();
 937                         chan.send(~i);
 938                         send(chan, i + 1);
 939                     }
 940                 }
 941
 942                 fn recv(port: Port<~int>, i: int) {
 943                     if i == 10 { return }
 944
 945                     let port_cell = Cell::new(port);
 946                     do spawntask_random {
 947                         let port = port_cell.take();
 948                         assert!(port.recv() == ~i);
 949                         recv(port, i + 1);
 950                     };
 951                 }
 952             }
 953         }
 954     }
 955
 956     #[test]
 957     fn recv_a_lot() {
 958         // Regression test that we don't run out of stack in scheduler context
 959         do run_in_newsched_task {
 960             let (port, chan) = stream();
 961             do 10000.times { chan.send(()) }
 962             do 10000.times { port.recv() }
 963         }
 964     }
 965
 966     #[test]
 967     fn shared_chan_stress() {
 968         do run_in_mt_newsched_task {
 969             let (port, chan) = stream();
 970             let chan = SharedChan::new(chan);
 971             let total = stress_factor() + 100;
 972             do total.times {
 973                 let chan_clone = chan.clone();
 974                 do spawntask_random {
 975                     chan_clone.send(());
 976                 }
 977             }
 978
 979             do total.times {
 980                 port.recv();
 981             }
 982         }
 983     }
 984
 985     #[test]
 986     fn shared_port_stress() {
 987         do run_in_mt_newsched_task {
 988             // XXX: Removing these type annotations causes an ICE
 989             let (end_port, end_chan) = stream::<()>();
 990             let (port, chan) = stream::<()>();
 991             let end_chan = SharedChan::new(end_chan);
 992             let port = SharedPort::new(port);
 993             let total = stress_factor() + 100;
 994             do total.times {
 995                 let end_chan_clone = end_chan.clone();
 996                 let port_clone = port.clone();
 997                 do spawntask_random {
 998                     port_clone.recv();
 999                     end_chan_clone.send(());
1000                 }
1001             }
1002
1003             do total.times {
1004                 chan.send(());
1005             }
1006
1007             do total.times {
1008                 end_port.recv();
1009             }
1010         }
1011     }
1012
1013     #[test]
1014     fn shared_port_close_simple() {
1015         do run_in_mt_newsched_task {
1016             let (port, chan) = stream::<()>();
1017             let port = SharedPort::new(port);
1018             { let _chan = chan; }
1019             assert!(port.try_recv().is_none());
1020         }
1021     }
1022
1023     #[test]
1024     fn shared_port_close() {
1025         do run_in_mt_newsched_task {
1026             let (end_port, end_chan) = stream::<bool>();
1027             let (port, chan) = stream::<()>();
1028             let end_chan = SharedChan::new(end_chan);
1029             let port = SharedPort::new(port);
1030             let chan = SharedChan::new(chan);
1031             let send_total = 10;
1032             let recv_total = 20;
1033             do spawntask_random {
1034                 do send_total.times {
1035                     let chan_clone = chan.clone();
1036                     do spawntask_random {
1037                         chan_clone.send(());
1038                     }
1039                 }
1040             }
1041             let end_chan_clone = end_chan.clone();
1042             do spawntask_random {
1043                 do recv_total.times {
1044                     let port_clone = port.clone();
1045                     let end_chan_clone = end_chan_clone.clone();
1046                     do spawntask_random {
1047                         let recvd = port_clone.try_recv().is_some();
1048                         end_chan_clone.send(recvd);
1049                     }
1050                 }
1051             }
1052
1053             let mut recvd = 0;
1054             do recv_total.times {
1055                 recvd += if end_port.recv() { 1 } else { 0 };
1056             }
1057
1058             assert!(recvd == send_total);
1059         }
1060     }
1061
1062     #[test]
1063     fn megapipe_stress() {
1064         use rand;
1065         use rand::RngUtil;
1066
1067         do run_in_mt_newsched_task {
1068             let (end_port, end_chan) = stream::<()>();
1069             let end_chan = SharedChan::new(end_chan);
1070             let pipe = megapipe();
1071             let total = stress_factor() + 10;
1072             let mut rng = rand::rng();
1073             do total.times {
1074                 let msgs = rng.gen_uint_range(0, 10);
1075                 let pipe_clone = pipe.clone();
1076                 let end_chan_clone = end_chan.clone();
1077                 do spawntask_random {
1078                     do msgs.times {
1079                         pipe_clone.send(());
1080                     }
1081                     do msgs.times {
1082                         pipe_clone.recv();
1083                     }
1084                 }
1085
1086                 end_chan_clone.send(());
1087             }
1088
1089             do total.times {
1090                 end_port.recv();
1091             }
1092         }
1093     }
1094
1095     #[test]
1096     fn send_deferred() {
1097         use unstable::sync::atomically;
1098
1099         // Tests no-rescheduling of send_deferred on all types of channels.
1100         do run_in_newsched_task {
1101             let (pone, cone) = oneshot();
1102             let (pstream, cstream) = stream();
1103             let (pshared, cshared) = stream();
1104             let cshared = SharedChan::new(cshared);
1105             let mp = megapipe();
1106
1107             let pone = Cell::new(pone);
1108             do spawntask { pone.take().recv(); }
1109             let pstream = Cell::new(pstream);
1110             do spawntask { pstream.take().recv(); }
1111             let pshared = Cell::new(pshared);
1112             do spawntask { pshared.take().recv(); }
1113             let p_mp = Cell::new(mp.clone());
1114             do spawntask { p_mp.take().recv(); }
1115
1116             let cs = Cell::new((cone, cstream, cshared, mp));
1117             unsafe {
1118                 do atomically {
1119                     let (cone, cstream, cshared, mp) = cs.take();
1120                     cone.send_deferred(());
1121                     cstream.send_deferred(());
1122                     cshared.send_deferred(());
1123                     mp.send_deferred(());
1124                 }
1125             }
1126         }
1127     }
1128
1129 }