dli

GenServer + DynamicSupervisor to track streaming workers per topic?

Hey there,

I am building a system that:

streams data from an external HTTP API endpoint for multiple topics
starts streaming processes under a supervisor
allows runtime introspection which topics are currently streaming (optional: + which ones errored)

The external endpoint has a param for passing multiple topics (topic count per request is limited though). To stream additional topics, I need to make a new request.

Something like StreamerManager.start(topics) would launch one or more streamer processes for topics that are not streaming yet. Topics that are already streamed by some process are skipped. FWIW, I already implemented the Streamer + Consumers using GenStage but this is likely irrelevant to my question.

I am considering a GenServer that tracks %{:topic_name => pid()} in its state and calls DynamicSupervisor to launch streamers as child processes with the appropriate args. However, the PIDs will become invalid when a streamer crashes and is restarted by the supervisor (not sure if they even matter if I rely on a supervisor).

Am I off the track here or what would be the appropriate architecture here?

7 comments

#supervisor #genserver

0 533 7

2024-01-15 15:00:53 UTC

First 7 of 7 Posts

shanesveller

You sound like you may be imagining something very similar to Registry and via-tuples, will that save you some wheel reinvention? It automatically deregisters processes that terminate.

It is single host only, not distributed, but scales very well IME especially if you’re on a new enough Elixir version to activate the partitioning support.

Post #1

dli

Registry looks almost perfect for the job. Can’t believe I didn’t see this before.

With via, I cannot register a process with multiple keys. I could use the topic list as a key, but then it would be hard to look up a single topic’s process without iterating through all keys.

Would it make sense to do something like this? Anything that speaks against registering inside an init callback? Having slight doubts about the coupling between Streamer and TopicRegistry, maybe not much of an issue though.

defmodule Streamer do
  use GenStage

  def init(topics) do
    for topic <- topics do
      # value does not matter at the moment
      Registry.register(MyApp.TopicRegistry, topic, nil)
    end

    # ...
    {:producer, ...}
  end
end

Post #2

shanesveller

I don’t see any concerns with this design off the top of my head, though in a standard GenServer I would possibly move the logic into a handle_continue instead of directly in init.

Post #3

Tyson

My understanding is that any work done in init will block the Supervisor, so better to offload expensive setup to handle_continue (which, unlike some earlier workarounds to that concern, is guaranteed to be the first message in the process mailbox).

Post #4

dli

offload expensive setup to handle_continue

This is not possible with GenStage

Edit: There is an open PR to GenStage that implements :continue and handle_continue: Add support for `handle_continue/2` callback by maartenvanvliet · Pull Request #300 · elixir-lang/gen_stage · GitHub

Post #5

Tyson

Interesting. I haven’t had an opportunity to use GenStage yet. Then I guess init will have to do!

[edit: Yay for the handle_continue PR, but looks like it’s been sitting there a while. And another PR before that for even longer!]

Post #6

dli

Here is what I am using now (thanks @shanesveller for the inspiration):

Start Registry in supervisor tree: {Registry, keys: :unique, name: MyApp.TopicRegistry}
When starting a :producer / “TopicStreamer” GenStage with topics:
- fetch all keys from Registry:
```
Registry.select(MyApp.TopicRegistry, [{{:"$1", :_, :_}, [], [:"$1"]}])
|> List.flatten()
|> MapSet.new()
```
- Enum.reject all topics that exist in set of keys
- MyApp.TopicStreamer.start_link(name: {:via, TopicRegistry, topics})