kaslo/synapse
1
0
Fork 0
Code Wiki Activity

Count total_events and total_event_bytes within the loop.

Browse Source 
In this case, we still update these counts if we get stuck in the loop
because the server is busy.

Signed-off-by: Olivier Wilkinson (reivilibre) <olivier@librepush.net>
This commit is contained in:
Olivier Wilkinson (reivilibre)
2019-08-29 14:10:21 +01:00
parent d7a692f860
commit 39dbee2a3e
2 changed files with 20 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
synapse/handlers/stats.py
View File

@@ -91,31 +91,31 @@ class StatsHandler(StateDeltasHandler):
return None
# Loop round handling deltas until we're up to date
with Measure(self.clock, "stats_delta"):
while True:
while True:
with Measure(self.clock, "stats_delta"):
deltas = yield self.store.get_current_state_deltas(
self.pos["state_delta_stream_id"]
)
if not deltas:
break
logger.debug("Handling %d state deltas", len(deltas))
yield self._handle_deltas(deltas)
self.pos["state_delta_stream_id"] = deltas[-1]["stream_id"]
yield self.store.update_stats_positions(self.pos)
event_processing_positions.labels("stats").set(
self.pos["state_delta_stream_id"]
)
if self.pos is not None:
yield self.store.update_stats_positions(self.pos)
# Then count deltas for total_events and total_event_bytes.
with Measure(self.clock, "stats_total_events_and_bytes"):
self.pos, had_counts = yield self.store.incremental_update_room_total_events_and_bytes(
self.pos
)
# Then count deltas for total_events and total_event_bytes.
with Measure(self.clock, "stats_total_events_and_bytes"):
self.pos = yield self.store.incremental_update_total_events_and_bytes(
self.pos
)
if not deltas and not had_counts:
break
@defer.inlineCallbacks
def _handle_deltas(self, deltas):

13
synapse/storage/stats.py
View File

@@ -487,9 +487,12 @@ class StatsStore(StateDeltasStore):
in_positions (dict): Positions,
as retrieved from L{get_stats_positions}.
Returns (dict):
The new positions. Note that this is for reference only –
the new positions WILL be committed by this function.
Returns (Deferred[tuple[dict, bool]]):
First element (dict):
The new positions. Note that this is for reference only –
the new positions WILL be committed by this function.
Second element (bool):
true iff there was a change to the positions, false otherwise
"""
def incremental_update_total_events_and_bytes_txn(txn):
@@ -518,7 +521,9 @@ class StatsStore(StateDeltasStore):
self._update_stats_positions_txn(txn, positions)
return positions
return positions, True
else:
return positions, False
return self.runInteraction(
"stats_incremental_total_events_and_bytes",