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