"""Compare cold/warm full Monaco initialization in two verified native images.
Actual PS/2 navigation, model readback, presented software-rendered glyphs, and
keyboard editing. No pre-warming, injected JS, host browser or substituted frame.
"""
import argparse
import hashlib
import json
from pathlib import Path
import re
import shutil
import socket
import statistics
import subprocess
import time
from PIL import Image
from TestBoot import Qmp
from IndependentBingEvidence import PRESENTED, window_geometry
from MultiprocessorEvidence import validate_multiprocessors
from TestBrowserPipeline import _monaco_frame_stats, _monaco_fourth_line_delta
ROOT = Path(__file__).resolve().parents[2]
ADDRESS = 'http://xfesystem.local/Browser/Monaco/Index.html'
VALUE = '// XFESystem | XFEstudio\nconst answer = 6 * 7;\nconsole.log(answer);'
MODELS = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)
def digest(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def run(receipt_path, output, iterations, responsiveness=False):
receipt = json.loads(receipt_path.read_text(encoding='utf-8-sig'))
assert receipt['Passed'] and receipt['ContainsCosmos'] is False
image = Path(receipt['ImagePath'])
assert digest(image) == receipt['ImageSha256']
output.mkdir(parents=True)
shutil.copyfile(__file__, output / 'Harness.py')
result = dict(Kind='XfeNativeFullMonacoStartup', Passed=False,
ImageSha256=receipt['ImageSha256'], VerifiedReceipt=str(receipt_path),
VerifiedReceiptSha256=digest(receipt_path), HarnessSha256=digest(__file__),
VirtualCpus=8, ExecutingCpus=None, ManagedSchedulerCpus=1, ManagedSmpVerified=False, HostBrowserUsed=False, Prewarmed=False,
HostScreenshotSubstitution=False, Cases=[])
serial_path = output / 'Serial.log'
qemu = qmp = None
def text():
return serial_path.read_text(encoding='utf-8', errors='replace') if serial_path.exists() else ''
def wait(predicate, seconds=120, stage='Full editor initialization'):
end = time.monotonic() + seconds
while time.monotonic() < end:
assert qemu.poll() is None, 'Guest exited'
log = text()
assert not any(m in log for m in ('XFE_NATIVE_FAIL', 'XFE_RUNTIME_FATAL', 'XFE_ZERO_FAILFAST',
'XFE_INDEPENDENT_APPLICATION_FAILED', 'XFE_EDITOR_BACKGROUND|SessionError|', 'XFE_BROWSER_PAINT_FAILED'))
if predicate(log):
return log
time.sleep(.02)
raise AssertionError(stage + ' timeout: ' + text()[-4096:])
def key(chord):
qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
time.sleep(.12)
def type_text(value):
codes = {':': 'shift-semicolon', '.': 'dot', '/': 'slash', ' ': 'spc'}
for char in value:
key(codes.get(char, 'shift-' + char.lower() if char.isupper() else char))
def frame(name):
qmp.call('screendump', {'filename': str(output / (name + '.ppm'))})
with Image.open(output / (name + '.ppm')) as image:
rgb = image.convert('RGB')
rgb.save(output / (name + '.png'))
return rgb
def browser_visible(log):
try:
window_geometry(log, 'Browser')
return True
except AssertionError:
return False
try:
shutil.copy2(ROOT / '.tools/firmware/edk2-i386-vars.fd', output / 'FirmwareVars.fd')
with socket.socket() as reservation:
reservation.bind(('127.0.0.1', 0))
port = reservation.getsockname()[1]
command = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35', '-accel', 'tcg',
'-cpu', 'max', '-smp', '8', '-m', '1024M', '-vga', 'std', '-global', 'VGA.vgamem_mb=64',
'-display', 'none', '-net', 'none', '-rtc', 'base=utc',
'-drive', f'if=pflash,format=raw,readonly=on,file={ROOT / ".tools/firmware/edk2-x86_64-code.fd"}',
'-drive', f'if=pflash,format=raw,file={output / "FirmwareVars.fd"}',
'-drive', f'file={image},format=raw,if=ide,snapshot=on', '-serial', f'file:{serial_path}',
'-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown']
result['QemuArguments'] = command
with (output / 'Qemu.log').open('w') as log:
qemu = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
end = time.monotonic() + 30
while qmp is None:
try:
qmp = Qmp(port)
except OSError:
assert time.monotonic() < end
time.sleep(.1)
wait(lambda s: 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME' in s)
wait(lambda s: 'XFE_DESKTOP_STATE_END' in s)
if 'XFE_DESKTOP_ACTIVE|Terminal' in text():
key('alt-f4')
key('ctrl-esc'); key('down'); key('down'); key('ret')
wait(browser_visible)
browser_id, _ = window_geometry(text(), 'Browser')
wait(lambda s: f'XFE_EDITOR_BACKGROUND|Ready|Window|{browser_id}|' in s)
initial_frame = None
viewport = None
for iteration in range(1, iterations + 1):
key('ctrl-l'); type_text('about:Monaco')
start = len(text())
begun = time.monotonic_ns()
qmp.call('human-monitor-command', {'command-line': 'sendkey ret'})
def ready(log):
models = MODELS.findall(log, start)
return models and models[-1] == VALUE and any(m[3] == ADDRESS for m in PRESENTED.finditer(log, start))
log = wait(ready)
_, (x, y, width, height) = window_geometry(log, 'Browser')
viewport = (x + 2, y + 67, width - 4, height - 96)
deadline = time.monotonic() + 60
previous_pixels = None
while True:
rgb = frame('Ready' + str(iteration))
stats = _monaco_frame_stats(rgb, viewport, 1)
web = rgb.crop((viewport[0], viewport[1], viewport[0] + viewport[2], viewport[1] + viewport[3]))
pixels = hashlib.sha256(web.tobytes()).hexdigest()
# A scanout can be sampled while it is being copied. Require
# the actual glyph frame to agree on two successive captures.
if stats['ReadyPainted'] and previous_pixels == pixels:
break
previous_pixels = pixels if stats['ReadyPainted'] else None
assert time.monotonic() < deadline, 'The full editor source and syntax glyphs were not painted'
wait(lambda s: True)
time.sleep(.1)
elapsed = (time.monotonic_ns() - begun) / 1_000_000
log = text()
web.save(output / ('Viewport' + str(iteration) + '.png'))
timings = []
for line in log[start:].splitlines():
if not line.startswith('XFE_BROWSER_SCRIPT_TIMING|'):
continue
fields = line.split('|')
values = dict(zip(fields[2::2], fields[3::2]))
frequency = int(values['Frequency'])
timings.append(dict(Resource=fields[1], CacheHit=values.get('CacheHit'),
**{('Total' if k == 'Ticks' else k.removesuffix('Ticks')): int(v) * 1000 / frequency
for k, v in values.items() if k.endswith('Ticks')}))
result['Cases'].append(dict(Iteration=iteration, Cold=iteration == 1,
EnterToFullPresentedMilliseconds=elapsed, ExactModel=VALUE, NativeGlyphs=stats,
ScriptTimingsMilliseconds=timings, WebPixelsSha256=hashlib.sha256(web.tobytes()).hexdigest()))
initial_frame = rgb
time.sleep(.15)
if responsiveness:
# Separate from startup timing: exercise the actual desktop
# while the worker executes a fresh editor initialization.
# The start menu must be processed AND presented by the UI.
key('ctrl-l'); type_text('about:Monaco')
start = len(text())
qmp.call('human-monitor-command', {'command-line': 'sendkey ret'})
wait(lambda s: 'XFE_BROWSER_SCRIPT_BEGIN|' in s[start:])
menu_start = len(text())
sent = time.monotonic_ns()
# Windows-key release is routed to the desktop even when a
# completed Monaco owns keyboard focus. Ctrl+Esc goes to the
# page in that state, so it is unsuitable for an idle control.
qmp.call('human-monitor-command', {'command-line': 'sendkey meta_l'})
menu_log = wait(lambda s: 'XFE_DESKTOP_FLAGS|1|' in s[menu_start:], stage='Menu during initialization')
menu_ms = (time.monotonic_ns() - sent) / 1_000_000
frame('MenuDuringInitialization')
time.sleep(.15)
close_start = len(text()); key('esc')
wait(lambda s: 'XFE_DESKTOP_FLAGS|0|' in s[close_start:])
wait(ready)
# Same input/presentation path while idle distinguishes guest
# stalls from QMP keyboard delivery and serial polling overhead.
idle_start = len(text()); idle_sent = time.monotonic_ns()
qmp.call('human-monitor-command', {'command-line': 'sendkey meta_l'})
wait(lambda s: 'XFE_DESKTOP_FLAGS|1|' in s[idle_start:], stage='Idle menu control')
idle_ms = (time.monotonic_ns() - idle_sent) / 1_000_000
time.sleep(.15)
close_start = len(text()); key('esc')
wait(lambda s: 'XFE_DESKTOP_FLAGS|0|' in s[close_start:])
result['ResponsivenessProbe'] = dict(Kind='ActualDesktopStartMenuDuringWorkerScript',
KeyboardToPresentedMenuMilliseconds=menu_ms,
IdleKeyboardToPresentedMenuMilliseconds=idle_ms,
ExtraInitializationLatencyMilliseconds=menu_ms-idle_ms,
InitializationStillRunningAtMenuPresentation='XFE_MONACO_VALUE_BEGIN' not in menu_log[start:],
TimingIncludesQmpAndSerialPolling=True, HostScreenshotSubstitution=False)
# Native key packets must change the actual model AND add new glyphs.
# Click the real text surface using relative PS/2 movement.
pointer = [640, 400]
target = [viewport[0] + 160, viewport[1] + 70]
while pointer != target:
delta = [max(-80, min(80, target[i] - pointer[i])) for i in range(2)]
qmp.call('input-send-event', {'events': [
{'type': 'rel', 'data': {'axis': axis, 'value': delta[i]}} for i, axis in enumerate(('x', 'y'))]})
pointer = [pointer[i] + delta[i] for i in range(2)]
time.sleep(.025)
for down in (True, False):
qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'left'}}]})
time.sleep(.1)
start = len(text()); key('ctrl-end'); key('ret'); type_text('xfenativeedit')
expected = VALUE + '\nxfenativeedit'
wait(lambda s: MODELS.findall(s, start) and MODELS.findall(s, start)[-1] == expected)
edited = frame('Edited')
stats = _monaco_frame_stats(edited, viewport, 1)
delta = _monaco_fourth_line_delta(initial_frame, edited, stats['Lines'][3]['NativeBounds'])
assert stats['Lines'][3]['HasTextGlyphDistribution'] and delta['NewInkPixels'] >= 35 and delta['NewInkColumns'] >= 24
if 'XFE_MULTIPROCESSOR|' in text():
result['ExecutingCpus'] = validate_multiprocessors(text(), 8)['ExecutingCpus']
else:
assert receipt.get('ExecutingCpus') == 1, 'Missing actual AP evidence'
result['ExecutingCpus'] = 1
result.update(Passed=True, KeyboardEditedExactModel=True, EditedModel=expected, EditedGlyphDelta=delta)
except Exception as error:
result['Error'] = str(error)
finally:
if qemu is not None:
if qemu.poll() is None and qmp is not None:
try:
qmp.call('quit')
except (OSError, RuntimeError):
pass
try:
qemu.wait(timeout=10)
except subprocess.TimeoutExpired:
qemu.terminate(); qemu.wait(timeout=10); result['ForcedCleanup'] = True
result['QemuExitCode'] = qemu.returncode
if qmp is not None:
qmp.close()
result['ImageUnchanged'] = digest(image) == receipt['ImageSha256']
result['Passed'] &= result.get('QemuExitCode') == 0 and result['ImageUnchanged']
(output / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
print(json.dumps({k: v for k, v in result.items() if k not in ('Cases', 'QemuArguments')}, ensure_ascii=True))
return result
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--baseline-proof', type=Path, required=True)
parser.add_argument('--optimized-proof', type=Path, required=True)
parser.add_argument('--iterations', type=int, default=5, choices=range(3, 10))
parser.add_argument('--responsiveness', action='store_true')
args = parser.parse_args()
output = ROOT / 'build/MonacoInit' / f'NativeComparison-{time.time_ns()}'
before = run(args.baseline_proof.resolve(), output / 'Before', args.iterations, args.responsiveness)
after = run(args.optimized_proof.resolve(), output / 'After', args.iterations, args.responsiveness)
passed = before['Passed'] and after['Passed']
comparison = dict(Kind='XfeNativeFullMonacoStartupComparison', Passed=passed, FullReleaseQualified=False,
Method='QEMU TCG max, 8 vCPUs / 1 executing BSP; sequential cold/warm navigation to the unchanged official Monaco editor',
Before=str(output / 'Before/Result.json'), After=str(output / 'After/Result.json'))
if passed:
old = [r['EnterToFullPresentedMilliseconds'] for r in before['Cases']]
new = [r['EnterToFullPresentedMilliseconds'] for r in after['Cases']]
models_equal = [r['ExactModel'] for r in before['Cases']] == [r['ExactModel'] for r in after['Cases']]
pixels_equal = [r['WebPixelsSha256'] for r in before['Cases']] == [r['WebPixelsSha256'] for r in after['Cases']]
passed = models_equal and pixels_equal
comparison['Passed'] = passed
comparison.update(BeforeColdMilliseconds=old[0], AfterColdMilliseconds=new[0], ColdSpeedup=old[0] / new[0],
BeforeWarmMedianMilliseconds=statistics.median(old[1:]), AfterWarmMedianMilliseconds=statistics.median(new[1:]),
AllNativeViewportPixelsEqual=pixels_equal, ExactModelsEqual=models_equal, BothImagesPassedActualKeyboardEditing=True)
if not passed:
comparison['Error'] = 'Native model or viewport pixel equality failed'
(output / 'Comparison.json').write_text(json.dumps(comparison, indent=2), encoding='utf-8')
print(json.dumps(comparison))
raise SystemExit(0 if passed else 1)
"""Compare cold/warm full Monaco initialization in two verified native images.
Actual PS/2 navigation, model readback, presented software-rendered glyphs, and
keyboard editing. No pre-warming, injected JS, host browser or substituted frame.
"""
import argparse
import hashlib
import json
from pathlib import Path
import re
import shutil
import socket
import statistics
import subprocess
import time
from PIL import Image
from TestBoot import Qmp
from IndependentBingEvidence import PRESENTED, window_geometry
from MultiprocessorEvidence import validate_multiprocessors
from TestBrowserPipeline import _monaco_frame_stats, _monaco_fourth_line_delta
ROOT = Path(__file__).resolve().parents[2]
ADDRESS = 'http://xfesystem.local/Browser/Monaco/Index.html'
VALUE = '// XFESystem | XFEstudio\nconst answer = 6 * 7;\nconsole.log(answer);'
MODELS = re.compile(r'XFE_MONACO_VALUE_BEGIN\r?\n(.*?)\r?\nXFE_MONACO_VALUE_END', re.S)
def digest(path):
return hashlib.sha256(Path(path).read_bytes()).hexdigest()
def run(receipt_path, output, iterations, responsiveness=False):
receipt = json.loads(receipt_path.read_text(encoding='utf-8-sig'))
assert receipt['Passed'] and receipt['ContainsCosmos'] is False
image = Path(receipt['ImagePath'])
assert digest(image) == receipt['ImageSha256']
output.mkdir(parents=True)
shutil.copyfile(__file__, output / 'Harness.py')
result = dict(Kind='XfeNativeFullMonacoStartup', Passed=False,
ImageSha256=receipt['ImageSha256'], VerifiedReceipt=str(receipt_path),
VerifiedReceiptSha256=digest(receipt_path), HarnessSha256=digest(__file__),
VirtualCpus=8, ExecutingCpus=None, ManagedSchedulerCpus=1, ManagedSmpVerified=False, HostBrowserUsed=False, Prewarmed=False,
HostScreenshotSubstitution=False, Cases=[])
serial_path = output / 'Serial.log'
qemu = qmp = None
def text():
return serial_path.read_text(encoding='utf-8', errors='replace') if serial_path.exists() else ''
def wait(predicate, seconds=120, stage='Full editor initialization'):
end = time.monotonic() + seconds
while time.monotonic() < end:
assert qemu.poll() is None, 'Guest exited'
log = text()
assert not any(m in log for m in ('XFE_NATIVE_FAIL', 'XFE_RUNTIME_FATAL', 'XFE_ZERO_FAILFAST',
'XFE_INDEPENDENT_APPLICATION_FAILED', 'XFE_EDITOR_BACKGROUND|SessionError|', 'XFE_BROWSER_PAINT_FAILED'))
if predicate(log):
return log
time.sleep(.02)
raise AssertionError(stage + ' timeout: ' + text()[-4096:])
def key(chord):
qmp.call('human-monitor-command', {'command-line': 'sendkey ' + chord})
time.sleep(.12)
def type_text(value):
codes = {':': 'shift-semicolon', '.': 'dot', '/': 'slash', ' ': 'spc'}
for char in value:
key(codes.get(char, 'shift-' + char.lower() if char.isupper() else char))
def frame(name):
qmp.call('screendump', {'filename': str(output / (name + '.ppm'))})
with Image.open(output / (name + '.ppm')) as image:
rgb = image.convert('RGB')
rgb.save(output / (name + '.png'))
return rgb
def browser_visible(log):
try:
window_geometry(log, 'Browser')
return True
except AssertionError:
return False
try:
shutil.copy2(ROOT / '.tools/firmware/edk2-i386-vars.fd', output / 'FirmwareVars.fd')
with socket.socket() as reservation:
reservation.bind(('127.0.0.1', 0))
port = reservation.getsockname()[1]
command = [str(ROOT / '.tools/qemu/qemu-system-x86_64.exe'), '-machine', 'q35', '-accel', 'tcg',
'-cpu', 'max', '-smp', '8', '-m', '1024M', '-vga', 'std', '-global', 'VGA.vgamem_mb=64',
'-display', 'none', '-net', 'none', '-rtc', 'base=utc',
'-drive', f'if=pflash,format=raw,readonly=on,file={ROOT / ".tools/firmware/edk2-x86_64-code.fd"}',
'-drive', f'if=pflash,format=raw,file={output / "FirmwareVars.fd"}',
'-drive', f'file={image},format=raw,if=ide,snapshot=on', '-serial', f'file:{serial_path}',
'-qmp', f'tcp:127.0.0.1:{port},server=on,wait=off', '-no-reboot', '-no-shutdown']
result['QemuArguments'] = command
with (output / 'Qemu.log').open('w') as log:
qemu = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
creationflags=getattr(subprocess, 'CREATE_NO_WINDOW', 0))
end = time.monotonic() + 30
while qmp is None:
try:
qmp = Qmp(port)
except OSError:
assert time.monotonic() < end
time.sleep(.1)
wait(lambda s: 'XFE_INDEPENDENT_DESKTOP_FIRST_FRAME' in s)
wait(lambda s: 'XFE_DESKTOP_STATE_END' in s)
if 'XFE_DESKTOP_ACTIVE|Terminal' in text():
key('alt-f4')
key('ctrl-esc'); key('down'); key('down'); key('ret')
wait(browser_visible)
browser_id, _ = window_geometry(text(), 'Browser')
wait(lambda s: f'XFE_EDITOR_BACKGROUND|Ready|Window|{browser_id}|' in s)
initial_frame = None
viewport = None
for iteration in range(1, iterations + 1):
key('ctrl-l'); type_text('about:Monaco')
start = len(text())
begun = time.monotonic_ns()
qmp.call('human-monitor-command', {'command-line': 'sendkey ret'})
def ready(log):
models = MODELS.findall(log, start)
return models and models[-1] == VALUE and any(m[3] == ADDRESS for m in PRESENTED.finditer(log, start))
log = wait(ready)
_, (x, y, width, height) = window_geometry(log, 'Browser')
viewport = (x + 2, y + 67, width - 4, height - 96)
deadline = time.monotonic() + 60
previous_pixels = None
while True:
rgb = frame('Ready' + str(iteration))
stats = _monaco_frame_stats(rgb, viewport, 1)
web = rgb.crop((viewport[0], viewport[1], viewport[0] + viewport[2], viewport[1] + viewport[3]))
pixels = hashlib.sha256(web.tobytes()).hexdigest()
# A scanout can be sampled while it is being copied. Require
# the actual glyph frame to agree on two successive captures.
if stats['ReadyPainted'] and previous_pixels == pixels:
break
previous_pixels = pixels if stats['ReadyPainted'] else None
assert time.monotonic() < deadline, 'The full editor source and syntax glyphs were not painted'
wait(lambda s: True)
time.sleep(.1)
elapsed = (time.monotonic_ns() - begun) / 1_000_000
log = text()
web.save(output / ('Viewport' + str(iteration) + '.png'))
timings = []
for line in log[start:].splitlines():
if not line.startswith('XFE_BROWSER_SCRIPT_TIMING|'):
continue
fields = line.split('|')
values = dict(zip(fields[2::2], fields[3::2]))
frequency = int(values['Frequency'])
timings.append(dict(Resource=fields[1], CacheHit=values.get('CacheHit'),
**{('Total' if k == 'Ticks' else k.removesuffix('Ticks')): int(v) * 1000 / frequency
for k, v in values.items() if k.endswith('Ticks')}))
result['Cases'].append(dict(Iteration=iteration, Cold=iteration == 1,
EnterToFullPresentedMilliseconds=elapsed, ExactModel=VALUE, NativeGlyphs=stats,
ScriptTimingsMilliseconds=timings, WebPixelsSha256=hashlib.sha256(web.tobytes()).hexdigest()))
initial_frame = rgb
time.sleep(.15)
if responsiveness:
# Separate from startup timing: exercise the actual desktop
# while the worker executes a fresh editor initialization.
# The start menu must be processed AND presented by the UI.
key('ctrl-l'); type_text('about:Monaco')
start = len(text())
qmp.call('human-monitor-command', {'command-line': 'sendkey ret'})
wait(lambda s: 'XFE_BROWSER_SCRIPT_BEGIN|' in s[start:])
menu_start = len(text())
sent = time.monotonic_ns()
# Windows-key release is routed to the desktop even when a
# completed Monaco owns keyboard focus. Ctrl+Esc goes to the
# page in that state, so it is unsuitable for an idle control.
qmp.call('human-monitor-command', {'command-line': 'sendkey meta_l'})
menu_log = wait(lambda s: 'XFE_DESKTOP_FLAGS|1|' in s[menu_start:], stage='Menu during initialization')
menu_ms = (time.monotonic_ns() - sent) / 1_000_000
frame('MenuDuringInitialization')
time.sleep(.15)
close_start = len(text()); key('esc')
wait(lambda s: 'XFE_DESKTOP_FLAGS|0|' in s[close_start:])
wait(ready)
# Same input/presentation path while idle distinguishes guest
# stalls from QMP keyboard delivery and serial polling overhead.
idle_start = len(text()); idle_sent = time.monotonic_ns()
qmp.call('human-monitor-command', {'command-line': 'sendkey meta_l'})
wait(lambda s: 'XFE_DESKTOP_FLAGS|1|' in s[idle_start:], stage='Idle menu control')
idle_ms = (time.monotonic_ns() - idle_sent) / 1_000_000
time.sleep(.15)
close_start = len(text()); key('esc')
wait(lambda s: 'XFE_DESKTOP_FLAGS|0|' in s[close_start:])
result['ResponsivenessProbe'] = dict(Kind='ActualDesktopStartMenuDuringWorkerScript',
KeyboardToPresentedMenuMilliseconds=menu_ms,
IdleKeyboardToPresentedMenuMilliseconds=idle_ms,
ExtraInitializationLatencyMilliseconds=menu_ms-idle_ms,
InitializationStillRunningAtMenuPresentation='XFE_MONACO_VALUE_BEGIN' not in menu_log[start:],
TimingIncludesQmpAndSerialPolling=True, HostScreenshotSubstitution=False)
# Native key packets must change the actual model AND add new glyphs.
# Click the real text surface using relative PS/2 movement.
pointer = [640, 400]
target = [viewport[0] + 160, viewport[1] + 70]
while pointer != target:
delta = [max(-80, min(80, target[i] - pointer[i])) for i in range(2)]
qmp.call('input-send-event', {'events': [
{'type': 'rel', 'data': {'axis': axis, 'value': delta[i]}} for i, axis in enumerate(('x', 'y'))]})
pointer = [pointer[i] + delta[i] for i in range(2)]
time.sleep(.025)
for down in (True, False):
qmp.call('input-send-event', {'events': [{'type': 'btn', 'data': {'down': down, 'button': 'left'}}]})
time.sleep(.1)
start = len(text()); key('ctrl-end'); key('ret'); type_text('xfenativeedit')
expected = VALUE + '\nxfenativeedit'
wait(lambda s: MODELS.findall(s, start) and MODELS.findall(s, start)[-1] == expected)
edited = frame('Edited')
stats = _monaco_frame_stats(edited, viewport, 1)
delta = _monaco_fourth_line_delta(initial_frame, edited, stats['Lines'][3]['NativeBounds'])
assert stats['Lines'][3]['HasTextGlyphDistribution'] and delta['NewInkPixels'] >= 35 and delta['NewInkColumns'] >= 24
if 'XFE_MULTIPROCESSOR|' in text():
result['ExecutingCpus'] = validate_multiprocessors(text(), 8)['ExecutingCpus']
else:
assert receipt.get('ExecutingCpus') == 1, 'Missing actual AP evidence'
result['ExecutingCpus'] = 1
result.update(Passed=True, KeyboardEditedExactModel=True, EditedModel=expected, EditedGlyphDelta=delta)
except Exception as error:
result['Error'] = str(error)
finally:
if qemu is not None:
if qemu.poll() is None and qmp is not None:
try:
qmp.call('quit')
except (OSError, RuntimeError):
pass
try:
qemu.wait(timeout=10)
except subprocess.TimeoutExpired:
qemu.terminate(); qemu.wait(timeout=10); result['ForcedCleanup'] = True
result['QemuExitCode'] = qemu.returncode
if qmp is not None:
qmp.close()
result['ImageUnchanged'] = digest(image) == receipt['ImageSha256']
result['Passed'] &= result.get('QemuExitCode') == 0 and result['ImageUnchanged']
(output / 'Result.json').write_text(json.dumps(result, indent=2), encoding='utf-8')
print(json.dumps({k: v for k, v in result.items() if k not in ('Cases', 'QemuArguments')}, ensure_ascii=True))
return result
if __name__ == '__main__':
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--baseline-proof', type=Path, required=True)
parser.add_argument('--optimized-proof', type=Path, required=True)
parser.add_argument('--iterations', type=int, default=5, choices=range(3, 10))
parser.add_argument('--responsiveness', action='store_true')
args = parser.parse_args()
output = ROOT / 'build/MonacoInit' / f'NativeComparison-{time.time_ns()}'
before = run(args.baseline_proof.resolve(), output / 'Before', args.iterations, args.responsiveness)
after = run(args.optimized_proof.resolve(), output / 'After', args.iterations, args.responsiveness)
passed = before['Passed'] and after['Passed']
comparison = dict(Kind='XfeNativeFullMonacoStartupComparison', Passed=passed, FullReleaseQualified=False,
Method='QEMU TCG max, 8 vCPUs / 1 executing BSP; sequential cold/warm navigation to the unchanged official Monaco editor',
Before=str(output / 'Before/Result.json'), After=str(output / 'After/Result.json'))
if passed:
old = [r['EnterToFullPresentedMilliseconds'] for r in before['Cases']]
new = [r['EnterToFullPresentedMilliseconds'] for r in after['Cases']]
models_equal = [r['ExactModel'] for r in before['Cases']] == [r['ExactModel'] for r in after['Cases']]
pixels_equal = [r['WebPixelsSha256'] for r in before['Cases']] == [r['WebPixelsSha256'] for r in after['Cases']]
passed = models_equal and pixels_equal
comparison['Passed'] = passed
comparison.update(BeforeColdMilliseconds=old[0], AfterColdMilliseconds=new[0], ColdSpeedup=old[0] / new[0],
BeforeWarmMedianMilliseconds=statistics.median(old[1:]), AfterWarmMedianMilliseconds=statistics.median(new[1:]),
AllNativeViewportPixelsEqual=pixels_equal, ExactModelsEqual=models_equal, BothImagesPassedActualKeyboardEditing=True)
if not passed:
comparison['Error'] = 'Native model or viewport pixel equality failed'
(output / 'Comparison.json').write_text(json.dumps(comparison, indent=2), encoding='utf-8')
print(json.dumps(comparison))
raise SystemExit(0 if passed else 1)
