From ee6b9937aabb9b810af3a518005c4dfa57a6e39f Mon Sep 17 00:00:00 2001
From: Sage Weil <sage@inktank.com>
Date: Wed, 20 Mar 2013 08:40:09 -0700
Subject: [PATCH] crush: add remove_item_under()

Remove only instances of item nested beneath a particular ancestor.

Signed-off-by: Sage Weil <sage@inktank.com>
---
 src/crush/CrushWrapper.cc | 81 +++++++++++++++++++++++++++++++++++++--
 src/crush/CrushWrapper.h  | 25 +++++++++++-
 2 files changed, 102 insertions(+), 4 deletions(-)

diff --git a/src/crush/CrushWrapper.cc b/src/crush/CrushWrapper.cc
index bb23901223612..e5af5405f72eb 100644
--- 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 6a12d95ff16cf..048eb3dee5cf0 100644
--- 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
    *
-- 
2.39.5