"""Постоянный владелец DebugSession и локальный JSON-RPC для адаптеров.""" from __future__ import annotations from collections import deque import json from pathlib import Path import socket import socketserver import string import threading import time from .session import DebugSession, SessionError from .transport import BridgeError from .macros import validate_log_message class SessionController: def __init__(self, session: DebugSession): self.session = session self.lock = threading.RLock() self.changed = threading.Condition() self.events = deque(maxlen=1024) self.dropped_events = 0 self.sequence = 0 self.running = False self.source_step = None self.closed = False self.source_breakpoints: dict[str, list[int]] = {} self.function_breakpoints: list[int] = [] self.breakpoint_info: dict[int, dict] = {} self.attached = session.attach() self.mame_console = bool(self.attached.get('capabilities', {}).get('console_print')) self._emit('stopped', {'reason': 'entry', 'location': self.attached['location']}) self._install_macro_logs() self.poller = threading.Thread(target=self._poll, name='sdbg-poller', daemon=True) self.poller.start() def _install_macro_logs(self) -> None: """Авторские точки принадлежат пакету сборки, не DAP source-набору.""" created = [] try: for macro in getattr(self.session.model, 'logpoints', []): if not macro['verified'] or macro['source'] in self.session.model.stale_sources: self._emit('output', {'category': 'stderr', 'output': 'sdbg: SDBG_LOG ' + macro['tag'] + ' не активирован: ' + str(macro.get('reason') or 'устаревший исходник') + '\n'}) continue self._validate_log_message(macro['message']) item = self.session.break_anchor(macro, enabled=False) created.append(item['id']) self.breakpoint_info[item['id']] = { 'kind': 'log', 'message': macro['message'], 'condition': macro['condition'], 'module': macro['module'], 'tag': macro['tag'], 'locations': item['locations'], 'hits': 0, } if created: self.session.activate_breakpoints() except BaseException: for identifier in created: self.breakpoint_info.pop(identifier, None) try: self.session.clear_breakpoint(identifier) except BaseException: pass raise def _emit(self, name: str, body: dict) -> None: with self.changed: if len(self.events) == self.events.maxlen: self.dropped_events += 1 self.sequence += 1 self.events.append({'seq': self.sequence, 'event': name, 'body': body}) self.changed.notify_all() def _poll(self) -> None: while not self.closed: if not self.running: time.sleep(.02) continue try: with self.lock: snapshot = self.session.bridge.request('snapshot') if snapshot['state'] == 'stopped': location = self.session.where(snapshot) if self.source_step is not None: self._source_step_stopped(location) elif self._handle_logpoints(location): self.session.bridge.request('continue') continue else: self.running = False self._emit('stopped', {'reason': 'breakpoint', 'location': location}) elif snapshot['state'] == 'invalidated': self.running = False self.source_step = None self.closed = True self._emit('invalidated', {'reason': 'reset_or_load'}) except (BridgeError, SessionError, ValueError, OSError) as error: self.running = False self.source_step = None self.closed = True self._emit('invalidated', {'reason': str(error)}) time.sleep(.01) def close(self) -> None: self.closed = True self.poller.join(timeout=1) self.session.bridge.close() def call(self, method: str, arguments: dict) -> object: if method == 'events': after = int(arguments.get('after', 0)) timeout = min(max(float(arguments.get('timeout', 0)), 0), 30) deadline = time.monotonic() + timeout with self.changed: while self.sequence <= after and not self.closed and time.monotonic() < deadline: self.changed.wait(deadline - time.monotonic()) return {'events': [event for event in self.events if event['seq'] > after], 'first': self.events[0]['seq'] if self.events else self.sequence + 1, 'last': self.sequence, 'lost': self.dropped_events, 'closed': self.closed} with self.lock: if self.closed: raise SessionError('Сессия закрыта или инвалидирована') if method == 'status': return {**self.attached, 'running': self.running, 'event_sequence': self.sequence} if method == 'snapshot': return self.session.bridge.request('snapshot') if method == 'mame_console_tail': return self.session.bridge.request( 'console_tail', count=int(arguments.get('count', 40))) if method == 'input_key': if not self.running and arguments['down'] is not False: raise SessionError('Нажатие возможно только при running CPU') return self.session.bridge.request( 'key', tag=arguments['tag'], mask=int(arguments['mask']), down=arguments['down']) if method == 'where': return self.session.where() if method == 'registers': mapping = self.session.refresh() return {'generation': mapping.generation, 'registers': mapping.registers, 'bank_pages': mapping.bank_pages} if method == 'variables': return self.session.model.variables if method == 'read_variable': return self.session.read_variable(arguments['name'], arguments.get('module')) if method == 'break_line': return self.session.break_line(arguments['file'], int(arguments['line'])) if method == 'break_function': return self.session.break_function(arguments['name']) if method == 'clear_breakpoint': return self.session.clear_breakpoint(int(arguments['id'])) if method == 'set_source_breakpoints': specifications = arguments.get('breakpoints') if specifications is None: specifications = [{'line': line} for line in arguments.get('lines', [])] return self._set_source_breakpoints(arguments['file'], specifications) if method == 'set_function_breakpoints': return self._set_function_breakpoints(arguments.get('names', [])) if method == 'continue': if self.running: raise SessionError('CPU уже выполняется; сначала Pause') result = self.session.bridge.request('continue') self.running = True self._emit('continued', {}) return result if method == 'pause': self.source_step = None self.session.bridge.request('pause') location = self.session.where(self.session.bridge.wait_stopped()) self.running = False self._emit('stopped', {'reason': 'pause', 'location': location}) return location if method == 'step': if self.running: raise SessionError('CPU уже выполняется; сначала Pause') self.session.bridge.request('step') self.running = True self._emit('continued', {'reason': 'step'}) location = self.session.where(self.session.bridge.wait_stopped()) self.running = False self._emit('stopped', {'reason': 'step', 'location': location}) return location if method == 'source_step': return self._source_step(arguments.get('kind', 'into')) raise SessionError('Неизвестный RPC-метод: ' + method) @staticmethod def _source_identity(location: dict) -> frozenset[tuple[str, int]]: return frozenset((item['file'], int(item['line'])) for item in location.get('sources', [])) def _has_stop_breakpoint(self, location: dict) -> bool: address = location.get('link_address') return any(info['kind'] == 'stop' and any(item.get('link_address') == address for item in info['locations']) for info in self.breakpoint_info.values()) def _source_step(self, kind: str) -> dict: commands = {'into': 'step', 'over': 'step_over', 'out': 'step_out'} if kind not in commands: raise SessionError('Неизвестный вид source step') if self.running: raise SessionError('CPU уже выполняется; сначала Pause') start = self.session.where() if start.get('stale_source'): raise SessionError('Исходник изменён после сборки; пересоберите программу') initial = self._source_identity(start) if not initial: raise SessionError('Текущий PC не имеет проверенной C-позиции') self.source_step = { 'initial': initial, 'kind': kind, 'command': commands[kind], 'instructions': 1, 'location': start, } try: self.session.bridge.request(commands[kind]) except BaseException: self.source_step = None raise if kind == 'out': # Первый out выходит из машинного frame. У банкового вызова # дальше идём через over до первой позиции C вызывающей функции. self.source_step['command'] = 'step_over' self.running = True self._emit('continued', {'reason': 'step'}) return {'accepted': True} def _source_step_stopped(self, location: dict) -> None: step = self.source_step self._handle_logpoints(location) reason = None if self._has_stop_breakpoint(location): reason = 'breakpoint' elif self._source_identity(location) and self._source_identity(location) != step['initial']: reason = 'step' elif step['instructions'] >= 512: reason = 'step' self._emit('output', {'category': 'stderr', 'output': 'sdbg: source step достиг лимита; CPU остановлен\n'}) if reason is not None: self.source_step = None self.running = False self._emit('stopped', {'reason': reason, 'location': location}) return # Машинный over может ждать клавишу сколь угодно долго. Пока CPU # выполняется, этот автомат не занимает session lock и не ставит # таймер; пользователь может направить ввод в MAME или нажать Pause. self.session.bridge.request(step['command']) step['instructions'] += 1 step['location'] = location def _set_source_breakpoints(self, filename: str, specifications) -> dict: filename = str(Path(filename).resolve()) requested = [] for value in specifications: if not isinstance(value, dict): raise SessionError('Описание breakpoint должно быть объектом') line = int(value['line']) if line <= 0 or any(item['line'] == line for item in requested): raise SessionError('Номер строки должен быть положительным и уникальным') if value.get('condition') or value.get('hitCondition'): raise SessionError('Условия и hitCondition пока не поддержаны') message = value.get('logMessage') if message is not None: self._validate_log_message(message) requested.append({'line': line, 'logMessage': message}) created = [] results = [] try: for specification in requested: line = specification['line'] item = self.session.break_line(filename, line, enabled=False) created.append(item['id']) kind = 'log' if specification['logMessage'] is not None else 'stop' self.breakpoint_info[item['id']] = { 'kind': kind, 'message': specification['logMessage'], 'locations': item['locations'], 'hits': 0, } results.append({'line': line, 'verified': True, 'logMessage': specification['logMessage'], **item}) except BaseException: for identifier in created: self.breakpoint_info.pop(identifier, None) try: self.session.clear_breakpoint(identifier) except BaseException: pass raise previous = self.source_breakpoints.get(filename, []) for identifier in previous: self.breakpoint_info.pop(identifier, None) self.session.clear_breakpoint(identifier) self.source_breakpoints[filename] = created self.session.activate_breakpoints() return {'file': filename, 'breakpoints': results} def _set_function_breakpoints(self, names) -> dict: requested = [] for value in names: name = str(value) if not name or name in requested: raise SessionError('Имя функции должно быть непустым и уникальным') requested.append(name) created = [] results = [] try: for name in requested: item = self.session.break_function(name, enabled=False) created.append(item['id']) self.breakpoint_info[item['id']] = { 'kind': 'stop', 'message': None, 'locations': item['locations'], 'hits': 0} results.append({'name': name, 'verified': True, **item}) except BaseException: for identifier in created: self.breakpoint_info.pop(identifier, None) try: self.session.clear_breakpoint(identifier) except BaseException: pass raise for identifier in self.function_breakpoints: self.breakpoint_info.pop(identifier, None) self.session.clear_breakpoint(identifier) self.function_breakpoints = created self.session.activate_breakpoints() return {'breakpoints': results} @staticmethod def _validate_log_message(message: str) -> None: try: validate_log_message(message) except ValueError as error: raise SessionError(str(error)) from error def _read_log_variable(self, name: str, module: str | None) -> dict: if module is None: return self.session.read_variable(name) candidates = [item for item in self.session.model.variables if item['name'] == name and item['module'] in (None, module)] if len(candidates) != 1: raise SessionError('Переменная не найдена или имя неоднозначно') return self.session.read_variable(name, candidates[0]['module']) def _render_log_message(self, message: str, module: str | None = None) -> str: output = [] for literal, name, _, _ in string.Formatter().parse(message): output.append(literal) if name is not None: try: output.append(str(self._read_log_variable(name, module)['value'])) except SessionError as error: output.append('') return ''.join(output) def _handle_logpoints(self, location: dict) -> bool: address = location.get('link_address') matched_logs = [] matched_any_log = False matched_stop = False for info in self.breakpoint_info.values(): if not any(item.get('link_address') == address for item in info['locations']): continue if info['kind'] == 'stop': matched_stop = True else: matched_any_log = True condition = info.get('condition') if condition is not None: try: if not self._read_log_variable(condition, info.get('module'))['value']: continue except SessionError as error: if not info.get('condition_warned'): info['condition_warned'] = True self._emit('output', {'category': 'stderr', 'output': 'sdbg: SDBG_LOGIF ' + info.get('tag', '') + ': ' + str(error) + '\n'}) continue info['hits'] += 1 matched_logs.append(info) for info in matched_logs: rendered = self._render_log_message(info['message'], info.get('module')) self._emit('output', {'category': 'console', 'output': rendered + '\n', 'location': location, 'hit': info['hits'], 'tag': info.get('tag')}) if self.mame_console: try: self.session.bridge.request('console_print', text=rendered) except (BridgeError, OSError) as error: if not info.get('console_warned'): info['console_warned'] = True self._emit('output', {'category': 'stderr', 'output': 'sdbg: MAME console: ' + str(error) + '\n'}) return matched_any_log and not matched_stop class _ThreadedUnixServer(socketserver.ThreadingMixIn, socketserver.UnixStreamServer): daemon_threads = True class SessionRpcServer: def __init__(self, path, controller: SessionController): self.path = Path(path) self.controller = controller self.path.parent.mkdir(parents=True, exist_ok=True) if self.path.exists(): try: with socket.socket(socket.AF_UNIX) as probe: probe.connect(str(self.path)) except OSError: self.path.unlink() else: raise OSError('RPC socket уже занят: ' + str(self.path)) controller_ref = controller class Handler(socketserver.StreamRequestHandler): def handle(self): raw = self.rfile.readline(1_048_577) response = {'id': None, 'ok': False} try: if len(raw) > 1_048_576: raise ValueError('RPC-запрос слишком велик') request = json.loads(raw) if not isinstance(request, dict) or not isinstance(request.get('method'), str): raise ValueError('Неверная структура RPC-запроса') arguments = request.get('arguments', {}) if not isinstance(arguments, dict): raise ValueError('RPC arguments должен быть объектом') response = {'id': request.get('id'), 'ok': True, 'result': controller_ref.call(request['method'], arguments)} except (BridgeError, SessionError, ValueError, TypeError, OSError, KeyError) as error: response.update(error=str(error)) self.wfile.write((json.dumps(response, ensure_ascii=False) + '\n').encode()) self.server = _ThreadedUnixServer(str(self.path), Handler) self.path.chmod(0o600) def serve_forever(self): try: self.server.serve_forever(poll_interval=.1) finally: self.close() def close(self): self.server.server_close() self.path.unlink(missing_ok=True) def rpc_call(path, method: str, arguments=None, timeout=10): request = {'id': 1, 'method': method, 'arguments': arguments or {}} with socket.socket(socket.AF_UNIX) as client: client.settimeout(timeout) client.connect(str(path)) client.sendall((json.dumps(request, ensure_ascii=False) + '\n').encode()) file = client.makefile('rb') raw = file.readline(1_048_577) if not raw: raise SessionError('RPC server закрыл соединение без ответа') response = json.loads(raw) if not response.get('ok'): raise SessionError(response.get('error', 'Неизвестная RPC-ошибка')) return response['result']