Changeset af75d48

Timestamp:

Mar 1, 2010 10:50:00 AM (14 years ago)

Author:

Ted Faber <faber@…>

Branches:

axis_example, compt_changes, info-ops, master, version-3.01, version-3.02

Children:

9c2e4e1

Parents:

b78c9ea

Message:

Add the ability to revoke a synch key

File:

• fedd/federation/synch_store.py (modified) (10 diffs)

fedd/federation/synch_store.py

@@ -20 +20 @@
     class BadKeyError(Exception): pass
 
+    # Thrown when a variable has been revoked.
+    class RevokedKeyError(Exception): pass
+
     # A value in the dict.  Each has its own condition (and lock) for threads
     # to wait on.
@@ -37 +40 @@
         self.main_lock = Lock()
         self.values = { }
+        self.revoked = set([])
 
     # Set the value in the dict.  If threads are waiting, wake them.
@@ -44 +48 @@
 
         self.main_lock.acquire()
+        if key in self.revoked:
+            self.main_lock.release()
+            raise synch_store.RevokedKeyError("Synch key %s revoked" % key)
         if self.values.has_key(key):
             v = self.values[key]
@@ -63 +70 @@
     def get_value(self, key, wait=True):
         self.main_lock.acquire()
+        if key in self.revoked:
+            self.main_lock.release()
+            raise synch_store.RevokedKeyError("Synch key %s revoked" % key)
         if not self.values.has_key(key):
             self.values[key] = synch_store.synch_value()
@@ -70 +80 @@
 
         while v.value is None:
-            if wait: v.condition.wait()
-            else: break
+            if wait: 
+                v.condition.wait()
+                v.condition.release()
+                self.main_lock.acquire()
+                if key in self.revoked:
+                    self.main_lock.release()
+                    raise synch_store.RevokedKeyError("Synch key %s revoked" % key)
+                else:
+                    self.main_lock.release()
+                v.condition.acquire()
+            else:
+                break
         rv = v.value
         v.condition.release()
@@ -78 +98 @@
     # Remove the key from the shared store.  If the key is not present or has
     # never been set, fail.  Otherwise, wake any waiting threads (there
-    # shouldn't be any) and remove the value and key from the store.
+    # shouldn't be any) and remove the value and key from the store.  Deleteing
+    # a revoked key removes its revocation.
     def del_value(self, key):
         self.main_lock.acquire()
+        if key in self.revoked:
+            self.revoked.remove(key)
+            self.main_lock.release()
+        else:
+            if self.values.has_key(key):
+                v = self.values[key]
+                v.condition.acquire()
+                if v.value is not None:
+                    del self.values[key]
+                    self.main_lock.release()
+                    v.condition.notifyAll()
+                    v.condition.release()
+                else:
+                    self.main_lock.release()
+                    v.condition.release()
+                    raise synch_store.BadDeletionError("%s has not been set" \
+                            % key)
+            else:
+                self.main_lock.release()
+                raise synch_store.BadDeletionError("%s does not exist yet" \
+                        % key)
+
+    # Invlaidate this synchronization point.  Those waiting for it will get
+    # exceptions, and those trying to wait for it in the future will as well.
+    def revoke_key(self, key):
+        self.main_lock.acquire()
+        self.revoked.add(key)
         if self.values.has_key(key):
             v = self.values[key]
+            del self.values[key]
+            self.main_lock.release()
             v.condition.acquire()
-            if v.value is not None:
-                del self.values[key]
-                self.main_lock.release()
-                v.condition.notifyAll()
-            else:
-                self.main_lock.release()
-                raise synch_store.BadDeletionError("%s has not been set" % key)
+            v.condition.notifyAll()
+            v.condition.release()
         else:
             self.main_lock.release()
-            raise synch_store.BadDeletionError("%s does not exist yet" % key)
+
 
     def all_keys(self):
@@ -145 +190 @@
     def get_val(k, ss, wait=True):
         print "%s: getting %s" % (currentThread(), k)
-        v = ss.get_value(k, wait)
+        try:
+            v = ss.get_value(k, wait)
+        except synch_store.RevokedKeyError, e:
+            print "Error: %s" %e
+            return
         print "%s: %s is %s" % (currentThread(), k, v)
 
@@ -156 +205 @@
             print "Error: %s" %e
             return
+        except synch_store.RevokedKeyError, e:
+            print "Error: %s" %e
+            return
         print "Done with set"
 
@@ -165 +217 @@
             print "Error: %s" %e
             return
+        except synch_store.RevokedKeyError, e:
+            print "Error: %s" %e
+            return
         print "Done with del"
+
+    def rev_val(k, ss):
+        print "revoking %s" % k
+        try:
+            ss.revoke_key(k)
+        except synch_store.BadDeletionError, e:
+            print "Error: %s" %e
+            return
+        except synch_store.RevokedKeyError, e:
+            print "Error: %s" %e
+            return
+        print "Done with revoking"
 
 
@@ -186 +253 @@
     print ss
 
-
+    for i in range(0,10):
+        t = Thread(target=get_val, args=('c', ss))
+        t.start()
+
+    
+    t = Thread(target=rev_val, args=('c', ss))
+    t.start()
+