Worker: Introduce per-global-scope task scheduler

third_party/WebKit/Source/core/workers/WorkerThread.cpp

 /*
  * Copyright (C) 2008 Apple Inc. All Rights Reserved.
  *
  * Redistribution and use in source and binary forms, with or without
  * modification, are permitted provided that the following conditions
  * are met:
  * 1. Redistributions of source code must retain the above copyright
  *    notice, this list of conditions and the following disclaimer.
  * 2. Redistributions in binary form must reproduce the above copyright
  *    notice, this list of conditions and the following disclaimer in the
@@ skipping 13 matching lines @@
  *
  */
 
 #include "core/workers/WorkerThread.h"
 
 #include <limits.h>
 #include <memory>
 #include "bindings/core/v8/Microtask.h"
 #include "bindings/core/v8/ScriptSourceCode.h"
 #include "bindings/core/v8/WorkerOrWorkletScriptController.h"
+#include "core/dom/TaskRunnerHelper.h"
 #include "core/inspector/ConsoleMessageStorage.h"
 #include "core/inspector/InspectorTaskRunner.h"
 #include "core/inspector/WorkerInspectorController.h"
 #include "core/inspector/WorkerThreadDebugger.h"
 #include "core/origin_trials/OriginTrialContext.h"
 #include "core/probe/CoreProbes.h"
 #include "core/workers/ThreadedWorkletGlobalScope.h"
 #include "core/workers/WorkerBackingThread.h"
 #include "core/workers/WorkerClients.h"
 #include "core/workers/WorkerGlobalScope.h"
 #include "core/workers/WorkerReportingProxy.h"
 #include "core/workers/WorkerThreadStartupData.h"
 #include "platform/CrossThreadFunctional.h"
 #include "platform/Histogram.h"
 #include "platform/WaitableEvent.h"
 #include "platform/WebThreadSupportingGC.h"
 #include "platform/heap/SafePoint.h"
 #include "platform/heap/ThreadState.h"
 #include "platform/weborigin/KURL.h"
+#include "public/platform/scheduler/child/webthread_impl_for_worker_scheduler.h"
 #include "wtf/Functional.h"
 #include "wtf/Noncopyable.h"
 #include "wtf/PtrUtil.h"
 #include "wtf/Threading.h"
 #include "wtf/text/WTFString.h"
 
 namespace blink {
 
 using ExitCode = WorkerThread::ExitCode;
 
@@ skipping 37 matching lines @@
   DEFINE_THREAD_SAFE_STATIC_LOCAL(
       EnumerationHistogram, exitCodeHistogram,
       new EnumerationHistogram("WorkerThread.ExitCode",
                                static_cast<int>(ExitCode::LastEnum)));
   exitCodeHistogram.count(static_cast<int>(m_exitCode));
 }
 
 void WorkerThread::start(std::unique_ptr<WorkerThreadStartupData> startupData,
                          ParentFrameTaskRunners* parentFrameTaskRunners) {
   DCHECK(isMainThread());
-
   if (m_requestedToStart)
     return;
 
   m_requestedToStart = true;
   m_parentFrameTaskRunners = parentFrameTaskRunners;
-  workerBackingThread().backingThread().postTask(
+  m_threadControlTaskRunner =
+      workerBackingThread().backingThread().platformThread().getWebTaskRunner();
+
+  // Synchronously initialize the per-global-scope scheduler to prevent someone
+  // from posting a task to the thread before the scheduler is ready.
+  WaitableEvent waitableEvent;
+  m_threadControlTaskRunner->postTask(
+      BLINK_FROM_HERE,
+      crossThreadBind(&WorkerThread::initializeSchedulerOnWorkerThread,
+                      crossThreadUnretained(this),
+                      crossThreadUnretained(&waitableEvent)));

[kinuko, 2017/04/10 06:30:06]

Is it the requirement of NewTaskQueue implementation that this needs to run on
the new worker thread? (Looks like yes)

[Sami, 2017/04/10 16:58:10]

Yes, that's the case.

+  waitableEvent.wait();
+
+  m_threadControlTaskRunner->postTask(
       BLINK_FROM_HERE, crossThreadBind(&WorkerThread::initializeOnWorkerThread,
                                        crossThreadUnretained(this),
                                        WTF::passed(std::move(startupData))));
 }
 
 void WorkerThread::terminate() {
   DCHECK(isMainThread());
   terminateInternal(TerminationMode::Graceful);
 }
 
@@ skipping 56 matching lines @@
 }
 
 v8::Isolate* WorkerThread::isolate() {
   return workerBackingThread().isolate();
 }
 
 bool WorkerThread::isCurrentThread() {
   return workerBackingThread().backingThread().isCurrentThread();
 }
 
-void WorkerThread::postTask(const WebTraceLocation& location,
-                            std::unique_ptr<WTF::Closure> task) {
-  DCHECK(isCurrentThread());
-  if (isInShutdown())
-    return;
-  workerBackingThread().backingThread().postTask(
-      location,
-      WTF::bind(
-          &WorkerThread::performTaskOnWorkerThread<WTF::SameThreadAffinity>,
-          WTF::unretained(this), WTF::passed(std::move(task))));
-}
-
-void WorkerThread::postTask(const WebTraceLocation& location,
-                            std::unique_ptr<WTF::CrossThreadClosure> task) {
-  if (isInShutdown())
-    return;
-  workerBackingThread().backingThread().postTask(
-      location,
-      crossThreadBind(
-          &WorkerThread::performTaskOnWorkerThread<WTF::CrossThreadAffinity>,
-          crossThreadUnretained(this), WTF::passed(std::move(task))));
-}
-
 void WorkerThread::appendDebuggerTask(
     std::unique_ptr<CrossThreadClosure> task) {
   DCHECK(isMainThread());
-  if (isInShutdown())
+  if (m_requestedToTerminate)
     return;
   m_inspectorTaskRunner->appendTask(crossThreadBind(
       &WorkerThread::performDebuggerTaskOnWorkerThread,
       crossThreadUnretained(this), WTF::passed(std::move(task))));
   {
     MutexLocker lock(m_threadStateMutex);
     if (isolate() && m_threadState != ThreadState::ReadyToShutdown)
       m_inspectorTaskRunner->interruptAndRunAllTasksDontWait(isolate());
   }
-  workerBackingThread().backingThread().postTask(
-      BLINK_FROM_HERE,
-      crossThreadBind(&WorkerThread::performDebuggerTaskDontWaitOnWorkerThread,
-                      crossThreadUnretained(this)));
+  TaskRunnerHelper::get(TaskType::Unthrottled, this)
+      ->postTask(BLINK_FROM_HERE,
+                 crossThreadBind(
+                     &WorkerThread::performDebuggerTaskDontWaitOnWorkerThread,
+                     crossThreadUnretained(this)));
 }
 
 void WorkerThread::startRunningDebuggerTasksOnPauseOnWorkerThread() {
   DCHECK(isCurrentThread());
   if (m_workerInspectorController)
     m_workerInspectorController->flushProtocolNotifications();
   m_pausedInDebugger = true;
   ThreadDebugger::idleStarted(isolate());
   std::unique_ptr<CrossThreadClosure> task;
   do {
@@ skipping 118 matching lines @@
                                 WTF::unretained(this)),
                       m_forcibleTerminationDelayInMs);
           break;
       }
     }
   }
 
   m_workerThreadLifecycleContext->notifyContextDestroyed();
   m_inspectorTaskRunner->kill();
 
-  workerBackingThread().backingThread().postTask(
+  m_threadControlTaskRunner->postTask(
       BLINK_FROM_HERE,
       crossThreadBind(&WorkerThread::prepareForShutdownOnWorkerThread,
                       crossThreadUnretained(this)));
-  workerBackingThread().backingThread().postTask(
+  m_threadControlTaskRunner->postTask(
       BLINK_FROM_HERE,
       crossThreadBind(&WorkerThread::performShutdownOnWorkerThread,
                       crossThreadUnretained(this)));
 }
 
 bool WorkerThread::shouldScheduleToTerminateExecution(const MutexLocker& lock) {
   DCHECK(isMainThread());
   DCHECK(isThreadStateMutexLocked(lock));
 
   switch (m_threadState) {
@@ skipping 37 matching lines @@
   DCHECK(isThreadStateMutexLocked(lock));
 
   DCHECK(exitCode == ExitCode::SyncForciblyTerminated ||
          exitCode == ExitCode::AsyncForciblyTerminated);
   setExitCode(lock, exitCode);
 
   isolate()->TerminateExecution();
   m_forcibleTerminationTaskHandle.cancel();
 }
 
-bool WorkerThread::isInShutdown() {
-  // Check if we've started termination or shutdown sequence. Avoid acquiring
-  // a lock here to avoid introducing a risk of deadlock. Note that accessing
-  // |m_requestedToTerminate| on the main thread or |m_threadState| on the
-  // worker thread is safe as the flag is set only on the thread.
-  if (isMainThread() && m_requestedToTerminate)
-    return true;
-  if (isCurrentThread() && m_threadState == ThreadState::ReadyToShutdown)
-    return true;
-  return false;
+void WorkerThread::initializeSchedulerOnWorkerThread(
+    WaitableEvent* waitableEvent) {
+  DCHECK(isCurrentThread());
+  scheduler::WebThreadImplForWorkerScheduler& workerScheduler =
+      static_cast<scheduler::WebThreadImplForWorkerScheduler&>(
+          workerBackingThread().backingThread().platformThread());
+  m_globalScopeScheduler = WTF::makeUnique<scheduler::GlobalScopeScheduler>(
+      workerScheduler.getWorkerScheduler());
+  waitableEvent->signal();
 }
 
 void WorkerThread::initializeOnWorkerThread(
     std::unique_ptr<WorkerThreadStartupData> startupData) {
   DCHECK(isCurrentThread());
   DCHECK_EQ(ThreadState::NotStarted, m_threadState);
 
   KURL scriptURL = startupData->m_scriptURL;
   String sourceCode = startupData->m_sourceCode;
   WorkerThreadStartMode startMode = startupData->m_startMode;
@@ skipping 77 matching lines @@
   probe::allAsyncTasksCanceled(globalScope());
 
   globalScope()->notifyContextDestroyed();
   if (m_workerInspectorController) {
     m_workerInspectorController->dispose();
     m_workerInspectorController.clear();
   }
   globalScope()->dispose();
   m_consoleMessageStorage.clear();
   workerBackingThread().backingThread().removeTaskObserver(this);
+
+  // Cancel all queued tasks.
+  m_globalScopeScheduler->dispose();
 }
 
 void WorkerThread::performShutdownOnWorkerThread() {
   DCHECK(isCurrentThread());
   DCHECK(checkRequestedToTerminateOnWorkerThread());
   DCHECK_EQ(ThreadState::ReadyToShutdown, m_threadState);
 
   // The below assignment will destroy the context, which will in turn notify
   // messaging proxy. We cannot let any objects survive past thread exit,
   // because no other thread will run GC or otherwise destroy them. If Oilpan
   // is enabled, we detach of the context/global scope, with the final heap
   // cleanup below sweeping it out.
   m_globalScope = nullptr;
 
   if (isOwningBackingThread())
     workerBackingThread().shutdown();
   // We must not touch workerBackingThread() from now on.
 
   // Notify the proxy that the WorkerOrWorkletGlobalScope has been disposed
   // of. This can free this thread object, hence it must not be touched
   // afterwards.
   workerReportingProxy().didTerminateWorkerThread();
 
   m_shutdownEvent->signal();
 }
 
-template <WTF::FunctionThreadAffinity threadAffinity>
-void WorkerThread::performTaskOnWorkerThread(
-    std::unique_ptr<Function<void(), threadAffinity>> task) {
-  DCHECK(isCurrentThread());
-  if (m_threadState != ThreadState::Running)
-    return;
-
-  {
-    DEFINE_THREAD_SAFE_STATIC_LOCAL(
-        CustomCountHistogram, scopedUsCounter,
-        new CustomCountHistogram("WorkerThread.Task.Time", 0, 10000000, 50));
-    ScopedUsHistogramTimer timer(scopedUsCounter);
-    (*task)();
-  }
-}
-
 void WorkerThread::performDebuggerTaskOnWorkerThread(
     std::unique_ptr<CrossThreadClosure> task) {
   DCHECK(isCurrentThread());
   InspectorTaskRunner::IgnoreInterruptsScope scope(m_inspectorTaskRunner.get());
   {
     MutexLocker lock(m_threadStateMutex);
     DCHECK_EQ(ThreadState::Running, m_threadState);
     m_runningDebuggerTask = true;
   }
   ThreadDebugger::idleFinished(isolate());
@@ skipping 64 matching lines @@
   MutexLocker lock(m_threadStateMutex);
   return m_requestedToTerminate;
 }
 
 ExitCode WorkerThread::getExitCodeForTesting() {
   MutexLocker lock(m_threadStateMutex);
   return m_exitCode;
 }
 
 }  // namespace blink