From: Sage Weil <sage@redhat.com>
Date: Fri, 20 Jan 2017 03:09:35 +0000 (-0600)
Subject: osdc/Objecter: infer ptruncated on old OSDs via max_entries
X-Git-Tag: v12.0.0~121^2~6
X-Git-Url: http://git-server-git.apps.pok.os.sepia.ceph.com/?a=commitdiff_plain;h=cfa6f7932ed8bd3f7c861991b40cffa12eb20fb4;p=ceph.git

osdc/Objecter: infer ptruncated on old OSDs via max_entries

If we do not get an explicit 'more' value from the OSD, infer it by
checking whether we got the max requested entries.  On old OSDs, which
don't enforce a limit, this will work.  On new OSDs, we will get the
explicit result.

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

diff --git a/src/osdc/Objecter.h b/src/osdc/Objecter.h
index b8131fabdce4..96af080e1b0a 100644
--- a/src/osdc/Objecter.h
+++ b/src/osdc/Objecter.h
@@ -391,13 +391,14 @@ struct ObjectOperation {
     out_rval[p] = prval;
   }
   struct C_ObjectOperation_decodevals : public Context {
+    uint64_t max_entries;
     bufferlist bl;
     std::map<std::string,bufferlist> *pattrs;
     bool *ptruncated;
     int *prval;
-    C_ObjectOperation_decodevals(std::map<std::string,bufferlist> *pa,
+    C_ObjectOperation_decodevals(uint64_t m, std::map<std::string,bufferlist> *pa,
 				 bool *pt, int *pr)
-      : pattrs(pa), ptruncated(pt), prval(pr) {
+      : max_entries(m), pattrs(pa), ptruncated(pt), prval(pr) {
       if (ptruncated) {
 	*ptruncated = false;
       }
@@ -409,12 +410,18 @@ struct ObjectOperation {
 	  if (pattrs)
 	    ::decode(*pattrs, p);
 	  if (ptruncated) {
+	    std::map<std::string,bufferlist> ignore;
 	    if (!pattrs) {
-	      std::map<std::string,bufferlist> ignore;
 	      ::decode(ignore, p);
+	      pattrs = &ignore;
 	    }
 	    if (!p.end()) {
 	      ::decode(*ptruncated, p);
+	    } else {
+	      // the OSD did not provide this.  since old OSDs do not
+	      // enfoce omap result limits either, we can infer it from
+	      // the size of the result
+	      *ptruncated = (pattrs->size() == max_entries);
 	    }
 	  }
 	}
@@ -426,13 +433,14 @@ struct ObjectOperation {
     }
   };
   struct C_ObjectOperation_decodekeys : public Context {
+    uint64_t max_entries;
     bufferlist bl;
     std::set<std::string> *pattrs;
     bool *ptruncated;
     int *prval;
-    C_ObjectOperation_decodekeys(std::set<std::string> *pa, bool *pt,
+    C_ObjectOperation_decodekeys(uint64_t m, std::set<std::string> *pa, bool *pt,
 				 int *pr)
-      : pattrs(pa), ptruncated(pt), prval(pr) {
+      : max_entries(m), pattrs(pa), ptruncated(pt), prval(pr) {
       if (ptruncated) {
 	*ptruncated = false;
       }
@@ -444,12 +452,18 @@ struct ObjectOperation {
 	  if (pattrs)
 	    ::decode(*pattrs, p);
 	  if (ptruncated) {
+	    std::set<std::string> ignore;
 	    if (!pattrs) {
-	      std::set<std::string> ignore;
 	      ::decode(ignore, p);
+	      pattrs = &ignore;
 	    }
 	    if (!p.end()) {
 	      ::decode(*ptruncated, p);
+	    } else {
+	      // the OSD did not provide this.  since old OSDs do not
+	      // enfoce omap result limits either, we can infer it from
+	      // the size of the result
+	      *ptruncated = (pattrs->size() == max_entries);
 	    }
 	  }
 	}
@@ -535,7 +549,7 @@ struct ObjectOperation {
     if (pattrs || prval) {
       unsigned p = ops.size() - 1;
       C_ObjectOperation_decodevals *h
-	= new C_ObjectOperation_decodevals(pattrs, nullptr, prval);
+	= new C_ObjectOperation_decodevals(0, pattrs, nullptr, prval);
       out_handler[p] = h;
       out_bl[p] = &h->bl;
       out_rval[p] = prval;
@@ -606,7 +620,7 @@ struct ObjectOperation {
     if (prval || ptruncated || out_set) {
       unsigned p = ops.size() - 1;
       C_ObjectOperation_decodekeys *h =
-	new C_ObjectOperation_decodekeys(out_set, ptruncated, prval);
+	new C_ObjectOperation_decodekeys(max_to_get, out_set, ptruncated, prval);
       out_handler[p] = h;
       out_bl[p] = &h->bl;
       out_rval[p] = prval;
@@ -630,7 +644,7 @@ struct ObjectOperation {
     if (prval || out_set || ptruncated) {
       unsigned p = ops.size() - 1;
       C_ObjectOperation_decodevals *h =
-	new C_ObjectOperation_decodevals(out_set, ptruncated, prval);
+	new C_ObjectOperation_decodevals(max_to_get, out_set, ptruncated, prval);
       out_handler[p] = h;
       out_bl[p] = &h->bl;
       out_rval[p] = prval;
@@ -649,7 +663,7 @@ struct ObjectOperation {
     if (prval || out_set) {
       unsigned p = ops.size() - 1;
       C_ObjectOperation_decodevals *h =
-	new C_ObjectOperation_decodevals(out_set, nullptr, prval);
+	new C_ObjectOperation_decodevals(0, out_set, nullptr, prval);
       out_handler[p] = h;
       out_bl[p] = &h->bl;
       out_rval[p] = prval;