]> git.lizzy.rs Git - rust.git/blob - src/libstd/sync/mpsc/select.rs
43554d7c335a4c4471084162593651789fcc2c64
[rust.git] / src / libstd / sync / mpsc / select.rs
1 // Copyright 2013-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 //! Selection over an array of receivers
12 //!
13 //! This module contains the implementation machinery necessary for selecting
14 //! over a number of receivers. One large goal of this module is to provide an
15 //! efficient interface to selecting over any receiver of any type.
16 //!
17 //! This is achieved through an architecture of a "receiver set" in which
18 //! receivers are added to a set and then the entire set is waited on at once.
19 //! The set can be waited on multiple times to prevent re-adding each receiver
20 //! to the set.
21 //!
22 //! Usage of this module is currently encouraged to go through the use of the
23 //! `select!` macro. This macro allows naturally binding of variables to the
24 //! received values of receivers in a much more natural syntax then usage of the
25 //! `Select` structure directly.
26 //!
27 //! # Example
28 //!
29 //! ```rust
30 //! use std::sync::mpsc::channel;
31 //!
32 //! let (tx1, rx1) = channel();
33 //! let (tx2, rx2) = channel();
34 //!
35 //! tx1.send(1i).unwrap();
36 //! tx2.send(2i).unwrap();
37 //!
38 //! select! {
39 //!     val = rx1.recv() => {
40 //!         assert_eq!(val.unwrap(), 1i);
41 //!     },
42 //!     val = rx2.recv() => {
43 //!         assert_eq!(val.unwrap(), 2i);
44 //!     }
45 //! }
46 //! ```
47
48 #![allow(dead_code)]
49 #![experimental = "This implementation, while likely sufficient, is unsafe and \
50                    likely to be error prone. At some point in the future this \
51                    module will likely be replaced, and it is currently \
52                    unknown how much API breakage that will cause. The ability \
53                    to select over a number of channels will remain forever, \
54                    but no guarantees beyond this are being made"]
55
56
57 use core::prelude::*;
58
59 use core::cell::Cell;
60 use core::kinds::marker;
61 use core::mem;
62 use core::uint;
63
64 use sync::mpsc::{Receiver, RecvError};
65 use sync::mpsc::blocking::{mod, SignalToken};
66
67 /// The "receiver set" of the select interface. This structure is used to manage
68 /// a set of receivers which are being selected over.
69 pub struct Select {
70     head: *mut Handle<'static, ()>,
71     tail: *mut Handle<'static, ()>,
72     next_id: Cell<uint>,
73     marker1: marker::NoSend,
74 }
75
76 /// A handle to a receiver which is currently a member of a `Select` set of
77 /// receivers.  This handle is used to keep the receiver in the set as well as
78 /// interact with the underlying receiver.
79 pub struct Handle<'rx, T:'rx> {
80     /// The ID of this handle, used to compare against the return value of
81     /// `Select::wait()`
82     id: uint,
83     selector: &'rx Select,
84     next: *mut Handle<'static, ()>,
85     prev: *mut Handle<'static, ()>,
86     added: bool,
87     packet: &'rx (Packet+'rx),
88
89     // due to our fun transmutes, we be sure to place this at the end. (nothing
90     // previous relies on T)
91     rx: &'rx Receiver<T>,
92 }
93
94 struct Packets { cur: *mut Handle<'static, ()> }
95
96 #[doc(hidden)]
97 #[deriving(PartialEq)]
98 pub enum StartResult {
99     Installed,
100     Abort,
101 }
102
103 #[doc(hidden)]
104 pub trait Packet {
105     fn can_recv(&self) -> bool;
106     fn start_selection(&self, token: SignalToken) -> StartResult;
107     fn abort_selection(&self) -> bool;
108 }
109
110 impl Select {
111     /// Creates a new selection structure. This set is initially empty and
112     /// `wait` will panic!() if called.
113     ///
114     /// Usage of this struct directly can sometimes be burdensome, and usage is
115     /// rather much easier through the `select!` macro.
116     pub fn new() -> Select {
117         Select {
118             marker1: marker::NoSend,
119             head: 0 as *mut Handle<'static, ()>,
120             tail: 0 as *mut Handle<'static, ()>,
121             next_id: Cell::new(1),
122         }
123     }
124
125     /// Creates a new handle into this receiver set for a new receiver. Note
126     /// that this does *not* add the receiver to the receiver set, for that you
127     /// must call the `add` method on the handle itself.
128     pub fn handle<'a, T: Send>(&'a self, rx: &'a Receiver<T>) -> Handle<'a, T> {
129         let id = self.next_id.get();
130         self.next_id.set(id + 1);
131         Handle {
132             id: id,
133             selector: self,
134             next: 0 as *mut Handle<'static, ()>,
135             prev: 0 as *mut Handle<'static, ()>,
136             added: false,
137             rx: rx,
138             packet: rx,
139         }
140     }
141
142     /// Waits for an event on this receiver set. The returned value is *not* an
143     /// index, but rather an id. This id can be queried against any active
144     /// `Handle` structures (each one has an `id` method). The handle with
145     /// the matching `id` will have some sort of event available on it. The
146     /// event could either be that data is available or the corresponding
147     /// channel has been closed.
148     pub fn wait(&self) -> uint {
149         self.wait2(true)
150     }
151
152     /// Helper method for skipping the preflight checks during testing
153     fn wait2(&self, do_preflight_checks: bool) -> uint {
154         // Note that this is currently an inefficient implementation. We in
155         // theory have knowledge about all receivers in the set ahead of time,
156         // so this method shouldn't really have to iterate over all of them yet
157         // again. The idea with this "receiver set" interface is to get the
158         // interface right this time around, and later this implementation can
159         // be optimized.
160         //
161         // This implementation can be summarized by:
162         //
163         //      fn select(receivers) {
164         //          if any receiver ready { return ready index }
165         //          deschedule {
166         //              block on all receivers
167         //          }
168         //          unblock on all receivers
169         //          return ready index
170         //      }
171         //
172         // Most notably, the iterations over all of the receivers shouldn't be
173         // necessary.
174         unsafe {
175             // Stage 1: preflight checks. Look for any packets ready to receive
176             if do_preflight_checks {
177                 for handle in self.iter() {
178                     if (*handle).packet.can_recv() {
179                         return (*handle).id();
180                     }
181                 }
182             }
183
184             // Stage 2: begin the blocking process
185             //
186             // Create a number of signal tokens, and install each one
187             // sequentially until one fails. If one fails, then abort the
188             // selection on the already-installed tokens.
189             let (wait_token, signal_token) = blocking::tokens();
190             for (i, handle) in self.iter().enumerate() {
191                 match (*handle).packet.start_selection(signal_token.clone()) {
192                     StartResult::Installed => {}
193                     StartResult::Abort => {
194                         // Go back and abort the already-begun selections
195                         for handle in self.iter().take(i) {
196                             (*handle).packet.abort_selection();
197                         }
198                         return (*handle).id;
199                     }
200                 }
201             }
202
203             // Stage 3: no messages available, actually block
204             wait_token.wait();
205
206             // Stage 4: there *must* be message available; find it.
207             //
208             // Abort the selection process on each receiver. If the abort
209             // process returns `true`, then that means that the receiver is
210             // ready to receive some data. Note that this also means that the
211             // receiver may have yet to have fully read the `to_wake` field and
212             // woken us up (although the wakeup is guaranteed to fail).
213             //
214             // This situation happens in the window of where a sender invokes
215             // increment(), sees -1, and then decides to wake up the task. After
216             // all this is done, the sending thread will set `selecting` to
217             // `false`. Until this is done, we cannot return. If we were to
218             // return, then a sender could wake up a receiver which has gone
219             // back to sleep after this call to `select`.
220             //
221             // Note that it is a "fairly small window" in which an increment()
222             // views that it should wake a thread up until the `selecting` bit
223             // is set to false. For now, the implementation currently just spins
224             // in a yield loop. This is very distasteful, but this
225             // implementation is already nowhere near what it should ideally be.
226             // A rewrite should focus on avoiding a yield loop, and for now this
227             // implementation is tying us over to a more efficient "don't
228             // iterate over everything every time" implementation.
229             let mut ready_id = uint::MAX;
230             for handle in self.iter() {
231                 if (*handle).packet.abort_selection() {
232                     ready_id = (*handle).id;
233                 }
234             }
235
236             // We must have found a ready receiver
237             assert!(ready_id != uint::MAX);
238             return ready_id;
239         }
240     }
241
242     fn iter(&self) -> Packets { Packets { cur: self.head } }
243 }
244
245 impl<'rx, T: Send> Handle<'rx, T> {
246     /// Retrieve the id of this handle.
247     #[inline]
248     pub fn id(&self) -> uint { self.id }
249
250     /// Block to receive a value on the underlying receiver, returning `Some` on
251     /// success or `None` if the channel disconnects. This function has the same
252     /// semantics as `Receiver.recv`
253     pub fn recv(&mut self) -> Result<T, RecvError> { self.rx.recv() }
254
255     /// Adds this handle to the receiver set that the handle was created from. This
256     /// method can be called multiple times, but it has no effect if `add` was
257     /// called previously.
258     ///
259     /// This method is unsafe because it requires that the `Handle` is not moved
260     /// while it is added to the `Select` set.
261     pub unsafe fn add(&mut self) {
262         if self.added { return }
263         let selector: &mut Select = mem::transmute(&*self.selector);
264         let me: *mut Handle<'static, ()> = mem::transmute(&*self);
265
266         if selector.head.is_null() {
267             selector.head = me;
268             selector.tail = me;
269         } else {
270             (*me).prev = selector.tail;
271             assert!((*me).next.is_null());
272             (*selector.tail).next = me;
273             selector.tail = me;
274         }
275         self.added = true;
276     }
277
278     /// Removes this handle from the `Select` set. This method is unsafe because
279     /// it has no guarantee that the `Handle` was not moved since `add` was
280     /// called.
281     pub unsafe fn remove(&mut self) {
282         if !self.added { return }
283
284         let selector: &mut Select = mem::transmute(&*self.selector);
285         let me: *mut Handle<'static, ()> = mem::transmute(&*self);
286
287         if self.prev.is_null() {
288             assert_eq!(selector.head, me);
289             selector.head = self.next;
290         } else {
291             (*self.prev).next = self.next;
292         }
293         if self.next.is_null() {
294             assert_eq!(selector.tail, me);
295             selector.tail = self.prev;
296         } else {
297             (*self.next).prev = self.prev;
298         }
299
300         self.next = 0 as *mut Handle<'static, ()>;
301         self.prev = 0 as *mut Handle<'static, ()>;
302
303         self.added = false;
304     }
305 }
306
307 #[unsafe_destructor]
308 impl Drop for Select {
309     fn drop(&mut self) {
310         assert!(self.head.is_null());
311         assert!(self.tail.is_null());
312     }
313 }
314
315 #[unsafe_destructor]
316 impl<'rx, T: Send> Drop for Handle<'rx, T> {
317     fn drop(&mut self) {
318         unsafe { self.remove() }
319     }
320 }
321
322 impl Iterator<*mut Handle<'static, ()>> for Packets {
323     fn next(&mut self) -> Option<*mut Handle<'static, ()>> {
324         if self.cur.is_null() {
325             None
326         } else {
327             let ret = Some(self.cur);
328             unsafe { self.cur = (*self.cur).next; }
329             ret
330         }
331     }
332 }
333
334 #[cfg(test)]
335 #[allow(unused_imports)]
336 mod test {
337     use prelude::v1::*;
338
339     use thread::Thread;
340     use sync::mpsc::*;
341
342     // Don't use the libstd version so we can pull in the right Select structure
343     // (std::comm points at the wrong one)
344     macro_rules! select {
345         (
346             $($name:pat = $rx:ident.$meth:ident() => $code:expr),+
347         ) => ({
348             let sel = Select::new();
349             $( let mut $rx = sel.handle(&$rx); )+
350             unsafe {
351                 $( $rx.add(); )+
352             }
353             let ret = sel.wait();
354             $( if ret == $rx.id() { let $name = $rx.$meth(); $code } else )+
355             { unreachable!() }
356         })
357     }
358
359     #[test]
360     fn smoke() {
361         let (tx1, rx1) = channel::<int>();
362         let (tx2, rx2) = channel::<int>();
363         tx1.send(1).unwrap();
364         select! {
365             foo = rx1.recv() => { assert_eq!(foo.unwrap(), 1); },
366             _bar = rx2.recv() => { panic!() }
367         }
368         tx2.send(2).unwrap();
369         select! {
370             _foo = rx1.recv() => { panic!() },
371             bar = rx2.recv() => { assert_eq!(bar.unwrap(), 2) }
372         }
373         drop(tx1);
374         select! {
375             foo = rx1.recv() => { assert!(foo.is_err()); },
376             _bar = rx2.recv() => { panic!() }
377         }
378         drop(tx2);
379         select! {
380             bar = rx2.recv() => { assert!(bar.is_err()); }
381         }
382     }
383
384     #[test]
385     fn smoke2() {
386         let (_tx1, rx1) = channel::<int>();
387         let (_tx2, rx2) = channel::<int>();
388         let (_tx3, rx3) = channel::<int>();
389         let (_tx4, rx4) = channel::<int>();
390         let (tx5, rx5) = channel::<int>();
391         tx5.send(4).unwrap();
392         select! {
393             _foo = rx1.recv() => { panic!("1") },
394             _foo = rx2.recv() => { panic!("2") },
395             _foo = rx3.recv() => { panic!("3") },
396             _foo = rx4.recv() => { panic!("4") },
397             foo = rx5.recv() => { assert_eq!(foo.unwrap(), 4); }
398         }
399     }
400
401     #[test]
402     fn closed() {
403         let (_tx1, rx1) = channel::<int>();
404         let (tx2, rx2) = channel::<int>();
405         drop(tx2);
406
407         select! {
408             _a1 = rx1.recv() => { panic!() },
409             a2 = rx2.recv() => { assert!(a2.is_err()); }
410         }
411     }
412
413     #[test]
414     fn unblocks() {
415         let (tx1, rx1) = channel::<int>();
416         let (_tx2, rx2) = channel::<int>();
417         let (tx3, rx3) = channel::<int>();
418
419         let _t = Thread::spawn(move|| {
420             for _ in range(0u, 20) { Thread::yield_now(); }
421             tx1.send(1).unwrap();
422             rx3.recv().unwrap();
423             for _ in range(0u, 20) { Thread::yield_now(); }
424         });
425
426         select! {
427             a = rx1.recv() => { assert_eq!(a.unwrap(), 1); },
428             _b = rx2.recv() => { panic!() }
429         }
430         tx3.send(1).unwrap();
431         select! {
432             a = rx1.recv() => { assert!(a.is_err()) },
433             _b = rx2.recv() => { panic!() }
434         }
435     }
436
437     #[test]
438     fn both_ready() {
439         let (tx1, rx1) = channel::<int>();
440         let (tx2, rx2) = channel::<int>();
441         let (tx3, rx3) = channel::<()>();
442
443         let _t = Thread::spawn(move|| {
444             for _ in range(0u, 20) { Thread::yield_now(); }
445             tx1.send(1).unwrap();
446             tx2.send(2).unwrap();
447             rx3.recv().unwrap();
448         });
449
450         select! {
451             a = rx1.recv() => { assert_eq!(a.unwrap(), 1); },
452             a = rx2.recv() => { assert_eq!(a.unwrap(), 2); }
453         }
454         select! {
455             a = rx1.recv() => { assert_eq!(a.unwrap(), 1); },
456             a = rx2.recv() => { assert_eq!(a.unwrap(), 2); }
457         }
458         assert_eq!(rx1.try_recv(), Err(TryRecvError::Empty));
459         assert_eq!(rx2.try_recv(), Err(TryRecvError::Empty));
460         tx3.send(()).unwrap();
461     }
462
463     #[test]
464     fn stress() {
465         static AMT: int = 10000;
466         let (tx1, rx1) = channel::<int>();
467         let (tx2, rx2) = channel::<int>();
468         let (tx3, rx3) = channel::<()>();
469
470         let _t = Thread::spawn(move|| {
471             for i in range(0, AMT) {
472                 if i % 2 == 0 {
473                     tx1.send(i).unwrap();
474                 } else {
475                     tx2.send(i).unwrap();
476                 }
477                 rx3.recv().unwrap();
478             }
479         });
480
481         for i in range(0, AMT) {
482             select! {
483                 i1 = rx1.recv() => { assert!(i % 2 == 0 && i == i1.unwrap()); },
484                 i2 = rx2.recv() => { assert!(i % 2 == 1 && i == i2.unwrap()); }
485             }
486             tx3.send(()).unwrap();
487         }
488     }
489
490     #[test]
491     fn cloning() {
492         let (tx1, rx1) = channel::<int>();
493         let (_tx2, rx2) = channel::<int>();
494         let (tx3, rx3) = channel::<()>();
495
496         let _t = Thread::spawn(move|| {
497             rx3.recv().unwrap();
498             tx1.clone();
499             assert_eq!(rx3.try_recv(), Err(TryRecvError::Empty));
500             tx1.send(2).unwrap();
501             rx3.recv().unwrap();
502         });
503
504         tx3.send(()).unwrap();
505         select! {
506             _i1 = rx1.recv() => {},
507             _i2 = rx2.recv() => panic!()
508         }
509         tx3.send(()).unwrap();
510     }
511
512     #[test]
513     fn cloning2() {
514         let (tx1, rx1) = channel::<int>();
515         let (_tx2, rx2) = channel::<int>();
516         let (tx3, rx3) = channel::<()>();
517
518         let _t = Thread::spawn(move|| {
519             rx3.recv().unwrap();
520             tx1.clone();
521             assert_eq!(rx3.try_recv(), Err(TryRecvError::Empty));
522             tx1.send(2).unwrap();
523             rx3.recv().unwrap();
524         });
525
526         tx3.send(()).unwrap();
527         select! {
528             _i1 = rx1.recv() => {},
529             _i2 = rx2.recv() => panic!()
530         }
531         tx3.send(()).unwrap();
532     }
533
534     #[test]
535     fn cloning3() {
536         let (tx1, rx1) = channel::<()>();
537         let (tx2, rx2) = channel::<()>();
538         let (tx3, rx3) = channel::<()>();
539         let _t = Thread::spawn(move|| {
540             let s = Select::new();
541             let mut h1 = s.handle(&rx1);
542             let mut h2 = s.handle(&rx2);
543             unsafe { h2.add(); }
544             unsafe { h1.add(); }
545             assert_eq!(s.wait(), h2.id);
546             tx3.send(()).unwrap();
547         });
548
549         for _ in range(0u, 1000) { Thread::yield_now(); }
550         drop(tx1.clone());
551         tx2.send(()).unwrap();
552         rx3.recv().unwrap();
553     }
554
555     #[test]
556     fn preflight1() {
557         let (tx, rx) = channel();
558         tx.send(()).unwrap();
559         select! {
560             _n = rx.recv() => {}
561         }
562     }
563
564     #[test]
565     fn preflight2() {
566         let (tx, rx) = channel();
567         tx.send(()).unwrap();
568         tx.send(()).unwrap();
569         select! {
570             _n = rx.recv() => {}
571         }
572     }
573
574     #[test]
575     fn preflight3() {
576         let (tx, rx) = channel();
577         drop(tx.clone());
578         tx.send(()).unwrap();
579         select! {
580             _n = rx.recv() => {}
581         }
582     }
583
584     #[test]
585     fn preflight4() {
586         let (tx, rx) = channel();
587         tx.send(()).unwrap();
588         let s = Select::new();
589         let mut h = s.handle(&rx);
590         unsafe { h.add(); }
591         assert_eq!(s.wait2(false), h.id);
592     }
593
594     #[test]
595     fn preflight5() {
596         let (tx, rx) = channel();
597         tx.send(()).unwrap();
598         tx.send(()).unwrap();
599         let s = Select::new();
600         let mut h = s.handle(&rx);
601         unsafe { h.add(); }
602         assert_eq!(s.wait2(false), h.id);
603     }
604
605     #[test]
606     fn preflight6() {
607         let (tx, rx) = channel();
608         drop(tx.clone());
609         tx.send(()).unwrap();
610         let s = Select::new();
611         let mut h = s.handle(&rx);
612         unsafe { h.add(); }
613         assert_eq!(s.wait2(false), h.id);
614     }
615
616     #[test]
617     fn preflight7() {
618         let (tx, rx) = channel::<()>();
619         drop(tx);
620         let s = Select::new();
621         let mut h = s.handle(&rx);
622         unsafe { h.add(); }
623         assert_eq!(s.wait2(false), h.id);
624     }
625
626     #[test]
627     fn preflight8() {
628         let (tx, rx) = channel();
629         tx.send(()).unwrap();
630         drop(tx);
631         rx.recv().unwrap();
632         let s = Select::new();
633         let mut h = s.handle(&rx);
634         unsafe { h.add(); }
635         assert_eq!(s.wait2(false), h.id);
636     }
637
638     #[test]
639     fn preflight9() {
640         let (tx, rx) = channel();
641         drop(tx.clone());
642         tx.send(()).unwrap();
643         drop(tx);
644         rx.recv().unwrap();
645         let s = Select::new();
646         let mut h = s.handle(&rx);
647         unsafe { h.add(); }
648         assert_eq!(s.wait2(false), h.id);
649     }
650
651     #[test]
652     fn oneshot_data_waiting() {
653         let (tx1, rx1) = channel();
654         let (tx2, rx2) = channel();
655         let _t = Thread::spawn(move|| {
656             select! {
657                 _n = rx1.recv() => {}
658             }
659             tx2.send(()).unwrap();
660         });
661
662         for _ in range(0u, 100) { Thread::yield_now() }
663         tx1.send(()).unwrap();
664         rx2.recv().unwrap();
665     }
666
667     #[test]
668     fn stream_data_waiting() {
669         let (tx1, rx1) = channel();
670         let (tx2, rx2) = channel();
671         tx1.send(()).unwrap();
672         tx1.send(()).unwrap();
673         rx1.recv().unwrap();
674         rx1.recv().unwrap();
675         let _t = Thread::spawn(move|| {
676             select! {
677                 _n = rx1.recv() => {}
678             }
679             tx2.send(()).unwrap();
680         });
681
682         for _ in range(0u, 100) { Thread::yield_now() }
683         tx1.send(()).unwrap();
684         rx2.recv().unwrap();
685     }
686
687     #[test]
688     fn shared_data_waiting() {
689         let (tx1, rx1) = channel();
690         let (tx2, rx2) = channel();
691         drop(tx1.clone());
692         tx1.send(()).unwrap();
693         rx1.recv().unwrap();
694         let _t = Thread::spawn(move|| {
695             select! {
696                 _n = rx1.recv() => {}
697             }
698             tx2.send(()).unwrap();
699         });
700
701         for _ in range(0u, 100) { Thread::yield_now() }
702         tx1.send(()).unwrap();
703         rx2.recv().unwrap();
704     }
705
706     #[test]
707     fn sync1() {
708         let (tx, rx) = sync_channel::<int>(1);
709         tx.send(1).unwrap();
710         select! {
711             n = rx.recv() => { assert_eq!(n.unwrap(), 1); }
712         }
713     }
714
715     #[test]
716     fn sync2() {
717         let (tx, rx) = sync_channel::<int>(0);
718         let _t = Thread::spawn(move|| {
719             for _ in range(0u, 100) { Thread::yield_now() }
720             tx.send(1).unwrap();
721         });
722         select! {
723             n = rx.recv() => { assert_eq!(n.unwrap(), 1); }
724         }
725     }
726
727     #[test]
728     fn sync3() {
729         let (tx1, rx1) = sync_channel::<int>(0);
730         let (tx2, rx2): (Sender<int>, Receiver<int>) = channel();
731         let _t = Thread::spawn(move|| { tx1.send(1).unwrap(); });
732         let _t = Thread::spawn(move|| { tx2.send(2).unwrap(); });
733         select! {
734             n = rx1.recv() => {
735                 let n = n.unwrap();
736                 assert_eq!(n, 1);
737                 assert_eq!(rx2.recv().unwrap(), 2);
738             },
739             n = rx2.recv() => {
740                 let n = n.unwrap();
741                 assert_eq!(n, 2);
742                 assert_eq!(rx1.recv().unwrap(), 1);
743             }
744         }
745     }
746 }