DawnNative: implement Instance::ProcessEvents

Since https://dawn-review.googlesource.com/c/dawn/+/120940, callbacks
will be deferred to be executed in next device.APITick() instead of
immediately.
However, if the device is already destroyed (last ref dropped),
user/wire_server has no chance to call device.APITick() anymore, leading
to the callbacks waiting in queue forever.

This is also possibly the cause of memory leaks in cluserfuzz tests.

This CL attempt to fix it by implementing Instance::ProcessEvents():
In this method, every created device will invoke APITick() even if it is
already lost/externally released.

bug: chromium:1422507
bug: dawn:752
Change-Id: Iec69ad3b547a7e88c6e1a2225b13ad060a501a4f
Reviewed-on: https://dawn-review.googlesource.com/c/dawn/+/123420
Commit-Queue: Quyen Le <lehoangquyen@chromium.org>
Kokoro: Kokoro <noreply+kokoro@google.com>
Reviewed-by: Austin Eng <enga@chromium.org>

dawn.json

@@ -1530,8 +1530,7 @@
                 ]
             },
             {
-                "name": "process events",
-                "tags": ["upstream", "emscripten"]
+                "name": "process events"
             },
             {
                 "name": "request adapter",

include/dawn/native/DawnNative.h

@@ -187,6 +187,8 @@ class DAWN_NATIVE_EXPORT Instance {
 
     uint64_t GetDeviceCountForTesting() const;
 
+    bool ProcessEvents();
+
     // Returns the underlying WGPUInstance object.
     WGPUInstance Get() const;
 

src/dawn/fuzzers/DawnWireServerFuzzer.cpp

@@ -133,6 +133,10 @@ int DawnWireServerFuzzer::Run(const uint8_t* data,
     wireServer->InjectInstance(sInstance->Get(), 1, 0);
     wireServer->HandleCommands(reinterpret_cast<const char*>(data), size);
 
+    // Flush remaining callbacks to avoid memory leaks.
+    do {
+    } while (sInstance->ProcessEvents());
+
     // Note: Deleting the server will release all created objects.
     // Deleted devices will wait for idle on destruction.
     wireServer = nullptr;

src/dawn/native/DawnNative.cpp

@@ -251,6 +251,10 @@ uint64_t Instance::GetDeviceCountForTesting() const {
     return mImpl->GetDeviceCountForTesting();
 }
 
+bool Instance::ProcessEvents() {
+    return mImpl->APIProcessEvents();
+}
+
 WGPUInstance Instance::Get() const {
     return ToAPI(mImpl);
 }

src/dawn/native/Device.cpp

@@ -174,7 +174,6 @@ DeviceBase::DeviceBase(AdapterBase* adapter,
                        const DeviceDescriptor* descriptor,
                        const TogglesState& deviceToggles)
     : mAdapter(adapter), mToggles(deviceToggles), mNextPipelineCompatibilityToken(1) {
-    mAdapter->GetInstance()->IncrementDeviceCountForTesting();
     ASSERT(descriptor != nullptr);
 
     AdapterProperties adapterProperties;
@@ -218,8 +217,9 @@ DeviceBase::~DeviceBase() {
     // Queue does not get destroyed after the Device.
     mQueue = nullptr;
     // mAdapter is not set for mock test devices.
+    // TODO(crbug.com/dawn/1702): using a mock adapter could avoid the null checking.
     if (mAdapter != nullptr) {
-        mAdapter->GetInstance()->DecrementDeviceCountForTesting();
+        mAdapter->GetInstance()->RemoveDevice(this);
     }
 }
 
@@ -282,6 +282,11 @@ MaybeError DeviceBase::Initialize(Ref<QueueBase> defaultQueue) {
                         CreateShaderModule(&descriptor));
     }
 
+    // TODO(crbug.com/dawn/1702): using a mock adapter could avoid the null checking.
+    if (mAdapter != nullptr) {
+        mAdapter->GetInstance()->AddDevice(this);
+    }
+
     return {};
 }
 

src/dawn/native/Instance.cpp

@@ -21,6 +21,7 @@
 #include "dawn/common/Log.h"
 #include "dawn/common/SystemUtils.h"
 #include "dawn/native/ChainUtils_autogen.h"
+#include "dawn/native/Device.h"
 #include "dawn/native/ErrorData.h"
 #include "dawn/native/Surface.h"
 #include "dawn/native/Toggles.h"
@@ -504,15 +505,35 @@ BlobCache* InstanceBase::GetBlobCache(bool enabled) {
 }
 
 uint64_t InstanceBase::GetDeviceCountForTesting() const {
-    return mDeviceCountForTesting.load();
+    std::lock_guard<std::mutex> lg(mDevicesListMutex);
+    return mDevicesList.size();
 }
 
-void InstanceBase::IncrementDeviceCountForTesting() {
-    mDeviceCountForTesting++;
+void InstanceBase::AddDevice(DeviceBase* device) {
+    std::lock_guard<std::mutex> lg(mDevicesListMutex);
+    mDevicesList.insert(device);
 }
 
-void InstanceBase::DecrementDeviceCountForTesting() {
-    mDeviceCountForTesting--;
+void InstanceBase::RemoveDevice(DeviceBase* device) {
+    std::lock_guard<std::mutex> lg(mDevicesListMutex);
+    mDevicesList.erase(device);
+}
+
+bool InstanceBase::APIProcessEvents() {
+    std::vector<Ref<DeviceBase>> devices;
+    {
+        std::lock_guard<std::mutex> lg(mDevicesListMutex);
+        for (auto device : mDevicesList) {
+            devices.push_back(device);
+        }
+    }
+
+    bool hasMoreEvents = false;
+    for (auto device : devices) {
+        hasMoreEvents = device->APITick() || hasMoreEvents;
+    }
+
+    return hasMoreEvents;
 }
 
 const std::vector<std::string>& InstanceBase::GetRuntimeSearchPaths() const {

src/dawn/native/Instance.h

@@ -17,6 +17,8 @@
 
 #include <array>
 #include <memory>
+#include <mutex>
+#include <set>
 #include <string>
 #include <unordered_map>
 #include <vector>
@@ -37,6 +39,7 @@ class Platform;
 
 namespace dawn::native {
 
+class DeviceBase;
 class Surface;
 class XlibXcbFunctions;
 
@@ -103,8 +106,8 @@ class InstanceBase final : public RefCountedWithExternalCount {
     BlobCache* GetBlobCache(bool enabled = true);
 
     uint64_t GetDeviceCountForTesting() const;
-    void IncrementDeviceCountForTesting();
-    void DecrementDeviceCountForTesting();
+    void AddDevice(DeviceBase* device);
+    void RemoveDevice(DeviceBase* device);
 
     const std::vector<std::string>& GetRuntimeSearchPaths() const;
 
@@ -113,6 +116,7 @@ class InstanceBase final : public RefCountedWithExternalCount {
 
     // Dawn API
     Surface* APICreateSurface(const SurfaceDescriptor* descriptor);
+    bool APIProcessEvents();
 
   private:
     explicit InstanceBase(const TogglesState& instanceToggles);
@@ -161,7 +165,8 @@ class InstanceBase final : public RefCountedWithExternalCount {
     std::unique_ptr<XlibXcbFunctions> mXlibXcbFunctions;
 #endif  // defined(DAWN_USE_X11)
 
-    std::atomic_uint64_t mDeviceCountForTesting{0};
+    std::set<DeviceBase*> mDevicesList;
+    mutable std::mutex mDevicesListMutex;
 };
 
 }  // namespace dawn::native

src/dawn/tests/DawnTest.cpp

@@ -1439,12 +1439,12 @@ std::ostringstream& DawnTestBase::ExpectAttachmentDepthStencilTestData(
     return EXPECT_TEXTURE_EQ(colorData.data(), colorTexture, {0, 0}, {width, height});
 }
 
-void DawnTestBase::WaitABit(wgpu::Device targetDevice) {
-    if (targetDevice == nullptr) {
-        targetDevice = this->device;
+void DawnTestBase::WaitABit(wgpu::Instance targetInstance) {
+    if (targetInstance == nullptr) {
+        targetInstance = mInstance;
     }
-    if (targetDevice != nullptr) {
-        targetDevice.Tick();
+    if (targetInstance != nullptr) {
+        targetInstance.ProcessEvents();
     }
     FlushWire();
 
@@ -1498,8 +1498,6 @@ void DawnTestBase::MapSlotsSynchronously() {
     // immediately.
     mNumPendingMapOperations = mReadbackSlots.size();
 
-    std::vector<wgpu::Device> pendingDevices;
-
     // Map all readback slots
     for (size_t i = 0; i < mReadbackSlots.size(); ++i) {
         MapReadUserdata* userdata = new MapReadUserdata{this, i};
@@ -1507,15 +1505,11 @@ void DawnTestBase::MapSlotsSynchronously() {
         const ReadbackSlot& slot = mReadbackSlots[i];
         slot.buffer.MapAsync(wgpu::MapMode::Read, 0, wgpu::kWholeMapSize, SlotMapCallback,
                              userdata);
-
-        pendingDevices.push_back(slot.device);
     }
 
     // Busy wait until all map operations are done.
     while (mNumPendingMapOperations != 0) {
-        for (const wgpu::Device& device : pendingDevices) {
-            WaitABit(device);
-        }
+        WaitABit();
     }
 }
 

src/dawn/tests/DawnTest.h

@@ -550,7 +550,7 @@ class DawnTestBase {
                                                     mipLevel, {}, &expectedStencil);
     }
 
-    void WaitABit(wgpu::Device = nullptr);
+    void WaitABit(wgpu::Instance = nullptr);
     void FlushWire();
     void WaitForAllOperations();
 

src/dawn/tests/end2end/DeviceLifetimeTests.cpp

@@ -189,10 +189,7 @@ TEST_P(DeviceLifetimeTests, DroppedBeforeMappedAtCreationBuffer) {
 
 // Test that the device can be dropped before a buffer created from it, then mapping the buffer
 // fails.
-// TODO(crbug.com/dawn/752): Re-enable this test once we implement Instance.ProcessEvents().
-// Currently the callbacks are called inside Device.Tick() only. However, since we drop the device,
-// there is no way to call Device.Tick() anymore.
-TEST_P(DeviceLifetimeTests, DISABLED_DroppedThenMapBuffer) {
+TEST_P(DeviceLifetimeTests, DroppedThenMapBuffer) {
     wgpu::BufferDescriptor desc = {};
     desc.size = 4;
     desc.usage = wgpu::BufferUsage::MapRead | wgpu::BufferUsage::CopyDst;
@@ -481,6 +478,36 @@ TEST_P(DeviceLifetimeTests, DroppedInsideCreatePipelineAsyncRaceCache) {
     }
 }
 
+// Tests that dropping 2nd device inside 1st device's callback triggered by instance.ProcessEvents
+// won't crash.
+TEST_P(DeviceLifetimeTests, DropDevice2InProcessEvents) {
+    wgpu::Device device2 = CreateDevice();
+
+    struct UserData {
+        wgpu::Device device2;
+        bool done = false;
+    } userdata;
+
+    userdata.device2 = std::move(device2);
+
+    device.PushErrorScope(wgpu::ErrorFilter::Validation);
+
+    // The following callback will drop the 2nd device. It won't be triggered until
+    // instance.ProcessEvents() is called.
+    device.PopErrorScope(
+        [](WGPUErrorType type, const char*, void* userdataPtr) {
+            auto userdata = static_cast<UserData*>(userdataPtr);
+
+            userdata->device2 = nullptr;
+            userdata->done = true;
+        },
+        &userdata);
+
+    while (!userdata.done) {
+        WaitABit();
+    }
+}
+
 DAWN_INSTANTIATE_TEST(DeviceLifetimeTests,
                       D3D12Backend(),
                       MetalBackend(),

src/dawn/tests/unittests/wire/WireTest.cpp

@@ -150,5 +150,6 @@ void WireTest::DeleteClient() {
 }
 
 void WireTest::SetupIgnoredCallExpectations() {
+    EXPECT_CALL(api, InstanceProcessEvents(_)).Times(AnyNumber());
     EXPECT_CALL(api, DeviceTick(_)).Times(AnyNumber());
 }