kaiyuan/chatgpt-on-wechat
1
0
Fork 0
mirror of https://github.com/zhayujie/chatgpt-on-wechat.git synced 2026-07-20 05:27:59 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: try to solve the unresponsiveness problem

Browse Source
This commit is contained in:
Saboteur7
2024-10-19 15:49:57 +08:00
parent 883f0d449b
commit 70e007d8ca
1 changed files with 14 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

25
channel/chat_channel.py
View File

@@ -337,24 +337,27 @@ class ChatChannel(Channel):
while True: while True:
with self.lock: with self.lock:
session_ids = list(self.sessions.keys()) session_ids = list(self.sessions.keys())
for session_id in session_ids: for session_id in session_ids:
with self.lock:
context_queue, semaphore = self.sessions[session_id] context_queue, semaphore = self.sessions[session_id]
if semaphore.acquire(blocking=False): # 等线程处理完毕才能删除 if semaphore.acquire(blocking=False): # 等线程处理完毕才能删除
if not context_queue.empty(): if not context_queue.empty():
context = context_queue.get() context = context_queue.get()
logger.debug("[chat_channel] consume context: {}".format(context)) logger.debug("[chat_channel] consume context: {}".format(context))
future: Future = handler_pool.submit(self._handle, context) future: Future = handler_pool.submit(self._handle, context)
future.add_done_callback(self._thread_pool_callback(session_id, context=context)) future.add_done_callback(self._thread_pool_callback(session_id, context=context))
with self.lock:
if session_id not in self.futures: if session_id not in self.futures:
self.futures[session_id] = [] self.futures[session_id] = []
self.futures[session_id].append(future) self.futures[session_id].append(future)
elif semaphore._initial_value == semaphore._value + 1: # 除了当前，没有任务再申请到信号量，说明所有任务都处理完毕 elif semaphore._initial_value == semaphore._value + 1: # 除了当前，没有任务再申请到信号量，说明所有任务都处理完毕
with self.lock:
self.futures[session_id] = [t for t in self.futures[session_id] if not t.done()] self.futures[session_id] = [t for t in self.futures[session_id] if not t.done()]
assert len(self.futures[session_id]) == 0, "thread pool error" assert len(self.futures[session_id]) == 0, "thread pool error"
del self.sessions[session_id] del self.sessions[session_id]
else: else:
semaphore.release() semaphore.release()
time.sleep(0.1) time.sleep(0.2)
# 取消session_id对应的所有任务，只能取消排队的消息和已提交线程池但未执行的任务 # 取消session_id对应的所有任务，只能取消排队的消息和已提交线程池但未执行的任务
def cancel_session(self, session_id): def cancel_session(self, session_id):