removing plural from model names. some cleanup

[plstackapi.git] / planetstack / ec2_observer / event_loop.py

diff --git a/planetstack/ec2_observer/event_loop.py b/planetstack/ec2_observer/event_loop.py
index 88725db..1319fec 100644 (file)
--- a/planetstack/ec2_observer/event_loop.py
+++ b/planetstack/ec2_observer/event_loop.py
@@ -6,6 +6,7 @@ import traceback
 import commands
 import threading
 import json
+import pdb
 
 from datetime import datetime
 from collections import defaultdict
@@ -48,7 +49,7 @@ def invert_graph(g):
        return ig
 
 class PlanetStackObserver:
-       #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivileges,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
+       #sync_steps = [SyncNetworks,SyncNetworkSlivers,SyncSites,SyncSitePrivilege,SyncSlices,SyncSliceMemberships,SyncSlivers,SyncSliverIps,SyncExternalRoutes,SyncUsers,SyncRoles,SyncNodes,SyncImages,GarbageCollector]
        sync_steps = []
 
        
@@ -221,7 +222,6 @@ class PlanetStackObserver:
                                self.last_deletion_run_times[e]=0
 
 
-
        def save_run_times(self):
                run_times = json.dumps(self.last_run_times)
                open('/tmp/observer_run_times','w').write(run_times)
@@ -241,20 +241,31 @@ class PlanetStackObserver:
                step = self.step_lookup[S]
                start_time=time.time()
                
+               dependency_graph = self.dependency_graph if not deletion else self.deletion_dependency_graph
+
                # Wait for step dependencies to be met
-               deps = self.dependency_graph[S]
-               for d in deps:
-                       cond = self.step_conditions[d]
-                       acquire(cond)
-                       if (self.step_status[S] is STEP_STATUS_WORKING):
-                               cond.wait()
-                       cond.release()
-
-               if (self.step_status[S] is not STEP_STATUS_OK):
+               try:
+                       deps = self.dependency_graph[S]
+                       has_deps = True
+               except KeyError:
+                       has_deps = False
+
+               if (has_deps):
+                       for d in deps:
+                               cond = self.step_conditions[d]
+                               cond.acquire()
+                               if (self.step_status[d] is STEP_STATUS_WORKING):
+                                       cond.wait()
+                               cond.release()
+                       go = self.step_status[d] == STEP_STATUS_OK
+               else:
+                       go = True
+
+               if (not go):
                        self.failed_steps.append(sync_step)
                        my_status = STEP_STATUS_KO
                else:
-                       sync_step = step(driver=self.driver,error_map=error_mapper)
+                       sync_step = step(driver=self.driver,error_map=self.error_mapper)
                        sync_step.__name__ = step.__name__
                        sync_step.dependencies = []
                        try:
@@ -347,7 +358,7 @@ class PlanetStackObserver:
                                self.step_status = {}
                                for p in list(providers):
                                        self.step_conditions[p] = threading.Condition()
-                                       self.step_status[p] = STEP_STATUS_IDLE
+                                       self.step_status[p] = STEP_STATUS_WORKING
 
 
                                logger.info('Waiting for event')
@@ -361,10 +372,11 @@ class PlanetStackObserver:
                                        logger.info('Deletion=%r...'%deletion)
                                        schedule = self.ordered_steps if not deletion else reversed(self.ordered_steps)
 
-                                       thread = threading.Thread(target=self.sync, args=(schedule.start_conditions, schedule.ordered_steps,deletion, schedule.signal_sem))
+                                       for S in schedule:
+                                               thread = threading.Thread(target=self.sync, args=(S, deletion))
 
-                                       logger.info('Deletion=%r...'%deletion)
-                                       threads.append(thread)
+                                               logger.info('Deletion=%r...'%deletion)
+                                               threads.append(thread)
 
                                        # Start threads 
                                        for t in threads: