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
   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 "services/svc_rados.h"
  22 #include "rgw_aio.h"
  23
  24 namespace rgw {
  25
  26 // a throttle for aio operations that enforces a maximum window on outstanding
  27 // bytes. only supports a single waiter, so all public functions must be called
  28 // from the same thread
  29 class AioThrottle : public Aio {
  30  protected:
  31   const uint64_t window;
  32   uint64_t pending_size = 0;
  33
  34   bool is_available() const { return pending_size <= window; }
  35   bool has_completion() const { return !completed.empty(); }
  36   bool is_drained() const { return pending.empty(); }
  37
  38   struct Pending : AioResultEntry {
  39     AioThrottle *parent = nullptr;
  40     uint64_t cost = 0;
  41     librados::AioCompletion *completion = nullptr;
  42   };
  43   OwningList<Pending> pending;
  44   AioResultList completed;
  45
  46   enum class Wait { None, Available, Completion, Drained };
  47   Wait waiter = Wait::None;
  48
  49   bool waiter_ready() const;
  50
  51   ceph::mutex mutex = ceph::make_mutex("AioThrottle");
  52   ceph::condition_variable cond;
  53
  54   void get(Pending& p);
  55   void put(Pending& p);
  56
  57   static void aio_cb(void *cb, void *arg);
  58
  59  public:
  60   AioThrottle(uint64_t window) : window(window) {}
  61
  62   virtual ~AioThrottle() {
  63     // must drain before destructing
  64     ceph_assert(pending.empty());
  65     ceph_assert(completed.empty());
  66   }
  67
  68   AioResultList submit(RGWSI_RADOS::Obj& obj,
  69                        librados::ObjectReadOperation *op,
  70                        uint64_t cost, uint64_t id) override;
  71
  72   AioResultList submit(RGWSI_RADOS::Obj& obj,
  73                        librados::ObjectWriteOperation *op,
  74                        uint64_t cost, uint64_t id) override;
  75
  76   AioResultList poll() override;
  77
  78   AioResultList wait() override;
  79
  80   AioResultList drain() override;
  81 };
  82
  83 } // namespace rgw