1

Distributed Internet Applications - DIA

Processes and Threads

2

Operating Systems Basics

The Processes

A process consists of an executing program, its current values, state information, and the resources used by the operating system to manage its execution.

A program is a text written by a programmer; a process is a dynamic entity which exists only when a program is executing.

3

Process StateAs a process executes, it changes state. Each process may be in following states:

New: The process is being created.

Running: Instructions are being executed.

Waiting: The process is waiting for some event to occur.

Ready: The process is waiting to be assigned to a processor.

Terminated: The process has finished the execution.

4

Process State Transition Diagram

Simplifed finite state diagram for a process's lifetime

new

readyrunning

blocked

terminated

dispatch

queued

event completion waitingfor an event

exit

5

Concurrent ProcessingOn operating systems, multiple processes appear to be executing concurrently on a machine by timesharing resources.

6

Parent and Child ProcessesAt run time, a process may spawn child processes. The original process, called the parent process, continues to run concurrently with the child processes.

A child process is a complete process.parent process

child processes

A parent process can be notified when a child

process has terminated.

7

Process CreationWhen a process create a child process, two possibilities exist in terms of execution:

The parent continues to execute concurrently with its children.

The parent waits until some or all of its children have terminated.

There are two possibilities in terms of the address space of the child process:

The child process is a duplicate of the parent process.

The child process has a program loaded into it

8

Creating a New Process in Ruby

fork splits a running program into two nearly identical processes that can be identified by the return value of the fork call.

There are to ways of using fork method:

pid = forkif pid.nil? # child’s code goes hereelse # parent’s code goes here . . . Process.waitend

pid = fork do # child’s code goes hereend# parent’s code goes here. . .Process.wait

9

Example 1global = 0pid = fork do global += 1 puts "Child: global = #{global}” exitendProcess.waitputs "Parent: Child Complete"puts "Parent: global = #{global}"

Child: global = 1Parent: Child CompleteParent: global = 0

10

Example 2global = 0pid = forkif pid.nil? global += 1 puts "Child: global = #{global}"else Process.wait puts "Parent: Child Complete" puts "Parent: global = #{global}”end

Child: global = 1Parent: Child Complete

Parent: global = 0

11

Example 3global = 0pid = forkif pid.nil? exec(“ls”, “-l”) global += 1 puts "Child: global = #{global}"else Process.wait puts "Parent: Child Complete" puts "Parent: global = #{global}”end

Output: see next slide

12

Example 3

total 552drwxr-xr-x 9 mortezan mortezan 306 8 Nov 23:41 dia-2004-rw-r--r-- 1 mortezan mortezan 5465 3 Nov 18:57 dia.html-rw-r--r-- 1 mortezan mortezan 292 8 Nov 23:26 example1.rb-rw-r--r-- 1 mortezan mortezan 180 6 Nov 16:08 example2.rb-rw-r--r-- 1 mortezan mortezan 195 9 Nov 14:00 process.rb-rw-r--r-- 1 mortezan mortezan 399 5 Nov 23:17 spinner1.rb-rw-r--r-- 1 mortezan mortezan 4486 3 Nov 17:44 web-portal.rtfParent: Child CompleteParent: global = 0

13

ThreadsA process may spawn threads, also known as light weight processes.

Threads carry a minimum of state information.

Threads behave the same as processes.

a process

main thread

child thread 1

child thread 2Concurrent processing within a process

14

Coordination of ThreadsThe concurrent execution of threads may result in a race conditions.

Critical section: A code segment that can be executed concurrently by more than one thread.

Mutual exclusion: A method to ensure that a critical section can only be executed by one thread at a time.

Programming using threads is called multi-threaded programming.

A multi-threaded program that written to guard against race conditions is said to be thread-safe.

15

Threads in RubyWe use multiple threads to split up cooperating tasks within the program.

Threads in Ruby are totally in-process, implemented within the Ruby interpreter.

Threads in Ruby are completely portable.

16

Class ThreadWhen a Ruby script starts up, there is usually a single thread, named main thread.

$ irb --simple-prompt

ruby> Thread.main

=> #<Thread:0x348f8 run>

From within a Ruby script, there are many ways to create a new

thread of execution.

17

Creating Threads (1)Using Thread::new

Thread.new([args]*) { | args |

thread body

}

or

Thread.new([args]*) do |args|

thread body

end

Thread::fork is a synonym for Thread::new.

18

Creating Threads (2)names = %w(one two three)threads = []

for name in names threads << Thread.new(name) { | myName | sleep 10*rand puts "I am thread number: #{myName}" }endthreads.each { | aThread | aThread.join}puts "I am the parent thread"

19

Passing Parameters (1)a = 1b = 4c = 3t1 = Thread.new(a,b,c) do |x,y,z| a = x**2 b = y**2 c = z**2endt1.joinputs "Parent: a = #{a}, b = #{b}, c = #{c}"

Parent thread: a = 1, b = 16, c = 9Output:

Wait for t1 to finish

20

Passing Parameters (2)a = 1b = 4c = 3thread = Thread.new(a,b,c) do |x,y,z| sleep(rand(0)) a = x**2 b = y**2 c = z**2endsleep(rand(0))puts "Parent: a = #{a}, b = #{b}, c = #{c}"

Output 1: Parent: a = 1, b = 16, c = 9Output 2: Parent: a = 1, b = 4, c = 3

Defined in main thread}

21

Manipulating ThreadsWhen a Ruby program terminates, all running threads are killed, regardless of their states.

The parent thread can wait for a child thread by calling that child thread’s Thread#join method. global = 0

t1 = Thread.new { t2 = Thread.new { sleep(rand(0)) global +=1 t3 = Thread.new { sleep(rand(0)) global +=1 } t3.join } t2.join}t1.joinputs "Top thread: global = #{global}"

Output:Top thread: global = 2

22

Using Thread#join (1) global = 0t1 = Thread.new { t2 = Thread.new { sleep(rand(0)) global +=1 t3 = Thread.new { sleep(rand(0)) global +=1 } }}sleep(rand(0))puts "Top thread: global = #{global}"

Possible outputs:Top thread: global = 0

orTop thread: global = 2

orTop thread: global = 1

23

Using Thread#join (2)

global = 0t1 = Thread.new { t2 = Thread.new { global +=1 t3 = Thread.new { global +=1 } t3.join } t2.join}t1.joinputs "Top thread: global = #{global}"

Possible output:Top thread: global = 2

Using Thread#join

24

Exporting Local Variables (1)a = 1000 # globalt1 = Thread.new { t = Thread.current a = 1 # global b = "local to this thread" t[:a] = "accessible variable"}b = t1[:a]puts "Parent: a = #{a}, b = #{b}"

Exported variables are accessible from other threads even after the thread that owned themis dead.

Output:Parent: a = 1, b = accessible variable

At this point is thread t1 dead

25

Exporting Local Variables (2)t1 = Thread.new { t = Thread.current x = "renilreB nie nib hcI" t[:a] = x x.reverse!}b = t1[:a]puts "Parent: #{b}"

Output:Parent: Ich bin ein Berliner

An object reference to a true localvariable can be used as a sort ofshorthand within the thread.

26

Thread.list and Thread.kill Methodst1 = Thread.new { sleep(1000) }t2 = Thread.new do if (Thread.current == Thread.main) puts "This is the main thread" end 1.upto(100) do sleep 0.1 endendcount = Thread.list.sizeputs "Number of living threads: #{count}"Thread.kill(t1)count = Thread.list.sizeputs "Number of living threads: #{count}"count = Thread.list.sizeputs "Number of living threads: #{count}"t2.joincount = Thread.list.sizeputs "Number of living threads: #{count}"

Output:Number of living threads: 3Number of living threads: 2Number of living threads: 2Number of living threads: 1

27

Controlling the Thread Scheduler (1)

Thread::stop stops the execution of the current thread, putting it into a sleep state

A thread that is stopped can be awakened by parent thread using the Thread#run or Thread#wakeup methods.

The status of a thread can be determined using Thread#status method.

28

Controlling the Thread Scheduler (2)

t1 = Thread.new { print "one\n" Thread.stop print "second\n"}puts "Status of thread t1: #{t1.status}"print "three\n"t1.run

oneStatus of thread t1: sleepthreesecond

Output:

29

Controlling the Thread Scheduler (3) t1 = Thread.new { sleep(1000) }t2 = Thread.new { loop {} }t3 = Thread.new { Thread.stop puts "I am thread t3" loop {}}t4 = Thread.new { Thread.exit }t5 = Thread.new { raise "exception" }puts "Status of thread t1: #{t1.status}"puts "Status of thread t2: #{t2.status}"puts "Status of thread t3: #{t3.status}"puts "Status of thread t4: #{t4.status}"puts "Status of thread t5: #{t5.status}"t3.wakeupsleep 0.1puts "Status of thread t3: #{t3.status}"Thread.kill(t3)puts "Status of thread t3: #{t3.status}"

Status of thread t1: sleepStatus of thread t2: runStatus of thread t3: sleepStatus of thread t4: falseStatus of thread t5: I am thread t3Status of thread t3: runStatus of thread t3: false

30

Controlling Threads with Thread::pass

t1 = Thread.new { print "one"; Thread.pass print "two"; Thread.pass print "three"}t2 = Thread.new { print "1"; Thread.pass print "2"; Thread.pass print "3"}t1.join; t2.join

one1two2three3Output:

Thread::pass method invokesThe scheduler to pass execution

To another thread.

31

Using Thread#alive? t1 = Thread.new { loop {} }t2 = Thread.new { Thread.stop }t3 = Thread.new { Thread.exit }puts "t1 alive? -> #{t1.alive?}"puts "t2 alive? -> #{t2.alive?}"puts "t3 alive? -> #{t3.alive?}"sleep 1if t1.alive? Thread.kill(t1) endt2.run if t2.alive? & t2.status == "sleep"puts "t1 alive? -> #{t1.alive?}"puts "t2 alive? -> #{t2.alive?}"

t1 alive? -> truet2 alive? -> truet3 alive? -> falset1 alive? -> falset2 alive? -> true

32

Changing Priority

t1 = Thread.new { loop {sleep 1} }t2 = Thread.new { loop {sleep 1} }t3 = Thread.new { loop {sleep 1} }puts "t1 priority? -> #{t1.priority}"puts "t2 priority? -> #{t2.priority}"puts "t3 priority? -> #{t3.priority}"t1.priority = rand(10)t2.priority = rand(10)t3.priority = rand(10)puts "t1 priority? -> #{t1.priority}"puts "t2 priority? -> #{t2.priority}"puts "t3 priority? -> #{t3.priority}"sleep 1

t1 priority? -> 0t2 priority? -> 0t3 priority? -> 0t1 priority? -> 9t2 priority? -> 9t3 priority? -> 5

t.priority returns the priority of tHigher-priority threads will runbefore lower-priority threads

33

Threads an ExceptionsWhat happens if a thread raises an unhandled exception?

It depends on a flag called abort_on_exception that operates both at the class and instance level.

If abort_on_exception is false, an unhandled exception terminates the current threads.

If abort_on_exception is true, an unhandled exception will terminate all running threads.

34

abort_on_exception

Thread#raise raises an exception and terminates the current thread.The flag abort_on_exceptionis false by default.

Status of t1: sleepStatus of t2: runt1 is terminated with an exceptionStatus of t2: run

a = 1b = 0t1 = Thread.new(a,b) { |x,y| Thread.stop raise "divide by zero!" if y == 0 a/b}t2 = Thread.new {loop{}}sleep 1puts "Status of t1: #{t1.status}"puts "Status of t2: #{t2.status}"t1.runsleep rand(0.5)if t1.status == nilputs "t1 is terminated with an exception"endputs "Status of t2: #{t2.status}"

35

abort_on_exceptiona = 1b = 0Thread.abort_on_exception = truet1 = Thread.new(a,b) { |x,y| Thread.stop raise "divide by zero!" if y == 0 a/b}t2 = Thread.new {loop{}}sleep 1puts "Status of t1: #{t1.status}"puts "Status of t2: #{t2.status}"t1.runputs "Status of t1: #{t1.status}"puts "Status of t2: #{t2.status}"

Status of t1: sleepStatus of t2: runexample1.rb:6: divide by zero! (RuntimeError) from example1.rb:4:in `initialize' from example1.rb:4:in `new' from example1.rb:4

All running threads will terminateif an unhandled exception arises.

36

Using Thread#abort_on_exption

threads = []10.times { |i| threads << Thread.new(i) { Thread.stop puts "Thread #{i}" raise "Thread #{i} raises an exception!" if i == 6 }}threads[6].abort_on_exception = truethreads.each {|t| t.run t.join}

Ex1.rb:6: Thread 6 raises an exception! (RuntimeError) from Ex1.rb:3:in `initialize' from Ex1.rb:3:in `new' from Ex1.rb:3 from Ex1.rb:2:in `times' from Ex1.rb:20123456

37

Synchronizing Threadsc1 = c2 = 0diff = 0t1 = Thread.new do loop do c1 += 1; c2 += 1 endendt2 = Thread.new do loop do diff += (c1 - c2).abs endendsleep 1Thread.critical = trueputs "c1 = #{c1}, c2 = #{c2}, diff = #{diff}"

c1 = 108665, c2 = 108665, diff = 55137

Race Condition

38

Synchronization with Critical Sectionc1 = c2 = 0diff = 0t1 = Thread.new do loop do Thread.critical = true c1 +=1; c2 += 1 Thread.critical = false endendt2 = Thread.new do loop do Thread.critical = true diff += (c1 - c2).abs Thread.critical = false endendputs "c1 = #{c1}, c2 = #{c2}, diff = #{diff}"

c1 = 1568, c2 = 1568, diff = 0

39

Mutual Exclusion (1)require 'thread'mutex = Mutex.newc1 = c2 = 0diff = 0t1 = Thread.new do loop do mutex.synchronize { c1 += 1; c2 += 1 } endendt2 = Thread.new do loop do mutex.synchronize { diff += (c1 - c2).abs } endendsleep 1mutex.lockputs "c1 = #{c1}, c2 = #{c2}, diff = #{diff}"

c1 = 15407, c2 = 15407, diff = 0

A Mutex object acting as a semaphore.

40

Mutual Exclusion (2)require "thread.rb"c1 = c2 = 0diff = 0$mutex = Mutex.newt1 = Thread.new do loop do $mutex.lock c1 +=1; c2 += 1 $mutex.unlock endendt2 = Thread.new do loop do $mutex.lock diff += (c1 - c2).abs $mutex.unlock endend#sleep 1puts "c1 = #{c1}, c2 = #{c2}, diff = #{diff}"

c1 = 1268, c2 = 1268, diff = 0

41

Monitor

Monitor is a fundamental high-level synchronization construct.

Monitor is implemented in Ruby in the form of the monitor.rb library.

A monitor presents a set of programmer defined operations that are provided mutual exclusion within the monitor.

42

Example: Bounded Buffer Monitor 1 # file: MonitorBuffer.rb 2 require "monitor.rb" 3 class Buffer 4 def initialize 5 @buffer = [] 6 @monitor = Monitor.new 7 @empty_condition = @monitor.new_cond 8 end 9 def enter(item) 10 @monitor.synchronize do 11 @buffer.push(item) 12 @empty_condition.signal 13 end 14 end 15 def remove 16 @monitor.synchronize do 17 while @buffer.empty? 18 @empty_condition.wait 19 end 20 return @buffer.shift 21 end 22 end 23 end

Wait until buffer is not empty

A condition variable

43

Example: Bounded Buffer Monitor 25 class BoundedBuffer<Buffer 26 attr :max_size 27 def initialize(max_value) 28 super() 29 @max_size = max_value 30 @full_condition = @monitor.new_cond 31 end 32 def enter(item) 33 @monitor.synchronize do 34 while @buffer.length >= @max_size 35 @full_condition.wait 36 end 37 super(item) 38 end 39 end

44

Example: Bounded Buffer Monitor 40 def remove 41 @monitor.synchronize do 42 item = super 43 if @buffer.length < @max_size 44 @full_condition.signal 45 end 46 return item 47 end 48 end 49 def max_size=(max_value) 50 @monitor.synchronize do 51 @max_size = max_value 52 @full_condition.broadcast 53 end 54 end 55 end

45

Using Bounded Buffer MonitorRequire “MonitorBuffer.rb” buffer = BoundedBuffer.new(5)consumer = Thread.new do loop { puts "Consumer: #{buffer.remove}" sleep rand(2) }endproducer = Thread.new do loop { item = "A"+(rand 10).to_s puts "Producer produces: #{item}" buffer.enter(item) sleep rand(3) }endproducer.joinconsumer.join

Consumer Thread

Producer Thread