git.maemo.org Git - gc-dialer/blob - hand_tests/threading.py

   1 #!/usr/bin/env python
   2
   3 from __future__ import with_statement
   4 from __future__ import division
   5
   6 import time
   7
   8 import sys
   9 sys.path.insert(0,"./src")
  10 from util import qt_compat
  11 QtCore = qt_compat.QtCore
  12
  13
  14 class QThread44(QtCore.QThread):
  15         """
  16         This is to imitate QThread in Qt 4.4+ for when running on older version
  17         See http://labs.trolltech.com/blogs/2010/06/17/youre-doing-it-wrong
  18         (On Lucid I have Qt 4.7 and this is still an issue)
  19         """
  20
  21         def __init__(self, parent = None):
  22                 QtCore.QThread.__init__(self, parent)
  23
  24         def run(self):
  25                 self.exec_()
  26
  27
  28 class _WorkerThread(QtCore.QObject):
  29
  30         _taskComplete  = qt_compat.Signal(object)
  31
  32         def __init__(self, futureThread):
  33                 QtCore.QObject.__init__(self)
  34                 self._futureThread = futureThread
  35                 self._futureThread._addTask.connect(self._on_task_added)
  36                 self._taskComplete.connect(self._futureThread._on_task_complete)
  37
  38         @qt_compat.Slot(object)
  39         def _on_task_added(self, task):
  40                 if not self._futureThread._isRunning:
  41                         print "Dropping task"
  42
  43                 func, args, kwds, on_success, on_error = task
  44
  45                 try:
  46                         result = func(*args, **kwds)
  47                         isError = False
  48                 except Exception, e:
  49                         print "Error, passing it back to the main thread"
  50                         result = e
  51                         isError = True
  52
  53                 taskResult = on_success, on_error, isError, result
  54                 self._taskComplete.emit(taskResult)
  55
  56
  57 class FutureThread(QtCore.QObject):
  58
  59         _addTask = qt_compat.Signal(object)
  60
  61         def __init__(self):
  62                 QtCore.QObject.__init__(self)
  63                 self._thread = QThread44()
  64                 self._isRunning = False
  65                 self._worker = _WorkerThread(self)
  66                 self._worker.moveToThread(self._thread)
  67
  68         def start(self):
  69                 self._thread.start()
  70                 self._isRunning = True
  71
  72         def stop(self):
  73                 self._isRunning = False
  74                 self._thread.quit()
  75
  76         def add_task(self, func, args, kwds, on_success, on_error):
  77                 assert self._isRunning, "Task queue not started"
  78                 task = func, args, kwds, on_success, on_error
  79                 self._addTask.emit(task)
  80
  81         @qt_compat.Slot(object)
  82         def _on_task_complete(self, taskResult):
  83                 on_success, on_error, isError, result = taskResult
  84                 if not self._isRunning:
  85                         if isError:
  86                                 print "Masking: %s" % (result, )
  87                         isError = True
  88                         result = StopIteration("Cancelling all callbacks")
  89                 callback = on_success if not isError else on_error
  90                 try:
  91                         callback(result)
  92                 except Exception:
  93                         print "Callback errored"
  94
  95
  96 class Producer(QtCore.QObject):
  97
  98         data = qt_compat.Signal(int)
  99         done = qt_compat.Signal()
 100
 101         def __init__(self):
 102                 QtCore.QObject.__init__(self)
 103
 104         @qt_compat.Slot()
 105         def process(self):
 106                 print "Starting producer"
 107                 for i in xrange(10):
 108                         self.data.emit(i)
 109                         time.sleep(0.1)
 110                 self.done.emit()
 111
 112
 113 class Consumer(QtCore.QObject):
 114
 115         def __init__(self):
 116                 QtCore.QObject.__init__(self)
 117
 118         @qt_compat.Slot()
 119         def process(self):
 120                 print "Starting consumer"
 121
 122         @qt_compat.Slot()
 123         def print_done(self):
 124                 print "Done"
 125
 126         @qt_compat.Slot(int)
 127         def print_data(self, i):
 128                 print i
 129
 130
 131 def run_producer_consumer():
 132         app = QtCore.QCoreApplication([])
 133
 134         producerThread = QThread44()
 135         producer = Producer()
 136         producer.moveToThread(producerThread)
 137         producerThread.started.connect(producer.process)
 138
 139         consumerThread = QThread44()
 140         consumer = Consumer()
 141         consumer.moveToThread(consumerThread)
 142         consumerThread.started.connect(consumer.process)
 143
 144         producer.data.connect(consumer.print_data)
 145         producer.done.connect(consumer.print_done)
 146
 147         @qt_compat.Slot()
 148         def producer_done():
 149                 print "Shutting down"
 150                 producerThread.quit()
 151                 consumerThread.quit()
 152                 print "Done"
 153         producer.done.connect(producer_done)
 154
 155         count = [0]
 156
 157         @qt_compat.Slot()
 158         def thread_done():
 159                 print "Thread done"
 160                 count[0] += 1
 161                 if count[0] == 2:
 162                         print "Quitting"
 163                         app.exit(0)
 164                 print "Done"
 165         producerThread.finished.connect(thread_done)
 166         consumerThread.finished.connect(thread_done)
 167
 168         producerThread.start()
 169         consumerThread.start()
 170         print "Status %s" % app.exec_()
 171
 172
 173 def run_task():
 174         app = QtCore.QCoreApplication([])
 175
 176         bright = FutureThread()
 177         def on_failure(*args):
 178                 print "Failure", args
 179
 180         def on_success(*args):
 181                 print "Success", args
 182
 183         def task(*args):
 184                 print "Task", args
 185
 186         timer = QtCore.QTimer()
 187         timeouts = [0]
 188         @qt_compat.Slot()
 189         def on_timeout():
 190                 timeouts[0] += 1
 191                 print timeouts[0]
 192                 bright.add_task(task, (timeouts[0], ), {}, on_success, on_failure)
 193                 if timeouts[0] == 5:
 194                         timer.stop()
 195                         bright.stop()
 196                         app.exit(0)
 197         timer.timeout.connect(on_timeout)
 198         timer.start(10)
 199         bright.start()
 200
 201         print "Status %s" % app.exec_()
 202
 203
 204 if __name__ == "__main__":
 205         run_producer_consumer()
 206         #run_task()