Checkpoint synchronization: Difference between revisions
| Line 342: | Line 342: | ||
<lang Java> |
<lang Java> |
||
import java.util.Scanner; |
import java.util.Scanner; |
||
import |
import java.util.Random; |
||
public class CheckpointSync{ |
public class CheckpointSync{ |
||
| Line 393: | Line 393: | ||
private synchronized void work(){ |
private synchronized void work(){ |
||
try { |
try { |
||
int workTime = rgen.nextInt( |
int workTime = rgen.nextInt(900) + 100; |
||
System.out.println("Worker " + threadID + " will work for " + workTime + " msec."); |
System.out.println("Worker " + threadID + " will work for " + workTime + " msec."); |
||
Thread.sleep(workTime); //work for 'workTime' |
Thread.sleep(workTime); //work for 'workTime' |
||
| Line 425: | Line 425: | ||
/* static variables */ |
/* static variables */ |
||
private static |
private static Random rgen = new Random(); |
||
private static int nFinished = 0; |
private static int nFinished = 0; |
||
public static int nWorkers = 0; |
public static int nWorkers = 0; |
||
Revision as of 07:25, 20 October 2010
You are encouraged to solve this task according to the task description, using any language you may know.
The checkpoint synchronization is a problem of synchronizing multiple tasks. Consider a workshop where several workers (tasks) assembly details of some mechanism. When each of them completes his work they put the details together. There is no store, so a worker who finished its part first must wait for others before starting another one. Putting details together is the checkpoint at which tasks synchronize themselves before going their paths apart.
The task
Implement checkpoint synchronization in your language.
Make sure that the solution is race condition-free. Note that a straightforward solution based on events is exposed to race condition. Let two tasks A and B need to be synchronized at a checkpoint. Each signals its event (EA and EB correspondingly), then waits for the AND-combination of the events (EA&EB) and resets its event. Consider the following scenario: A signals EA first and gets blocked waiting for EA&EB. Then B signals EB and loses the processor. Then A is released (both events are signaled) and resets EA. Now if B returns and enters waiting for EA&EB, it gets lost.
When a worker is ready it shall not continue before others finish. A typical implementation bug is when a worker is counted twice within one working cycle causing its premature completion. This happens when the quickest worker serves its cycle two times while the laziest one is lagging behind.
If you can, implement workers joining and leaving.
Ada
<lang Ada>with Ada.Calendar; use Ada.Calendar; with Ada.Numerics.Float_Random; use Ada.Numerics.Float_Random; with Ada.Text_IO; use Ada.Text_IO;
procedure Test_Checkpoint is
protected Checkpoint is
entry Deliver;
entry Join (Label : out Character);
entry Leave;
private
Signaling : Boolean := False;
Ready_Count : Natural := 0;
Worker_Count : Natural := 0;
Unused_Label : Character := 'A';
entry Lodge;
end Checkpoint;
protected body Checkpoint is
entry Join (Label : out Character) when not Signaling is
begin
Label := Unused_Label;
Unused_Label := Character'Succ (Unused_Label);
Worker_Count := Worker_Count + 1;
end Join;
entry Leave when not Signaling is
begin
Worker_Count := Worker_Count - 1;
end Leave;
entry Deliver when not Signaling is
begin
Ready_Count := Ready_Count + 1;
requeue Lodge;
end Deliver;
entry Lodge when Ready_Count = Worker_Count or Signaling is
begin
Ready_Count := Ready_Count - 1;
Signaling := Ready_Count /= 0;
end Lodge;
end Checkpoint;
task type Worker;
task body Worker is
Dice : Generator;
Label : Character;
Shift_End : Time := Clock + 2.0; -- Trade unions are hard!
begin
Reset (Dice);
Checkpoint.Join (Label);
loop
Put_Line (Label & " is working");
delay Duration (Random (Dice) * 0.200);
Put_Line (Label & " is ready");
Checkpoint.Deliver;
exit when Clock >= Shift_End;
end loop;
Checkpoint.Leave;
end Worker;
Set : array (1..4) of Worker;
begin
null; -- Nothing to do here
end Test_Checkpoint;</lang> Sample output:
A is working B is working C is working D is working A is ready C is ready B is ready D is ready D is working C is working A is working B is working D is ready B is ready A is ready C is ready C is working B is working D is working A is working D is ready A is ready C is ready B is ready D is working A is working C is working B is working A is ready D is ready B is ready C is ready A is working D is working C is working B is working A is ready D is ready C is ready B is ready A is working C is working D is working B is working A is ready C is ready D is ready B is ready A is working D is working C is working B is working D is ready C is ready B is ready A is ready A is working D is working C is working B is working D is ready C is ready B is ready A is ready D is working C is working B is working A is working C is ready B is ready D is ready A is ready C is working B is working D is working A is working C is ready B is ready A is ready D is ready B is working A is working D is working C is working B is ready A is ready D is ready C is ready C is working D is working B is working A is working C is ready D is ready B is ready A is ready C is working D is working B is working A is working C is ready B is ready D is ready A is ready C is working B is working D is working A is working C is ready D is ready B is ready A is ready C is working D is working A is working B is working C is ready D is ready B is ready A is ready C is working D is working A is working B is working C is ready B is ready A is ready D is ready B is working A is working D is working C is working B is ready D is ready C is ready A is ready
E
The problem as stated is somewhat unnatural in E. We would prefer to define the control flow in association with the data flow; for example, such that the workers return values that are combined at the checkpoint; the availability of that result value naturally defines when the workers should proceed with the next round.
That said, here is an implementation of the task as stated. We start by defining a 'flag set' data structure (which is hopefully also useful for other problems), which allows us to express the checkpoint algorithm straightforwardly while being protected against the possibility of a task calling deliver or leave too many times. Note also that each task gets its own reference denoting its membership in the checkpoint group; thus it can only speak for itself and not break any global invariants.
<lang e>/** A flagSet solves this problem: There are N things, each in a true or false
* state, and we want to know whether they are all true (or all false), and be * able to bulk-change all of them, and all this without allowing double- * counting -- setting a flag twice is idempotent. */
def makeFlagSet() {
# Each flag object is either in the true set or the false set.
def trues := [].asSet().diverge()
def falses := [].asSet().diverge()
return def flagSet {
/** Add a flag to the set. */
to join() {
def flag {
/** Get the value of this flag. */
to get() :boolean {
}
/** Set the value of this flag. */
to put(v :boolean) {
def [del,add] := if (v) { [falses,trues] } else { [trues,falses] }
if (del.contains(flag)) {
del.remove(flag)
add.addElement(flag)
}
}
/** Remove this flag from the set. */
to leave() :void {
trues.remove(flag)
falses.remove(flag)
}
}
falses.addElement(flag)
return flag
}
/** Are all the flags true (none false)? */
to allTrue() { return falses.size().isZero() }
/** Are all the flags false (none true)? */
to allFalse() { return trues.size().isZero() }
/** Set all the flags to the same value. */
to setAll(v :boolean) {
def [del,add] := if (v) { [falses,trues] } else { [trues,falses] }
add.addAll(del)
del.removeAll(del)
}
}
}
def makeCheckpoint() {
def [var continueSignal, var continueRes] := Ref.promise()
def readies := makeFlagSet()
/** Check whether all tasks have reached the checkpoint, and if so send the
* signal and go to the next round. */
def check() {
if (readies.allTrue()) {
readies.setAll(false)
continueRes.resolve(null) # send the continue signal
def [p, r] := Ref.promise() # prepare a new continue signal
continueSignal := p
continueRes := r
}
}
return def checkpoint {
to join() {
def &flag := readies.join()
return def membership {
to leave() {
(&flag).leave()
check <- ()
}
to deliver() {
flag := true
check <- ()
return continueSignal
}
}
}
}
}
def makeWorker(piece, checkpoint) {
def stops := timer.now() + 3000 + entropy.nextInt(2000)
var count := 0
def checkpointMember := checkpoint <- join()
def stopped
def run() {
# Pretend to do something lengthy; up to 1000 ms.
timer.whenPast(timer.now() + entropy.nextInt(1000), fn {
if (timer.now() >= stops) {
checkpointMember <- leave()
bind stopped := true
} else {
count += 1
println(`Delivering $piece#$count`)
when (checkpointMember <- deliver()) -> {
println(`Delivered $piece#$count`)
run()
}
}
})
}
run()
return stopped
}
def checkpoint := makeCheckpoint() var waits := [] for piece in 1..5 {
waits with= makeWorker(piece, checkpoint)
} interp.waitAtTop(promiseAllFulfilled(waits))</lang>
Java
<lang Java> import java.util.Scanner; import java.util.Random;
public class CheckpointSync{ public static void main(String[] args){ System.out.print("Enter number of workers to use: "); Scanner in = new Scanner(System.in); Worker.nWorkers = in.nextInt(); System.out.print("Enter number of tasks to complete:"); runTasks(in.nextInt()); }
/* * Informs that workers started working on the task and * starts running threads. Prior to proceeding with next * task syncs using static Worker.checkpoint() method. */ private static void runTasks(int nTasks){ for(int i = 0; i < nTasks; i++){ System.out.println("Starting task number " + (i+1) + "."); runThreads(); Worker.checkpoint(); } }
/* * Creates a thread for each worker and runs it. */ private static void runThreads(){ for(int i = 0; i < Worker.nWorkers; i ++){ new Thread(new Worker(i+1)).start(); } }
/* * Worker inner static class. */ public static class Worker implements Runnable{ public Worker(int threadID){ this.threadID = threadID; } public void run(){ work(); }
/* * Notifies that thread started running for 100 to 1000 msec. * Once finished increments static counter 'nFinished' * that counts number of workers finished their work. */ private synchronized void work(){ try { int workTime = rgen.nextInt(900) + 100; System.out.println("Worker " + threadID + " will work for " + workTime + " msec."); Thread.sleep(workTime); //work for 'workTime' nFinished++; //increases work finished counter System.out.println("Worker " + threadID + " is ready"); } catch (InterruptedException e) { System.err.println("Error: thread execution interrupted"); e.printStackTrace(); } }
/* * Used to synchronize Worker threads using 'nFinished' static integer. * Waits (with step of 10 msec) until 'nFinished' equals to 'nWorkers'. * Once they are equal resets 'nFinished' counter. */ public static synchronized void checkpoint(){ while(nFinished != nWorkers){ try { Thread.sleep(10); } catch (InterruptedException e) { System.err.println("Error: thread execution interrupted"); e.printStackTrace(); } } nFinished = 0; }
/* inner class instance variables */ private int threadID;
/* static variables */ private static Random rgen = new Random(); private static int nFinished = 0; public static int nWorkers = 0; } }</lang> Output:
Enter number of workers to use: 5 Enter number of tasks to complete:3 Starting task number 1. Worker 1 will work for 882 msec. Worker 2 will work for 330 msec. Worker 3 will work for 618 msec. Worker 4 will work for 949 msec. Worker 5 will work for 805 msec. Worker 2 is ready Worker 3 is ready Worker 5 is ready Worker 1 is ready Worker 4 is ready Starting task number 2. Worker 1 will work for 942 msec. Worker 2 will work for 247 msec. Worker 3 will work for 545 msec. Worker 4 will work for 850 msec. Worker 5 will work for 888 msec. Worker 2 is ready Worker 3 is ready Worker 4 is ready Worker 5 is ready Worker 1 is ready Starting task number 3. Worker 2 will work for 976 msec. Worker 1 will work for 194 msec. Worker 4 will work for 532 msec. Worker 3 will work for 515 msec. Worker 5 will work for 326 msec. Worker 1 is ready Worker 5 is ready Worker 3 is ready Worker 4 is ready Worker 2 is ready
PureBasic
PureBasic normally uses Semaphores and Mutex’s to synchronize parallel systems. This system only relies on semaphores between each thread and the controller (CheckPoint-procedure). For exchanging data a Mutex based message stack could easily be added, both synchronized according to this specific task or non-blocking if each worker could be allowed that freedom. <lang PureBasic>#MaxWorktime=8000 ; "Workday" in msec
- Structure that each thread uses
Structure MyIO
ThreadID.i Semaphore_Joining.i Semaphore_Release.i Semaphore_Deliver.i Semaphore_Leaving.i
EndStructure
- Array of used threads
Global Dim Comm.MyIO(0)
- Master loop synchronizing the threads via semaphores
Procedure CheckPoint()
Protected i, j, maxthreads=ArraySize(Comm())
Protected Worker_count, Deliver_count
Repeat
For i=1 To maxthreads
With Comm(i)
If TrySemaphore(\Semaphore_Leaving)
Worker_count-1
ElseIf TrySemaphore(\Semaphore_Deliver)
Deliver_count+1
If Deliver_count=Worker_count
PrintN("All Workers reported in, starting next task.")
Deliver_count=0
For j=1 To maxthreads
SignalSemaphore(Comm(j)\Semaphore_Release)
Next j
EndIf
ElseIf TrySemaphore(\Semaphore_Joining)
PrintN("A new Worker joined the force.")
Worker_count+1: SignalSemaphore(\Semaphore_Release)
ElseIf Worker_count=0
ProcedureReturn
EndIf
Next i
EndWith
ForEver
StartAll=0
EndProcedure
- A worker thread, all orchestrated by the Checkpoint() routine
Procedure Worker(ID)
Protected EndTime=ElapsedMilliseconds()+#MaxWorktime, n
With Comm(ID)
SignalSemaphore(\Semaphore_Joining)
Repeat
Repeat ; Use a non-blocking semaphore check to avoid dead-locking at shutdown.
If ElapsedMilliseconds()>EndTime
SignalSemaphore(\Semaphore_Leaving)
PrintN("Thread #"+Str(ID)+" is done.")
ProcedureReturn
EndIf
Delay(1)
Until TrySemaphore(\Semaphore_Release)
n=Random(1000)
PrintN("Thread #"+Str(ID)+" will work for "+Str(n)+" msec.")
Delay(n): PrintN("Thread #"+Str(ID)+" delivering")
SignalSemaphore(\Semaphore_Deliver)
ForEver
EndWith
EndProcedure
- User IO & init
If OpenConsole()
Define i, j
Repeat
Print("Enter number of workers to use [2-2000]: ")
j=Val(Input())
Until j>=2 And j<=2000
ReDim Comm(j)
For i=1 To j
With Comm(i)
\Semaphore_Release =CreateSemaphore()
\Semaphore_Joining =CreateSemaphore()
\Semaphore_Deliver =CreateSemaphore()
\Semaphore_Leaving =CreateSemaphore()
\ThreadID = CreateThread(@Worker(),i)
EndWith
Next
PrintN("Work started, "+Str(j)+" workers has been called.")
CheckPoint()
Print("Press ENTER to exit"): Input()
EndIf</lang>
Enter number of workers to use [2-2000]: 5 Work started, 5 workers has been called. A new Worker joined the force. A new Worker joined the force. A new Worker joined the force. A new Worker joined the force. A new Worker joined the force. Thread #5 will work for 908 msec. Thread #3 will work for 405 msec. Thread #1 will work for 536 msec. Thread #2 will work for 632 msec. Thread #4 will work for 202 msec. Thread #4 delivering Thread #3 delivering Thread #1 delivering Thread #2 delivering Thread #5 delivering All Workers reported in, starting next task. Thread #2 will work for 484 msec. Thread #4 will work for 836 msec. Thread #5 will work for 464 msec. Thread #3 will work for 251 msec. Thread #1 will work for 734 msec. Thread #3 delivering Thread #5 delivering Thread #2 delivering Thread #1 delivering Thread #4 delivering All Workers reported in, starting next task. Thread #3 will work for 864 msec. Thread #1 will work for 526 msec. Thread #5 will work for 145 msec. Thread #2 will work for 762 msec. Thread #4 will work for 283 msec. Thread #5 delivering Thread #4 delivering Thread #1 delivering Thread #2 delivering Thread #3 delivering All Workers reported in, starting next task. Thread #2 will work for 329 msec. Thread #4 will work for 452 msec. Thread #1 will work for 176 msec. Thread #5 will work for 702 msec. Thread #3 will work for 500 msec. Thread #1 delivering Thread #2 delivering Thread #4 delivering Thread #3 delivering Thread #5 delivering All Workers reported in, starting next task. Thread #5 will work for 681 msec. Thread #3 will work for 71 msec. Thread #2 will work for 267 msec. Thread #1 will work for 151 msec. Thread #4 will work for 252 msec. Thread #3 delivering Thread #1 delivering Thread #4 delivering Thread #2 delivering Thread #5 delivering All Workers reported in, starting next task. Thread #5 will work for 963 msec. Thread #3 will work for 378 msec. Thread #1 will work for 209 msec. Thread #4 will work for 897 msec. Thread #2 will work for 736 msec. Thread #1 delivering Thread #3 delivering Thread #2 delivering Thread #5 delivering Thread #4 delivering All Workers reported in, starting next task. Thread #2 will work for 44 msec. Thread #4 will work for 973 msec. Thread #1 will work for 700 msec. Thread #3 will work for 505 msec. Thread #5 will work for 256 msec. Thread #2 delivering Thread #5 delivering Thread #3 delivering Thread #1 delivering Thread #4 delivering All Workers reported in, starting next task. Thread #2 will work for 703 msec. Thread #4 will work for 296 msec. Thread #1 will work for 702 msec. Thread #3 will work for 99 msec. Thread #5 will work for 114 msec. Thread #3 delivering Thread #5 delivering Thread #4 delivering Thread #1 delivering Thread #2 delivering All Workers reported in, starting next task. Thread #3 will work for 97 msec. Thread #5 will work for 192 msec. Thread #2 will work for 762 msec. Thread #1 will work for 232 msec. Thread #4 will work for 484 msec. Thread #3 delivering Thread #5 delivering Thread #1 delivering Thread #4 delivering Thread #2 delivering All Workers reported in, starting next task. Thread #1 will work for 790 msec. Thread #5 will work for 602 msec. Thread #3 will work for 105 msec. Thread #2 will work for 449 msec. Thread #4 will work for 180 msec. Thread #3 delivering Thread #4 delivering Thread #2 delivering Thread #2 is done. Thread #4 is done. Thread #3 is done. Thread #5 delivering Thread #5 is done. Thread #1 delivering Thread #1 is done. Press ENTER to exit
Tcl
This implementation works by having a separate thread handle the synchronization (inter-thread message delivery already being serialized). The alternative, using a read-write mutex, is more complex and more likely to run into trouble with multi-core machines. <lang tcl>package require Tcl 8.5 package require Thread
namespace eval checkpoint {
namespace export {[a-z]*}
namespace ensemble create
variable members {}
variable waiting {}
variable event
# Back-end of join operation
proc Join {id} {
variable members variable counter if {$id ni $members} { lappend members $id } return $id
}
# Back-end of leave operation
proc Leave {id} {
variable members set idx [lsearch -exact $members $id] if {$idx > -1} { set members [lreplace $members $idx $idx] variable event if {![info exists event]} { set event [after idle ::checkpoint::Release] } } return
}
# Back-end of deliver operation
proc Deliver {id} {
variable waiting lappend waiting $id
variable event if {![info exists event]} { set event [after idle ::checkpoint::Release] } return
}
# Releasing is done as an "idle" action to prevent deadlocks
proc Release {} {
variable members variable waiting variable event unset event if {[llength $members] != [llength $waiting]} return set w $waiting set waiting {} foreach id $w { thread::send -async $id {incr ::checkpoint::Delivered} }
}
# Make a thread and attach it to the public API of the checkpoint proc makeThread Template:Script "" {
set id [thread::create thread::wait] thread::send $id { namespace eval checkpoint { namespace export {[a-z]*} namespace ensemble create
# Call to actually join the checkpoint group proc join {} { variable checkpoint thread::send $checkpoint [list \ ::checkpoint::Join [thread::id]] } # Call to actually leave the checkpoint group proc leave {} { variable checkpoint thread::send $checkpoint [list \ ::checkpoint::Leave [thread::id]] } # Call to wait for checkpoint synchronization proc deliver {} { variable checkpoint # Do this from within the [vwait] to ensure that we're already waiting after 0 [list thread::send $checkpoint [list \ ::checkpoint::Deliver [thread::id]]] vwait ::checkpoint::Delivered } } } thread::send $id [list set ::checkpoint::checkpoint [thread::id]] thread::send $id $script return $id
}
# Utility to help determine whether the checkpoint is in use
proc anyJoined {} {
variable members expr {[llength $members] > 0}
}
}</lang> Demonstration of how this works.
<lang tcl># Build the workers foreach worker {A B C D} {
dict set ids $worker [checkpoint makeThread {
proc task {name} { checkpoint join set deadline [expr {[clock seconds] + 2}] while {[clock seconds] <= $deadline} { puts "$name is working" after [expr {int(500 * rand())}] puts "$name is ready" checkpoint deliver } checkpoint leave thread::release; # Ask the thread to finish }
}]
}
- Set them all processing in the background
dict for {name id} $ids {
thread::send -async $id "task $name"
}
- Wait until all tasks are done (i.e., they have unregistered)
while 1 {
after 100 set s 1; vwait s; # Process events for 100ms
if {![checkpoint anyJoined]} {
break
}
}</lang> Output:
A is working C is working B is working D is working B is ready A is ready D is ready C is ready B is working A is working D is working C is working D is ready A is ready C is ready B is ready B is working D is working A is working C is working D is ready C is ready B is ready A is ready D is working C is working B is working A is working D is ready A is ready C is ready B is ready D is working C is working A is working B is working C is ready A is ready B is ready D is ready