2
\$\begingroup\$

This is my first shot at asyncio. I'm looking for any pointers you might have. Especially:

  • Are there any race conditions?
  • This code doesn't scale as expected. In my project I need to call caget 4k times.

As a background thread I need to make 4k network calls (caget, which is a third party function I can't change) which normally return a result in a timely manner but if the serves is unavailable might time out. I need to be able to cancel the calls at any point. In addition I need a live update of how many calls succeeded. Results of the calls need to be only available if the process was not interrupted and finished.

#!/usr/bin/env python3

import sys
from time import sleep
from asyncio import Task, new_event_loop, gather, set_event_loop
from concurrent.futures import ThreadPoolExecutor, CancelledError
from threading import Timer
from collections import namedtuple
from random import random

from PyQt5.QtWidgets import (QWidget, QApplication, QLabel, QPushButton,
    QVBoxLayout)
from PyQt5.QtCore import pyqtSignal, QThread


def caget(delay):
    ''' This is a non-corutine third party function. It does a local network call
    and waits for its result to return. '''
    # The actual function call would be epics.caget for interested
    # http://cars9.uchicago.edu/software/python/pyepics3/
    # The results are then taken further used for an optics model.
    # The sleep call is placed here to simplify the problem and keep the
    # focus on the asyncio part rather than the lengthy rest of the code
    # which was omited for that exact purpose. (Part of a framework with about 10k lines)
    delay = delay / 50 + random()
    sleep(delay)
    return delay


# Is this wraper function for caget really necessary? Respectively is there a
# a more resource friendly way to do it? I potentially have to ask for about
# 4k of values.
async def caget_wrapper(loop, num):
    return await loop.run_in_executor(ThreadPoolExecutor(), caget, num)


def update_factory(signal, loop):
    ''' Returns a str of the form (done / total) '''
    def intern():
        tasks = Task.all_tasks(loop=loop)
        done = sum(task.done() for task in tasks)
        signal.emit(f'{done} / {len(tasks)}')
    return intern


class IntervalTimer(namedtuple('IntervalTimer', ['interval', 'fun'])):
    ''' Repetitive timer function. Note that like for the normal timer it needs
    to be started from the thread it lives in. '''

    def start(self):
        self.timer = Timer(self.interval, self.run)
        self.timer.start()

    def stop(self):
        self.timer.cancel()

    def run(self):
        self.start()
        self.fun()


class Worker(QThread):
    TOTAL = 100

    def run(self):
        # We need to create our own loop since QThread doesn't start with having
        # one running.
        self.loop = new_event_loop()
        set_event_loop(self.loop)

        timer = IntervalTimer(1 / 25, update_factory(self.parent().log, self.loop))
        timer.start()

        tasks = [caget_wrapper(self.loop, i) for i in range(self.TOTAL)]
        try:
            result = self.loop.run_until_complete(gather(*tasks))
            timer.stop()
            self.parent().log.emit(f'Mean {sum(result) / self.TOTAL:.3f}')
        except CancelledError:
            timer.stop()
            self.parent().log.emit(f'Abort')

    def abort(self):
        ''' Kill all the tasks. Note that we do not stop the event loop. This will
        most likely raise an CancelledError in the thread the event loop runs. '''
        for task in Task.all_tasks(loop=self.loop):
            task.cancel()


class Main(QWidget):
    log = pyqtSignal(str)

    def __init__(self):
        super().__init__()
        self.worker = Worker(self)
        layout = QVBoxLayout(self)

        self.ctrl = QPushButton(self)
        self.stop_worker()
        self.ctrl.clicked.connect(self.ctrl_callback)
        layout.addWidget(self.ctrl)

        self.log_label = QLabel('init', self)
        layout.addWidget(self.log_label)
        self.log.connect(lambda txt: self.log_label.setText(txt))

        self.worker.started.connect(self.start_worker)
        self.worker.finished.connect(self.stop_worker)

    def ctrl_callback(self):
        self._ctrl_callback()

    def start_worker(self)  :
        self.ctrl.setText('Abort')
        self._ctrl_callback = self.worker.abort

    def stop_worker(self):
        self.ctrl.setText('Start')
        self._ctrl_callback = self.worker.start


if __name__ == '__main__':
    app = QApplication(sys.argv)
    main = Main()
    main.show()
    sys.exit(app.exec_())

enter image description here

\$\endgroup\$
  • 2
    \$\begingroup\$ This doesn't look like real code (a function named foo?) can you please clarify or edit to include your real code? \$\endgroup\$ – Phrancis Oct 9 '17 at 19:04
  • \$\begingroup\$ @Phrancis. This will be a part of a bigger framework (framework is still not ready for public release but already has about 10k loc). I wanted to keep it simple so that people can focus on the async/threading part. However I added some additional information as comments. \$\endgroup\$ – magu_ Oct 9 '17 at 20:36
  • \$\begingroup\$ You might make this on-topic by providing a suitable mock for the foo()operation that shows its unpredictable timing and return value. \$\endgroup\$ – Toby Speight Oct 10 '17 at 12:21
  • \$\begingroup\$ @TobySpeight I try to get this question on topic, however I seem to fail to see the root cause of the problem (I read the faq of off-topic on CR). I changed foo to caget and added its original source location plus a random factor. If needed I can also write import epics however I don't believe that most people have EPICS installed and the channel access doesn't actually provide any added value for the problem, at least in my opinion. Is my mockup function (foo/caget) the only problem or is this only the tip of the iceberg? \$\endgroup\$ – magu_ Oct 10 '17 at 16:38

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.