Add rate limiting example. (#249)

socketry · Jul 14, 2024 · 341aa8c · 341aa8c
1 parent cb4bd27
commit 341aa8c
Show file tree

Hide file tree

Showing 4 changed files with 240 additions and 0 deletions.
diff --git a/examples/pushback/Gemfile b/examples/pushback/Gemfile
@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2024, by Peter Schrammel.
+# Copyright, 2024, by Samuel Williams.
+
+source 'https://rubygems.org'
+
+gem "falcon"# frozen_string_literal: true
diff --git a/examples/pushback/Gemfile.lock b/examples/pushback/Gemfile.lock
@@ -0,0 +1,80 @@
+GEM
+  remote: https://rubygems.org/
+  specs:
+    async (2.13.0)
+      console (~> 1.25, >= 1.25.2)
+      fiber-annotation
+      io-event (~> 1.6, >= 1.6.5)
+    async-container (0.18.2)
+      async (~> 2.10)
+    async-http (0.69.0)
+      async (>= 2.10.2)
+      async-pool (~> 0.7)
+      io-endpoint (~> 0.11)
+      io-stream (~> 0.4)
+      protocol-http (~> 0.26)
+      protocol-http1 (~> 0.19)
+      protocol-http2 (~> 0.18)
+      traces (>= 0.10)
+    async-http-cache (0.4.3)
+      async-http (~> 0.56)
+    async-pool (0.7.0)
+      async (>= 1.25)
+    async-service (0.12.0)
+      async
+      async-container (~> 0.16)
+    console (1.25.2)
+      fiber-annotation
+      fiber-local (~> 1.1)
+      json
+    falcon (0.47.7)
+      async
+      async-container (~> 0.18)
+      async-http (~> 0.66, >= 0.66.3)
+      async-http-cache (~> 0.4.0)
+      async-service (~> 0.10)
+      bundler
+      localhost (~> 1.1)
+      openssl (~> 3.0)
+      process-metrics (~> 0.2.0)
+      protocol-rack (~> 0.5)
+      samovar (~> 2.3)
+    fiber-annotation (0.2.0)
+    fiber-local (1.1.0)
+      fiber-storage
+    fiber-storage (0.1.2)
+    io-endpoint (0.11.0)
+    io-event (1.6.5)
+    io-stream (0.4.0)
+    json (2.7.2)
+    localhost (1.3.1)
+    mapping (1.1.1)
+    openssl (3.2.0)
+    process-metrics (0.2.1)
+      console (~> 1.8)
+      samovar (~> 2.1)
+    protocol-hpack (1.4.3)
+    protocol-http (0.27.0)
+    protocol-http1 (0.19.1)
+      protocol-http (~> 0.22)
+    protocol-http2 (0.18.0)
+      protocol-hpack (~> 1.4)
+      protocol-http (~> 0.18)
+    protocol-rack (0.6.0)
+      protocol-http (~> 0.23)
+      rack (>= 1.0)
+    rack (3.1.7)
+    samovar (2.3.0)
+      console (~> 1.0)
+      mapping (~> 1.0)
+    traces (0.11.1)
+
+PLATFORMS
+  ruby
+  x86_64-linux
+
+DEPENDENCIES
+  falcon
+
+BUNDLED WITH
+   2.5.11
diff --git a/examples/pushback/README.md b/examples/pushback/README.md
@@ -0,0 +1,19 @@
+# Pushback
+
+The problem to solve is to have long running requests limited to a
+certain number of concurrent request.
+
+If more requests are comming in they are responded with code 429.
+
+# Running it
+```
+bundle install
+bundle exec falcon -n 1 --bind http://localhost:9292
+```
+
+in another console you can run an apache-ab test:
+
+```
+ab -c 17 -n 17  localhost:9292/1
+```
+
diff --git a/examples/pushback/config.ru b/examples/pushback/config.ru
@@ -0,0 +1,132 @@
+# frozen_string_literal: true
+
+# https://tmr08c.github.io/2020/05/concurrent-ruby-hello-async/
+#
+require 'async'
+
+class Pool
+  def initialize(size:, workers:)
+    @size = size
+    @workers = workers
+    @running = []
+    @waiting = []
+    @mutex = Mutex.new
+    watch
+  end
+
+  def push(path)
+    body = Async::HTTP::Body::Writable.new
+    if @waiting.size >= @size
+      puts "pushback"
+      body.close
+      [429, {}, body]
+    else
+      Async do |a_task|
+        task = Task.new(a_task, body, annotation: path)
+        @mutex.synchronize do
+          @waiting << task
+          puts "Pushed: waiting: #{@waiting.size} running: #{@running.size}"
+        end
+        task.wait_for_worker # after this call we're in a yield loop until we're contined by the scheduler
+      end
+      [200, {}, body]
+    end
+  end
+
+  private
+
+  def schedule
+    @mutex.synchronize do
+      @running = @running.reject(&:finished?)
+      available = (@workers - @running.size)
+      waiting = @waiting.size
+      fill = [waiting, available].min
+      if fill.positive?
+        puts "Filling: #{waiting} available: #{available} fill: #{fill}"
+        fill.times do
+          task = @waiting.shift
+          @running << task
+          task.continue
+        end
+      end
+    end
+  end
+
+  def watch
+    Thread.new do
+      loop do
+        sleep 0.1
+        #        puts "Pool: #{@running.size} #{@waiting.size}"
+        schedule
+      end
+    end
+  end
+end
+
+class Task
+  def initialize(task, body, annotation:)
+    @task = task
+    @body = body
+    @fiber = Fiber.current
+    @annotation = annotation
+    @wait = true
+    @timings = {}
+  end
+
+  def step(comment)
+    @timings[comment] = Time.now
+    puts "#{comment}: #{@annotation}"
+  end
+
+  def work
+    step 'working'
+    sleep 5
+    @body.write "(#{Time.now}) Hello World #{Process.pid} #{@task}\n"
+  ensure
+    finish
+  end
+
+  def finished?
+    @task.finished?
+  end
+
+  def wait_for_worker
+    step('waiting')
+    Fiber.scheduler.yield while @wait
+    work
+    #    @fiber.yield
+    #    @task.yield
+  end
+
+  def continue
+    step('continue')
+    @wait = false
+  end
+
+  def finish
+    step('finish')
+    @body.close
+    print_timings
+  end
+
+  def pushback
+    step('finish')
+    @body.close
+  end
+
+  def print_timings
+    puts "Timings: Queued for #{timings_diff('waiting', 'working')}, Running: #{timings_diff('working', 'finish')}"
+  end
+
+  def timings_diff(start, stop)
+    @timings[stop] - @timings[start]
+  end
+end
+
+pool = Pool.new(size: 15, workers: 10)
+
+run do |env|
+  request = env['protocol.http.request']
+  path = request.path
+  pool.push(path) # returns [code, {}, body]
+end