crush: add remove_item_under()

Remove only instances of item nested beneath a particular ancestor.

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

diff --git a/src/crush/CrushWrapper.cc b/src/crush/CrushWrapper.cc
index bb23901223612cb4b2a53c3dbe39ec31c5b0bd59..e5af5405f72eb16c69387b416c9788752ba9ce58 100644 (file)
--- a/src/crush/CrushWrapper.cc
+++ b/src/crush/CrushWrapper.cc
@@ -58,13 +58,15 @@ int CrushWrapper::remove_item(CephContext *cct, int item)
        if (item < 0) {
          crush_bucket *t = get_bucket(item);
          if (t && t->size) {
-           ldout(cct, 1) << "remove_device bucket " << item << " has " << t->size << " items, not empty" << dendl;
+           ldout(cct, 1) << "remove_item bucket " << item << " has " << t->size
+                         << " items, not empty" << dendl;
            return -ENOTEMPTY;
          }
          was_bucket = t;
        }
        adjust_item_weight(cct, item, 0);
-       ldout(cct, 5) << "remove_device removing item " << item << " from bucket " << b->id << dendl;
+       ldout(cct, 5) << "remove_item removing item " << item
+                     << " from bucket " << b->id << dendl;
        crush_bucket_remove_item(b, item);
        ret = 0;
       }
@@ -72,7 +74,7 @@ int CrushWrapper::remove_item(CephContext *cct, int item)
   }
 
   if (was_bucket) {
-    ldout(cct, 5) << "remove_device removing bucket " << item << dendl;
+    ldout(cct, 5) << "remove_item removing bucket " << item << dendl;
     crush_remove_bucket(crush, was_bucket);
   }
   if (item >= 0 && name_map.count(item)) {
@@ -84,6 +86,79 @@ int CrushWrapper::remove_item(CephContext *cct, int item)
   return ret;
 }
 
+bool CrushWrapper::_search_item_exists(int item)
+{
+  for (int i = 0; i < crush->max_buckets; i++) {
+    if (!crush->buckets[i])
+      continue;
+    crush_bucket *b = crush->buckets[i];
+    for (unsigned i=0; i<b->size; ++i) {
+      if (b->items[i] == item)
+       return true;
+    }
+  }
+  return false;
+}
+
+int CrushWrapper::_remove_item_under(CephContext *cct, int item, int ancestor)
+{
+  ldout(cct, 5) << "remove_item_under " << item << " under " << ancestor << dendl;
+
+  if (ancestor >= 0) {
+    return -EINVAL;
+  }
+
+  if (!bucket_exists(ancestor))
+    return -EINVAL;
+
+  int ret = -ENOENT;
+
+  crush_bucket *b = get_bucket(ancestor);
+  for (unsigned i=0; i<b->size; ++i) {
+    int id = b->items[i];
+    if (id == item) {
+      if (item < 0) {
+       crush_bucket *t = get_bucket(item);
+       if (t && t->size) {
+         ldout(cct, 1) << "remove_item_under bucket " << item << " has " << t->size << " items, not empty" << dendl;
+         return -ENOTEMPTY;
+       }
+      }
+      adjust_item_weight(cct, item, 0);
+      ldout(cct, 5) << "remove_item_under removing item " << item << " from bucket " << b->id << dendl;
+      crush_bucket_remove_item(b, item);
+      ret = 0;
+    } else if (id < 0) {
+      int r = remove_item_under(cct, item, id);
+      if (r == 0)
+       ret = 0;
+    }
+  }
+  return ret;
+}
+
+int CrushWrapper::remove_item_under(CephContext *cct, int item, int ancestor)
+{
+  int ret = _remove_item_under(cct, item, ancestor);
+  if (ret < 0)
+    return ret;
+
+  // last instance?
+  if (!_search_item_exists(item)) {
+    if (item < 0) {
+      ldout(cct, 5) << "remove_item removing bucket " << item << dendl;
+      crush_bucket *t = get_bucket(item);
+      crush_remove_bucket(crush, t);
+    }
+    if (item >= 0 && name_map.count(item)) {
+      name_map.erase(item);
+      have_rmaps = false;
+    }
+  }
+
+  return ret;
+}
+
 bool CrushWrapper::check_item_loc(CephContext *cct, int item, const map<string,string>& loc,
                                  int *weight)
 {

diff --git a/src/crush/CrushWrapper.h b/src/crush/CrushWrapper.h
index 6a12d95ff16cf21c5adf85ad89dc4ee6df8a410b..048eb3dee5cf07a0071ecc39e0ece53989731281 100644 (file)
--- a/src/crush/CrushWrapper.h
+++ b/src/crush/CrushWrapper.h
@@ -241,6 +241,16 @@ public:
    */
   bool subtree_contains(int root, int item) const;
 
+private:
+  /**
+   * search for an item in any bucket
+   *
+   * @param i item
+   * @return true if present
+   */
+  bool _search_item_exists(int i);
+public:
+
   /**
    * see if item is located where we think it is
    *
@@ -395,7 +405,7 @@ public:
                          const map<string,string>& loc);
 
   /**
-   * remove an item from the map
+   * remove all instances of an item from the map
    *
    * @param cct cct
    * @param id item id to remove
@@ -403,6 +413,19 @@ public:
    */
   int remove_item(CephContext *cct, int id);
 
+  /**
+   * remove all instances of an item nested beneath a certain point from the map
+   *
+   * @param cct cct
+   * @param id item id to remove
+   * @param ancestor ancestor item id under which to search for id
+   * @return 0 on success, negative on error
+   */
+private:
+  int _remove_item_under(CephContext *cct, int id, int ancestor);
+public:
+  int remove_item_under(CephContext *cct, int id, int ancestor);
+
   /**
    * get an item's weight
    *