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
|
##
# Copyright (c) 2010-2016 Apple Inc. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
##
import sys
from Queue import Queue
from twisted.python.failure import Failure
from twisted.internet.defer import Deferred
_DONE = object()
_STATE_STARTING = 'STARTING'
_STATE_RUNNING = 'RUNNING'
_STATE_STOPPING = 'STOPPING'
_STATE_STOPPED = 'STOPPED'
class ThreadHolder(object):
"""
A queue which will hold a reactor threadpool thread open until all of the
work in that queue is done.
"""
def __init__(self, reactor):
self._reactor = reactor
self._state = _STATE_STOPPED
self._stopper = None
self._q = None
self._retryCallback = None
def _run(self):
"""
Worker function which runs in a non-reactor thread.
"""
self._state = _STATE_RUNNING
while self._qpull():
pass
def _qpull(self):
"""
Pull one item off the queue and react appropriately.
Return whether or not to keep going.
"""
work = self._q.get()
if work is _DONE:
def finishStopping():
self._state = _STATE_STOPPED
self._q = None
s = self._stopper
self._stopper = None
s.callback(None)
self._reactor.callFromThread(finishStopping)
return False
self._oneWorkUnit(*work)
return True
def _oneWorkUnit(self, deferred, instruction):
try:
result = instruction()
except:
etype, evalue, etb = sys.exc_info()
def relayFailure():
f = Failure(evalue, etype, etb)
deferred.errback(f)
self._reactor.callFromThread(relayFailure)
else:
self._reactor.callFromThread(deferred.callback, result)
def submit(self, work):
"""
Submit some work to be run.
@param work: a 0-argument callable, which will be run in a thread.
@return: L{Deferred} that fires with the result of L{work}
"""
if self._state not in (_STATE_RUNNING, _STATE_STARTING):
raise RuntimeError("not running")
d = Deferred()
self._q.put((d, work))
return d
def start(self):
"""
Start this thing, if it's stopped.
"""
if self._state != _STATE_STOPPED:
raise RuntimeError("Not stopped.")
self._state = _STATE_STARTING
self._q = Queue(0)
self._reactor.callInThread(self._run)
self.retry()
def retry(self):
if self._state == _STATE_STARTING:
if self._retryCallback is not None:
self._reactor.threadpool.adjustPoolsize()
self._retryCallback = self._reactor.callLater(0.1, self.retry)
else:
self._retryCallback = None
def stop(self):
"""
Stop this thing and release its thread, if it's running.
"""
if self._state not in (_STATE_RUNNING, _STATE_STARTING):
raise RuntimeError("Not running.")
s = self._stopper = Deferred()
self._state = _STATE_STOPPING
self._q.put(_DONE)
if self._retryCallback:
self._retryCallback.cancel()
self._retryCallback = None
return s
|
