Coverage for pyTooling/Process/__init__.py: 91%

1# ==================================================================================================================== # 

2# _____ _ _ ____ # 

3# _ __ _ |_ _|__ ___ | (_)_ __ __ _ | _ \ _ __ ___ ___ ___ ___ ___ # 

4# | '_ \| | | || |/ _ \ / _ \| | | '_ \ / _` | | |_) | '__/ _ \ / __/ _ \/ __/ __| # 

5# | |_) | |_| || | (_) | (_) | | | | | | (_| |_| __/| | | (_) | (_| __/\__ \__ \ # 

6# | .__/ \__, ||_|\___/ \___/|_|_|_| |_|\__, (_)_| |_| \___/ \___\___||___/___/ # 

7# |_| |___/ |___/ # 

8# ==================================================================================================================== # 

9# Authors: # 

10# Patrick Lehmann # 

11# # 

12# License: # 

13# ==================================================================================================================== # 

14# Copyright 2026-2026 Patrick Lehmann - Bötzingen, Germany # 

15# # 

16# Licensed under the Apache License, Version 2.0 (the "License"); # 

17# you may not use this file except in compliance with the License. # 

18# You may obtain a copy of the License at # 

19# # 

20# http://www.apache.org/licenses/LICENSE-2.0 # 

21# # 

22# Unless required by applicable law or agreed to in writing, software # 

23# distributed under the License is distributed on an "AS IS" BASIS, # 

24# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # 

25# See the License for the specific language governing permissions and # 

26# limitations under the License. # 

27# # 

28# SPDX-License-Identifier: Apache-2.0 # 

29# ==================================================================================================================== # 

30# 

31 

32from ctypes import Structure, c_void_p, c_size_t, c_int, c_int32, c_uint64 

33from os import getpid, strerror 

34from pathlib import Path 

35from typing import ClassVar, Any 

36 

37from pyTooling.Decorators import export, readonly 

38from pyTooling.MetaClasses import ExtendedType 

39from pyTooling.Platform import PlatformException, CurrentPlatform 

40 

41if CurrentPlatform.IsNativeWindows or CurrentPlatform.IsMSYS2Environment: 

42 from ctypes import WinDLL 

43 from ctypes.wintypes import HANDLE, BOOL, DWORD 

44 

45 

46@export 

47class MemoryInfo(metaclass=ExtendedType, slots=True): 

48 _ResidentMemory: int #: Resident Set Size (VmRSS) – physical pages currently mapped. Memory usage in bytes. 

49 _VirtualMemory: int #: Virtual Memory Size (VmS) – total virtual address space used. Memory usage in bytes. 

50 

51 def __init__(self, residentMemory: int, virtualMemory: int) -> None: 

52 """ 

53 Initializes the memory info object with **Resident Set Size** and **Virtual Memory Size**. 

54 

55 :param residentMemory: Resident Memory Size (VmRSS) in bytes. 

56 :param virtualMemory: Virtual Memory Size (VmS) in bytes. 

57 """ 

58 self._ResidentMemory = residentMemory 

59 self._VirtualMemory = virtualMemory 

60 

61 @readonly 

62 def ResidentMemory(self) -> int: 

63 """ 

64 Read-only property to access the **Resident Set Size** (used physical memory). 

65 

66 :returns: Resident Set Size (VmRSS) in bytes. 

67 """ 

68 return self._ResidentMemory 

69 

70 @readonly 

71 def VirtualMemory(self) -> int: 

72 """ 

73 Read-only property to access the **Virtual Memory Size** (used virtual memory). 

74 

75 :returns: Virtual Memory Size (VmS) in bytes. 

76 """ 

77 return self._VirtualMemory 

78 

79 def __str__(self) -> str: 

80 return f"Physical Memory (VmRSS): {self.ResidentMemory / 2**20:.3f} MiB / Virtual Memory (VmS): {self.VirtualMemory / 2**20:.3f} MiB" 

81 

82 

83@export 

84class ProcessInformation(metaclass=ExtendedType, slots=True): 

85 if CurrentPlatform.IsNativeWindows or CurrentPlatform.IsMSYS2Environment: 

86 _psapi: WinDLL 

87 _kernel32: WinDLL 

88 _processHandle: Any 

89 elif CurrentPlatform.IsNativeLinux: 

90 _processStatusFile: ClassVar[Path] = Path(f"/proc/self/statm") 

91 

92 if CurrentPlatform.IsNativeWindows or CurrentPlatform.IsMSYS2Environment: 

93 def __init__(self) -> None: 

94 self._psapi = WinDLL("psapi", use_last_error=True) 

95 self._kernel32 = WinDLL("kernel32", use_last_error=True) 

96 

97 self._kernel32.GetCurrentProcess.restype = HANDLE 

98 self._kernel32.GetCurrentProcess.argtypes = [] 

99 

100 self._processHandle = self._kernel32.GetCurrentProcess() 

101 else: 

102 def __init__(self) -> None: 

103 pass 

104 

105 if CurrentPlatform.IsNativeLinux: 

106 from os import sysconf 

107 _pageSize: ClassVar[int] = sysconf("SC_PAGESIZE") 

108 

109 def GetMemoryUsage(self) -> MemoryInfo: 

110 """ 

111 Get the memory usage of this Python process on a Linux system. 

112 

113 Read the `/proc/self/statm` memory statistic file (space separated) for the current process: 

114 

115 [0] size 

116 VmSize (total virtual address space) 

117 [1] resident 

118 VmRSS - Virtual memory Resident Set Size (pages currently resident in RAM) = used physical memory 

119 [2] shared 

120 shared pages (mapped from files) 

121 [3] text 

122 code segment pages 

123 [4] lib 

124 unused (always 0 since Linux 2.6) 

125 [5] data 

126 data + stack pages 

127 [6] dt 

128 dirty pages (always 0 since Linux 2.6) 

129 

130 ``SC_PAGESIZE`` is typically 4096 bytes, but can be 16kiB (ARM64) or 64kiB (PowerPC/RHEL9+). :func:`os.sysconf` 

131 reads it from the aux vector — no syscall overhead. 

132 

133 :returns: Physical memory usage (VmRSS) in bytes. 

134 """ 

135 

136 try: 

137 with self._processStatusFile.open("rb") as f: 

138 fields = f.read().split() 

139 except FileNotFoundError as ex: 

140 raise PlatformException(f"Can't open '{self._processStatusFile}' to extract the process' physical memory usage.") from ex 

141 

142 vms = int(fields[0]) * self._pageSize #: VmSize 

143 rss = int(fields[1]) * self._pageSize #: VmRSS 

144 

145 return MemoryInfo(rss, vms) 

146 

147 elif CurrentPlatform.IsNativeMacOS: 

148 class _ProcTaskInfo(Structure): 

149 """ 

150 ``struct proc_taskinfo`` from ``<sys/proc_info.h>`` 

151 """ 

152 _fields_ = [ 

153 ("pti_virtual_size", c_uint64), 

154 ("pti_resident_size", c_uint64), 

155 ("pti_total_user", c_uint64), 

156 ("pti_total_system", c_uint64), 

157 ("pti_threads_user", c_uint64), 

158 ("pti_threads_system", c_uint64), 

159 ("pti_policy", c_int32), 

160 ("pti_faults", c_int32), 

161 ("pti_pageins", c_int32), 

162 ("pti_cow_faults", c_int32), 

163 ("pti_messages_sent", c_int32), 

164 ("pti_messages_received", c_int32), 

165 ("pti_syscalls_mach", c_int32), 

166 ("pti_syscalls_unix", c_int32), 

167 ("pti_csw", c_int32), 

168 ("pti_threadnum", c_int32), 

169 ("pti_numrunning", c_int32), 

170 ("pti_priority", c_int32), 

171 ] 

172 

173 def GetMemoryUsage(self) -> MemoryInfo: 

174 """ 

175 Call libproc.proc_pidinfo(PROC_PIDTASKINFO) – the same route psutil takes. 

176 

177 struct proc_taskinfo (<sys/proc_info.h>): 

178 pti_virtual_size uint64 – virtual address space in bytes 

179 pti_resident_size uint64 – resident (physical) memory in bytes 

180 … 16 further fields (timing, policy, fault/syscall counters) 

181 

182 proc_pidinfo() returns the number of bytes written; ≤ 0 means error 

183 (errno is set). PROC_PIDTASKINFO = 4. 

184 """ 

185 from ctypes import CDLL, byref, sizeof, get_errno 

186 from ctypes.util import find_library 

187 

188 PROC_PIDTASKINFO = 4 

189 

190 _libproc_path = find_library("proc") # or "/usr/lib/libproc.dylib" 

191 _libproc = CDLL(_libproc_path, use_errno=True) 

192 _libproc.proc_pidinfo.restype = c_int 

193 _libproc.proc_pidinfo.argtypes = [ 

194 c_int, # pid 

195 c_int, # flavor 

196 c_uint64, # arg (unused for PROC_PIDTASKINFO) 

197 c_void_p, # buffer 

198 c_int, # buffersize 

199 ] 

200 

201 taskInfo = self._ProcTaskInfo() 

202 ret = _libproc.proc_pidinfo(getpid(), PROC_PIDTASKINFO, 0, byref(taskInfo), sizeof(taskInfo)) 

203 if ret <= 0: 203 ↛ 204line 203 didn't jump to line 204 because the condition on line 203 was never true

204 err = get_errno() 

205 raise PlatformException(f"Failed to get current process' information.") from OSError(err, strerror(err), "proc_pidinfo") 

206 

207 return MemoryInfo(taskInfo.pti_resident_size, taskInfo.pti_virtual_size) 

208 

209 elif CurrentPlatform.IsNativeWindows or CurrentPlatform.IsMSYS2Environment: 209 ↛ 255line 209 didn't jump to line 255 because the condition on line 209 was always true

210 class _ProcessMemoryCounters(Structure): 

211 from ctypes.wintypes import DWORD 

212 

213 _fields_ = [ 

214 ("cb", DWORD), 

215 ("PageFaultCount", DWORD), 

216 ("PeakWorkingSetSize", c_size_t), 

217 ("WorkingSetSize", c_size_t), 

218 ("QuotaPeakPagedPoolUsage", c_size_t), 

219 ("QuotaPagedPoolUsage", c_size_t), 

220 ("QuotaPeakNonPagedPoolUsage", c_size_t), 

221 ("QuotaNonPagedPoolUsage", c_size_t), 

222 ("PagefileUsage", c_size_t), 

223 ("PeakPagefileUsage", c_size_t), 

224 ] 

225 

226 del DWORD 

227 

228 def GetMemoryUsage(self) -> MemoryInfo: 

229 """ 

230 Call psapi.GetProcessMemoryInfo() with a PROCESS_MEMORY_COUNTERS struct. 

231 

232 WorkingSetSize – physical pages currently mapped → RSS 

233 PagefileUsage – private committed bytes → VMS (= "Private Bytes" 

234 in Task Manager; mirrors psutil's vms on Windows) 

235 

236 GetCurrentProcess() returns a pseudo-handle (-1) requiring no CloseHandle. 

237 use_last_error=True routes SetLastError / GetLastError through ctypes so 

238 WinError() picks up the correct code without a race. 

239 """ 

240 

241 from ctypes import WinDLL, WinError, POINTER, sizeof, byref, get_last_error 

242 

243 self._psapi.GetProcessMemoryInfo.restype = BOOL 

244 self._psapi.GetProcessMemoryInfo.argtypes = [HANDLE, POINTER(self._ProcessMemoryCounters), DWORD] 

245 

246 processMemoryCounters = self._ProcessMemoryCounters() 

247 processMemoryCounters.cb = sizeof(processMemoryCounters) 

248 

249 if not self._psapi.GetProcessMemoryInfo(self._processHandle, byref(processMemoryCounters), processMemoryCounters.cb): 249 ↛ 250line 249 didn't jump to line 250 because the condition on line 249 was never true

250 raise WinError(get_last_error()) 

251 

252 return MemoryInfo(processMemoryCounters.WorkingSetSize, processMemoryCounters.PagefileUsage) 

253 

254 else: 

255 raise PlatformException(f"Unsupported platform: '{CurrentPlatform}'.")