ios/web/web_thread_impl.cc

   1 // Copyright 2014 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "ios/web/web_thread_impl.h"
   6
   7 #include <string>
   8
   9 #include "base/atomicops.h"
  10 #include "base/bind.h"
  11 #include "base/compiler_specific.h"
  12 #include "base/lazy_instance.h"
  13 #include "base/message_loop/message_loop.h"
  14 #include "base/message_loop/message_loop_proxy.h"
  15 #include "base/threading/sequenced_worker_pool.h"
  16 #include "base/threading/thread_restrictions.h"
  17 #include "ios/web/public/web_thread_delegate.h"
  18 #include "net/disk_cache/simple/simple_backend_impl.h"
  19 #include "net/url_request/url_fetcher.h"
  20
  21 namespace web {
  22
  23 namespace {
  24
  25 // Friendly names for the well-known threads.
  26 const char* g_web_thread_names[WebThread::ID_COUNT] = {
  27     "Web_UIThread",                // UI
  28     "Web_DBThread",                // DB
  29     "Web_FileThread",              // FILE
  30     "Web_FileUserBlockingThread",  // FILE_USER_BLOCKING
  31     "Web_CacheThread",             // CACHE
  32     "Web_IOThread",                // IO
  33 };
  34
  35 // An implementation of MessageLoopProxy to be used in conjunction
  36 // with WebThread.
  37 class WebThreadMessageLoopProxy : public base::MessageLoopProxy {
  38  public:
  39   explicit WebThreadMessageLoopProxy(WebThread::ID identifier)
  40       : id_(identifier) {}
  41
  42   // MessageLoopProxy implementation.
  43   bool PostDelayedTask(const tracked_objects::Location& from_here,
  44                        const base::Closure& task,
  45                        base::TimeDelta delay) override {
  46     return WebThread::PostDelayedTask(id_, from_here, task, delay);
  47   }
  48
  49   bool PostNonNestableDelayedTask(const tracked_objects::Location& from_here,
  50                                   const base::Closure& task,
  51                                   base::TimeDelta delay) override {
  52     return WebThread::PostNonNestableDelayedTask(id_, from_here, task, delay);
  53   }
  54
  55   bool RunsTasksOnCurrentThread() const override {
  56     return WebThread::CurrentlyOn(id_);
  57   }
  58
  59  protected:
  60   ~WebThreadMessageLoopProxy() override {}
  61
  62  private:
  63   WebThread::ID id_;
  64   DISALLOW_COPY_AND_ASSIGN(WebThreadMessageLoopProxy);
  65 };
  66
  67 // A separate helper is used just for the proxies, in order to avoid needing
  68 // to initialize the globals to create a proxy.
  69 struct WebThreadProxies {
  70   WebThreadProxies() {
  71     for (int i = 0; i < WebThread::ID_COUNT; ++i) {
  72       proxies[i] = new WebThreadMessageLoopProxy(static_cast<WebThread::ID>(i));
  73     }
  74   }
  75
  76   scoped_refptr<base::MessageLoopProxy> proxies[WebThread::ID_COUNT];
  77 };
  78
  79 base::LazyInstance<WebThreadProxies>::Leaky g_proxies =
  80     LAZY_INSTANCE_INITIALIZER;
  81
  82 struct WebThreadGlobals {
  83   WebThreadGlobals()
  84       : blocking_pool(new base::SequencedWorkerPool(3, "WebBlocking")) {
  85     memset(threads, 0, WebThread::ID_COUNT * sizeof(threads[0]));
  86     memset(thread_delegates, 0,
  87            WebThread::ID_COUNT * sizeof(thread_delegates[0]));
  88   }
  89
  90   // This lock protects |threads|. Do not read or modify that array
  91   // without holding this lock. Do not block while holding this lock.
  92   base::Lock lock;
  93
  94   // This array is protected by |lock|. The threads are not owned by this
  95   // array. Typically, the threads are owned on the UI thread by
  96   // WebMainLoop. WebThreadImpl objects remove themselves from this
  97   // array upon destruction.
  98   WebThreadImpl* threads[WebThread::ID_COUNT];
  99
 100   // Only atomic operations are used on this array. The delegates are not owned
 101   // by this array, rather by whoever calls WebThread::SetDelegate.
 102   WebThreadDelegate* thread_delegates[WebThread::ID_COUNT];
 103
 104   const scoped_refptr<base::SequencedWorkerPool> blocking_pool;
 105 };
 106
 107 base::LazyInstance<WebThreadGlobals>::Leaky g_globals =
 108     LAZY_INSTANCE_INITIALIZER;
 109
 110 }  // namespace
 111
 112 WebThreadImpl::WebThreadImpl(ID identifier)
 113     : Thread(g_web_thread_names[identifier]), identifier_(identifier) {
 114   Initialize();
 115 }
 116
 117 WebThreadImpl::WebThreadImpl(ID identifier, base::MessageLoop* message_loop)
 118     : Thread(message_loop->thread_name()), identifier_(identifier) {
 119   set_message_loop(message_loop);
 120   Initialize();
 121 }
 122
 123 // static
 124 void WebThreadImpl::ShutdownThreadPool() {
 125   // The goal is to make it impossible to 'infinite loop' during shutdown,
 126   // but to reasonably expect that all BLOCKING_SHUTDOWN tasks queued during
 127   // shutdown get run. There's nothing particularly scientific about the
 128   // number chosen.
 129   const int kMaxNewShutdownBlockingTasks = 1000;
 130   WebThreadGlobals& globals = g_globals.Get();
 131   globals.blocking_pool->Shutdown(kMaxNewShutdownBlockingTasks);
 132 }
 133
 134 // static
 135 void WebThreadImpl::FlushThreadPoolHelperForTesting() {
 136   // We don't want to create a pool if none exists.
 137   if (g_globals == nullptr)
 138     return;
 139   g_globals.Get().blocking_pool->FlushForTesting();
 140   disk_cache::SimpleBackendImpl::FlushWorkerPoolForTesting();
 141 }
 142
 143 void WebThreadImpl::Init() {
 144   WebThreadGlobals& globals = g_globals.Get();
 145
 146   using base::subtle::AtomicWord;
 147   AtomicWord* storage =
 148       reinterpret_cast<AtomicWord*>(&globals.thread_delegates[identifier_]);
 149   AtomicWord stored_pointer = base::subtle::NoBarrier_Load(storage);
 150   WebThreadDelegate* delegate =
 151       reinterpret_cast<WebThreadDelegate*>(stored_pointer);
 152   if (delegate) {
 153     delegate->Init();
 154     message_loop()->PostTask(FROM_HERE,
 155                              base::Bind(&WebThreadDelegate::InitAsync,
 156                                         // Delegate is expected to exist for the
 157                                         // duration of the thread's lifetime
 158                                         base::Unretained(delegate)));
 159   }
 160
 161   if (WebThread::CurrentlyOn(WebThread::IO)) {
 162     // Though this thread is called the "IO" thread, it actually just routes
 163     // messages around; it shouldn't be allowed to perform any blocking disk
 164     // I/O.
 165     base::ThreadRestrictions::SetIOAllowed(false);
 166     base::ThreadRestrictions::DisallowWaiting();
 167   }
 168 }
 169
 170 NOINLINE void WebThreadImpl::UIThreadRun(base::MessageLoop* message_loop) {
 171   volatile int line_number = __LINE__;
 172   Thread::Run(message_loop);
 173   CHECK_GT(line_number, 0);
 174 }
 175
 176 NOINLINE void WebThreadImpl::DBThreadRun(base::MessageLoop* message_loop) {
 177   volatile int line_number = __LINE__;
 178   Thread::Run(message_loop);
 179   CHECK_GT(line_number, 0);
 180 }
 181
 182 NOINLINE void WebThreadImpl::FileThreadRun(base::MessageLoop* message_loop) {
 183   volatile int line_number = __LINE__;
 184   Thread::Run(message_loop);
 185   CHECK_GT(line_number, 0);
 186 }
 187
 188 NOINLINE void WebThreadImpl::FileUserBlockingThreadRun(
 189     base::MessageLoop* message_loop) {
 190   volatile int line_number = __LINE__;
 191   Thread::Run(message_loop);
 192   CHECK_GT(line_number, 0);
 193 }
 194
 195 NOINLINE void WebThreadImpl::CacheThreadRun(base::MessageLoop* message_loop) {
 196   volatile int line_number = __LINE__;
 197   Thread::Run(message_loop);
 198   CHECK_GT(line_number, 0);
 199 }
 200
 201 NOINLINE void WebThreadImpl::IOThreadRun(base::MessageLoop* message_loop) {
 202   volatile int line_number = __LINE__;
 203   Thread::Run(message_loop);
 204   CHECK_GT(line_number, 0);
 205 }
 206
 207 void WebThreadImpl::Run(base::MessageLoop* message_loop) {
 208   WebThread::ID thread_id = ID_COUNT;
 209   if (!GetCurrentThreadIdentifier(&thread_id))
 210     return Thread::Run(message_loop);
 211
 212   switch (thread_id) {
 213     case WebThread::UI:
 214       return UIThreadRun(message_loop);
 215     case WebThread::DB:
 216       return DBThreadRun(message_loop);
 217     case WebThread::FILE:
 218       return FileThreadRun(message_loop);
 219     case WebThread::FILE_USER_BLOCKING:
 220       return FileUserBlockingThreadRun(message_loop);
 221     case WebThread::CACHE:
 222       return CacheThreadRun(message_loop);
 223     case WebThread::IO:
 224       return IOThreadRun(message_loop);
 225     case WebThread::ID_COUNT:
 226       CHECK(false);  // This shouldn't actually be reached!
 227       break;
 228   }
 229   Thread::Run(message_loop);
 230 }
 231
 232 void WebThreadImpl::CleanUp() {
 233   if (WebThread::CurrentlyOn(WebThread::IO))
 234     IOThreadPreCleanUp();
 235
 236   WebThreadGlobals& globals = g_globals.Get();
 237
 238   using base::subtle::AtomicWord;
 239   AtomicWord* storage =
 240       reinterpret_cast<AtomicWord*>(&globals.thread_delegates[identifier_]);
 241   AtomicWord stored_pointer = base::subtle::NoBarrier_Load(storage);
 242   WebThreadDelegate* delegate =
 243       reinterpret_cast<WebThreadDelegate*>(stored_pointer);
 244
 245   if (delegate)
 246     delegate->CleanUp();
 247 }
 248
 249 void WebThreadImpl::Initialize() {
 250   WebThreadGlobals& globals = g_globals.Get();
 251
 252   base::AutoLock lock(globals.lock);
 253   DCHECK(identifier_ >= 0 && identifier_ < ID_COUNT);
 254   DCHECK(globals.threads[identifier_] == nullptr);
 255   globals.threads[identifier_] = this;
 256 }
 257
 258 void WebThreadImpl::IOThreadPreCleanUp() {
 259   // Kill all things that might be holding onto
 260   // net::URLRequest/net::URLRequestContexts.
 261
 262   // Destroy all URLRequests started by URLFetchers.
 263   net::URLFetcher::CancelAll();
 264 }
 265
 266 WebThreadImpl::~WebThreadImpl() {
 267   // All Thread subclasses must call Stop() in the destructor. This is
 268   // doubly important here as various bits of code check they are on
 269   // the right WebThread.
 270   Stop();
 271
 272   WebThreadGlobals& globals = g_globals.Get();
 273   base::AutoLock lock(globals.lock);
 274   globals.threads[identifier_] = nullptr;
 275 #ifndef NDEBUG
 276   // Double check that the threads are ordered correctly in the enumeration.
 277   for (int i = identifier_ + 1; i < ID_COUNT; ++i) {
 278     DCHECK(!globals.threads[i])
 279         << "Threads must be listed in the reverse order that they die";
 280   }
 281 #endif
 282 }
 283
 284 // static
 285 bool WebThreadImpl::PostTaskHelper(WebThread::ID identifier,
 286                                    const tracked_objects::Location& from_here,
 287                                    const base::Closure& task,
 288                                    base::TimeDelta delay,
 289                                    bool nestable) {
 290   DCHECK(identifier >= 0 && identifier < ID_COUNT);
 291   // Optimization: to avoid unnecessary locks, we listed the ID enumeration in
 292   // order of lifetime.  So no need to lock if we know that the target thread
 293   // outlives current thread.
 294   // Note: since the array is so small, ok to loop instead of creating a map,
 295   // which would require a lock because std::map isn't thread safe, defeating
 296   // the whole purpose of this optimization.
 297   WebThread::ID current_thread = ID_COUNT;
 298   bool target_thread_outlives_current =
 299       GetCurrentThreadIdentifier(&current_thread) &&
 300       current_thread >= identifier;
 301
 302   WebThreadGlobals& globals = g_globals.Get();
 303   if (!target_thread_outlives_current)
 304     globals.lock.Acquire();
 305
 306   base::MessageLoop* message_loop =
 307       globals.threads[identifier] ? globals.threads[identifier]->message_loop()
 308                                   : nullptr;
 309   if (message_loop) {
 310     if (nestable) {
 311       message_loop->PostDelayedTask(from_here, task, delay);
 312     } else {
 313       message_loop->PostNonNestableDelayedTask(from_here, task, delay);
 314     }
 315   }
 316
 317   if (!target_thread_outlives_current)
 318     globals.lock.Release();
 319
 320   return !!message_loop;
 321 }
 322
 323 // static
 324 bool WebThread::PostBlockingPoolTask(const tracked_objects::Location& from_here,
 325                                      const base::Closure& task) {
 326   return g_globals.Get().blocking_pool->PostWorkerTask(from_here, task);
 327 }
 328
 329 // static
 330 bool WebThread::PostBlockingPoolTaskAndReply(
 331     const tracked_objects::Location& from_here,
 332     const base::Closure& task,
 333     const base::Closure& reply) {
 334   return g_globals.Get().blocking_pool->PostTaskAndReply(from_here, task,
 335                                                          reply);
 336 }
 337
 338 // static
 339 bool WebThread::PostBlockingPoolSequencedTask(
 340     const std::string& sequence_token_name,
 341     const tracked_objects::Location& from_here,
 342     const base::Closure& task) {
 343   return g_globals.Get().blocking_pool->PostNamedSequencedWorkerTask(
 344       sequence_token_name, from_here, task);
 345 }
 346
 347 // static
 348 base::SequencedWorkerPool* WebThread::GetBlockingPool() {
 349   return g_globals.Get().blocking_pool.get();
 350 }
 351
 352 // static
 353 bool WebThread::IsThreadInitialized(ID identifier) {
 354   if (g_globals == nullptr)
 355     return false;
 356
 357   WebThreadGlobals& globals = g_globals.Get();
 358   base::AutoLock lock(globals.lock);
 359   DCHECK(identifier >= 0 && identifier < ID_COUNT);
 360   return globals.threads[identifier] != nullptr;
 361 }
 362
 363 // static
 364 bool WebThread::CurrentlyOn(ID identifier) {
 365   // This shouldn't use MessageLoop::current() since it uses LazyInstance which
 366   // may be deleted by ~AtExitManager when a WorkerPool thread calls this
 367   // function.
 368   // http://crbug.com/63678
 369   base::ThreadRestrictions::ScopedAllowSingleton allow_singleton;
 370   WebThreadGlobals& globals = g_globals.Get();
 371   base::AutoLock lock(globals.lock);
 372   DCHECK(identifier >= 0 && identifier < ID_COUNT);
 373   return globals.threads[identifier] &&
 374          globals.threads[identifier]->message_loop() ==
 375              base::MessageLoop::current();
 376 }
 377
 378 static const char* GetThreadName(WebThread::ID thread) {
 379   if (WebThread::UI <= thread && thread < WebThread::ID_COUNT)
 380     return g_web_thread_names[thread];
 381   return "Unknown Thread";
 382 }
 383
 384 // static
 385 std::string WebThread::GetDCheckCurrentlyOnErrorMessage(ID expected) {
 386   const base::MessageLoop* message_loop = base::MessageLoop::current();
 387   ID actual_web_thread;
 388   const char* actual_name = "Unknown Thread";
 389   if (message_loop && !message_loop->thread_name().empty()) {
 390     actual_name = message_loop->thread_name().c_str();
 391   } else if (GetCurrentThreadIdentifier(&actual_web_thread)) {
 392     actual_name = GetThreadName(actual_web_thread);
 393   }
 394   std::string result = "Must be called on ";
 395   result += GetThreadName(expected);
 396   result += "; actually called on ";
 397   result += actual_name;
 398   result += ".";
 399   return result;
 400 }
 401
 402 // static
 403 bool WebThread::IsMessageLoopValid(ID identifier) {
 404   if (g_globals == nullptr)
 405     return false;
 406
 407   WebThreadGlobals& globals = g_globals.Get();
 408   base::AutoLock lock(globals.lock);
 409   DCHECK(identifier >= 0 && identifier < ID_COUNT);
 410   return globals.threads[identifier] &&
 411          globals.threads[identifier]->message_loop();
 412 }
 413
 414 // static
 415 bool WebThread::PostTask(ID identifier,
 416                          const tracked_objects::Location& from_here,
 417                          const base::Closure& task) {
 418   return WebThreadImpl::PostTaskHelper(identifier, from_here, task,
 419                                        base::TimeDelta(), true);
 420 }
 421
 422 // static
 423 bool WebThread::PostDelayedTask(ID identifier,
 424                                 const tracked_objects::Location& from_here,
 425                                 const base::Closure& task,
 426                                 base::TimeDelta delay) {
 427   return WebThreadImpl::PostTaskHelper(identifier, from_here, task, delay,
 428                                        true);
 429 }
 430
 431 // static
 432 bool WebThread::PostNonNestableTask(ID identifier,
 433                                     const tracked_objects::Location& from_here,
 434                                     const base::Closure& task) {
 435   return WebThreadImpl::PostTaskHelper(identifier, from_here, task,
 436                                        base::TimeDelta(), false);
 437 }
 438
 439 // static
 440 bool WebThread::PostNonNestableDelayedTask(
 441     ID identifier,
 442     const tracked_objects::Location& from_here,
 443     const base::Closure& task,
 444     base::TimeDelta delay) {
 445   return WebThreadImpl::PostTaskHelper(identifier, from_here, task, delay,
 446                                        false);
 447 }
 448
 449 // static
 450 bool WebThread::PostTaskAndReply(ID identifier,
 451                                  const tracked_objects::Location& from_here,
 452                                  const base::Closure& task,
 453                                  const base::Closure& reply) {
 454   return GetMessageLoopProxyForThread(identifier)
 455       ->PostTaskAndReply(from_here, task, reply);
 456 }
 457
 458 // static
 459 bool WebThread::GetCurrentThreadIdentifier(ID* identifier) {
 460   if (g_globals == nullptr)
 461     return false;
 462
 463   // This shouldn't use MessageLoop::current() since it uses LazyInstance which
 464   // may be deleted by ~AtExitManager when a WorkerPool thread calls this
 465   // function.
 466   // http://crbug.com/63678
 467   base::ThreadRestrictions::ScopedAllowSingleton allow_singleton;
 468   base::MessageLoop* cur_message_loop = base::MessageLoop::current();
 469   WebThreadGlobals& globals = g_globals.Get();
 470   for (int i = 0; i < ID_COUNT; ++i) {
 471     if (globals.threads[i] &&
 472         globals.threads[i]->message_loop() == cur_message_loop) {
 473       *identifier = globals.threads[i]->identifier_;
 474       return true;
 475     }
 476   }
 477
 478   return false;
 479 }
 480
 481 // static
 482 scoped_refptr<base::MessageLoopProxy> WebThread::GetMessageLoopProxyForThread(
 483     ID identifier) {
 484   return g_proxies.Get().proxies[identifier];
 485 }
 486
 487 // static
 488 base::MessageLoop* WebThread::UnsafeGetMessageLoopForThread(ID identifier) {
 489   if (g_globals == nullptr)
 490     return nullptr;
 491
 492   WebThreadGlobals& globals = g_globals.Get();
 493   base::AutoLock lock(globals.lock);
 494   base::Thread* thread = globals.threads[identifier];
 495   DCHECK(thread);
 496   base::MessageLoop* loop = thread->message_loop();
 497   return loop;
 498 }
 499
 500 // static
 501 void WebThreadImpl::SetDelegate(ID identifier, WebThreadDelegate* delegate) {
 502   using base::subtle::AtomicWord;
 503   WebThreadGlobals& globals = g_globals.Get();
 504   AtomicWord* storage =
 505       reinterpret_cast<AtomicWord*>(&globals.thread_delegates[identifier]);
 506   AtomicWord old_pointer = base::subtle::NoBarrier_AtomicExchange(
 507       storage, reinterpret_cast<AtomicWord>(delegate));
 508
 509   // This catches registration when previously registered.
 510   DCHECK(!delegate || !old_pointer);
 511 }
 512
 513 }  // namespace web