Eryk Sun <[email protected]> added the comment:
I implemented a ctypes prototype that replaces the registry-based
implementation with the API calls PdhOpenQueryW(), PdhAddEnglishCounterW(),
PdhCollectQueryData(), PdhGetRawCounterValue(), and PdhCloseQuery(). I'm
attaching the script, but here's the class itself without the ctypes
definitions:
class WindowsLoadTracker():
"""
This class asynchronously reads the system "Processor Queue Length"
counter to calculate the system load on Windows. A raw thread is
used to avoid interfering with tests of the threading module.
"""
def __init__(self):
self._values = []
self._load = None
self._hrunning = kernel32.CreateEventW(None, True, False, None)
self._hstopped = kernel32.CreateEventW(None, True, False, None)
self._hquery = wintypes.HANDLE()
self._hcounter = wintypes.HANDLE()
pdh.PdhOpenQueryW(None, None, ctypes.byref(self._hquery))
pdh.PdhAddEnglishCounterW(self._hquery,
r"\System\Processor Queue Length",
None,
ctypes.byref(self._hcounter))
pdh.PdhCollectQueryData(self._hquery)
_thread.start_new_thread(self._update_load, (), {})
def _update_load(self,
# Localize module access to prevent shutdown errors.
WaitForSingleObject=_winapi.WaitForSingleObject,
SetEvent=kernel32.SetEvent):
# run until signaled to stop
while WaitForSingleObject(self._hrunning, 1000):
self._calculate_load()
# notify stopped
SetEvent(self._hstopped)
def _calculate_load(self,
# Lcalize module access to prevent shutdown errors.
PdhCollectQueryData=pdh.PdhCollectQueryData,
PdhGetRawCounterValue=pdh.PdhGetRawCounterValue):
counter_type = wintypes.DWORD()
raw = PDH_RAW_COUNTER()
PdhCollectQueryData(self._hquery)
PdhGetRawCounterValue(self._hcounter,
ctypes.byref(counter_type),
ctypes.byref(raw))
if raw.CStatus < 0:
return
processor_queue_length = raw.FirstValue
# 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 getloadavg(self):
return self._load
def close(self,
# Localize module access to prevent shutdown errors.
WaitForSingleObject=_winapi.WaitForSingleObject,
INFINITE=_winapi.INFINITE,
SetEvent=kernel32.SetEvent,
CloseHandle=_winapi.CloseHandle,
PdhCloseQuery=pdh.PdhCloseQuery):
if self._hrunning is None:
return
# Tell the update thread to quit.
SetEvent(self._hrunning)
# Wait for the update thread to stop before cleanup.
WaitForSingleObject(self._hstopped, INFINITE)
CloseHandle(self._hrunning)
CloseHandle(self._hstopped)
PdhCloseQuery(self._hquery)
self._hrunning = self._hstopped = None
def __del__(self):
self.close()
return
----------
Added file: https://bugs.python.org/file50695/loadtracker.py
_______________________________________
Python tracker <[email protected]>
<https://bugs.python.org/issue46788>
_______________________________________
_______________________________________________
Python-bugs-list mailing list
Unsubscribe:
https://mail.python.org/mailman/options/python-bugs-list/archive%40mail-archive.com
