Issue 5179 - Implement asynchronous executor with a controllable lifetime

src/Platform.cpp

 /*
 * This file is part of Adblock Plus <https://adblockplus.org/>,
 * Copyright (C) 2006-present eyeo GmbH
 *
 * Adblock Plus is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License version 3 as
 * published by the Free Software Foundation.
 *
 * Adblock Plus is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with Adblock Plus.  If not, see <http://www.gnu.org/licenses/>.
 */
 #include <AdblockPlus/Platform.h>
 #include <AdblockPlus/JsEngine.h>
 #include <AdblockPlus/FilterEngine.h>
 #include <AdblockPlus/DefaultLogSystem.h>
+#include <AdblockPlus/AsyncExecutor.h>
 #include "DefaultTimer.h"
 #include "DefaultWebRequest.h"
 #include "DefaultFileSystem.h"
 #include <stdexcept>
 
 using namespace AdblockPlus;
 
 namespace
 {
-  void DummyScheduler(const AdblockPlus::SchedulerTask& task)
-  {
-    std::thread(task).detach();
-  }
-
   template<typename T>
   void ValidatePlatformCreationParameter(const std::unique_ptr<T>& param, const char* paramName)
   {
     if (!param)
       throw std::logic_error(paramName + std::string(" must not be nullptr"));
   }
 }
 
 #define ASSIGN_PLATFORM_PARAM(param) ValidatePlatformCreationParameter(param = std::move(creationParameters.param), #param)
 
@@ skipping 69 matching lines @@
 }
 
 void Platform::WithLogSystem(const WithLogSystemCallback& callback)
 {
   if (logSystem && callback)
     callback(*logSystem);
 }
 
 namespace
 {
+  class SharedAsyncExecutor {

[sergei, 2018/02/23 11:16:42]

It's rather a compatibility thing to simplify the updating of libadblockplus in
libadblockplus-android which will already cause a lot of work (because they will
have to figure out what to do with infinitely running operations). In
libadblockplus-android we are currently using
DefaultPlatformBuilder::GetDefaultAsyncExecutor() to obtain a scheduler
(std::function) which is used to safely run asynchronous tasks related to
FilterEngine. In general it's not a public API and under the question to be
changed later.

When DefaultPlatform is being destroyed we invalidate that asynchronous executor
and ensure that no any task is running there, so whatever scheduled by
libadblocklpus-android via the function defaultScheduler will be either able to
normally (without access violations) finish because FilterEngine is not
destroyed yet or ignored at the scheduling stage, what is also safe.

See how it's used with `defaultScheduler` and that it's "invalidated" in the
destructor.

+  public:
+    SharedAsyncExecutor()
+      : executor(new AsyncExecutor())
+    {
+    }
+    void Dispatch(const std::function<void()>& task) {
+      std::lock_guard<std::mutex> lock(asyncExecutorMutex);
+      if (!executor)
+        return;
+      executor->Dispatch(task);
+    }
+    void Invalidate() {
+      std::unique_ptr<AsyncExecutor> tmp;

[sergei, 2018/02/23 11:16:42]

JIC, it's needed to prevent dead locks. If `executor` were destroyed directly
under the `lock` then it could happen that one of the tasks of the `executor` is
waiting for the lock on mutex in order to dispatch a next task and the
`executor` is waiting for the finishing of that task and is not releasing the
mutex.

+      {
+        std::lock_guard<std::mutex> lock(asyncExecutorMutex);
+        tmp = move(executor);
+      }
+    }
+  private:
+    std::mutex asyncExecutorMutex;
+    std::unique_ptr<AsyncExecutor> executor;
+  };
+
   class DefaultPlatform : public Platform
   {
   public:
-    typedef std::shared_ptr<Scheduler> AsyncExecutorPtr;
-    explicit DefaultPlatform(const AsyncExecutorPtr& asyncExecutor, CreationParameters&& creationParams)
+    typedef std::shared_ptr<SharedAsyncExecutor> AsyncExecutorPtr;
+    explicit DefaultPlatform(const std::shared_ptr<SharedAsyncExecutor>& asyncExecutor, CreationParameters&& creationParams)

[hub, 2018/02/23 22:59:35]

Since there is `typedef std::shared_ptr<SharedAsyncExecutor> AsyncExecutorPtr`
above, why isn't AsyncExecutorPtr used here. (and below line 160) ?

[sergei, 2018/03/01 11:18:37]

Fixed, I somehow forgot about it ))

       : Platform(std::move(creationParams)), asyncExecutor(asyncExecutor)
     {
     }
     ~DefaultPlatform();
 
     void WithTimer(const WithTimerCallback&) override;
     void WithFileSystem(const WithFileSystemCallback&) override;
     void WithWebRequest(const WithWebRequestCallback&) override;
     void WithLogSystem(const WithLogSystemCallback&) override;
 
   private:
-    AsyncExecutorPtr asyncExecutor;
+    std::shared_ptr<SharedAsyncExecutor> asyncExecutor;
     std::recursive_mutex interfacesMutex;
   };
 
   DefaultPlatform::~DefaultPlatform()
   {
-    asyncExecutor.reset();
+    asyncExecutor->Invalidate();
     LogSystemPtr tmpLogSystem;
     TimerPtr tmpTimer;
     FileSystemPtr tmpFileSystem;
     WebRequestPtr tmpWebRequest;
     {
       std::lock_guard<std::recursive_mutex> lock(interfacesMutex);
       tmpLogSystem = std::move(logSystem);
       tmpTimer = std::move(timer);
       tmpFileSystem = std::move(fileSystem);
       tmpWebRequest = std::move(webRequest);
@@ skipping 20 matching lines @@
 
   void DefaultPlatform::WithLogSystem(const WithLogSystemCallback& callback)
   {
     std::lock_guard<std::recursive_mutex> lock(interfacesMutex);
     Platform::WithLogSystem(callback);
   }
 }
 
 Scheduler DefaultPlatformBuilder::GetDefaultAsyncExecutor()
 {
-  if (!defaultScheduler)
-  {
-    asyncExecutor = std::make_shared<Scheduler>(::DummyScheduler);
-    std::weak_ptr<Scheduler> weakExecutor = asyncExecutor;
-    defaultScheduler = [weakExecutor](const SchedulerTask& task)
-    {
-      if (auto executor = weakExecutor.lock())
-      {
-        (*executor)(task);
-      }
-    };
-  }
   return defaultScheduler;
 }
 
 void DefaultPlatformBuilder::CreateDefaultTimer()
 {
   timer.reset(new DefaultTimer());
 }
 
 void DefaultPlatformBuilder::CreateDefaultFileSystem(const std::string& basePath)
 {
   fileSystem.reset(new DefaultFileSystem(GetDefaultAsyncExecutor(), std::unique_ptr<DefaultFileSystemSync>(new DefaultFileSystemSync(basePath))));
 }
 
 void DefaultPlatformBuilder::CreateDefaultWebRequest(std::unique_ptr<IWebRequestSync> webRequest)
 {
   if (!webRequest)
     webRequest.reset(new DefaultWebRequestSync());
   this->webRequest.reset(new DefaultWebRequest(GetDefaultAsyncExecutor(), std::move(webRequest)));
 }
 
 void DefaultPlatformBuilder::CreateDefaultLogSystem()
 {
   logSystem.reset(new DefaultLogSystem());
 }
 
 std::unique_ptr<Platform> DefaultPlatformBuilder::CreatePlatform()
 {
+  auto sharedAsyncExecutor = std::make_shared<SharedAsyncExecutor>();
+  defaultScheduler = [sharedAsyncExecutor](const SchedulerTask& task)
+  {
+    sharedAsyncExecutor->Dispatch(task);
+  };
   if (!logSystem)
     CreateDefaultLogSystem();
   if (!timer)
     CreateDefaultTimer();
   if (!fileSystem)
     CreateDefaultFileSystem();
   if (!webRequest)
     CreateDefaultWebRequest();
 
-  std::unique_ptr<Platform> platform(new DefaultPlatform(asyncExecutor, std::move(*this)));
-  asyncExecutor.reset();
+  std::unique_ptr<Platform> platform(new DefaultPlatform(sharedAsyncExecutor, std::move(*this)));
   return platform;
 }