ThreadEventQueue.cpp 3.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132
  1. /*
  2. * Copyright (c) 2023, Andreas Kling <kling@serenityos.org>
  3. *
  4. * SPDX-License-Identifier: BSD-2-Clause
  5. */
  6. #include <AK/Vector.h>
  7. #include <LibCore/DeferredInvocationContext.h>
  8. #include <LibCore/EventLoopImplementation.h>
  9. #include <LibCore/EventReceiver.h>
  10. #include <LibCore/Promise.h>
  11. #include <LibCore/ThreadEventQueue.h>
  12. #include <LibThreading/Mutex.h>
  13. #include <errno.h>
  14. namespace Core {
  15. struct ThreadEventQueue::Private {
  16. struct QueuedEvent {
  17. AK_MAKE_NONCOPYABLE(QueuedEvent);
  18. AK_MAKE_DEFAULT_MOVABLE(QueuedEvent);
  19. public:
  20. QueuedEvent(EventReceiver& receiver, NonnullOwnPtr<Event> event)
  21. : receiver(receiver)
  22. , event(move(event))
  23. {
  24. }
  25. ~QueuedEvent() = default;
  26. WeakPtr<EventReceiver> receiver;
  27. NonnullOwnPtr<Event> event;
  28. };
  29. Threading::Mutex mutex;
  30. Vector<QueuedEvent, 128> queued_events;
  31. Vector<NonnullRefPtr<Promise<NonnullRefPtr<EventReceiver>>>, 16> pending_promises;
  32. bool warned_promise_count { false };
  33. };
  34. static thread_local ThreadEventQueue* s_current_thread_event_queue;
  35. ThreadEventQueue& ThreadEventQueue::current()
  36. {
  37. if (!s_current_thread_event_queue) {
  38. // FIXME: Don't leak these.
  39. s_current_thread_event_queue = new ThreadEventQueue;
  40. }
  41. return *s_current_thread_event_queue;
  42. }
  43. ThreadEventQueue::ThreadEventQueue()
  44. : m_private(make<Private>())
  45. {
  46. }
  47. ThreadEventQueue::~ThreadEventQueue() = default;
  48. void ThreadEventQueue::post_event(Core::EventReceiver& receiver, NonnullOwnPtr<Core::Event> event)
  49. {
  50. {
  51. Threading::MutexLocker lock(m_private->mutex);
  52. m_private->queued_events.empend(receiver, move(event));
  53. }
  54. Core::EventLoopManager::the().did_post_event();
  55. }
  56. void ThreadEventQueue::add_job(NonnullRefPtr<Promise<NonnullRefPtr<EventReceiver>>> promise)
  57. {
  58. Threading::MutexLocker lock(m_private->mutex);
  59. m_private->pending_promises.append(move(promise));
  60. }
  61. void ThreadEventQueue::cancel_all_pending_jobs()
  62. {
  63. Threading::MutexLocker lock(m_private->mutex);
  64. for (auto const& promise : m_private->pending_promises)
  65. promise->reject(Error::from_errno(ECANCELED));
  66. m_private->pending_promises.clear();
  67. }
  68. size_t ThreadEventQueue::process()
  69. {
  70. decltype(m_private->queued_events) events;
  71. {
  72. Threading::MutexLocker locker(m_private->mutex);
  73. events = move(m_private->queued_events);
  74. m_private->pending_promises.remove_all_matching([](auto& job) { return job->is_resolved() || job->is_rejected(); });
  75. }
  76. size_t processed_events = 0;
  77. for (size_t i = 0; i < events.size(); ++i) {
  78. auto& queued_event = events.at(i);
  79. auto receiver = queued_event.receiver.strong_ref();
  80. auto& event = *queued_event.event;
  81. if (!receiver) {
  82. switch (event.type()) {
  83. case Event::Quit:
  84. VERIFY_NOT_REACHED();
  85. default:
  86. // Receiver disappeared, drop the event on the floor.
  87. break;
  88. }
  89. } else if (event.type() == Event::Type::DeferredInvoke) {
  90. static_cast<DeferredInvocationEvent&>(event).m_invokee();
  91. } else {
  92. NonnullRefPtr<EventReceiver> protector(*receiver);
  93. receiver->dispatch_event(event);
  94. }
  95. ++processed_events;
  96. }
  97. {
  98. Threading::MutexLocker locker(m_private->mutex);
  99. if (m_private->pending_promises.size() > 30 && !m_private->warned_promise_count) {
  100. m_private->warned_promise_count = true;
  101. dbgln("ThreadEventQueue::process: Job queue wasn't designed for this load ({} promises)", m_private->pending_promises.size());
  102. }
  103. }
  104. return processed_events;
  105. }
  106. bool ThreadEventQueue::has_pending_events() const
  107. {
  108. Threading::MutexLocker locker(m_private->mutex);
  109. return !m_private->queued_events.is_empty();
  110. }
  111. }