Fix possible race condition in topic common (#769)

Author: vgvoleg

examples/coordination/example.py

@@ -43,7 +43,7 @@ def run(endpoint, database):
         threads = []
 
         for i in range(4):
-            worker_name = f"worker {i+1}"
+            worker_name = f"worker {i + 1}"
             if i < 2:
                 thread = threading.Thread(target=linear_workload, args=(driver.coordination_client, worker_name))
             else:

examples/topic/writer_example.py

@@ -1,7 +1,7 @@
 import concurrent.futures
 import datetime
 from typing import Dict, List
-from concurrent.futures import Future, wait
+from concurrent.futures import Future, wait  # noqa: F401
 
 import ydb
 from ydb import TopicWriterMessage

test-requirements.txt

@@ -38,7 +38,7 @@ zipp==3.19.1
 aiohttp>=3.9.0
 pytest-pep8
 pytest-flake8
-flake8==3.9.2
+flake8==6.1.0
 sqlalchemy==1.4.26
 pylint-protobuf
 cython

tests/iam/test_auth.py

@@ -14,7 +14,7 @@ def __init__(self, account_id, key_id, private_key, iam_endpoint=None, iam_chann
         self.iam_channel_credentials = iam_channel_credentials
 
     def __eq__(self, other):
-        return self.__dict__ == other.__dict__ if type(self) == type(other) else False
+        return self.__dict__ == other.__dict__ if type(self) is type(other) else False
 
 
 @patch("builtins.open", new_callable=mock_open, read_data=CONTENT1)

tests/scheme/scheme_test.py

@@ -1,4 +1,4 @@
-import typing
+import typing  # noqa: F401
 import pytest
 
 import ydb

tests/test_errors.py

@@ -11,7 +11,7 @@ def test_scheme_error(driver_sync, database):
 
     server_code = ydb.issues.StatusCode.SCHEME_ERROR
 
-    assert type(exc.value) == ydb.issues.SchemeError
+    assert type(exc.value) is ydb.issues.SchemeError
     assert exc.value.status == server_code
     assert f"server_code: {server_code}" in str(exc.value)
     assert "Path does not exist" in str(exc.value)

tests/topics/test_topic_writer.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 import asyncio
-from typing import List
+from typing import List  # noqa: F401
 
 import pytest
 

ydb/_topic_common/common.py

@@ -36,21 +36,26 @@ def wrapper(rpc_state, response_pb, driver=None):
 
 
 def _get_shared_event_loop() -> asyncio.AbstractEventLoop:
-    global _shared_event_loop
-
     if _shared_event_loop is not None:
         return _shared_event_loop
 
     with _shared_event_loop_lock:
         if _shared_event_loop is not None:
             return _shared_event_loop
 
-        event_loop_set_done: concurrent.futures.Future[asyncio.AbstractEventLoop] = concurrent.futures.Future()
+        loop_ready: threading.Event = threading.Event()
 
         def start_event_loop():
             event_loop = asyncio.new_event_loop()
-            event_loop_set_done.set_result(event_loop)
             asyncio.set_event_loop(event_loop)
+
+            def on_loop_started():
+                # Set global only when loop is actually running
+                global _shared_event_loop
+                _shared_event_loop = event_loop
+                loop_ready.set()
+
+            event_loop.call_soon(on_loop_started)
             event_loop.run_forever()
 
         t = threading.Thread(
@@ -60,7 +65,11 @@ def start_event_loop():
         )
         t.start()
 
-        _shared_event_loop = event_loop_set_done.result()
+        loop_ready.wait()
+
+        if _shared_event_loop is None:
+            raise RuntimeError("Event loop was not properly initialized")
+
         return _shared_event_loop
 
 

ydb/retries.py

@@ -77,7 +77,7 @@ def __init__(self, timeout: float) -> None:
 
     def __eq__(self, other: object) -> bool:
         return (
-            type(self) == type(other) and isinstance(other, YdbRetryOperationSleepOpt) and self.timeout == other.timeout
+            type(self) is type(other) and isinstance(other, YdbRetryOperationSleepOpt) and self.timeout == other.timeout
         )
 
     def __repr__(self) -> str:
@@ -91,7 +91,7 @@ def __init__(self, result: Any) -> None:
 
     def __eq__(self, other: object) -> bool:
         return (
-            type(self) == type(other)
+            type(self) is type(other)
             and isinstance(other, YdbRetryOperationFinalResult)
             and self.result == other.result
             and self.exc == other.exc

ydb/table_test.py

@@ -14,7 +14,7 @@ def test_retry_operation_impl(monkeypatch):
     monkeypatch.setattr(
         issues.Error,
         "__eq__",
-        lambda self, other: type(self) == type(other) and self.message == other.message,
+        lambda self, other: type(self) is type(other) and self.message == other.message,
     )
 
     retry_once_settings = RetrySettings(
@@ -43,7 +43,7 @@ def __init__(self, message):
             self.message = message
 
         def __eq__(self, other):
-            return type(self) == type(other) and self.message == other.message
+            return type(self) is type(other) and self.message == other.message
 
     def check_unretriable_error(err_type, call_ydb_handler):
         retry_once_settings.on_ydb_error_callback.reset_mock()