import abc import os import selectors as _selectors import socket as _socket import sys import threading as _threading from concurrent.futures import Executor, ThreadPoolExecutor as _ThreadPoolExecutor from selectors import DefaultSelector from threading import Thread from time import time from typing import Set, Type, Union, TypeVar, Tuple, Any, Optional from biopro import NAME, VERSION, COPYRIGHT from biopro.controller import ComplexResponse from biopro.util.console import pp, foreground as _fc, LINE_BREAK from biopro.util.json import JsonSerialize from biopro.util.logger import LoggerFlag, Logger from biopro.util.sleep import long_sleep from biopro.util.socket import socket_send, socket_recv, Socket, SocketTimeoutError from biopro.util.stack import * from .dispatcher import APIDispatcher _RUNTIME_COMPILE = False if sys.version_info.minor < 6: # noinspection PyUnusedLocal def ThreadPoolExecutor(max_workers: int = 16, thread_name_prefix = None): return _ThreadPoolExecutor(max_workers=max_workers) else: ThreadPoolExecutor = _ThreadPoolExecutor class ServerThreadClosedInterrupt(RuntimeError): """server thread has closed""" __slots__ = () class ServerShutdownException(RuntimeError): """server shutdown RuntimeError as signal""" __slots__ = () class ServerRestartException(RuntimeError): """server restart RuntimeError as signal.""" SERVER_RESTART_EXIT = 4 """process exit code for restart server""" SERVER_DUMMY_EXIT = 5 """process exit code for restart in dummy mode""" __slots__ = ('mode',) def __init__(self, mode: int = SERVER_RESTART_EXIT): if mode not in (4, 5): raise ValueError() self.mode = mode T_ServerThread = TypeVar('T_ServerThread', bound='ServerThread') class SocketServer(Logger, metaclass=abc.ABCMeta): """the server which provide the (file) socket protocol. """ SOCKET_ACCEPT_KEY = 'SOCKET_ACCEPT_KEY' __slots__ = ('__socket_file', '__selector', '__socket', '__socket_listen', '__server_name', '__dispatcher', '__log_level', '__worker_thread', '__executor', '__response_counter', '__close_flag', '__interrupt', '__thread_pool_count', '__mqtt_server_ip', '__mqtt_server_port' '__db_host', '__db_name', '__db_user', '__db_password', '_threads') def __init__(self, api_class: Union[Type, Tuple[Type, ...]], server_name: str, socket_file: str, socket_listen = 1, thread_pool_count=16): """ :param api_class: api provide. :param server_name: server name :param socket_file: socket filename :param socket_listen: number of listen socket port :param thread_pool_count: number of thread in pool. """ self.__server_name = server_name # socket relative self.__socket_file = socket_file self.__selector = None self.__socket = None self.__socket_listen = socket_listen # method dispatcher self.__dispatcher = APIDispatcher(api_class, self) # general server thread self.__worker_thread = set() # self._executor = ThreadPoolExecutor(max_workers=8) self._threads = [] self._event = _threading.Event() # response thread self.__executor = None self.__response_counter = 0 # close flag self.__close_flag = False self.__interrupt = None # worker control self.__thread_pool_count = thread_pool_count # mqtt control options self.__mqtt_server_ip = '52.194.17.114' self.__mqtt_server_port = 1883 # database constrol options self.__db_host = '127.0.0.1' self.__db_name = 'postgres' self.__db_user = 'biopro' self.__db_password = 'BioProControlBox' @property def server_name(self) -> str: return self.__server_name @property def socket_file(self) -> str: return self.__socket_file @property def socket_db_host(self) -> str: return self.__db_host @property def socket_db_name(self) -> str: return self.__db_name @property def socket_db_user(self) -> str: return self.__db_user @property def socket_db_password(self) -> str: return self.__db_password @property def socket_mqtt_port(self) -> int: return self.__mqtt_server_port @property def socket_mqtt_ip(self) -> str: return self.__mqtt_server_ip def interrupt(self, exception: BaseException) -> None: """interrupt server main thread with non-rescue ``exception``. :param exception: error be raised in the main server thread. :return: None """ if not _RUNTIME_COMPILE: self.log_info('interrupt with', exception.__class__.__name__) self.__interrupt = exception def is_interrupted(self) -> bool: if self.__interrupt is not None: if isinstance(self.__interrupt, ServerRestartException): raise ServerRestartException(self.__interrupt.mode) else: return True return self.__close_flag @property def is_closed(self) -> bool: return self.__close_flag @abc.abstractmethod def setup(self): """call when server starting. """ pass @abc.abstractmethod def close(self, external_interrupt: bool): """call when server closing. :param external_interrupt: interrupt from external Ctrl+C """ pass def shutdown_server(self): self.log_info('shutdown server') self.interrupt(ServerShutdownException()) def restart_server(self): self.log_info('restart server') self.interrupt(ServerRestartException()) # noinspection PyBroadException def main(self) -> int: """server main function. :return: exit code. """ # print server information self.log_info(LINE_BREAK) self.log_info(NAME, VERSION, self.__server_name) self.log_info(COPYRIGHT) self.log_info(LINE_BREAK) self.log_verbose('start', self.__server_name, 'process') # initialize thread pool. if self.__thread_pool_count > 0: executor = ThreadPoolExecutor(thread_name_prefix=self.__server_name, max_workers=self.__thread_pool_count) else: executor = None self.__executor = executor self.__close_flag = False exit_code = 0 external_interrupt = False # this function focus on error handling. try: self._handle_socket_open() try: self.setup() self._handle_socket_routine_forever() finally: self._handle_socket_close() # close response thread self.__close_flag = True if executor is not None: executor.shutdown(wait=False) except KeyboardInterrupt: # user or system send SIGTERM signal print("") self.log_warn('interrupt captured') external_interrupt = True except ServerShutdownException: self.log_verbose('catch server shutdown signal') exit_code = 0 except ServerRestartException as e: self.log_verbose('catch server restart signal') exit_code = e.mode except OSError: e_type, e_value, e_stack = sys.exc_info() if '[Errno 98] Address already in use' in str(e_value): # socket file already in used, may has another process is using it # or previous process forget to delete it. self.log_error('server fatal error : process has already run') exit_code = 1 if not _RUNTIME_COMPILE: print_exception(self) except: # unknown error if not _RUNTIME_COMPILE: print_exception(self) finally: self.log_verbose('close service') try: self.broadcast_command('shutdown') except: self.log_verbose('broadcast_command shutdown error') try: # close server thread # copy a list avoid worker closing change the content # which would raise RuntimeError: Set changed size during iteration current_worker = list(self.__worker_thread) for worker in current_worker: worker.close() self.close(external_interrupt) if external_interrupt: # wait all thread if server shutdown by interrupt for worker in current_worker: worker.join(5) for worker in current_worker: if worker.thread.is_alive(): self.log_verbose(worker.log_name, 'still alive') except KeyboardInterrupt: pass except: if not _RUNTIME_COMPILE: print_exception(self) self.log_info('server', self.__server_name, 'exit', exit_code) return exit_code def _handle_socket_open(self): """open file socket. """ # create selector, # let OS notify us when the file-object (in here, is socket) is ready to be read. # Therefore, we can write code without being blocked by IO function. if self.__selector is not None or self.__socket is not None: raise RuntimeError('SocketHandler has opened') if os.path.exists(self.__socket_file): os.remove(self.__socket_file) self.__selector = DefaultSelector() self.__socket = Socket(_socket.AF_UNIX, _socket.SOCK_STREAM) logger = self if isinstance(logger, LoggerFlag): logger.log_verbose('listen', self.__socket_file) self.__socket.bind(self.__socket_file) self.__socket.listen(self.__socket_listen) self.__socket.setblocking(False) # register socket in selector. # When it is ready to read, we can get the event at selector.select() self.__selector.register(self.__socket, _selectors.EVENT_READ, self.SOCKET_ACCEPT_KEY) def _handle_socket_close(self): """close socket. """ selector = self.__selector socket = self.__socket if socket is not None: socket.close() if selector is not None: selector.close() if os.path.exists(self.__socket_file): os.remove(self.__socket_file) self.__selector = None self.__socket = None def _handle_socket_routine_once(self, timeout: int = 1): """listen socket incoming at least *timeout* second. Once socket connected, call :func:`handle_socket_accept`. :param timeout: """ selector = self.__selector if selector is None: raise RuntimeError('SocketHandler not open') e = selector.select(timeout) if self.is_interrupted(): return for k, _ in e: if k.data == self.SOCKET_ACCEPT_KEY: socket = k.fileobj cnt, _ = socket.accept() cnt.setblocking(False) self._handle_socket_accept(cnt) def _handle_socket_routine_forever(self): """listen socket incoming forever, unit :func:`is_interrupted` not True. Once socket connected, call :func:`handle_socket_accept`. """ selector = self.__selector if selector is None: raise RuntimeError('SocketHandler not open') while not self.is_interrupted(): e = selector.select(1) if self.is_interrupted(): return for k, _ in e: if k.data == self.SOCKET_ACCEPT_KEY: socket = k.fileobj cnt, _ = socket.accept() cnt.setblocking(False) self._handle_socket_accept(cnt) def _handle_socket_accept(self, socket: Socket): c = self.__response_counter self.__response_counter += 1 # pass incoming socket to task thread. self.submit_thread(self.__handle_socket_accept, c, socket, time()) def __handle_socket_accept(self, counter: int, socket: _socket.socket, init_time: float): start_time = time() wait_time = (start_time - init_time) * 1000 thread = _threading.current_thread() # give task thread a attribute 'log_header' # make it work like a LoggerFlag in LoggerFormatter. # it will replace the default thread name with our 'header' in logging message. header = pp('%04d' % counter, _fc(counter), ']') thread.log_header = header try: self._handle_socket_communicate(socket) except (ServerRestartException, ServerShutdownException) as e: self.interrupt(e) except BrokenPipeError: self.log_verbose(header, 'broken pipe') except SocketTimeoutError: self.log_verbose(header, 'timeout') except: print_exception(self) finally: if not _RUNTIME_COMPILE: close_time = time() live_time = (close_time - start_time) * 1000 total_time = (close_time - init_time) * 1000 if wait_time > 1000: wait_time = '%.1f s,' % (wait_time / 1000) else: wait_time = '%.2f ms,' % wait_time if live_time > 1000: live_time = '%.1f s,' % (live_time / 1000) else: live_time = '%.2f ms,' % live_time if total_time > 1000: total_time = '%.1f s' % (total_time / 1000) else: total_time = '%.2f ms' % total_time self.log_verbose('wait', wait_time, 'live', live_time, 'total', total_time) del thread.log_header def _handle_socket_communicate(self, socket: Socket): """communicate with socket. :param socket: """ try: go = True while not self.is_interrupted() and go: recv = socket_recv(socket, 3) if recv is None: break send, go = self._handle_socket_receive(recv) socket_send(socket, send) except SocketTimeoutError: pass finally: socket.close() def _handle_socket_receive(self, content: Any) -> Tuple[Any, bool]: if not isinstance(content, tuple) or len(content) == 0: return None, False try: method, argument, options = content except ValueError: return None, False try: ret = self.__dispatcher.method_dispatch(method, argument, options) except BaseException as e: if isinstance(self, Logger): print_exception(self) return RuntimeError(e.__class__.__name__, *e.args), True else: if isinstance(ret, ComplexResponse): return ComplexResponse(JsonSerialize.to_json(ret.response), ret), True else: return JsonSerialize.to_json(ret), True def submit_thread(self, f, *args): """submit task to thread pool. :param f: callable :param args: *f* 's arguments """ if self.__executor is not None: self.__executor.submit(f, *args) else: # do not make exception impact main thread, # ket its behavior as same as using executor try: f(*args) except ServerShutdownException as e: self.interrupt(e) except ServerThreadClosedInterrupt: pass except: print_exception(self) def run_thread(self, thread: T_ServerThread) -> T_ServerThread: """start daemon thread. :param thread: server thread instance :return: *thread* itself. """ if thread is None: raise TypeError('NoneType for ServerThread') thread.log_verbose('start') if self.__executor is not None: self._threads.append(self.__executor.submit(self._run_thread, thread)) # Thread(target=self._run_thread, # name=thread.log_name, # args=(thread,), # daemon=True).start() return thread def _run_thread(self, thread: 'ServerThread'): try: self.__worker_thread.add(thread) thread.log_verbose('setup') thread.setup() if not thread._event.is_set(): thread.log_verbose('routine') while not thread._event.is_set(): thread.routine(thread._event) except ServerShutdownException as e: self.interrupt(e) except ServerThreadClosedInterrupt: pass except: print_exception(self) finally: try: thread.log_verbose('closing') thread.shutdown() except KeyboardInterrupt: thread.log_warn('closing interrupt') finally: try: self.__worker_thread.remove(thread) except KeyError: pass thread.log_verbose('closed') class ServerThread(LoggerFlag, metaclass=abc.ABCMeta): """server thread.""" def __init__(self, name: str, header: Union[str, int, Tuple[Any, ...]] = None): super().__init__(name, header) self.__thread = None self.__closed = False self.__condition = _threading.Condition() self._event = _threading.Event() @property def thread(self) -> Optional[_threading.Thread]: return self.__thread def join(self, timeout: Optional[float] = None): if self.__thread is not None: self.__thread.join(timeout) def setup(self) -> None: """call when thread is started""" # reset close flag self.__closed = False # get current thread self.__thread = _threading.current_thread() return @abc.abstractmethod def routine(self, event = None) -> None: """routine logic""" return def shutdown(self) -> None: """call when thread is closing""" return def close(self) -> None: """close this thread""" if not self.__closed: self.log_verbose('close') self.__closed = True self._event.set() self.notify_all() @property def is_startup(self): """ :return: is this thread setup """ return self.__thread is not None and not self.__closed @property def is_closed(self): """ :return: is this thread closing """ return self.__closed def sleep(self, second: float, disable_interrupt=False): """long period sleep""" if long_sleep(second, lambda: self.__closed): if not disable_interrupt: raise ServerThreadClosedInterrupt() def wait(self, timeout: Optional[float] = None): with self.__condition: self.log_verbose('wait') self.__condition.wait(timeout) self.log_verbose('wake') def notify_all(self): with self.__condition: self.log_verbose('notify') self.__condition.notify_all() def __str__(self): return type(self).__name__ def __repr__(self): return str(self)