osd: allow extra reqids to be stashed in each pg_log_entry_t

Signed-off-by: Sage Weil <sage@redhat.com>

diff --git a/src/include/unordered_map.h b/src/include/unordered_map.h
index adcb89ad5557d5699f3efda2c8345d033beb9841..30b09149a8132c2422bbf92bb08a5e9e53796b1c 100644 (file)
--- a/src/include/unordered_map.h
+++ b/src/include/unordered_map.h
@@ -9,6 +9,7 @@
 
 namespace ceph {
   using std::unordered_map;
+  using std::unordered_multimap;
 }
 
 #else
@@ -17,6 +18,7 @@ namespace ceph {
 
 namespace ceph {
   using std::tr1::unordered_map;
+  using std::tr1::unordered_multimap;
 }
 
 #endif

diff --git a/src/osd/PGLog.h b/src/osd/PGLog.h
index 1961ec5fc8d252b061e4c0985056674e13ac97cd..55e90ec8775035c296b5b64d987112a1840243fe 100644 (file)
--- a/src/osd/PGLog.h
+++ b/src/osd/PGLog.h
@@ -57,6 +57,7 @@ struct PGLog {
   struct IndexedLog : public pg_log_t {
     ceph::unordered_map<hobject_t,pg_log_entry_t*> objects;  // ptrs into log.  be careful!
     ceph::unordered_map<osd_reqid_t,pg_log_entry_t*> caller_ops;
+    ceph::unordered_multimap<osd_reqid_t,pg_log_entry_t*> extra_caller_ops;
 
     // recovery pointers
     list<pg_log_entry_t>::iterator complete_to;  // not inclusive of referenced item
@@ -123,18 +124,25 @@ struct PGLog {
       return objects.count(oid);
     }
     bool logged_req(const osd_reqid_t &r) const {
-      return caller_ops.count(r);
+      return caller_ops.count(r) || extra_caller_ops.count(r);
     }
     const pg_log_entry_t *get_request(const osd_reqid_t &r) const {
-      ceph::unordered_map<osd_reqid_t,pg_log_entry_t*>::const_iterator p = caller_ops.find(r);
-      if (p == caller_ops.end())
-       return NULL;
-      return p->second;
+      ceph::unordered_map<osd_reqid_t,pg_log_entry_t*>::const_iterator p;
+      p = caller_ops.find(r);
+      if (p != caller_ops.end())
+       return p->second;
+      // warning: we will return *a* request for this reqid, but not
+      // necessarily the most recent.
+      p = extra_caller_ops.find(r);
+      if (p != extra_caller_ops.end())
+       return p->second;
+      return NULL;
     }
 
     void index() {
       objects.clear();
       caller_ops.clear();
+      extra_caller_ops.clear();
       for (list<pg_log_entry_t>::iterator i = log.begin();
            i != log.end();
            ++i) {
@@ -142,6 +150,11 @@ struct PGLog {
        if (i->reqid_is_indexed()) {
          //assert(caller_ops.count(i->reqid) == 0);  // divergent merge_log indexes new before unindexing old
          caller_ops[i->reqid] = &(*i);
+         for (vector<osd_reqid_t>::const_iterator j = i->extra_reqids.begin();
+              j != i->extra_reqids.end();
+              ++j) {
+           extra_caller_ops.insert(make_pair(*j, &(*i)));
+         }
        }
       }
 
@@ -158,20 +171,40 @@ struct PGLog {
       if (e.reqid_is_indexed()) {
        //assert(caller_ops.count(i->reqid) == 0);  // divergent merge_log indexes new before unindexing old
        caller_ops[e.reqid] = &e;
+       for (vector<osd_reqid_t>::const_iterator j = e.extra_reqids.begin();
+            j != e.extra_reqids.end();
+            ++j) {
+         extra_caller_ops.insert(make_pair(*j, &e));
+       }
       }
     }
     void unindex() {
       objects.clear();
       caller_ops.clear();
+      extra_caller_ops.clear();
     }
     void unindex(pg_log_entry_t& e) {
       // NOTE: this only works if we remove from the _tail_ of the log!
       if (objects.count(e.soid) && objects[e.soid]->version == e.version)
         objects.erase(e.soid);
-      if (e.reqid_is_indexed() &&
-         caller_ops.count(e.reqid) &&  // divergent merge_log indexes new before unindexing old
-         caller_ops[e.reqid] == &e)
-       caller_ops.erase(e.reqid);
+      if (e.reqid_is_indexed()) {
+       if (caller_ops.count(e.reqid) &&  // divergent merge_log indexes new before unindexing old
+           caller_ops[e.reqid] == &e)
+         caller_ops.erase(e.reqid);
+       for (vector<osd_reqid_t>::const_iterator j = e.extra_reqids.begin();
+            j != e.extra_reqids.end();
+            ++j) {
+         for (ceph::unordered_multimap<osd_reqid_t,pg_log_entry_t*>::iterator k =
+                extra_caller_ops.find(*j);
+              k != extra_caller_ops.end() && k->first == *j;
+              ++j) {
+           if (k->second == &e) {
+             extra_caller_ops.erase(k);
+             break;
+           }
+         }
+       }
+      }
     }
 
     // actors
@@ -189,8 +222,14 @@ struct PGLog {
 
       // to our index
       objects[e.soid] = &(log.back());
-      if (e.reqid_is_indexed())
+      if (e.reqid_is_indexed()) {
        caller_ops[e.reqid] = &(log.back());
+       for (vector<osd_reqid_t>::const_iterator j = e.extra_reqids.begin();
+            j != e.extra_reqids.end();
+            ++j) {
+         extra_caller_ops.insert(make_pair(*j, &(log.back())));
+       }
+      }
     }
 
     void trim(

diff --git a/src/osd/osd_types.cc b/src/osd/osd_types.cc
index 2c313cdc699cdb59546bc7720814a94ccc4518a9..ab4253833c2db2033149dd29127a61bb25175639 100644 (file)
--- a/src/osd/osd_types.cc
+++ b/src/osd/osd_types.cc
@@ -2847,7 +2847,7 @@ void pg_log_entry_t::decode_with_checksum(bufferlist::iterator& p)
 
 void pg_log_entry_t::encode(bufferlist &bl) const
 {
-  ENCODE_START(9, 4, bl);
+  ENCODE_START(10, 4, bl);
   ::encode(op, bl);
   ::encode(soid, bl);
   ::encode(version, bl);
@@ -2871,12 +2871,13 @@ void pg_log_entry_t::encode(bufferlist &bl) const
   ::encode(snaps, bl);
   ::encode(user_version, bl);
   ::encode(mod_desc, bl);
+  ::encode(extra_reqids, bl);
   ENCODE_FINISH(bl);
 }
 
 void pg_log_entry_t::decode(bufferlist::iterator &bl)
 {
-  DECODE_START_LEGACY_COMPAT_LEN(8, 4, 4, bl);
+  DECODE_START_LEGACY_COMPAT_LEN(10, 4, 4, bl);
   ::decode(op, bl);
   if (struct_v < 2) {
     sobject_t old_soid;
@@ -2897,6 +2898,7 @@ void pg_log_entry_t::decode(bufferlist::iterator &bl)
     ::decode(prior_version, bl);
 
   ::decode(reqid, bl);
+
   ::decode(mtime, bl);
   if (struct_v < 5)
     invalid_pool = true;
@@ -2922,6 +2924,8 @@ void pg_log_entry_t::decode(bufferlist::iterator &bl)
     ::decode(mod_desc, bl);
   else
     mod_desc.mark_unrollbackable();
+  if (struct_v >= 10)
+    ::decode(extra_reqids, bl);
 
   DECODE_FINISH(bl);
 }
@@ -2933,6 +2937,12 @@ void pg_log_entry_t::dump(Formatter *f) const
   f->dump_stream("version") << version;
   f->dump_stream("prior_version") << prior_version;
   f->dump_stream("reqid") << reqid;
+  f->open_array_section("extra_reqids");
+  for (vector<osd_reqid_t>::const_iterator p = extra_reqids.begin();
+       p != extra_reqids.end();
+       ++p)
+    f->dump_stream("reqid") << *p;
+  f->close_section();
   f->dump_stream("mtime") << mtime;
   if (snaps.length() > 0) {
     vector<snapid_t> v;

diff --git a/src/osd/osd_types.h b/src/osd/osd_types.h
index 10728fa6c60ca3971d67b8d3233fafea32c9eeac..979e69f182c2a2eafa64c6fed3b4f9f65631af97 100644 (file)
--- a/src/osd/osd_types.h
+++ b/src/osd/osd_types.h
@@ -2144,7 +2144,9 @@ struct pg_log_entry_t {
 
   /// describes state for a locally-rollbackable entry
   ObjectModDesc mod_desc;
-      
+
+  vector<osd_reqid_t> extra_reqids;
+
   pg_log_entry_t()
     : op(0), user_version(0),
       invalid_hash(false), invalid_pool(false), offset(0) {}