Merge branch 'wip-osdmap'

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

src/crush/CrushWrapper.cc

@@ -172,6 +172,12 @@ int CrushWrapper::insert_item(CephContext *cct, int item, float weight, string n
 
     // now that we've added the (0-weighted) item and any parent buckets, adjust the weight.
     adjust_item_weightf(cct, item, weight);
+
+    if (item >= crush->max_devices) {
+      crush->max_devices = item + 1;
+      ldout(cct, 5) << "insert_item max_devices now " << crush->max_devices << dendl;
+    }
+
     return 0;
   }
 

src/crush/CrushWrapper.h

@@ -471,7 +471,7 @@ public:
   void do_rule(int rule, int x, vector<int>& out, int maxout,
 	       const vector<__u32>& weight) const {
     int rawout[maxout];
-    int numrep = crush_do_rule(crush, rule, x, rawout, maxout, &weight[0]);
+    int numrep = crush_do_rule(crush, rule, x, rawout, maxout, &weight[0], weight.size());
     if (numrep < 0)
       numrep = 0;
     out.resize(numrep);

src/crush/mapper.c

@@ -263,8 +263,10 @@ static int crush_bucket_choose(struct crush_bucket *in, int x, int r)
  * true if device is marked "out" (failed, fully offloaded)
  * of the cluster
  */
-static int is_out(const struct crush_map *map, const __u32 *weight, int item, int x)
+static int is_out(const struct crush_map *map, const __u32 *weight, int weight_max, int item, int x)
 {
+	if (item >= weight_max)
+		return 1;
 	if (weight[item] >= 0x10000)
 		return 0;
 	if (weight[item] == 0)
@@ -290,7 +292,7 @@ static int is_out(const struct crush_map *map, const __u32 *weight, int item, in
  */
 static int crush_choose(const struct crush_map *map,
 			struct crush_bucket *bucket,
-			const __u32 *weight,
+			const __u32 *weight, int weight_max,
 			int x, int numrep, int type,
 			int *out, int outpos,
 			int firstn, int recurse_to_leaf,
@@ -394,7 +396,7 @@ static int crush_choose(const struct crush_map *map,
 					if (item < 0) {
 						if (crush_choose(map,
 							 map->buckets[-1-item],
-							 weight,
+							 weight, weight_max,
 							 x, outpos+1, 0,
 							 out2, outpos,
 							 firstn, 0,
@@ -410,7 +412,7 @@ static int crush_choose(const struct crush_map *map,
 				if (!reject) {
 					/* out? */
 					if (itemtype == 0)
-						reject = is_out(map, weight,
+						reject = is_out(map, weight, weight_max,
 								item, x);
 					else
 						reject = 0;
@@ -466,7 +468,7 @@ reject:
  */
 int crush_do_rule(const struct crush_map *map,
 		  int ruleno, int x, int *result, int result_max,
-		  const __u32 *weight)
+		  const __u32 *weight, int weight_max)
 {
 	int result_len;
 	int a[CRUSH_MAX_SET];
@@ -537,7 +539,7 @@ int crush_do_rule(const struct crush_map *map,
 				j = 0;
 				osize += crush_choose(map,
 						      map->buckets[-1-w[i]],
-						      weight,
+						      weight, weight_max,
 						      x, numrep,
 						      curstep->arg2,
 						      o+osize, j,

src/crush/mapper.h

@@ -14,6 +14,6 @@ extern int crush_find_rule(const struct crush_map *map, int ruleset, int type, i
 extern int crush_do_rule(const struct crush_map *map,
 			 int ruleno,
 			 int x, int *result, int result_max,
-			 const __u32 *weights);
+			 const __u32 *weights, int weight_max);
 
 #endif

src/mon/OSDMonitor.cc

@@ -1757,8 +1757,14 @@ bool OSDMonitor::prepare_command(MMonCommand *m)
     }
     else if (m->cmd.size() >= 6 && m->cmd[1] == "crush" && m->cmd[2] == "set") {
       do {
-	// osd crush update <id> <name> <weight> [<loc1> [<loc2> ...]]
+	// osd crush set <id> <name> <weight> [<loc1> [<loc2> ...]]
 	int id = atoi(m->cmd[3].c_str());
+	if (!osdmap.exists(id)) {
+	  err = -ENOENT;
+	  ss << "osd." << id << " does not exist.  create it before updating the crush map";
+	  goto out;
+	}
+
 	string name = m->cmd[4];
 	float weight = atof(m->cmd[5].c_str());
 	map<string,string> loc;

src/osd/OSDMap.cc

@@ -816,6 +816,22 @@ ceph_object_layout OSDMap::make_object_layout(object_t oid, int pg_pool) const
   return ol;
 }
 
+void OSDMap::_remove_nonexistent_osds(vector<int>& osds) const
+{
+  unsigned removed = 0;
+  for (unsigned i = 0; i < osds.size(); i++) {
+    if (!exists(osds[i])) {
+      removed++;
+      continue;
+    }
+    if (removed) {
+      osds[i - removed] = osds[i];
+    }
+  }
+  if (removed)
+    osds.resize(osds.size() - removed);
+}
+
 int OSDMap::_pg_to_osds(const pg_pool_t& pool, pg_t pg, vector<int>& osds) const
 {
   // map to osds[]
@@ -826,7 +842,9 @@ int OSDMap::_pg_to_osds(const pg_pool_t& pool, pg_t pg, vector<int>& osds) const
   int ruleno = crush->find_rule(pool.get_crush_ruleset(), pool.get_type(), size);
   if (ruleno >= 0)
     crush->do_rule(ruleno, pps, osds, size, osd_weight);
-  
+
+  _remove_nonexistent_osds(osds);
+
   return osds.size();
 }
 

src/osd/OSDMap.h

@@ -405,6 +405,7 @@ public:
 private:
   /// pg -> (raw osd list)
   int _pg_to_osds(const pg_pool_t& pool, pg_t pg, vector<int>& osds) const;
+  void _remove_nonexistent_osds(vector<int>& osds) const;
 
   /// pg -> (up osd list)
   void _raw_to_up_osds(pg_t pg, vector<int>& raw, vector<int>& up) const;