-
Notifications
You must be signed in to change notification settings - Fork 5.5k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
4 changed files
with
236 additions
and
24 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
115 changes: 115 additions & 0 deletions
115
tests/pytests/functional/transport/ipc/test_subscriber.py
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,115 @@ | ||
| import sys | ||
|
|
||
| import attr | ||
| import pytest | ||
| import salt.ext.tornado.gen | ||
| import salt.transport.client | ||
| import salt.transport.ipc | ||
| import salt.transport.server | ||
| from salt.ext.tornado import locks | ||
|
|
||
| pytestmark = [ | ||
| # Windows does not support POSIX IPC | ||
| pytest.mark.skip_on_windows, | ||
| pytest.mark.skipif( | ||
| sys.version_info < (3, 6), reason="The IOLoop blocks under Py3.5 on these tests" | ||
| ), | ||
| ] | ||
|
|
||
|
|
||
| @attr.s(frozen=True, slots=True) | ||
| class PayloadHandler: | ||
| payloads = attr.ib(init=False, default=attr.Factory(list)) | ||
|
|
||
| async def handle_payload(self, payload, reply_func): | ||
| self.payloads.append(payload) | ||
| await reply_func(payload) | ||
|
|
||
| def __enter__(self): | ||
| return self | ||
|
|
||
| def __exit__(self, *args): | ||
| self.payloads.clear() | ||
|
|
||
|
|
||
| @attr.s(frozen=True, slots=True) | ||
| class IPCTester: | ||
| io_loop = attr.ib() | ||
| socket_path = attr.ib() | ||
| publisher = attr.ib() | ||
| subscriber = attr.ib() | ||
| payloads = attr.ib(default=attr.Factory(list)) | ||
| payload_ack = attr.ib(default=attr.Factory(locks.Condition)) | ||
|
|
||
| @subscriber.default | ||
| def _subscriber_default(self): | ||
| return salt.transport.ipc.IPCMessageSubscriber( | ||
| self.socket_path, io_loop=self.io_loop, | ||
| ) | ||
|
|
||
| @publisher.default | ||
| def _publisher_default(self): | ||
| return salt.transport.ipc.IPCMessagePublisher( | ||
| {"ipc_write_buffer": 0}, self.socket_path, io_loop=self.io_loop, | ||
| ) | ||
|
|
||
| async def handle_payload(self, payload, reply_func): | ||
| self.payloads.append(payload) | ||
| await reply_func(payload) | ||
| self.payload_ack.notify() | ||
|
|
||
| def new_client(self): | ||
| return IPCTester( | ||
| io_loop=self.io_loop, | ||
| socket_path=self.socket_path, | ||
| server=self.server, | ||
| payloads=self.payloads, | ||
| payload_ack=self.payload_ack, | ||
| ) | ||
|
|
||
| async def publish(self, payload, timeout=60): | ||
| self.publisher.publish(payload) | ||
|
|
||
| async def read(self, timeout=60): | ||
| ret = await self.subscriber.read(timeout) | ||
| return ret | ||
|
|
||
| def __enter__(self): | ||
| self.publisher.start() | ||
| self.io_loop.add_callback(self.subscriber.connect) | ||
| return self | ||
|
|
||
| def __exit__(self, *args): | ||
| self.subscriber.close() | ||
| self.publisher.close() | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def ipc_socket_path(tmp_path): | ||
| _socket_path = tmp_path / "ipc-test.ipc" | ||
| try: | ||
| yield _socket_path | ||
| finally: | ||
| if _socket_path.exists(): | ||
| _socket_path.unlink() | ||
|
|
||
|
|
||
| @pytest.fixture | ||
| def channel(io_loop, ipc_socket_path): | ||
| _ipc_tester = IPCTester(io_loop=io_loop, socket_path=str(ipc_socket_path)) | ||
| with _ipc_tester: | ||
| yield _ipc_tester | ||
|
|
||
|
|
||
| async def test_basic_send(channel): | ||
| msg = {"foo": "bar", "stop": True} | ||
| # XXX: IPCClient connect and connected methods need to be cleaned up as | ||
| # this should not be needed. | ||
| while not channel.subscriber._connecting_future.done(): | ||
| await salt.ext.tornado.gen.sleep(0.01) | ||
| while not channel.subscriber.connected(): | ||
| await salt.ext.tornado.gen.sleep(0.01) | ||
| assert channel.subscriber.connected() | ||
| await channel.publish(msg) | ||
| ret = await channel.read() | ||
| assert ret == msg |
4cf62fbThere was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@dwoz Does the only way to wean off IOLoop.run_sync() lead through breaking get_event_block()? _read() will actually return a future which then will break in self.unpack() where it tries to access raw.partition. get_event() uses _get_event() which uses read() instead of _read() and the former somehow makes it work. I'm not as proficient in the old-style yield-based async code so I can't fully explain why but IPCMessageSubscriber.read() yields the result of _read() which combined with the fact it is marked as a coroutine (even though it is called directly from non-async code) makes the result contain the actual result of the future. I hope it makes sense, anyway you rendered get_event_block() useless with this change, at least in 3003.4.
My colleague opened #62015 for this some time ago but it got little attention. There's some simple code there showing that it just breaks get_event_block(). That code used to work from 2016.* through 3000 with no problems.
I understand you wean off IOLoop.run_sync() which was used IPCMessageSubscriber.read_sync() and you wrote that new read() function which using the aforementioned "hocus pocus" makes coroutines called from synchronous code work but then you just can't start calling that _read() directly because that won't work. My gut feeling is that using that IPCMessageSubscriber.read_sync() to run the async coroutine _read() was the right way to do it but if it works the way that new read() func does it that's also fine, just as I said it does not work if _read() is called directly as it now is from get_event_block() and also get_event_noblock(), which is most likely broken too, we just don't use that one so we don't care xD
Obviously it doesn't seem to make sense you create that new read() function and then replace read_sync() for read() in _get_event() while replacing the same read_sync() for the existing function _read() in two other places that can't just possibly work neither in 3003 nor 3000. That read_sync() wrapped that _read() in IOLoop.run_sync() for a reason right? And you seem to have discerned this reason partly by writing that read() function as a replacement.
I don't think doing it the way you did will even be portable to new-style asyncio - you can't mix sync/async code easily there like that - I don't think it's possible to call a coroutine there just like a normal function and get the result, which ultimately is the case in your code (_get_event() calls your new read() function which is a coroutine). As I said I haven't fully figured out why wrapping that _read() coroutine in your read() coroutine makes it return the actual result instead of a Future object (which is what _read() returns and hence the bug) but that seems to be the case. I tried to do that in native Python3 async but that won't work - you'll always get a coroutine object plus a warning eventually that it was never awaited. It normally takes asyncio.run() to run a coroutine which is some equivalent of the IOLoop.run_sync() of which the proces of weaning off this PR started.
@Ch3LL are you also happy with this PR introducing bugs like this?