chaintool/chainqueue
4
0
Fork 0
Code Issues 1 Pull Requests Projects Releases Wiki Activity

Reinstate block on recv, shorter timeout when have data or submits

Browse Source
This commit is contained in:
nolash 2021-06-02 21:58:35 +02:00
parent 3872d0d9b4
commit 18412cfd41
Signed by: lash
GPG Key ID: 21D2E7BB88C2A746
1 changed files with 33 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
chainqueue/runnable/server.py
View File

@ -76,7 +76,7 @@ class SessionController:
self.srv = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM) self.srv = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM)
self.srv.bind(config.get('SESSION_SOCKET_PATH')) self.srv.bind(config.get('SESSION_SOCKET_PATH'))
self.srv.listen(2) self.srv.listen(2)
self.srv.settimeout(1.0) self.srv.settimeout(4.0)
def shutdown(self, signo, frame): def shutdown(self, signo, frame):
if self.dead: if self.dead:
@ -111,11 +111,14 @@ adapter = EthAdapter(backend)
rpc = EthHTTPConnection(url=config.get('RPC_ENDPOINT'), chain_spec=chain_spec) rpc = EthHTTPConnection(url=config.get('RPC_ENDPOINT'), chain_spec=chain_spec)
if __name__ == '__main__': if __name__ == '__main__':
havesends = 0
while True: while True:
srvs = None srvs = None
try: try:
logg.debug('getting connection')
(srvs, srvs_addr) = ctrl.get_connection() (srvs, srvs_addr) = ctrl.get_connection()
except OSError as e: except OSError as e:
havesends = 0
try: try:
fi = os.stat(config.get('SESSION_SOCKET_PATH')) fi = os.stat(config.get('SESSION_SOCKET_PATH'))
except FileNotFoundError: except FileNotFoundError:
@ -125,16 +128,38 @@ if __name__ == '__main__':
logg.error('entity on socket path is not a socket') logg.error('entity on socket path is not a socket')
break break
if srvs == None: if srvs == None:
logg.debug('timeout (remote socket is none)')
txs = adapter.upcoming(chain_spec) txs = adapter.upcoming(chain_spec)
for k in txs.keys(): for k in txs.keys():
havesends += 1
logg.debug('txs {} {}'.format(k, txs[k])) logg.debug('txs {} {}'.format(k, txs[k]))
adapter.dispatch(chain_spec, rpc, k, txs[k]) adapter.dispatch(chain_spec, rpc, k, txs[k])
if havesends > 0:
ctrl.srv.settimeout(0.1)
else:
ctrl.srv.settimeout(4.0)
continue continue
srvs.setblocking(False) ctrl.srv.settimeout(0.1)
srvs.settimeout(0.1)
data_in = None
try:
data_in = srvs.recv(1024) data_in = srvs.recv(1024)
except BlockingIOError as e:
logg.debug('block io error: {}'.format(e))
continue
data = None
try:
data_in_str = data_in.decode('utf-8') data_in_str = data_in.decode('utf-8')
data = bytes.fromhex(strip_0x(data_in_str)) data = bytes.fromhex(strip_0x(data_in_str))
except ValueError:
logg.error('invalid input "{}"'.format(data_in_str))
continue
logg.debug('recv {} bytes'.format(len(data)))
r = adapter.add(chain_spec, data) r = adapter.add(chain_spec, data)
srvs.send(r.to_bytes(4, byteorder='big')) r = srvs.send(r.to_bytes(4, byteorder='big'))
logg.debug('{} bytes sent'.format(r))
srvs.close()
ctrl.shutdown(None, None) ctrl.shutdown(None, None)