Repository navigation
Duplicate requests to the resampling actor cause it to go out of sync #115
Description
What happened?
When the same request is sent to the resampling actor multiple times, the output streams start to go out of sync, and such metrics start receiving data at higher rates.
For example, after starting the actors, if we run the below code, we'll see that the output streams are out of sync, and soon we'll start receiving warning messages saying one of the channels is overflowing.
## Create two requests for the resampling actor.
req_2002 = ComponentMetricRequest(
namespace="hell0",
component_id=2002,
metric_id=ComponentMetricId.ACTIVE_POWER,
start_time=None,
)
req_1001 = ComponentMetricRequest(
namespace="hell0",
component_id=1001,
metric_id=ComponentMetricId.ACTIVE_POWER,
start_time=None,
)
# Send one of the request two times to the resampling actor.
await resampling_actor_request_sender.send(req_2002)
await resampling_actor_request_sender.send(req_1001)
await resampling_actor_request_sender.send(req_1001)
recv_1001 = channel_registry.new_receiver(req_1001.get_channel_name())
recv_2002 = channel_registry.new_receiver(req_2002.get_channel_name())
while True:
print()
print(f"1001: {(await recv_1001.receive()).timestamp}")
print(f"2002: {(await recv_2002.receive()).timestamp}")will produce output:
1001: 2022-12-13 14:33:45.467626+00:00
2002: 2022-12-13 14:33:45.467626+00:00
1001: 2022-12-13 14:33:45.467626+00:00
2002: 2022-12-13 14:33:46.468322+00:00
1001: 2022-12-13 14:33:46.468322+00:00
2002: 2022-12-13 14:33:47.470336+00:00
1001: 2022-12-13 14:33:46.468322+00:00
2002: 2022-12-13 14:33:48.471591+00:00
...
1001: 2022-12-13 14:34:09.494318+00:00
2002: 2022-12-13 14:34:33.522342+00:00
1001: 2022-12-13 14:34:09.494318+00:00
2002: 2022-12-13 14:34:34.523423+00:00
1001: 2022-12-13 14:34:10.495892+00:00
Broadcast receiver [Microgrid Channel Registry-1001::ACTIVE_POWER::None::hell0:52ba648d-9d57-4c49-b899-78bf383bdd3c] is full. Oldest message was dropped.
2002: 2022-12-13 14:34:35.524349+00:00
1001: 2022-12-13 14:34:11.496506+00:00
Broadcast receiver [Microgrid Channel Registry-1001::ACTIVE_POWER::None::hell0:52ba648d-9d57-4c49-b899-78bf383bdd3c] is full. Oldest message was dropped.
2002: 2022-12-13 14:34:36.525981+00:00
What did you expect instead?
Values shouldn't diverge if the actor receives the same request multiple times.
Affected version(s)
v0.14.0
Affected part(s)
Data pipeline (part:data-pipeline)
Extra information
A minimal example to reproduce the issue
import asyncio
from frequenz.channels import Broadcast
from frequenz.sdk import microgrid
from frequenz.sdk.actor import (
ChannelRegistry,
ComponentMetricRequest,
ComponentMetricsResamplingActor,
DataSourcingActor,
)
from frequenz.sdk.microgrid.component import ComponentCategory, ComponentMetricId
async def test_resampler() -> None:
"""Test resampler."""
await microgrid.initialize(HOST, PORT)
channel_registry = ChannelRegistry(name="Microgrid Channel Registry")
data_source_request_channel = Broadcast[ComponentMetricRequest](
"Data Source Request Channel"
)
data_source_request_sender = data_source_request_channel.new_sender()
data_source_request_receiver = data_source_request_channel.new_receiver()
resampling_actor_request_channel = Broadcast[ComponentMetricRequest](
"Resampling Actor Request Channel"
)
resampling_actor_request_sender = resampling_actor_request_channel.new_sender()
resampling_actor_request_receiver = resampling_actor_request_channel.new_receiver()
DataSourcingActor(
request_receiver=data_source_request_receiver, registry=channel_registry
)
ComponentMetricsResamplingActor(
channel_registry=channel_registry,
data_sourcing_request_sender=data_source_request_sender,
resampling_request_receiver=resampling_actor_request_receiver,
resampling_period_s=1.0,
)
## Create two requests for the resampling actor.
req_2002 = ComponentMetricRequest(
namespace="hell0",
component_id=2002,
metric_id=ComponentMetricId.ACTIVE_POWER,
start_time=None,
)
await resampling_actor_request_sender.send(req_2002)
req_1001 = ComponentMetricRequest(
namespace="hell0",
component_id=1001,
metric_id=ComponentMetricId.ACTIVE_POWER,
start_time=None,
)
# Send one of the request two times to the resampling actor.
await resampling_actor_request_sender.send(req_1001)
await resampling_actor_request_sender.send(req_1001)
recv_1001 = channel_registry.new_receiver(req_1001.get_channel_name())
recv_2002 = channel_registry.new_receiver(req_2002.get_channel_name())
while True:
print()
print(f"1001: {(await recv_1001.receive()).timestamp}")
print(f"2002: {(await recv_2002.receive()).timestamp}")
asyncio.run(test_resampler())Metadata
Metadata
Assignees
Type
Fields
Priority
None yet
Projects
- Status
Activity
leandro-lucarella-frequenz commentedon Dec 19, 2022
I'm not sure why the timer gets messed up, but there is something definitely wrong with how sources are handled, if the same request is received twice, we set up a new resampler unconditionally so we'll end up resampling the same source twice and putting both into the same output channel too.
Just keeping track of which channel names we are already subscribed to, and ignoring active subscriptions if a new subscription for the same stuff comes in fixes the issue, but I would know understand why the timer gets messed up. Will probably send a PR with the fix ASAP though.
leandro-lucarella-frequenz commentedon Dec 19, 2022
OK, the timer doesn't get messed up, we are sending the same resampled data for the same component twice, to the same channel. So when the test consumes from the channel, it gets first the first copy and in the next loop the second copy, so we basically have the duplicated component falling behind because it has duplicated data in the channel. I didn't notice the 1001 timestamps were duplicated, I just thought 2002 was getting ahead :)
resampling: Don't subcribe twice to the same source
resampling: Don't subcribe twice to the same source
resampling: Don't subscribe twice to the same source (#123)
resampling: Don't subcribe twice to the same source
resampling: Don't subcribe twice to the same source
resampling: Don't subcribe twice to the same source