jcs.org
Serenity Operating System
1/*
2 * Copyright (c) 2018-2020, Andreas Kling <kling@serenityos.org>
3 * All rights reserved.
4 *
5 * Redistribution and use in source and binary forms, with or without
6 * modification, are permitted provided that the following conditions are met:
7 *
8 * 1. Redistributions of source code must retain the above copyright notice, this
9 * list of conditions and the following disclaimer.
10 *
11 * 2. Redistributions in binary form must reproduce the above copyright notice,
12 * this list of conditions and the following disclaimer in the documentation
13 * and/or other materials provided with the distribution.
14 *
15 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
16 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
17 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
18 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
19 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
20 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
21 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
22 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
23 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
24 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
25 */
26
27#include <AK/Badge.h>
28#include <AK/IDAllocator.h>
29#include <AK/JsonObject.h>
30#include <AK/JsonValue.h>
31#include <AK/Time.h>
32#include <LibCore/Event.h>
33#include <LibCore/EventLoop.h>
34#include <LibCore/LocalServer.h>
35#include <LibCore/LocalSocket.h>
36#include <LibCore/Notifier.h>
37#include <LibCore/Object.h>
38#include <LibCore/SyscallUtils.h>
39#include <LibThread/Lock.h>
40#include <errno.h>
41#include <fcntl.h>
42#include <stdio.h>
43#include <stdlib.h>
44#include <string.h>
45#include <sys/select.h>
46#include <sys/socket.h>
47#include <sys/time.h>
48#include <time.h>
49#include <unistd.h>
50
51//#define CEVENTLOOP_DEBUG
52//#define DEFERRED_INVOKE_DEBUG
53
54namespace Core {
55
56class RPCClient;
57
58struct EventLoopTimer {
59 int timer_id { 0 };
60 int interval { 0 };
61 timeval fire_time { 0, 0 };
62 bool should_reload { false };
63 TimerShouldFireWhenNotVisible fire_when_not_visible { TimerShouldFireWhenNotVisible::No };
64 WeakPtr<Object> owner;
65
66 void reload(const timeval& now);
67 bool has_expired(const timeval& now) const;
68};
69
70struct EventLoop::Private {
71 LibThread::Lock lock;
72};
73
74static EventLoop* s_main_event_loop;
75static Vector<EventLoop*>* s_event_loop_stack;
76static IDAllocator s_id_allocator;
77static HashMap<int, NonnullOwnPtr<EventLoopTimer>>* s_timers;
78static HashTable<Notifier*>* s_notifiers;
79int EventLoop::s_wake_pipe_fds[2];
80static RefPtr<LocalServer> s_rpc_server;
81HashMap<int, RefPtr<RPCClient>> s_rpc_clients;
82
83class RPCClient : public Object {
84 C_OBJECT(RPCClient)
85public:
86 explicit RPCClient(RefPtr<LocalSocket> socket)
87 : m_socket(move(socket))
88 , m_client_id(s_id_allocator.allocate())
89 {
90 s_rpc_clients.set(m_client_id, this);
91 add_child(*m_socket);
92 m_socket->on_ready_to_read = [this] {
93 u32 length;
94 int nread = m_socket->read((u8*)&length, sizeof(length));
95 if (nread == 0) {
96 dbg() << "RPC client disconnected";
97 shutdown();
98 return;
99 }
100 ASSERT(nread == sizeof(length));
101 auto request = m_socket->read(length);
102
103 auto request_json = JsonValue::from_string(request);
104 if (!request_json.is_object()) {
105 dbg() << "RPC client sent invalid request";
106 shutdown();
107 return;
108 }
109
110 handle_request(request_json.as_object());
111 };
112 }
113 virtual ~RPCClient() override
114 {
115 }
116
117 void send_response(const JsonObject& response)
118 {
119 auto serialized = response.to_string();
120 u32 length = serialized.length();
121 m_socket->write((const u8*)&length, sizeof(length));
122 m_socket->write(serialized);
123 }
124
125 void handle_request(const JsonObject& request)
126 {
127 auto type = request.get("type").as_string_or({});
128
129 if (type.is_null()) {
130 dbg() << "RPC client sent request without type field";
131 return;
132 }
133
134 if (type == "Identify") {
135 JsonObject response;
136 response.set("type", type);
137 response.set("pid", getpid());
138#ifdef __serenity__
139 char buffer[1024];
140 if (get_process_name(buffer, sizeof(buffer)) >= 0) {
141 response.set("process_name", buffer);
142 } else {
143 response.set("process_name", JsonValue());
144 }
145#endif
146 send_response(response);
147 return;
148 }
149
150 if (type == "GetAllObjects") {
151 JsonObject response;
152 response.set("type", type);
153 JsonArray objects;
154 for (auto& object : Object::all_objects()) {
155 JsonObject json_object;
156 object.save_to(json_object);
157 objects.append(move(json_object));
158 }
159 response.set("objects", move(objects));
160 send_response(response);
161 return;
162 }
163
164 if (type == "Disconnect") {
165 shutdown();
166 return;
167 }
168 }
169
170 void shutdown()
171 {
172 s_rpc_clients.remove(m_client_id);
173 s_id_allocator.deallocate(m_client_id);
174 }
175
176private:
177 RefPtr<LocalSocket> m_socket;
178 int m_client_id { -1 };
179};
180
181EventLoop::EventLoop()
182 : m_private(make<Private>())
183{
184 if (!s_event_loop_stack) {
185 s_event_loop_stack = new Vector<EventLoop*>;
186 s_timers = new HashMap<int, NonnullOwnPtr<EventLoopTimer>>;
187 s_notifiers = new HashTable<Notifier*>;
188 }
189
190 if (!s_main_event_loop) {
191 s_main_event_loop = this;
192#if defined(SOCK_NONBLOCK)
193 int rc = pipe2(s_wake_pipe_fds, O_CLOEXEC);
194#else
195 int rc = pipe(s_wake_pipe_fds);
196 fcntl(s_wake_pipe_fds[0], F_SETFD, FD_CLOEXEC);
197 fcntl(s_wake_pipe_fds[1], F_SETFD, FD_CLOEXEC);
198
199#endif
200 ASSERT(rc == 0);
201 s_event_loop_stack->append(this);
202
203 auto rpc_path = String::format("/tmp/rpc.%d", getpid());
204 rc = unlink(rpc_path.characters());
205 if (rc < 0 && errno != ENOENT) {
206 perror("unlink");
207 ASSERT_NOT_REACHED();
208 }
209 s_rpc_server = LocalServer::construct();
210 s_rpc_server->set_name("Core::EventLoop_RPC_server");
211 bool listening = s_rpc_server->listen(rpc_path);
212 ASSERT(listening);
213
214 s_rpc_server->on_ready_to_accept = [&] {
215 RPCClient::construct(s_rpc_server->accept());
216 };
217 }
218
219#ifdef CEVENTLOOP_DEBUG
220 dbg() << getpid() << " Core::EventLoop constructed :)";
221#endif
222}
223
224EventLoop::~EventLoop()
225{
226}
227
228EventLoop& EventLoop::main()
229{
230 ASSERT(s_main_event_loop);
231 return *s_main_event_loop;
232}
233
234EventLoop& EventLoop::current()
235{
236 EventLoop* event_loop = s_event_loop_stack->last();
237 ASSERT(event_loop != nullptr);
238 return *event_loop;
239}
240
241void EventLoop::quit(int code)
242{
243 dbg() << "Core::EventLoop::quit(" << code << ")";
244 m_exit_requested = true;
245 m_exit_code = code;
246}
247
248void EventLoop::unquit()
249{
250 dbg() << "Core::EventLoop::unquit()";
251 m_exit_requested = false;
252 m_exit_code = 0;
253}
254
255struct EventLoopPusher {
256public:
257 EventLoopPusher(EventLoop& event_loop)
258 : m_event_loop(event_loop)
259 {
260 if (&m_event_loop != s_main_event_loop) {
261 m_event_loop.take_pending_events_from(EventLoop::current());
262 s_event_loop_stack->append(&event_loop);
263 }
264 }
265 ~EventLoopPusher()
266 {
267 if (&m_event_loop != s_main_event_loop) {
268 s_event_loop_stack->take_last();
269 EventLoop::current().take_pending_events_from(m_event_loop);
270 }
271 }
272
273private:
274 EventLoop& m_event_loop;
275};
276
277int EventLoop::exec()
278{
279 EventLoopPusher pusher(*this);
280 for (;;) {
281 if (m_exit_requested)
282 return m_exit_code;
283 pump();
284 }
285 ASSERT_NOT_REACHED();
286}
287
288void EventLoop::pump(WaitMode mode)
289{
290 if (m_queued_events.is_empty())
291 wait_for_event(mode);
292
293 decltype(m_queued_events) events;
294 {
295 LOCKER(m_private->lock);
296 events = move(m_queued_events);
297 }
298
299 for (size_t i = 0; i < events.size(); ++i) {
300 auto& queued_event = events.at(i);
301#ifndef __clang__
302 ASSERT(queued_event.event);
303#endif
304 auto* receiver = queued_event.receiver.ptr();
305 auto& event = *queued_event.event;
306#ifdef CEVENTLOOP_DEBUG
307 if (receiver)
308 dbg() << "Core::EventLoop: " << *receiver << " event " << (int)event.type();
309#endif
310 if (!receiver) {
311 switch (event.type()) {
312 case Event::Quit:
313 ASSERT_NOT_REACHED();
314 return;
315 default:
316 dbg() << "Event type " << event.type() << " with no receiver :(";
317 }
318 } else if (event.type() == Event::Type::DeferredInvoke) {
319#ifdef DEFERRED_INVOKE_DEBUG
320 printf("DeferredInvoke: receiver=%s{%p}\n", receiver->class_name(), receiver);
321#endif
322 static_cast<DeferredInvocationEvent&>(event).m_invokee(*receiver);
323 } else {
324 NonnullRefPtr<Object> protector(*receiver);
325 receiver->dispatch_event(event);
326 }
327
328 if (m_exit_requested) {
329 LOCKER(m_private->lock);
330#ifdef CEVENTLOOP_DEBUG
331 dbg() << "Core::EventLoop: Exit requested. Rejigging " << (events.size() - i) << " events.";
332#endif
333 decltype(m_queued_events) new_event_queue;
334 new_event_queue.ensure_capacity(m_queued_events.size() + events.size());
335 for (; i < events.size(); ++i)
336 new_event_queue.unchecked_append(move(events[i]));
337 new_event_queue.append(move(m_queued_events));
338 m_queued_events = move(new_event_queue);
339 return;
340 }
341 }
342}
343
344void EventLoop::post_event(Object& receiver, NonnullOwnPtr<Event>&& event)
345{
346 LOCKER(m_private->lock);
347#ifdef CEVENTLOOP_DEBUG
348 dbg() << "Core::EventLoop::post_event: {" << m_queued_events.size() << "} << receiver=" << receiver << ", event=" << event;
349#endif
350 m_queued_events.empend(receiver, move(event));
351}
352
353void EventLoop::wait_for_event(WaitMode mode)
354{
355 fd_set rfds;
356 fd_set wfds;
357 FD_ZERO(&rfds);
358 FD_ZERO(&wfds);
359
360 int max_fd = 0;
361 auto add_fd_to_set = [&max_fd](int fd, fd_set& set) {
362 FD_SET(fd, &set);
363 if (fd > max_fd)
364 max_fd = fd;
365 };
366
367 int max_fd_added = -1;
368 add_fd_to_set(s_wake_pipe_fds[0], rfds);
369 max_fd = max(max_fd, max_fd_added);
370 for (auto& notifier : *s_notifiers) {
371 if (notifier->event_mask() & Notifier::Read)
372 add_fd_to_set(notifier->fd(), rfds);
373 if (notifier->event_mask() & Notifier::Write)
374 add_fd_to_set(notifier->fd(), wfds);
375 if (notifier->event_mask() & Notifier::Exceptional)
376 ASSERT_NOT_REACHED();
377 }
378
379 bool queued_events_is_empty;
380 {
381 LOCKER(m_private->lock);
382 queued_events_is_empty = m_queued_events.is_empty();
383 }
384
385 timeval now;
386 struct timeval timeout = { 0, 0 };
387 bool should_wait_forever = false;
388 if (mode == WaitMode::WaitForEvents) {
389 if (!s_timers->is_empty() && queued_events_is_empty) {
390 gettimeofday(&now, nullptr);
391 get_next_timer_expiration(timeout);
392 timeval_sub(timeout, now, timeout);
393 if (timeout.tv_sec < 0) {
394 timeout.tv_sec = 0;
395 timeout.tv_usec = 0;
396 }
397 } else {
398 should_wait_forever = true;
399 }
400 } else {
401 should_wait_forever = false;
402 }
403
404 int marked_fd_count = Core::safe_syscall(select, max_fd + 1, &rfds, &wfds, nullptr, should_wait_forever ? nullptr : &timeout);
405 if (FD_ISSET(s_wake_pipe_fds[0], &rfds)) {
406 char buffer[32];
407 auto nread = read(s_wake_pipe_fds[0], buffer, sizeof(buffer));
408 if (nread < 0) {
409 perror("read from wake pipe");
410 ASSERT_NOT_REACHED();
411 }
412 ASSERT(nread > 0);
413 }
414
415 if (!s_timers->is_empty()) {
416 gettimeofday(&now, nullptr);
417 }
418
419 for (auto& it : *s_timers) {
420 auto& timer = *it.value;
421 if (!timer.has_expired(now))
422 continue;
423 if (it.value->fire_when_not_visible == TimerShouldFireWhenNotVisible::No
424 && it.value->owner
425 && !it.value->owner->is_visible_for_timer_purposes()) {
426 continue;
427 }
428#ifdef CEVENTLOOP_DEBUG
429 dbg() << "Core::EventLoop: Timer " << timer.timer_id << " has expired, sending Core::TimerEvent to " << timer.owner;
430#endif
431 post_event(*timer.owner, make<TimerEvent>(timer.timer_id));
432 if (timer.should_reload) {
433 timer.reload(now);
434 } else {
435 // FIXME: Support removing expired timers that don't want to reload.
436 ASSERT_NOT_REACHED();
437 }
438 }
439
440 if (!marked_fd_count)
441 return;
442
443 for (auto& notifier : *s_notifiers) {
444 if (FD_ISSET(notifier->fd(), &rfds)) {
445 if (notifier->on_ready_to_read)
446 post_event(*notifier, make<NotifierReadEvent>(notifier->fd()));
447 }
448 if (FD_ISSET(notifier->fd(), &wfds)) {
449 if (notifier->on_ready_to_write)
450 post_event(*notifier, make<NotifierWriteEvent>(notifier->fd()));
451 }
452 }
453}
454
455bool EventLoopTimer::has_expired(const timeval& now) const
456{
457 return now.tv_sec > fire_time.tv_sec || (now.tv_sec == fire_time.tv_sec && now.tv_usec >= fire_time.tv_usec);
458}
459
460void EventLoopTimer::reload(const timeval& now)
461{
462 fire_time = now;
463 fire_time.tv_sec += interval / 1000;
464 fire_time.tv_usec += (interval % 1000) * 1000;
465}
466
467void EventLoop::get_next_timer_expiration(timeval& soonest)
468{
469 ASSERT(!s_timers->is_empty());
470 bool has_checked_any = false;
471 for (auto& it : *s_timers) {
472 auto& fire_time = it.value->fire_time;
473 if (it.value->fire_when_not_visible == TimerShouldFireWhenNotVisible::No
474 && it.value->owner
475 && !it.value->owner->is_visible_for_timer_purposes()) {
476 continue;
477 }
478 if (!has_checked_any || fire_time.tv_sec < soonest.tv_sec || (fire_time.tv_sec == soonest.tv_sec && fire_time.tv_usec < soonest.tv_usec))
479 soonest = fire_time;
480 has_checked_any = true;
481 }
482}
483
484int EventLoop::register_timer(Object& object, int milliseconds, bool should_reload, TimerShouldFireWhenNotVisible fire_when_not_visible)
485{
486 ASSERT(milliseconds >= 0);
487 auto timer = make<EventLoopTimer>();
488 timer->owner = object.make_weak_ptr();
489 timer->interval = milliseconds;
490 timeval now;
491 gettimeofday(&now, nullptr);
492 timer->reload(now);
493 timer->should_reload = should_reload;
494 timer->fire_when_not_visible = fire_when_not_visible;
495 int timer_id = s_id_allocator.allocate();
496 timer->timer_id = timer_id;
497 s_timers->set(timer_id, move(timer));
498 return timer_id;
499}
500
501bool EventLoop::unregister_timer(int timer_id)
502{
503 s_id_allocator.deallocate(timer_id);
504 auto it = s_timers->find(timer_id);
505 if (it == s_timers->end())
506 return false;
507 s_timers->remove(it);
508 return true;
509}
510
511void EventLoop::register_notifier(Badge<Notifier>, Notifier& notifier)
512{
513 s_notifiers->set(¬ifier);
514}
515
516void EventLoop::unregister_notifier(Badge<Notifier>, Notifier& notifier)
517{
518 s_notifiers->remove(¬ifier);
519}
520
521void EventLoop::wake()
522{
523 char ch = '!';
524 int nwritten = write(s_wake_pipe_fds[1], &ch, 1);
525 if (nwritten < 0) {
526 perror("EventLoop::wake: write");
527 ASSERT_NOT_REACHED();
528 }
529}
530
531EventLoop::QueuedEvent::QueuedEvent(Object& receiver, NonnullOwnPtr<Event> event)
532 : receiver(receiver.make_weak_ptr())
533 , event(move(event))
534{
535}
536
537EventLoop::QueuedEvent::QueuedEvent(QueuedEvent&& other)
538 : receiver(other.receiver)
539 , event(move(other.event))
540{
541}
542
543EventLoop::QueuedEvent::~QueuedEvent()
544{
545}
546
547}