win_utils.py

import_winapi
importmath
importmsvcrt
importos
importsubprocess
importuuid
importwinreg
fromtestimportsupport
fromtest.libregrtest.utilsimportprint_warning


# Max size of asynchronous reads
BUFSIZE=8192
# 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
# Windows registry subkey of HKEY_LOCAL_MACHINE where the counter names
# of typeperf are registered
COUNTER_REGISTRY_KEY= (r"SOFTWARE\Microsoft\Windows NT\CurrentVersion"
r"\Perflib\CurrentLanguage")


classWindowsLoadTracker():
"""
 This class asynchronously interacts with the `typeperf` command to read
 the system load on Windows. Multiprocessing and threads can't be used
 here because they interfere with the test suite's cases for those
 modules.
 """

def__init__(self):
self._values= []
self._load=None
self._buffer=''
self._popen=None
self.start()

defstart(self):
# Create a named pipe which allows for asynchronous IO in Windows
pipe_name=r'\\.\pipe\typeperf_output_'+str(uuid.uuid4())

open_mode=_winapi.PIPE_ACCESS_INBOUND
open_mode|=_winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
open_mode|=_winapi.FILE_FLAG_OVERLAPPED

# This is the read end of the pipe, where we will be grabbing output
self.pipe=_winapi.CreateNamedPipe(
pipe_name, open_mode, _winapi.PIPE_WAIT,
1, BUFSIZE, BUFSIZE, _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL
 )
# The write end of the pipe which is passed to the created process
pipe_write_end=_winapi.CreateFile(
pipe_name, _winapi.GENERIC_WRITE, 0, _winapi.NULL,
_winapi.OPEN_EXISTING, 0, _winapi.NULL
 )
# Open up the handle as a python file object so we can pass it to
# subprocess
command_stdout=msvcrt.open_osfhandle(pipe_write_end, 0)

# Connect to the read end of the pipe in overlap/async mode
overlap=_winapi.ConnectNamedPipe(self.pipe, overlapped=True)
overlap.GetOverlappedResult(True)

# Spawn off the load monitor
counter_name=self._get_counter_name()
command= ['typeperf', counter_name, '-si', str(SAMPLING_INTERVAL)]
self._popen=subprocess.Popen(' '.join(command), stdout=command_stdout, cwd=support.SAVEDCWD)

# Close our copy of the write end of the pipe
os.close(command_stdout)

def_get_counter_name(self):
# accessing the registry to get the counter localization name
withwinreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, COUNTER_REGISTRY_KEY) asperfkey:
counters=winreg.QueryValueEx(perfkey, 'Counter')[0]

# Convert [key1, value1, key2, value2, ...] list
# to {key1: value1, key2: value2, ...} dict
counters=iter(counters)
counters_dict=dict(zip(counters, counters))

# System counter has key '2' and Processor Queue Length has key '44'
system=counters_dict['2']
process_queue_length=counters_dict['44']
returnf'"\\{system}\\{process_queue_length}"'

defclose(self, kill=True):
ifself._popenisNone:
return

self._load=None

ifkill:
self._popen.kill()
self._popen.wait()
self._popen=None

def__del__(self):
self.close()

def_parse_line(self, line):
# typeperf outputs in a CSV format like this:
# "07/19/2018 01:32:26.605","3.000000"
# (date, process queue length)
tokens=line.split(',')
iflen(tokens) !=2:
raiseValueError

value=tokens[1]
ifnotvalue.startswith('"') ornotvalue.endswith('"'):
raiseValueError
value=value[1:-1]
returnfloat(value)

def_read_lines(self):
overlapped, _=_winapi.ReadFile(self.pipe, BUFSIZE, True)
bytes_read, res=overlapped.GetOverlappedResult(False)
ifres!=0:
return ()

output=overlapped.getbuffer()
output=output.decode('oem', 'replace')
output=self._buffer+output
lines=output.splitlines(True)

# bpo-36670: typeperf only writes a newline *before* writing a value,
# not after. Sometimes, the written line in incomplete (ex: only
# timestamp, without the process queue length). Only pass the last line
# to the parser if it's a valid value, otherwise store it in
# self._buffer.
try:
self._parse_line(lines[-1])
exceptValueError:
self._buffer=lines.pop(-1)
else:
self._buffer=''

returnlines

defgetloadavg(self):
ifself._popenisNone:
returnNone

returncode=self._popen.poll()
ifreturncodeisnotNone:
self.close(kill=False)
returnNone

try:
lines=self._read_lines()
exceptBrokenPipeError:
self.close()
returnNone

forlineinlines:
line=line.rstrip()

# Ignore the initial header:
# "(PDH-CSV 4.0)","\\\\WIN\\System\\Processor Queue Length"
if'PDH-CSV'inline:
continue

# Ignore blank lines
ifnotline:
continue

try:
processor_queue_length=self._parse_line(line)
exceptValueError:
print_warning("Failed to parse typeperf output: %a"%line)
continue

# 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
ifself._loadisnotNone:
self._load= (self._load*LOAD_FACTOR_1
+processor_queue_length* (1.0-LOAD_FACTOR_1))
eliflen(self._values) <NVALUE:
self._values.append(processor_queue_length)
else:
self._load=sum(self._values) /len(self._values)

returnself._load