memcached: add support for group deletion

Change-Id: Iff0e41180a212b56040f1142c499266a1c522f3d

3 files changed, 79 insertions, 3 deletions

diff --git a/tooz/coordination.py b/tooz/coordination.py
index 86fbcc5..0cdbef1 100644
--- a/tooz/coordination.py
+++ b/tooz/coordination.py
@@ -244,6 +244,17 @@ class CoordinationDriver(object):
         raise tooz.NotImplemented
 
     @staticmethod
+    def delete_group(group_id):
+        """Delete a group asynchronously.
+
+        :param group_id: the id of the group to leave
+        :type group_id: str
+        :returns: Result
+        :rtype: CoordAsyncResult
+        """
+        raise tooz.NotImplemented
+
+    @staticmethod
     def get_members(group_id):
         """Return the list of all members ids of the specified group
         asynchronously.
@@ -403,3 +414,10 @@ class MemberNotJoined(ToozError):
         self.member_id = member_id
         super(MemberNotJoined, self).__init__("Member %s has not joined %s" %
                                               (member_id, group_id))
+
+
+class GroupNotEmpty(ToozError):
+    "Exception raised when the caller try to delete a group with members."
+    def __init__(self, group_id):
+        self.group_id = group_id
+        super(GroupNotEmpty, self).__init__("Group %s is not empty" % group_id)
diff --git a/tooz/drivers/memcached.py b/tooz/drivers/memcached.py
index 661438f..779ceb7 100644
--- a/tooz/drivers/memcached.py
+++ b/tooz/drivers/memcached.py
@@ -175,6 +175,20 @@ class MemcachedDriver(coordination.CoordinationDriver):
                 # Someone updated the group list before us, try again!
                 raise _retry.Retry
 
+    @_retry.retry
+    def _remove_from_group_list(self, group_id):
+        """Remove group from the group list.
+
+        :param group_id: The group id
+        """
+        group_list, cas = self.client.gets(self._GROUP_LIST_KEY)
+        group_list = set(group_list)
+        group_list.remove(group_id)
+        if not self.client.cas(self._GROUP_LIST_KEY,
+                               list(group_list), cas):
+            # Someone updated the group list before us, try again!
+            raise _retry.Retry
+
     def create_group(self, group_id):
         encoded_group = self._encode_group_id(group_id)
 
@@ -196,7 +210,7 @@ class MemcachedDriver(coordination.CoordinationDriver):
         @_retry.retry
         def _join_group():
             group_members, cas = self.client.gets(encoded_group)
-            if not cas:
+            if group_members is None:
                 raise coordination.GroupNotCreated(group_id)
             if self._member_id in group_members:
                 raise coordination.MemberAlreadyExist(group_id,
@@ -217,7 +231,7 @@ class MemcachedDriver(coordination.CoordinationDriver):
         @_retry.retry
         def _leave_group():
             group_members, cas = self.client.gets(encoded_group)
-            if not cas:
+            if group_members is None:
                 raise coordination.GroupNotCreated(group_id)
             if self._member_id not in group_members:
                 raise coordination.MemberNotJoined(group_id, self._member_id)
@@ -232,6 +246,25 @@ class MemcachedDriver(coordination.CoordinationDriver):
     def _destroy_group(self, group_id):
         self.client.delete(self._encode_group_id(group_id))
 
+    def delete_group(self, group_id):
+        encoded_group = self._encode_group_id(group_id)
+
+        @_retry.retry
+        def _delete_group():
+            group_members, cas = self.client.gets(encoded_group)
+            if group_members is None:
+                raise coordination.GroupNotCreated(group_id)
+            if group_members != {}:
+                raise coordination.GroupNotEmpty(group_id)
+            # Delete is not atomic, so we first set the group to
+            # using CAS, and then we delete it, to avoid race conditions.
+            if not self.client.cas(encoded_group, None, cas):
+                raise _retry.Retry
+            self.client.delete(encoded_group)
+            self._remove_from_group_list(group_id)
+
+        return MemcachedFutureResult(self._executor.submit(_delete_group))
+
     @_retry.retry
     def _get_members(self, group_id):
         encoded_group = self._encode_group_id(group_id)
@@ -271,7 +304,7 @@ class MemcachedDriver(coordination.CoordinationDriver):
         @_retry.retry
         def _update_capabilities():
             group_members, cas = self.client.gets(encoded_group)
-            if cas is None:
+            if group_members is None:
                 raise coordination.GroupNotCreated(group_id)
             if self._member_id not in group_members:
                 raise coordination.MemberNotJoined(group_id, self._member_id)
diff --git a/tooz/tests/test_coordination.py b/tooz/tests/test_coordination.py
index 0890256..db385e8 100644
--- a/tooz/tests/test_coordination.py
+++ b/tooz/tests/test_coordination.py
@@ -114,6 +114,31 @@ class TestAPI(testscenarios.TestWithScenarios,
         for group_id in groups_ids:
             self.assertTrue(group_id in created_groups)
 
+    def test_delete_group(self):
+        self._coord.create_group(self.group_id).get()
+        all_group_ids = self._coord.get_groups().get()
+        self.assertTrue(self.group_id in all_group_ids)
+        self._coord.delete_group(self.group_id).get()
+        all_group_ids = self._coord.get_groups().get()
+        self.assertFalse(self.group_id in all_group_ids)
+        join_group = self._coord.join_group(self.group_id)
+        self.assertRaises(tooz.coordination.GroupNotCreated,
+                          join_group.get)
+
+    def test_delete_group_non_existent(self):
+        delete = self._coord.delete_group(self.group_id)
+        self.assertRaises(tooz.coordination.GroupNotCreated,
+                          delete.get)
+
+    def test_delete_group_non_empty(self):
+        self._coord.create_group(self.group_id).get()
+        self._coord.join_group(self.group_id).get()
+        delete = self._coord.delete_group(self.group_id)
+        self.assertRaises(tooz.coordination.GroupNotEmpty,
+                          delete.get)
+        self._coord.leave_group(self.group_id)
+        self._coord.delete_group(self.group_id).get()
+
     def test_join_group(self):
         self._coord.create_group(self.group_id).get()
         self._coord.join_group(self.group_id).get()