import{_ as i,c as a,a7 as n,o as e}from"./chunks/framework.jrU0lkHV.js";const o=JSON.parse('{"title":"liteyuki.core.manager","description":"","frontmatter":{"title":"liteyuki.core.manager"},"headers":[],"relativePath":"en/dev/api/core/manager.md","filePath":"en/dev/api/core/manager.md","lastUpdated":1725101868000}'),t={name:"en/dev/api/core/manager.md"};function l(h,s,k,p,r,d){return e(),a("div",null,s[0]||(s[0]=[n(`

Module liteyuki.core.manager ​

Copyright (C) 2020-2024 LiteyukiStudio. All Rights Reserved

@Time : 2024/7/27 上午11:12 @Author : snowykami @Email : snowykami@outlook.com @File : manager.py @Software: PyCharm

class ChannelDeliver ​

func __init__(self, active: Channel[Any], passive: Channel[Any], channel_deliver_active: Channel[Channel[Any]], channel_deliver_passive: Channel[tuple[str, dict]], publish: Channel[tuple[str, Any]]) ​

Source code or View on GitHub
python
def __init__(self, active: Channel[Any], passive: Channel[Any], channel_deliver_active: Channel[Channel[Any]], channel_deliver_passive: Channel[tuple[str, dict]], publish: Channel[tuple[str, Any]]):
    self.active = active
    self.passive = passive
    self.channel_deliver_active = channel_deliver_active
    self.channel_deliver_passive = channel_deliver_passive
    self.publish = publish

class ProcessManager ​

func __init__(self, lifespan: Lifespan) ​

Source code or View on GitHub
python
def __init__(self, lifespan: 'Lifespan'):
    self.lifespan = lifespan
    self.targets: dict[str, tuple[Callable, tuple, dict]] = {}
    self.processes: dict[str, Process] = {}

func start_all(self) ​

Description: 对外启动方法，启动所有进程，创建asyncio task

Source code or View on GitHub
python
def start_all(self):
    for name in self.targets:
        logger.debug(f'Starting process {name}')
        threading.Thread(target=self._run_process, args=(name,), daemon=True).start()

func add_target(self, name: str, target: TARGET_FUNC, args: tuple = (), kwargs = None) ​

Description: 添加进程

Arguments:

Source code or View on GitHub
python
def add_target(self, name: str, target: TARGET_FUNC, args: tuple=(), kwargs=None):
    if kwargs is None:
        kwargs = {}
    chan_active: Channel = Channel(name=f'{name}-active')
    chan_passive: Channel = Channel(name=f'{name}-passive')
    channel_deliver = ChannelDeliver(active=chan_active, passive=chan_passive, channel_deliver_active=channel_deliver_active_channel, channel_deliver_passive=channel_deliver_passive_channel, publish=publish_channel)
    self.targets[name] = (_delivery_channel_wrapper, (target, channel_deliver, shared_memory, *args), kwargs)

func join_all(self) ​

Source code or View on GitHub
python
def join_all(self):
    for name, process in self.targets:
        process.join()

func terminate(self, name: str) ​

Description: 终止进程并从进程字典中删除

Arguments:

Source code or View on GitHub
python
def terminate(self, name: str):
    if name not in self.processes:
        logger.warning(f'Process {name} not found.')
        return
    process = self.processes[name]
    process.terminate()
    process.join(TIMEOUT)
    if process.is_alive():
        process.kill()
    logger.success(f'Process {name} terminated.')

func terminate_all(self) ​

Source code or View on GitHub
python
def terminate_all(self):
    for name in self.targets:
        self.terminate(name)

func is_process_alive(self, name: str) -> bool ​

Description: 检查进程是否存活

Arguments:

Source code or View on GitHub
python
def is_process_alive(self, name: str) -> bool:
    if name not in self.targets:
        logger.warning(f'Process {name} not found.')
    return self.processes[name].is_alive()

class _SubProcessManager ​

func __init__(self) ​

Source code or View on GitHub
python
def __init__(self):
    self.processes: dict[str, SubProcess] = {}

func new_process(self, name: str, *args, **kwargs) -> Callable[[ProcessFuncType], None] ​

Source code or View on GitHub
python
def new_process(self, name: str, *args, **kwargs) -> Callable[[ProcessFuncType], None]:

    def decorator(func: ProcessFuncType):
        self.processes[name] = SubProcess(name, func, *args, **kwargs)
    return decorator

func add(self, name: str, func: ProcessFuncType, *args, **kwargs) ​

Description: 添加子进程

Arguments:

Source code or View on GitHub
python
def add(self, name: str, func: ProcessFuncType, *args, **kwargs):
    self.processes[name] = SubProcess(name, func, *args, **kwargs)

func start(self, name: str) ​

Description: 启动指定子进程

Arguments:

Source code or View on GitHub
python
def start(self, name: str):
    if name not in self.processes:
        raise KeyError(f'Process {name} not found.')
    self.processes[name].start()

func start_all(self) ​

Description: 启动所有子进程

Source code or View on GitHub
python
def start_all(self):
    for name, process in self.processes.items():
        process.start()
        logger.debug(f'Starting process {name}')
`,62)]))}const E=i(t,[["render",l]]);export{o as __pageData,E as default};