timely_communication/allocator/
counters.rs

1//! Push and Pull wrappers to maintain counts of messages in channels.
2
3use std::rc::Rc;
4use std::cell::RefCell;
5use std::sync::mpsc::Sender;
6
7use crate::{Push, Pull};
8
9/// The push half of an intra-thread channel.
10pub struct Pusher<T, P: Push<T>> {
11    index: usize,
12    // count: usize,
13    events: Rc<RefCell<Vec<usize>>>,
14    pusher: P,
15    phantom: ::std::marker::PhantomData<T>,
16}
17
18impl<T, P: Push<T>>  Pusher<T, P> {
19    /// Wraps a pusher with a message counter.
20    pub fn new(pusher: P, index: usize, events: Rc<RefCell<Vec<usize>>>) -> Self {
21        Pusher {
22            index,
23            // count: 0,
24            events,
25            pusher,
26            phantom: ::std::marker::PhantomData,
27        }
28    }
29}
30
31impl<T, P: Push<T>> Push<T> for Pusher<T, P> {
32    #[inline]
33    fn push(&mut self, element: &mut Option<T>) {
34        // if element.is_none() {
35        //     if self.count != 0 {
36        //         self.events
37        //             .borrow_mut()
38        //             .push_back(self.index);
39        //         self.count = 0;
40        //     }
41        // }
42        // else {
43        //     self.count += 1;
44        // }
45        // TODO: Version above is less chatty, but can be a bit late in
46        //       moving information along. Better, but needs cooperation.
47        // A `None` is a flush, and the wrapped pusher is unbuffered: nothing
48        // is enqueued, and so there is nothing to announce.
49        if element.is_some() {
50            self.events
51                .borrow_mut()
52                .push(self.index);
53        }
54
55        self.pusher.push(element)
56    }
57}
58
59/// The push half of an intra-thread channel.
60pub struct ArcPusher<T, P: Push<T>> {
61    index: usize,
62    // count: usize,
63    events: Sender<usize>,
64    pusher: P,
65    phantom: ::std::marker::PhantomData<T>,
66    buzzer: crate::buzzer::Buzzer,
67}
68
69impl<T, P: Push<T>>  ArcPusher<T, P> {
70    /// Wraps a pusher with a message counter.
71    pub fn new(pusher: P, index: usize, events: Sender<usize>, buzzer: crate::buzzer::Buzzer) -> Self {
72        ArcPusher {
73            index,
74            // count: 0,
75            events,
76            pusher,
77            phantom: ::std::marker::PhantomData,
78            buzzer,
79        }
80    }
81}
82
83impl<T, P: Push<T>> Push<T> for ArcPusher<T, P> {
84    #[inline]
85    fn push(&mut self, element: &mut Option<T>) {
86        // if element.is_none() {
87        //     if self.count != 0 {
88        //         self.events
89        //             .send((self.index, Event::Pushed(self.count)))
90        //             .expect("Failed to send message count");
91        //         self.count = 0;
92        //     }
93        // }
94        // else {
95        //     self.count += 1;
96        // }
97
98        // A `None` is a flush, and the wrapped pusher is unbuffered: nothing
99        // is enqueued, and so there is nothing to announce or to awaken for.
100        if element.is_none() { return; }
101
102        // These three calls should happen in this order, to ensure that
103        // we first enqueue data, second enqueue interest in the channel,
104        // and finally awaken the thread. Other orders are defective when
105        // multiple threads are involved.
106        self.pusher.push(element);
107        let _ = self.events.send(self.index);
108            // TODO : Perhaps this shouldn't be a fatal error (e.g. in shutdown).
109            // .expect("Failed to send message count");
110        self.buzzer.buzz();
111    }
112}
113
114/// The pull half of an intra-thread channel.
115pub struct Puller<T, P: Pull<T>> {
116    index: usize,
117    count: usize,
118    events: Rc<RefCell<Vec<usize>>>,
119    puller: P,
120    phantom: ::std::marker::PhantomData<T>,
121}
122
123impl<T, P: Pull<T>>  Puller<T, P> {
124    /// Wraps a puller with a message counter.
125    pub fn new(puller: P, index: usize, events: Rc<RefCell<Vec<usize>>>) -> Self {
126        Puller {
127            index,
128            count: 0,
129            events,
130            puller,
131            phantom: ::std::marker::PhantomData,
132        }
133    }
134}
135impl<T, P: Pull<T>> Pull<T> for Puller<T, P> {
136    #[inline]
137    fn pull(&mut self) -> &mut Option<T> {
138        let result = self.puller.pull();
139        if result.is_none() {
140            if self.count != 0 {
141                self.events
142                    .borrow_mut()
143                    .push(self.index);
144                self.count = 0;
145            }
146        }
147        else {
148            self.count += 1;
149        }
150
151        result
152    }
153}
timely_communication/allocator/counters.rs

timely_communication/allocator/
counters.rs
