ceph/src/rgw/rgw_aio_throttle.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab ft=cpp
   3
   4 /*
   5  * Ceph - scalable distributed file system
   6  *
   7  * Copyright (C) 2018 Red Hat, Inc.
   8  *
   9  * This is free software; you can redistribute it and/or
  10  * modify it under the terms of the GNU Lesser General Public
  11  * License version 2.1, as published by the Free Software
  12  * Foundation. See file COPYING.
  13  *
  14  */
  15
  16 #pragma once
  17
  18 #include "include/rados/librados_fwd.hpp"
  19 #include <memory>
  20 #include "common/ceph_mutex.h"
  21 #include "common/async/completion.h"
  22 #include "common/async/yield_context.h"
  23 #include "services/svc_rados.h"
  24 #include "rgw_aio.h"
  25
  26 namespace rgw {
  27
  28 class Throttle {
  29  protected:
  30   const uint64_t window;
  31   uint64_t pending_size = 0;
  32
  33   AioResultList pending;
  34   AioResultList completed;
  35
  36   bool is_available() const { return pending_size <= window; }
  37   bool has_completion() const { return !completed.empty(); }
  38   bool is_drained() const { return pending.empty(); }
  39
  40   enum class Wait { None, Available, Completion, Drained };
  41   Wait waiter = Wait::None;
  42
  43   bool waiter_ready() const;
  44
  45  public:
  46   Throttle(uint64_t window) : window(window) {}
  47
  48   virtual ~Throttle() {
  49     // must drain before destructing
  50     ceph_assert(pending.empty());
  51     ceph_assert(completed.empty());
  52   }
  53 };
  54
  55 // a throttle for aio operations. all public functions must be called from
  56 // the same thread
  57 class BlockingAioThrottle final : public Aio, private Throttle {
  58   ceph::mutex mutex = ceph::make_mutex("AioThrottle");
  59   ceph::condition_variable cond;
  60
  61   struct Pending : AioResultEntry {
  62     BlockingAioThrottle *parent = nullptr;
  63     uint64_t cost = 0;
  64     librados::AioCompletion *completion = nullptr;
  65   };
  66  public:
  67   BlockingAioThrottle(uint64_t window) : Throttle(window) {}
  68
  69   virtual ~BlockingAioThrottle() override {};
  70
  71   AioResultList get(const RGWSI_RADOS::Obj& obj, OpFunc&& f,
  72                     uint64_t cost, uint64_t id) override final;
  73
  74   void put(AioResult& r) override final;
  75
  76   AioResultList poll() override final;
  77
  78   AioResultList wait() override final;
  79
  80   AioResultList drain() override final;
  81 };
  82
  83 // a throttle that yields the coroutine instead of blocking. all public
  84 // functions must be called within the coroutine strand
  85 class YieldingAioThrottle final : public Aio, private Throttle {
  86   boost::asio::io_context& context;
  87   yield_context yield;
  88   struct Handler;
  89
  90   // completion callback associated with the waiter
  91   using Completion = ceph::async::Completion<void(boost::system::error_code)>;
  92   std::unique_ptr<Completion> completion;
  93
  94   template <typename CompletionToken>
  95   auto async_wait(CompletionToken&& token);
  96
  97   struct Pending : AioResultEntry { uint64_t cost = 0; };
  98
  99  public:
 100   YieldingAioThrottle(uint64_t window, boost::asio::io_context& context,
 101                       yield_context yield)
 102     : Throttle(window), context(context), yield(yield)
 103   {}
 104
 105   virtual ~YieldingAioThrottle() override {};
 106
 107   AioResultList get(const RGWSI_RADOS::Obj& obj, OpFunc&& f,
 108                     uint64_t cost, uint64_t id) override final;
 109
 110   void put(AioResult& r) override final;
 111
 112   AioResultList poll() override final;
 113
 114   AioResultList wait() override final;
 115
 116   AioResultList drain() override final;
 117 };
 118
 119 // return a smart pointer to Aio
 120 inline auto make_throttle(uint64_t window_size, optional_yield y)
 121 {
 122   std::unique_ptr<Aio> aio;
 123   if (y) {
 124     aio = std::make_unique<YieldingAioThrottle>(window_size,
 125                                                 y.get_io_context(),
 126                                                 y.get_yield_context());
 127   } else {
 128     aio = std::make_unique<BlockingAioThrottle>(window_size);
 129   }
 130   return aio;
 131 }
 132
 133 } // namespace rgw