summaryrefslogtreecommitdiff
path: root/client/src/taskthread/__init__.py
blob: a734a829863c4f9cdb56d1d171069aed43e40da0 (plain)
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
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
# Copyright 2013 Hewlett-Packard Development Company, L.P.
#
# 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 logging
import threading

__version__ = '1.4'


logger = logging.getLogger(__name__)


class TaskInProcessException(BaseException):
    pass


class TaskThread(threading.Thread):
    """
    A thread object that repeats a task.

    Usage example::

        from taskthread import TaskThread

        import time

        def my_task(*args, **kwargs):
            print args, kwargs

        task_thread = TaskThread(my_task)
        task_thread.start()
        for i in xrange(10):
            task_thread.run_task()
            task_thread.join_task()
        task_thread.join()

    .. note:: If :py:meth:`~TaskThread.run_task` is
        invoked while run_task is in progress,
        :py:class:`~.TaskInProcessException` will
        be raised.

    :param task:
        A ``function``. This param is the task to execute when
         run_task is called.
    :param event:
        A ``threading.Event``. This event will be set when run_task
         is called. The default value is a new event, but may be
         specified for testing purposes.
    """

    daemon = True
    '''
    Threads marked as daemon will be terminated.
    '''
    def __init__(self, task, event=threading.Event(),
                 *args, **kwargs):
        super(TaskThread, self).__init__()
        self.task = task
        self.task_event = event
        self.running = True
        self.running_lock = threading.Lock()
        self.in_task = False
        self.task_complete = threading.Event()
        self.args = args
        self.kwargs = kwargs

    def run(self):
        """
        Called by threading.Thread, this runs in the new thread.
        """
        while True:
            self.task_event.wait()
            if not self.running:
                logger.debug("TaskThread exiting")
                return
            logger.debug("TaskThread starting task")
            with self.running_lock:
                self.task_event.clear()
            self.task_complete.clear()
            self.task(*self.args, **self.kwargs)
            with self.running_lock:
                self.in_task = False
            self.task_complete.set()

    def run_task(self, *args, **kwargs):
        """
        Run an instance of the task.

        :param args:
            The arguments to pass to the task.

        :param kwargs:
            The keyword arguments to pass to the task.
        """
        # Don't allow this call if the thread is currently
        # in a task.
        with self.running_lock:
            if self.in_task:
                raise TaskInProcessException()
            self.in_task = True
        logger.debug("Waking up the thread")
        self.args = args
        self.kwargs = kwargs
        # Wake up the thread to do it's thing
        self.task_event.set()

    def join_task(self, time_out):
        """
        Wait for the currently running task to complete.

        :param time_out:
            An ``int``. The amount of time to wait for the
            task to finish.
        """
        with self.running_lock:
            if not self.in_task:
                return

        success = self.task_complete.wait(time_out)
        if success:
            self.task_complete.clear()
        return success

    def join(self, timeout=None):
        """
        Wait for the task to finish
        """
        self.running = False
        self.task_event.set()
        super(TaskThread, self).join(timeout=timeout)


class TimerTask(object):
    """
    An object that executes a commit function at a given interval.
    This class is driven by a TaskThread. A new TaskThread will be
    created the first time :py:meth:`.~start` is called. All
    subsequent calls to start will reuse the same thread.

    Usage example::

        from taskthread import TimerTask
        import time

        count = 0
        def get_count():
            return count
        def execute():
            print "Count: %d" % count

        task = TimerTask(execute,
                         timeout=10,
                         count_fcn=get_count,
                         threshold=1)

        task.start()

        for i in xrange(100000):
            count += 1
            time.sleep(1)
        task.stop()
        count = 0
        task.start()
        for i in xrange(100000):
            count += 1
            time.sleep(1)
        task.shutdown()

    :param execute_fcn:
        A `function`. This function will be executed on each time interval.

    :param delay:
        An `int`. The delay in **seconds** invocations of
        `execute_fcn`. Default: `10`.

    :param count_fcn:
        A `function`. This function returns a current count. If the count
        has not changed more the `threshold` since the last invocation of
        `execute_fcn`, `execute_fcn` will not be called again. If not
        specified, `execute_fcn` will be called each time the timer fires.
        **Optional**. If count_fcn is specified, ``threshold`` is required.

    :param threshold:
        An `int`. Specifies the minimum delta in `count_fcn` that must be
        met for `execute_fcn` to be invoked. **Optional**. Must be
        specified in conjunction with `count_fcn`.

    """
    def __init__(self, execute_fcn, delay=10, count_fcn=None, threshold=None):
        self.running = True
        self.execute_fcn = execute_fcn
        self.last_count = 0
        self.event = threading.Event()
        self.delay = delay
        self.thread = None
        self.running_lock = threading.RLock()
        if bool(threshold) != bool(count_fcn):
            raise ValueError("Must specify threshold "
                             "and count_fcn, or neither")

        self.count_fcn = count_fcn
        self.threshold = threshold

    def start(self):
        """
        Start the task. This starts a :py:class:`.~TaskThread`, and starts
        running run_threshold_timer on the thread.

        """
        if not self.thread:
            logger.debug('Starting up the taskthread')
            self.thread = TaskThread(self._run_threshold_timer)
            self.thread.start()

        if self.threshold:
            self.last_count = 0

        logger.debug('Running the task')
        self.running = True
        self.thread.run_task()

    def stop(self):
        """
        Stop the task, leaving the :py:class:`.~TaskThread` running
        so start can be called again.

        """
        logger.debug('Stopping the task')
        wait = False
        with self.running_lock:
            if self.running:
                wait = True
                self.running = False
        if wait:
            self.event.set()
            self.thread.join_task(2)

    def shutdown(self):
        """
        Close down the task thread and stop the task if it is running.

        """
        logger.debug('Shutting down the task')
        self.stop()
        self.thread.join(2)

    def _exec_if_threshold_met(self):
        new_count = self.count_fcn()
        logger.debug('new_count: %d' % new_count)
        if new_count >= self.last_count + self.threshold:
            self.execute_fcn()
            self.last_count = new_count

    def _exec(self):
        if self.count_fcn:
            self._exec_if_threshold_met()
        else:
            self.execute_fcn()

    def _wait(self):
        self.event.wait(timeout=self.delay)
        self.event.clear()
        logger.debug('Task woken up')

    def _exit_loop(self):
        """
        If self.running is false, it means the task should shut down.
        """
        exit_loop = False
        with self.running_lock:
            if not self.running:
                exit_loop = True
                logger.debug('Task shutting down')
        return exit_loop

    def _run_threshold_timer(self):
        """
        Main loop of the timer task

        """
        logger.debug('In Task')
        while True:
            self._wait()
            if self._exit_loop():
                return
            self._exec()