Author: tmckay
Date: 2011-07-29 20:05:00 +0000 (Fri, 29 Jul 2011)
New Revision: 4896
Modified:
trunk/sage/python/sage/wallaby/wallabyoperations.py
Log:
Updates to WallabyOperations interface and functionality.
Modified: trunk/sage/python/sage/wallaby/wallabyoperations.py
===================================================================
--- trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-07-28 12:48:36 UTC (rev
4895)
+++ trunk/sage/python/sage/wallaby/wallabyoperations.py 2011-07-29 20:05:00 UTC (rev
4896)
@@ -9,14 +9,47 @@
try:
import wallaby
from wallaby_collections import StorePatches
+ from tagging import StorePatches, NodePatches
except:
imports_ok = False
log = logging.getLogger("sage.wallaby")
+class WBTypes():
+ '''
+ Wrap symbolic names for data items tracked by WallabyOperations.
+ May be helpful in avoiding typos, good for interpreter help
+ '''
+ NODES = "nodes"
+ GROUPS = "groups"
+ TAGS = "tags"
+
if imports_ok:
class WallabyOperations(object):
+ '''
+ Wrapper around the Wallaby client library.
+ '''
def __init__(self, broker_uri, refresh_interval=None):
+ '''
+ Constructor.
+
+ broker_uri -- the URI used to connect to a QMF message broker
+ where a Wallaby agent is connected. The simplest URI is just a
+ hostname but a full URI can specify scheme://user/password@host:port
+ or a subset of those components as long as the host is included.
+ Examples:
+
+ localhost
+ localhost:5672
+ amqp://fred/flintsone@quarry.bedrock.com:1234
+
+ refresh_interval -- default refresh interval in seconds for all items
+ maintained by WallabyOperations' internal caching thread. A value of
+ None causes the caching thread to wait forever before refreshing an
+ item after a successful call unless the refresh() method is used.
+ The refresh interval may be set for items individually with the
+ set_interval() method.
+ '''
self.broker_uri = broker_uri
# A wallaby Store object
@@ -26,57 +59,40 @@
self._broker = None
# The cache maintenance thread
- self.maintain_cache = None
+ self._maintain_cache = None
# Stop the maintenance thread
self._stop = False
- # Length of time the maintenance thread will sleep
- # (in seconds) before it next attempts to refresh cached
- # objects after a successful attempt. Failed attempts
- # result in a 5 second retry time.
- self.refresh_interval = refresh_interval
+ # Cached data. Each of the keys in this dictionary is the name of
+ # an attribute on the Wallaby Store object.
+ self._cache = {WBTypes.NODES: self.CacheData(refresh_interval),
+ WBTypes.GROUPS: self.CacheData(refresh_interval)}
- # Cached node list
- self.nodes = []
+ # Lock is used for synchronization with the caching thread and
+ # for thread safety of any and all data that could be accessed
+ # by multiple threads.
+ self._lock = Lock()
+ self._condition = Condition(self._lock)
- # Protect cached info, give the cache maintenance
- # thread a condition to sleep on that it may be woken
- # up early.
- self.lock = Lock()
- self.condition = Condition(self.lock)
-
- def stop(self, wait=False, timeout=None):
+ def start(self, retry_secs=5):
'''
- Wake the caching thread if asleep and cause it to exit.
- If wait is True, the call will return after the thread
- has exited or timeout seconds has passed if timeout is
- not None.
- '''
- self._stop = True
- self.refresh()
- if wait:
- self.maintain_cache.join(timeout)
- log.debug("WallabyOperations: stopped cache maintenance thread")
-
- def refresh(self):
- '''
- Wake the caching thread if asleep and cause it to iterate.
- '''
- self.condition.acquire()
- self.condition.notify()
- self.condition.release()
-
- def start(self):
- '''
Start the caching thread.
This thread will attempt to connect to the broker and retrieve
- a wallaby Store object. If successful, it will periodically
- retrieve and cache necessary wallaby objects.
+ a Store object from the Wallaby agent. If successful, it will
+ periodically retrieve and cache data from Wallaby.
Only one caching thread may run at a time. The thread may
be restarted if it has previously been stopped.
+
+ Note, for the moment start() and stop() are not thread safe. They
+ should only be called from a single thread.
+
+ retry_secs -- how often the caching thread will retry failed
+ operations. This includes attempts to connect to the broker
+ and retrieve a Store object as well as calls to Wallaby that
+ return no data.
'''
# The connection to the broker can actually take a long
# time to complete. We don't want to hang a calling function,
@@ -95,6 +111,7 @@
# function, but it refers to the WallabyOperations object
# so the code reads nicely
def maintain_cache(self):
+
# Get initinal connection and Store obect
self.session = Session(manageConnections=True)
self.broker = self.session.addBroker(self.broker_uri)
@@ -104,71 +121,218 @@
log.debug("WallabyOperations: found wallaby store
object")
break
- # Allow us to be woken up early...
- self.condition.acquire()
- self.condition.wait(5)
- self.condition.release()
+ # Check stop inside the lock to make sure that we don't miss
+ # a signal or a "stop" that was set while we were
iterating.
+ self._condition.acquire()
+ if not self._stop:
+ self._condition.wait(retry_secs)
+ self._condition.release()
- # Okay, we have a Store object so now we can
- # get node lists and such
+ # Init remaining time til next update to 0 for each
+ # cached item in case the thread was restarted
+ for attr, val in self._cache.iteritems():
+ val.remaining = 0
+
+ # Okay, now we're ready to retrieve data
while not self._stop:
- # Uses API extensions to __getattr__ on the Store to
- # retrieve a list of nodes from the Broker and wrap them
- # in Wallaby proxy objects. This might take a while.
- try:
- nodes = self._store.nodes
- except:
- nodes = []
- self._set_node_list(nodes)
- if nodes in ([], None):
- # Hmm, try again shortly
- sleep = 5
- else:
- # We found some nodes so cache them and
- # sleep for a normal interval
- sleep = self.refresh_interval
- self.condition.acquire()
- self.condition.wait(sleep)
- self.condition.release()
+ start_processing = time.time()
+ for attr, val in self._cache.iteritems():
+ # val.remaining is the number of seconds left before
+ # the next update of this data item. None is
"forever".
+ if val.remaining is not None and val.remaining <= 0:
+ log.debug("WallabyOperations: refreshing %s" %
attr)
+ try:
+ # Wallaby API uses extensions to __getattr__ on
+ # the Store to retrieve objects from the Broker
+ # and return a list of proxy objects.
+ d = getattr(self._store, attr, [])
+ except:
+ d = []
- # And wipe out the node list if we have been
- # stopped....
- self._set_node_list([])
+ # If the data is empty, _set_cache will leave the
+ # remaining field set to 0 for the attribute so we
+ # will try to get it again on our next retry.
+ # Otherwise, remaining will be reset to the full
+ # interval for this attribute.
+ self._set_cache(attr, d)
- if self.maintain_cache is not None and \
- self.maintain_cache.isAlive():
+ log.debug("WallabyOperations: refresh processing time %s"
\
+ % (time.time() - start_processing))
+
+ # Find out how long we should sleep for.
+ # Based on min remaining times for all items
+ # If minimum is 0 because we have items waiting
+ # for a retry, we fall back on retry_secs as a minimum.
+ sleep_time = self._find_min_remaining(min=retry_secs)
+
+ self._condition.acquire()
+ if not self._stop:
+ # Could be signaled, so track the actual sleep time
+ log.debug("WallabyOperations: cache thread sleeping
for"\
+ " %s seconds" % sleep_time)
+ bed_time = time.time()
+ self._condition.wait(sleep_time)
+ slept = time.time() - bed_time
+ log.debug("WallabyOperations: cache thread slept for"\
+ " %s seconds" % slept)
+
+ # When we wake up from sleep here, we already
+ # have the lock so we might as well check refresh
+ # and adjust the "remaining" values
+ for attr, val in self._cache.iteritems():
+ if val.refresh: # Force an update
+ val.remaining = 0
+ val.refresh = False
+ elif val.remaining is not None:
+ val.remaining -= slept
+ self._condition.release()
+
+ # Clear cache if we have been stopped....
+ for attr in self._cache:
+ self._set_cache(attr, [])
+ #end maintain_cache
+
+ if self._maintain_cache is not None and \
+ self._maintain_cache.isAlive():
# No, you can't start another one.
return False
self._stop = False
- self.maintain_cache = CallThread(maintain_cache, None, self)
- self.maintain_cache.daemon = True
- self.maintain_cache.start()
+ self._maintain_cache = CallThread(maintain_cache, None, self)
+ self._maintain_cache.daemon = True
+ self._maintain_cache.start()
log.debug("WallabyOperations: start cache maintenance thread")
return True
- def get_node_list(self):
+ def stop(self, wait=False, timeout=None):
'''
- Return the latest list of wallaby.Node objects retrieved by
- the caching thread.
+ Stop the caching thread.
+
+ Wake the caching thread if asleep and cause it to exit.
+ The thread may be restarted again with a call to start()
+ once it has successfully exited. On exit, the thread will
+ null out cached data.
+
+ wait -- if True the call will block until the thread exits or
+ "timeout" seconds has passed if "timeout" is not None.
+
+ timeout -- how long to wait for the thread to exit if "wait" is
True.
+ A value of None means wait forever.
+
+ Note, for the moment start() and stop() are not thread safe. They
+ should only be called from a single thread.
'''
- # Who really knows for sure how the GIL and byte code
- # works in Python? Play it safe, bump the reference
- # count on self.nodes inside the lock and return.
- self.lock.acquire()
- n = self.nodes
- self.lock.release()
- return n
+ if self.maintain_cache is not None:
+ self._condition.acquire()
+ self._stop = True
+ self._condition.notify()
+ self._condition.release()
+ if wait:
+ self._maintain_cache.join(timeout)
+ log.debug("WallabyOperations: stopped cache maintenance
thread")
+ def refresh(self, items=()):
+ '''
+ Wake the caching thread if asleep and cause it to iterate.
+
+ items -- what data to refresh. If "items" is an empty
+ tuple, refresh all data otherwise refresh only the data specified.
+ Attributes of WBTypes define valid values for elements of "items"
+ '''
+ self._condition.acquire()
+ if len(items) == 0:
+ do_notify = True
+ for attr, val in self._cache.iteritems():
+ val.refresh = True
+ else:
+ do_notify = False
+ for attr in items:
+ if attr == WBTypes.TAGS:
+ # Tags are really just groups with a filter
+ # in get_data()
+ attr = WBTypes.GROUPS
+ if attr in self._cache:
+ do_notify = True
+ self._cache[attr].refresh = True
+
+ if do_notify:
+ self._condition.notify()
+ self._condition.release()
+
+ def get_data(self, which):
+ '''
+ Return the most recently cached value for the specified data.
+
+ which -- specifies the data item. Attributes of WBTypes
+ define valid values for "which"
+ '''
+ d = []
+ if which == WBTypes.TAGS:
+ # Tags are special for the time being.
+ # Tags are defined as non-internal groups with no parameters.
+ # Get groups and filter.
+ target = WBTypes.GROUPS
+ else:
+ target = which
+
+ if target in self._cache:
+ self._lock.acquire()
+ d = self._cache[target].data
+ self._lock.release()
+
+ if which == WBTypes.TAGS:
+ groups = d
+ d = []
+ for g in groups:
+ if g.getConfig() == {} and not g.name.startswith("+++"):
+ d.append(g)
+ return d
+
+ def set_interval(self, which, refresh):
+ '''
+ Set an individual refresh interval for a data item.
+
+ The interval set here will override the initial default
+ interval for the item created in WallabyOperations' constructor.
+
+ which -- specifies the data item. Attributes of WBTypes
+ define valid values for "which"
+
+ refresh -- the interval in seconds. If None, the specified item
+ will be updated iff refresh() is called.
+ '''
+ if which == WBTypes.TAGS:
+ which = WBTypes.GROUPS
+
+ if which in self._cache:
+ self._lock.acquire()
+ self._cache[which].interval = refresh
+ self._lock.release()
+
# Super secret private implementation stuff. Don't look!
- def _set_node_list(self, nodes):
- self.lock.acquire()
- self.nodes = nodes
- self.lock.release()
- log.debug("WallabyOperations: Node list updated (%s nodes)" %
len(nodes))
+ def _find_min_remaining(self, min):
+ # None indicates forever, the biggest value
+ # Note though that None < int is True in Python!
+ t = None
+ for attr, val in self._cache.iteritems():
+ if t == None or \
+ (val.remaining is not None and val.remaining < t):
+ t = val.remaining
+ # Put a floor on this
+ if t is not None and t < min:
+ t = min
+ return t
+
+ def _set_cache(self, attr, data):
+ self._lock.acquire()
+ self._cache[attr].data = data
+ self._cache[attr].reset_remaining(len(data) == 0)
+ self._lock.release()
+ log.debug("WallabyOperations: %s list updated (%s items)" % (attr,
len(data)))
+
def _get_store(self):
# Ideally there should only be a single Store object from a single agent.
# We constrain the search a bit, hopefully more efficient
@@ -185,52 +349,46 @@
except:
pass
return store
+
+ class CacheData(object):
+ def __init__(self, interval):
+ # This attribure is only referenced from the cache thread
+ # so it does not need to be protected
+ self.remaining = 0
+
+ # Use of these these items need to be protected by the lock
+ self.refresh = False
+ self.interval = interval
+ self.data = []
+
+ def reset_remaining(self, retry):
+ # If retry is True, we want to try to get the data
+ # at the next opportunity, otherwise we will wait.
+ if retry:
+ self.remaining = 0
+ else:
+ self.remaining = self.interval
+
else:
class WallabyOperations(object):
'''
Dummy object when wallaby imports fail
'''
- def __init__(self, broker_uri, refresh_interval=60):
- self.broker_uri = broker_uri
- self.refresh_interval = refresh_interval
+ def __init__(self, *args, **kwargs):
+ pass
- def stop(self, wait=False, timeout=None):
- '''
- Wake the caching thread if asleep and cause it to exit.
- If wait is True, the call will return after the thread
- has exited or timeout seconds has passed if timeout is
- not None.
- '''
+ def start(self, *args, **kwargs):
+ log.debug("WallabyOperations: uing dummy implementation, imports
failed")
+ return False
+
+ def stop(self, *args, **kwargs):
pass
- def refresh(self):
- '''
- Wake the caching thread if asleep and cause it to iterate.
- '''
+ def refresh(self, *args, **kwargs):
pass
- def start(self):
- '''
- Start the caching thread.
-
- This thread will attempt to connect to the broker and retrieve
- a wallaby Store object. If successful, it will periodically
- retrieve and cache necessary wallaby objects.
-
- Only one caching thread may run at a time. The thread may
- be restarted if it has previously been stopped.
- '''
- log.debug("WallabyOperations: using dummy implementation, imports
failed")
- return False
-
- def get_node_list(self):
- '''
- Return the latest list of wallaby.Node objects retrieved by
- the caching thread.
- '''
+ def get_data(self, *args, **kwargs):
return []
-
-
-
-
+ def set_interval(self, *args, **kwargs):
+ pass