Playing with queues and Discrete-Event Simulations in Crystal

readme.md

Prerequisite

Your system has an installation of crystal 0.33.0 or higher.

How to run

Clone the files in this gist, then run the following once, to install the code dependencies.

shards install

You can now run a simulation with

crystal des.cr

You can now tweak the nodes and run your own simulation!

des.cr

require "statistics"
require "tablo"
extend Statistics::Distributions

# Launches a DES that will halt once the simulation's time exceeds `max_t`
def simulate(nodes : Array(Node), max_t : Int32)
  t = 0_f64

  loop do
    next_node = nodes.map { |a| {a, a.next(t) || Float64::MAX} }
      .min_by(&.last).first
    next_event = next_node.get

    next_node.output.push next_event

    t = next_event
    
    break if t > max_t
  end
end

# Prints a human-friendly report featuring machine utilisation rate and final queue sizes
def report(nodes, iterations)
  data = nodes.map { |m|
    server_utilization = [m.processing_times.sum/iterations * 100, 100].min
    [m.name, server_utilization.round(2).to_f, m.output.size]
  }

  table = Tablo::Table.new(data) do |t|
    t.add_column("node") { |n| n[0] }
    t.add_column("utilisation") { |n| "#{n[1]}%" }
    t.add_column("output queue size") { |n| n[2] }
  end

  puts table
end

# Represent either a generator or a processing node in the system.
# A generators is a `Node` where `input` is an empty list, i.e. a `Node` with no dependencies 
class Node
  getter name, input, output, gen
  getter processing_times = [] of Float64

  def initialize(@name : String, @input : Array(Deque(Float64)),
    @output : Deque(Float64), @gen : Distribution(Float64))
  end
  @next : Float64? = nil
  def next(t : Float64)
    return @next.not_nil! unless @next.nil?
    
    unless @input.any?(&.first?.nil?) # all the dependencies are satisfied, we can process the input
      args = @input.map(&.shift)
      processing_time = Math.max(0_f64, @gen.rand)
      processing_times << processing_time
      # puts "#{t.round(3)} #{@name}: processed #{args.map(&.round(3))} in #{processing_time.round(3)}"
      @next = processing_time + t
    end
  end

  def get
    event = @next.not_nil!
    @next = nil
    event
  end
end

# Nodes push Float64 - representing simulation clock timestamps - into queues
q1 = Deque(Float64).new
q2 = Deque(Float64).new
q3 = Deque(Float64).new
q4 = Deque(Float64).new
q5 = Deque(Float64).new
sink = Deque(Float64).new

x = Node.new "x", input: [] of Deque(Float64), output: q1, gen: Uniform.new(0.4,0.6)
y = Node.new "y", input: [] of Deque(Float64), output: q2, gen: Normal.new(0.5,0.2)
m1 = Node.new "m1", input: [q1], output: q3, gen: Normal.new(0.1,0.03)
m2 = Node.new "m2", input: [q2], output: q4, gen: Normal.new(0.15,0.04)
m3 = Node.new "m3", input: [q3, q4], output: q5, gen: Constant.new(0.3)
m4a = Node.new "m4a", input: [q5], output: sink, gen: Normal.new(0.6, 0.13)
m4b = Node.new "m4b", input: [q5], output: sink, gen: Normal.new(0.6, 0.13)

# remember to update `nodes` as you add/remove nodes to your system
nodes = [x, y, m1, m2, m3, m4a, m4b]

T = 1000 # The simulation time after which the simulation should stop

simulate nodes, T
report nodes, T

shard.yml

name: des.cr
version: 0.1.0

authors:
  - lbarasti <lbarasti@users.noreply.github.com>

dependencies:
  tablo:
    github: hutou/tablo
    version: 0.9.3
  statistics:
    github: lbarasti/statistics
    version: 0.2.0

crystal: 0.33.0

license: MIT