1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
|
"""A eventlet based handler."""
from __future__ import absolute_import
import contextlib
import logging
import eventlet
from eventlet.green import select as green_select
from eventlet.green import socket as green_socket
from eventlet.green import time as green_time
from eventlet.green import threading as green_threading
from eventlet import queue as green_queue
from kazoo.handlers import utils
import kazoo.python2atexit as python2atexit
LOG = logging.getLogger(__name__)
# sentinel objects
_STOP = object()
@contextlib.contextmanager
def _yield_before_after():
# Yield to any other co-routines...
#
# See: http://eventlet.net/doc/modules/greenthread.html
# for how this zero sleep is really a cooperative yield to other potential
# co-routines...
eventlet.sleep(0)
try:
yield
finally:
eventlet.sleep(0)
class TimeoutError(Exception):
pass
class AsyncResult(utils.AsyncResult):
"""A one-time event that stores a value or an exception"""
def __init__(self, handler):
super(AsyncResult, self).__init__(handler,
green_threading.Condition,
TimeoutError)
class SequentialEventletHandler(object):
"""Eventlet 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 greenthread 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 greenthread 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_eventlet_handler"
def __init__(self):
"""Create a :class:`SequentialEventletHandler` instance"""
self.callback_queue = green_queue.LightQueue()
self.completion_queue = green_queue.LightQueue()
self._workers = []
self._started = False
@staticmethod
def sleep_func(wait):
green_time.sleep(wait)
@property
def running(self):
return self._started
timeout_exception = TimeoutError
def _process_completion_queue(self):
while True:
cb = self.completion_queue.get()
if cb is _STOP:
break
try:
with _yield_before_after():
cb()
except Exception:
LOG.warning("Exception in worker completion queue greenlet",
exc_info=True)
def _process_callback_queue(self):
while True:
cb = self.callback_queue.get()
if cb is _STOP:
break
try:
with _yield_before_after():
cb()
except Exception:
LOG.warning("Exception in worker callback queue greenlet",
exc_info=True)
def start(self):
if not self._started:
# Spawn our worker threads, we have
# - A callback worker for watch events to be called
# - A completion worker for completion events to be called
w = eventlet.spawn(self._process_completion_queue)
self._workers.append((w, self.completion_queue))
w = eventlet.spawn(self._process_callback_queue)
self._workers.append((w, self.callback_queue))
self._started = True
python2atexit.register(self.stop)
def stop(self):
while self._workers:
w, q = self._workers.pop()
q.put(_STOP)
w.wait()
self._started = False
python2atexit.unregister(self.stop)
def socket(self, *args, **kwargs):
return utils.create_tcp_socket(green_socket)
def create_socket_pair(self):
return utils.create_socket_pair(green_socket)
def event_object(self):
return green_threading.Event()
def lock_object(self):
return green_threading.Lock()
def rlock_object(self):
return green_threading.RLock()
def create_connection(self, *args, **kwargs):
return utils.create_tcp_connection(green_socket, *args, **kwargs)
def select(self, *args, **kwargs):
with _yield_before_after():
return green_select.select(*args, **kwargs)
def async_result(self):
return AsyncResult(self)
def spawn(self, func, *args, **kwargs):
t = green_threading.Thread(target=func, args=args, kwargs=kwargs)
t.daemon = True
t.start()
return t
def dispatch_callback(self, callback):
self.callback_queue.put(lambda: callback.func(*callback.args))
|
