ceph/src/common/Continuation.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) 2014 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 #include "include/Context.h"
  16
  17 /**
  18  * The Continuation interface is designed to help easily create multi-step
  19  * operations that share data without having to pass it around or create
  20  * custom Context classes for each step. To write a Continuation:
  21  * 1) create a child class with a function for each stage.
  22  * 2) Put all your shared data members into the class.
  23  * 3) In the constructor, register each function stage with set_callback().
  24  * 4) Whenever you need to provide a Context callback that activates the next
  25  * stage, call get_callback(stage_number). If you need to proceed to another
  26  * stage immediately, call immediate(stage, retcode) and return its result.
  27  *
  28  * To use a class:
  29  * 1) Construct the child class on the heap.
  30  * 2) Call begin().
  31  * 3) The destructor will be called once one of your functions returns true to
  32  * indicate it is done.
  33  *
  34  * Please note that while you can skip stages and get multiple Callback
  35  * objects at once, you *cannot* have any stage report that the Continuation
  36  * is completed while any other stage Callbacks are outstanding. It's best to
  37  * be serial unless you want to maintain your own metadata about which stages
  38  * are still pending.
  39  *
  40  * In fact, there are only two situations in which a stage should return
  41  * true while others are running:
  42  * 1) A Callback was issued and completed in the same thread,
  43  * 2) you called immediate(stage) and it is returning true.
  44  */
  45
  46 class Continuation {
  47   std::set<int> stages_in_flight;
  48   std::set<int> stages_processing;
  49   int rval;
  50   Context *on_finish;
  51   bool reported_done;
  52
  53   class Callback : public Context {
  54     Continuation *continuation;
  55     int stage_to_activate;
  56   public:
  57     Callback(Continuation *c, int stage) :
  58       continuation(c),
  59       stage_to_activate(stage) {}
  60     void finish(int r) override {
  61       continuation->continue_function(r, stage_to_activate);
  62     }
  63   };
  64
  65 protected:
  66   typedef bool (Continuation::*stagePtr)(int r);
  67   /**
  68    * Continue immediately to the given stage. It will be executed
  69    * immediately, in the given thread.
  70    * @pre You are in a callback function.
  71    * @param stage The stage to execute
  72    * @param r The return code that will be provided to the next stage
  73    */
  74   bool immediate(int stage, int r) {
  75     assert(!stages_in_flight.count(stage));
  76     assert(!stages_processing.count(stage));
  77     stages_in_flight.insert(stage);
  78     stages_processing.insert(stage);
  79     return _continue_function(r, stage);
  80   }
  81
  82   /**
  83    * Obtain a Context * that when complete()ed calls back into the given stage.
  84    * @pre You are in a callback function.
  85    * @param stage The stage this Context should activate
  86    */
  87   Context *get_callback(int stage) {
  88     stages_in_flight.insert(stage);
  89     return new Callback(this, stage);
  90   }
  91
  92   /**
  93    * Set the return code that is passed to the finally-activated Context.
  94    * @param new_rval The return code to use.
  95    */
  96   void set_rval(int new_rval) { rval = new_rval; }
  97   int get_rval() { return rval; }
  98
  99   /**
 100    * Register member functions as associated with a given stage. Start
 101    * your stage IDs at 0 and make that one the setup phase.
 102    * @pre There are no other functions associated with the stage.
 103    * @param stage The stage to associate this function with
 104    * @param func The function to use
 105    */
 106   void set_callback(int stage, stagePtr func) {
 107     assert(callbacks.find(stage) == callbacks.end());
 108     callbacks[stage] = func;
 109   }
 110
 111   /**
 112    * Called when the Continuation is done, as determined by a stage returning
 113    * true and us having finished all the currently-processing ones.
 114    */
 115    virtual void _done() {
 116      on_finish->complete(rval);
 117      on_finish = NULL;
 118      return;
 119    }
 120
 121 private:
 122   std::map<int, Continuation::stagePtr> callbacks;
 123
 124   bool _continue_function(int r, int n) {
 125     set<int>::iterator stage_iter = stages_in_flight.find(n);
 126     assert(stage_iter != stages_in_flight.end());
 127     assert(callbacks.count(n));
 128     stagePtr p = callbacks[n];
 129
 130     pair<set<int>::iterator,bool> insert_r = stages_processing.insert(n);
 131
 132     bool done = (this->*p)(r);
 133     if (done)
 134       reported_done = true;
 135
 136     stages_processing.erase(insert_r.first);
 137     stages_in_flight.erase(stage_iter);
 138     return done;
 139   }
 140
 141   void continue_function(int r, int stage) {
 142     bool done = _continue_function(r, stage);
 143
 144     assert (!done ||
 145             stages_in_flight.size() == stages_processing.size());
 146
 147     if (done ||
 148         (reported_done && stages_processing.empty())) {
 149       _done();
 150       delete this;
 151     }
 152   }
 153
 154
 155
 156 public:
 157   /**
 158    * Construct a new Continuation object. Call this from your child class,
 159    * obviously.
 160    *
 161    * @Param c The Context which should be complete()ed when this Continuation
 162    * is done.
 163    */
 164   Continuation(Context *c) :
 165     rval(0), on_finish(c), reported_done(false) {}
 166   /**
 167    * Clean up.
 168    */
 169   virtual ~Continuation() { assert(on_finish == NULL); }
 170   /**
 171    * Begin running the Continuation.
 172    */
 173   void begin() { stages_in_flight.insert(0); continue_function(0, 0); }
 174 };