Merge pull request #45 from Logflare/feat/performance-tweaks

feat: tweak concurrency, pool count, startup sequence

.github/workflows/deploy.yml

@@ -10,6 +10,6 @@ jobs:
     steps:
       - uses: actions/checkout@v3
       - uses: superfly/flyctl-actions/setup-flyctl@master
-      - run: flyctl deploy --remote-only
+      - run: flyctl deploy --remote-only --vm-size=shared-cpu-8x -y
         env:
           FLY_API_TOKEN: ${{ secrets.FLY_ACCESS_TOKEN }}

config/prod.exs

@@ -12,4 +12,5 @@ config :loadfest,
     "loadfest.test.4",
     "loadfest.test.5"
   ]
-  # max_rps: 40
+
+# max_rps: 40

lib/application.ex

@@ -23,10 +23,10 @@ defmodule Loadfest.Application do
           [
             {Task.Supervisor, name: Loadfest.TaskSupervisor},
             :hackney_pool.child_spec(:loadfest_pool, timeout: 15_000, max_connections: 10_000),
-            {Finch, name: Loadfest.Finch, size: 10_000, count: 10},
+            {Finch, name: Loadfest.Finch, size: 10_000, count: 1},
             # Loadfest.Worker,
-            Loadfest.Pipeline,
-            Loadfest.Counter
+            Loadfest.Counter,
+            Loadfest.Pipeline
 
             # {Telemetry.Metrics.ConsoleReporter, metrics: metrics()}
           ]

lib/counter.ex

@@ -20,6 +20,7 @@ defmodule Loadfest.Counter do
   def add(batch_size) do
     GenServer.cast(__MODULE__, {:add, batch_size})
   end
+
   def requests() do
     GenServer.call(__MODULE__, :requests)
   end
@@ -34,7 +35,7 @@ defmodule Loadfest.Counter do
 
   def handle_call(:requests, _caller, state) do
     value = :counters.get(state.ref, idx(:requests))
-  {:reply, value, state}
+    {:reply, value, state}
   end
 
   def handle_info(:log, state) do

lib/pipeline.ex

@@ -17,7 +17,8 @@ defmodule Loadfest.Pipeline do
     end
 
     def handle_demand(demand, state) when demand > 0 do
-      events = Loadfest.Worker.make_batch()
+      events = Loadfest.Worker.make_batch(Enum.random(50..100))
+      # events = Loadfest.Worker.make_batch_stream() |> Enum.to_list() |> dbg()
       {:noreply, [%Broadway.Message{data: events, acknowledger: {__MODULE__, :ack, 3}}], state}
     end
 
@@ -27,13 +28,15 @@ defmodule Loadfest.Pipeline do
   end
 
   def start_link(_opts) do
+    Logger.info("Starting pipeline with #{4 * System.schedulers_online()} concurrency")
+
     Broadway.start_link(__MODULE__,
       name: __MODULE__,
       producer: [
         module: {Loadfest.Pipeline.Producer, []}
       ],
       processors: [
-        default: [concurrency: min(@max_rps, 50), max_demand: 1]
+        default: [concurrency: 4 * System.schedulers_online(), max_demand: 1]
       ]
     )
   end
@@ -47,6 +50,7 @@ defmodule Loadfest.Pipeline do
     }
 
     rps = Loadfest.Counter.requests()
+
     if rps < @max_rps do
       request = Loadfest.Client.send(name, body)