Questions & Help>Questions

Broadway Message Distribution Strategy

#genstage#flow#broadway
mgwidmann

mgwidmann

Here is my system loaded up with work:

https://imgur.com/a/RWN6G4d

It seems Broadway is favoring several workers and giving them lots of messages to process (when I have max_demand: 1). Others are getting no messages, so turning up the number of workers has no effect on throughput. CPU & Network never hit maximum, the system just processes slowly.

Does Broadway use a round-robin message distribution strategy? Can I change it? How would I ensure each worker only ever has 1 message to work on at a time?

13 comments
#genstage#flow#broadway
3400 13

Showing Posts 1 to 10

josevalim

josevalim

Creator of Elixir

It uses the default demand dispatcher from GenStage. Can you please post your topology definition? A high number of workers and low max_demand should be enough to make it process one by one. Also, you may be running into the issue that the producer cannot produce fast enough, so there is not enough work for all cores. More information would be appreciated.

2
Post #1
mgwidmann

mgwidmann OP

Here is the broadway setup I tried in the above screenshot. Previously I tried 200 workers with the same effect. I can try 20k but I definitely don’t want to be processing that many files at a time. The files size range unfortunately from a few kb to 200-300 MB so their weights are probably not being respected.

    Broadway.start_link(__MODULE__,
      name: __MODULE__,
      producers: [
        default: [
          module:
            {BroadwaySQS.Producer,
             queue_name: "my-pipeline-prototype",
             max_number_of_messages: 10,
             wait_time_seconds: 20,
             visibility_timeout: 300,
             receive_interval: 10},
          stages: 50
        ]
      ],
      processors: [
        # downloaders: [
        #   max_demand: 1,
        #   stages: System.schedulers_online()
        # ],
        default: [
          max_demand: 1,
          # + div(System.schedulers_online(), 2)
          stages: 2000
        ]
      ],
      batchers: [
        default: [
          batch_size: 10,
          batch_timeout: 10_000,
          stages: 50
        ]
      ]
    )

There is more information in my last post:

1
Post #2
josevalim

josevalim

Creator of Elixir

Yeah, I think the issue is that we can’t get that much data that fast. Try starting two instances of the same broadway (with different names) and see if you can max it out this way.

1
Post #3
josevalim

josevalim

Creator of Elixir

Actually, ignore me, Broadway.SQS uses one poller per producer stage.

1
Post #4
mgwidmann

mgwidmann OP

Starting multiple instances doesn’t help anything. It just ends up wasting a lot of CPU/Network resources. I’ve already tried running with up to 3 of these instances.

Why would a processor ask for more work when it has 40+ messages in its message queue already? Are messages getting distributed to the worker when they did not ask for it?

0
Post #5
josevalim

josevalim

Creator of Elixir

A processor should not receive more than max_demand, if it is doing so, it is a bug. Add an IO.inspect(length(events)) here and if you receive more than 1 for max_demand == 1, then it is a bug. Maybe we are not passing the max_demand forward?

EDIT: added link.

1
Post #6
mgwidmann

mgwidmann OP

Do you mean messages?

1
Post #7
josevalim

josevalim

Creator of Elixir

Yes, apparently I can’t internet today.

2
Post #8
mgwidmann

mgwidmann OP

The value is always 1 but the PID is almost always the same few PIDs. Remember, there are 2000 processors here waiting around for work, but look at which PIDs are actually working:

2019-07-24T02:56:51.756000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:56:51.756000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:56:52.259000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:56:52.876000 [warn] Broadway Producer (#PID<0.547.0>) received 1 messages

2019-07-24T02:56:56.760000 [warn] Broadway Producer (#PID<0.547.0>) received 1 messages

2019-07-24T02:56:59.487000 [warn] Broadway Producer (#PID<0.547.0>) received 1 messages

2019-07-24T02:57:10.134000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:10.374000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:10.515000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:10.745000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:10.898000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:12.190000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:12.141000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:14.315000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:14.934000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:15.429000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:15.685000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:16.833000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:19.277000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:25.162000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:27.335000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:29.114000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:30.339000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:30.792000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:30.953000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T02:57:35.670000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:37.250000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:37.514000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:38.204000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:39.228000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:39.706000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:39.892000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:40.836000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:41.252000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:41.940000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:44.214000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:45.612000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:57:55.786000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:57.739000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:57:58.921000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:58:00.790000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:58:00.938000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:58:01.430000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:58:01.178000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:58:02.589000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:58:03.206000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:58:04.298000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:58:05.262000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:58:07.560000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:58:10.996000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T02:58:16.141000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T02:58:31.597000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

Heres a screenshot from observer while this is ongoing (sorted by MsgQ desc). There are only a few processors which get lots of messages:

https://imgur.com/a/uWd79tT

When running again and sorting by Reds to see which processes did the most work, it looks a bit lopsided as well.

https://imgur.com/a/iUJpeXL

Any ideas?

0
Post #9
mgwidmann

mgwidmann OP

So I tried with only big files (300-400 files queued up >= 100mb, running from my laptop w/ wifi) to see if smaller files was causing an issue and it seems not to be the case.

2019-07-24T03:22:58.852000 [warn] Broadway Producer (#PID<0.545.0>) received 1 messages

2019-07-24T03:22:58.852000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T03:22:59.520000 [warn] Broadway Producer (#PID<0.546.0>) received 1 messages

2019-07-24T03:23:00.480000 [info] (#PID<0.545.0>) Begin downloading via curl, 302490264 bytes

2019-07-24T03:23:00.480000 [info] (#PID<0.544.0>) Begin downloading via curl, 173432832 bytes

2019-07-24T03:23:00.480000 [info] (#PID<0.546.0>) Begin downloading via curl, 267965903 bytes

2019-07-24T03:23:02.594000 [warn] Broadway Producer (#PID<0.547.0>) received 1 messages

2019-07-24T03:23:02.594000 [info] (#PID<0.547.0>) Begin downloading via curl, 175853568 bytes

2019-07-24T03:23:04.415000 [warn] Broadway Producer (#PID<0.548.0>) received 1 messages

2019-07-24T03:23:04.415000 [info] (#PID<0.548.0>) Begin downloading via curl, 273748300 bytes

2019-07-24T03:23:06.648000 [warn] Broadway Producer (#PID<0.549.0>) received 1 messages

2019-07-24T03:23:06.648000 [info] (#PID<0.549.0>) Begin downloading via curl, 179757056 bytes

2019-07-24T03:23:10.412000 [warn] Broadway Producer (#PID<0.551.0>) received 1 messages

2019-07-24T03:23:10.412000 [info] (#PID<0.551.0>) Begin downloading via curl, 198276268 bytes

2019-07-24T03:23:11.582000 [warn] Broadway Producer (#PID<0.550.0>) received 1 messages

2019-07-24T03:23:11.582000 [info] (#PID<0.550.0>) Begin downloading via curl, 278604105 bytes

2019-07-24T03:23:11.582000 [warn] Broadway Producer (#PID<0.552.0>) received 1 messages

2019-07-24T03:23:11.582000 [info] (#PID<0.552.0>) Begin downloading via curl, 287281012 bytes

2019-07-24T03:23:12.409000 [warn] Broadway Producer (#PID<0.553.0>) received 1 messages

2019-07-24T03:23:12.409000 [info] (#PID<0.553.0>) Begin downloading via curl, 200230071 bytes

2019-07-24T03:23:13.285000 [warn] Broadway Producer (#PID<0.554.0>) received 1 messages

2019-07-24T03:23:13.285000 [info] (#PID<0.554.0>) Begin downloading via curl, 204331743 bytes

2019-07-24T03:23:14.558000 [warn] Broadway Producer (#PID<0.555.0>) received 1 messages

2019-07-24T03:23:14.558000 [info] (#PID<0.555.0>) Begin downloading via curl, 112947200 bytes

2019-07-24T03:23:14.942000 [warn] Broadway Producer (#PID<0.556.0>) received 1 messages

2019-07-24T03:23:14.942000 [info] (#PID<0.556.0>) Begin downloading via curl, 289457789 bytes

2019-07-24T03:23:25.440000 [info] {"atom_memory":1057985,"binary_memory":43145840,"code_memory":17526429,"ets_memory":7952160,"metrics":"health","process_memory":683837488,"system_memory":84343400,"total_memory":768180888}

2019-07-24T03:24:06.857000 [info] (#PID<0.555.0>) Finish downloading in 52.294123 seconds (112947200 bytes)

2019-07-24T03:24:06.857000 [info] (#PID<0.555.0>) Begin processing 112947200 bytes

2019-07-24T03:24:11.355000 [info] (#PID<0.555.0>) Finish processing in 4.497762 seconds (112947200 bytes)

2019-07-24T03:24:11.360000 [warn] Broadway Producer (#PID<0.555.0>) received 1 messages

2019-07-24T03:24:11.360000 [info] (#PID<0.555.0>) Begin downloading via curl, 211469787 bytes

2019-07-24T03:24:21.363000 [info] Delete 1 messages from SQS queue

2019-07-24T03:24:25.440000 [info] {"atom_memory":1057985,"binary_memory":44927608,"code_memory":17560940,"ets_memory":7961552,"metrics":"health","process_memory":684351832,"system_memory":86185056,"total_memory":770536888}

2019-07-24T03:24:30.100000 [info] (#PID<0.544.0>) Finish downloading in 90.051605 seconds (173432832 bytes)

2019-07-24T03:24:30.100000 [info] (#PID<0.544.0>) Begin processing 173432832 bytes

2019-07-24T03:24:36.463000 [info] (#PID<0.544.0>) Finish processing in 6.362783 seconds (173432832 bytes)

2019-07-24T03:24:36.463000 [warn] Broadway Producer (#PID<0.544.0>) received 1 messages

2019-07-24T03:24:36.463000 [info] (#PID<0.544.0>) Begin downloading via curl, 303573205 bytes

2019-07-24T03:24:46.464000 [info] Delete 1 messages from SQS queue

The odd thing is the logs there look all fine since no one worker is actually working more than one file at a time, but when looking in observer, a bunch of messages are in the mailbox for a few processes still. When looking at one of the workers with a lot of the messages, several of the messages (maybe half) look like this:

{'$gen_consumer',
    {<0.516.0>,#Ref<0.352140224.506462210.91230>},
    [#{'__struct__' => 'Elixir.Broadway.Message',
       acknowledger =>
           {'Elixir.BroadwaySQS.ExAwsClient',
               #Ref<0.352140224.506462209.83744>,
               #{receipt =>
                     #{id => <<"...">>,
                       receipt_handle =>
                           <<"...">>}}},
       batch_key => default,batcher => default,
       data =>
           <<"{\"receivedTime\":\"2019-07-24T03:29:18.668259Z\",\"presignedUrl\":\"https://...\",\"fileSize\":202984451}">>,
       metadata =>
           #{attributes => [],
             md5_of_body => <<"...">>,
             message_attributes => [],
             message_id => <<"...">>,
             receipt_handle =>
                 <<"...">>},
       status => ok}]}

One message like this: (could this one be a problem? why is it asking for 5? maybe this is the batcher?)

{'$gen_producer',{<0.2648.0>,#Ref<0.352140224.506462209.111402>},{ask,5}}

And a few like this: (none of these look to be that interesting…)

{'EXIT',#Port<0.69>,normal}

{system,{<0.8279.0>,#Ref<0.352140224.506462210.165688>},get_status}
0
Post #10

Where Next? Top

View thread on forum (has 13 replies!)
genstage
flow
broadway
Home Questions & Help>Questions
12 3351 13
Last post
REPLY VIEW THREAD WITH 13 POSTS

Trending in Questions Top

Blokh
Questions & Help>Questions

Huge CSV processing that needs to be processed hourly - any tips?
Hey guys, I’ve got a huge CSV ( around 10 GB ) that needs to be processed hourly Do you guys have any suggestions what is the best prac...
/phoenix
10 1191 5
New
kszambelanczyk
Questions & Help>Questions

Delete file on S3 with Waffle
Hello! Could someone please give me a help/sample code, how to delete a file from s3 using waffle/waffle_ecto from Phoenix app. I creat...
/phoenix#waffle#waffle-ecto
1 695 1
New
Onor.io
Questions & Help>Questions

How To Create A Select Control From A Lookup Table?
I have what I’ve heard referred to as a “lookup table” in my database. This is a way of assigning codes to common values. One common lo...
/phoenix
3 243 5
New
Trolleger
Questions & Help>Questions

Proper way to send channel events to multiple different people for live updates?
What approach to take when sending live updates to “random” users Hi! I have a question, I have a little chat app, and when I create a DM...
/phoenix#channels
2 689 2
New
matt-savvy
Questions & Help>Questions

Honeybadger and Bandit, Phoenix, Plug errors
Anyone here using Honeybadger? My Honeybadger account is being overwhelmed with noise from some bots. Seeing a lot of Bandit.HTTPError...
/phoenix#bandit
1 222 2
New
RemyXRenard
Questions & Help>Questions

Livebook DataTable displaying charlist as list
I’m seeing that a list inside a Kino.DataTable will be interpreted as a charlist, even if the Kino.configure() is set to charlists: :as_l...
/livebook#charlist
2 325 3
New
samoloth
Questions & Help>Questions

Prevent_hijacking? on email change confirmation with magic link strategy
Hi, I’ve just set up an application with ash_authentication. There is only magic link strategy for now, so there is no confirmation add o...
/ash#ash-authentication
0 196 1
New

Other Trending Topics Top

mudasobwa
Chat & Discussions>Discussions

Cure - a new language compiled to BEAM
I am happy to introduce the very α version of the new programming language compiled to BEAM. Welcome Cure. It has literally three kille...
#static-types#type-checking#types#fsm
168 5507 69
New
garrison
News>Announcing

Hobbes - a low-level distributed database for the Elixir programming language
Hobbes is a low-level distributed database for the Elixir programming language. Hobbes provides a simple, safe, and scalable storage lay...
#library#database#distributed#hobbes
302 4824 87
New
mcass19
News>Announcing

ExRatatui - Elixir bindings for the Rust ratatui terminal UI library
ExRatatui lets you cook up rich terminal UIs in Elixir, powered by Rust’s ratatui via Rustler NIFs. Build interactive terminal applicatio...
#library#rust#terminal#tui
58 2534 15
New
Damirados
News>Announcing

Emerge & Solve - a GUI framework for Elixir
Hello everyone. After busy few months I am happy to announce v0.1.0 of Emerge &amp; Solve. They are GUI (Emerge) and State management (S...
/nerves#library#frameworks#gui#native#state
59 2300 23
New
netoum
News>Announcing

Corex - Accessible and unstyled UI Phoenix components
Corex is an accessible, unstyled UI component library for Phoenix that integrates Zag.js state machines using Vanilla JavaScript and Live...
/phoenix#library#ui#liveview#hooks#accessibility#component#tableau
61 2202 19
New
wintermeyer
News>Announcing

Vutuv.de - an open-source LinkedIn alternative built in Phoenix
There are three potential reasons for members of this forum to have a look at https://vutuv.de You are tired or annoyed of LinkedIn. Yo...
/phoenix#fediverse#project
76 1722 40
New
Back to top

Questions & Help>Questions

Questions Questions ❯

Latest on Elixir Forum

Elixir Forum ❯

Sponsor Spotlight

Potions

Deploy Phoenix to your own VPS with PaaS ease: push-to-deploy, zero downtime, auto SSL, managed Postgres. 14-day free trial at potions.io

Our Sponsors

EEF

Supporting innovation across the BEAM ecosystem.

Coders51

We build reliable cloud platforms for business-critical systems.

AppSignal

Catch errors, track performance, monitor hosts and more.

Honeybadger

Error tracking, logs, uptime, and Just Enough APM. Built for developers.

Code Sync

Elixir, Erlang & BEAM conferences from Code Sync - join us in 2026!

Pragmatic Bookshelf

Practical resources that improve the lives of professional developers.

ElixirCasts

Producing high quality Elixir screencasts since 2017.

WyeWorks

Enabling companies to succeed by building software people love.

Software Mansion

The team behind Membrane, Popcorn, LiveDebugger. Available for hire.

Pragmatic Studio

Courses that'll move you from confusion to "Aha, now I get it!"

View all our sponsors ❯

Categories:

Sub Categories:

Forums

Popular Tags

Our Sponsors

EEF

Supporting innovation across the BEAM ecosystem.

Coders51

We build reliable cloud platforms for business-critical systems.

AppSignal

Catch errors, track performance, monitor hosts and more.

Honeybadger

Error tracking, logs, uptime, and Just Enough APM. Built for developers.

Code Sync

Elixir, Erlang & BEAM conferences from Code Sync - join us in 2026!

Pragmatic Bookshelf

Practical resources that improve the lives of professional developers.

ElixirCasts

Producing high quality Elixir screencasts since 2017.

WyeWorks

Enabling companies to succeed by building software people love.

Software Mansion

The team behind Membrane, Popcorn, LiveDebugger. Available for hire.

Pragmatic Studio

Courses that'll move you from confusion to "Aha, now I get it!"

View all our sponsors ❯

We're in Beta

About us Mission Statement

Options

Thread Display Mode




Thread Preview

Skip Thread Previews