src/librustuv/homing.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 //! Homing I/O implementation
  12 //!
  13 //! In libuv, whenever a handle is created on an I/O loop it is illegal to use
  14 //! that handle outside of that I/O loop. We use libuv I/O with our green
  15 //! scheduler, and each green scheduler corresponds to a different I/O loop on a
  16 //! different OS thread. Green tasks are also free to roam among schedulers,
  17 //! which implies that it is possible to create an I/O handle on one event loop
  18 //! and then attempt to use it on another.
  19 //!
  20 //! In order to solve this problem, this module implements the notion of a
  21 //! "homing operation" which will transplant a task from its currently running
  22 //! scheduler back onto the original I/O loop. This is accomplished entirely at
  23 //! the librustuv layer with very little cooperation from the scheduler (which
  24 //! we don't even know exists technically).
  25 //!
  26 //! These homing operations are completed by first realizing that we're on the
  27 //! wrong I/O loop, then descheduling ourselves, sending ourselves to the
  28 //! correct I/O loop, and then waking up the I/O loop in order to process its
  29 //! local queue of tasks which need to run.
  30 //!
  31 //! This enqueueing is done with a concurrent queue from libstd, and the
  32 //! signalling is achieved with an async handle.
  33
  34 use std::rt::local::Local;
  35 use std::rt::rtio::LocalIo;
  36 use std::rt::task::{Task, BlockedTask};
  37
  38 use ForbidUnwind;
  39 use queue::{Queue, QueuePool};
  40
  41 /// A handle to a remote libuv event loop. This handle will keep the event loop
  42 /// alive while active in order to ensure that a homing operation can always be
  43 /// completed.
  44 ///
  45 /// Handles are clone-able in order to derive new handles from existing handles
  46 /// (very useful for when accepting a socket from a server).
  47 pub struct HomeHandle {
  48     priv queue: Queue,
  49     priv id: uint,
  50 }
  51
  52 impl HomeHandle {
  53     pub fn new(id: uint, pool: &mut QueuePool) -> HomeHandle {
  54         HomeHandle { queue: pool.queue(), id: id }
  55     }
  56
  57     fn send(&mut self, task: BlockedTask) {
  58         self.queue.push(task);
  59     }
  60 }
  61
  62 impl Clone for HomeHandle {
  63     fn clone(&self) -> HomeHandle {
  64         HomeHandle {
  65             queue: self.queue.clone(),
  66             id: self.id,
  67         }
  68     }
  69 }
  70
  71 pub trait HomingIO {
  72     fn home<'r>(&'r mut self) -> &'r mut HomeHandle;
  73
  74     /// This function will move tasks to run on their home I/O scheduler. Note
  75     /// that this function does *not* pin the task to the I/O scheduler, but
  76     /// rather it simply moves it to running on the I/O scheduler.
  77     fn go_to_IO_home(&mut self) -> uint {
  78         let _f = ForbidUnwind::new("going home");
  79
  80         let mut cur_task: ~Task = Local::take();
  81         let cur_loop_id = {
  82             let mut io = cur_task.local_io().expect("libuv must have I/O");
  83             io.get().id()
  84         };
  85
  86         // Try at all costs to avoid the homing operation because it is quite
  87         // expensive. Hence, we only deschedule/send if we're not on the correct
  88         // event loop. If we're already on the home event loop, then we're good
  89         // to go (remember we have no preemption, so we're guaranteed to stay on
  90         // this event loop as long as we avoid the scheduler).
  91         if cur_loop_id != self.home().id {
  92             cur_task.deschedule(1, |task| {
  93                 self.home().send(task);
  94                 Ok(())
  95             });
  96
  97             // Once we wake up, assert that we're in the right location
  98             let cur_loop_id = {
  99                 let mut io = LocalIo::borrow().expect("libuv must have I/O");
 100                 io.get().id()
 101             };
 102             assert_eq!(cur_loop_id, self.home().id);
 103
 104             cur_loop_id
 105         } else {
 106             Local::put(cur_task);
 107             cur_loop_id
 108         }
 109     }
 110
 111     /// Fires a single homing missile, returning another missile targeted back
 112     /// at the original home of this task. In other words, this function will
 113     /// move the local task to its I/O scheduler and then return an RAII wrapper
 114     /// which will return the task home.
 115     fn fire_homing_missile(&mut self) -> HomingMissile {
 116         HomingMissile { io_home: self.go_to_IO_home() }
 117     }
 118 }
 119
 120 /// After a homing operation has been completed, this will return the current
 121 /// task back to its appropriate home (if applicable). The field is used to
 122 /// assert that we are where we think we are.
 123 struct HomingMissile {
 124     priv io_home: uint,
 125 }
 126
 127 impl HomingMissile {
 128     /// Check at runtime that the task has *not* transplanted itself to a
 129     /// different I/O loop while executing.
 130     pub fn check(&self, msg: &'static str) {
 131         let mut io = LocalIo::borrow().expect("libuv must have I/O");
 132         assert!(io.get().id() == self.io_home, "{}", msg);
 133     }
 134 }
 135
 136 impl Drop for HomingMissile {
 137     fn drop(&mut self) {
 138         let _f = ForbidUnwind::new("leaving home");
 139
 140         // It would truly be a sad day if we had moved off the home I/O
 141         // scheduler while we were doing I/O.
 142         self.check("task moved away from the home scheduler");
 143     }
 144 }
 145
 146 #[cfg(test)]
 147 mod test {
 148     use green::sched;
 149     use green::{SchedPool, PoolConfig};
 150     use std::rt::rtio::RtioUdpSocket;
 151     use std::io::test::next_test_ip4;
 152     use std::task::TaskOpts;
 153
 154     use net::UdpWatcher;
 155     use super::super::local_loop;
 156
 157     // On one thread, create a udp socket. Then send that socket to another
 158     // thread and destroy the socket on the remote thread. This should make sure
 159     // that homing kicks in for the socket to go back home to the original
 160     // thread, close itself, and then come back to the last thread.
 161     #[test]
 162     fn test_homing_closes_correctly() {
 163         let (port, chan) = Chan::new();
 164         let mut pool = SchedPool::new(PoolConfig {
 165             threads: 1,
 166             event_loop_factory: None,
 167         });
 168
 169         do pool.spawn(TaskOpts::new()) {
 170             let listener = UdpWatcher::bind(local_loop(), next_test_ip4());
 171             chan.send(listener.unwrap());
 172         }
 173
 174         let task = do pool.task(TaskOpts::new()) {
 175             port.recv();
 176         };
 177         pool.spawn_sched().send(sched::TaskFromFriend(task));
 178
 179         pool.shutdown();
 180     }
 181
 182     #[test]
 183     fn test_homing_read() {
 184         let (port, chan) = Chan::new();
 185         let mut pool = SchedPool::new(PoolConfig {
 186             threads: 1,
 187             event_loop_factory: None,
 188         });
 189
 190         do pool.spawn(TaskOpts::new()) {
 191             let addr1 = next_test_ip4();
 192             let addr2 = next_test_ip4();
 193             let listener = UdpWatcher::bind(local_loop(), addr2);
 194             chan.send((listener.unwrap(), addr1));
 195             let mut listener = UdpWatcher::bind(local_loop(), addr1).unwrap();
 196             listener.sendto([1, 2, 3, 4], addr2);
 197         }
 198
 199         let task = do pool.task(TaskOpts::new()) {
 200             let (mut watcher, addr) = port.recv();
 201             let mut buf = [0, ..10];
 202             assert_eq!(watcher.recvfrom(buf).unwrap(), (4, addr));
 203         };
 204         pool.spawn_sched().send(sched::TaskFromFriend(task));
 205
 206         pool.shutdown();
 207     }
 208 }