ReplicatedPG: fix pgls listing, add max listing size

Previously, a client requesting a large pgls could tie up the
osd for an unacceptable amount of time.  Also, it's possible
for the osd to return less than the requested number of
entries anyway, so we now return 1 when we have completed the
listing.

Signed-off-by: Samuel Just <sam.just@inktank.com>
Reviewed-by: Sage Weil <sage@inktank.com>

diff --git a/src/common/config_opts.h b/src/common/config_opts.h
index 418436de18561d40e5676cba20452951fb0513bd..7144ef4427b5b976052e48248c2a4d88b83dffea 100644 (file)
--- a/src/common/config_opts.h
+++ b/src/common/config_opts.h
@@ -256,6 +256,7 @@ OPTION(osd_data, OPT_STR, "/var/lib/ceph/osd/$cluster-$id")
 OPTION(osd_journal, OPT_STR, "/var/lib/ceph/osd/$cluster-$id/journal")
 OPTION(osd_journal_size, OPT_INT, 0)         // in mb
 OPTION(osd_max_write_size, OPT_INT, 90)
+OPTION(osd_max_pgls, OPT_U64, 1024) // max number of pgls entries to return
 OPTION(osd_balance_reads, OPT_BOOL, false)
 OPTION(osd_shed_reads, OPT_INT, false)     // forward from primary to replica
 OPTION(osd_shed_reads_min_latency, OPT_DOUBLE, .01)       // min local latency

diff --git a/src/osd/ReplicatedPG.cc b/src/osd/ReplicatedPG.cc
index 2fe4246255f00e517ed0d72ec46e32cec4d74759..beeb85a944890066e458886a0452ad476f36f8f2 100644 (file)
--- a/src/osd/ReplicatedPG.cc
+++ b/src/osd/ReplicatedPG.cc
@@ -440,7 +440,9 @@ void ReplicatedPG::do_pg_op(OpRequestRef op)
         dout(10) << " pgls pg=" << m->get_pg() << " != " << info.pgid << dendl;
        result = 0; // hmm?
       } else {
-        dout(10) << " pgls pg=" << m->get_pg() << " count " << p->op.pgls.count << dendl;
+       unsigned list_size = MIN(g_conf->osd_max_pgls, p->op.pgls.count);
+
+        dout(10) << " pgls pg=" << m->get_pg() << " count " << list_size << dendl;
        // read into a buffer
         vector<hobject_t> sentries;
         pg_ls_response_t response;
@@ -457,8 +459,8 @@ void ReplicatedPG::do_pg_op(OpRequestRef op)
        hobject_t current = response.handle;
        osr.flush();
        int r = osd->store->collection_list_partial(coll, current,
-                                                   p->op.pgls.count,
-                                                   p->op.pgls.count,
+                                                   list_size,
+                                                   list_size,
                                                    snapid,
                                                    &sentries,
                                                    &next);
@@ -484,8 +486,9 @@ void ReplicatedPG::do_pg_op(OpRequestRef op)
            candidate = (missing_iter++)->first;
          }
 
-         if (response.entries.size() == p->op.pgls.count) {
+         if (response.entries.size() == list_size) {
            next = candidate;
+           break;
          }
 
          // skip snapdir objects
@@ -520,6 +523,11 @@ void ReplicatedPG::do_pg_op(OpRequestRef op)
          response.entries.push_back(make_pair(candidate.oid,
                                               candidate.get_key()));
        }
+       if (next.is_max() &&
+           missing_iter == missing.missing.end() &&
+           ls_iter == sentries.end()) {
+         result = 1;
+       }
        response.handle = next;
        ::encode(response, outdata);
        if (filter)