senset/vnpy
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

[bug fix] 兼容linux

Browse Source
This commit is contained in:
msincenselee 2020-03-19 23:39:25 +08:00
parent d228ca9559
commit d653da2097
1 changed files with 32 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

41
vnpy/api/rest/rest_client.py
View File

@ -7,7 +7,10 @@ import traceback
import uuid import uuid
from datetime import datetime from datetime import datetime
from enum import Enum from enum import Enum
import platform
from multiprocessing.dummy import Pool from multiprocessing.dummy import Pool
from concurrent.futures import ThreadPoolExecutor
from threading import Lock, Thread from threading import Lock, Thread
from types import TracebackType from types import TracebackType
from typing import Any, Callable, List, Optional, Type, Union from typing import Any, Callable, List, Optional, Type, Union
@ -138,8 +141,16 @@ class RestClient(object):
self.proxies = None self.proxies = None
self.thread_executor = ThreadPoolExecutor(max_workers=os.cpu_count() * 20)
self._tasks_lock = Lock() self._tasks_lock = Lock()
self._tasks: List[multiprocessing.pool.AsyncResult] = []
#self._tasks: List[multiprocessing.pool.AsyncResult] = []
self._tasks = []
self._executor_lock = Lock()
self._executors = []
self._sessions_lock = Lock() self._sessions_lock = Lock()
self._sessions: List[requests.Session] = [] self._sessions: List[requests.Session] = []
@ -275,20 +286,32 @@ class RestClient(object):
extra=extra, extra=extra,
client=self, client=self,
) )
task = pool.apply_async( if str(platform.system()) == 'Windows':
self._process_request, task = pool.apply_async(
args=[request, ], self._process_request,
callback=self._clean_finished_tasks, args=[request, ],
# error_callback=lambda e: self.on_error(type(e), e, e.__traceback__, request), callback=self._clean_finished_tasks,
) # error_callback=lambda e: self.on_error(type(e), e, e.__traceback__, request),
self._push_task(task) )
self._push_task(task)
else:
self._clean_finished_executors()
executor = self.thread_executor.submit(self._process_request, request)
with self._executor_lock:
self._executors.append(executor)
return request return request
def _push_task(self, task): def _push_task(self, task):
with self._tasks_lock: with self._tasks_lock:
self._tasks.append(task) self._tasks.append(task)
def _clean_finished_tasks(self, result: None): def _clean_finished_executors(self,):
with self._executor_lock:
not_finished_executors = [i for i in self._executors if not i.done()]
self._executors = not_finished_executors
def _clean_finished_tasks(self, result = None):
with self._tasks_lock: with self._tasks_lock:
not_finished_tasks = [i for i in self._tasks if not i.ready()] not_finished_tasks = [i for i in self._tasks if not i.ready()]
self._tasks = not_finished_tasks self._tasks = not_finished_tasks