import{_ as i,c as a,a7 as n,o as e}from"./chunks/framework.Z6UMW4JZ.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(`
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
ChannelDeliver
__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]])
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
ProcessManager
__init__(self, lifespan: Lifespan)
def __init__(self, lifespan: 'Lifespan'):
self.lifespan = lifespan
self.targets: dict[str, tuple[Callable, tuple, dict]] = {}
self.processes: dict[str, Process] = {}
start_all(self)
Description: 对外启动方法,启动所有进程,创建asyncio task
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()
add_target(self, name: str, target: TARGET_FUNC, args: tuple = (), kwargs = None)
Description: 添加进程
Arguments:
- name: 进程名,用于获取和唯一标识
- target: 进程函数
- args: 进程函数参数
- kwargs: 进程函数关键字参数,通常会默认传入chan_active和chan_passive
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)
join_all(self)
def join_all(self):
for name, process in self.targets:
process.join()
terminate(self, name: str)
Description: 终止进程并从进程字典中删除
Arguments:
- name:
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.')
terminate_all(self)
def terminate_all(self):
for name in self.targets:
self.terminate(name)
is_process_alive(self, name: str) -> bool
Description: 检查进程是否存活
Arguments:
- name:
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()
_SubProcessManager
__init__(self)
def __init__(self):
self.processes: dict[str, SubProcess] = {}
add(self, name: str, func: ProcessFuncType, *args, **kwargs)
Description: 添加子进程
Arguments:
- func: 子进程函数
- name: 子进程名称
- args: 子进程函数参数
- kwargs: 子进程函数关键字参数
def add(self, name: str, func: ProcessFuncType, *args, **kwargs):
self.processes[name] = SubProcess(name, func, *args, **kwargs)
start(self, name: str)
Description: 启动指定子进程
Arguments:
- name: 子进程名称
def start(self, name: str):
if name not in self.processes:
raise KeyError(f'Process {name} not found.')
self.processes[name].start()
start_all(self)
Description: 启动所有子进程
def start_all(self):
for name, process in self.processes.items():
process.start()
logger.debug(f'Starting process {name}')
terminate(self, name: str)
Description: 终止指定子进程
Arguments:
- name: 子进程名称
def terminate(self, name: str):
if name not in self.processes:
raise KeyError(f'Process {name} not found.')
self.processes[name].terminate()
terminate_all(self)
Description: 终止所有子进程
def terminate_all(self):
for name, process in self.processes.items():
process.terminate()
logger.debug(f'Terminating process {name}')
get_process(self, name: str) -> SubProcess | None
Description: 获取指定子进程
Arguments:
- name: 子进程名称
def get_process(self, name: str) -> SubProcess | None:
return self.processes.get(name, None)