"""A threading based handler.
The :class:`SequentialThreadingHandler` is intended for regular Python
environments that use threads.
.. warning::
Do not use :class:`SequentialThreadingHandler` with applications
using asynchronous event loops (like gevent). Use the
:class:`~kazoo.handlers.gevent.SequentialGeventHandler` instead.
"""
from __future__ import absolute_import
import atexit
import logging
import select
import socket
import threading
import time
try:
import Queue
except ImportError: # pragma: nocover
import queue as Queue
from kazoo.handlers.utils import create_tcp_socket, create_tcp_connection
# sentinel objects
_NONE = object()
_STOP = object()
log = logging.getLogger(__name__)
[docs]class TimeoutError(Exception):
pass
[docs]class AsyncResult(object):
"""A one-time event that stores a value or an exception"""
def __init__(self, handler):
self._handler = handler
self.value = None
self._exception = _NONE
self._condition = threading.Condition()
self._callbacks = []
[docs] def ready(self):
"""Return true if and only if it holds a value or an
exception"""
return self._exception is not _NONE
[docs] def successful(self):
"""Return true if and only if it is ready and holds a value"""
return self._exception is None
@property
def exception(self):
if self._exception is not _NONE:
return self._exception
[docs] def set(self, value=None):
"""Store the value. Wake up the waiters."""
with self._condition:
self.value = value
self._exception = None
for callback in self._callbacks:
self._handler.completion_queue.put(
lambda: callback(self)
)
self._condition.notify_all()
[docs] def set_exception(self, exception):
"""Store the exception. Wake up the waiters."""
with self._condition:
self._exception = exception
for callback in self._callbacks:
self._handler.completion_queue.put(
lambda: callback(self)
)
self._condition.notify_all()
[docs] def get(self, block=True, timeout=None):
"""Return the stored value or raise the exception.
If there is no value raises TimeoutError.
"""
with self._condition:
if self._exception is not _NONE:
if self._exception is None:
return self.value
raise self._exception
elif block:
self._condition.wait(timeout)
if self._exception is not _NONE:
if self._exception is None:
return self.value
raise self._exception
# if we get to this point we timeout
raise TimeoutError()
[docs] def get_nowait(self):
"""Return the value or raise the exception without blocking.
If nothing is available, raises TimeoutError
"""
return self.get(block=False)
[docs] def wait(self, timeout=None):
"""Block until the instance is ready."""
with self._condition:
self._condition.wait(timeout)
return self._exception is not _NONE
[docs] def rawlink(self, callback):
"""Register a callback to call when a value or an exception is
set"""
with self._condition:
# Are we already set? Dispatch it now
if self.ready():
self._handler.completion_queue.put(
lambda: callback(self)
)
return
if callback not in self._callbacks:
self._callbacks.append(callback)
[docs] def unlink(self, callback):
"""Remove the callback set by :meth:`rawlink`"""
with self._condition:
if self.ready():
# Already triggered, ignore
return
if callback in self._callbacks:
self._callbacks.remove(callback)
[docs]class SequentialThreadingHandler(object):
"""Threading handler for sequentially executing callbacks.
This handler executes callbacks in a sequential manner. A queue is
created for each of the callback events, so that each type of event
has its callback type run sequentially. These are split into two
queues, one for watch events and one for async result completion
callbacks.
Each queue type has a thread worker that pulls the callback event
off the queue and runs it in the order the client sees it.
This split helps ensure that watch callbacks won't block session
re-establishment should the connection be lost during a Zookeeper
client call.
Watch and completion callbacks should avoid blocking behavior as
the next callback of that type won't be run until it completes. If
you need to block, spawn a new thread and return immediately so
callbacks can proceed.
.. note::
Completion callbacks can block to wait on Zookeeper calls, but
no other completion callbacks will execute until the callback
returns.
"""
name = "sequential_threading_handler"
timeout_exception = TimeoutError
sleep_func = staticmethod(time.sleep)
queue_impl = Queue.Queue
queue_empty = Queue.Empty
def __init__(self):
"""Create a :class:`SequentialThreadingHandler` instance"""
self.callback_queue = self.queue_impl()
self.completion_queue = self.queue_impl()
self._running = False
self._state_change = threading.Lock()
self._workers = []
def _create_thread_worker(self, queue):
def thread_worker(): # pragma: nocover
while True:
try:
func = queue.get()
try:
if func is _STOP:
break
func()
except Exception:
log.exception("Exception in worker queue thread")
finally:
queue.task_done()
except self.queue_empty:
continue
t = threading.Thread(target=thread_worker)
# Even though these should be joined, it's possible stop might
# not issue in time so we set them to daemon to let the program
# exit anyways
t.daemon = True
t.start()
return t
[docs] def start(self):
"""Start the worker threads."""
with self._state_change:
if self._running:
return
# Spawn our worker threads, we have
# - A callback worker for watch events to be called
# - A completion worker for completion events to be called
for queue in (self.completion_queue, self.callback_queue):
w = self._create_thread_worker(queue)
self._workers.append(w)
self._running = True
atexit.register(self.stop)
[docs] def stop(self):
"""Stop the worker threads and empty all queues."""
with self._state_change:
if not self._running:
return
self._running = False
for queue in (self.completion_queue, self.callback_queue):
queue.put(_STOP)
self._workers.reverse()
while self._workers:
worker = self._workers.pop()
worker.join()
# Clear the queues
self.callback_queue = self.queue_impl()
self.completion_queue = self.queue_impl()
if hasattr(atexit, "unregister"):
atexit.unregister(self.stop)
def select(self, *args, **kwargs):
return select.select(*args, **kwargs)
def socket(self):
return create_tcp_socket(socket)
def create_connection(self, *args, **kwargs):
return create_tcp_connection(socket, *args, **kwargs)
[docs] def event_object(self):
"""Create an appropriate Event object"""
return threading.Event()
[docs] def lock_object(self):
"""Create a lock object"""
return threading.Lock()
[docs] def rlock_object(self):
"""Create an appropriate RLock object"""
return threading.RLock()
[docs] def async_result(self):
"""Create a :class:`AsyncResult` instance"""
return AsyncResult(self)
def spawn(self, func, *args, **kwargs):
t = threading.Thread(target=func, args=args, kwargs=kwargs)
t.daemon = True
t.start()
return t
[docs] def dispatch_callback(self, callback):
"""Dispatch to the callback object
The callback is put on separate queues to run depending on the
type as documented for the :class:`SequentialThreadingHandler`.
"""
self.callback_queue.put(lambda: callback.func(*callback.args))