apunkt/trustgraph
1
0
Fork 0
mirror of https://github.com/trustgraph-ai/trustgraph.git synced 2026-07-21 11:11:03 +02:00
Code Issues Projects Releases Packages Wiki Activity Actions 2

Tweak timeouts, reduce stop time

Browse source
This commit is contained in:
Cyber MacGeddon 2025-04-29 21:43:37 +01:00
parent 027b52cd7c
commit 215665ba16
2 changed files with 22 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

8
trustgraph-base/trustgraph/base/publisher.py
View file

@ -21,6 +21,7 @@ class Publisher:
async def stop(self): async def stop(self):
self.running = False self.running = False
await self.task
async def join(self): async def join(self):
await self.stop() await self.stop()
@ -42,7 +43,7 @@ class Publisher:
try: try:
id, item = await asyncio.wait_for( id, item = await asyncio.wait_for(
self.q.get(), self.q.get(),
timeout=0.5 timeout=0.25
) )
except asyncio.TimeoutError: except asyncio.TimeoutError:
continue continue
@ -57,8 +58,11 @@ class Publisher:
except Exception as e: except Exception as e:
print("Exception:", e, flush=True) print("Exception:", e, flush=True)
if not self.running:
return
# If handler drops out, sleep a retry # If handler drops out, sleep a retry
time.sleep(2) await asyncio.sleep(1)
async def send(self, id, item): async def send(self, id, item):
await self.q.put((id, item)) await self.q.put((id, item))

21
trustgraph-base/trustgraph/base/subscriber.py
View file

@ -28,6 +28,7 @@ class Subscriber:
async def stop(self): async def stop(self):
self.running = False self.running = False
await self.task
async def join(self): async def join(self):
await self.stop() await self.stop()
@ -35,6 +36,8 @@ class Subscriber:
async def run(self): async def run(self):
consumer = None
while self.running: while self.running:
if self.metrics: if self.metrics:
@ -59,7 +62,7 @@ class Subscriber:
try: try:
msg = await asyncio.to_thread( msg = await asyncio.to_thread(
consumer.receive, consumer.receive,
timeout_millis=2000 timeout_millis=250
) )
except _pulsar.Timeout: except _pulsar.Timeout:
continue continue
@ -91,7 +94,7 @@ class Subscriber:
# FIXME: Timeout means data goes missing # FIXME: Timeout means data goes missing
await asyncio.wait_for( await asyncio.wait_for(
self.q[id].put(value), self.q[id].put(value),
timeout=2 timeout=1
) )
except Exception as e: except Exception as e:
@ -103,7 +106,7 @@ class Subscriber:
# FIXME: Timeout means data goes missing # FIXME: Timeout means data goes missing
await asyncio.wait_for( await asyncio.wait_for(
q.put(value), q.put(value),
timeout=2 timeout=1
) )
except Exception as e: except Exception as e:
self.metrics.dropped() self.metrics.dropped()
@ -112,13 +115,21 @@ class Subscriber:
except Exception as e: except Exception as e:
print("Subscriber exception:", e, flush=True) print("Subscriber exception:", e, flush=True)
consumer.close() finally:
if consumer:
consumer.close()
consumer = None
if self.metrics: if self.metrics:
self.metrics.state("stopped") self.metrics.state("stopped")
if not self.running:
return
# If handler drops out, sleep a retry # If handler drops out, sleep a retry
time.sleep(2) await asyncio.sleep(1)
async def subscribe(self, id): async def subscribe(self, id):