A new synchronization primitive is introduced: AbortableTaskQueue, which allows to send work to the main thread from a background thread with the option to perform two-phase cancellation. Tests for AbortableTaskQueue are included. This new primitive has been used to overhaul GstBus messaging in AppendPipeline. A lot of code made redundant has been deleted in the process and lots of internal functions were now able to be made private.
Created attachment 353088 [details] Patch
Created attachment 353114 [details] Patch
Created attachment 353116 [details] Patch
Created attachment 353118 [details] Patch
Created attachment 353136 [details] Patch
Comment on attachment 353116 [details] Patch View in context: https://bugs.webkit.org/attachment.cgi?id=353116&action=review WIP review written by calvaris on my laptop: > Source/WebCore/platform/AbortableTaskQueue.h:68 > +class AbortableTaskQueue final { > + WTF_MAKE_NONCOPYABLE(AbortableTaskQueue); fast allocated? > Source/WebCore/platform/AbortableTaskQueue.h:72 > + ASSERT(WTF::isMainThread()); I think you can just call isMainThread() without prefixing the WTF. Please check in all places. > Source/WebCore/platform/AbortableTaskQueue.h:83 > + Too many capitals. > Source/WebCore/platform/AbortableTaskQueue.h:119 > + void tryEnqueueTask(std::function<void()> mainThreadHandler) WTF::Function, please check in all places > Source/WebCore/platform/AbortableTaskQueue.h:133 > + std::optional<R> tryEnqueueTaskAndWaitForReply(std::function<R()> mainThreadHandler) I think mainThreadHandler is kind of misleading. Shouldn't it be more like a task? > Source/WebCore/platform/AbortableTaskQueue.h:135 > + ASSERT(!WTF::isMainThread()); // don't deadlock main thread with itself 1) comment should start with uppercase and be in its own line IIRC 2) Whether we should allow main thread usage and run immediately the task > Source/WebCore/platform/AbortableTaskQueue.h:139 > + return std::optional<R>(); Shouldn't this be std::nullopt? > Source/WebCore/platform/AbortableTaskQueue.h:157 > + // This is class is provided for convenience when you want to use tryEnqueueTaskAndWaitForReply() but > + // you don't need any particular data from the main thread in return and just knowing that it finished > + // running the handler function is enough. > + class DummyResponse { }; void does not work, but Void could do, as when we use this we'll have to specify AsyncTaskQueue::Void. I think this gives a better idea, conceptually speaking. > Source/WebCore/platform/AbortableTaskQueue.h:164 > + class Task { > + WTF_MAKE_NONCOPYABLE(Task); fast allocated? > Source/WebCore/platform/AbortableTaskQueue.h:172 > + return m_taskQueue == nullptr; return m_taskQueue; > Source/WebCore/platform/AbortableTaskQueue.h:177 > + ASSERT(!isCancelled()); Is this really necessary? > Source/WebCore/platform/AbortableTaskQueue.h:215 > + while (!m_channel.empty()) { > + Task* task = m_channel.front(); > + task->cancel(); > + m_channel.pop(); > + } Could you use an iterator and then clear? For efficiency. > Source/WebCore/platform/AbortableTaskQueue.h:218 > + bool m_aborting = false; bool m_aborting { false }; > Source/WebCore/platform/AbortableTaskQueue.h:221 > + std::queue<Task*> m_channel; WTF::Deque > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:35 > +#include <functional> WTF::Function does not require this.
Comment on attachment 353136 [details] Patch View in context: https://bugs.webkit.org/attachment.cgi?id=353136&action=review You might need to rebase as well. Amazing patch Alicia, thanks for writing it, it is a very good refactor. Just some things need to be worked out and we will be good to go. > Source/WebCore/platform/AbortableTaskQueue.h:24 > +#include <functional> > +#include <memory> > +#include <queue> I don't think you need all these, do you? > Source/WebCore/platform/AbortableTaskQueue.h:31 > +#include <wtf/StdLibExtras.h> > +#include <wtf/ThreadSafeRefCounted.h> > +#include <wtf/Vector.h> > +#include <wtf/glib/GRefPtr.h> Ditto? > Source/WebCore/platform/AbortableTaskQueue.h:106 > + void stopAborting() I think I would call this finishAborting. Besides this, I see this is not used in the AppendPipeline code yet, it's for short term future patches. We should comment this in the changelog about this. > Source/WebCore/platform/AbortableTaskQueue.h:149 > + return m_aborting || response.has_value(); has_value -> operator bool in all places > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:137 > + g_signal_connect(m_bus.get(), "sync-message::need-context", G_CALLBACK(+[](GstBus*, GstMessage* message, AppendPipeline* appendPipeline) { > + appendPipeline->handleNeedContextSyncMessage(message); > + }), this); From what I see, this is not an integral part of this patch and it could be done in a separate one. I don't dislike it being here, it can fit, but please, add a comment to the changelog. > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:191 > + g_signal_connect(m_demux.get(), "pad-added", G_CALLBACK(+[](GstElement*, GstPad* demuxerSrcPad, AppendPipeline* appendPipeline) { > + appendPipeline->connectDemuxerSrcPadToAppsinkFromAnyThread(demuxerSrcPad); > + }), this); This *FromAnyThread functions could suffer a name change. Feel free to do it now or consider it for another patch. > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:203 > + ASSERT(!WTF::isMainThread()); This is redundant since it is checked in the method called bellow. > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:30 > +#include <thread> Not needed with WTF::Thread > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:40 > + bool testFinished = false; > + int currentStep = 0; You might want to use { } to initialize. > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:57 > + std::thread(backgroundThreadFunction).detach(); You might want to use WTF::Thread. > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:87 > + bool testFinished = false; > + int currentStep = 0; Ditto > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:96 > + FancyResponse ret(100); returnValue > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:167 > + bool testFinished = false; > + int currentStep = 0; Ditto > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:181 > + // Main thread has called startAborting() Period at the end. > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:187 > + // This call must return immediately because we are aborting Ditto > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:197 > + // Main thread has called stopAborting() Ditto
Comment on attachment 353136 [details] Patch View in context: https://bugs.webkit.org/attachment.cgi?id=353136&action=review Nice patch :) Initial drive-by review. > Source/WebCore/platform/AbortableTaskQueue.h:36 > + * thread(s) that need to send requests to the main thread. Nit, but does this rephrasing capture the intention, it seems a little more to the point: "AbortableTaskQueue is a high-level synchronization object to manage abortable tasks posted from background threads", since it has little to do with processes from what I can tell. > Source/WebCore/platform/AbortableTaskQueue.h:38 > + * The requests sent by the background thread(s) may be asynchronous, using tryEnqueueTask(), which returns The tasks .. > Source/WebCore/platform/AbortableTaskQueue.h:68 > + WTF_MAKE_NONCOPYABLE(AbortableTaskQueue); We may as well WTF_MAKE_FAST_ALLOCATED; > Source/WebCore/platform/AbortableTaskQueue.h:119 > + void tryEnqueueTask(std::function<void()> mainThreadHandler) You'll probably want a WTF::Function instead of a std::function here and elsewhere in the patch. It has two advantages, 1) by-value captures are not copied and 2) it uses the fast allocator rather than system malloc. Point 1) can cause some unpleasant thread-safety bugs. I see you by-value capture append pipeline variables in the usage code, which could trip this up if I've understood correctly. I don't think you rely on the copying features of std::function for this classes' intended use-case, so it seems a good change to switch to WTF::Function and move the tasks into the task queue. > Source/WebCore/platform/AbortableTaskQueue.h:163 > + class Task { Perhaps this should be thread safe ref-counted to avoid using unique_ptr. It might even be nicer to use WTF::SharedTask's somehow. > Source/WebCore/platform/AbortableTaskQueue.h:172 > + return m_taskQueue == nullptr; return !m_taskQueue > Source/WebCore/platform/AbortableTaskQueue.h:221 > + std::queue<Task*> m_channel; I'd use a WTF queue here. > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:43 > + auto backgroundThreadFunction = [&testFinished, &taskQueue, ¤tStep]() { You may as well use [&] { ... } here rather than listing the closure capture individually. Ditto for other places in this patch. > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:50 > + g_assert(WTF::isMainThread()); Why g_assert now and not EXPECT_TRUE? > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:91 > + EXPECT_FALSE(WTF::isMainThread()); Perhaps this check should be in the above test as well. > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:101 > + EXPECT_TRUE(reply.has_value()); EXPECT_TRUE(reply) > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:193 > + EXPECT_FALSE(response.has_value()); EXPECT_FALSE(response) > Tools/TestWebKitAPI/Tests/WebCore/AbortableTaskQueue.cpp:219 > +} Another test that might be nice is if a background thread posts a task that never returns using the synchronous api before a call to start aborting, to make sure the early return in tryEnqueueTaskAndWaitForReply on m_aborting works correctly. IIUC, that case isn't covered so far, but was the one that initially concerned me.
I've found this patch regresses media/media-source/media-source-stpp-crash.html -- it deadlocked when parsing a file with an unsupported track. That stall was caused by dangling usages of the padAdded lock/condition pair that is not needed anymore with ATQ. These are the changes to fix it, in case someone wants to look at them in isolation: diff --git a/Source/WebCore/platform/AbortableTaskQueue.h b/Source/WebCore/platform/AbortableTaskQueue.h index f1741db2d80..be89c7720a7 100644 --- a/Source/WebCore/platform/AbortableTaskQueue.h +++ b/Source/WebCore/platform/AbortableTaskQueue.h @@ -87,13 +87,14 @@ public: // // Until stopAborting is called, all present and future calls to postRequestAndWait() will immediately // return an empty optional. + // + // This method is idempotent. void startAborting() { ASSERT(WTF::isMainThread()); { LockHolder lockHolder(m_mutex); - ASSERT(!m_aborting); m_aborting = true; cancelAllTasks(); } diff --git a/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp b/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp index 4ac0a2ffd65..b34cb70992d 100644 --- a/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp +++ b/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp @@ -226,12 +226,6 @@ AppendPipeline::~AppendPipeline() // Forget all pending tasks and unblock the streaming thread if it was blocked. m_taskQueue.startAborting(); - { - LockHolder locker(m_padAddRemoveLock); - m_playerPrivate = nullptr; - m_padAddRemoveCondition.notifyOne(); - } - GST_TRACE("Destroying AppendPipeline (%p)", this); // FIXME: Maybe notify appendComplete here? @@ -308,12 +302,7 @@ void AppendPipeline::clearPlayerPrivate() // Make sure that AppendPipeline won't process more data from now on and // instruct handleNewSample to abort itself from now on as well. setAppendState(AppendState::Invalid); - - { - LockHolder locker(m_padAddRemoveLock); - m_playerPrivate = nullptr; - m_padAddRemoveCondition.notifyOne(); - } + m_taskQueue.startAborting(); // And now that no handleNewSample operations will remain stalled waiting // for the main thread, stop the pipeline. @@ -935,7 +924,6 @@ void AppendPipeline::connectDemuxerSrcPadToAppsink(GstPad* demuxerSrcPad) if (type.endsWith("webm")) gst_pad_add_probe(demuxerSrcPad, GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM, matroskademuxForceSegmentStartToEqualZero, nullptr, nullptr); - LockHolder locker(m_padAddRemoveLock); GRefPtr<GstPad> sinkSinkPad = adoptGRef(gst_element_get_static_pad(m_appsink.get(), "sink")); // Only one stream per demuxer is supported. @@ -944,7 +932,6 @@ void AppendPipeline::connectDemuxerSrcPadToAppsink(GstPad* demuxerSrcPad) GRefPtr<GstCaps> caps = adoptGRef(gst_pad_get_current_caps(GST_PAD(demuxerSrcPad))); if (!caps || m_appendState == AppendState::Invalid || !m_playerPrivate) { - m_padAddRemoveCondition.notifyOne(); return; } @@ -980,8 +967,6 @@ void AppendPipeline::connectDemuxerSrcPadToAppsink(GstPad* demuxerSrcPad) // This is going to cause an error which will detach the SourceBuffer and tear down this // AppendPipeline, so we need the padAddRemove lock released before continuing. m_track = nullptr; - m_padAddRemoveCondition.notifyOne(); - locker.unlockEarly(); didReceiveInitializationSegment(); return; default: @@ -992,8 +977,6 @@ void AppendPipeline::connectDemuxerSrcPadToAppsink(GstPad* demuxerSrcPad) m_appsinkCaps = WTFMove(caps); if (m_playerPrivate) m_playerPrivate->trackDetected(this, m_track, true); - - m_padAddRemoveCondition.notifyOne(); } void AppendPipeline::disconnectDemuxerSrcPadFromAppsinkFromAnyThread(GstPad*) diff --git a/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.h b/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.h index 0546a50c675..e27aada76ab 100644 --- a/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.h +++ b/Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.h @@ -133,9 +133,6 @@ private: // queue, instead of it growing unbounded. std::atomic_flag m_wasBusAlreadyNotifiedOfAvailableSamples; - Lock m_padAddRemoveLock; - Condition m_padAddRemoveCondition; - GRefPtr<GstCaps> m_appsinkCaps; GRefPtr<GstCaps> m_demuxerSrcPadCaps; FloatSize m_presentationSize;
Created attachment 353533 [details] Patch
Comment on attachment 353533 [details] Patch Note to self: never again `webkit-patch upload -g` anything else than HEAD. It's completely broken.
Alicia, I find a bit confusing because no patch is marked for review. Should I review what?
(In reply to Xabier Rodríguez Calvar from comment #12) > Alicia, I find a bit confusing because no patch is marked for review. Should > I review what? There is no new patch (yet). I used webkit-patch to upload a completely unrelated patch and because it's buggy, it uploaded it here, obsoleting the previous (ATQ related) patch, so I had to un-obsolete it.
(In reply to Alicia Boya García from comment #6) > Comment on attachment 353116 [details] > Patch > > View in context: > https://bugs.webkit.org/attachment.cgi?id=353116&action=review > > WIP review written by calvaris on my laptop: > > > Source/WebCore/platform/AbortableTaskQueue.h:68 > > +class AbortableTaskQueue final { > > + WTF_MAKE_NONCOPYABLE(AbortableTaskQueue); > > fast allocated? Not necessary, since AbortableTaskQueue is not constructed with `new`, but rather as a member of the class that makes use of it. > > Source/WebCore/platform/AbortableTaskQueue.h:164 > > + class Task { > > + WTF_MAKE_NONCOPYABLE(Task); > > fast allocated? Here it is actually useful. > > Source/WebCore/platform/AbortableTaskQueue.h:177 > > + ASSERT(!isCancelled()); > > Is this really necessary? It checks the task is not cancelled twice, which should not happen... but that's what asserts are for, crashing in Debug when things that should not happen, happen. (In reply to Xabier Rodríguez Calvar from comment #7) > > Source/WebCore/platform/AbortableTaskQueue.h:31 > > +#include <wtf/StdLibExtras.h> This one is needed for std::optional pre-C++17 > > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:191 > > + g_signal_connect(m_demux.get(), "pad-added", G_CALLBACK(+[](GstElement*, GstPad* demuxerSrcPad, AppendPipeline* appendPipeline) { > > + appendPipeline->connectDemuxerSrcPadToAppsinkFromAnyThread(demuxerSrcPad); > > + }), this); > > This *FromAnyThread functions could suffer a name change. Feel free to do it > now or consider it for another patch. Done. The remaining *FromAnyThread function can actually be called from both threads. (In reply to Charlie Turner from comment #8) > Comment on attachment 353136 [details] > Patch > > View in context: > https://bugs.webkit.org/attachment.cgi?id=353136&action=review > > Nice patch :) Initial drive-by review. > > > Source/WebCore/platform/AbortableTaskQueue.h:36 > > + * thread(s) that need to send requests to the main thread. > > Nit, but does this rephrasing capture the intention, it seems a little more > to the point: "AbortableTaskQueue is a high-level synchronization object to > manage abortable tasks posted from background threads", since it has little > to do with processes from what I can tell. There "process" means "work, sequence of steps to accomplish something", not "Operating system process". I've made changes in the phrasing to hopefully make it clearer. > Another test that might be nice is if a background thread posts a task that > never returns using the synchronous api before a call to start aborting, to > make sure the early return in tryEnqueueTaskAndWaitForReply on m_aborting > works correctly. IIUC, that case isn't covered so far, but was the one that > initially concerned me. I have added such a test. Unfortunately, this one cannot be scheduled exactly because we need to wait for the background thread to be suspended and there is no easy cross-platform way to do that in an accurate manner... A small sleep is used instead.
Created attachment 354478 [details] Patch
Comment on attachment 354478 [details] Patch View in context: https://bugs.webkit.org/attachment.cgi?id=354478&action=review > Source/WebCore/platform/AbortableTaskQueue.h:122 > + void tryEnqueueTask(WTF::Function<void()>&& mainThreadTaskHandler) I would just name this enqueueTask. I don't think we need the try here, it is obvious because of the nature of this ATQ that we'll fail if it's aborting. > Source/WebCore/platform/AbortableTaskQueue.h:124 > + LockHolder lockHolder(m_mutex); I'd suggest to ASSERT(!isMainThread()); before this as well. > Source/WebCore/platform/AbortableTaskQueue.h:136 > + std::optional<R> tryEnqueueTaskAndWaitForReply(WTF::Function<R()>&& mainThreadTaskHandler) Similar to what I commented above, I would call this enqueueTaskAndWait. I think we neither need the try nor the ForReply. > Source/WebCore/platform/AbortableTaskQueue.h:191 > + if (!isCancelled()) { We better bail out here if it'd cancelled. > Source/WebCore/platform/AbortableTaskQueue.h:229 > + WTF::Deque<Ref<Task>> m_channel; I guess what made you switch to Ref was actually switching to Deque, right? > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:140 > + g_signal_connect(m_bus.get(), "sync-message::need-context", G_CALLBACK(+[](GstBus*, GstMessage* message, AppendPipeline* appendPipeline) { > + appendPipeline->handleNeedContextSyncMessage(message); > + }), this); > + g_signal_connect(m_bus.get(), "message::state-changed", G_CALLBACK(+[](GstBus*, GstMessage* message, AppendPipeline* appendPipeline) { > + appendPipeline->handleStateChangeMessage(message); > + }), this); I still see no references to this on the changelog. > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:333 > - ASSERT(WTF::isMainThread()); > + ASSERT(isMainThread()); Please, comment about this in the Changelog as well. > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:872 > + if (!response.has_value()) { if (!response) { ... } > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.h:60 > +private: You made lost of methods private, please comment it on the changelog.
(In reply to Xabier Rodríguez Calvar from comment #16) > > Source/WebCore/platform/AbortableTaskQueue.h:229 > > + WTF::Deque<Ref<Task>> m_channel; > > I guess what made you switch to Ref was actually switching to Deque, right? No, it was Charlie preferring ThreadSafeRefCounted over unique_ptr. > > > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:140 > > + g_signal_connect(m_bus.get(), "sync-message::need-context", G_CALLBACK(+[](GstBus*, GstMessage* message, AppendPipeline* appendPipeline) { > > + appendPipeline->handleNeedContextSyncMessage(message); > > + }), this); > > + g_signal_connect(m_bus.get(), "message::state-changed", G_CALLBACK(+[](GstBus*, GstMessage* message, AppendPipeline* appendPipeline) { > > + appendPipeline->handleStateChangeMessage(message); > > + }), this); > > I still see no references to this on the changelog. Emphasis in caps: This new primitive has been used to overhaul GstBus messaging in AppendPipeline. A lot of code made redundant has been deleted in the process and lots of internal functions were now able to be made private. AS PART OF THE REFACTOR ALL GLIB SIGNALS IN APPENDPIPELINE NOW USE LAMBDAS. > > > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.cpp:333 > > - ASSERT(WTF::isMainThread()); > > + ASSERT(isMainThread()); > > Please, comment about this in the Changelog as well. OK. > > > Source/WebCore/platform/graphics/gstreamer/mse/AppendPipeline.h:60 > > +private: > > You made lost of methods private, please comment it on the changelog. Emphasis in caps: This new primitive has been used to overhaul GstBus messaging in AppendPipeline. A lot of code made redundant has been deleted in the process and LOTS OF INTERNAL FUNCTIONS WERE NOW ABLE TO BE MADE PRIVATE. As part of the refactor all glib signals in AppendPipeline now use lambdas.
Created attachment 354552 [details] Patch for landing
Comment on attachment 354552 [details] Patch for landing Clearing flags on attachment: 354552 Committed r238084: <https://trac.webkit.org/changeset/238084>
All reviewed patches have been landed. Closing bug.
<rdar://problem/45988278>