Ruby Multithreading

Every program running on the system is a process. Each process contains one or more threads.

A thread is a single sequential control flow in a program. Running multiple threads simultaneously in a single program to accomplish different tasks is called multithreading.

In Ruby, we can create multithreading through the Thread class. Ruby threads are lightweight and can implement parallel code in an efficient way.

Creating Ruby Threads

To start a new thread, simply call Thread.new:

#Thread #1 code section Thread.new { #Thread #2 executing code } #Thread #1 executing code

Example

The following example demonstrates how to use multithreading in a Ruby program:

Example

#!/usr/bin/ruby def func1 i=0 while i<=2 puts "func1 at: #{Time.now}" sleep(2) i=i+1 end end def func2 j=0 while j<=2 puts "func2 at: #{Time.now}" sleep(1) j=j+1 end end puts "Started At #{Time.now}" t1=Thread.new{func1()} t2=Thread.new{func2()} t1.join t2.join puts "End at #{Time.now}"

Try it »

The execution result of the above code is:

Started At Wed May 14 08:21:54 -0700 2014
func1 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:54 -0700 2014
func2 at: Wed May 14 08:21:55 -0700 2014
func1 at: Wed May 14 08:21:56 -0700 2014
func2 at: Wed May 14 08:21:56 -0700 2014
func1 at: Wed May 14 08:21:58 -0700 2014
End at Wed May 14 08:22:00 -0700 2014

Thread Lifecycle

1. Thread.new can be used to create a thread. Similarly, Thread.start or Thread.fork can be used with the same syntax to create threads.

2. After a thread is created, there is no need to start it; the thread will execute automatically.

3. The Thread class defines some methods to control threads. A thread executes the code block in Thread.new.

4. The last statement in the thread's code block is the thread's value, which can be called via thread methods. If the thread has finished executing, it returns the thread value; otherwise, no value is returned until the thread completes.

5. The Thread.current method returns the object representing the current thread. The Thread.main method returns the main thread.

6. Use the Thread.join method to execute threads. This method suspends the main thread until the current thread finishes execution.

Thread States

Threads have 5 states:

Thread StateReturn Value
Runnablerun
SleepingSleeping
Exitingaborting
Normal terminationfalse
Terminated due to exceptionnil

Threads and Exceptions

When an exception occurs in a thread and is not caught by rescue, the thread is usually terminated without warning. However, if other threads are waiting for that thread due to Thread#join, the waiting threads will also have the same exception raised.

begin t = Thread.new do Thread.pass #The main thread is indeed waiting for join raise "unhandled exception" end t.join rescue p $! # => "unhandled exception" end

Using the following 3 methods, you can make the interpreter interrupt execution when a thread terminates due to an exception.

  • Specify when starting the script-doption, and run in debug mode.
  • useThread.abort_on_exceptionSet the flag.
  • UseThread#abort_on_exceptionSet the flag for the specified thread.

After using one of the above three methods, the entire interpreter will be interrupted.

t = Thread.new { ... } t.abort_on_exception = true

Thread Synchronization Control

In Ruby, there are three ways to achieve synchronization, namely:

1. Implementing thread synchronization through the Mutex class

2. Implementing thread synchronization through the Queue class for data handover

3. Using ConditionVariable for synchronization control

Implementing Thread Synchronization with the Mutex Class

To implement thread synchronization control through the Mutex class, if multiple threads need a program variable at the same time, you can lock that variable using lock. The code is as follows:

Example

#!/usr/bin/ruby require "thread" puts "Synchronize Thread" @num=200 @mutex=Mutex.new def buyTicket(num) @mutex.lock if @num>=num @num=@num-num puts "you have successfully bought #{num} tickets" else puts "sorry,no enough tickets" end @mutex.unlock end ticket1=Thread.new 10 do 10.times do |value| ticketNum=15 buyTicket(ticketNum) sleep 0.01 end end ticket2=Thread.new 10 do 10.times do |value| ticketNum=20 buyTicket(ticketNum) sleep 0.01 end end sleep 1 ticket1.join ticket2.join

Try it »

The execution result of the above code is:

Synchronize Thread
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
you have successfully bought 20 tickets
you have successfully bought 15 tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets
sorry,no enough tickets

In addition to using lock to lock a variable, you can also use try_lock to lock a variable, and you can use Mutex.synchronize to synchronize access to a variable.

Implementing Thread Synchronization with the Queue Class for Data Handover

The Queue class represents a thread-safe queue that can synchronize access to the end of the queue. Different threads can use the same queue class without worrying about whether the data in the queue can be synchronized. In addition, using the SizedQueue class can limit the length of the queue.

The SizedQueue class can very conveniently help us develop thread-synchronized applications, because as long as items are added to this queue, there is no need to worry about thread synchronization issues.

The classic producer-consumer problem:

Example

#!/usr/bin/ruby require "thread" puts "SizedQuee Test" queue = Queue.new producer = Thread.new do 10.times do |i| sleep rand(i) #Let the thread sleep for a while queue << i puts "#{i} produced" end end consumer = Thread.new do 10.times do |i| value = queue.pop sleep rand(i/2) puts "consumed #{value}" end end consumer.join

Try it »

The output of the program:

SizedQuee Test
0 produced
1 produced
consumed 0
2 produced
consumed 1
consumed 2
3 produced
consumed 34 produced

consumed 4
5 produced
consumed 5
6 produced
consumed 6
7 produced
consumed 7
8 produced
9 produced
consumed 8
consumed 9

Thread Variables

Threads can have their own private variables. A thread's private variables are written into the thread when it is created. They can be used within the thread's scope, but cannot be shared outside the thread.

But sometimes, what if a thread's local variables need to be accessed by other threads or the main thread? Ruby provides a way to create thread variables by name, treating a thread like a hash-style lookup table. Use []= to write data and [] to read data. Let's look at the following code:

Example

#!/usr/bin/ruby count = 0 arr = [] 10.times do |i| arr[i] = Thread.new { sleep(rand(0)/10.0) Thread.current["mycount"] = count count += 1 } end arr.each {|t| t.join; print t["mycount"], ", " } puts "count = #{count}"

The output of the above code is:

8, 0, 3, 7, 2, 1, 6, 5, 4, 9, count = 10

The main thread waits for child threads to finish executing, then outputs each value separately.

Thread Priority

Thread priority is the main factor affecting thread scheduling. Other factors include the length of CPU execution time, thread group scheduling, and so on.

You can use the Thread.priority method to get a thread's priority and use the Thread.priority= method to adjust the thread's priority.

The default thread priority is 0. Higher-priority threads execute faster.

A Thread can access all data within its own scope, but what if you need to access other threads' data from within a certain thread? The Thread class provides methods for threads to access each other's data. You can simply treat a thread as a Hash table, and use []= in any thread to write data, and [] to read data.

athr = Thread.new { Thread.current["name"] = "Thread A"; Thread.stop } bthr = Thread.new { Thread.current["name"] = "Thread B"; Thread.stop } cthr = Thread.new { Thread.current["name"] = "Thread C"; Thread.stop } Thread.list.each {|x| puts "#{x.inspect}: #{x["name"]}" }

It can be seen that by treating a thread as a Hash table and using the [] and []= methods, we have achieved data sharing between threads.

Thread Mutual Exclusion

Mutex (Mutual Exclusion = mutex lock) is a mechanism used in multithreaded programming to prevent two threads from simultaneously reading and writing the same shared resource (such as a global variable).

Example without Using Mutex

Example

#!/usr/bin/ruby require 'thread' count1 = count2 = 0 difference = 0 counter = Thread.new do loop do count1 += 1 count2 += 1 end end spy = Thread.new do loop do difference += (count1 - count2).abs end end sleep 1 puts "count1 : #{count1}" puts "count2 : #{count2}" puts "difference : #{difference}"

The output of the above example is:

count1 :  9712487
count2 :  12501239
difference : 0

Example Using mutex

Example

#!/usr/bin/ruby require 'thread' mutex = Mutex.new count1 = count2 = 0 difference = 0 counter = Thread.new do loop do mutex.synchronize do count1 += 1 count2 += 1 end end end spy = Thread.new do loop do mutex.synchronize do difference += (count1 - count2).abs end end end sleep 1 mutex.lock puts "count1 : #{count1}" puts "count2 : #{count2}" puts "difference : #{difference}"

The output of the above example is:

count1 :  1336406
count2 :  1336406
difference : 0

Deadlock

When two or more computing units are each waiting for the other to stop running in order to obtain system resources, but neither exits in advance, this situation is called deadlock.

For example, if process p1 occupies the monitor and must also use the printer, but the printer is occupied by process p2, and p2 must use the monitor, this creates a deadlock.

When using Mutex objects, we need to be aware of thread deadlocks.

Example

#!/usr/bin/ruby require 'thread' mutex = Mutex.new cv = ConditionVariable.new a = Thread.new { mutex.synchronize { puts "A: I have critical section, but will wait for cv" cv.wait(mutex) puts "A: I have critical section again! I rule!" } } puts "(Later, back at the ranch...)" b = Thread.new { mutex.synchronize { puts "B: Now I am critical, but am done with cv" cv.signal puts "B: I am still critical, finishing up" } } a.join b.join

The output of the above example is:

A: I have critical section, but will wait for cv
(Later, back at the ranch...)
B: Now I am critical, but am done with cv
B: I am still critical, finishing up
A: I have critical section again! I rule!

Thread Class Methods

The complete Thread class methods are as follows:

Serial NumberMethod Description
1Thread.abort_on_exception
If its value is true, once a thread terminates due to an exception, the entire interpreter will be interrupted. Its default value is false. That is, under normal circumstances, if a thread raises an exception that is not detected by Thread#join or similar, the thread will be terminated without warning.
2Thread.abort_on_exception=
If set totrue, once a thread terminates due to an exception, the entire interpreter will be interrupted. Returns the new state.
3Thread.critical
Returns a boolean value.
4Thread.critical=
When its value is true, thread switching will not occur. If the current thread is suspended (stop) or intervened by a signal, its value will automatically become false.
5Thread.current
Returns the currently running thread (current thread).
6Thread.exit
Terminates the current thread's execution. Returns the current thread. If the current thread is the only thread, it will use exit(0) to terminate its execution.
7Thread.fork { block }
Creates a thread just like Thread.new.
8Thread.kill( aThread )
Terminates the thread's execution.
9Thread.list
Returns an array of live threads that are in the running or suspended state.
10Thread.main
Returns the main thread.
11Thread.new( [ arg ]* ) {| args | block }
Creates a thread and starts execution. The arguments are passed to the block unchanged. This makes it possible to pass values to the thread's local variables when starting the thread.
12Thread.pass
Yields execution to other threads. It does not change the state of the running thread, but hands control over to other runnable threads (explicit thread scheduling).
13Thread.start( [ args ]* ) {| args | block }
Creates a thread and starts execution. The arguments are passed to the block unchanged. This makes it possible to pass values to the thread's local variables when starting the thread.
14Thread.stop
Suspends the current thread until another thread uses the run method to wake it again.

Thread Instance Methods

The following example calls the thread instantiation method join:

Example

#!/usr/bin/ruby thr = Thread.new do #Instantiation puts "In second thread" raise "Raise exception" end thr.join #Calls the instantiation method join

The following is the complete list of instantiation methods:

No.Method description
1thr[ name ]
Retrieves the thread-local data corresponding to name. name can be a string or a symbol. If there is no data corresponding to name, nil is returned.
2thr[ name ] =
Sets the value of the thread-local data corresponding to name. name can be a string or a symbol. If set to nil, the corresponding data in the thread is deleted.
3thr.abort_on_exception
Returns a boolean value.
4thr.abort_on_exception=
If its value is true, once a thread terminates due to an exception, the entire interpreter is interrupted.
5thr.alive?
Returns true if the thread is alive.
6thr.exit
Terminates the thread's execution. Returns self.
7thr.join
Suspends the current thread until the self thread terminates. If self terminates due to an exception, the current thread will raise the same exception.
8thr.key?
Returns true if the thread-local data corresponding to name has already been defined.
9thr.kill
Similar toThread.exit 。
10thr.priority
Returns the thread's priority. The default priority is 0. The larger the value, the higher the priority.
11thr.priority=
Sets the thread's priority. It can also be set to a negative number.
12thr.raise( anException )
Forcibly raises an exception within the thread.
13thr.run
Restarts a suspended (stop) thread. Unlike wakeup, it performs a thread switch immediately. If this method is used on a dead thread, a ThreadError exception is raised.
14thr.safe_level
Returns the security level of self. The current thread's safe_level is the same as $SAFE.
15thr.status
Represents the status of a live thread using the strings "run", "sleep", or "aborting". Returns false if a thread has terminated normally. Returns nil if it terminated due to an exception.
16thr.stop?
Returns true if the thread is dead or suspended (stop).
17thr.value
Waits until the self thread terminates (equivalent to join), then returns the return value of the thread's block. If an exception occurred during the thread's execution, that exception is raised again.
18thr.wakeup
Changes the state of a suspended (stop) thread to runnable (run). If this method is executed on a dead thread, a ThreadError exception is raised.
Other extensions