LockStore: fix acquiring a lock via LockStore.try_acquire_lock

[sumnerevans]

This PR fixes acquiring a lock via LockStore.try_acquire_lock which is currently broken.

• Adds a test that demonstrates the issue. See MSC2246 async uploads #12484 (comment)
• Adds a Linearizer to LockStore to make the test pass.

Outstanding questions:

• This change causes a regression in LockTestCase.test_drop and I'm not quite sure what to do about it. It's not obvious where the error is coming from (seems to be in the internals of twisted which seems concerning):

old error

$ trial tests.storage.databases.main.test_lock
tests.storage.databases.main.test_lock
  LockTestCase
    test_acquire_contention ...                                            [OK]
    test_drop ...                                                       [ERROR]
    test_maintain_lock ...                                                 [OK]
    test_shutdown ...                                                      [OK]
    test_simple_lock ...                                                   [OK]
    test_timeout_lock ...                                                  [OK]

===============================================================================
[ERROR]
Traceback (most recent call last):
  File "/home/sumner/projects/beeper/.venv/lib/python3.9/site-packages/twisted/internet/defer.py", line 1660, in _inlineCallbacks
    result = current_context.run(gen.send, result)
builtins.RuntimeError: cannot reuse already awaited coroutine

tests.storage.databases.main.test_lock.LockTestCase.test_drop
-------------------------------------------------------------------------------
Ran 6 tests in 5.219s

FAILED (errors=1, successes=5)

Note to Reviewer

If you checkout the branch at the commit where I add the initial test case, and run the test, it will fail.

$ trial tests.storage.databases.main.test_lock 

After fast-forwarding to the HEAD of the branch, running the same tests will pass.

Sign-off

Signed-off-by: Sumner Evans <sumner@beeper.com>

Pull Request Checklist

• Pull request is based on the develop branch
• Pull request includes a changelog file. The entry should:
  • Be a short description of your change which makes sense to users. "Fixed a bug that prevented receiving messages from other servers." instead of "Moved X method from EventStore to EventWorkerStore.".
  • Use markdown where necessary, mostly for code blocks.
  • End with either a period (.) or an exclamation mark (!).
  • Start with a capital letter.
  • Feel free to credit yourself, by adding a sentence "Contributed by @github_username." or "Contributed by [Your Name]." to the end of the entry.
• Pull request includes a sign off
• Code style is correct
  (run the linters)

[squahtx]

It turns out that the test I posted is poorly written: it doesn't run the three tasks to completion.

Previously all three tasks would acquire the worker lock and be waiting for a database transaction to release the lock at the end of the test case.

After the fix, at the end of test_acquire_contention:

• task 1 is waiting for a database transaction to release the worker lock(?)
• task 2 holds the Linearizer lock and is waiting for a database transaction. It hasn't actually acquired the worker lock yet.
• task 3 is waiting for the Linearizer lock and also hasn't acquired the worker lock.

Python's GC runs and throws a GeneratorExit exception inside each task. Task 3 gets the exception, resumes and finishes.
Task 2 gets the exception after and exits the Linearizer lock, which tries to resume task 3. But task 3 has already completed, so we get "RuntimeError: cannot reuse already awaited coroutine".

We can ensure all three tasks have finished like so:

--- a/tests/storage/databases/main/test_lock.py
+++ b/tests/storage/databases/main/test_lock.py
@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

-from twisted.internet import defer
+from twisted.internet import defer, reactor
+from twisted.internet.base import ReactorBase
 from twisted.internet.defer import Deferred

 from synapse.server import HomeServer
@@ -51,9 +52,9 @@ class LockTestCase(unittest.HomeserverTestCase):
                 in_lock -= 1

         # Start 3 tasks.
-        defer.ensureDeferred(task())
-        defer.ensureDeferred(task())
-        defer.ensureDeferred(task())
+        task1 = defer.ensureDeferred(task())
+        task2 = defer.ensureDeferred(task())
+        task3 = defer.ensureDeferred(task())

         # All three are now waiting for the database.
         # Give the reactor a kick so that the database transaction returns.
@@ -61,6 +62,18 @@ class LockTestCase(unittest.HomeserverTestCase):

         release_lock.callback(None)

+        # Run the tasks to completion.
+        # To work around `Linearizer`s using a different reactor to sleep when
+        # contended (#12841), we call `runUntilCurrent` on
+        # `twisted.internet.reactor`, which is a different reactor to that used
+        # by the homeserver.
+        assert isinstance(reactor, ReactorBase)
+        self.get_success(task1)
+        reactor.runUntilCurrent()
+        self.get_success(task2)
+        reactor.runUntilCurrent()
+        self.get_success(task3)
+
         # At most one task should have held the lock at a time.
         self.assertEqual(max_in_lock, 1)

but note that the "All three are now waiting for the database." comment is no longer accurate.

[squahtx]

nb: currently blocks #12484