ceph/src/seastar/include/seastar/testing/exchanger.hh

   1 /*
   2  * This file is open source software, licensed to you under the terms
   3  * of the Apache License, Version 2.0 (the "License").  See the NOTICE file
   4  * distributed with this work for additional information regarding copyright
   5  * ownership.  You may not use this file except in compliance with the License.
   6  *
   7  * You may obtain a copy of the License at
   8  *
   9  *   http://www.apache.org/licenses/LICENSE-2.0
  10  *
  11  * Unless required by applicable law or agreed to in writing,
  12  * software distributed under the License is distributed on an
  13  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  14  * KIND, either express or implied.  See the License for the
  15  * specific language governing permissions and limitations
  16  * under the License.
  17  */
  18 /*
  19  * Copyright (C) 2015 Cloudius Systems, Ltd.
  20  */
  21
  22 #pragma once
  23
  24 #include <mutex>
  25 #include <condition_variable>
  26 #include <seastar/util/std-compat.hh>
  27
  28 namespace seastar {
  29
  30 namespace testing {
  31
  32 // Single-element blocking queue
  33 template <typename T>
  34 class exchanger {
  35 private:
  36     std::mutex _mutex;
  37     std::condition_variable _cv;
  38     seastar::compat::optional<T> _element;
  39     std::exception_ptr _exception;
  40 private:
  41     void interrupt_ptr(std::exception_ptr e) {
  42         std::unique_lock<std::mutex> lock(_mutex);
  43         if (!_exception) {
  44             _exception = e;
  45             _cv.notify_all();
  46         }
  47         // FIXME: log if already interrupted
  48     }
  49 public:
  50     template <typename Exception>
  51     void interrupt(Exception e) {
  52         try {
  53             throw e;
  54         } catch (...) {
  55             interrupt_ptr(std::current_exception());
  56         }
  57     }
  58     void give(T value) {
  59         std::unique_lock<std::mutex> lock(_mutex);
  60         _cv.wait(lock, [this] { return !_element || _exception; });
  61         if (_exception) {
  62             std::rethrow_exception(_exception);
  63         }
  64         _element = value;
  65         _cv.notify_one();
  66     }
  67     T take() {
  68         std::unique_lock<std::mutex> lock(_mutex);
  69         _cv.wait(lock, [this] { return bool(_element) || _exception; });
  70         if (_exception) {
  71             std::rethrow_exception(_exception);
  72         }
  73         auto v = *_element;
  74         _element = {};
  75         _cv.notify_one();
  76         return v;
  77     }
  78 };
  79
  80 }
  81
  82 }