Checkpoint synchronization: Difference between revisions

The checkpoint synchronization is a problem of synchronizing multiple [[task]]s. Consider a workshop where several workers ([[task]]s) 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 [[task]]s synchronize themselves before going their paths apart.

 

 

If you can, implement workers joining and leaving.

=={{header|Ada}}==

with Ada.Numerics.Float_Random;

with Ada.Text_IO; use Ada.Text_IO;

end Test_Checkpoint;

 

Sample output:

<pre style="height: 200px;overflow:scroll">

F ends shift

D ends shift

=={{header|BBC BASIC}}==

{{works with|BBC BASIC for Windows}}

nWorkers% = 3

DIM tID%(nWorkers%)

PROC_killtimer(tID%(I%))

NEXT

'''Output:'''

<pre>

Worker 2 starting (5 ticks)

</pre>

=={{header|C}}==

Using OpenMP. Compiled with <code>gcc -Wall -fopenmp</code>.

#include <stdlib.h>

#include <unistd.h>

 

return 0;

 

 

=={{header|C++}}==

{{works with|C++11}}

#include <chrono>

#include <atomic>

for(auto& t: threads) t.join();

std::cout << "Assembly is finished";

<pre>

Worker Mary finished work

Assembly is finished

</pre>

=={{header|Clojure}}==

With a fixed number of workers, this would be very straightforward in Clojure by using a ''CyclicBarrier'' from ''java.util.concurrent''.

So to make it interesting, this version supports workers dynamically joining and parting, and uses the new (2013) ''core.async'' library to use Go-like channels.

Also, each worker passes a value to the checkpoint, so that some ''combine'' function could consume them once they're all received.

(:gen-class)

(:require [clojure.core.async :as async :refer [go <! >! <!! >!! alts! close!]]

(worker ckpt 10 (monitor 2))))

 

=={{header|D}}==

import std.parallelism: taskPool, defaultPoolThreads, totalCPUs;

 

buildMechanism(42);

buildMechanism(11);

{{out|Example output}}

<pre>Build detail 0

Checkpoint reached. Assemble details ...

Mechanism with 11 parts finished: 55</pre>

=={{header|E}}==

 

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 <code>deliver</code> or <code>leave</code> 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.

 

* 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-

waits with= makeWorker(piece, checkpoint)

}

=={{header|Erlang}}==

A team of 5 workers assemble 3 items. The time it takes to assemble 1 item is 0 - 100 milliseconds.

-module( checkpoint_synchronization ).

 

end,

worker_loop( Worker, N - 1, Checkpoint ).

{{out}}

<pre>

Worker 5 item 1

</pre>

 

=={{header|Go}}==

 

 

import (

}

l.Println("worker", id, "leaves shop")

Output:

<pre>worker 1 contracted to assemble 2 details

worker 6 leaves shop

mechanism 5 completed</pre>

=={{header|Haskell}}==

<p>Although not being sure if the approach might be right, this example shows several workers performing a series of tasks simultaneously and synchronizing themselves before starting the next task.</p>

<li>For effectful computations, you should use concurrent threads (forkIO and MVar from the module Control.Concurrent), software transactional memory (STM) or alternatives provided by other modules.</li>

</ul>

 

data Task a = Idle | Make a

 

main = workshop sum tasks

<p>The following version works with the concurrency model provided by the module Control.Concurrent</p>

<p>A workshop is an MVar that holds three values: the number of workers doing something, the number of workers ready for the next task and the total number of workers at the moment.</p>

<p>Other than the parallel version above, this code runs in the IO Monad and makes it possible to perform IO actions such as accessing the hardware. However, all actions must have the return type IO (). If the workers must return some useful values, the MVar should be extended with the necessary fields and the workers should use those fields to store the results they produce.</p>

<p>Note: This code has been tested on GHC 7.6.1 and will most probably not run under other Haskell implementations due to the use of some functions from the module Control.Concurrent. It won't work if compiled with the -O2 compiler switch. Compile with the -threaded compiler switch if you want to run the threads in parallel.</p>

import Control.Monad -- needed for "forM", "forM_"

 

-- kill all worker threads before exit, if they're still running

'''Output:'''

<pre style="height: 200px;overflow:scroll">

The following only works in Unicon:

 

 

procedure main(A)

wait(cv)

}

 

Sample run:

->

</pre>

=={{header|J}}==

 

 

 

=={{header|Java}}==

import java.util.Random;

 

public static int nWorkers = 0;

}

Output:

<pre style="height: 200px;overflow:scroll">

</pre>

{{works with|Java|1.5+}}

import java.util.concurrent.CountDownLatch;

 

}

}

Output:

<pre style="height: 200px;overflow:scroll">Starting task 1

Worker 2 is ready

Task 3 complete</pre>

 

=={{header|Logtalk}}==

The following example can be found in the Logtalk distribution and is used here with permission. It's based on the Erlang solution for this task. Works when using SWI-Prolog, XSB, or YAP as the backend compiler.

:- object(checkpoint).

 

 

:- end_object.

Output:

| ?- checkpoint::run.

Worker 1 item 3

All assemblies done.

yes

 

 

Checkpoint is implemented as a task. It :

 

- And waits for $allDone checkpoint return on its personal channel.

 

while(true) [

System.Out "TASK " << n << " : Beginning my work..." << cr

jobs send($jobDone) drop

myChannel receive drop

 

while(true) [

#[ jobs receive drop ] times(n)

"CHECKPOINT : All jobs done, sending done to all tasks" println

channels apply(#[ send($allDone) drop ])

 

| jobs channels i |

Channel new ->jobs

 

#[ checkPoint(n, jobs, channels) ] &

=={{header|Perl}}==

 

The perlipc man page details several approaches to interprocess communication. Here's one of my favourites: socketpair and fork. I've omitted some error-checking for brevity.

 

use warnings;

use strict;

# workers had terminate, it would need to reap them to avoid zombies:

 

 

A sample run:

msl@64Lucid:~/perl$

</pre>

=={{header|PicoLisp}}==

The following solution implements each worker as a coroutine. Therefore, it

'worker' takes a number of steps to perform. It "works" by printing each step,

and returning NIL when done.

(for P Projects

(prinl "Starting project number " P ":")

(yield ID)

(prinl "Worker " ID " step " N) )

Output:

<pre>: (checkpoints 2 3) # Start two projects with 3 workers

Worker 1 step 4

Project number 2 is done.</pre>

=={{header|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.

 

; Structure that each thread uses

CheckPoint()

Print("Press ENTER to exit"): Input()

<pre style="height: 200px;overflow:scroll">Enter number of workers to use [2-2000]: 5

Work started, 5 workers has been called.

Thread #1 is done.

Press ENTER to exit</pre>

 

=={{header|Racket}}==

This solution uses a double barrier to synchronize the five threads.

The method can be found on page 41 of the delightful book

[http://greenteapress.com/semaphores/downey08semaphores.pdf "The Little Book of Semaphores"] by Allen B. Downey.

#lang racket

(define t 5) ; total number of threads

(displayln (for/list ([_ t]) (channel-get ch)))

(loop))

Output:

(1 4 2 0 3)

(6 9 7 8 5)

(97 98 99 95 96)

...

 

=={{header|Ruby}}==

{{needs-review|Ruby|This code might or might not do the correct task. See comment at [[Talk:{{PAGENAME}}]].}}

 

 

# A Workshop runs all of its workers, then collects their results. Use

# Remove all workers.

wids.each { |wid| shop.remove wid }

 

Example of output: <pre>{23187=>[0, 1346269],

4494=>[5, 4, 1166220]}

{}</pre>

 

 

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.

package require Thread

 

expr {[llength $members] > 0}

}

Demonstration of how this works.

{{trans|Ada}}

foreach worker {A B C D} {

dict set ids $worker [checkpoint makeThread {

break

}

Output:

<pre>

A is ready

B is ready

</pre>

 

The consumer requests a part it doesn't have, waits for a part and puts the received part (which might not be the requested one (if buggy code)) in a bin and assembles the parts into a product.

Repeat until all requested products are made.

var requested=Atomic.Int(-1); // the id of the part the consumer needs

var pipe=Thread.Pipe(); // "conveyor belt" of parts to consumer

foreach n in (NUM_PARTS){ product[n]-=1 } // remove parts from bin

}

An AtomicInt is an integer that does its operations in an atomic fashion. It is used to serialize the producers and consumer.

 

Done

</pre>

{{omit from|ML/I}}