From 891dfd90bd97ad485d54dfae3e510c640de8e978 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Fri, 14 Aug 2015 15:42:52 +0100 Subject: [PATCH 1/3] Fix pending_calls metric to not lie --- synapse/metrics/__init__.py | 21 ++++++++++++++++++--- 1 file changed, 18 insertions(+), 3 deletions(-) diff --git a/synapse/metrics/__init__.py b/synapse/metrics/__init__.py index 0be9772991..7b8a20108b 100644 --- a/synapse/metrics/__init__.py +++ b/synapse/metrics/__init__.py @@ -158,18 +158,33 @@ def runUntilCurrentTimer(func): @functools.wraps(func) def f(*args, **kwargs): - pending_calls = len(reactor.getDelayedCalls()) + now = reactor.seconds() + num_pending = 0 + + # _newTimedCalls is one long list of *all* pending calls. Below loop + # is based off of impl of reactor.runUntilCurrent + for p in reactor._newTimedCalls: + if p.time > now: + break + + if p.delayed_time > 0: + continue + + num_pending += 1 + + num_pending += len(reactor.threadCallQueue) + start = time.time() * 1000 ret = func(*args, **kwargs) end = time.time() * 1000 tick_time.inc_by(end - start) - pending_calls_metric.inc_by(pending_calls) + pending_calls_metric.inc_by(num_pending) return ret return f -if hasattr(reactor, "runUntilCurrent"): +if hasattr(reactor, "runUntilCurrent") and hasattr(reactor, "_newTimedCalls"): # runUntilCurrent is called when we have pending calls. It is called once # per iteratation after fd polling. reactor.runUntilCurrent = runUntilCurrentTimer(reactor.runUntilCurrent) From d3da63f7667f93e30f6071c1f5a0a0ffb9419e79 Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 18 Aug 2015 11:47:00 +0100 Subject: [PATCH 2/3] Use more helpful variable names --- synapse/metrics/__init__.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/synapse/metrics/__init__.py b/synapse/metrics/__init__.py index 7b8a20108b..2e307a03ad 100644 --- a/synapse/metrics/__init__.py +++ b/synapse/metrics/__init__.py @@ -163,11 +163,11 @@ def runUntilCurrentTimer(func): # _newTimedCalls is one long list of *all* pending calls. Below loop # is based off of impl of reactor.runUntilCurrent - for p in reactor._newTimedCalls: - if p.time > now: + for delayed_call in reactor._newTimedCalls: + if delayed_call.time > now: break - if p.delayed_time > 0: + if delayed_call.delayed_time > 0: continue num_pending += 1 From 6e7d36a72c5778be6d1a79d296b1fcdce667839d Mon Sep 17 00:00:00 2001 From: Erik Johnston Date: Tue, 18 Aug 2015 11:51:08 +0100 Subject: [PATCH 3/3] Also check for presence of 'threadCallQueue' in reactor --- synapse/metrics/__init__.py | 9 ++++++++- 1 file changed, 8 insertions(+), 1 deletion(-) diff --git a/synapse/metrics/__init__.py b/synapse/metrics/__init__.py index 2e307a03ad..d7bcad8a8a 100644 --- a/synapse/metrics/__init__.py +++ b/synapse/metrics/__init__.py @@ -184,7 +184,14 @@ def runUntilCurrentTimer(func): return f -if hasattr(reactor, "runUntilCurrent") and hasattr(reactor, "_newTimedCalls"): +try: + # Ensure the reactor has all the attributes we expect + reactor.runUntilCurrent + reactor._newTimedCalls + reactor.threadCallQueue + # runUntilCurrent is called when we have pending calls. It is called once # per iteratation after fd polling. reactor.runUntilCurrent = runUntilCurrentTimer(reactor.runUntilCurrent) +except AttributeError: + pass