Performance additions surrounding log parsing

autoprimenet.py

@@ -1,4 +1,4 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
 
 # /// script
@@ -90,10 +90,10 @@
 import time
 import timeit
 import uuid
 import xml.etree.ElementTree as ET
 import zipfile
 from array import array
-from collections import deque, namedtuple
+from collections import namedtuple
 from ctypes.util import find_library
 from datetime import datetime
 from decimal import Decimal
@@ -106,9 +106,9 @@
 from hashlib import md5
 from itertools import chain, count, starmap
 
 try:
 	# Python 2
 	from future_builtins import map, zip  # ascii, filter, hex, oct
 except ImportError:
 	pass
 
@@ -183,7 +183,7 @@
 		if n < 0:
 			msg = "isqrt() argument must be nonnegative"
 			raise ValueError(msg)
 		if n == 0:
 			return 0
 
 		c = (n.bit_length() - 1) // 2
@@ -205,7 +205,7 @@
 
 	def log2(x):
 		"""Calculate the base-2 logarithm of a given number."""
 		return math.log(x, 2)
 
 
 try:
@@ -238,15 +238,15 @@
 		# Python 3+
 		import winreg
 	except ImportError:
 		import _winreg as winreg
 
 	kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
 	advapi32 = ctypes.WinDLL("advapi32", use_last_error=True)
 
 	class GROUP_AFFINITY(ctypes.Structure):
 		_fields_ = (("Mask", wintypes.WPARAM), ("Group", wintypes.WORD), ("Reserved", wintypes.WORD * 3))
 
 	class PROCESSOR_RELATIONSHIP(ctypes.Structure):
 		_fields_ = (
 			("Flags", wintypes.BYTE),
 			("EfficiencyClass", wintypes.BYTE),
@@ -255,10 +255,10 @@
 			("GroupMask", GROUP_AFFINITY * 1),
 		)
 
 	class union(ctypes.Union):
 		_fields_ = (("GroupMask", GROUP_AFFINITY), ("GroupMasks", GROUP_AFFINITY * 1))
 
 	class NUMA_NODE_RELATIONSHIP(ctypes.Structure):
 		_anonymous_ = ("union",)
 		_fields_ = (
 			("NodeNumber", wintypes.DWORD),
@@ -267,7 +267,7 @@
 			("union", union),
 		)
 
 	class CACHE_RELATIONSHIP(ctypes.Structure):
 		_anonymous_ = ("union",)
 		_fields_ = (
 			("Level", wintypes.BYTE),
@@ -280,7 +280,7 @@
 			("union", union),
 		)
 
 	class PROCESSOR_GROUP_INFO(ctypes.Structure):
 		_fields_ = (
 			("MaximumProcessorCount", wintypes.BYTE),
 			("ActiveProcessorCount", wintypes.BYTE),
@@ -288,7 +288,7 @@
 			("ActiveProcessorMask", wintypes.WPARAM),
 		)
 
 	class GROUP_RELATIONSHIP(ctypes.Structure):
 		_fields_ = (
 			("MaximumGroupCount", wintypes.WORD),
 			("ActiveGroupCount", wintypes.WORD),
@@ -296,7 +296,7 @@
 			("GroupInfo", PROCESSOR_GROUP_INFO * 1),
 		)
 
 	class union(ctypes.Union):
 		_fields_ = (
 			("Processor", PROCESSOR_RELATIONSHIP),
 			("NumaNode", NUMA_NODE_RELATIONSHIP),
@@ -304,7 +304,7 @@
 			("Group", GROUP_RELATIONSHIP),
 		)
 
 	class SYSTEM_LOGICAL_PROCESSOR_INFORMATION_EX(ctypes.Structure):
 		_anonymous_ = ("union",)
 		_fields_ = (("Relationship", wintypes.DWORD), ("Size", wintypes.DWORD), ("union", union))
 
@@ -323,7 +323,7 @@
 			("Type", wintypes.DWORD),
 		)
 
 	class union(ctypes.Union):
 		_fields_ = (
 			("ProcessorCore", ProcessorCore),
 			("NumaNode", NumaNode),
@@ -393,7 +393,7 @@
 					if regtype == winreg.REG_SZ:
 						output = product_id
 						break
 			except WindowsError:
 				pass
 
 		return output
@@ -1040,7 +1040,7 @@
 
 	def watch_files(adir, cpu_num):
 		"""Monitors a directory for file changes and processes specific file actions."""
 		fun = kernel32.CreateFileW if sys.version_info >= (3,) or isinstance(adir, unicode) else kernel32.CreateFileA
 		handle = fun(
 			adir,
 			1,  # FILE_LIST_DIRECTORY
@@ -1227,7 +1227,7 @@
 		"""Adds a watch on the specified path for the given file flags."""
 		fd = os.open(path, 0x8000)  # os.O_EVTONLY
 
 		event = select.kevent(fd, select.KQ_FILTER_VNODE, select.KQ_EV_ADD | select.KQ_EV_CLEAR, fflags)
 		return fd, event
 
 	def watch_files(adir, cpu_num):
@@ -1242,10 +1242,10 @@
 		proof_fd = None
 		changelist = []
 
 		kq = select.kqueue()
 
 		try:
 			dir_fd, event = add_watch(adir, select.KQ_NOTE_DELETE | select.KQ_NOTE_WRITE)
 			fds[dir_fd] = adir, event
 			changelist.append(event)
 			logging.info("Watching the directory: %r.", os.path.abspath(adir))
@@ -1639,7 +1639,7 @@
 		def utcoffset(self, _dt):
 			return self.ZERO
 
 		def tzname(self, _dt):
 			return "UTC"
 
 		def dst(self, _dt):
@@ -1947,15 +1947,15 @@
 
 # Python 2
 if hasattr(__builtins__, "unicode"):
 	str = unicode
 
 # Python 2
 if hasattr(__builtins__, "raw_input"):
 	input = raw_input
 
 # Python 2
 if hasattr(__builtins__, "xrange"):
 	range = xrange
 
 
 def exponent_to_str(assignment):
@@ -2998,6 +2998,51 @@
 			logging.exception("Failed to read the %r file: %s: %s", filename, type(e).__name__, e, exc_info=args.debug)
 
 
+def iter_lines_reversed(filename, encoding="utf-8", errors="strict", chunk_size=256 * 1024):
+	"""Backward chunked read; raises OSError if getsize/seek/read fails; nothing if size is 0."""
+	buffer = bytearray(chunk_size)
+	view = memoryview(buffer)
+	incomplete = None
+	try:
+		with open(filename, "rb", buffering=0) as f:
+			f.seek(0, 2)  # os.SEEK_END
+			pos = size = f.tell()
+			while pos > 0:
+				step = min(chunk_size, pos)
+				f.seek(pos - step)
+				n = f.readinto(view[:step])
+				if not incomplete:
+					buf = buffer[:n]
+				else:
+					buf = b"".join((view[:n], incomplete))
+				lines = buf.splitlines()
+				if pos != size and buf.endswith((b"\r", b"\n")):
+					lines.append(b"")
+				incomplete = lines.pop(0)
+				for line in reversed(lines):
+					yield line.decode(encoding, errors)
+				pos -= n
+			if incomplete:
+				yield incomplete.decode(encoding, errors)
+	# Python 3.3+: FileNotFoundError
+	except (IOError, OSError) as e:
+		if e.errno != errno.ENOENT:
+			logging.exception("Failed to read the %r file: %s: %s", filename, type(e).__name__, e, exc_info=args.debug)
+
+
+def read_last_n_lines(filename, n, encoding="utf-8", errors="strict"):
+	"""Return the last n lines (oldest first)."""
+	if n <= 0:
+		return []
+	out = []
+	for i, line in enumerate(iter_lines_reversed(filename, encoding=encoding, errors=errors), 1):
+		out.append(line)
+		if i >= n:
+			break
+	out.reverse()
+	return out
+
+
 # region Config
 attr_to_copy = {
 	SEC.PrimeNet: {
@@ -4018,10 +4063,10 @@
 	"""Returns the last 'lines' lines from a file, or an appropriate message if the file is not found or empty."""
 	if not os.path.isfile(filename):
 		return "> (File not found)"
-	w = deque(readonly_list_file(filename, errors="replace"), lines)
-	if not w:
+	last = read_last_n_lines(filename, lines, errors="replace")
+	if not last:
 		return "> (File is empty)"
-	return "\n".join("> " + line for line in w)
+	return "\n".join("> " + line for line in last)
 
 
 def send(subject, message, attachments=None, to=None, cc=None, bcc=None, priority=None):
@@ -4308,7 +4353,7 @@
 		with open("/proc/cpuinfo") as f:
 			for line in f:
 				if line.startswith("model name"):
-					output = re.sub(r"^.*: *", "", line.rstrip(), count=1)
+					output = line.split(":", 1)[1].strip()
 					break
 	return output
 
@@ -4378,7 +4423,7 @@
 			frequency = output // 1000 // 1000
 	elif sys.platform.startswith("linux"):
 		with open("/proc/cpuinfo") as f:
-			freqs = [float(re.sub(r"^.*: *", "", line.rstrip(), count=1)) for line in f if line.startswith("cpu MHz")]
+			freqs = [float(line.split(":", 1)[1]) for line in f if line.startswith("cpu MHz")]
 		if freqs:
 			freq = set(freqs)
 			if len(freq) == 1:
@@ -5712,6 +5757,12 @@
 
 MLUCAS_RE = re.compile(r"^p([0-9]+)(?:\.s([12]))?$")
 
+MLUCAS_STAT_ITER_RE = re.compile(
+	r"(Iter#|S1|S2)(?: bit| at q)? = ([0-9]+) \[ ?([0-9]+\.[0-9]+)% complete\] .*\[ *([0-9]+\.[0-9]+) (m?sec)/iter\]"
+)
+MLUCAS_STAT_FFT_RE = re.compile(r"FFT length [0-9]{3,}K = ([0-9]{6,})")
+MLUCAS_STAT_S2Q0_RE = re.compile(r"Stage 2 q0 = ([0-9]+)")
+
 
 def parse_stat_file(adapter, adir, p):
 	"""Parse the Mlucas stat file for the progress of the assignment."""
@@ -5738,20 +5789,15 @@
 		adapter.debug("stat file %r does not exist", statfile)
 		return iteration, iterations, None, None, stage, fftlen
 
-	w = readonly_list_file(statfile)  # appended line by line, no lock needed
+	# appended line by line, no lock needed
 	found = 0
-	regex = re.compile(
-		r"(Iter#|S1|S2)(?: bit| at q)? = ([0-9]+) \[ ?([0-9]+\.[0-9]+)% complete\] .*\[ *([0-9]+\.[0-9]+) (m?sec)/iter\]"
-	)
-	fft_regex = re.compile(r"FFT length [0-9]{3,}K = ([0-9]{6,})")
-	s2_regex = re.compile(r"Stage 2 q0 = ([0-9]+)")
 	list_msec_per_iter = []
 	s2 = bits = 0
 	# get the 5 most recent Iter line
-	for line in reversed(list(w)):
-		res = regex.search(line)
-		fft_res = fft_regex.search(line)
-		s2_res = s2_regex.search(line)
+	for line in iter_lines_reversed(statfile):
+		res = MLUCAS_STAT_ITER_RE.search(line)
+		fft_res = MLUCAS_STAT_FFT_RE.search(line)
+		s2_res = MLUCAS_STAT_S2Q0_RE.search(line)
 		if res and found < 5:
 			astage = res.group(1)
 			# keep the last iteration to compute the percent of progress
@@ -5805,6 +5851,17 @@
 
 
 GPUOWL_RE = re.compile(r"^(?:(?:[0-9]+)(?:-([0-9]+)\.(ll|prp|p1final|p2)|(?:-[0-9]+-([0-9]+))?\.p1|\.(?:(ll|p[12])\.)?owl))$")
+GPUOWL_LOG_MAIN_RE = re.compile(r"([0-9]{6,}) (LL|P1|OK|EE)? +([0-9]{4,})")
+GPUOWL_LOG_AP1_RE = re.compile(r"([0-9]{6,})P1 +([0-9]+\.[0-9]+)% ([KE])? +[0-9a-f]{16} +([0-9]+)")
+GPUOWL_LOG_US_PER_RE = re.compile(r"\b([0-9]+) us/it;?")
+GPUOWL_LOG_FFT_RE = re.compile(r"\b([0-9]{6,}) FFT: ([0-9]+(?:\.[0-9]+)?[KM])\b")
+GPUOWL_LOG_P1_BITS_RE = re.compile(r"\b[0-9]{6,} P1(?: B1=[0-9]+, B2=[0-9]+;|\([0-9]+(?:\.[0-9])?M?\)) ([0-9]+) bits;?\b")
+GPUOWL_LOG_AP1_BITS_RE = re.compile(r"\b[0-9]{6,}P1 +[0-9]+\.[0-9]+% @([0-9]+)/([0-9]+) B1\([0-9]+\)")
+GPUOWL_LOG_P2_BLOCKS_RE = re.compile(
+	r"[0-9]{6,} P2\([0-9]+(?:\.[0-9])?M?,[0-9]+(?:\.[0-9])?M?\) ([0-9]+) blocks: ([0-9]+) - ([0-9]+);"
+)
+GPUOWL_LOG_P1_PIPE_RE = re.compile(r"\| P1\([0-9]+(?:\.[0-9])?M?\)")
+GPUOWL_LOG_P2_OK_RE = re.compile(r"[0-9]{6,} P2(?: ([0-9]+)/([0-9]+)|\([0-9]+(?:\.[0-9])?M?,[0-9]+(?:\.[0-9])?M?\) OK @([0-9]+)):")
 
 
 def parse_gpuowl_log_file(adapter, adir, p):
@@ -5840,30 +5897,20 @@
 		adapter.debug("Log file %r does not exist", logfile)
 		return iteration, iterations, None, None, stage, fftlen
 
-	w = readonly_list_file(logfile, errors="replace")
 	found = 0
-	regex = re.compile(r"([0-9]{6,}) (LL|P1|OK|EE)? +([0-9]{4,})")
-	aregex = re.compile(r"([0-9]{6,})P1 +([0-9]+\.[0-9]+)% ([KE])? +[0-9a-f]{16} +([0-9]+)")
-	us_per_regex = re.compile(r"\b([0-9]+) us/it;?")
-	fft_regex = re.compile(r"\b([0-9]{6,}) FFT: ([0-9]+(?:\.[0-9]+)?[KM])\b")
-	p1_bits_regex = re.compile(r"\b[0-9]{6,} P1(?: B1=[0-9]+, B2=[0-9]+;|\([0-9]+(?:\.[0-9])?M?\)) ([0-9]+) bits;?\b")
-	ap1_bits_regex = re.compile(r"\b[0-9]{6,}P1 +[0-9]+\.[0-9]+% @([0-9]+)/([0-9]+) B1\([0-9]+\)")
-	p2_blocks_regex = re.compile(r"[0-9]{6,} P2\([0-9]+(?:\.[0-9])?M?,[0-9]+(?:\.[0-9])?M?\) ([0-9]+) blocks: ([0-9]+) - ([0-9]+);")
-	p1_regex = re.compile(r"\| P1\([0-9]+(?:\.[0-9])?M?\)")
-	p2_regex = re.compile(r"[0-9]{6,} P2(?: ([0-9]+)/([0-9]+)|\([0-9]+(?:\.[0-9])?M?,[0-9]+(?:\.[0-9])?M?\) OK @([0-9]+)):")
 	list_usec_per_iter = []
 	p1 = p2 = False
 	buffs = bits = 0
 	# get the 5 most recent Iter line
-	for line in reversed(list(w)):
-		res = regex.search(line)
-		ares = aregex.search(line)
-		us_res = us_per_regex.search(line)
-		fft_res = fft_regex.search(line)
-		p1_bits_res = p1_bits_regex.search(line)
-		ap1_bits_res = ap1_bits_regex.search(line)
-		blocks_res = p2_blocks_regex.search(line)
-		p2_res = p2_regex.search(line)
+	for line in iter_lines_reversed(logfile, errors="replace"):
+		res = GPUOWL_LOG_MAIN_RE.search(line)
+		ares = GPUOWL_LOG_AP1_RE.search(line) if "P1" in line else None
+		us_res = GPUOWL_LOG_US_PER_RE.search(line) if "us/it" in line else None
+		fft_res = GPUOWL_LOG_FFT_RE.search(line) if "FFT:" in line else None
+		p1_bits_res = GPUOWL_LOG_P1_BITS_RE.search(line) if "P1" in line and "bits" in line else None
+		ap1_bits_res = GPUOWL_LOG_AP1_BITS_RE.search(line) if "P1" in line and "B1" in line else None
+		blocks_res = GPUOWL_LOG_P2_BLOCKS_RE.search(line) if "P2" in line and "blocks:" in line else None
+		p2_res = GPUOWL_LOG_P2_OK_RE.search(line) if "P2" in line else None
 		if res or ares:
 			num = int(res.group(1) if res else ares.group(1))
 			if num != p:
@@ -5892,7 +5939,7 @@
 			elif aiteration > iteration:
 				break
 			if not p1 and not (p2 or buffs):
-				p1_res = p1_regex.search(line)
+				p1_res = GPUOWL_LOG_P1_PIPE_RE.search(line) if "P1" in line else None
 				p1 = res.group(2) == "OK" and bool(p1_res)
 				if p1:
 					stage = 1
@@ -6700,7 +6747,7 @@
 
 
 # endregion
 # TODO -- have people set their own program options for commented out portions
 def program_options(send=False, start=-1, retry_count=0):
 	"""Sets the program options on the PrimeNet server."""
 	adapter = logging.LoggerAdapter(logger, None)
@@ -7166,7 +7213,7 @@
 		PRIMENET_WORK_TYPE.CERT,
 	}:
 		adapter.error("Returned assignment from server is not a supported worktype %s.", assignment.work_type)
 		# TODO: Unreserve assignment
 		# assignment_unreserve()
 		# return None
 	if assignment.work_type in {PRIMENET_WORK_TYPE.FIRST_LL, PRIMENET_WORK_TYPE.DBLCHK}:
@@ -7189,7 +7236,7 @@
 						adapter.error("PRP base (%s) is not 3", assignment.prp_base)
 					if assignment.prp_residue_type not in {1, 5}:
 						adapter.error("PRP residue type (%s) is not 1 or 5", assignment.prp_residue_type)
 					# TODO: Unreserve assignment
 					# assignment_unreserve()
 		if "kf" in r:
 			# Workaround Mlucas bug: https://github.com/primesearch/Mlucas/issues/30
@@ -7326,7 +7373,7 @@
 			adapter.debug("New assignment: %r", new_task)
 			file.write(new_task + "\n")  # changed = True
 
 			# TODO: Something better here
 			cpu_quota_used = test.cert_squarings / 110000
 			cpu_quota_used *= 2.1 ** log2(test.n / 97300000)
 			cpu_limit_remaining -= cpu_quota_used
@@ -7522,7 +7569,7 @@
 			register_instance(guid)
 		# In all other error case, the submission must not be retried
 		elif rc in {PRIMENET_ERROR.INVALID_ASSIGNMENT_KEY, PRIMENET_ERROR.WORK_NO_LONGER_NEEDED, PRIMENET_ERROR.NO_ASSIGNMENT}:
 			# TODO: Delete assignment from workfile if it is not done
 			return rc, ghd
 		elif rc in {PRIMENET_ERROR.INVALID_RESULT_TYPE, PRIMENET_ERROR.ILLEGAL_RESIDUE}:
 			return rc, ghd
@@ -8841,7 +8888,7 @@
 			# adapter.debug("Update correctly sent to server")
 			pass
 		elif rc in {PRIMENET_ERROR.INVALID_ASSIGNMENT_KEY, PRIMENET_ERROR.WORK_NO_LONGER_NEEDED}:
 			# TODO: Delete assignment from workfile
 			pass
 		elif rc == PRIMENET_ERROR.UNREGISTERED_CPU:
 			register_instance()
@@ -9993,7 +10040,7 @@
 group.add_argument("--proxy-username", help="Proxy server username")
 group.add_argument("--proxy-password", help="Proxy server password")
 
 # TODO: add detection for most parameter, including automatic change of the hardware
 memory = get_physical_memory() or 1024
 cores, threads = get_cpu_cores_threads()
 cache_sizes = get_cpu_cache_sizes()