mirror of
https://github.com/element-hq/synapse.git
synced 2026-07-12 16:10:15 +00:00
Eric Eastwood
7da21a715d
This means you can use `get_success(...)` anywhere regardless of what kind of work needs to be done. Spawning from adding some more async Rust things in https://github.com/element-hq/synapse/pull/19846 and wanting something more standard instead of the custom `till_deferred_has_result(...)` that has crept in to a few files. Alternative to https://github.com/element-hq/synapse/pull/19867 spurred on by [this comment](https://github.com/element-hq/synapse/pull/19867#discussion_r3441774685) from @erikjohnston ### How does this work? Previously, `get_success(...)` just ran in a hot-loop advancing the Twisted reactor clock which didn't give any time for other threads to do some work or acquire the GIL if necessary (whenever there is a hand-off from Rust to Python, we need the GIL). Now, `get_success(...)` loops until we see a result (until we hit the ~0.1s real-time timeout). In the loop, we call [`time.sleep(0)`](https://docs.python.org/3/library/time.html#time.sleep) which will "Suspend execution of the calling thread [...]" (CPU and GIL) to allow other threads to do some work. Then like before, we advance the Twisted reactor clock to run any scheduled callbacks which includes anything the other threads may have scheduled. ### Does this slow down the entire test suite? Seems just as fast as before. There is minutes variance in what we had before and after but both are within the same range of each other. (see PR for actual before/after timings)
224 lines
8.2 KiB
Python
224 lines
8.2 KiB
Python
#
|
|
# This file is licensed under the Affero General Public License (AGPL) version 3.
|
|
#
|
|
# Copyright 2022 The Matrix.org Foundation C.I.C.
|
|
# Copyright (C) 2023 New Vector, Ltd
|
|
#
|
|
# This program is free software: you can redistribute it and/or modify
|
|
# it under the terms of the GNU Affero General Public License as
|
|
# published by the Free Software Foundation, either version 3 of the
|
|
# License, or (at your option) any later version.
|
|
#
|
|
# See the GNU Affero General Public License for more details:
|
|
# <https://www.gnu.org/licenses/agpl-3.0.html>.
|
|
#
|
|
# Originally licensed under the Apache License, Version 2.0:
|
|
# <http://www.apache.org/licenses/LICENSE-2.0>.
|
|
#
|
|
# [This file includes modifications made by New Vector Limited]
|
|
#
|
|
#
|
|
|
|
from twisted.internet import defer
|
|
|
|
from synapse.replication.tcp.commands import PositionCommand
|
|
|
|
from tests.replication._base import BaseMultiWorkerStreamTestCase
|
|
|
|
|
|
class ChannelsTestCase(BaseMultiWorkerStreamTestCase):
|
|
def test_subscribed_to_enough_redis_channels(self) -> None:
|
|
# The default main process is subscribed to the USER_IP channel.
|
|
self.assertCountEqual(
|
|
self.hs.get_replication_command_handler()._channels_to_subscribe_to,
|
|
["USER_IP"],
|
|
)
|
|
|
|
def test_background_worker_subscribed_to_user_ip(self) -> None:
|
|
# The default main process is subscribed to the USER_IP channel.
|
|
worker1 = self.make_worker_hs(
|
|
"synapse.app.generic_worker",
|
|
extra_config={
|
|
"worker_name": "worker1",
|
|
"run_background_tasks_on": "worker1",
|
|
"redis": {"enabled": True},
|
|
},
|
|
)
|
|
self.assertIn(
|
|
"USER_IP",
|
|
worker1.get_replication_command_handler()._channels_to_subscribe_to,
|
|
)
|
|
|
|
# Advance so the Redis subscription gets processed
|
|
self.pump(0.1)
|
|
|
|
# The counts are 2 because both the main process and the worker are subscribed.
|
|
self.assertEqual(len(self._redis_server._subscribers_by_channel[b"test"]), 2)
|
|
self.assertEqual(
|
|
len(self._redis_server._subscribers_by_channel[b"test/USER_IP"]), 2
|
|
)
|
|
|
|
def test_non_background_worker_not_subscribed_to_user_ip(self) -> None:
|
|
# The default main process is subscribed to the USER_IP channel.
|
|
worker2 = self.make_worker_hs(
|
|
"synapse.app.generic_worker",
|
|
extra_config={
|
|
"worker_name": "worker2",
|
|
"run_background_tasks_on": "worker1",
|
|
"redis": {"enabled": True},
|
|
},
|
|
)
|
|
self.assertNotIn(
|
|
"USER_IP",
|
|
worker2.get_replication_command_handler()._channels_to_subscribe_to,
|
|
)
|
|
|
|
# Advance so the Redis subscription gets processed
|
|
self.pump(0.1)
|
|
|
|
# The count is 2 because both the main process and the worker are subscribed.
|
|
self.assertEqual(len(self._redis_server._subscribers_by_channel[b"test"]), 2)
|
|
# For USER_IP, the count is 1 because only the main process is subscribed.
|
|
self.assertEqual(
|
|
len(self._redis_server._subscribers_by_channel[b"test/USER_IP"]), 1
|
|
)
|
|
|
|
def test_wait_for_stream_position(self) -> None:
|
|
"""Check that wait for stream position correctly waits for an update from the
|
|
correct instance.
|
|
"""
|
|
store = self.hs.get_datastores().main
|
|
cmd_handler = self.hs.get_replication_command_handler()
|
|
data_handler = self.hs.get_replication_data_handler()
|
|
|
|
worker1 = self.make_worker_hs(
|
|
"synapse.app.generic_worker",
|
|
extra_config={
|
|
"worker_name": "worker1",
|
|
"run_background_tasks_on": "worker1",
|
|
"redis": {"enabled": True},
|
|
},
|
|
)
|
|
|
|
cache_id_gen = worker1.get_datastores().main._cache_id_gen
|
|
assert cache_id_gen is not None
|
|
|
|
self.replicate()
|
|
|
|
# First, make sure the master knows that `worker1` exists.
|
|
initial_token = cache_id_gen.get_current_token()
|
|
cmd_handler.send_command(
|
|
PositionCommand("caches", "worker1", initial_token, initial_token)
|
|
)
|
|
self.replicate()
|
|
|
|
# Next send out a normal RDATA, and check that waiting for that stream
|
|
# ID returns immediately.
|
|
ctx = cache_id_gen.get_next()
|
|
next_token = self.get_success(ctx.__aenter__())
|
|
self.get_success(ctx.__aexit__(None, None, None))
|
|
|
|
self.get_success(
|
|
data_handler.wait_for_stream_position("worker1", "caches", next_token)
|
|
)
|
|
|
|
# `wait_for_stream_position` should only return once master receives a
|
|
# notification that `next_token` has persisted.
|
|
ctx_worker1 = cache_id_gen.get_next()
|
|
next_token = self.get_success(ctx_worker1.__aenter__())
|
|
|
|
d = defer.ensureDeferred(
|
|
data_handler.wait_for_stream_position("worker1", "caches", next_token)
|
|
)
|
|
self.assertFalse(d.called)
|
|
|
|
# ... updating the cache ID gen on the master still shouldn't cause the
|
|
# deferred to wake up.
|
|
assert store._cache_id_gen is not None
|
|
ctx = store._cache_id_gen.get_next()
|
|
self.get_success(ctx.__aenter__())
|
|
self.get_success(ctx.__aexit__(None, None, None))
|
|
|
|
d = defer.ensureDeferred(
|
|
data_handler.wait_for_stream_position("worker1", "caches", next_token)
|
|
)
|
|
self.assertFalse(d.called)
|
|
|
|
# ... but worker1 finishing (and so sending an update) should.
|
|
self.get_success(ctx_worker1.__aexit__(None, None, None))
|
|
|
|
# Wait for the stream position to be replicated to the master process
|
|
#
|
|
# Replication travels over `FakeTransport` and we're specifically flushing the
|
|
# write
|
|
self.reactor.advance(0)
|
|
|
|
self.assertTrue(d.called)
|
|
|
|
def test_wait_for_stream_position_rdata(self) -> None:
|
|
"""Check that wait for stream position correctly waits for an update
|
|
from the correct instance, when RDATA is sent.
|
|
"""
|
|
store = self.hs.get_datastores().main
|
|
cmd_handler = self.hs.get_replication_command_handler()
|
|
data_handler = self.hs.get_replication_data_handler()
|
|
|
|
worker1 = self.make_worker_hs(
|
|
"synapse.app.generic_worker",
|
|
extra_config={
|
|
"worker_name": "worker1",
|
|
"run_background_tasks_on": "worker1",
|
|
"redis": {"enabled": True},
|
|
},
|
|
)
|
|
|
|
cache_id_gen = worker1.get_datastores().main._cache_id_gen
|
|
assert cache_id_gen is not None
|
|
|
|
self.replicate()
|
|
|
|
# First, make sure the master knows that `worker1` exists.
|
|
initial_token = cache_id_gen.get_current_token()
|
|
cmd_handler.send_command(
|
|
PositionCommand("caches", "worker1", initial_token, initial_token)
|
|
)
|
|
self.replicate()
|
|
|
|
# `wait_for_stream_position` should only return once master receives a
|
|
# notification that `next_token2` has persisted.
|
|
ctx_worker1 = cache_id_gen.get_next_mult(2)
|
|
next_token1, next_token2 = self.get_success(ctx_worker1.__aenter__())
|
|
|
|
d = defer.ensureDeferred(
|
|
data_handler.wait_for_stream_position("worker1", "caches", next_token2)
|
|
)
|
|
self.assertFalse(d.called)
|
|
|
|
# Insert an entry into the cache stream with token `next_token1`, but
|
|
# not `next_token2`.
|
|
self.get_success(
|
|
store.db_pool.simple_insert(
|
|
table="cache_invalidation_stream_by_instance",
|
|
values={
|
|
"stream_id": next_token1,
|
|
"instance_name": "worker1",
|
|
"cache_func": "foo",
|
|
"keys": [],
|
|
"invalidation_ts": 0,
|
|
},
|
|
)
|
|
)
|
|
|
|
# Finish the context manager, triggering the data to be sent to master.
|
|
self.get_success(ctx_worker1.__aexit__(None, None, None))
|
|
|
|
# Wait for the stream position to be replicated to the master process
|
|
#
|
|
# Replication travels over `FakeTransport` and we're specifically flushing the
|
|
# write
|
|
self.reactor.advance(0)
|
|
|
|
# Master should get told about `next_token2`, so the deferred should
|
|
# resolve.
|
|
self.assertTrue(d.called)
|