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/NeverDestroyed.h>
32#include <AK/Time.h>
33#include <LibCore/Event.h>
34#include <LibCore/EventLoop.h>
35#include <LibCore/LocalServer.h>
36#include <LibCore/LocalSocket.h>
37#include <LibCore/Notifier.h>
38#include <LibCore/Object.h>
39#include <LibCore/SyscallUtils.h>
40#include <LibThread/Lock.h>
41#include <errno.h>
42#include <fcntl.h>
43#include <stdio.h>
44#include <stdlib.h>
45#include <string.h>
46#include <sys/select.h>
47#include <sys/socket.h>
48#include <sys/time.h>
49#include <time.h>
50#include <unistd.h>
51
52//#define CEVENTLOOP_DEBUG
53//#define DEFERRED_INVOKE_DEBUG
54
55namespace Core {
56
57class RPCClient;
58
59struct EventLoopTimer {
60 int timer_id { 0 };
61 int interval { 0 };
62 timeval fire_time { 0, 0 };
63 bool should_reload { false };
64 TimerShouldFireWhenNotVisible fire_when_not_visible { TimerShouldFireWhenNotVisible::No };
65 WeakPtr<Object> owner;
66
67 void reload(const timeval& now);
68 bool has_expired(const timeval& now) const;
69};
70
71struct EventLoop::Private {
72 LibThread::Lock lock;
73};
74
75static EventLoop* s_main_event_loop;
76static Vector<EventLoop*>* s_event_loop_stack;
77static NeverDestroyed<IDAllocator> s_id_allocator;
78static HashMap<int, NonnullOwnPtr<EventLoopTimer>>* s_timers;
79static HashTable<Notifier*>* s_notifiers;
80int EventLoop::s_wake_pipe_fds[2];
81static RefPtr<LocalServer> s_rpc_server;
82HashMap<int, RefPtr<RPCClient>> s_rpc_clients;
83
84class RPCClient : public Object {
85 C_OBJECT(RPCClient)
86public:
87 explicit RPCClient(RefPtr<LocalSocket> socket)
88 : m_socket(move(socket))
89 , m_client_id(s_id_allocator->allocate())
90 {
91 s_rpc_clients.set(m_client_id, this);
92 add_child(*m_socket);
93 m_socket->on_ready_to_read = [this] {
94 u32 length;
95 int nread = m_socket->read((u8*)&length, sizeof(length));
96 if (nread == 0) {
97 dbg() << "RPC client disconnected";
98 shutdown();
99 return;
100 }
101 ASSERT(nread == sizeof(length));
102 auto request = m_socket->read(length);
103
104 auto request_json = JsonValue::from_string(request);
105 if (!request_json.is_object()) {
106 dbg() << "RPC client sent invalid request";
107 shutdown();
108 return;
109 }
110
111 handle_request(request_json.as_object());
112 };
113 }
114 virtual ~RPCClient() override
115 {
116 if (m_inspected_object)
117 m_inspected_object->decrement_inspector_count({});
118 }
119
120 void send_response(const JsonObject& response)
121 {
122 auto serialized = response.to_string();
123 u32 length = serialized.length();
124 m_socket->write((const u8*)&length, sizeof(length));
125 m_socket->write(serialized);
126 }
127
128 void handle_request(const JsonObject& request)
129 {
130 auto type = request.get("type").as_string_or({});
131
132 if (type.is_null()) {
133 dbg() << "RPC client sent request without type field";
134 return;
135 }
136
137 if (type == "Identify") {
138 JsonObject response;
139 response.set("type", type);
140 response.set("pid", getpid());
141#ifdef __serenity__
142 char buffer[1024];
143 if (get_process_name(buffer, sizeof(buffer)) >= 0) {
144 response.set("process_name", buffer);
145 } else {
146 response.set("process_name", JsonValue());
147 }
148#endif
149 send_response(response);
150 return;
151 }
152
153 if (type == "GetAllObjects") {
154 JsonObject response;
155 response.set("type", type);
156 JsonArray objects;
157 for (auto& object : Object::all_objects()) {
158 JsonObject json_object;
159 object.save_to(json_object);
160 objects.append(move(json_object));
161 }
162 response.set("objects", move(objects));
163 send_response(response);
164 return;
165 }
166
167 if (type == "SetInspectedObject") {
168 auto address = request.get("address").to_number<FlatPtr>();
169 for (auto& object : Object::all_objects()) {
170 if ((FlatPtr)&object == address) {
171 if (m_inspected_object)
172 m_inspected_object->decrement_inspector_count({});
173 m_inspected_object = object.make_weak_ptr();
174 m_inspected_object->increment_inspector_count({});
175 break;
176 }
177 }
178 return;
179 }
180
181 if (type == "SetProperty") {
182 auto address = request.get("address").to_number<FlatPtr>();
183 for (auto& object : Object::all_objects()) {
184 if ((FlatPtr)&object == address) {
185 bool success = object.set_property(request.get("name").to_string(), request.get("value"));
186 JsonObject response;
187 response.set("type", "SetProperty");
188 response.set("success", success);
189 send_response(response);
190 break;
191 }
192 }
193 return;
194 }
195
196 if (type == "Disconnect") {
197 shutdown();
198 return;
199 }
200 }
201
202 void shutdown()
203 {
204 s_rpc_clients.remove(m_client_id);
205 s_id_allocator->deallocate(m_client_id);
206 }
207
208private:
209 RefPtr<LocalSocket> m_socket;
210 WeakPtr<Object> m_inspected_object;
211 int m_client_id { -1 };
212};
213
214EventLoop::EventLoop()
215 : m_private(make<Private>())
216{
217 if (!s_event_loop_stack) {
218 s_event_loop_stack = new Vector<EventLoop*>;
219 s_timers = new HashMap<int, NonnullOwnPtr<EventLoopTimer>>;
220 s_notifiers = new HashTable<Notifier*>;
221 }
222
223 if (!s_main_event_loop) {
224 s_main_event_loop = this;
225#if defined(SOCK_NONBLOCK)
226 int rc = pipe2(s_wake_pipe_fds, O_CLOEXEC);
227#else
228 int rc = pipe(s_wake_pipe_fds);
229 fcntl(s_wake_pipe_fds[0], F_SETFD, FD_CLOEXEC);
230 fcntl(s_wake_pipe_fds[1], F_SETFD, FD_CLOEXEC);
231
232#endif
233 ASSERT(rc == 0);
234 s_event_loop_stack->append(this);
235
236 auto rpc_path = String::format("/tmp/rpc.%d", getpid());
237 rc = unlink(rpc_path.characters());
238 if (rc < 0 && errno != ENOENT) {
239 perror("unlink");
240 ASSERT_NOT_REACHED();
241 }
242 s_rpc_server = LocalServer::construct();
243 s_rpc_server->set_name("Core::EventLoop_RPC_server");
244 bool listening = s_rpc_server->listen(rpc_path);
245 ASSERT(listening);
246
247 s_rpc_server->on_ready_to_accept = [&] {
248 RPCClient::construct(s_rpc_server->accept());
249 };
250 }
251
252#ifdef CEVENTLOOP_DEBUG
253 dbg() << getpid() << " Core::EventLoop constructed :)";
254#endif
255}
256
257EventLoop::~EventLoop()
258{
259}
260
261EventLoop& EventLoop::main()
262{
263 ASSERT(s_main_event_loop);
264 return *s_main_event_loop;
265}
266
267EventLoop& EventLoop::current()
268{
269 EventLoop* event_loop = s_event_loop_stack->last();
270 ASSERT(event_loop != nullptr);
271 return *event_loop;
272}
273
274void EventLoop::quit(int code)
275{
276 dbg() << "Core::EventLoop::quit(" << code << ")";
277 m_exit_requested = true;
278 m_exit_code = code;
279}
280
281void EventLoop::unquit()
282{
283 dbg() << "Core::EventLoop::unquit()";
284 m_exit_requested = false;
285 m_exit_code = 0;
286}
287
288struct EventLoopPusher {
289public:
290 EventLoopPusher(EventLoop& event_loop)
291 : m_event_loop(event_loop)
292 {
293 if (&m_event_loop != s_main_event_loop) {
294 m_event_loop.take_pending_events_from(EventLoop::current());
295 s_event_loop_stack->append(&event_loop);
296 }
297 }
298 ~EventLoopPusher()
299 {
300 if (&m_event_loop != s_main_event_loop) {
301 s_event_loop_stack->take_last();
302 EventLoop::current().take_pending_events_from(m_event_loop);
303 }
304 }
305
306private:
307 EventLoop& m_event_loop;
308};
309
310int EventLoop::exec()
311{
312 EventLoopPusher pusher(*this);
313 for (;;) {
314 if (m_exit_requested)
315 return m_exit_code;
316 pump();
317 }
318 ASSERT_NOT_REACHED();
319}
320
321void EventLoop::pump(WaitMode mode)
322{
323 if (m_queued_events.is_empty())
324 wait_for_event(mode);
325
326 decltype(m_queued_events) events;
327 {
328 LOCKER(m_private->lock);
329 events = move(m_queued_events);
330 }
331
332 for (size_t i = 0; i < events.size(); ++i) {
333 auto& queued_event = events.at(i);
334#ifndef __clang__
335 ASSERT(queued_event.event);
336#endif
337 auto* receiver = queued_event.receiver.ptr();
338 auto& event = *queued_event.event;
339#ifdef CEVENTLOOP_DEBUG
340 if (receiver)
341 dbg() << "Core::EventLoop: " << *receiver << " event " << (int)event.type();
342#endif
343 if (!receiver) {
344 switch (event.type()) {
345 case Event::Quit:
346 ASSERT_NOT_REACHED();
347 return;
348 default:
349 dbg() << "Event type " << event.type() << " with no receiver :(";
350 }
351 } else if (event.type() == Event::Type::DeferredInvoke) {
352#ifdef DEFERRED_INVOKE_DEBUG
353 printf("DeferredInvoke: receiver=%s{%p}\n", receiver->class_name(), receiver);
354#endif
355 static_cast<DeferredInvocationEvent&>(event).m_invokee(*receiver);
356 } else {
357 NonnullRefPtr<Object> protector(*receiver);
358 receiver->dispatch_event(event);
359 }
360
361 if (m_exit_requested) {
362 LOCKER(m_private->lock);
363#ifdef CEVENTLOOP_DEBUG
364 dbg() << "Core::EventLoop: Exit requested. Rejigging " << (events.size() - i) << " events.";
365#endif
366 decltype(m_queued_events) new_event_queue;
367 new_event_queue.ensure_capacity(m_queued_events.size() + events.size());
368 for (++i; i < events.size(); ++i)
369 new_event_queue.unchecked_append(move(events[i]));
370 new_event_queue.append(move(m_queued_events));
371 m_queued_events = move(new_event_queue);
372 return;
373 }
374 }
375}
376
377void EventLoop::post_event(Object& receiver, NonnullOwnPtr<Event>&& event)
378{
379 LOCKER(m_private->lock);
380#ifdef CEVENTLOOP_DEBUG
381 dbg() << "Core::EventLoop::post_event: {" << m_queued_events.size() << "} << receiver=" << receiver << ", event=" << event;
382#endif
383 m_queued_events.empend(receiver, move(event));
384}
385
386void EventLoop::wait_for_event(WaitMode mode)
387{
388 fd_set rfds;
389 fd_set wfds;
390 FD_ZERO(&rfds);
391 FD_ZERO(&wfds);
392
393 int max_fd = 0;
394 auto add_fd_to_set = [&max_fd](int fd, fd_set& set) {
395 FD_SET(fd, &set);
396 if (fd > max_fd)
397 max_fd = fd;
398 };
399
400 int max_fd_added = -1;
401 add_fd_to_set(s_wake_pipe_fds[0], rfds);
402 max_fd = max(max_fd, max_fd_added);
403 for (auto& notifier : *s_notifiers) {
404 if (notifier->event_mask() & Notifier::Read)
405 add_fd_to_set(notifier->fd(), rfds);
406 if (notifier->event_mask() & Notifier::Write)
407 add_fd_to_set(notifier->fd(), wfds);
408 if (notifier->event_mask() & Notifier::Exceptional)
409 ASSERT_NOT_REACHED();
410 }
411
412 bool queued_events_is_empty;
413 {
414 LOCKER(m_private->lock);
415 queued_events_is_empty = m_queued_events.is_empty();
416 }
417
418 timeval now;
419 struct timeval timeout = { 0, 0 };
420 bool should_wait_forever = false;
421 if (mode == WaitMode::WaitForEvents) {
422 if (!s_timers->is_empty() && queued_events_is_empty) {
423 timespec now_spec;
424 clock_gettime(CLOCK_MONOTONIC, &now_spec);
425 now.tv_sec = now_spec.tv_sec;
426 now.tv_usec = now_spec.tv_nsec / 1000;
427 get_next_timer_expiration(timeout);
428 timeval_sub(timeout, now, timeout);
429 if (timeout.tv_sec < 0) {
430 timeout.tv_sec = 0;
431 timeout.tv_usec = 0;
432 }
433 } else {
434 should_wait_forever = true;
435 }
436 } else {
437 should_wait_forever = false;
438 }
439
440 int marked_fd_count = Core::safe_syscall(select, max_fd + 1, &rfds, &wfds, nullptr, should_wait_forever ? nullptr : &timeout);
441 if (FD_ISSET(s_wake_pipe_fds[0], &rfds)) {
442 char buffer[32];
443 auto nread = read(s_wake_pipe_fds[0], buffer, sizeof(buffer));
444 if (nread < 0) {
445 perror("read from wake pipe");
446 ASSERT_NOT_REACHED();
447 }
448 ASSERT(nread > 0);
449 }
450
451 if (!s_timers->is_empty()) {
452 timespec now_spec;
453 clock_gettime(CLOCK_MONOTONIC, &now_spec);
454 now.tv_sec = now_spec.tv_sec;
455 now.tv_usec = now_spec.tv_nsec / 1000;
456 }
457
458 for (auto& it : *s_timers) {
459 auto& timer = *it.value;
460 if (!timer.has_expired(now))
461 continue;
462 if (it.value->fire_when_not_visible == TimerShouldFireWhenNotVisible::No
463 && it.value->owner
464 && !it.value->owner->is_visible_for_timer_purposes()) {
465 continue;
466 }
467#ifdef CEVENTLOOP_DEBUG
468 dbg() << "Core::EventLoop: Timer " << timer.timer_id << " has expired, sending Core::TimerEvent to " << timer.owner;
469#endif
470 post_event(*timer.owner, make<TimerEvent>(timer.timer_id));
471 if (timer.should_reload) {
472 timer.reload(now);
473 } else {
474 // FIXME: Support removing expired timers that don't want to reload.
475 ASSERT_NOT_REACHED();
476 }
477 }
478
479 if (!marked_fd_count)
480 return;
481
482 for (auto& notifier : *s_notifiers) {
483 if (FD_ISSET(notifier->fd(), &rfds)) {
484 if (notifier->on_ready_to_read)
485 post_event(*notifier, make<NotifierReadEvent>(notifier->fd()));
486 }
487 if (FD_ISSET(notifier->fd(), &wfds)) {
488 if (notifier->on_ready_to_write)
489 post_event(*notifier, make<NotifierWriteEvent>(notifier->fd()));
490 }
491 }
492}
493
494bool EventLoopTimer::has_expired(const timeval& now) const
495{
496 return now.tv_sec > fire_time.tv_sec || (now.tv_sec == fire_time.tv_sec && now.tv_usec >= fire_time.tv_usec);
497}
498
499void EventLoopTimer::reload(const timeval& now)
500{
501 fire_time = now;
502 fire_time.tv_sec += interval / 1000;
503 fire_time.tv_usec += (interval % 1000) * 1000;
504}
505
506void EventLoop::get_next_timer_expiration(timeval& soonest)
507{
508 ASSERT(!s_timers->is_empty());
509 bool has_checked_any = false;
510 for (auto& it : *s_timers) {
511 auto& fire_time = it.value->fire_time;
512 if (it.value->fire_when_not_visible == TimerShouldFireWhenNotVisible::No
513 && it.value->owner
514 && !it.value->owner->is_visible_for_timer_purposes()) {
515 continue;
516 }
517 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))
518 soonest = fire_time;
519 has_checked_any = true;
520 }
521}
522
523int EventLoop::register_timer(Object& object, int milliseconds, bool should_reload, TimerShouldFireWhenNotVisible fire_when_not_visible)
524{
525 ASSERT(milliseconds >= 0);
526 auto timer = make<EventLoopTimer>();
527 timer->owner = object.make_weak_ptr();
528 timer->interval = milliseconds;
529 timeval now;
530 timespec now_spec;
531 clock_gettime(CLOCK_MONOTONIC, &now_spec);
532 now.tv_sec = now_spec.tv_sec;
533 now.tv_usec = now_spec.tv_nsec / 1000;
534 timer->reload(now);
535 timer->should_reload = should_reload;
536 timer->fire_when_not_visible = fire_when_not_visible;
537 int timer_id = s_id_allocator->allocate();
538 timer->timer_id = timer_id;
539 s_timers->set(timer_id, move(timer));
540 return timer_id;
541}
542
543bool EventLoop::unregister_timer(int timer_id)
544{
545 s_id_allocator->deallocate(timer_id);
546 auto it = s_timers->find(timer_id);
547 if (it == s_timers->end())
548 return false;
549 s_timers->remove(it);
550 return true;
551}
552
553void EventLoop::register_notifier(Badge<Notifier>, Notifier& notifier)
554{
555 s_notifiers->set(¬ifier);
556}
557
558void EventLoop::unregister_notifier(Badge<Notifier>, Notifier& notifier)
559{
560 s_notifiers->remove(¬ifier);
561}
562
563void EventLoop::wake()
564{
565 char ch = '!';
566 int nwritten = write(s_wake_pipe_fds[1], &ch, 1);
567 if (nwritten < 0) {
568 perror("EventLoop::wake: write");
569 ASSERT_NOT_REACHED();
570 }
571}
572
573EventLoop::QueuedEvent::QueuedEvent(Object& receiver, NonnullOwnPtr<Event> event)
574 : receiver(receiver.make_weak_ptr())
575 , event(move(event))
576{
577}
578
579EventLoop::QueuedEvent::QueuedEvent(QueuedEvent&& other)
580 : receiver(other.receiver)
581 , event(move(other.event))
582{
583}
584
585EventLoop::QueuedEvent::~QueuedEvent()
586{
587}
588
589}