mon: add new get_bl_[sn|ss]_safe functions

These functions are like the non-safe versions, but assert that
there were no disk errors and have void return types. Change a
bunch of callers who weren't checking the return code to use
these variants instead.
(Unfortunately we can't make them default safe because several of
the callers depend on getting back the length, and are perfectly happy
with ENOENT producing a 0 return value.)

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

8 files changed, 35 insertions, 19 deletions

diff --git a/src/ceph_mon.cc b/src/ceph_mon.cc
index da4529f2765..eee048027aa 100644
--- a/src/ceph_mon.cc
+++ b/src/ceph_mon.cc
@@ -284,14 +284,14 @@ int main(int argc, const char **argv)
   {
     bufferlist mapbl;
     bufferlist latest;
-    store.get_bl_ss(latest, "monmap", "latest");
+    store.get_bl_ss_safe(latest, "monmap", "latest");
     if (latest.length() > 0) {
       bufferlist::iterator p = latest.begin();
       version_t v;
       ::decode(v, p);
       ::decode(mapbl, p);
     } else {
-      store.get_bl_ss(mapbl, "mkfs", "monmap");
+      store.get_bl_ss_safe(mapbl, "mkfs", "monmap");
       if (mapbl.length() == 0) {
 	cerr << "mon fs missing 'monmap/latest' and 'mkfs/monmap'" << std::endl;
 	exit(1);
diff --git a/src/mon/AuthMonitor.cc b/src/mon/AuthMonitor.cc
index c43d7e988db..aec2ba04d86 100644
--- a/src/mon/AuthMonitor.cc
+++ b/src/mon/AuthMonitor.cc
@@ -96,7 +96,7 @@ void AuthMonitor::create_initial()
 
   KeyRing keyring;
   bufferlist bl;
-  mon->store->get_bl_ss(bl, "mkfs", "keyring");
+  mon->store->get_bl_ss_safe(bl, "mkfs", "keyring");
   bufferlist::iterator p = bl.begin();
   ::decode(keyring, p);
 
diff --git a/src/mon/MDSMonitor.cc b/src/mon/MDSMonitor.cc
index dd256b9b14a..72168ac9638 100644
--- a/src/mon/MDSMonitor.cc
+++ b/src/mon/MDSMonitor.cc
@@ -555,7 +555,7 @@ bool MDSMonitor::preprocess_command(MMonCommand *m)
       MDSMap *p = &mdsmap;
       if (epoch) {
 	bufferlist b;
-	mon->store->get_bl_sn(b, "mdsmap", epoch);
+	mon->store->get_bl_sn_safe(b, "mdsmap", epoch);
 	if (!b.length()) {
 	  p = 0;
 	  r = -ENOENT;
@@ -597,7 +597,7 @@ bool MDSMonitor::preprocess_command(MMonCommand *m)
 	}
 	epoch_t e = l;
 	bufferlist b;
-	mon->store->get_bl_sn(b,"mdsmap",e);
+	mon->store->get_bl_sn_safe(b,"mdsmap",e);
 	if (!b.length()) {
 	  r = -ENOENT;
 	} else {
diff --git a/src/mon/Monitor.cc b/src/mon/Monitor.cc
index 79f257783e7..14fe963cec6 100644
--- a/src/mon/Monitor.cc
+++ b/src/mon/Monitor.cc
@@ -362,7 +362,7 @@ int Monitor::check_features(MonitorStore *store)
   CompatSet ondisk;
 
   bufferlist features;
-  store->get_bl_ss(features, COMPAT_SET_LOC, 0);
+  store->get_bl_ss_safe(features, COMPAT_SET_LOC, 0);
   if (features.length() == 0) {
     generic_dout(0) << "WARNING: mon fs missing feature list.\n"
 	    << "Assuming it is old-style and introducing one." << dendl;
@@ -391,7 +391,7 @@ int Monitor::check_features(MonitorStore *store)
 void Monitor::read_features()
 {
   bufferlist bl;
-  store->get_bl_ss(bl, COMPAT_SET_LOC, 0);
+  store->get_bl_ss_safe(bl, COMPAT_SET_LOC, 0);
   assert(bl.length());
 
   bufferlist::iterator p = bl.begin();
@@ -497,7 +497,7 @@ int Monitor::preinit()
   if (authmon()->paxos->get_version() == 0) {
     dout(10) << "loading initial keyring to bootstrap authentication for mkfs" << dendl;
     bufferlist bl;
-    store->get_bl_ss(bl, "mkfs", "keyring");
+    store->get_bl_ss_safe(bl, "mkfs", "keyring");
     KeyRing keyring;
     bufferlist::iterator p = bl.begin();
     ::decode(keyring, p);
@@ -1032,12 +1032,14 @@ MMonProbe *Monitor::fill_probe_data(MMonProbe *m, Paxos *pax)
   version_t v = MAX(pax->get_first_committed(), m->newest_version + 1);
   int len = 0;
   for (; v <= pax->get_version(); v++) {
-    len += store->get_bl_sn(r->paxos_values[m->machine_name][v], m->machine_name.c_str(), v);
+    store->get_bl_sn_safe(r->paxos_values[m->machine_name][v], m->machine_name.c_str(), v);
+    len += r->paxos_values[m->machine_name][v].length();
     r->gv[m->machine_name][v] = store->get_global_version(m->machine_name.c_str(), v);
     for (list<string>::iterator p = pax->extra_state_dirs.begin();
          p != pax->extra_state_dirs.end();
          ++p) {
-      len += store->get_bl_sn(r->paxos_values[*p][v], p->c_str(), v);      
+      store->get_bl_sn_safe(r->paxos_values[*p][v], p->c_str(), v);
+      len += r->paxos_values[*p][v].length();
     }
     if (len >= g_conf->mon_slurp_bytes)
       break;
diff --git a/src/mon/MonitorStore.h b/src/mon/MonitorStore.h
index cf748360456..2f2426e587b 100644
--- a/src/mon/MonitorStore.h
+++ b/src/mon/MonitorStore.h
@@ -20,6 +20,7 @@
 
 #include <iosfwd>
 #include <string.h>
+#include <errno.h>
 
 class MonitorStore {
   string dir;
@@ -49,6 +50,10 @@ public:
   // ss and sn varieties.
   bool exists_bl_ss(const char *a, const char *b=0);
   int get_bl_ss(bufferlist& bl, const char *a, const char *b);
+  void get_bl_ss_safe(bufferlist& bl, const char *a, const char *b) {
+    int ret = get_bl_ss(bl, a, b);
+    assert (ret >= 0 || ret == -ENOENT);
+  }
   int put_bl_ss(bufferlist& bl, const char *a, const char *b) {
     return write_bl_ss(bl, a, b, false);
   }
@@ -65,6 +70,10 @@ public:
     snprintf(bs, sizeof(bs), "%llu", (unsigned long long)b);
     return get_bl_ss(bl, a, bs);
   }
+  void get_bl_sn_safe(bufferlist& bl, const char *a, version_t b) {
+    int ret = get_bl_sn(bl, a, b);
+    assert(ret >= 0 || ret == -ENOENT);
+  }
   int put_bl_sn(bufferlist& bl, const char *a, version_t b) {
     char bs[20];
     snprintf(bs, sizeof(bs), "%llu", (unsigned long long)b);
diff --git a/src/mon/OSDMonitor.cc b/src/mon/OSDMonitor.cc
index e72929b7d43..94e08b87088 100644
--- a/src/mon/OSDMonitor.cc
+++ b/src/mon/OSDMonitor.cc
@@ -75,7 +75,7 @@ void OSDMonitor::create_initial()
   OSDMap newmap;
 
   bufferlist bl;
-  mon->store->get_bl_ss(bl, "mkfs", "osdmap");
+  mon->store->get_bl_ss_safe(bl, "mkfs", "osdmap");
   if (bl.length()) {
     newmap.decode(bl);
     newmap.set_fsid(mon->monmap->fsid);
@@ -1309,7 +1309,8 @@ void OSDMonitor::send_incremental(PaxosServiceMessage *req, epoch_t first)
   if (first < paxos->get_first_committed()) {
     first = paxos->get_first_committed();
     bufferlist bl;
-    mon->store->get_bl_sn(bl, "osdmap_full", first);
+    mon->store->get_bl_sn_safe(bl, "osdmap_full", first);
+    assert(bl.length());
     dout(20) << "send_incremental starting with base full " << first << " " << bl.length() << " bytes" << dendl;
     MOSDMap *m = new MOSDMap(osdmap.get_fsid());
     m->oldest_map = paxos->get_first_committed();
@@ -1336,7 +1337,8 @@ void OSDMonitor::send_incremental(epoch_t first, entity_inst_t& dest, bool oneti
   if (first < paxos->get_first_committed()) {
     first = paxos->get_first_committed();
     bufferlist bl;
-    mon->store->get_bl_sn(bl, "osdmap_full", first);
+    mon->store->get_bl_sn_safe(bl, "osdmap_full", first);
+    assert(bl.length());
     dout(20) << "send_incremental starting with base full " << first << " " << bl.length() << " bytes" << dendl;
     MOSDMap *m = new MOSDMap(osdmap.get_fsid());
     m->oldest_map = paxos->get_first_committed();
@@ -1701,7 +1703,7 @@ bool OSDMonitor::preprocess_command(MMonCommand *m)
       OSDMap *p = &osdmap;
       if (epoch) {
 	bufferlist b;
-	mon->store->get_bl_sn(b,"osdmap_full", epoch);
+	mon->store->get_bl_sn_safe(b,"osdmap_full", epoch);
 	if (!b.length()) {
 	  p = 0;
 	  r = -ENOENT;
diff --git a/src/mon/PGMonitor.cc b/src/mon/PGMonitor.cc
index dcf9380613b..6a6dd60b84b 100644
--- a/src/mon/PGMonitor.cc
+++ b/src/mon/PGMonitor.cc
@@ -579,7 +579,7 @@ void PGMonitor::check_osd_map(epoch_t epoch)
        e++) {
     dout(10) << "check_osd_map applying osdmap e" << e << " to pg_map" << dendl;
     bufferlist bl;
-    mon->store->get_bl_sn(bl, "osdmap", e);
+    mon->store->get_bl_sn_safe(bl, "osdmap", e);
     assert(bl.length());
     OSDMap::Incremental inc(bl);
     for (map<int32_t,uint32_t>::iterator p = inc.new_weight.begin();
diff --git a/src/mon/Paxos.cc b/src/mon/Paxos.cc
index 349908789e3..2c16e9b2849 100644
--- a/src/mon/Paxos.cc
+++ b/src/mon/Paxos.cc
@@ -73,7 +73,8 @@ void Paxos::collect(version_t oldpn)
   if (mon->store->exists_bl_sn(machine_name, last_committed+1)) {
     uncommitted_v = last_committed+1;
     uncommitted_pn = accepted_pn;
-    mon->store->get_bl_sn(uncommitted_value, machine_name, last_committed+1);
+    mon->store->get_bl_sn_safe(uncommitted_value, machine_name, last_committed+1);
+    assert(uncommitted_value.length());
     dout(10) << "learned uncommitted " << (last_committed+1)
 	     << " (" << uncommitted_value.length() << " bytes) from myself" 
 	     << dendl;
@@ -146,7 +147,7 @@ void Paxos::handle_collect(MMonPaxos *collect)
   //  (it'll be at last_committed+1)
   bufferlist bl;
   if (mon->store->exists_bl_sn(machine_name, last_committed+1)) {
-    mon->store->get_bl_sn(bl, machine_name, last_committed+1);
+    mon->store->get_bl_sn_safe(bl, machine_name, last_committed+1);
     assert(bl.length() > 0);
     dout(10) << " sharing our accepted but uncommitted value for " 
 	     << last_committed+1 << " (" << bl.length() << " bytes)" << dendl;
@@ -184,7 +185,8 @@ void Paxos::share_state(MMonPaxos *m, version_t peer_first_committed,
   // include incrementals
   for ( ; v <= last_committed; v++) {
     if (mon->store->exists_bl_sn(machine_name, v)) {
-      mon->store->get_bl_sn(m->values[v], machine_name, v);
+      mon->store->get_bl_sn_safe(m->values[v], machine_name, v);
+      assert(m->values[v].length());
       m->gv[v] = mon->store->get_global_version(machine_name, v);
       dout(10) << " sharing " << v << " (" 
 	       << m->values[v].length() << " bytes)" << dendl;
@@ -1051,7 +1053,8 @@ void Paxos::stash_latest(version_t v, bufferlist& bl)
 version_t Paxos::get_stashed(bufferlist& bl)
 {
   bufferlist full;
-  if (mon->store->get_bl_ss(full, machine_name, "latest") <= 0) {
+  mon->store->get_bl_ss_safe(full, machine_name, "latest");
+  if (!full.length()) {
     dout(10) << "get_stashed not found" << dendl;
     return 0;
   }