CAIOHOBORGHI

CAIOHOBORGHI

Best Way to Broadcast data to all workers

Hello guys,

I`m starting to learn Elixir with the purpose to create a distributed application to do some validations in a list that is updated every min.

Here`s my ideia:

Create a DynamicSupervisor with N workers to isolate the validations(one process one validation of some item of the list)

The list contains about 1.000 items, where each item contains 10 properties.

A process need to validate some of the properties of one item. (eg: If property A from item N is bigger than 10, print it)

The problem is that this list is updated every min and I dont know which is the best way to emit the updates
to every worker. My first ideia is to use ETS but I`m a little concerned about every process doing a select every min in the database, doesn´t look like a good idea.

The complete flux is(I think it should be):
-> Supervisor gets the update from an external socket application
-> Filter the list according to each process item
-> Send to process only the properties of its item

But it needs to be fast, and scalable :smiley:

Marked As Solved

lud

lud

is “Not B” related to be, or does it mean anything besides “A” or “B”? Please explain your problem fully, if you add new concepts on each message we will not be able to help you properly :smiley:

Well, I though of sending it separated for performance improvement…

It may work, but spawning processes for each item will definitely not improve performances to just run “a quick validation” and a print on a list of 1000 items.

For now I would stick to the last proposal I made above, spawning a process for each item as required.

First, a producer that receive the whole list (as in the code I have posted before). You socket receives the whole list, and now you have that in memory. With any solution you will have to copy this data to other processes memory anyway, so you just send it to the producer and you can garbage collect on the socket. If you want to improve, you can stream events from the socket to the producer, but that is optimization, so for a future version, if ever needed.

So, you have this producer that receives the whole list. It uses the default dispatcher (not the broadcast or partition ones).

Then you add a ConsumerSupervisor that will receive the items according to its max_demand, and will spawn a process for each list item. In the handler function, you check the code and call the appropriate validator for this code.

defmodule SocketListUpdater do
  def start_link(producer \\ nil) do
    spawn_link(fn -> init(producer) end)
  end

  defp init(producer) do
    loop(producer)
  end

  defp loop(producer) do
    IO.puts("sending list")
    send(producer, {:new_list, generate_list()})
    # Send a list every minute, here every second
    Process.sleep(1000)
    loop(producer)
  end

  defp generate_list() do
    Stream.cycle(["A", "B", "C"])
    # The length of the list
    |> Enum.take(1000)
    |> Enum.shuffle()
    |> Enum.map(fn code -> %{code: code, infos: generate_infos()} end)
  end

  defp generate_infos do
    [
      %{name: "mary"},
      %{name: "john"},
      %{name: "robert"},
      %{name: "ava"},
      %{name: "simon"},
      %{name: "simon"}
    ]
    |> Enum.take(Enum.random(2..6))
  end
end

defmodule GenstageExample.Producer do
  use GenStage

  def start_link() do
    GenStage.start_link(__MODULE__, [], name: __MODULE__)
  end

  def init(_arg) do
    {:producer, :some_state}
  end

  def handle_info({:new_list, list}, state) do
    {:noreply, list, state}
  end

  def handle_demand(_demand, state) do
    {:noreply, [], state}
  end
end

defmodule GenstageExample.Consumer do
  use ConsumerSupervisor

  # Starts consumer with his code
  def start_link(arg) do
    ConsumerSupervisor.start_link(__MODULE__, arg)
  end

  def init(_arg) do
    # Note: By default the restart for a child is set to :permanent
    # which is not supported in ConsumerSupervisor. You need to explicitly
    # set the :restart option either to :temporary or :transient.
    children = [
      %{
        id: GenstageExample.Validator,
        start: {GenstageExample.Validator, :start_link, []},
        restart: :transient
      }
    ]

    opts = [strategy: :one_for_one, subscribe_to: [{GenstageExample.Producer, max_demand: 50}]]
    ConsumerSupervisor.init(children, opts)
  end
end

defmodule GenstageExample.Validator do
  def start_link(event) do
    Task.start_link(fn -> validate(event) end)
  end

  # Here you can call any validator function depending on the code. In this
  # example I implemented the validators directly in the function body

  defp validate(%{code: "A", infos: infos}) do
    if length(infos) < 3 do
      IO.puts("item with code A is not valid")
    end
  end

  defp validate(%{code: "B", infos: infos}) do
    if length(infos) == 2 do
      IO.puts("item with code B is not valid")
    end
  end

  defp validate(%{code: code}) do
    IO.puts("received unknown code #{code}")
  end
end

GenstageExample.Producer.start_link()
GenstageExample.Consumer.start_link([])
SocketListUpdater.start_link(GenstageExample.Producer)

So you do not have to start a validator for “B” or “Not B”, you just have a single validate/1 function that will check the code and do the appropriate validation. It is simpler.

Also Liked

jkmrto

jkmrto

hey!
Normally it is not good to do any task on the supervisor.
Instead a valid approach would be using GenStage.
So, with this approach you could think about having a producer, the process that will receive the messages, and many consumers to process messages.
In the Producer you just will need to receive the messages and broadcast them. The filtering part can be done almost automatically using the selector option for the subscription of the consumers.
Hope it helps!

lud

lud

I see,

In that case GenStage can do the job with a consumer per code.

But my code is far from good:

  • I did not use any buffering for the demands in the consumer, that should be handled properly
  • Items are not handled in isolation, as you requested, the processes still receive a list of items.

That is why I would not use GenStage if I required process isolation for handling each item. Although, as it is a functional immutable language, I guess you are fine with handling lists of items.

defmodule ListUpdater do
  def start_link(producer \\ nil) do
    spawn_link(fn -> init(producer) end)
  end

  defp init(producer) do
    loop(producer)
  end

  defp loop(producer) do
    IO.puts("sending list")
    send(producer, {:new_list, generate_list()})
    # Send a list every minute, here every second
    Process.sleep(1000)
    loop(producer)
  end

  defp generate_list() do
    Stream.cycle(["A", "B", "C"])
    # The length of the list
    |> Enum.take(1000)
    |> Enum.shuffle()
    |> Enum.with_index()
    |> Enum.map(fn {key, index} -> %{key: key, id: index} end)
  end
end

defmodule GenstageExample.Producer do
  use GenStage

  def start_link() do
    GenStage.start_link(__MODULE__, [], name: __MODULE__)
  end

  def init(_arg) do
    {:producer, :some_state, dispatcher: GenStage.BroadcastDispatcher}
  end

  def handle_info({:new_list, list}, state) do
    {:noreply, list, state}
  end

  def handle_demand(_demand, state) do
    IO.puts("got demand")
    {:noreply, [], state}
  end
end

defmodule GenstageExample.Consumer do
  use GenStage

  # Starts consumer with his code
  def start_link(code) do
    GenStage.start_link(__MODULE__, code)
  end

  def init(code) do
    # Next line should filter messages sent from producer
    state = %{code: code}

    {:consumer, state,
     subscribe_to: [
       {GenstageExample.Producer, selector: fn %{key: key} -> code == key end, max_demand: 1000}
     ]}
  end

  def handle_events(events, _from, state) do
    # Prints every message received
    for event <- events do
      IO.puts(
        "[#{state.code}] #{inspect(self)} Received: #{inspect(event)}, Expected: #{state.code}"
      )
    end

    # As a consumer we never emit events
    {:noreply, [], state}
  end
end

GenstageExample.Producer.start_link()
GenstageExample.Consumer.start_link("A")
GenstageExample.Consumer.start_link("B")
GenstageExample.Consumer.start_link("C")
ListUpdater.start_link(GenstageExample.Producer)


Edit: You could also use the PartitionDispatcher, with the same code as above except for those differences:

  # GenstageExample.Producer
  def init(_arg) do
    {:producer, :some_state,
     dispatcher:
       {GenStage.PartitionDispatcher,
        partitions: ["A", "B", "C"], hash: fn event -> {event, event.key} end}}
  end


  # GenstageExample.Consumer
  def init(code) do
    # Next line should filter messages sent from producer
    state = %{code: code}

    {:consumer, state,
     subscribe_to: [
       {GenstageExample.Producer, partition: code, max_demand: 1000}
     ]}
  end

It makes more sense to me, but all the possible codes must be known beforehand. For example if you write partitions: ["A", "B"], then a list item with "C" will crash the producer. On the other hand, with the broadcast dispatcher, it is the opposite, as if you declare A,B,C, then a list item that would have D would be ignored (an maybe stay in the producer memory for ever ? I don’t know). So I would rather have the process to crash and use partitions that have unhandled events.

CAIOHOBORGHI

CAIOHOBORGHI

Hey lud, thanks for the code, I think that this will work with a few adjustments.

The only think is that I cant write different validate functions, because I dont know which code will be on the list(And even if I do, I wouldnt write 800 versions).

But with validate, I can get the code and the infos, and do what I need.

The final ideia is to do something like

defp validate({%{code: code, infos: infos}) do
    IO.puts("Checking code #{code}...")
    {age, name, nickname} = infos
    if age > 18 do
         IO.puts("#{nickname} is an adult ")
    end

end

Thats a silly demonstration, but the goal is to do specific commands according to properties inside “infos”.

Thanks for your time and effort

Where Next?

Popular in Questions Top

sergio
In Ruby, I can go: User.find_by(email: "foobar@email.com").update(email: "hello@email.com") How can I do something similar in Elixir? ...
New
Harrisonl
We have an ECS cluster with 4 services, where each task joins a single cluster, via discovery ECS discovery service. Currently when I de...
New
logicmason
Hi there, I'm working through my first release with elixir/phoenix. I've built a release with distillery and found that it crashes when I...
New
freewebwithme
Using vs code and installed ElixirLS: support and debugger. And I got an error popped up on start up says Failed to run ‘elixir’ comma...
New
chensan
I have a User schema with a :from_id field set to type :string: defmodule TweetBot.Repo.Migrations.CreateUsers do use Ecto.Migration ...
New
mathew4509
I have a list say x = ["23gh", "56kh", "97mh"] I would like to pass each element to Val in each iteration. Say, in iteration 1 -------...
New
WestKeys
Currently suffering from paralysis by [HTTP client] analysis. This is rather unusual in Elixirland as there tends to be consensus on the ...
New
beno
I will often find my self writing things similar to: case some_value do nil -&gt; something() "" -&gt; something() _ -&gt; someth...
New
skosch
To my knowledge, put_in, Map.update etc. all have the one limitation of not automatically creating intermediate keys when needed (for exa...
New
vrod
I am using the Starship cross-shell prompt – it seems pretty nice, but I get some errors: [WARN] - (starship::utils): Executing command ...
New

Other popular topics Top

sergio_101
I am VERY much an elixir newbie. I have taken one elixir course and one phoenix course on Udemy. During that course, I saw the instructor...
New
KronicDeth
Elixir plugin for JetBrain’s IntelliJ Platform (including Rubymine) This is a plugin that adds support for Elixir to JetBrains IntelliJ...
289 35421 110
New
SoCreat
i’m a new one to elixir which editor can i use vs code? or atom? Thanks! :smiley:
New
lk-geimfari
What is most correct way to open, read and parse JSON file with poison? For example if we have example.json file in root of some projec...
New
vac
Hi, I'm quite new in Elixir and I'm trying to format a string to a PEM format. I have the certificate value like MIIDBTCCAe2...... and ...
New
myronmarston
The Elixir Typespec docs show the following syntax for keyword lists in typespecs: # ... | [key: type] # keyword lis...
New
stefanluptak
Hello everybody, usually, I use a 29" ultra-wide monitor for VSCode which can easily accomodate explorer (files panel) + file with code ...
New
9mm
I am constructing a JSON object (map) and I need to conditionally set a field. I’m trying to write proper elixir-way code… and I’m at a l...
New
AstonJ
by Lance Halvorsen Elixir and Phoenix are generating tremendous excitement as an unbeatable platform for building modern web application...
460 27162 124
New
romenigld
I am trying to run a deploy with docker and I successfully runned with this command: docker build -t romenigld/blog-prod . but when I t...
New

We're in Beta

About us Mission Statement