brige/human
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 2 Wiki Activity

modify mutil-thread

Browse Source
This commit is contained in:
brige 2024-11-15 21:23:34 +08:00
parent 7eccc99c2a
commit aa554b1209
2 changed files with 18 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
asr/sherpa_ncnn_asr.py
View File

@ -58,7 +58,7 @@ class SherpaNcnnAsr(AsrBase):
def _recognize_loop(self): def _recognize_loop(self):
segment_id = 0 segment_id = 0
time.sleep(3) time.sleep(9)
last_result = "" last_result = ""
logger.info(f'_recognize_loop') logger.info(f'_recognize_loop')
print(f'_recognize_loop') print(f'_recognize_loop')

39
utils/async_task_queue.py
View File

@ -1,6 +1,5 @@
#encoding = utf8 #encoding = utf8
import asyncio
import logging import logging
from queue import Queue from queue import Queue
import threading import threading
@ -14,24 +13,14 @@ class AsyncTaskQueue:
self._worker_num = work_num self._worker_num = work_num
self._current_worker_num = work_num self._current_worker_num = work_num
self._name = name self._name = name
self._thread = threading.Thread(target=self._run_loop, name=name) self._threads = []
self._thread.start() self._lock = threading.Lock()
self.__loop = None for _ in range(work_num):
thread = threading.Thread(target=self._worker, name=f'{name}_worker_{_}')
thread.start()
self._threads.append(thread)
def _run_loop(self): def _worker(self):
logging.info(f'{self._name}, _run_loop')
self.__loop = asyncio.new_event_loop()
asyncio.set_event_loop(self.__loop)
self._tasks = [self.__loop.create_task(self._worker()) for _ in range(self._worker_num)]
try:
self.__loop.run_forever()
finally:
logging.info(f'{self._name}, exit run')
self.__loop.run_until_complete(asyncio.gather(*asyncio.all_tasks(self.__loop)))
self.__loop.close()
logging.info(f'{self._name}, close loop')
async def _worker(self):
logging.info(f'{self._name}, _worker') logging.info(f'{self._name}, _worker')
while True: while True:
try: try:
@ -43,16 +32,17 @@ class AsyncTaskQueue:
if func is None: # None as a stop signal if func is None: # None as a stop signal
break break
await func(*args) # Execute async function func(*args) # Execute function
except Exception as e: except Exception as e:
logging.error(f'{self._name} error: {repr(e)}') logging.error(f'{self._name} error: {repr(e)}')
finally: finally:
self._queue.task_done() self._queue.task_done()
logging.info(f'{self._name}, _worker finish') logging.info(f'{self._name}, _worker finish')
with self._lock:
self._current_worker_num -= 1 self._current_worker_num -= 1
if self._current_worker_num == 0: if self._current_worker_num == 0:
self.__loop.call_soon_threadsafe(self.__loop.stop) self._queue.put(None) # Send stop signal to remaining workers
def add_task(self, func, *args): def add_task(self, func, *args):
self._queue.put((func, *args)) self._queue.put((func, *args))
@ -62,10 +52,11 @@ class AsyncTaskQueue:
self.add_task(None) # Send stop signal self.add_task(None) # Send stop signal
def clear(self): def clear(self):
while not self._queue.empty(): with self._queue.mutex:
self._queue.get_nowait() self._queue.queue.clear()
self._queue.task_done()
def stop(self): def stop(self):
self.stop_workers() self.stop_workers()
self._thread.join() for thread in self._threads:
thread.join()