Threads are implemented by ruby to serve concurrent programming . Threads exist in an operating system process and have access to its memory. And practically any piece of code written by ruby is executed by a thread, also known as the main thread.
Threads are useful for scripts that can be executed independently, especially when the code spends time waiting for external events. This type of situation occurs when you deal with I / O (I / O) operations.
We have an following example:
We have a Worker class:
1 2 3 | class Worker end |
The purpose of this Worker is to do something, but what makes it so useful is whether it can be done in a separate thread or maybe multiple threads.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | class Worker attr_reader :num_threads, :threads private :threads def initialize(num_threads:) @num_threads = num_threads @threads = [] end def spawn_threads num_threads.times do threads << Thread.new do # there will be work that the worker performs end end end end |
Because threads are useful for heavy I / O operations, this Worker is perfect for making HTTP requests , manipulating files on disk or making DB requests. We can have an example of an operation stream as follows:
“Hey, woker! Please send this data to the API and fetch some data from another API, after saving some of that data to my database and don’t forget to save everything you’ve done. show in a log file “- this is a perfect job for the worker.
So how can we leave a job to a worker? This is simple if you have to perform a monotonous task multiple times:
1 2 3 4 5 6 7 8 9 10 11 | class Worker # rest of the class omitted def spawn_threads num_threads.times do threads << Thread.new do HealthService::API.ping # send a HTTP request end end end end |
But if you need to perform different types of work depending on the external context, then we can use Queues to solve this problem.
1 2 3 4 5 6 7 8 9 10 11 12 13 | class Worker def initialize(num_threads:) @num_threads = num_threads @threads = [] @queue = Queue.new end # rest of the class omitted def enqueue(action, payload) queue.push([action, payload]) end end |
With the Worker # enqueue method, we can now transfer work to Worker. This can be done in many ways. For example, the action could be a Proc and the payload could be an argument to Proc.
1 2 3 4 5 6 7 8 9 10 11 | require "net/http" require "json" action = proc do |data| Net::HTTP.post( URI("https://api.some-ping-service.com"), data.to_json, "Content-Type" => "applicatoin/json", ) end worker_instance.enqueue(action, { ok: true }) |
The great thing about deploying Queues with ruby is that they are intrinsically safe for the thread sequence to be deployed.
To perform actions on the Worker without consuming all the CPU resources, we need to arrange the dequeue algorithm – intelligently and effectively remove jobs from the queue. For example, we have the following example:
1 2 3 4 5 | queue = Queue.new loop do puts "we need dequeue actions and do some job" unless queue.empty? end |
Such a loop will eat up your free CPU time. When we check the processes on the machine, the result is as follows:
1 2 3 4 | $ top -o cpu PID COMMAND %CPU TIME 56681 ruby 99.9 01:58.17 |
The most common approach to solving this problem is to use sleep :
1 2 3 4 5 6 | queue = Queue.new loop do puts "we need dequeue actions and do some job" unless queue.empty? sleep 5 end |
This is useful but not perfect. Imagine how the Ruby interpreter takes time to switch between the main thread and the workflow thread during each sleep period to realize that we have nothing to do because the worker queue is empty. This problem will be multiplied by a number of topics and will get worse as the sleep period must become smaller. Sleep not an effective way to capture something in the future.
And again, Queues is used to solve:
The Queue#pop(non_block = false) , when non_block = false , will pause the current thread if the queue is empty until the data is pushed onto the queue. This means that the worker thread has nothing to do waiting for the next action. No need to sleep.
And for convenience Worker has specific methods describing its state:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 | class Worker # rest of the class omitted private attr_reader :queue, :threads def actions? !queue.empty? end def running? !queue.closed? end def dequeue_action queue.pop(true) end def wait_for_action queue.pop(false) end end |
Notice the wait_for_action method. It suspends a Worker thread, as described above, when we take no action in the queue.
Finally, it’s time for the main part of the Worker class:
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 | class Worker # rest of the class omitted def spawn_threads num_threads.times do threads << Thread.new do while running? || actions? action_proc, action_payload = wait_for_action action_proc.call(action_payload) if action_proc end end end end # rest of the class omitted end |
With # wait_for_action equivalent to queue.pop (false), the worker will start exiting the queue exactly when it starts to fill up.
The last one. Workers need a method to stop it
1 2 3 4 5 6 7 8 9 10 11 | class Worker # rest of the class omitted def stop queue.close threads.each(&:exit) threads.clear true end # rest of the class omitted end |
So I just introduced you about threads and queues in ruby and some notes when working with it.
References
https://medium.com/@shvetsovdm/playing-with-ruby-threads-and-queues-52beb6e8613c