Как использовать asyncio внутри Thread Python?

Question

[r1dddy4sv]

У меня есть приложение на pyqt5, внутри функции которого я создаю потоки через измененный класс, с целью остановки потока по желанию пользователя. После чего внутри потока используется asyncio, что вызывает конфликт.

results=[]
def get_tasks(session):
    tasks = []
    
    for i in range(0,requestsNumber):
        tasks.append(asyncio.create_task(session.post(url, data = json.dumps(js), ssl=False)))
    return tasks

async def get_symbols(headers):
    async with aiohttp.ClientSession(headers=headers) as session:
        tasks = get_tasks(session)
        responses = await asyncio.gather(*tasks)
        for response in responses:
            results.append(await response.text())

def startSsc(headers):
    asyncio.get_event_loop().run_until_complete(get_symbols(headers))


class thread_with_exception(threading.Thread):
    def __init__(self,name):
        threading.Thread.__init__(self)
        self.name = name
        
   def run(self):
        #some code (основная функция для запуска в многопоточности)
        startSsc(head)

    def get_id(self):
 
        # returns id of the respective thread
        if hasattr(self, '_thread_id'):
            return self._thread_id
        for id, thread in threading._active.items():
            if thread is self:
                return id
  
    def raise_exception(self):
        thread_id = self.get_id()
        res = ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id,
              ctypes.py_object(SystemExit))
        if res > 1:
            ctypes.pythonapi.PyThreadState_SetAsyncExc(thread_id, 0)
            print('Exception raise failure')

class Ui_Dialog(object):
    def setupUi(self,code,end, Dialog):
        #some code

    def start(self):
         s = thread_with_exception(some_args)
         s.start()
         self.ap[i]=s

    def stop_all_tasks(self):
         for i in self.ap:
              self.ap[i].raise_exception()
              self.ap[i].join()
if __name__ == "__main__":
    import sys
    app = QtWidgets.QApplication(sys.argv)
    MainWindow = QtWidgets.QDialog()
    ui = Ui_Dialog()
    ui.setupUi(code,end,MainWindow)
    MainWindow.show()        
    sys.exit(app.exec_())

и внутри функции для запуска в многопоточности нужно использовать функции с использованием asyncio, из-за чего происходит конфликт

ошибка:

RuntimeError("There is no current event loop in thread '0'.")

решаю её следующими изменениями:

loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)

тогда периодически возникает следующая проблема:

ClientOSError(22, 'Указанное сетевое имя более недоступно', None, 64, None)

Есть ли способ адаптировать мой код ?

Comments

[Сергей Горностаев]

Практически всегда смешивание многопоточности с асинхронностью - это ошибка. Зачем вам это?

[r1dddy4sv]

Сергей Горностаев, основная функция, описанная в классе threading_with_exception, вызывает сессию selenium, внутри которой используется asyncio

хотелось бы что б код поддерживал запуск нескольких сессий одновременно, с
этой целью обращаюсь к вам)

Answer 1

[Vindicar]

Во-первых, несколько асинхронных задач прекрасно уживаются друг с другом. На кой тебе вообще нужны потоки для этого? Чем не устроил loop.create_task(), или asyncio.gather()?

Во-вторых, куда проще завернуть долгий синхронный код в поток, а уже поток - в асинхронную задачу, чем наоборот. Подумай над этим.

В-третьих, технически ты можешь создавать свой цикл реактора (asyncio loop) в каждом потоке. Почитай доки на asyncio как это делается. Но нужно помнить, что async-объекты из разных реакторов не дружат друг с другом. Так что подчеркиваю красным: не надо это делать. Лучше сначала попробуй по-иному подойти к проблеме.

Comments

[r1dddy4sv]

Дело в том, что основной код приложения должен продолжать выполняться после запуска всех параллельных функций, для этого я и использую потоки (если я вас правильно понял)

[Vindicar]

r1dddy4sv, у вас точно все параллельные функции - асинхронные?
Если да, то просто делаете тело программы ещё одной асинхронной функцией. Остальные запускаете через loop.create_task().
Если же тело программы строго синхронное (например, использует input() или выполняет длительные вычисления), то заворачиваете его в отдельный поток через loop.run_in_executor(). На выходе получите асинхронную функцию, которую можно запустить как обычно.

[r1dddy4sv]

Vindicar, я, видимо, не совсем смышленый. Не могли бы вы наглядно продемонстрировать выше сказанное в коде ?

[Vindicar]

r1dddy4sv, т.е. у тебя в главном потоке GUI? Что ж ты мозги-то полоскал...
Вообще посмотри в сторону qasync, оно вроде как раз про то, как подружить Qt с asyncio. Вот пример.
Либо можно поколхозить что-то свое.

def main():
    input() #тут синхронный код тела программы

async def background_task():
    #тут код долгоиграющей асинхронной задачи
    while True:
        print('beep boop')
        await asyncio.sleep(1) 

if __name__ == '__main__':
    loop = asyncio.get_event_loop()
    #создаём фоновую задачу
    loop.create_task(background_task())
    #создаём задачу для исполнения main() в отдельном потоке
    main_task = loop.run_in_executor(None, main)
    #выполняемся пока main() не отработает
    loop.run_until_complete(main_task)

С запуском новых асинхронных задач из другого потока могут быть проблемы - читай тут. И даже если задачу запустишь, получить её результат может быть еще сложнее.