Changeset 284671 in webkit

Timestamp:

Oct 22, 2021, 1:18:15 AM (5 years ago)

Author:

commit-queue@webkit.org

Message:

StreamConnectionWorkQueue is not thread-safe
​https://bugs.webkit.org/show_bug.cgi?id=232068

Patch by Kimmo Kinnunen <​kkinnunen@apple.com> on 2021-10-22
Reviewed by Wenson Hsieh.

Make StreamConnectionWorkQueue thread-safe so that
addStreamConnection works from multiple threads.
In the future, this will be called when a new RemoteDisplayListRecorder is
added during the StreamConnectionWorkQueue message processing
invocation for RemoteRenderingBackend::createRemoteImageBuffer.

• Platform/IPC/StreamConnectionWorkQueue.cpp:

(IPC::StreamConnectionWorkQueue::~StreamConnectionWorkQueue):
(IPC::StreamConnectionWorkQueue::dispatch):
(IPC::StreamConnectionWorkQueue::addStreamConnection):
(IPC::StreamConnectionWorkQueue::removeStreamConnection):
Change semantics so that adding and removing connections during
shutdown is ok. This is required for simpler code since
during stop we must dispatch all pending messages, but during
message code we might have unconditional start listening when
the message creates a new RemoteDisplayListRecorder.
As a consequence it's not an error to add a connection,
stop and destroy the work queue without removing it.
(IPC::StreamConnectionWorkQueue::stop):
(IPC::StreamConnectionWorkQueue::startProcessingThread):
(IPC::StreamConnectionWorkQueue::wakeUpProcessingThread): Deleted.

• Platform/IPC/StreamConnectionWorkQueue.h:

Location:

trunk/Source/WebKit

Files:

• ChangeLog (modified) (1 diff)
• Platform/IPC/StreamConnectionWorkQueue.cpp (modified) (4 diffs)
• Platform/IPC/StreamConnectionWorkQueue.h (modified) (3 diffs)

trunk/Source/WebKit/ChangeLog

@@ +1 @@
+2021-10-22  Kimmo Kinnunen  <kkinnunen@apple.com>
+
+        StreamConnectionWorkQueue is not thread-safe
+        https://bugs.webkit.org/show_bug.cgi?id=232068
+
+        Reviewed by Wenson Hsieh.
+
+        Make StreamConnectionWorkQueue thread-safe so that
+        addStreamConnection works from multiple threads.
+        In the future, this will be called when a new RemoteDisplayListRecorder is
+        added during the StreamConnectionWorkQueue message processing
+        invocation for RemoteRenderingBackend::createRemoteImageBuffer.
+
+        * Platform/IPC/StreamConnectionWorkQueue.cpp:
+        (IPC::StreamConnectionWorkQueue::~StreamConnectionWorkQueue):
+        (IPC::StreamConnectionWorkQueue::dispatch):
+        (IPC::StreamConnectionWorkQueue::addStreamConnection):
+        (IPC::StreamConnectionWorkQueue::removeStreamConnection):
+        Change semantics so that adding and removing connections during
+        shutdown is ok. This is required for simpler code since
+        during stop we must dispatch all pending messages, but during
+        message code we might have unconditional start listening when
+        the message creates a new RemoteDisplayListRecorder.
+        As a consequence it's not an error to add a connection,
+        stop and destroy the work queue without removing it.
+        (IPC::StreamConnectionWorkQueue::stop):
+        (IPC::StreamConnectionWorkQueue::startProcessingThread):
+        (IPC::StreamConnectionWorkQueue::wakeUpProcessingThread): Deleted.
+        * Platform/IPC/StreamConnectionWorkQueue.h:
+
 2021-10-21  Alex Christensen  <achristensen@webkit.org>
 

trunk/Source/WebKit/Platform/IPC/StreamConnectionWorkQueue.cpp

@@ -34 +34 @@
 }
 
+StreamConnectionWorkQueue::~StreamConnectionWorkQueue()
+{
+    // `StreamConnectionWorkQueue::stop()` should be called if anything has been dispatched or listened to.
+    ASSERT(!m_processingThread);
+}
+
 void StreamConnectionWorkQueue::dispatch(WTF::Function<void()>&& function)
 {
+    ASSERT(!m_shouldQuit); // Re-entering during shutdown not supported.
     {
         Locker locker { m_lock };
         m_functions.append(WTFMove(function));
-        ASSERT(!m_shouldQuit); // Re-entering during shutdown not supported.
+        if (!m_shouldQuit && !m_processingThread) {
+            startProcessingThread();
+            return;
+        }
     }
-    wakeUpProcessingThread();
+    wakeUp();
+}
 
-}
 void StreamConnectionWorkQueue::addStreamConnection(StreamServerConnectionBase& connection)
 {
@@ -49 +59 @@
         Locker locker { m_lock };
         m_connections.add(connection);
-        ASSERT(!m_shouldQuit); // Re-entering during shutdown not supported.
+        if (!m_shouldQuit && !m_processingThread) {
+            startProcessingThread();
+            return;
+        }
     }
-    wakeUpProcessingThread();
+    wakeUp();
 }
 
 void StreamConnectionWorkQueue::removeStreamConnection(StreamServerConnectionBase& connection)
 {
-    ASSERT(m_processingThread);
     {
         Locker locker { m_lock };
         m_connections.remove(connection);
-        ASSERT(!m_shouldQuit); // Re-entering during shutdown not supported.
     }
-    m_wakeUpSemaphore.signal();
+    wakeUp();
 }
 
@@ -68 +79 @@
 {
     m_shouldQuit = true;
-    if (!m_processingThread)
+    RefPtr<Thread> processingThread;
+    {
+        Locker locker { m_lock };
+        processingThread = WTFMove(m_processingThread);
+    }
+    if (!processingThread)
         return;
-    m_wakeUpSemaphore.signal();
-    m_processingThread->waitForCompletion();
-    m_processingThread = nullptr;
+    ASSERT(Thread::current().uid() != processingThread->uid());
+    wakeUp();
+    processingThread->waitForCompletion();
 }
 
@@ -85 +101 @@
 }
 
-void StreamConnectionWorkQueue::wakeUpProcessingThread()
+void StreamConnectionWorkQueue::startProcessingThread()
 {
-    if (m_processingThread) {
-        m_wakeUpSemaphore.signal();
-        return;
-    }
-
     auto task = [this]() mutable {
         for (;;) {

trunk/Source/WebKit/Platform/IPC/StreamConnectionWorkQueue.h

@@ -45 +45 @@
 
     StreamConnectionWorkQueue(const char*);
-    ~StreamConnectionWorkQueue() = default;
+    ~StreamConnectionWorkQueue();
     void addStreamConnection(StreamServerConnectionBase&);
     void removeStreamConnection(StreamServerConnectionBase&);
@@ -56 +56 @@
     Semaphore& wakeUpSemaphore();
 private:
-    void wakeUpProcessingThread();
+    void startProcessingThread() WTF_REQUIRES_LOCK(m_lock);
     void processStreams();
 
@@ -62 +62 @@
 
     Semaphore m_wakeUpSemaphore;
-    RefPtr<Thread> m_processingThread;
-
     std::atomic<bool> m_shouldQuit { false };
 
     Lock m_lock;
+    RefPtr<Thread> m_processingThread WTF_GUARDED_BY_LOCK(m_lock);
     Deque<Function<void()>> m_functions WTF_GUARDED_BY_LOCK(m_lock);
     HashSet<Ref<StreamServerConnectionBase>> m_connections WTF_GUARDED_BY_LOCK(m_lock);