src/libsync/comm/sync.rs

   1 // Copyright 2014 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 /// Synchronous channels/ports
  12 ///
  13 /// This channel implementation differs significantly from the asynchronous
  14 /// implementations found next to it (oneshot/stream/share). This is an
  15 /// implementation of a synchronous, bounded buffer channel.
  16 ///
  17 /// Each channel is created with some amount of backing buffer, and sends will
  18 /// *block* until buffer space becomes available. A buffer size of 0 is valid,
  19 /// which means that every successful send is paired with a successful recv.
  20 ///
  21 /// This flavor of channels defines a new `send_opt` method for channels which
  22 /// is the method by which a message is sent but the task does not panic if it
  23 /// cannot be delivered.
  24 ///
  25 /// Another major difference is that send() will *always* return back the data
  26 /// if it couldn't be sent. This is because it is deterministically known when
  27 /// the data is received and when it is not received.
  28 ///
  29 /// Implementation-wise, it can all be summed up with "use a mutex plus some
  30 /// logic". The mutex used here is an OS native mutex, meaning that no user code
  31 /// is run inside of the mutex (to prevent context switching). This
  32 /// implementation shares almost all code for the buffered and unbuffered cases
  33 /// of a synchronous channel. There are a few branches for the unbuffered case,
  34 /// but they're mostly just relevant to blocking senders.
  35
  36 use core::prelude::*;
  37
  38 use alloc::boxed::Box;
  39 use collections::Vec;
  40 use core::mem;
  41 use core::cell::UnsafeCell;
  42 use rustrt::local::Local;
  43 use rustrt::mutex::{NativeMutex, LockGuard};
  44 use rustrt::task::{Task, BlockedTask};
  45
  46 use atomic;
  47
  48 pub struct Packet<T> {
  49     /// Only field outside of the mutex. Just done for kicks, but mainly because
  50     /// the other shared channel already had the code implemented
  51     channels: atomic::AtomicUint,
  52
  53     /// The state field is protected by this mutex
  54     lock: NativeMutex,
  55     state: UnsafeCell<State<T>>,
  56 }
  57
  58 struct State<T> {
  59     disconnected: bool, // Is the channel disconnected yet?
  60     queue: Queue,       // queue of senders waiting to send data
  61     blocker: Blocker,   // currently blocked task on this channel
  62     buf: Buffer<T>,     // storage for buffered messages
  63     cap: uint,          // capacity of this channel
  64
  65     /// A curious flag used to indicate whether a sender failed or succeeded in
  66     /// blocking. This is used to transmit information back to the task that it
  67     /// must dequeue its message from the buffer because it was not received.
  68     /// This is only relevant in the 0-buffer case. This obviously cannot be
  69     /// safely constructed, but it's guaranteed to always have a valid pointer
  70     /// value.
  71     canceled: Option<&'static mut bool>,
  72 }
  73
  74 /// Possible flavors of tasks who can be blocked on this channel.
  75 enum Blocker {
  76     BlockedSender(BlockedTask),
  77     BlockedReceiver(BlockedTask),
  78     NoneBlocked
  79 }
  80
  81 /// Simple queue for threading tasks together. Nodes are stack-allocated, so
  82 /// this structure is not safe at all
  83 struct Queue {
  84     head: *mut Node,
  85     tail: *mut Node,
  86 }
  87
  88 struct Node {
  89     task: Option<BlockedTask>,
  90     next: *mut Node,
  91 }
  92
  93 /// A simple ring-buffer
  94 struct Buffer<T> {
  95     buf: Vec<Option<T>>,
  96     start: uint,
  97     size: uint,
  98 }
  99
 100 #[deriving(Show)]
 101 pub enum Failure {
 102     Empty,
 103     Disconnected,
 104 }
 105
 106 /// Atomically blocks the current task, placing it into `slot`, unlocking `lock`
 107 /// in the meantime. This re-locks the mutex upon returning.
 108 fn wait(slot: &mut Blocker, f: fn(BlockedTask) -> Blocker,
 109         lock: &NativeMutex) {
 110     let me: Box<Task> = Local::take();
 111     me.deschedule(1, |task| {
 112         match mem::replace(slot, f(task)) {
 113             NoneBlocked => {}
 114             _ => unreachable!(),
 115         }
 116         unsafe { lock.unlock_noguard(); }
 117         Ok(())
 118     });
 119     unsafe { lock.lock_noguard(); }
 120 }
 121
 122 /// Wakes up a task, dropping the lock at the correct time
 123 fn wakeup(task: BlockedTask, guard: LockGuard) {
 124     // We need to be careful to wake up the waiting task *outside* of the mutex
 125     // in case it incurs a context switch.
 126     mem::drop(guard);
 127     task.wake().map(|t| t.reawaken());
 128 }
 129
 130 impl<T: Send> Packet<T> {
 131     pub fn new(cap: uint) -> Packet<T> {
 132         Packet {
 133             channels: atomic::AtomicUint::new(1),
 134             lock: unsafe { NativeMutex::new() },
 135             state: UnsafeCell::new(State {
 136                 disconnected: false,
 137                 blocker: NoneBlocked,
 138                 cap: cap,
 139                 canceled: None,
 140                 queue: Queue {
 141                     head: 0 as *mut Node,
 142                     tail: 0 as *mut Node,
 143                 },
 144                 buf: Buffer {
 145                     buf: Vec::from_fn(cap + if cap == 0 {1} else {0}, |_| None),
 146                     start: 0,
 147                     size: 0,
 148                 },
 149             }),
 150         }
 151     }
 152
 153     // Locks this channel, returning a guard for the state and the mutable state
 154     // itself. Care should be taken to ensure that the state does not escape the
 155     // guard!
 156     //
 157     // Note that we're ok promoting an & reference to an &mut reference because
 158     // the lock ensures that we're the only ones in the world with a pointer to
 159     // the state.
 160     fn lock<'a>(&'a self) -> (LockGuard<'a>, &'a mut State<T>) {
 161         unsafe {
 162             let guard = self.lock.lock();
 163             (guard, &mut *self.state.get())
 164         }
 165     }
 166
 167     pub fn send(&self, t: T) -> Result<(), T> {
 168         let (guard, state) = self.lock();
 169
 170         // wait for a slot to become available, and enqueue the data
 171         while !state.disconnected && state.buf.size() == state.buf.cap() {
 172             state.queue.enqueue(&self.lock);
 173         }
 174         if state.disconnected { return Err(t) }
 175         state.buf.enqueue(t);
 176
 177         match mem::replace(&mut state.blocker, NoneBlocked) {
 178             // if our capacity is 0, then we need to wait for a receiver to be
 179             // available to take our data. After waiting, we check again to make
 180             // sure the port didn't go away in the meantime. If it did, we need
 181             // to hand back our data.
 182             NoneBlocked if state.cap == 0 => {
 183                 let mut canceled = false;
 184                 assert!(state.canceled.is_none());
 185                 state.canceled = Some(unsafe { mem::transmute(&mut canceled) });
 186                 wait(&mut state.blocker, BlockedSender, &self.lock);
 187                 if canceled {Err(state.buf.dequeue())} else {Ok(())}
 188             }
 189
 190             // success, we buffered some data
 191             NoneBlocked => Ok(()),
 192
 193             // success, someone's about to receive our buffered data.
 194             BlockedReceiver(task) => { wakeup(task, guard); Ok(()) }
 195
 196             BlockedSender(..) => panic!("lolwut"),
 197         }
 198     }
 199
 200     pub fn try_send(&self, t: T) -> Result<(), super::TrySendError<T>> {
 201         let (guard, state) = self.lock();
 202         if state.disconnected {
 203             Err(super::RecvDisconnected(t))
 204         } else if state.buf.size() == state.buf.cap() {
 205             Err(super::Full(t))
 206         } else if state.cap == 0 {
 207             // With capacity 0, even though we have buffer space we can't
 208             // transfer the data unless there's a receiver waiting.
 209             match mem::replace(&mut state.blocker, NoneBlocked) {
 210                 NoneBlocked => Err(super::Full(t)),
 211                 BlockedSender(..) => unreachable!(),
 212                 BlockedReceiver(task) => {
 213                     state.buf.enqueue(t);
 214                     wakeup(task, guard);
 215                     Ok(())
 216                 }
 217             }
 218         } else {
 219             // If the buffer has some space and the capacity isn't 0, then we
 220             // just enqueue the data for later retrieval, ensuring to wake up
 221             // any blocked receiver if there is one.
 222             assert!(state.buf.size() < state.buf.cap());
 223             state.buf.enqueue(t);
 224             match mem::replace(&mut state.blocker, NoneBlocked) {
 225                 BlockedReceiver(task) => wakeup(task, guard),
 226                 NoneBlocked => {}
 227                 BlockedSender(..) => unreachable!(),
 228             }
 229             Ok(())
 230         }
 231     }
 232
 233     // Receives a message from this channel
 234     //
 235     // When reading this, remember that there can only ever be one receiver at
 236     // time.
 237     pub fn recv(&self) -> Result<T, ()> {
 238         let (guard, state) = self.lock();
 239
 240         // Wait for the buffer to have something in it. No need for a while loop
 241         // because we're the only receiver.
 242         let mut waited = false;
 243         if !state.disconnected && state.buf.size() == 0 {
 244             wait(&mut state.blocker, BlockedReceiver, &self.lock);
 245             waited = true;
 246         }
 247         if state.disconnected && state.buf.size() == 0 { return Err(()) }
 248
 249         // Pick up the data, wake up our neighbors, and carry on
 250         assert!(state.buf.size() > 0);
 251         let ret = state.buf.dequeue();
 252         self.wakeup_senders(waited, guard, state);
 253         return Ok(ret);
 254     }
 255
 256     pub fn try_recv(&self) -> Result<T, Failure> {
 257         let (guard, state) = self.lock();
 258
 259         // Easy cases first
 260         if state.disconnected { return Err(Disconnected) }
 261         if state.buf.size() == 0 { return Err(Empty) }
 262
 263         // Be sure to wake up neighbors
 264         let ret = Ok(state.buf.dequeue());
 265         self.wakeup_senders(false, guard, state);
 266
 267         return ret;
 268     }
 269
 270     // Wake up pending senders after some data has been received
 271     //
 272     // * `waited` - flag if the receiver blocked to receive some data, or if it
 273     //              just picked up some data on the way out
 274     // * `guard` - the lock guard that is held over this channel's lock
 275     fn wakeup_senders(&self, waited: bool,
 276                       guard: LockGuard,
 277                       state: &mut State<T>) {
 278         let pending_sender1: Option<BlockedTask> = state.queue.dequeue();
 279
 280         // If this is a no-buffer channel (cap == 0), then if we didn't wait we
 281         // need to ACK the sender. If we waited, then the sender waking us up
 282         // was already the ACK.
 283         let pending_sender2 = if state.cap == 0 && !waited {
 284             match mem::replace(&mut state.blocker, NoneBlocked) {
 285                 NoneBlocked => None,
 286                 BlockedReceiver(..) => unreachable!(),
 287                 BlockedSender(task) => {
 288                     state.canceled.take();
 289                     Some(task)
 290                 }
 291             }
 292         } else {
 293             None
 294         };
 295         mem::drop((state, guard));
 296
 297         // only outside of the lock do we wake up the pending tasks
 298         pending_sender1.map(|t| t.wake().map(|t| t.reawaken()));
 299         pending_sender2.map(|t| t.wake().map(|t| t.reawaken()));
 300     }
 301
 302     // Prepares this shared packet for a channel clone, essentially just bumping
 303     // a refcount.
 304     pub fn clone_chan(&self) {
 305         self.channels.fetch_add(1, atomic::SeqCst);
 306     }
 307
 308     pub fn drop_chan(&self) {
 309         // Only flag the channel as disconnected if we're the last channel
 310         match self.channels.fetch_sub(1, atomic::SeqCst) {
 311             1 => {}
 312             _ => return
 313         }
 314
 315         // Not much to do other than wake up a receiver if one's there
 316         let (guard, state) = self.lock();
 317         if state.disconnected { return }
 318         state.disconnected = true;
 319         match mem::replace(&mut state.blocker, NoneBlocked) {
 320             NoneBlocked => {}
 321             BlockedSender(..) => unreachable!(),
 322             BlockedReceiver(task) => wakeup(task, guard),
 323         }
 324     }
 325
 326     pub fn drop_port(&self) {
 327         let (guard, state) = self.lock();
 328
 329         if state.disconnected { return }
 330         state.disconnected = true;
 331
 332         // If the capacity is 0, then the sender may want its data back after
 333         // we're disconnected. Otherwise it's now our responsibility to destroy
 334         // the buffered data. As with many other portions of this code, this
 335         // needs to be careful to destroy the data *outside* of the lock to
 336         // prevent deadlock.
 337         let _data = if state.cap != 0 {
 338             mem::replace(&mut state.buf.buf, Vec::new())
 339         } else {
 340             Vec::new()
 341         };
 342         let mut queue = mem::replace(&mut state.queue, Queue {
 343             head: 0 as *mut Node,
 344             tail: 0 as *mut Node,
 345         });
 346
 347         let waiter = match mem::replace(&mut state.blocker, NoneBlocked) {
 348             NoneBlocked => None,
 349             BlockedSender(task) => {
 350                 *state.canceled.take().unwrap() = true;
 351                 Some(task)
 352             }
 353             BlockedReceiver(..) => unreachable!(),
 354         };
 355         mem::drop((state, guard));
 356
 357         loop {
 358             match queue.dequeue() {
 359                 Some(task) => { task.wake().map(|t| t.reawaken()); }
 360                 None => break,
 361             }
 362         }
 363         waiter.map(|t| t.wake().map(|t| t.reawaken()));
 364     }
 365
 366     ////////////////////////////////////////////////////////////////////////////
 367     // select implementation
 368     ////////////////////////////////////////////////////////////////////////////
 369
 370     // If Ok, the value is whether this port has data, if Err, then the upgraded
 371     // port needs to be checked instead of this one.
 372     pub fn can_recv(&self) -> bool {
 373         let (_g, state) = self.lock();
 374         state.disconnected || state.buf.size() > 0
 375     }
 376
 377     // Attempts to start selection on this port. This can either succeed or fail
 378     // because there is data waiting.
 379     pub fn start_selection(&self, task: BlockedTask) -> Result<(), BlockedTask>{
 380         let (_g, state) = self.lock();
 381         if state.disconnected || state.buf.size() > 0 {
 382             Err(task)
 383         } else {
 384             match mem::replace(&mut state.blocker, BlockedReceiver(task)) {
 385                 NoneBlocked => {}
 386                 BlockedSender(..) => unreachable!(),
 387                 BlockedReceiver(..) => unreachable!(),
 388             }
 389             Ok(())
 390         }
 391     }
 392
 393     // Remove a previous selecting task from this port. This ensures that the
 394     // blocked task will no longer be visible to any other threads.
 395     //
 396     // The return value indicates whether there's data on this port.
 397     pub fn abort_selection(&self) -> bool {
 398         let (_g, state) = self.lock();
 399         match mem::replace(&mut state.blocker, NoneBlocked) {
 400             NoneBlocked => true,
 401             BlockedSender(task) => {
 402                 state.blocker = BlockedSender(task);
 403                 true
 404             }
 405             BlockedReceiver(task) => { task.trash(); false }
 406         }
 407     }
 408 }
 409
 410 #[unsafe_destructor]
 411 impl<T: Send> Drop for Packet<T> {
 412     fn drop(&mut self) {
 413         assert_eq!(self.channels.load(atomic::SeqCst), 0);
 414         let (_g, state) = self.lock();
 415         assert!(state.queue.dequeue().is_none());
 416         assert!(state.canceled.is_none());
 417     }
 418 }
 419
 420
 421 ////////////////////////////////////////////////////////////////////////////////
 422 // Buffer, a simple ring buffer backed by Vec<T>
 423 ////////////////////////////////////////////////////////////////////////////////
 424
 425 impl<T> Buffer<T> {
 426     fn enqueue(&mut self, t: T) {
 427         let pos = (self.start + self.size) % self.buf.len();
 428         self.size += 1;
 429         let prev = mem::replace(self.buf.get_mut(pos), Some(t));
 430         assert!(prev.is_none());
 431     }
 432
 433     fn dequeue(&mut self) -> T {
 434         let start = self.start;
 435         self.size -= 1;
 436         self.start = (self.start + 1) % self.buf.len();
 437         self.buf.get_mut(start).take().unwrap()
 438     }
 439
 440     fn size(&self) -> uint { self.size }
 441     fn cap(&self) -> uint { self.buf.len() }
 442 }
 443
 444 ////////////////////////////////////////////////////////////////////////////////
 445 // Queue, a simple queue to enqueue tasks with (stack-allocated nodes)
 446 ////////////////////////////////////////////////////////////////////////////////
 447
 448 impl Queue {
 449     fn enqueue(&mut self, lock: &NativeMutex) {
 450         let task: Box<Task> = Local::take();
 451         let mut node = Node {
 452             task: None,
 453             next: 0 as *mut Node,
 454         };
 455         task.deschedule(1, |task| {
 456             node.task = Some(task);
 457             if self.tail.is_null() {
 458                 self.head = &mut node as *mut Node;
 459                 self.tail = &mut node as *mut Node;
 460             } else {
 461                 unsafe {
 462                     (*self.tail).next = &mut node as *mut Node;
 463                     self.tail = &mut node as *mut Node;
 464                 }
 465             }
 466             unsafe { lock.unlock_noguard(); }
 467             Ok(())
 468         });
 469         unsafe { lock.lock_noguard(); }
 470         assert!(node.next.is_null());
 471     }
 472
 473     fn dequeue(&mut self) -> Option<BlockedTask> {
 474         if self.head.is_null() {
 475             return None
 476         }
 477         let node = self.head;
 478         self.head = unsafe { (*node).next };
 479         if self.head.is_null() {
 480             self.tail = 0 as *mut Node;
 481         }
 482         unsafe {
 483             (*node).next = 0 as *mut Node;
 484             Some((*node).task.take().unwrap())
 485         }
 486     }
 487 }