folly/executors/task_queue/UnboundedBlockingQueue.h

   1 /*
   2  * Copyright 2017 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 #pragma once
  18
  19 #include <folly/LifoSem.h>
  20 #include <folly/Synchronized.h>
  21 #include <folly/executors/task_queue/BlockingQueue.h>
  22 #include <queue>
  23
  24 namespace folly {
  25
  26 // Warning: this is effectively just a std::deque wrapped in a single mutex
  27 // We are aiming to add a more performant concurrent unbounded queue in the
  28 // future, but this class is available if you must have an unbounded queue
  29 // and can tolerate any contention.
  30 template <class T>
  31 class UnboundedBlockingQueue : public BlockingQueue<T> {
  32  public:
  33   virtual ~UnboundedBlockingQueue() {}
  34
  35   void add(T item) override {
  36     queue_.wlock()->push(std::move(item));
  37     sem_.post();
  38   }
  39
  40   T take() override {
  41     while (true) {
  42       {
  43         auto ulockedQueue = queue_.ulock();
  44         if (!ulockedQueue->empty()) {
  45           auto wlockedQueue = ulockedQueue.moveFromUpgradeToWrite();
  46           T item = std::move(wlockedQueue->front());
  47           wlockedQueue->pop();
  48           return item;
  49         }
  50       }
  51       sem_.wait();
  52     }
  53   }
  54
  55   size_t size() override {
  56     return queue_.rlock()->size();
  57   }
  58
  59  private:
  60   LifoSem sem_;
  61   Synchronized<std::queue<T>> queue_;
  62 };
  63
  64 } // namespace folly