ULTRA [ www.resource.optimyar.com ]

Linux uranus.pws-dns.net 5.14.0-427.42.1.el9_4.x86_64 #1 SMP PREEMPT_DYNAMIC Fri Nov 1 14:58:02 EDT 2024 x86_64

LiteSpeed

31.214.174.197 | Your IP: 216.73.216.208

Cant read [ /etc/named.conf ]

[ drwxr-xr-x ]

/ proc / self / root / opt / alt / python39 / lib64 / python3.9 / test / libregrtest /

Home Upload Mass Deface Mass Delete Back Connect Notify

win_utils.py

import _overlapped
import _thread
import _winapi
import math
import struct
import winreg

# Seconds per measurement
SAMPLING_INTERVAL = 1
# Exponential damping factor to compute exponentially weighted moving average
# on 1 minute (60 seconds)
LOAD_FACTOR_1 = 1 / math.exp(SAMPLING_INTERVAL / 60)
# Initialize the load using the arithmetic mean of the first NVALUE values
# of the Processor Queue Length
NVALUE = 5

class WindowsLoadTracker():
    """
    This class asynchronously reads the performance counters to calculate
    the system load on Windows.  A "raw" thread is used here to prevent
    interference with the test suite's cases for the threading module.
    """

def __init__(self):
        # Pre-flight test for access to the performance data;
        # `PermissionError` will be raised if not allowed
        winreg.QueryInfoKey(winreg.HKEY_PERFORMANCE_DATA)

self._values = []
        self._load = None
        self._running = _overlapped.CreateEvent(None, True, False, None)
        self._stopped = _overlapped.CreateEvent(None, True, False, None)

_thread.start_new_thread(self._update_load, (), {})

def _update_load(self,
                    # localize module access to prevent shutdown errors
                     _wait=_winapi.WaitForSingleObject,
                     _signal=_overlapped.SetEvent):
        # run until signaled to stop
        while _wait(self._running, 1000):
            self._calculate_load()
        # notify stopped
        _signal(self._stopped)

def _calculate_load(self,
                        # localize module access to prevent shutdown errors
                        _query=winreg.QueryValueEx,
                        _hkey=winreg.HKEY_PERFORMANCE_DATA,
                        _unpack=struct.unpack_from):
        # get the 'System' object
        data, _ = _query(_hkey, '2')
        # PERF_DATA_BLOCK {
        #   WCHAR Signature[4]      8 +
        #   DWOWD LittleEndian      4 +
        #   DWORD Version           4 +
        #   DWORD Revision          4 +
        #   DWORD TotalByteLength   4 +
        #   DWORD HeaderLength      = 24 byte offset
        #   ...
        # }
        obj_start, = _unpack('L', data, 24)
        # PERF_OBJECT_TYPE {
        #   DWORD TotalByteLength
        #   DWORD DefinitionLength
        #   DWORD HeaderLength
        #   ...
        # }
        data_start, defn_start = _unpack('4xLL', data, obj_start)
        data_base = obj_start + data_start
        defn_base = obj_start + defn_start
        # find the 'Processor Queue Length' counter (index=44)
        while defn_base < data_base:
            # PERF_COUNTER_DEFINITION {
            #   DWORD ByteLength
            #   DWORD CounterNameTitleIndex
            #   ... [7 DWORDs/28 bytes]
            #   DWORD CounterOffset
            # }
            size, idx, offset = _unpack('LL28xL', data, defn_base)
            defn_base += size
            if idx == 44:
                counter_offset = data_base + offset
                # the counter is known to be PERF_COUNTER_RAWCOUNT (DWORD)
                processor_queue_length, = _unpack('L', data, counter_offset)
                break
        else:
            return

# We use an exponentially weighted moving average, imitating the
        # load calculation on Unix systems.
        # https://en.wikipedia.org/wiki/Load_(computing)#Unix-style_load_calculation
        # https://en.wikipedia.org/wiki/Moving_average#Exponential_moving_average
        if self._load is not None:
            self._load = (self._load * LOAD_FACTOR_1
                            + processor_queue_length  * (1.0 - LOAD_FACTOR_1))
        elif len(self._values) < NVALUE:
            self._values.append(processor_queue_length)
        else:
            self._load = sum(self._values) / len(self._values)

def close(self, kill=True):
        self.__del__()
        return

def __del__(self,
                # localize module access to prevent shutdown errors
                _wait=_winapi.WaitForSingleObject,
                _close=_winapi.CloseHandle,
                _signal=_overlapped.SetEvent):
        if self._running is not None:
            # tell the update thread to quit
            _signal(self._running)
            # wait for the update thread to signal done
            _wait(self._stopped, -1)
            # cleanup events
            _close(self._running)
            _close(self._stopped)
            self._running = self._stopped = None

def getloadavg(self):
        return self._load

Name	Type	Size	Owner/Group	Permission	Last Modified	Actions
.	dir	-	root/linksafe	drwxr-xr-x	2025-06-29 12:11:52
..	dir	-	root/linksafe	drwxr-xr-x	2025-06-29 12:11:53
__pycache__	dir	-	root/linksafe	drwxr-xr-x	2025-06-29 12:11:52
__init__.py	text/x-script.python	119 B	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
cmdline.py	text/x-script.python	19.09 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
main.py	text/x-script.python	25.05 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
pgo.py	text/x-script.python	1.31 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
refleak.py	text/x-script.python	8 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
runtest.py	text/x-script.python	12.06 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
runtest_mp.py	text/x-script.python	17.13 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
save_env.py	text/x-script.python	11.54 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
setup.py	text/x-script.python	5.2 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
utils.py	text/x-script.python	2.05 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52
win_utils.py	text/x-script.python	4.45 KB	root/linksafe	-rw-r--r--	2025-06-03 06:47:52