Add event broadcasting capability #672
Conversation
ankona
changed the title
XxxCommChannel
Codecov ReportAttention: Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## mli-feature #672 +/- ##
==============================================
Coverage ? 69.84%
==============================================
Files ? 103
Lines ? 8708
Branches ? 0
==============================================
Hits ? 6082
Misses ? 2626
Partials ? 0
|
| # todo: consider that this could (under load) never exit. do we need | ||
| # to configure a maximum number to pull at once? |
There was a problem hiding this comment.
Hmmmm didn't think about this. We'd want all of the messages within the channel though, right?
There was a problem hiding this comment.
i think it's possible with a while to never run out of new messages to receive. it doesn't even need to be a ton of buffered messages, just constant.
i think a for i in range(max_message_retrievals) may ensure that it doesn't get stuck and any loop triggering retrieval would end up getting called again, anyway...
There was a problem hiding this comment.
I wonder if this should be the responsibility of the caller and not DragonCommChannel. It seems like recv might be less prone to edge case problems like the one described if it just simply returned the next message on the channel or None if the channel is empty. It is then up to the caller to takes as many messages as it should based on the context of what it is doing. Is this a good idea / bad idea / too big of a change / nonsense? Another approach maybe is that the timeout is not reset but it is decremented each iteration so it is applied across all messages? Something strikes me as too many responsibilities or unintuitive about the way this is designed now.
| # The tests in this file belong to the group_a group | ||
| pytestmark = pytest.mark.group_a |
There was a problem hiding this comment.
I think these tests are getting skipped in group a
There was a problem hiding this comment.
it's ok for them to get skipped if dragon isn't installed). that will be done on purpose with dragon = pytest.importorskip("dragon")
However, if the dragon library is there, we can run all of these tests without needing a "real dragon" environment (only need the ability to import the dragon components/modules successfully). For example, these will run on hotlum w/a normal pytest like pytest ./tests/test_featurestore.py
There was a problem hiding this comment.
Okay so we're cool with these tests being skipped in github actions then. Just wanted to double check.
There was a problem hiding this comment.
I think you've still got a valid point... I didn't achieve what I thought I did, after reviewing the gh action. Investigating...
smartsim/_core/mli/infrastructure/storage/backbonefeaturestore.py
Outdated
Show resolved
Hide resolved
| :param channel: a channel to use for communications | ||
| :param recv_timeout: a default timeout to apply to receive calls""" | ||
| serialized_ch = channel.serialize() | ||
| safe_descriptor = base64.b64encode(serialized_ch).decode("utf-8") |
There was a problem hiding this comment.
What do you mean by a "safe" descriptor? Have we run into a bug/issue?
| # todo: consider that this could (under load) never exit. do we need | ||
| # to configure a maximum number to pull at once? |
There was a problem hiding this comment.
I wonder if this should be the responsibility of the caller and not DragonCommChannel. It seems like recv might be less prone to edge case problems like the one described if it just simply returned the next message on the channel or None if the channel is empty. It is then up to the caller to takes as many messages as it should based on the context of what it is doing. Is this a good idea / bad idea / too big of a change / nonsense? Another approach maybe is that the timeout is not reset but it is decremented each iteration so it is applied across all messages? Something strikes me as too many responsibilities or unintuitive about the way this is designed now.
| return f"{self.uid}|{self.category}" | ||
|
|
||
|
|
||
| class OnCreateConsumer(EventBase): |
There was a problem hiding this comment.
Consumer is used throughout this class (and the EventCategory). Would Subscriber be more accurate?
|
|
||
| CONSUMER_CREATED: str = "consumer-created" | ||
| FEATURE_STORE_WRITTEN: str = "feature-store-written" | ||
| UNKNOWN: str = "unknown" |
There was a problem hiding this comment.
Is UNKNOWN used by other packages? I was wondering if DEFAULT is better
| :raises SmartSimError: if any unexpected error occurs during send""" | ||
| try: | ||
| self._save_to_buffer(event) | ||
|
|
| try: | ||
| comm_channel.send(next_event) | ||
| num_sent += 1 | ||
| except Exception as ex: |
There was a problem hiding this comment.
If one channel broadcast fails, then an error is raised. This means that other channels that may be OK will never get the message because it was already popped and not placed back on the deque. I wonder if we should stored failed sends and still go through other channels? We would still need to think about if the failed broadcasts should be retired or just thrown away...
| :raises SmartSimError: if any unexpected error occurs during send""" | ||
| try: | ||
| self._save_to_buffer(event) | ||
|
|
No description provided.