OpTracker: start making the OpTracker into a generic

We're starting by templating the create_request function and having
the OpTracker work in terms of TrackedOp[Ref]s intead of
OpRequest[Ref]s. This involves moving a bunch of stuff into the
base-class TrackedOp.

Signed-off-by: Greg Farnum <greg@inktank.com>

4 files changed, 87 insertions, 68 deletions

diff --git a/src/common/TrackedOp.h b/src/common/TrackedOp.h
index 753331df7f3..835cf86e9d3 100644
--- a/src/common/TrackedOp.h
+++ b/src/common/TrackedOp.h
@@ -22,9 +22,41 @@
 #include <tr1/memory>
 
 class TrackedOp {
+protected:
+  list<pair<utime_t, string> > events; /// list of events and their times
+  Mutex lock; /// to protect the events list
 public:
-  virtual void mark_event(const string &event) = 0;
+  // move these to private once friended OpTracker
+  Message *request;
+  xlist<TrackedOp*>::item xitem;
+  utime_t received_time;
+  // figure out how to get rid of this one?
+  uint8_t warn_interval_multiplier;
+  string current;
+  uint64_t seq;
+
+  TrackedOp(Message *req) :
+    lock("TrackedOp::lock"),
+    request(req),
+    xitem(this),
+    warn_interval_multiplier(1),
+    seq(0) {}
+
   virtual ~TrackedOp() {}
+
+  utime_t get_arrived() const {
+    return received_time;
+  }
+  // This function maybe needs some work; assumes last event is completion time
+  double get_duration() const {
+    return events.size() ?
+      (events.rbegin()->first - received_time) :
+      0.0;
+  }
+
+  virtual void mark_event(const string &event) = 0;
+  virtual const char *state_string() const = 0;
+  virtual void dump(utime_t now, Formatter *f) const = 0;
 };
 typedef std::tr1::shared_ptr<TrackedOp> TrackedOpRef;
 
diff --git a/src/osd/OSD.cc b/src/osd/OSD.cc
index ff1276969d8..d51e9dc4434 100644
--- a/src/osd/OSD.cc
+++ b/src/osd/OSD.cc
@@ -4724,7 +4724,7 @@ void OSD::_dispatch(Message *m)
 
   default:
     {
-      OpRequestRef op = op_tracker.create_request(m);
+      OpRequestRef op = op_tracker.create_request<OpRequest, OpRequestRef>(m);
       op->mark_event("waiting_for_osdmap");
       // no map?  starting up?
       if (!osdmap) {
diff --git a/src/osd/OpRequest.cc b/src/osd/OpRequest.cc
index 1ffe3073051..388faf36763 100644
--- a/src/osd/OpRequest.cc
+++ b/src/osd/OpRequest.cc
@@ -21,13 +21,10 @@ static ostream& _prefix(std::ostream* _dout)
 }
 
 OpRequest::OpRequest(Message *req, OpTracker *tracker) :
-  request(req), xitem(this),
+  TrackedOp(req),
   rmw_flags(0),
-  warn_interval_multiplier(1),
-  lock("OpRequest::lock"),
   tracker(tracker),
-  hit_flag_points(0), latest_flag_point(0),
-  seq(0) {
+  hit_flag_points(0), latest_flag_point(0) {
   received_time = request->get_recv_stamp();
   tracker->register_inflight_op(&xitem);
   if (req->get_priority() < tracker->cct->_conf->osd_client_op_priority) {
@@ -43,7 +40,7 @@ void OpHistory::on_shutdown()
   shutdown = true;
 }
 
-void OpHistory::insert(utime_t now, OpRequestRef op)
+void OpHistory::insert(utime_t now, TrackedOpRef op)
 {
   if (shutdown)
     return;
@@ -79,7 +76,7 @@ void OpHistory::dump_ops(utime_t now, Formatter *f)
   f->dump_int("duration to keep", tracker->cct->_conf->osd_op_history_duration);
   {
     f->open_array_section("Ops");
-    for (set<pair<utime_t, OpRequestRef> >::const_iterator i =
+    for (set<pair<utime_t, TrackedOpRef> >::const_iterator i =
 	   arrived.begin();
 	 i != arrived.end();
 	 ++i) {
@@ -104,32 +101,32 @@ void OpTracker::dump_ops_in_flight(Formatter *f)
   Mutex::Locker locker(ops_in_flight_lock);
   f->open_object_section("ops_in_flight"); // overall dump
   f->dump_int("num_ops", ops_in_flight.size());
-  f->open_array_section("ops"); // list of OpRequests
+  f->open_array_section("ops"); // list of TrackedOps
   utime_t now = ceph_clock_now(cct);
-  for (xlist<OpRequest*>::iterator p = ops_in_flight.begin(); !p.end(); ++p) {
+  for (xlist<TrackedOp*>::iterator p = ops_in_flight.begin(); !p.end(); ++p) {
     f->open_object_section("op");
     (*p)->dump(now, f);
-    f->close_section(); // this OpRequest
+    f->close_section(); // this TrackedOp
   }
-  f->close_section(); // list of OpRequests
+  f->close_section(); // list of TrackedOps
   f->close_section(); // overall dump
 }
 
-void OpTracker::register_inflight_op(xlist<OpRequest*>::item *i)
+void OpTracker::register_inflight_op(xlist<TrackedOp*>::item *i)
 {
   Mutex::Locker locker(ops_in_flight_lock);
   ops_in_flight.push_back(i);
   ops_in_flight.back()->seq = seq++;
 }
 
-void OpTracker::unregister_inflight_op(OpRequest *i)
+void OpTracker::unregister_inflight_op(TrackedOp *i)
 {
   Mutex::Locker locker(ops_in_flight_lock);
   assert(i->xitem.get_list() == &ops_in_flight);
   utime_t now = ceph_clock_now(cct);
   i->xitem.remove_myself();
   i->request->clear_data();
-  history.insert(now, OpRequestRef(i));
+  history.insert(now, TrackedOpRef(i));
 }
 
 bool OpTracker::check_ops_in_flight(std::vector<string> &warning_vector)
@@ -151,7 +148,7 @@ bool OpTracker::check_ops_in_flight(std::vector<string> &warning_vector)
   if (oldest_secs < cct->_conf->osd_op_complaint_time)
     return false;
 
-  xlist<OpRequest*>::iterator i = ops_in_flight.begin();
+  xlist<TrackedOp*>::iterator i = ops_in_flight.begin();
   warning_vector.reserve(cct->_conf->osd_op_log_threshold + 1);
 
   int slow = 0;     // total slow
@@ -205,7 +202,7 @@ void OpTracker::get_age_ms_histogram(pow2_hist_t *h)
   unsigned bin = 30;
   uint32_t lb = 1 << (bin-1);  // lower bound for this bin
   int count = 0;
-  for (xlist<OpRequest*>::iterator i = ops_in_flight.begin(); !i.end(); ++i) {
+  for (xlist<TrackedOp*>::iterator i = ops_in_flight.begin(); !i.end(); ++i) {
     utime_t age = now - (*i)->received_time;
     uint32_t ms = (long)(age * 1000.0);
     if (ms >= lb) {
@@ -257,44 +254,28 @@ void OpRequest::dump(utime_t now, Formatter *f) const
   }
 }
 
-void OpTracker::mark_event(OpRequest *op, const string &dest)
+void OpTracker::mark_event(TrackedOp *op, const string &dest)
 {
   utime_t now = ceph_clock_now(cct);
   return _mark_event(op, dest, now);
 }
 
-void OpTracker::_mark_event(OpRequest *op, const string &evt,
+void OpTracker::_mark_event(TrackedOp *op, const string &evt,
 			    utime_t time)
 {
   Mutex::Locker locker(ops_in_flight_lock);
-  dout(5) << "reqid: " << op->get_reqid() << ", seq: " << op->seq
+  dout(5) << //"reqid: " << op->get_reqid() <<
+	     ", seq: " << op->seq
 	  << ", time: " << time << ", event: " << evt
 	  << ", request: " << *op->request << dendl;
 }
 
-void OpTracker::RemoveOnDelete::operator()(OpRequest *op) {
+void OpTracker::RemoveOnDelete::operator()(TrackedOp *op) {
   op->mark_event("done");
   tracker->unregister_inflight_op(op);
   // Do not delete op, unregister_inflight_op took control
 }
 
-OpRequestRef OpTracker::create_request(Message *ref)
-{
-  OpRequestRef retval(new OpRequest(ref, this),
-		      RemoveOnDelete(this));
-
-  if (ref->get_type() == CEPH_MSG_OSD_OP) {
-    retval->reqid = static_cast<MOSDOp*>(ref)->get_reqid();
-  } else if (ref->get_type() == MSG_OSD_SUBOP) {
-    retval->reqid = static_cast<MOSDSubOp*>(ref)->reqid;
-  }
-  _mark_event(retval.get(), "header_read", ref->get_recv_stamp());
-  _mark_event(retval.get(), "throttled", ref->get_throttle_stamp());
-  _mark_event(retval.get(), "all_read", ref->get_recv_complete_stamp());
-  _mark_event(retval.get(), "dispatched", ref->get_dispatch_stamp());
-  return retval;
-}
-
 void OpRequest::mark_event(const string &event)
 {
   utime_t now = ceph_clock_now(tracker->cct);
diff --git a/src/osd/OpRequest.h b/src/osd/OpRequest.h
index d429e6e717f..f501c649699 100644
--- a/src/osd/OpRequest.h
+++ b/src/osd/OpRequest.h
@@ -25,13 +25,14 @@
 #include <tr1/memory>
 #include "common/TrackedOp.h"
 #include "osd/osd_types.h"
+// FIXME: augh, get these outta here!
+#include "messages/MOSDOp.h"
+#include "messages/MOSDSubOp.h"
 
-struct OpRequest;
 class OpTracker;
-typedef std::tr1::shared_ptr<OpRequest> OpRequestRef;
 class OpHistory {
-  set<pair<utime_t, OpRequestRef> > arrived;
-  set<pair<double, OpRequestRef> > duration;
+  set<pair<utime_t, TrackedOpRef> > arrived;
+  set<pair<double, TrackedOpRef> > duration;
   void cleanup(utime_t now);
   bool shutdown;
   OpTracker *tracker;
@@ -42,7 +43,7 @@ public:
     assert(arrived.empty());
     assert(duration.empty());
   }
-  void insert(utime_t now, OpRequestRef op);
+  void insert(utime_t now, TrackedOpRef op);
   void dump_ops(utime_t now, Formatter *f);
   void on_shutdown();
 };
@@ -52,14 +53,14 @@ class OpTracker {
     OpTracker *tracker;
   public:
     RemoveOnDelete(OpTracker *tracker) : tracker(tracker) {}
-    void operator()(OpRequest *op);
+    void operator()(TrackedOp *op);
   };
   friend class RemoveOnDelete;
   friend class OpRequest;
   friend class OpHistory;
   uint64_t seq;
   Mutex ops_in_flight_lock;
-  xlist<OpRequest *> ops_in_flight;
+  xlist<TrackedOp *> ops_in_flight;
   OpHistory history;
 
 protected:
@@ -69,8 +70,8 @@ public:
   OpTracker(CephContext *cct_) : seq(0), ops_in_flight_lock("OpTracker mutex"), history(this), cct(cct_) {}
   void dump_ops_in_flight(Formatter *f);
   void dump_historic_ops(Formatter *f);
-  void register_inflight_op(xlist<OpRequest*>::item *i);
-  void unregister_inflight_op(OpRequest *i);
+  void register_inflight_op(xlist<TrackedOp*>::item *i);
+  void unregister_inflight_op(TrackedOp *i);
 
   void get_age_ms_histogram(pow2_hist_t *h);
 
@@ -83,9 +84,9 @@ public:
    * @return True if there are any Ops to warn on, false otherwise.
    */
   bool check_ops_in_flight(std::vector<string> &warning_strings);
-  void mark_event(OpRequest *op, const string &evt);
-  void _mark_event(OpRequest *op, const string &evt, utime_t now);
-  OpRequestRef create_request(Message *req);
+  void mark_event(TrackedOp *op, const string &evt);
+  void _mark_event(TrackedOp *op, const string &evt, utime_t now);
+
   void on_shutdown() {
     Mutex::Locker l(ops_in_flight_lock);
     history.on_shutdown();
@@ -93,6 +94,24 @@ public:
   ~OpTracker() {
     assert(ops_in_flight.empty());
   }
+
+  template <typename T, typename TRef>
+  TRef create_request(Message *ref)
+  {
+    TRef retval(new T(ref, this),
+                RemoveOnDelete(this));
+
+    if (ref->get_type() == CEPH_MSG_OSD_OP) {
+      retval->reqid = static_cast<MOSDOp*>(ref)->get_reqid();
+    } else if (ref->get_type() == MSG_OSD_SUBOP) {
+      retval->reqid = static_cast<MOSDSubOp*>(ref)->reqid;
+    }
+    _mark_event(retval.get(), "header_read", ref->get_recv_stamp());
+    _mark_event(retval.get(), "throttled", ref->get_throttle_stamp());
+    _mark_event(retval.get(), "all_read", ref->get_recv_complete_stamp());
+    _mark_event(retval.get(), "dispatched", ref->get_dispatch_stamp());
+    return retval;
+  }
 };
 
 /**
@@ -102,8 +121,6 @@ public:
 struct OpRequest : public TrackedOp {
   friend class OpTracker;
   friend class OpHistory;
-  Message *request;
-  xlist<OpRequest*>::item xitem;
 
   // rmw flags
   int rmw_flags;
@@ -132,28 +149,13 @@ struct OpRequest : public TrackedOp {
   void set_class_write() { rmw_flags |= CEPH_OSD_RMW_FLAG_CLASS_WRITE; }
   void set_pg_op() { rmw_flags |= CEPH_OSD_RMW_FLAG_PGOP; }
 
-  utime_t received_time;
-  uint8_t warn_interval_multiplier;
-  utime_t get_arrived() const {
-    return received_time;
-  }
-  double get_duration() const {
-    return events.size() ?
-      (events.rbegin()->first - received_time) :
-      0.0;
-  }
-
   void dump(utime_t now, Formatter *f) const;
 
 private:
-  list<pair<utime_t, string> > events;
-  string current;
-  Mutex lock;
   OpTracker *tracker;
   osd_reqid_t reqid;
   uint8_t hit_flag_points;
   uint8_t latest_flag_point;
-  uint64_t seq;
   static const uint8_t flag_queued_for_pg=1 << 0;
   static const uint8_t flag_reached_pg =  1 << 1;
   static const uint8_t flag_delayed =     1 << 2;
@@ -162,6 +164,7 @@ private:
   static const uint8_t flag_commit_sent = 1 << 5;
 
   OpRequest(Message *req, OpTracker *tracker);
+
 public:
   ~OpRequest() {
     assert(request);
@@ -237,4 +240,7 @@ public:
   }
 };
 
+typedef std::tr1::shared_ptr<OpRequest> OpRequestRef;
+
+
 #endif /* OPREQUEST_H_ */