Eryk Sun <eryk...@gmail.com> 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 <rep...@bugs.python.org> <https://bugs.python.org/issue46788> _______________________________________ _______________________________________________ Python-bugs-list mailing list Unsubscribe: https://mail.python.org/mailman/options/python-bugs-list/archive%40mail-archive.com