1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456 457 458 459 460 461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476 477 478 479 480 481 482 483 484 485 486 487 488 489 490 491 492 493 494 495 496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637 638 639 640 641 642 643 644 645 646 647 648 649 650 651 652 653 654 655 656 657 658 659 660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677 678 679 680 681 682 683 684 685 686 687 688 689 690 691 692 693 694 695 696 697 698 699 700 701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718 719 720 721 722 723 724 725 726 727 728 729 730 731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746 747 748 749 750 751 752 753 754 755 756 757 758 759 760 761 762 763 764 765 766 767 768 769 770 771 772 773 774 775 776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801 802 803 804 805 806 807 808 809 810 811 812 813 814 815 816 817 818 819 820 821 822 823 824 825 826 827 828 829 830 831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850 851 852 853 854 855 856 857 858 859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892 893 894 895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924 925 926 927 928 929 930 931 932 933 934 935 936 937 938 939 940 941 942 943 944 945 946 947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974 975 976 977 978 979 980 981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031 1032 1033 1034 1035 1036 1037 1038 1039 1040 1041 1042 1043 1044 1045 1046 1047 1048 1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060 1061 1062 1063 1064 1065 1066 1067 1068 1069 1070 1071 1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083 1084 1085 1086 1087 1088 1089 1090 1091 1092 1093 1094 1095 1096 1097 1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113 1114 1115 1116 1117 1118 1119 1120 1121 1122 1123 1124 1125 1126 1127 1128 1129 1130 1131 1132 1133 1134 1135 1136 1137 1138 1139 1140 1141 1142 1143 1144 1145 1146 1147 1148 1149 1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169 1170 1171 1172 1173 1174 1175 1176 1177 1178 1179 1180 1181 1182 1183 1184 1185 1186 1187 1188 1189 1190 1191 1192 1193 1194 1195 1196 1197 1198 1199 1200 1201 1202 1203 1204 1205 1206 1207 1208 1209 1210 1211 1212 1213 1214 1215 1216 1217 1218 1219 1220 1221 1222 1223 1224 1225 1226 1227 1228 1229 1230 1231 1232 1233 1234 1235 1236 1237 1238 1239 1240 1241 1242 1243 1244 1245 1246 1247 1248 1249 1250 1251 1252 1253 1254 1255 1256 1257 1258 1259 1260 1261 1262 1263 1264 1265 1266 1267 1268 1269
//! Provides a message dispatch service where each receiver is aware of messages passed to other //! peers. In particular, if a message is sent to some receiver `r`, another receiver `r'` will be //! aware that one message has been dispatched when it does a subsequent read. Furthermore, the //! dispatcher ensures that messages are delivered in order by not emitting data until all input //! sources have confirmed that they will not send data with lower sequence numbers. //! //! The library ensures that a sender will not block due to the slowness of a receiver that is not //! the intended recipient of the message in question. For example, if there are two receivers, `r` //! and `r'`, `r.send(v)` will not block even though `r'` is not currently reading from its input //! channel. //! //! The library is implemented by routing all messages through a single dispatcher. //! This central dispatcher operates in one of two modes, *forwarding* or *serializing*. //! //! - In serializing mode, it assigns a monotonically increasing timestamp to each message, and //! forwards it to the intended recipient's queue. //! - In forwarding mode, it accepts timestamped messages from sources, and outputs them to the //! intended recipients *in order*. Messages are buffered by the dispatcher until each of the //! receiver's sources are at least as up-to-date as the message's timestamp. These timestamps //! *must* be sequentially assigned, but *may* be sent to the dispatcher in any order. The //! dispatcher guarantees that they are delivered in-order. //! //! This dual-mode operation allows dispatchers to be composed in a hierarchical fashion, with a //! serializing dispatcher at the "top", and forwarding dispatchers "below". //! //! # Examples: //! //! Simple usage: //! //! ``` //! use std::thread; //! use clocked_dispatch; //! //! // Create a dispatcher //! let d = clocked_dispatch::new(1); //! //! // Create a simple streaming channel //! let (tx, rx) = d.new("atx1", "arx"); //! thread::spawn(move|| { //! tx.send(10); //! }); //! assert_eq!(rx.recv().unwrap().0.unwrap(), 10); //! ``` //! //! Shared usage: //! //! ``` //! use std::thread; //! use clocked_dispatch; //! //! // Create a dispatcher. //! // Notice that we need more buffer space to the dispatcher here. //! // This is because clone() needs to talk to the dispatcher, but the buffer to the dispatcher //! // may already have been filled up by the sends in the threads we spawned. //! let d = clocked_dispatch::new(10); //! //! // Create a shared channel that can be sent along from many threads //! // where tx is the sending half (tx for transmission), and rx is the receiving //! // half (rx for receiving). //! let (tx, rx) = d.new("atx", "arx"); //! for i in 0..10 { //! let tx = tx.clone(format!("atx{}", i)); //! thread::spawn(move|| { //! tx.send(i); //! }); //! } //! //! for _ in 0..10 { //! let j = rx.recv().unwrap().0.unwrap(); //! assert!(0 <= j && j < 10); //! } //! ``` //! //! Accessing timestamps: //! //! ``` //! use clocked_dispatch; //! let m = clocked_dispatch::new(10); //! let (tx_a, rx_a) = m.new("atx1", "a"); //! //! // notice that we can't use _ here even though tx_b is unused because //! // then tx_b would be dropped, causing rx_b to be closed immediately //! let (tx_b, rx_b) = m.new("btx1", "b"); //! let _ = tx_b; //! //! tx_a.send("a1"); //! let x = rx_a.recv().unwrap(); //! assert_eq!(x.0, Some("a1")); //! assert_eq!(rx_b.recv(), Ok((None, x.1))); //! //! tx_a.send("a2"); //! tx_a.send("a3"); //! //! let a1 = rx_a.recv().unwrap(); //! assert_eq!(a1.0, Some("a2")); //! //! let a2 = rx_a.recv().unwrap(); //! assert_eq!(a2.0, Some("a3")); //! //! // b must see the timestamp from either a1 or a2 //! // it could see a1 if a2 hasn't yet been delivered //! let b = rx_b.recv().unwrap(); //! assert_eq!(b.0, None); //! assert!(b.1 == a1.1 || b.1 == a2.1); //! ``` //! //! In-order delivery //! //! ``` //! use clocked_dispatch; //! use std::sync::mpsc; //! //! let m = clocked_dispatch::new(10); //! let (tx1, rx) = m.new("tx1", "a"); //! let tx2 = tx1.clone("tx2"); //! //! tx1.forward(Some("a1"), 1); //! assert_eq!(rx.try_recv(), Err(mpsc::TryRecvError::Empty)); //! //! tx2.forward(None, 1); //! assert_eq!(rx.recv(), Ok((Some("a1"), 1))); //! ``` extern crate rand; use std::sync::{Arc, Mutex, Condvar}; use std::cmp::Ordering; use std::collections::HashMap; use std::collections::HashSet; use std::collections::VecDeque; use std::collections::BinaryHeap; use std::sync::mpsc; use std::thread; use std::sync; macro_rules! debug { ( $fmt:expr ) => { // println!($fmt); }; ( $fmt:expr, $( $args:expr ),+ ) => { // println!($fmt, $($args),*); $(let _ = $args;)*; }; } struct TaggedData<T> { from: String, to: Option<String>, ts: Option<usize>, data: Option<T>, } /// A message intended for the dispatcher. enum Message<T> { Data(TaggedData<T>), ReceiverJoin(String, Arc<ReceiverInner<T>>), ReceiverLeave(String), SenderJoin(Option<String>, String), SenderLeave(Option<String>, String), } /// The sending half of a clocked synchronous channel. /// This half can only be owned by one thread, but it can be cloned to send to other threads. /// /// Sending on a clocked channel will deliver the given message to the appropriate receiver, but /// also notify all other receivers about the timestamp assigned to the message. The sending will /// never block on a receiver that is not the destination of the message. /// /// Beware that dropping a clocked sender incurs control messages to the dispatcher, and that those /// control messages may result in messages being sent to receivers. If the dispatch channel is not /// sufficiently buffered, this means that dropping a `ClockedSender` before the corresponding /// `ClockedReceiver` is receiving on its end of the channel may deadlock. /// /// When the last `ClockedSender` is dropped for a target, and there are no `ClockedBroadcaster`s, /// the dispatcher will automatically be notified, and the recipient will see a disconnected /// channel error once it has read all buffered messages. /// /// ``` /// use clocked_dispatch; /// use std::thread; /// /// let m = clocked_dispatch::new(10); /// let (tx_a, rx_a) = m.new("atx", "arx"); /// /// let tx_a1 = tx_a.clone("atx1"); /// thread::spawn(move || { /// tx_a1.send("a1"); /// }); /// /// let tx_a2 = tx_a.clone("atx2"); /// thread::spawn(move || { /// tx_a2.send("a2"); /// }); /// /// drop(tx_a); /// assert_eq!(rx_a.count(), 2); /// ``` pub struct ClockedSender<T> { target: String, source: String, dispatcher: mpsc::SyncSender<Message<T>>, } impl<T> Drop for ClockedSender<T> { fn drop(&mut self) { self.dispatcher .send(Message::SenderLeave(Some(self.target.clone()), self.source.clone())) .unwrap(); } } impl<T> ClockedSender<T> { /// Sends a value on this synchronous channel, and notifies all other recipients of the /// timestamp it is assigned by the dispatcher. /// /// This function will *block* until space in the internal buffer becomes available, or a /// receiver is available to hand off the message to. /// /// Note that a successful send does *not* guarantee that the receiver will ever see the data if /// there is a buffer on this channel. Items may be enqueued in the internal buffer for the /// receiver to receive at a later time. If the buffer size is 0, however, it can be guaranteed /// that the receiver has indeed received the data if this function returns success. pub fn send(&self, data: T) { // XXX: would be really neat if we could return the ts here, but that'll probably be tricky // TODO: This function will never panic, but it may return `Err` if the `Receiver` has // disconnected and is no longer able to receive information. self.dispatcher .send(Message::Data(TaggedData { from: self.source.clone(), to: Some(self.target.clone()), ts: None, data: Some(data), })) .unwrap() } /// Sends an already-sequenced value to the associated receiver. The message may be buffered /// by the dispatcher until it can guarantee that no other sender will later try to send /// messages with a lower sequence number. /// /// It is optional to include data when forwarding. If no data is included, this message /// conveys to the dispatcher that this sender promises not to send later messages with a /// higher sequence number than the one given. pub fn forward(&self, data: Option<T>, ts: usize) { self.dispatcher .send(Message::Data(TaggedData { from: self.source.clone(), to: Some(self.target.clone()), ts: Some(ts), data: data, })) .unwrap() } /// Creates a new clocked sender for this sender's receiver. /// /// Clocked dispatch requires that all senders have a unique name so that the "up-to-date-ness" /// of the senders can be tracked reliably. pub fn clone<V: Into<String>>(&self, source: V) -> ClockedSender<T> { let source = source.into(); self.dispatcher .send(Message::SenderJoin(Some(self.target.clone()), source.clone())) .unwrap(); ClockedSender { source: source, target: self.target.clone(), dispatcher: self.dispatcher.clone(), } } } impl<T: Clone> ClockedSender<T> { /// Converts this sender into a broadcast sender. /// /// Doing so detaches the sender from its receiver, and means all future sends will be /// broadcast to all receivers. Note that the existence of a broadcaster prevents the closing /// of all channels. pub fn into_broadcaster(self) -> ClockedBroadcaster<T> { let dispatcher = self.dispatcher.clone(); let source = format!("{}_bcast", self.source); dispatcher.send(Message::SenderJoin(None, source.clone())).unwrap(); // NOTE: the drop of self causes a Message::SenderLeave to be sent for this sender ClockedBroadcaster { source: source, dispatcher: dispatcher, } } } /// A sending half of a clocked synchronous channel that only allows broadcast. This half can only /// be owned by one thread, but it can be cloned to send to other threads. A `ClockedBroadcaster` /// can be constructed from a `ClockedSender` using `ClockedSender::into_broadcaster`. /// /// Sending on a clocked channel will deliver the given message to the appropriate receiver, but /// also notify all other receivers about the timestamp assigned to the message. The sending will /// never block on a receiver that is not the destination of the message. /// /// Beware that dropping a clocked sender incurs control messages to the dispatcher, and that those /// control messages may result in messages being sent to receivers. If the dispatch channel is not /// sufficiently buffered, this means that dropping a `ClockedSender` before the corresponding /// `ClockedReceiver` is receiving on its end of the channel may deadlock. /// /// Note that the existence of a `ClockedBroadcater` prevents the closing of any clocked channels /// managed by this dispatcher. /// /// # Examples /// /// Regular broadcast: /// /// ``` /// use clocked_dispatch; /// use std::sync::mpsc; /// /// let m = clocked_dispatch::new(10); /// let (tx_a, rx_a) = m.new("atx", "arx"); /// let tx = tx_a.into_broadcaster(); /// // note that the A channel is still open since there now exists a broadcaster, /// // even though all A senders have been dropped. /// /// let (tx_b, rx_b) = m.new("btx", "brx"); /// /// tx.broadcast("1"); /// /// let x = rx_a.recv().unwrap(); /// assert_eq!(x.0, Some("1")); /// assert_eq!(rx_b.recv(), Ok(x)); /// /// // non-broadcasts still work /// tx_b.send("2"); /// let x = rx_b.recv().unwrap(); /// assert_eq!(x.0, Some("2")); /// assert_eq!(rx_a.recv(), Ok((None, x.1))); /// /// // drop broadcaster /// drop(tx); /// /// // A is now closed because there are no more senders /// assert_eq!(rx_a.recv(), Err(mpsc::RecvError)); /// /// // rx_b is *not* closed because tx_b still exists /// assert_eq!(rx_b.try_recv(), Err(mpsc::TryRecvError::Empty)); /// /// drop(tx_b); /// // rx_b is now closed because its senders have all gone away /// assert_eq!(rx_b.recv(), Err(mpsc::RecvError)); /// ``` /// /// Forwarding broadcast: /// /// ``` /// use clocked_dispatch; /// use std::sync::mpsc; /// /// let m = clocked_dispatch::new(10); /// let (tx_a, rx_a) = m.new("atx", "arx"); /// let (tx_b, rx_b) = m.new("btx", "brx"); /// let (tx_c, rx_c) = m.new("ctx", "crx"); /// /// let tx = tx_a.into_broadcaster(); /// tx.broadcast_forward(Some("1"), 1); /// /// assert_eq!(rx_a.recv().unwrap(), (Some("1"), 1)); /// assert_eq!(rx_b.recv().unwrap(), (Some("1"), 1)); /// assert_eq!(rx_c.recv().unwrap(), (Some("1"), 1)); /// /// // non-broadcasts still work /// tx_c.forward(Some("c"), 2); /// assert_eq!(rx_a.recv().unwrap(), (None, 2)); /// assert_eq!(rx_b.recv().unwrap(), (None, 2)); /// assert_eq!(rx_c.recv().unwrap(), (Some("c"), 2)); /// ``` pub struct ClockedBroadcaster<T: Clone> { source: String, dispatcher: mpsc::SyncSender<Message<T>>, } impl<T: Clone> Drop for ClockedBroadcaster<T> { fn drop(&mut self) { self.dispatcher.send(Message::SenderLeave(None, self.source.clone())).unwrap(); } } impl<T: Clone> ClockedBroadcaster<T> { /// Sends a value to all receivers known to this dispatcher. The value will be assigned a /// sequence number by the dispatcher. /// /// This function will *block* until space in the internal buffer becomes available, or a /// receiver is available to hand off the message to. /// /// Note that a successful send does *not* guarantee that the receiver will ever see the data if /// there is a buffer on this channel. Items may be enqueued in the internal buffer for the /// receiver to receive at a later time. If the buffer size is 0, however, it can be guaranteed /// that the receiver has indeed received the data if this function returns success. pub fn broadcast(&self, data: T) { self.dispatcher .send(Message::Data(TaggedData { from: self.source.clone(), to: None, ts: None, data: Some(data), })) .unwrap() } /// Sends an already-sequenced value to all receivers known to this dispatcher. The message may /// be buffered by the dispatcher until it can guarantee that no other sender will later try to /// send messages with a lower sequence number. /// /// This function will *block* until space in the internal buffer becomes available, or a /// receiver is available to hand off the message to. /// /// Note that a successful send does *not* guarantee that the receiver will ever see the data if /// there is a buffer on this channel. Items may be enqueued in the internal buffer for the /// receiver to receive at a later time. If the buffer size is 0, however, it can be guaranteed /// that the receiver has indeed received the data if this function returns success. /// /// It is optional to include data when forwarding. If no data is included, this message /// conveys to the dispatcher that this sender promises not to send later messages with a /// higher sequence number than the one given. pub fn broadcast_forward(&self, data: Option<T>, ts: usize) { self.dispatcher .send(Message::Data(TaggedData { from: self.source.clone(), to: None, ts: Some(ts), data: data, })) .unwrap() } /// Creates a new clocked broadcast sender. /// /// Clocked dispatch requires that all senders have a unique name so that the "up-to-date-ness" /// of the senders can be tracked reliably. pub fn clone<V: Into<String>>(&self, source: V) -> ClockedBroadcaster<T> { let source = source.into(); self.dispatcher.send(Message::SenderJoin(None, source.clone())).unwrap(); ClockedBroadcaster { source: source, dispatcher: self.dispatcher.clone(), } } } struct QueueState<T> { queue: VecDeque<(T, usize)>, ts_head: usize, ts_tail: usize, closed: bool, left: bool, } struct ReceiverInner<T> { mx: Mutex<QueueState<T>>, cond: Condvar, } /// The receiving half of a clocked synchronous channel. /// /// A clocked receiver will receive all messages sent by one of its associated senders. It will /// also receive notifications whenever a message with a higher timestamp than any it has seen has /// been sent to another receiver under the same dispatcher. /// /// Dropping it will unblock any senders trying to send to this receiver. pub struct ClockedReceiver<T: Send + 'static> { leave: mpsc::SyncSender<String>, inner: Arc<ReceiverInner<T>>, name: String, } impl<T: Send + 'static> ClockedReceiver<T> { fn new<V: Into<String>>(name: V, leave: mpsc::SyncSender<String>, bound: usize) -> ClockedReceiver<T> { ClockedReceiver { leave: leave, inner: Arc::new(ReceiverInner { mx: Mutex::new(QueueState { queue: VecDeque::with_capacity(bound), ts_head: 0, ts_tail: 0, closed: false, left: false, }), cond: Condvar::new(), }), name: name.into(), } } } impl<T: Send + 'static> Iterator for ClockedReceiver<T> { type Item = (Option<T>, usize); fn next(&mut self) -> Option<Self::Item> { self.recv().ok() } } impl<T: Send + 'static> Drop for ClockedReceiver<T> { fn drop(&mut self) { use std::mem; let name = mem::replace(&mut self.name, String::new()); self.leave.send(name).unwrap(); // wait until we've actually been dropped self.count(); } } impl<T: Send + 'static> ClockedReceiver<T> { /// Attempts to wait for a value on this receiver, returning an error if the corresponding /// channel has hung up. /// /// This function will always block the current thread if there is no data available, the /// receiver has seen the latest timestamp handled by the dispatcher, and it's possible for /// more data to be sent. Once a message is sent to a corresponding `ClockedSender`, then this /// receiver will wake up and return that message. If a message is sent by a `ClockedSender` /// connected to a different receiver under the same dispatcher, this receiver will wake up and /// receive the timestamp assigned to that message. /// /// If all corresponding `ClockedSender` have disconnected, or disconnect while this call is /// blocking, this call will wake up and return `Err` to indicate that no more messages can /// ever be received on this channel. However, since channels are buffered, messages sent /// before the disconnect will still be properly received. pub fn recv(&self) -> Result<(Option<T>, usize), mpsc::RecvError> { let mut state = self.inner.mx.lock().unwrap(); while state.ts_head == state.ts_tail && state.queue.is_empty() && !state.closed { // NOTE: is is *not* sufficient to use head == tail as an indicator that there are no // messages. specifically, if there are duplicates for a given timestamp, the equality // may work out while there are still elements in the queue. state = self.inner.cond.wait(state).unwrap(); } // if there's something at the head of the queue, return it if let Some((t, ts)) = state.queue.pop_front() { state.ts_head = ts; self.inner.cond.notify_one(); return Ok((Some(t), ts)); } if state.ts_head == state.ts_tail { // we must be closed assert_eq!(state.closed, true); return Err(mpsc::RecvError); } // otherwise, notify about the newest available timestamp state.ts_head = state.ts_tail; self.inner.cond.notify_one(); Ok((None, state.ts_head)) } /// Attempts to return a pending value on this receiver without blocking /// /// This method will never block the caller in order to wait for data to become available. /// Instead, this will always return immediately with a possible option of pending data on the /// channel. /// /// This is useful for a flavor of "optimistic check" before deciding to block on a receiver. pub fn try_recv(&self) -> Result<(Option<T>, usize), mpsc::TryRecvError> { let mut state = self.inner.mx.lock().unwrap(); if state.ts_head == state.ts_tail && !state.closed { // we have observed all timestamps, so the queue must be empty return Err(mpsc::TryRecvError::Empty); } if state.ts_head == state.ts_tail { // we must be closed assert_eq!(state.closed, true); return Err(mpsc::TryRecvError::Disconnected); } // if there's something at the head of the queue, return it if let Some((t, ts)) = state.queue.pop_front() { state.ts_head = ts; self.inner.cond.notify_one(); return Ok((Some(t), ts)); } // otherwise, notify about the newest available timestamp state.ts_head = state.ts_tail; self.inner.cond.notify_one(); Ok((None, state.ts_head)) } } /// Dispatch coordinator for adding additional clocked channels. pub struct Dispatcher<T: Send> { dispatcher: mpsc::SyncSender<Message<T>>, leave: mpsc::SyncSender<String>, bound: usize, } impl<T: Send> Dispatcher<T> { /// Creates a new named, synchronous, bounded, clocked channel managed by this dispatcher. /// /// The given receiver and sender names *must* be unique for this dispatch. /// /// The `ClockedReceiver` will block until a message or a new timestamp becomes available. /// /// The receiver's incoming channel has an internal buffer on which messages will be queued. /// Its size is inherited from the dispatch bound. When this buffer becomes full, future /// messages from the dispatcher will block waiting for the buffer to open up. Note that a /// buffer size of 0 is valid, but its behavior differs from that of synchronous Rust channels. /// Because the dispatcher sits between the sender and the receiver, a bound of 0 will not /// guarantee a "rendezvous" between the sender and the receiver, but rather between the sender /// and the dispatcher (and subsequently, the dispatcher and the receiver). pub fn new<S1: Into<String>, S2: Into<String>>(&self, sender: S1, receiver: S2) -> (ClockedSender<T>, ClockedReceiver<T>) { let source = sender.into(); let target = receiver.into(); let send = ClockedSender { source: source.clone(), target: target.clone(), dispatcher: self.dispatcher.clone(), }; let recv = ClockedReceiver::new(target.clone(), self.leave.clone(), self.bound); self.dispatcher.send(Message::ReceiverJoin(target.clone(), recv.inner.clone())).unwrap(); self.dispatcher.send(Message::SenderJoin(Some(target.clone()), source)).unwrap(); (send, recv) } } /// `Delayed` is used to keep track of messages that cannot yet be safely delivered because it /// would violate the in-order guarantees. /// /// `Delayed` structs are ordered by their timestamp such that the *lowest* is the "highest". This /// is so that `Delayed` can easily be used in a `BinaryHeap`. struct Delayed<T> { ts: usize, data: T, } impl<T> PartialEq for Delayed<T> { fn eq(&self, other: &Delayed<T>) -> bool { other.ts == self.ts } } impl<T> PartialOrd for Delayed<T> { fn partial_cmp(&self, other: &Delayed<T>) -> Option<Ordering> { Some(self.cmp(other)) } } impl<T> Eq for Delayed<T> {} impl<T> Ord for Delayed<T> { fn cmp(&self, other: &Delayed<T>) -> Ordering { other.ts.cmp(&self.ts) } } struct Target<T> { // the receiver's channel channel: Arc<ReceiverInner<T>>, // messages for this receiver that have high timestamps, and must be delayed // the mutex is so that we can allow the control thread to have an &Target without T: Sync delayed: sync::Mutex<BinaryHeap<Delayed<T>>>, // known senders for this receiver senders: HashSet<String>, } // TODO: // It seems like dispatchers are always used in one of the following ways: // // - Multi-in, multi-out, unicast, assigning timestamps, no buffering // - Multi-in, single-out, unicast, forwarding, buffering // - Single-in, multi-out, broadcast, forwarding, no buffering // // We could specialize for each of these, which might increase performance and further modularize // the code. This would also allow restricting the API such that you can't start using one kind of // dispatcher in another mode. Another potentially good example of this is forcing T: Clone only // for broadcast dispatchers. struct DispatchInner<T> { // per-receiver information // needs to be locked so that control thread can access ReceiverInner Arcs targets: sync::Arc<sync::RwLock<HashMap<String, Target<T>>>>, // essentially targets.keys() // kept separate so we can mutably use targets while iterating over all receiver names destinations: HashSet<String>, // known broadcasters broadcasters: HashSet<String>, // broadcast messages that have high timestamps, and must be delayed bdelay: BinaryHeap<Delayed<T>>, // whether we are operating in forwarding or serializing mode // in the former, the senders assign timestamps // in the latter, we assign the timestamps // the first message we receive dictate the mode forwarding: Option<bool>, // queue bound bound: usize, id: String, // Sequence number counter. // If we are in forwarding mode this is the highest consecutive sequence number we have // received. If we are in serializing mode, this is the last sequence number we have assigned. counter: usize, } impl<T: Clone> DispatchInner<T> { /// Notifies all receivers of the given timestamp, and sends any given data to the intended /// recipients. /// /// If `data == None`, `ts` is sent all receivers. /// If `to == None`, `data.unwrap()` is sent to all receivers. /// If `to == Some(t)`, `data.unwrap()` is sent to the the receiver named `t`. fn notify(&self, to: Option<&String>, ts: usize, data: Option<T>) { let tgts = self.targets.read().unwrap(); for (tn, t) in tgts.iter() { let mut state = t.channel.mx.lock().unwrap(); debug!("{}: notifying {} about {}", self.id, tn, ts); if data.is_some() && (to.is_none() || to.unwrap() == tn.as_str()) { debug!("{}: including data", self.id); while state.queue.len() == self.bound && !state.left { state = t.channel.cond.wait(state).unwrap(); } if state.left { t.channel.cond.notify_one(); continue; } // TODO: avoid clone() for the last send state.queue.push_back((data.clone().unwrap(), ts)); } state.ts_tail = ts; t.channel.cond.notify_one(); drop(state); } // if data.is_some() && to.is_some() && !self.targets.contains_key(to.unwrap().as_str()) // this seems like a bad case, but it could just be that the receiver has left // TODO: would be nice if we had some way of notifying the sender that this is the case } /// Find any delayed messages that are now earlier than the minimum sender sequence number, and /// send them in-order. Will check both broadcast messages and messages to a given sender if /// `to.is_some()`. fn process_delayed(&mut self) { assert!(self.forwarding.unwrap_or(false)); debug!("{}: processing delayed after {}", self.id, self.counter); // keep looking for a candidate to send loop { // we need to find the message in `[bdelay + targets[to].delayed]` with the lowest // timestamp. we do this by: // // 1. finding the smallest in `bdelay` let next = self.bdelay.peek().map(|d| d.ts); debug!("{}: next from bcast is {:?}", self.id, next); // 2. finding the smallest in `targets[*].delayed` let tnext = { let tgts = self.targets.read().unwrap(); let t = self.destinations .iter() .map(|to| { let t = &tgts[to]; (to, t.delayed .lock() .unwrap() .peek() .map(|d| d.ts)) }) .filter_map(|(to, ts)| ts.map(move |ts| (to, ts))) .min_by_key(|&(_, ts)| ts); t.map(|(to, ts)| (to.to_owned(), ts)) }; debug!("{}: next from tdelay is {:?}", self.id, tnext); // 3. using the message from 2 if it is the next message if let Some((to, tnext)) = tnext { if tnext == self.counter + 1 { debug!("{}: forwarding from tdelay", self.id); let d = { let tgts = self.targets.read().unwrap(); let mut x = tgts[to.as_str()].delayed.lock().unwrap(); x.pop().unwrap() }; self.notify(Some(&to), d.ts, Some(d.data)); self.counter += 1; continue; } } // 4. using the message from 1 if it is the next message if let Some(ts) = next { if ts == self.counter + 1 { debug!("{}: forwarding from bdelay", self.id); let d = self.bdelay.pop().unwrap(); self.notify(None, d.ts, Some(d.data)); self.counter += 1; continue; } } // no delayed message has a sequence number <= min break; } debug!("{}: done replaying", self.id); } /// Takes a message from any sender, handles control messages, and delays or delivers data /// messages. /// /// The first data message sets which mode the dispatcher operates in. If the first message has /// a sequence number, the dispatcher will operate in forwarding mode. If it does not, it will /// operate in assignment mode. In the former, it expects every data message to be numbered, /// and delays too-new messages until all inputs are at least that up-to-date. In the latter, /// it will deliver all messages immediately, and will assign sequence numbers to each one. fn absorb(&mut self, m: Message<T>) { match m { Message::Data(td) => { debug!("{}: got message with ts {:?} from {} for {:?}", self.id, td.ts, td.from, td.to); if self.forwarding.is_some() { assert!(self.forwarding.unwrap() == td.ts.is_some(), "one sender sent timestamp, another did not"); } else { self.forwarding = Some(td.ts.is_some()) } if let Some(ts) = td.ts { // if we are forwarding (which must be the case here), this message may be the // next to be sent out. in that case we should increase the sequence number // tracker so that any later messages will also be released assert!(ts >= self.counter); if ts == self.counter + 1 { self.counter = ts; } } if td.ts.is_none() { // the sender leaves it up to us to pick timestamps, so we know we're always up // to date. note that this latter case assumes that the senders will *never* // give us timestamps once they have let us pick once. self.counter += 1; self.notify(td.to.as_ref(), self.counter, td.data); return; } // we're in forwarding mode let ts = td.ts.unwrap(); if ts == self.counter { // this messages is the next to be sent out, so we can send it immediately self.notify(td.to.as_ref(), ts, td.data); // since this messages must also have incremented the counter above, there may // be other messages that can now be sent out. self.process_delayed(); return; } // need to buffer this message until the other views are sufficiently up-to-date. if let Some(data) = td.data { if let Some(ref to) = td.to { debug!("{}: delayed in {:?}", self.id, to); let tgts = self.targets.read().unwrap(); tgts[to].delayed.lock().unwrap().push(Delayed { ts: ts, data: data, }); drop(tgts); } else { debug!("{}: delayed in bcast", self.id); self.bdelay.push(Delayed { ts: ts, data: data, }); } } } Message::ReceiverJoin(name, inner) => { debug!("{}: receiver {} joined", self.id, name); if !self.destinations.insert(name.clone()) { panic!("receiver {} already exists!", name); } let mut tgts = self.targets.write().unwrap(); tgts.insert(name, Target { channel: inner, senders: HashSet::new(), delayed: sync::Mutex::new(BinaryHeap::new()), }); } Message::ReceiverLeave(name) => { debug!("{}: receiver {} left", self.id, name); // NOTE: Control thread has already unblocked senders and set .left // Deregister the receiver let mut tgts = self.targets.write().unwrap(); tgts.remove(&*name); self.destinations.remove(&*name); // TODO: ensure that subsequent send()'s return an error (somehow?) instead of just // crashing and burning (panic) like what happens now. } Message::SenderJoin(target, source) => { debug!("{}: sender {} for {:?} joined", self.id, source, target); if let Some(target) = target { let mut tgts = self.targets.write().unwrap(); tgts.get_mut(&*target).unwrap().senders.insert(source); } else { self.broadcasters.insert(source); } } Message::SenderLeave(target, source) => { debug!("{}: sender {} for {:?} left", self.id, source, target); if let Some(ref target) = target { // NOTE: target may not exist because receiver has left let mut tgts = self.targets.write().unwrap(); if let Some(target) = tgts.get_mut(target.as_str()) { target.senders.remove(&*source); } drop(tgts); } else { self.broadcasters.remove(&*source); } if self.broadcasters.is_empty() { // if there are broadcasters, no channel is closed let mut tgts = self.targets.write().unwrap(); for (tn, t) in tgts.iter_mut() .filter(|&(_, ref t)| { t.senders.is_empty() && t.delayed.lock().unwrap().is_empty() }) { debug!("{}: closing now-done channel {}", self.id, tn); // having no senders when there are no broadcasters means the channel is closed let mut state = t.channel.mx.lock().unwrap(); state.closed = true; t.channel.cond.notify_one(); drop(state); } } } } } } /// Creates a new clocked dispatch. Dispatch channels can be constructed by calling `new` on the /// returned dispatcher. /// /// The dispatcher has an internal buffer for incoming messages. When this buffer becomes full, /// future sends to the dispatcher will block waiting for the buffer to open up. Note that a buffer /// size of 0 is valid, but its behavior differs from that of synchronous Rust channels. Because /// the dispatcher sits between the sender and the receiver, a bound of 0 will not guarantee a /// "rendezvous" between the sender and the receiver, but rather between the sender and the /// dispatcher (and subsequently, the dispatcher and the receiver). /// /// Be aware that a bound of 0 means that it is not safe to drop a `ClockedSender` before the /// corresponding `ClockedReceiver` is reading from its end of the channel. pub fn new<T: Clone + Send + 'static>(bound: usize) -> Dispatcher<T> { new_with_seed(bound, 0) } /// Creates a new clocked dispatch whose automatically assigned sequence numbers start at a given /// value. /// /// This method is useful for programs that wish to maintain monotonic sequence numbers between /// multiple executions of the application. Such an application should track received sequence /// numbers, store the latest one upon exiting, and then use this method to resume the sequence /// numbers from that point onward upon resuming. pub fn new_with_seed<T: Clone + Send + 'static>(bound: usize, seed: usize) -> Dispatcher<T> { use rand::{thread_rng, Rng}; let (stx, srx) = mpsc::sync_channel(bound); let mut d = DispatchInner { targets: sync::Arc::new(sync::RwLock::new(HashMap::new())), destinations: HashSet::new(), bdelay: BinaryHeap::new(), broadcasters: HashSet::new(), forwarding: None, bound: bound, id: thread_rng().gen_ascii_chars().take(2).collect(), counter: seed, }; let id = d.id.clone(); let c_targets = d.targets.clone(); let c_stx = stx.clone(); let (ctx, crx) = mpsc::sync_channel::<String>(0); thread::spawn(move || { // this thread handles leaving receivers. // it is *basically* the following loop: // // ``` // for left in crx { // c_targets[left].channel.left = true; // ctx.send(ReceiverLeave(left)); // } // ``` // // unfortunately, it gets complicated by two factors: // // - if a receiver is created and then dropped immediately, the Leave could reach us // before the Join reaches the dispatcher. in this case, targets[left] doesn't exist // yet. we thus need to wait for the dispatcher to catch up to the join. // // - the dispatcher can't be blocking on a send to the receiver we are dropping because // it doesn't know about it yet, and thus must process the join message before it can // block waiting on us). however, there *is* a possibility of the dispatcher being // blocked on a send to a channel that is queued to be dropped *behind* this one. this // is the source of much of the complexity below. // // essentially, we keep track of leaving receivers that we haven't successfully handled // yet, but also keep reading from crx to see if there are other receivers that are also // trying to leave. // receivers that are trying to leave let mut leaving = Vec::new(); // temp for keeping track of nodes are *still* trying to leave while draining `leaving` let mut leaving_ = Vec::new(); 'recv: loop { // are there more receivers trying to leave? let left = crx.try_recv(); match left { Ok(left) => { // yes -- deal with them too leaving.push(left); } Err(..) if !leaving.is_empty() => { // no, but deal with the receivers that wanted to leave } Err(mpsc::TryRecvError::Disconnected) => { // no, and there will never be more // we must also have dealt with all receivers who tried to leave // it's safe for us to exit break 'recv; } Err(mpsc::TryRecvError::Empty) => { // no, and there also aren't any for us to retry // to avoid busy looping, we can now do a blocking receive let left = crx.recv(); if let Ok(left) = left { // someone tried to leave -- let's try to deal with that leaving.push(left); } else { // channel closed, and no one is waiting -- safe to exit break 'recv; } } } // try to process any receivers trying to leave for left in leaving.drain(..) { debug!("{} control: dealing with departure of receiver {}", id, left); let targets = c_targets.read().unwrap(); if let Some(t) = targets.get(&*left) { // the receiver exists, so we can remove it let mut state = t.channel.mx.lock().unwrap(); state.left = true; state.closed = true; t.channel.cond.notify_one(); drop(state); // kick off a message to the dispatcher in the background. // we don't want it to be in the foreground, because we may have other things // to close that could block sending to the dispatcher. let ctx = c_stx.clone(); thread::spawn(move || { ctx.send(Message::ReceiverLeave(left)).unwrap(); }); } else { // dispatcher doesn't know about this receiver yet leaving_.push(left); } } leaving.extend(leaving_.drain(..)); } }); thread::spawn(move || { for m in srx.iter() { d.absorb(m); } }); Dispatcher { dispatcher: stx, leave: ctx, bound: bound, } } #[cfg(test)] mod tests { #[test] fn can_send_after_recv_drop() { // Create a dispatcher let d = super::new(1); // Create two channels let (tx_a, rx_a) = d.new("atx", "arx"); let (tx_b, rx_b) = d.new("btx", "brx"); let _ = tx_a; // Drop a receiver drop(rx_a); // Ensure that sending doesn't block forever tx_b.send(10); // And that messages are still delivered assert_eq!(rx_b.recv().unwrap().0.unwrap(), 10); } #[test] fn recv_drop_unblocks_sender() { use std::thread; use std::time::Duration; // Create a dispatcher let d = super::new(1); // Create two channels let (tx_a, rx_a) = d.new("atx", "arx"); let (tx_b, rx_b) = d.new("btx", "brx"); // Make tx_a a broadcaster (so it would block on b) // Note that we have to do this *before* we saturate the channel to the dispatcher let tx_a = tx_a.into_broadcaster(); // Fill rx_b thread::spawn(move || { for _ in 0..20 { tx_b.send("b"); } }); thread::sleep(Duration::from_millis(200)); // Drop b's receiver drop(rx_b); // All of tx_b's sends should be dropped, and tx_a should be able to send tx_a.broadcast("a"); // And that messages are still delivered loop { let rx = rx_a.recv(); assert!(rx.is_ok()); let rx = rx.unwrap(); if rx.0.is_some() { assert_eq!(rx.0, Some("a")); break; } } } #[test] fn can_forward_after_recv_drop() { // Create a dispatcher let d = super::new(1); // Create two channels let (tx_a, rx_a) = d.new("atx", "arx"); let (tx_b, rx_b) = d.new("btx", "brx"); let _ = tx_a; // Drop a receiver drop(rx_a); // Ensure that forwarding doesn't block forever tx_b.forward(Some(10), 1); // note that dropping the receiver kills the senders too! // And that messages are still delivered assert_eq!(rx_b.recv(), Ok((Some(10), 1))); } #[test] fn forward_with_no_senders() { use std::sync::mpsc; let d = super::new(1); let (tx_a, rx_a) = d.new("atx", "arx"); let (tx_b, rx_b) = d.new("btx", "brx"); tx_a.forward(Some(1), 1); // the message is queued because tx_b hasn't sent anything // drop both senders, freeing Some(1) from the delayed queue. we specifically want to // explore the case where Some(1) is freed when there are no senders, so we have to drop // tx_a first (since tx_b is holding up the system). drop(tx_a); drop(tx_b); // Ensure that receiver still gets notified of messages assert_eq!(rx_a.recv(), Ok((Some(1), 1))); // And that other still gets a None assert_eq!(rx_b.recv(), Ok((None, 1))); // And that no more entries are sent assert_eq!(rx_a.recv(), Err(mpsc::RecvError)); assert_eq!(rx_b.recv(), Err(mpsc::RecvError)); } #[test] fn broadcast_dupe_termination() { use std::sync::mpsc; let d = super::new(1); let (tx, rx) = d.new("tx", "rx"); let tx = tx.into_broadcaster(); tx.broadcast_forward(Some("a"), 1); tx.broadcast_forward(Some("b"), 2); drop(tx); assert_eq!(rx.recv(), Ok((Some("a"), 1))); assert_eq!(rx.recv(), Ok((Some("b"), 2))); assert_eq!(rx.recv(), Err(mpsc::RecvError)); } #[test] fn multisend_thread_interleaving() { use std::thread; for _ in 0..1000 { let d = super::new(20); let (tx_a, rx) = d.new("tx_a", "rx"); let tx_b = tx_a.clone("tx_b"); let t_a = thread::spawn(move || { tx_a.forward(Some("c_1"), 1); tx_a.forward(Some("c_3"), 3); tx_a.forward(Some("a_1"), 5); }); let t_b = thread::spawn(move || { tx_b.forward(Some("c_2"), 2); tx_b.forward(Some("b_1"), 4); tx_b.forward(Some("a_2"), 6); }); assert_eq!(rx.recv(), Ok((Some("c_1"), 1))); assert_eq!(rx.recv(), Ok((Some("c_2"), 2))); assert_eq!(rx.recv(), Ok((Some("c_3"), 3))); assert_eq!(rx.recv(), Ok((Some("b_1"), 4))); assert_eq!(rx.recv(), Ok((Some("a_1"), 5))); assert_eq!(rx.recv(), Ok((Some("a_2"), 6))); t_a.join().unwrap(); t_b.join().unwrap(); } } #[test] fn test_new_with_seed() { let d = super::new_with_seed(1, 69105); let (tx, rx) = d.new("tx", "rx"); tx.send("a"); assert_eq!(rx.recv(), Ok((Some("a"), 69106))); } }