PyWebIO/pywebio/session/threadbased.py

import logging
import queue
import threading
from functools import wraps

from .base import Session
from ..exceptions import SessionNotFoundException, SessionClosedException, SessionException
from ..utils import random_str, LimitedSizeQueue, isgeneratorfunction, iscoroutinefunction, \
    get_function_name

logger = logging.getLogger(__name__)

"""
基于线程的会话实现

当任务函数返回并且会话内所有的通过 register_thread(thread) 注册的线程都退出后，会话结束，连接关闭。
正在等待PyWebIO输入的线程会在输入函数中抛出SessionClosedException异常，
其他线程若调用PyWebIO输入输出函数会引发异常SessionException
"""


# todo 线程安全
class ThreadBasedSession(Session):
    thread2session = {}  # thread_id -> session

    unhandled_task_mq_maxsize = 1000
    event_mq_maxsize = 100
    callback_mq_maxsize = 100

    @classmethod
    def get_current_session(cls) -> "ThreadBasedSession":
        curr = id(threading.current_thread())
        session = cls.thread2session.get(curr)
        if session is None:
            raise SessionNotFoundException("Can't find current session. "
                                           "Maybe session closed or forget to use `register_thread()`.")
        return session

    @classmethod
    def get_current_task_id(cls):
        return cls._get_task_id(threading.current_thread())

    @staticmethod
    def _get_task_id(thread: threading.Thread):
        tname = getattr(thread, '_target', 'task')
        tname = getattr(tname, '__name__', tname)
        return '%s-%s' % (tname, id(thread))

    def __init__(self, target, session_info, on_task_command=None, on_session_close=None, loop=None):
        """
        :param target: 会话运行的函数. 为None时表示Script mode
        :param on_task_command: 当Task内发送Command给session的时候触发的处理函数
        :param on_session_close: 会话结束的处理函数
        :param loop: 事件循环。若 on_task_command 或者 on_session_close 中有调用使用asyncio事件循环的调用，
            则需要事件循环实例来将回调在事件循环的线程中执行
        """
        assert target is None or (not iscoroutinefunction(target)) and (not isgeneratorfunction(target)), ValueError(
            "ThreadBasedSession only accept a simple function as task function, "
            "not coroutine function or generator function. ")

        super().__init__(session_info)

        self._on_task_command = on_task_command or (lambda _: None)
        self._on_session_close = on_session_close or (lambda: None)
        self._loop = loop

        self.threads = []  # 注册到当前会话的线程集合
        self.unhandled_task_msgs = LimitedSizeQueue(maxsize=self.unhandled_task_mq_maxsize)

        self.task_mqs = {}  # task_id -> event msg queue
        self._closed = False

        # 用于实现回调函数的注册
        self.callback_mq = None
        self.callback_thread = None
        self.callbacks = {}  # callback_id -> (callback_func, is_mutex)

        if target is not None:
            self._start_main_task(target)

    def _start_main_task(self, target):

        @wraps(target)
        def main_task(target):
            try:
                target()
            except Exception as e:
                if not isinstance(e, SessionException):
                    self.on_task_exception()
            finally:
                for t in self.threads:
                    if t.is_alive() and t is not threading.current_thread():
                        t.join()

                try:
                    if self.need_keep_alive():
                        from ..session import hold
                        hold()
                    else:
                        self.send_task_command(dict(command='close_session'))
                except SessionException:  # ignore SessionException error
                    pass
                finally:
                    # we need first trigger close event and then perform close operation,
                    # because close operation will clean up all resources in this session,
                    # which may need to be accessed in close event
                    self._trigger_close_event()
                    self.close()

        thread = threading.Thread(target=main_task, kwargs=dict(target=target),
                                  daemon=True, name='main_task')
        self.register_thread(thread)

        thread.start()

    def send_task_command(self, command):
        """向会话发送来自pywebio应用的消息

        :param dict command: 消息
        """
        if self.closed():
            raise SessionClosedException()

        self.unhandled_task_msgs.put(command)

        if self._loop:
            self._loop.call_soon_threadsafe(self._on_task_command, self)
        else:
            self._on_task_command(self)

    def next_client_event(self):
        # 函数开始不需要判断 self.closed()
        # 如果会话关闭，对 get_current_session().next_client_event() 的调用会抛出SessionNotFoundException

        task_id = self.get_current_task_id()
        event_mq = self.get_current_session().task_mqs.get(task_id)
        if event_mq is None:
            raise SessionNotFoundException
        event = event_mq.get()
        if event is None:
            raise SessionClosedException
        return event

    def send_client_event(self, event):
        """向会话发送来自用户浏览器的事件️

        :param dict event: 事件️消息
        """
        task_id = event['task_id']
        mq = self.task_mqs.get(task_id)
        if not mq and task_id in self.callbacks:
            mq = self.callback_mq

        if not mq:
            logger.error('event_mqs not found, task_id:%s', task_id)
            return

        try:
            mq.put_nowait(event)  # disable blocking, because this is call by backend
        except queue.Full:
            logger.error('Message queue is full, discard new messages')  # todo: alert user

    def get_task_commands(self):
        return self.unhandled_task_msgs.get()

    def _trigger_close_event(self):
        """触发Backend on_session_close callback"""
        if self.closed():
            return
        if self._loop:
            self._loop.call_soon_threadsafe(self._on_session_close)
        else:
            self._on_session_close()

    def _cleanup(self, nonblock=False):
        cls = type(self)
        if not nonblock:
            self.unhandled_task_msgs.wait_empty(8)

        if not self.unhandled_task_msgs.empty():
            msg = self.unhandled_task_msgs.get()
            logger.warning("%d unhandled task messages when session close. [%s]", len(msg), threading.current_thread())

        for t in self.threads:
            # delete registered thread
            # so the `get_current_session()` call in those thread will raise SessionNotFoundException
            del cls.thread2session[id(t)]

        if self.callback_thread:
            del cls.thread2session[id(self.callback_thread)]

        def try_best_to_add_item_to_mq(mq, item, try_count=10):
            for _ in range(try_count):
                try:
                    mq.put(item, block=False)
                    return True
                except queue.Full:
                    try:
                        mq.get(block=False)
                    except queue.Empty:
                        pass

        if self.callback_mq is not None:  # 回调功能已经激活, 结束回调线程
            try_best_to_add_item_to_mq(self.callback_mq, None)

        for mq in self.task_mqs.values():
            try_best_to_add_item_to_mq(mq, None)  # 消费端接收到None消息会抛出SessionClosedException异常
        self.task_mqs = {}

    def close(self, nonblock=False):
        """关闭当前Session。由Backend调用"""
        # todo self._closed 会有竞争条件
        if self.closed():
            return

        super().close()

        self._cleanup(nonblock=nonblock)

    def _activate_callback_env(self):
        """激活回调功能

        ThreadBasedSession 的回调实现原理是：创建一个单独的线程用于接收回调事件，进而调用相关的回调函数。
        当用户Task中并没有使用到回调功能时，不必开启此线程，可以节省资源
        """

        if self.callback_mq is not None:  # 回调功能已经激活
            return

        self.callback_mq = queue.Queue(maxsize=self.callback_mq_maxsize)
        self.callback_thread = threading.Thread(target=self._dispatch_callback_event,
                                                daemon=True, name='callback-' + random_str(10))
        # self.register_thread(self.callback_thread)
        self.thread2session[id(self.callback_thread)] = self  # 用于在线程内获取会话
        event_mq = queue.Queue(maxsize=self.event_mq_maxsize)  # 回调线程内的用户事件队列
        self.task_mqs[self._get_task_id(self.callback_thread)] = event_mq

        self.callback_thread.start()
        logger.debug('Callback thread start')

    def _dispatch_callback_event(self):
        while not self.closed():
            event = self.callback_mq.get()
            if event is None:  # 结束信号
                logger.debug('Callback thread exit')
                break

            callback_info = self.callbacks.get(event['task_id'])
            if not callback_info:
                logger.error("No callback for callback_id:%s", event['task_id'])
                return
            callback, mutex = callback_info

            @wraps(callback)
            def run(callback):
                try:
                    callback(event['data'])
                except Exception as e:
                    # 子类可能会重写 get_current_session ，所以不要用 ThreadBasedSession.get_current_session 来调用
                    if not isinstance(e, SessionException):
                        self.on_task_exception()

                # todo: good to have -> clean up from `register_thread()`

            if mutex:
                run(callback)
            else:
                t = threading.Thread(target=run, kwargs=dict(callback=callback),
                                     daemon=True)
                self.register_thread(t)
                t.start()

    def register_callback(self, callback, serial_mode=False):
        """ 向Session注册一个回调函数，返回回调id

        :param Callable callback: 回调函数. 函数签名为 ``callback(data)``. ``data`` 参数为回调事件的值
        :param bool serial_mode: 串行模式模式。若为 ``True`` ，则对于同一组件的点击事件，串行执行其回调函数
        """
        assert (not iscoroutinefunction(callback)) and (not isgeneratorfunction(callback)), ValueError(
            "In ThreadBasedSession.register_callback, `callback` must be a simple function, "
            "not coroutine function or generator function. ")

        self._activate_callback_env()
        callback_id = 'CB-%s-%s' % (get_function_name(callback, 'callback'), random_str(10))
        self.callbacks[callback_id] = (callback, serial_mode)
        return callback_id

    def register_thread(self, t: threading.Thread):
        """将线程注册到当前会话，以便在线程内调用 pywebio 交互函数。
        会话会一直保持直到所有通过 `register_thread` 注册的线程以及当前会话的主任务线程退出

        :param threading.Thread thread: 线程对象
        """
        self.threads.append(t)  # 保存 registered thread，用于主任务线程退出后等待注册线程结束
        self.thread2session[id(t)] = self  # 用于在线程内获取会话
        event_mq = queue.Queue(maxsize=self.event_mq_maxsize)  # 线程内的用户事件队列
        self.task_mqs[self._get_task_id(t)] = event_mq

    def need_keep_alive(self) -> bool:
        # if callback thread is activated, then the session need to keep alive
        return self.callback_thread is not None


class ScriptModeSession(ThreadBasedSession):
    """Script mode的会话实现"""

    @classmethod
    def get_current_session(cls) -> "ScriptModeSession":
        if cls.instance is None:
            raise SessionNotFoundException("Can't find current session. It might be a bug.")
        if cls.instance.closed():
            raise SessionClosedException()
        return cls.instance

    @classmethod
    def get_current_task_id(cls):
        task_id = super().get_current_task_id()
        session = cls.get_current_session()
        if task_id not in session.task_mqs:
            session.register_thread(threading.current_thread())
        return task_id

    instance = None

    def __init__(self, thread, session_info, on_task_command=None, loop=None):
        """
        :param thread: 第一次调用PyWebIO交互函数的线程 todo 貌似本参数并不必要
        :param on_task_command: 会话结束的处理函数。后端Backend在相应on_session_close时关闭连接时，
            需要保证会话内的所有消息都传送到了客户端
        :param loop: 事件循环。若 on_task_command 或者on_session_close中有调用使用asyncio事件循环的调用，
            则需要事件循环实例来将回调在事件循环的线程中执行
        """
        if ScriptModeSession.instance is not None:
            raise RuntimeError("ScriptModeSession can only be created once.")
        ScriptModeSession.instance = self

        super().__init__(target=None, session_info=session_info, on_task_command=on_task_command, loop=loop)

        tid = id(thread)
        event_mq = queue.Queue(maxsize=self.event_mq_maxsize)
        self.task_mqs[tid] = event_mq