File size: 7,339 Bytes
450ac2f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 | """Bounded, counted JSON-lines access to one explicitly owned subprocess."""
import json
import math
from pathlib import Path
import queue
import subprocess
import sys
import threading
import time
class InterfaceError(RuntimeError):
def __init__(self, code, message=None):
self.code = code
super().__init__(message or code.replace('_', ' '))
class TargetProcess:
def __init__(self, private, timeout=2.0, *, startup_timeout=10.0):
if isinstance(timeout,bool) or not isinstance(timeout,(int,float)) or not math.isfinite(timeout) or timeout <= 0 or timeout > 30:
raise ValueError('timeout must be finite and between zero and 30 seconds')
if (isinstance(startup_timeout, bool) or not isinstance(startup_timeout, (int, float))
or not math.isfinite(startup_timeout) or not 0 < startup_timeout <= 30):
raise ValueError('startup_timeout must be finite and between zero and 30 seconds')
self.timeout = timeout
self.startup_timeout = startup_timeout
self.counts = dict(reset_attempts=0,step_attempts=0,reset_successes=0,step_successes=0,target_calls=0)
self.last_trace = {'word':[],'outputs':[],'complete':False}
self._sequence, self._closed = 0, False
self._responses = queue.Queue()
self._process = subprocess.Popen([sys.executable, '-m', 'crosswire.target_worker'],
cwd=str(Path(__file__).resolve().parent.parent), stdin=subprocess.PIPE,
stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, bufsize=0,
creationflags=getattr(subprocess,'CREATE_NO_WINDOW',0))
self.pid = self._process.pid
self._reader = threading.Thread(target=self._read, daemon=True)
self._reader.start()
try:
# Interpreter/module startup is not a reset or action attempt. Keep
# its bounded handshake separate from the strict per-call deadline.
self._request('init', request_timeout=self.startup_timeout, config=private)
except BaseException:
self.close()
raise
def _read(self):
try:
while True:
line = self._process.stdout.readline(65537)
if not line:
self._responses.put(('error','worker_exit'))
return
if len(line) > 65536 or not line.endswith(b'\n'):
self._responses.put(('error','malformed_response'))
return
self._responses.put(('line',line))
except (OSError, ValueError):
self._responses.put(('error','worker_exit'))
def _request(self, operation, *, request_timeout=None, **payload):
try:
if self._closed:
raise InterfaceError('closed')
self._sequence += 1
message = {'id':self._sequence,'op':operation,**payload}
encoded = (json.dumps(message,separators=(',',':'),allow_nan=False)+'\n').encode('utf-8')
if len(encoded)>16384:
raise InterfaceError('invalid_request')
limit = self.timeout if request_timeout is None else request_timeout
deadline = time.monotonic() + limit
written = queue.Queue(maxsize=1)
def write():
try:
self._process.stdin.write(encoded)
self._process.stdin.flush()
written.put(None)
except (OSError,ValueError) as error:
written.put(error)
threading.Thread(target=write,daemon=True).start()
try:
if written.get(timeout=limit) is not None:
raise InterfaceError('worker_exit')
kind, raw = self._responses.get(timeout=max(0,deadline-time.monotonic()))
except queue.Empty:
raise InterfaceError('timeout') from None
if kind == 'error':
raise InterfaceError(raw)
try:
response = json.loads(raw,parse_constant=lambda value: (_ for _ in ()).throw(ValueError(value)))
except (ValueError,UnicodeError):
raise InterfaceError('malformed_response') from None
if not isinstance(response,dict) or type(response.get('id')) is not int or response['id'] != self._sequence:
raise InterfaceError('malformed_response')
if response.get('ok') is not True:
raise InterfaceError('target_error')
expected = {'id','ok','output'} if operation == 'step' else {'id','ok'}
if set(response) != expected:
raise InterfaceError('malformed_response')
if operation == 'step':
output = response['output']
if type(output) is not int or not 0 <= output < 4:
raise InterfaceError('malformed_response')
return output
return None
except InterfaceError:
self.close()
raise
except (OSError,ValueError,TypeError):
self.close()
raise InterfaceError('worker_exit') from None
def reset(self):
self.counts['reset_attempts'] += 1
self.counts['target_calls'] += 1
self._request('reset')
self.counts['reset_successes'] += 1
def step(self, input):
self.counts['step_attempts'] += 1
self.counts['target_calls'] += 1
if type(input) is not int or not 0 <= input < 3:
self.close()
raise InterfaceError('invalid_input')
output = self._request('step',input=input)
self.counts['step_successes'] += 1
return output
def query(self, word):
self.last_trace = {'word':list(word),'outputs':[],'complete':False}
try:
self.reset()
for symbol in self.last_trace['word']:
self.last_trace['outputs'].append(self.step(symbol))
except InterfaceError as error:
error.trace = {'word':list(self.last_trace['word']),
'outputs':list(self.last_trace['outputs']),'complete':False}
raise
self.last_trace['complete'] = True
return list(self.last_trace['outputs'])
def set_phase(self, phase):
self._request('set_phase', phase=phase)
def close(self):
if self._closed:
return
self._closed = True
process = self._process
if process.poll() is None:
# Kill only this Popen object, never a name, process group or tree.
process.terminate()
try:
process.wait(timeout=0.5)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=0.5)
for pipe in (process.stdin,process.stdout):
try:
pipe.close()
except OSError:
pass
if threading.current_thread() is not self._reader:
self._reader.join(timeout=0.5)
def __enter__(self):
return self
def __exit__(self, *args):
self.close()
|