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