kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Remove session leak better

Browse Source
This commit is contained in:
nolash 2021-07-14 14:43:51 +02:00
parent 8b2652e551
commit 0131a29417
Signed by untrusted user who does not match committer: lash
GPG Key ID: 21D2E7BB88C2A746
1 changed files with 9 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
apps/cic-eth/cic_eth/runnable/daemons/dispatcher.py
View File

@ -90,6 +90,7 @@ class DispatchSyncer:
def __init__(self, chain_spec): def __init__(self, chain_spec):
self.chain_spec = chain_spec self.chain_spec = chain_spec
self.session = None
def chain(self): def chain(self):
@ -100,7 +101,7 @@ class DispatchSyncer:
c = len(txs.keys()) c = len(txs.keys())
logg.debug('processing {} txs {}'.format(c, list(txs.keys()))) logg.debug('processing {} txs {}'.format(c, list(txs.keys())))
chain_str = str(self.chain_spec) chain_str = str(self.chain_spec)
session = SessionBase.create_session() self.session = SessionBase.create_session()
for k in txs.keys(): for k in txs.keys():
tx_raw = txs[k] tx_raw = txs[k]
tx_raw_bytes = bytes.fromhex(strip_0x(tx_raw)) tx_raw_bytes = bytes.fromhex(strip_0x(tx_raw))
@ -108,8 +109,10 @@ class DispatchSyncer:
try: try:
set_reserved(self.chain_spec, tx['hash'], session=session) set_reserved(self.chain_spec, tx['hash'], session=session)
self.session.commit()
except NotLocalTxError as e: except NotLocalTxError as e:
logg.warning('dispatcher was triggered with non-local tx {}'.format(tx['hash'])) logg.warning('dispatcher was triggered with non-local tx {}'.format(tx['hash']))
self.session.rollback()
continue continue
s_check = celery.signature( s_check = celery.signature(
@ -132,7 +135,8 @@ class DispatchSyncer:
s_check.link(s_send) s_check.link(s_send)
t = s_check.apply_async() t = s_check.apply_async()
logg.info('processed {}'.format(k)) logg.info('processed {}'.format(k))
session.close() self.session.close()
self.session = None
def loop(self, interval): def loop(self, interval):
@ -146,6 +150,9 @@ class DispatchSyncer:
conn = RPCConnection.connect(self.chain_spec, 'default') conn = RPCConnection.connect(self.chain_spec, 'default')
self.process(conn, txs) self.process(conn, txs)
except ConnectionError as e: except ConnectionError as e:
if self.session != None:
self.session.close()
self.session = None
logg.error('connection to node failed: {}'.format(e)) logg.error('connection to node failed: {}'.format(e))
if len(utxs) > 0: if len(utxs) > 0: