| Server IP : 104.21.30.238 / Your IP : 216.73.216.145 Web Server : LiteSpeed System : Linux premium151.web-hosting.com 4.18.0-553.44.1.lve.el8.x86_64 #1 SMP Thu Mar 13 14:29:12 UTC 2025 x86_64 User : tempvsty ( 647) PHP Version : 8.0.30 Disable Function : NONE MySQL : OFF | cURL : ON | WGET : ON | Perl : ON | Python : ON | Sudo : OFF | Pkexec : OFF Directory : /././././proc/self/root/opt/cloudlinux/venv/lib/python3.11/site-packages/ssa/internal/ |
Upload File : |
# -*- coding: utf-8 -*-
# Copyright © Cloud Linux GmbH & Cloud Linux Software, Inc 2010-2021 All Rights Reserved
#
# Licensed under CLOUD LINUX LICENSE AGREEMENT
# http://cloudlinux.com/docs/LICENSE.TXT
import logging
import signal
import sys
from typing import Union, Iterable, Callable
from .utils import switch_schedstats, no_xray_active_tasks
from ..modules.processor import RequestProcessor
logger = logging.getLogger('tools')
def log_on_receive(signum: int, frame) -> None:
"""
Signal handler to log received signal
"""
logger.info('Received sig: %i (%s)', signum, signum)
def sigterm_received(signum: int, frame) -> None:
"""
Signal handler to dump collected statistics and switch off schedstats
"""
logger.info('Received sig: %i', signum)
RequestProcessor().flush_buffer()
logger.info('Dumped successfully after %i signal, exiting...', signum)
# disable throttling detection kernel mechanism
# if there is no X-Ray tracing tasks active
if no_xray_active_tasks():
logger.info('No X-Ray tasks running, switching schedstats off')
switch_schedstats(enabled=False)
sys.exit(0)
def register_a_signal(signum: int, sighandler: Callable) -> None:
"""
Register given sighandler for given signum
"""
try:
signal.signal(signum, sighandler)
logger.info('Registered %s', signum)
except (OSError, ValueError) as e:
logger.info('Skipped %s because of %s', signum, e)
def register_sigterm():
"""
Register a dump handler for SIGTERM signal
"""
register_a_signal(signal.SIGTERM, sigterm_received)
def register_signals(signals: Union[int, Iterable] = None,
handler: Callable = log_on_receive) -> None:
"""
Register given handler for given signal or range of signals
If no signals given, add given handler for all available signals
"""
if signals is None:
# register all available signals
for sig in signal.Signals:
register_a_signal(sig, handler)
else:
if isinstance(signals, int):
# register single signal
register_a_signal(signals, handler)
else:
# register range of signals
for sig in signals:
register_a_signal(sig, handler)