tenant.openagents/omega

1use calloop::{
2    EventLoop, PostAction,
3    channel::{self, Sender},
4    timer::TimeoutAction,
5};
6use gpui_util::ResultExt;
7
8use std::{mem::MaybeUninit, thread, time::Duration};
9
10use gpui::{
11    PlatformDispatcher, Priority, PriorityQueueReceiver, PriorityQueueSender, RunnableVariant,
12    profiler,
13};
14
15struct TimerAfter {
16    duration: Duration,
17    runnable: RunnableVariant,
18}
19
20pub(crate) struct LinuxDispatcher {
21    main_sender: PriorityQueueCalloopSender<RunnableVariant>,
22    timer_sender: Sender<TimerAfter>,
23    background_sender: PriorityQueueSender<RunnableVariant>,
24    _background_threads: Vec<thread::JoinHandle<()>>,
25    main_thread_id: thread::ThreadId,
26}
27
28const MIN_THREADS: usize = 2;
29
30impl LinuxDispatcher {
31    pub fn new(main_sender: PriorityQueueCalloopSender<RunnableVariant>) -> Self {
32        let (background_sender, background_receiver) = PriorityQueueReceiver::new();
33        let thread_count =
34            std::thread::available_parallelism().map_or(MIN_THREADS, |i| i.get().max(MIN_THREADS));
35
36        let mut background_threads = (0..thread_count)
37            .map(|i| {
38                let receiver: PriorityQueueReceiver<RunnableVariant> = background_receiver.clone();
39                std::thread::Builder::new()
40                    .name(format!("Worker-{i}"))
41                    .spawn(move || {
42                        for runnable in receiver.iter() {
43                            let location = runnable.metadata().location;
44                            let spawned = runnable.metadata().spawned;
45                            profiler::update_running_task(spawned, location);
46                            runnable.run();
47                            profiler::save_task_timing();
48                        }
49                    })
50                    .unwrap()
51            })
52            .collect::<Vec<_>>();
53
54        let (timer_sender, timer_channel) = calloop::channel::channel::<TimerAfter>();
55        let timer_thread = std::thread::Builder::new()
56            .name("Timer".to_owned())
57            .spawn(move || {
58                let mut event_loop: EventLoop<()> =
59                    EventLoop::try_new().expect("Failed to initialize timer loop!");
60
61                let handle = event_loop.handle();
62                let timer_handle = event_loop.handle();
63                handle
64                    .insert_source(timer_channel, move |e, _, _| {
65                        if let channel::Event::Msg(timer) = e {
66                            let mut runnable = Some(timer.runnable);
67                            timer_handle
68                                .insert_source(
69                                    calloop::timer::Timer::from_duration(timer.duration),
70                                    move |_, _, _| {
71                                        if let Some(runnable) = runnable.take() {
72                                            let location = runnable.metadata().location;
73                                            let spawned = runnable.metadata().spawned;
74                                            profiler::update_running_task(spawned, location);
75                                            runnable.run();
76                                            profiler::save_task_timing();
77                                        }
78                                        TimeoutAction::Drop
79                                    },
80                                )
81                                .expect("Failed to start timer");
82                        }
83                    })
84                    .expect("Failed to start timer thread");
85
86                event_loop.run(None, &mut (), |_| {}).log_err();
87            })
88            .unwrap();
89
90        background_threads.push(timer_thread);
91
92        Self {
93            main_sender,
94            timer_sender,
95            background_sender,
96            _background_threads: background_threads,
97            main_thread_id: thread::current().id(),
98        }
99    }
100}
101
102impl PlatformDispatcher for LinuxDispatcher {
103    fn is_main_thread(&self) -> bool {
104        thread::current().id() == self.main_thread_id
105    }
106
107    fn dispatch(&self, runnable: RunnableVariant, priority: Priority) {
108        self.background_sender
109            .send(priority, runnable)
110            .unwrap_or_else(|_| panic!("blocking sender returned without value"));
111    }
112
113    fn dispatch_on_main_thread(&self, runnable: RunnableVariant, priority: Priority) {
114        self.main_sender
115            .send(priority, runnable)
116            .unwrap_or_else(|runnable| {
117                // NOTE: Runnable may wrap a Future that is !Send.
118                //
119                // This is usually safe because we only poll it on the main thread.
120                // However if the send fails, we know that:
121                // 1. main_receiver has been dropped (which implies the app is shutting down)
122                // 2. we are on a background thread.
123                // It is not safe to drop something !Send on the wrong thread, and
124                // the app will exit soon anyway, so we must forget the runnable.
125                std::mem::forget(runnable);
126            });
127    }
128
129    fn dispatch_after(&self, duration: Duration, runnable: RunnableVariant) {
130        if let Err(err) = self.timer_sender.send(TimerAfter { duration, runnable }) {
131            // The timer thread has shut down. Dropping a scheduled runnable cancels its task
132            // and makes the next poll of any awaiter panic. Leaking leaves the task pending,
133            // which is acceptable during shutdown.
134            std::mem::forget(err);
135        }
136    }
137
138    fn spawn_realtime(&self, f: Box<dyn FnOnce() + Send>) {
139        std::thread::spawn(move || {
140            // SAFETY: always safe to call
141            let thread_id = unsafe { libc::pthread_self() };
142
143            let policy = libc::SCHED_FIFO;
144            let sched_priority = 65;
145
146            // SAFETY: all sched_param members are valid when initialized to zero.
147            let mut sched_param =
148                unsafe { MaybeUninit::<libc::sched_param>::zeroed().assume_init() };
149            sched_param.sched_priority = sched_priority;
150            // SAFETY: sched_param is a valid initialized structure
151            let result = unsafe { libc::pthread_setschedparam(thread_id, policy, &sched_param) };
152            if result != 0 {
153                log::warn!("failed to set realtime thread priority");
154            }
155
156            f();
157        });
158    }
159}
160
161pub struct PriorityQueueCalloopSender<T> {
162    sender: PriorityQueueSender<T>,
163    ping: calloop::ping::Ping,
164}
165
166impl<T> PriorityQueueCalloopSender<T> {
167    fn new(tx: PriorityQueueSender<T>, ping: calloop::ping::Ping) -> Self {
168        Self { sender: tx, ping }
169    }
170
171    fn send(&self, priority: Priority, item: T) -> Result<(), gpui::queue::SendError<T>> {
172        let res = self.sender.send(priority, item);
173        if res.is_ok() {
174            self.ping.ping();
175        }
176        res
177    }
178}
179
180impl<T> Drop for PriorityQueueCalloopSender<T> {
181    fn drop(&mut self) {
182        self.ping.ping();
183    }
184}
185
186pub struct PriorityQueueCalloopReceiver<T> {
187    receiver: PriorityQueueReceiver<T>,
188    source: calloop::ping::PingSource,
189    ping: calloop::ping::Ping,
190}
191
192impl<T> PriorityQueueCalloopReceiver<T> {
193    pub fn new() -> (PriorityQueueCalloopSender<T>, Self) {
194        let (ping, source) = calloop::ping::make_ping().expect("Failed to create a Ping.");
195
196        let (tx, rx) = PriorityQueueReceiver::new();
197
198        (
199            PriorityQueueCalloopSender::new(tx, ping.clone()),
200            Self {
201                receiver: rx,
202                source,
203                ping,
204            },
205        )
206    }
207}
208
209use calloop::channel::Event;
210
211#[derive(Debug)]
212pub struct ChannelError(calloop::ping::PingError);
213
214impl std::fmt::Display for ChannelError {
215    #[cfg_attr(feature = "nightly_coverage", coverage(off))]
216    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
217        std::fmt::Display::fmt(&self.0, f)
218    }
219}
220
221impl std::error::Error for ChannelError {
222    #[cfg_attr(feature = "nightly_coverage", coverage(off))]
223    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
224        Some(&self.0)
225    }
226}
227
228impl<T> calloop::EventSource for PriorityQueueCalloopReceiver<T> {
229    type Event = Event<T>;
230    type Metadata = ();
231    type Ret = ();
232    type Error = ChannelError;
233
234    fn process_events<F>(
235        &mut self,
236        readiness: calloop::Readiness,
237        token: calloop::Token,
238        mut callback: F,
239    ) -> Result<calloop::PostAction, Self::Error>
240    where
241        F: FnMut(Self::Event, &mut Self::Metadata) -> Self::Ret,
242    {
243        let mut clear_readiness = false;
244        let mut disconnected = false;
245
246        let action = self
247            .source
248            .process_events(readiness, token, |(), &mut ()| {
249                let mut is_empty = true;
250
251                let receiver = self.receiver.clone();
252                for runnable in receiver.try_iter() {
253                    match runnable {
254                        Ok(r) => {
255                            callback(Event::Msg(r), &mut ());
256                            is_empty = false;
257                        }
258                        Err(_) => {
259                            disconnected = true;
260                        }
261                    }
262                }
263
264                if disconnected {
265                    callback(Event::Closed, &mut ());
266                }
267
268                if is_empty {
269                    clear_readiness = true;
270                }
271            })
272            .map_err(ChannelError)?;
273
274        if disconnected {
275            Ok(PostAction::Remove)
276        } else if clear_readiness {
277            Ok(action)
278        } else {
279            // Re-notify the ping source so we can try again.
280            self.ping.ping();
281            Ok(PostAction::Continue)
282        }
283    }
284
285    fn register(
286        &mut self,
287        poll: &mut calloop::Poll,
288        token_factory: &mut calloop::TokenFactory,
289    ) -> calloop::Result<()> {
290        self.source.register(poll, token_factory)
291    }
292
293    fn reregister(
294        &mut self,
295        poll: &mut calloop::Poll,
296        token_factory: &mut calloop::TokenFactory,
297    ) -> calloop::Result<()> {
298        self.source.reregister(poll, token_factory)
299    }
300
301    fn unregister(&mut self, poll: &mut calloop::Poll) -> calloop::Result<()> {
302        self.source.unregister(poll)
303    }
304}
305
306#[cfg(test)]
307mod tests {
308    use super::*;
309
310    #[test]
311    fn calloop_works() {
312        let mut event_loop = calloop::EventLoop::try_new().unwrap();
313        let handle = event_loop.handle();
314
315        let (tx, rx) = PriorityQueueCalloopReceiver::new();
316
317        struct Data {
318            got_msg: bool,
319            got_closed: bool,
320        }
321
322        let mut data = Data {
323            got_msg: false,
324            got_closed: false,
325        };
326
327        let _channel_token = handle
328            .insert_source(rx, move |evt, &mut (), data: &mut Data| match evt {
329                Event::Msg(()) => {
330                    data.got_msg = true;
331                }
332
333                Event::Closed => {
334                    data.got_closed = true;
335                }
336            })
337            .unwrap();
338
339        // nothing is sent, nothing is received
340        event_loop
341            .dispatch(Some(::std::time::Duration::ZERO), &mut data)
342            .unwrap();
343
344        assert!(!data.got_msg);
345        assert!(!data.got_closed);
346        // a message is send
347
348        tx.send(Priority::Medium, ()).unwrap();
349        event_loop
350            .dispatch(Some(::std::time::Duration::ZERO), &mut data)
351            .unwrap();
352
353        assert!(data.got_msg);
354        assert!(!data.got_closed);
355
356        // the sender is dropped
357        drop(tx);
358        event_loop
359            .dispatch(Some(::std::time::Duration::ZERO), &mut data)
360            .unwrap();
361
362        assert!(data.got_msg);
363        assert!(data.got_closed);
364    }
365}
366