crush: sum and check quantized weights for bucket

Sum the quantized weights for each bucket, and check that for overflow.

This could change the results of a compile marginally if the map is using
non-divisible weight values that quantize funny.  The old code might
calculate a bucket sum that is not the actual sum of the quantized weights.

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

4 files changed, 16 insertions, 15 deletions

diff --git a/src/crush/CrushCompiler.cc b/src/crush/CrushCompiler.cc
index 0896995a7ae..5a0cdbd0848 100644
--- a/src/crush/CrushCompiler.cc
+++ b/src/crush/CrushCompiler.cc
@@ -440,9 +440,9 @@ int CrushCompiler::parse_bucket(iter_t const& i)
   vector<int> weights(size);
 
   int curpos = 0;
-  float bucketweight = 0;
+  unsigned bucketweight = 0;
   bool have_uniform_weight = false;
-  float uniform_weight = 0;
+  unsigned uniform_weight = 0;
   for (unsigned p=3; p<i->children.size()-1; p++) {
     iter_t sub = i->children.begin() + p;
     string tag = string_node(sub->children[0]);
@@ -455,7 +455,7 @@ int CrushCompiler::parse_bucket(iter_t const& i)
       }
       int itemid = item_id[iname];
 
-      float weight = 1.0;
+      unsigned weight = 0x10000;
       if (item_weight.count(itemid))
 	weight = item_weight[itemid];
 
@@ -463,13 +463,13 @@ int CrushCompiler::parse_bucket(iter_t const& i)
       for (unsigned q = 2; q < sub->children.size(); q++) {
 	string tag = string_node(sub->children[q++]);
 	if (tag == "weight") {
-	  weight = float_node(sub->children[q]);
+	  weight = float_node(sub->children[q]) * (float)0x10000;
 	  if (weight > CRUSH_MAX_DEVICE_WEIGHT && itemid >= 0) {
-	    err << "device weight limited to " << CRUSH_MAX_DEVICE_WEIGHT << std::endl;
+	    err << "device weight limited to " << CRUSH_MAX_DEVICE_WEIGHT / 0x10000 << std::endl;
 	    return -ERANGE;
 	  }
 	  else if (weight > CRUSH_MAX_BUCKET_WEIGHT && itemid < 0) {
-	    err << "bucket weight limited to " << CRUSH_MAX_BUCKET_WEIGHT
+	    err << "bucket weight limited to " << CRUSH_MAX_BUCKET_WEIGHT / 0x10000
 	        << " to prevent overflow" << std::endl;
 	    return -ERANGE;
 	  }
@@ -487,7 +487,7 @@ int CrushCompiler::parse_bucket(iter_t const& i)
 	} else {
 	  if (uniform_weight != weight) {
 	    err << "item '" << iname << "' in uniform bucket '" << name << "' has weight " << weight
-		<< " but previous item(s) have weight " << uniform_weight
+		<< " but previous item(s) have weight " << (float)uniform_weight/(float)0x10000
 		<< "; uniform bucket items must all have identical weights." << std::endl;
 	    return -1;
 	  }
@@ -504,9 +504,9 @@ int CrushCompiler::parse_bucket(iter_t const& i)
       }
       //err << " item " << iname << " (" << itemid << ") pos " << pos << " weight " << weight << std::endl;
       items[pos] = itemid;
-      weights[pos] = (unsigned)(weight * 0x10000);
+      weights[pos] = weight;
 
-      if (crush_addition_is_unsafe((int) bucketweight, (int) weight)) {
+      if (crush_addition_is_unsafe(bucketweight, weight)) {
         err << "oh no! our bucket weights are overflowing all over the place, better lower the item weights" << std::endl;
         return -ERANGE;
       }
@@ -520,7 +520,8 @@ int CrushCompiler::parse_bucket(iter_t const& i)
     //err << "assigned id " << id << std::endl;
   }
 
-  if (verbose) err << "bucket " << name << " (" << id << ") " << size << " items and weight " << bucketweight << std::endl;
+  if (verbose) err << "bucket " << name << " (" << id << ") " << size << " items and weight "
+		   << (float)bucketweight / (float)0x10000 << std::endl;
   id_item[id] = name;
   item_id[name] = id;
   item_weight[id] = bucketweight;
diff --git a/src/crush/CrushCompiler.h b/src/crush/CrushCompiler.h
index 9d07d76afa9..d488735bb9e 100644
--- a/src/crush/CrushCompiler.h
+++ b/src/crush/CrushCompiler.h
@@ -36,7 +36,7 @@ class CrushCompiler {
 
   map<string, int> item_id;
   map<int, string> id_item;
-  map<int, float> item_weight;
+  map<int, unsigned> item_weight;
   map<string, int> type_id;
   map<string, int> rule_id;
 
diff --git a/src/crush/crush.h b/src/crush/crush.h
index 06f4bde8e62..de22386ed70 100644
--- a/src/crush/crush.h
+++ b/src/crush/crush.h
@@ -28,8 +28,8 @@
 #define CRUSH_MAX_DEPTH 10  /* max crush hierarchy depth */
 #define CRUSH_MAX_SET   10  /* max size of a mapping result */
 
-#define CRUSH_MAX_DEVICE_WEIGHT 10000
-#define CRUSH_MAX_BUCKET_WEIGHT 10000000
+#define CRUSH_MAX_DEVICE_WEIGHT (100u * 0x10000u)
+#define CRUSH_MAX_BUCKET_WEIGHT (65535u * 0x10000u)
 
 /*
  * CRUSH uses user-defined "rules" to describe how inputs should be
diff --git a/src/crushtool.cc b/src/crushtool.cc
index dd1484d6c2f..d628f6e1e96 100644
--- a/src/crushtool.cc
+++ b/src/crushtool.cc
@@ -370,7 +370,7 @@ int main(int argc, const char **argv)
   }
 
   if (decompile) {
-    CrushCompiler cc(crush, cerr);
+    CrushCompiler cc(crush, cerr, (int)verbose);
     if (!outfn.empty()) {
       ofstream o;
       o.open(outfn.c_str(), ios::out | ios::binary | ios::trunc);
@@ -395,7 +395,7 @@ int main(int argc, const char **argv)
       return -ENOENT;
     }
 
-    CrushCompiler cc(crush, cerr);
+    CrushCompiler cc(crush, cerr, (int)verbose);
     if (unsafe_tunables)
       cc.enable_unsafe_tunables();
     int r = cc.compile(in, srcfn.c_str());