folly/executors/SerialExecutor.cpp

   1 /*
   2  * Copyright 2017-present Facebook, Inc.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *   http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 #include "SerialExecutor.h"
  18
  19 #include <mutex>
  20 #include <queue>
  21
  22 #include <glog/logging.h>
  23
  24 #include <folly/ExceptionString.h>
  25
  26 namespace folly {
  27
  28 class SerialExecutor::TaskQueueImpl {
  29  public:
  30   void add(Func&& func) {
  31     std::lock_guard<std::mutex> lock(mutex_);
  32     queue_.push(std::move(func));
  33   }
  34
  35   void run() {
  36     std::unique_lock<std::mutex> lock(mutex_);
  37
  38     ++scheduled_;
  39
  40     if (scheduled_ > 1) {
  41       return;
  42     }
  43
  44     do {
  45       DCHECK(!queue_.empty());
  46       Func func = std::move(queue_.front());
  47       queue_.pop();
  48       lock.unlock();
  49
  50       try {
  51         func();
  52       } catch (std::exception const& ex) {
  53         LOG(ERROR) << "SerialExecutor: func threw unhandled exception "
  54                    << folly::exceptionStr(ex);
  55       } catch (...) {
  56         LOG(ERROR) << "SerialExecutor: func threw unhandled non-exception "
  57                       "object";
  58       }
  59
  60       // Destroy the function (and the data it captures) before we acquire the
  61       // lock again.
  62       func = {};
  63
  64       lock.lock();
  65       --scheduled_;
  66     } while (scheduled_);
  67   }
  68
  69  private:
  70   std::mutex mutex_;
  71   std::size_t scheduled_{0};
  72   std::queue<Func> queue_;
  73 };
  74
  75 SerialExecutor::SerialExecutor(std::shared_ptr<folly::Executor> parent)
  76     : parent_(std::move(parent)),
  77       taskQueueImpl_(std::make_shared<TaskQueueImpl>()) {}
  78
  79 void SerialExecutor::add(Func func) {
  80   taskQueueImpl_->add(std::move(func));
  81   parent_->add([impl = taskQueueImpl_] { impl->run(); });
  82 }
  83
  84 void SerialExecutor::addWithPriority(Func func, int8_t priority) {
  85   taskQueueImpl_->add(std::move(func));
  86   parent_->addWithPriority([impl = taskQueueImpl_] { impl->run(); }, priority);
  87 }
  88
  89 } // namespace folly