ceph/src/mds/PurgeQueue.h

   1 // -*- mode:C++; tab-width:8; c-basic-offset:2; indent-tabs-mode:t -*-
   2 // vim: ts=8 sw=2 smarttab
   3 /*
   4  * Ceph - scalable distributed file system
   5  *
   6  * Copyright (C) 2015 Red Hat
   7  *
   8  * This is free software; you can redistribute it and/or
   9  * modify it under the terms of the GNU Lesser General Public
  10  * License version 2.1, as published by the Free Software
  11  * Foundation.  See file COPYING.
  12  *
  13  */
  14
  15 #ifndef PURGE_QUEUE_H_
  16 #define PURGE_QUEUE_H_
  17
  18 #include "include/compact_set.h"
  19 #include "mds/MDSMap.h"
  20 #include "osdc/Journaler.h"
  21
  22
  23 /**
  24  * Descriptor of the work associated with purging a file.  We record
  25  * the minimal amount of information from the inode such as the size
  26  * and layout: all other un-needed inode metadata (times, permissions, etc)
  27  * has been discarded.
  28  */
  29 class PurgeItem
  30 {
  31 public:
  32   enum Action : uint8_t {
  33     NONE = 0,
  34     PURGE_FILE = 1,
  35     TRUNCATE_FILE,
  36     PURGE_DIR
  37   };
  38
  39   Action action;
  40   inodeno_t ino;
  41   uint64_t size;
  42   file_layout_t layout;
  43   compact_set<int64_t> old_pools;
  44   SnapContext snapc;
  45   fragtree_t fragtree;
  46
  47   PurgeItem()
  48    : action(NONE), ino(0), size(0)
  49   {}
  50
  51   void encode(bufferlist &bl) const;
  52   void decode(bufferlist::iterator &p);
  53 };
  54 WRITE_CLASS_ENCODER(PurgeItem)
  55
  56 enum {
  57   l_pq_first = 3500,
  58
  59   // How many items have been finished by PurgeQueue
  60   l_pq_executing_ops,
  61   l_pq_executing,
  62   l_pq_executed,
  63   l_pq_last
  64 };
  65
  66 /**
  67  * A persistent queue of PurgeItems.  This class both writes and reads
  68  * to the queue.  There is one of these per MDS rank.
  69  *
  70  * Note that this class does not take a reference to MDSRank: we are
  71  * independent of all the metadata structures and do not need to
  72  * take mds_lock for anything.
  73  */
  74 class PurgeQueue
  75 {
  76 protected:
  77   CephContext *cct;
  78   const mds_rank_t rank;
  79   Mutex lock;
  80
  81   int64_t metadata_pool;
  82
  83   // Don't use the MDSDaemon's Finisher and Timer, because this class
  84   // operates outside of MDSDaemon::mds_lock
  85   Finisher finisher;
  86   SafeTimer timer;
  87   Filer filer;
  88   Objecter *objecter;
  89   std::unique_ptr<PerfCounters> logger;
  90
  91   Journaler journaler;
  92
  93   Context *on_error;
  94
  95   // Map of Journaler offset to PurgeItem
  96   std::map<uint64_t, PurgeItem> in_flight;
  97
  98   // Throttled allowances
  99   uint64_t ops_in_flight;
 100
 101   // Dynamic op limit per MDS based on PG count
 102   uint64_t max_purge_ops;
 103
 104   uint32_t _calculate_ops(const PurgeItem &item) const;
 105
 106   bool can_consume();
 107
 108   // How many bytes were remaining when drain() was first called,
 109   // used for indicating progress.
 110   uint64_t drain_initial;
 111
 112   // Has drain() ever been called on this instance?
 113   bool draining;
 114
 115   // recover the journal write_pos (drop any partial written entry)
 116   void _recover();
 117
 118   /**
 119    * @return true if we were in a position to try and consume something:
 120    *         does not mean we necessarily did.
 121    */
 122   bool _consume();
 123
 124   // Do we currently have a flush timer event waiting?
 125   Context *delayed_flush;
 126
 127   void _execute_item(
 128       const PurgeItem &item,
 129       uint64_t expire_to);
 130   void _execute_item_complete(
 131       uint64_t expire_to);
 132
 133   bool recovered;
 134   std::list<Context*> waiting_for_recovery;
 135
 136 public:
 137   void init();
 138   void activate();
 139   void shutdown();
 140
 141   void create_logger();
 142
 143   // Write an empty queue, use this during MDS rank creation
 144   void create(Context *completion);
 145
 146   // Read the Journaler header for an existing queue and start consuming
 147   void open(Context *completion);
 148
 149   void wait_for_recovery(Context *c);
 150
 151   // Submit one entry to the work queue.  Call back when it is persisted
 152   // to the queue (there is no callback for when it is executed)
 153   void push(const PurgeItem &pi, Context *completion);
 154
 155   // If the on-disk queue is empty and we are not currently processing
 156   // anything.
 157   bool is_idle() const;
 158
 159   /**
 160    * Signal to the PurgeQueue that you would like it to hurry up and
 161    * finish consuming everything in the queue.  Provides progress
 162    * feedback.
 163    *
 164    * @param progress: bytes consumed since we started draining
 165    * @param progress_total: max bytes that were outstanding during purge
 166    * @param in_flight_count: number of file purges currently in flight
 167    *
 168    * @returns true if drain is complete
 169    */
 170   bool drain(
 171     uint64_t *progress,
 172     uint64_t *progress_total,
 173     size_t *in_flight_count);
 174
 175   void update_op_limit(const MDSMap &mds_map);
 176
 177   void handle_conf_change(const struct md_config_t *conf,
 178                           const std::set <std::string> &changed,
 179                           const MDSMap &mds_map);
 180
 181   PurgeQueue(
 182       CephContext *cct_,
 183       mds_rank_t rank_,
 184       const int64_t metadata_pool_,
 185       Objecter *objecter_,
 186       Context *on_error);
 187   ~PurgeQueue();
 188 };
 189
 190
 191 #endif
 192