async_utils.py at master ⋅ lneuhaus/pyrpl

"""
This file contains a number of methods for asynchronous operations.
"""
import logging
from qtpy import QtCore, QtWidgets
from timeit import default_timer
logger = logging.getLogger(name=__name__)

from . import APP  # APP is only created once at the startup of PyRPL
MAIN_THREAD = APP.thread()

try:
    from asyncio import Future, ensure_future, CancelledError, \
        set_event_loop, TimeoutError
except ImportError:  # this occurs in python 2.7
    logger.debug("asyncio not found, we will use concurrent.futures "
                  "instead of python 3.5 Futures.")
    from concurrent.futures import Future, CancelledError, TimeoutError
else:
    import quamash
    set_event_loop(quamash.QEventLoop())



class MainThreadTimer(QtCore.QTimer):
    """
    To be able to start a timer from any (eventually non Qt) thread,
    we have to make sure that the timer is living in the main thread (in Qt,
    each thread potentially has a distinct eventloop...).

    For example, this is required to use the timer within timeit.

    we might decide one day to allow 2 event loops to run concurrently in
    separate threads, but
    1. That should be QThreads and not python threads
    2. We would have to make sure that all the gui operations are performed
    in the main thread (for instance, by moving all widgets in the
    mainthread, and probably, we would have to change some connections in
    QueuedConnections)
    ==> but this is not a supported feature for the moment and I don't see
    the advantage because the whole point of using an eventloop is to
    avoid multi-threading.

    For conveniance, MainThreadTimer is also SingleShot by default and is
    initialized with an interval as only argument.

    Benchmark:
    ----------

     1. keep starting the same timer over and over --> 5 microsecond/call::

            n = [0]
            tics = [default_timer()]
            timers = [None]
            N = 100000
            timer = MainThreadTimer(0)
            timer.timeout.connect(func)
            def func():
                n[0]+=1
                if n[0] > N:
                    print('done', (default_timer() - tics[0])/N)
                    return
                timer.start()
                timers[0] = timer
                return
            func() ---> 5 microseconds per call

     2. Instantiating a new timer at each call --> 15 microsecond/call::

            n = [0]
            tics = [default_timer()]
            timers = [None]
            N = 100000
            def func():
                n[0]+=1
                if n[0] > N:
                    print('done', (default_timer() - tics[0])/N)
                    return
                timer = MainThreadTimer(0)
                timer.timeout.connect(func)
                timer.start()
                timers[0] = timer
                return
            func() ---> 15 microseconds per call

    Moreover, no catastrophe occurs when instantiating >10e6 timers
    successively

    Conclusion: it is OK to instantiate a new timer every time it is needed
    as long as a 10 microsecond overhead is not a problem.
    """

    def __init__(self, interval):
        super(MainThreadTimer, self).__init__()
        self.moveToThread(MAIN_THREAD)
        self.setSingleShot(True)
        self.setInterval(interval)


class PyrplFuture(Future):
    """
    A promise object compatible with the Qt event loop.

    We voluntarily use an object that is different from the native QFuture
    because we want a promise object that is compatible with the python 3.5+
    asyncio patterns (for instance, it implements an __await__ method...).

    Attributes:
        cancelled: Returns whether the promise has been cancelled.
        exception: Blocks until:
                a. the result is ready --> returns None
                b. an exception accured in the execution --> returns the exception the Qt event-loop is allowed to run in parallel.
        done: Checks whether the result is ready or not.
        add_done_callback (callback function): add a callback to execute when result becomes available. The callback function takes 1 argument (the result of the promise).

    Methods to implement in derived class:
        _set_data_as_result(): set
    """

    def __init__(self):
        super(PyrplFuture, self).__init__()
        self._timer_timeout = None  # timer that will be instantiated if
        #  result(timeout) is called with a >0 value

    def result(self):
        """
        Blocks until the result is ready while running the event-loop in the background.

        Returns:
            The result of the future.
        """
        try: #  concurrent.futures.Future (python 2)
            return super(PyrplFuture, self).result(timeout=0)
        except TypeError: #  asyncio.Future (python 3)
            return super(PyrplFuture, self).result()

    def _exit_loop(self, x=None):
        """
        Parameter x=None is there such that the function can be set as
        a callback at the same time for timer_timeout.timeout (no
        argument) and for self.done (1 argument).
        """
        self.loop.quit()

    def _wait_for_done(self, timeout):
        """
        Will not return until either timeout expires or future becomes "done".
        There is one potential deadlock situation here:

        The deadlock occurs if we await_result while at the same
        time, this future needs to await_result from another future
        ---> To be safe, don't use await_result() in a Qt slot...
        """
        if self.cancelled():
            raise CancelledError("Future was cancelled")  # pragma: no-cover
        if not self.done():
            self.timer_timeout = None
            if (timeout is not None) and timeout > 0:
                self._timer_timeout = MainThreadTimer(timeout*1000)
                self._timer_timeout.timeout.connect(self._exit_loop)
                self._timer_timeout.start()
            self.loop = QtCore.QEventLoop()
            self.add_done_callback(self._exit_loop)
            self.loop.exec_()
            if self._timer_timeout is not None:
                if not self._timer_timeout.isActive():
                    return TimeoutError("Timeout occured")  # pragma: no-cover
                else:
                    self._timer_timeout.stop()

    def await_result(self, timeout=None):
        """
        Return the result of the call that the future represents.
        Will not return until either timeout expires or future becomes "done".

        There is one potential deadlock situation here:
        The deadlock occurs if we await_result while at the same
        time, this future needs to await_result from another future since
        the eventloop will be blocked.
        ---> To be safe, don't use await_result() in a Qt slot. You should
        rather use result() and add_done_callback() instead.

        Args:
            timeout: The number of seconds to wait for the result if the future
                isn't done. If None, then there is no limit on the wait time.

        Returns:
            The result of the call that the future represents.

        Raises:
            CancelledError: If the future was cancelled.
            TimeoutError: If the future didn't finish executing before the
                          given timeout.
            Exception: If the call raised then that exception will be raised.
        """

        self._wait_for_done(timeout)
        return self.result()

    def await_exception(self, timeout=None):  # pragma: no-cover
        """
        Return the exception raised by the call that the future represents.

        Args:
            timeout: The number of seconds to wait for the exception if the
                future isn't done. If None, then there is no limit on the wait
                time.

        Returns:
            The exception raised by the call that the future represents or None
            if the call completed without raising.

        Raises:
            CancelledError: If the future was cancelled.
            TimeoutError: If the future didn't finish executing before the
            given  timeout.
        """
        self._wait_for_done(timeout)
        return self.exception()

    def cancel(self):
        """
        Cancels the future.
        """
        if self._timer_timeout is not None:
            self._timer_timeout.stop()
        super(PyrplFuture, self).cancel()


def sleep(delay):
    """
    Sleeps for :code:`delay` seconds + runs the event loop in the background.

        * This function will never return until the specified delay in seconds is elapsed.
        * During the execution of this function, the qt event loop (== asyncio event-loop in pyrpl) continues to process events from the gui, or from other coroutines.
        * Contrary to time.sleep() or async.sleep(), this function will try to achieve a precision much better than 1 millisecond (of course, occasionally, the real delay can be longer than requested), but on average, the precision is in the microsecond range.
        * Finally, care has been taken to use low level system-functions to reduce CPU-load when no events need to be processed.

    More details on the implementation can be found on the page: `<https://github.com/lneuhaus/pyrpl/wiki/Benchmark-asynchronous-sleep-functions>`_.
    """
    tic = default_timer()
    end_time = tic + delay

    # 1. CPU-free sleep for delay - 1ms
    if delay > 1e-3:
        new_delay = delay - 1e-3
        loop = QtCore.QEventLoop()
        timer = MainThreadTimer(new_delay * 1000)
        timer.timeout.connect(loop.quit)
        timer.start()
        try:
            loop.exec_()
        except KeyboardInterrupt as e:  # pragma: no-cover
            # try to recover from KeyboardInterrupt by finishing the current task
            timer.setInterval(1)
            timer.start()
            loop.exec_()
            raise e
    # 2. For high-precision, manually process events 1-by-1 during the last ms
    while default_timer() < end_time:
        APP.processEvents()

Read our documentation on viewing source code .

Navigation	Overlay
`t` Navigate files	`h` Toggle hits
`y` Change url to tip of branch	`m` Toggle misses
`b / v` Jump to prev/next hit line	`p` Toggle partial
`z / x` Jump to prev/next missed or partial line	`1..9` Toggle flags
`shift + o` Open current page in GitHub	`a` Toggle all on
`/ or ?` Show keyboard shortcuts dialog	`c` Toggle context lines or commits

1		"""
2		This file contains a number of methods for asynchronous operations.
3		"""
4	3	import logging
5	3	from qtpy import QtCore, QtWidgets
6	3	from timeit import default_timer
7	3	logger = logging.getLogger(name=__name__)
8
9	3	from . import APP # APP is only created once at the startup of PyRPL
10	3	MAIN_THREAD = APP.thread()
11
12	3	try:
13	3	from asyncio import Future, ensure_future, CancelledError, \
14		set_event_loop, TimeoutError
15	1	except ImportError: # this occurs in python 2.7
16	1	logger.debug("asyncio not found, we will use concurrent.futures "
17		"instead of python 3.5 Futures.")
18	1	from concurrent.futures import Future, CancelledError, TimeoutError
19		else:
20	2	import quamash
21	2	set_event_loop(quamash.QEventLoop())
22
23
24
25	3	class MainThreadTimer(QtCore.QTimer):
26		"""
27		To be able to start a timer from any (eventually non Qt) thread,
28		we have to make sure that the timer is living in the main thread (in Qt,
29		each thread potentially has a distinct eventloop...).
30
31		For example, this is required to use the timer within timeit.
32
33		we might decide one day to allow 2 event loops to run concurrently in
34		separate threads, but
35		1. That should be QThreads and not python threads
36		2. We would have to make sure that all the gui operations are performed
37		in the main thread (for instance, by moving all widgets in the
38		mainthread, and probably, we would have to change some connections in
39		QueuedConnections)
40		==> but this is not a supported feature for the moment and I don't see
41		the advantage because the whole point of using an eventloop is to
42		avoid multi-threading.
43
44		For conveniance, MainThreadTimer is also SingleShot by default and is
45		initialized with an interval as only argument.
46
47		Benchmark:
48		----------
49
50		1. keep starting the same timer over and over --> 5 microsecond/call::
51
52		n = [0]
53		tics = [default_timer()]
54		timers = [None]
55		N = 100000
56		timer = MainThreadTimer(0)
57		timer.timeout.connect(func)
58		def func():
59		n[0]+=1
60		if n[0] > N:
61		print('done', (default_timer() - tics[0])/N)
62		return
63		timer.start()
64		timers[0] = timer
65		return
66		func() ---> 5 microseconds per call
67
68		2. Instantiating a new timer at each call --> 15 microsecond/call::
69
70		n = [0]
71		tics = [default_timer()]
72		timers = [None]
73		N = 100000
74		def func():
75		n[0]+=1
76		if n[0] > N:
77		print('done', (default_timer() - tics[0])/N)
78		return
79		timer = MainThreadTimer(0)
80		timer.timeout.connect(func)
81		timer.start()
82		timers[0] = timer
83		return
84		func() ---> 15 microseconds per call
85
86		Moreover, no catastrophe occurs when instantiating >10e6 timers
87		successively
88
89		Conclusion: it is OK to instantiate a new timer every time it is needed
90		as long as a 10 microsecond overhead is not a problem.
91		"""
92
93	3	def __init__(self, interval):
94	3	super(MainThreadTimer, self).__init__()
95	3	self.moveToThread(MAIN_THREAD)
96	3	self.setSingleShot(True)
97	3	self.setInterval(interval)
98
99
100	3	class PyrplFuture(Future):
101		"""
102		A promise object compatible with the Qt event loop.
103
104		We voluntarily use an object that is different from the native QFuture
105		because we want a promise object that is compatible with the python 3.5+
106		asyncio patterns (for instance, it implements an __await__ method...).
107
108		Attributes:
109		cancelled: Returns whether the promise has been cancelled.
110		exception: Blocks until:
111		a. the result is ready --> returns None
112		b. an exception accured in the execution --> returns the exception the Qt event-loop is allowed to run in parallel.
113		done: Checks whether the result is ready or not.
114		add_done_callback (callback function): add a callback to execute when result becomes available. The callback function takes 1 argument (the result of the promise).
115
116		Methods to implement in derived class:
117		_set_data_as_result(): set
118		"""
119
120	3	def __init__(self):
121	0	super(PyrplFuture, self).__init__()
122	0	self._timer_timeout = None # timer that will be instantiated if
123		# result(timeout) is called with a >0 value
124
125	3	def result(self):
126		"""
127		Blocks until the result is ready while running the event-loop in the background.
128
129		Returns:
130		The result of the future.
131		"""
132	0	try: # concurrent.futures.Future (python 2)
133	0	return super(PyrplFuture, self).result(timeout=0)
134	0	except TypeError: # asyncio.Future (python 3)
135	0	return super(PyrplFuture, self).result()
136
137	3	def _exit_loop(self, x=None):
138		"""
139		Parameter x=None is there such that the function can be set as
140		a callback at the same time for timer_timeout.timeout (no
141		argument) and for self.done (1 argument).
142		"""
143	0	self.loop.quit()
144
145	3	def _wait_for_done(self, timeout):
146		"""
147		Will not return until either timeout expires or future becomes "done".
148		There is one potential deadlock situation here:
149
150		The deadlock occurs if we await_result while at the same
151		time, this future needs to await_result from another future
152		---> To be safe, don't use await_result() in a Qt slot...
153		"""
154	0	if self.cancelled():
155	0	raise CancelledError("Future was cancelled") # pragma: no-cover
156	0	if not self.done():
157	0	self.timer_timeout = None
158	0	if (timeout is not None) and timeout > 0:
159	0	self._timer_timeout = MainThreadTimer(timeout*1000)
160	0	self._timer_timeout.timeout.connect(self._exit_loop)
161	0	self._timer_timeout.start()
162	0	self.loop = QtCore.QEventLoop()
163	0	self.add_done_callback(self._exit_loop)
164	0	self.loop.exec_()
165	0	if self._timer_timeout is not None:
166	0	if not self._timer_timeout.isActive():
167	0	return TimeoutError("Timeout occured") # pragma: no-cover
168		else:
169	0	self._timer_timeout.stop()
170
171	3	def await_result(self, timeout=None):
172		"""
173		Return the result of the call that the future represents.
174		Will not return until either timeout expires or future becomes "done".
175
176		There is one potential deadlock situation here:
177		The deadlock occurs if we await_result while at the same
178		time, this future needs to await_result from another future since
179		the eventloop will be blocked.
180		---> To be safe, don't use await_result() in a Qt slot. You should
181		rather use result() and add_done_callback() instead.
182
183		Args:
184		timeout: The number of seconds to wait for the result if the future
185		isn't done. If None, then there is no limit on the wait time.
186
187		Returns:
188		The result of the call that the future represents.
189
190		Raises:
191		CancelledError: If the future was cancelled.
192		TimeoutError: If the future didn't finish executing before the
193		given timeout.
194		Exception: If the call raised then that exception will be raised.
195		"""
196
197	0	self._wait_for_done(timeout)
198	0	return self.result()
199
200	3	def await_exception(self, timeout=None): # pragma: no-cover
201		"""
202		Return the exception raised by the call that the future represents.
203
204		Args:
205		timeout: The number of seconds to wait for the exception if the
206		future isn't done. If None, then there is no limit on the wait
207		time.
208
209		Returns:
210		The exception raised by the call that the future represents or None
211		if the call completed without raising.
212
213		Raises:
214		CancelledError: If the future was cancelled.
215		TimeoutError: If the future didn't finish executing before the
216		given timeout.
217		"""
218	0	self._wait_for_done(timeout)
219	0	return self.exception()
220
221	3	def cancel(self):
222		"""
223		Cancels the future.
224		"""
225	0	if self._timer_timeout is not None:
226	0	self._timer_timeout.stop()
227	0	super(PyrplFuture, self).cancel()
228
229
230	3	def sleep(delay):
231		"""
232		Sleeps for :code:`delay` seconds + runs the event loop in the background.
233
234		* This function will never return until the specified delay in seconds is elapsed.
235		* During the execution of this function, the qt event loop (== asyncio event-loop in pyrpl) continues to process events from the gui, or from other coroutines.
236		* Contrary to time.sleep() or async.sleep(), this function will try to achieve a precision much better than 1 millisecond (of course, occasionally, the real delay can be longer than requested), but on average, the precision is in the microsecond range.
237		* Finally, care has been taken to use low level system-functions to reduce CPU-load when no events need to be processed.
238
239		More details on the implementation can be found on the page: `<https://github.com/lneuhaus/pyrpl/wiki/Benchmark-asynchronous-sleep-functions>`_.
240		"""
241	3	tic = default_timer()
242	3	end_time = tic + delay
243
244		# 1. CPU-free sleep for delay - 1ms
245	3	if delay > 1e-3:
246	3	new_delay = delay - 1e-3
247	3	loop = QtCore.QEventLoop()
248	3	timer = MainThreadTimer(new_delay * 1000)
249	3	timer.timeout.connect(loop.quit)
250	3	timer.start()
251	3	try:
252	3	loop.exec_()
253	0	except KeyboardInterrupt as e: # pragma: no-cover
254		# try to recover from KeyboardInterrupt by finishing the current task
255	0	timer.setInterval(1)
256	0	timer.start()
257	0	loop.exec_()
258	0	raise e
259		# 2. For high-precision, manually process events 1-by-1 during the last ms
260	3	while default_timer() < end_time:
261	3	APP.processEvents()