Synchronous concurrency: Difference between revisions

=={{header|Ada}}==

This Ada example starts by creating a package defining a single instance of a printing task. Ada requires packages to be separated into two parts. The package specification defines the interface to all public members of the package.

task Printer is

entry Put(Item : in String);

entry Get_Count(Count : out Natural);

end Printer;

The package body contains the implementation of all the subprograms and tasks defined in the specification.

with Ada.Strings.Unbounded; use Ada.Strings.Unbounded;

end Printer;

Note that the task body contains an ''accept'' block for each ''entry'' defined in the task specification. When some other task calls an entry in the Printer task the communication between the tasks is synchronized.

 

 

The next file contains the ''main'' procedure for this program. The main or entry-point procedure for a program always runs in the ''environment'' task. For this program, the ''environment'' task is takes on the role of the file reading concurrent unit while the Printer task takes on the role of the printing concurrent unit.

with Ada.Text_Io; use Ada.Text_Io;

New_Line;

Put_Line("The task wrote" & Natural'Image(Num_Strings) & " strings.");

In this example only the environment task can call the entries on the Printer task. When the environment task completes the ''terminate'' option of the Printer task applies, terminating the Printer task. The environment task completes right after printing the number of lines sent through the Printer task. Because of the ''terminate'' option, the Printer task terminates just after the environment task prints the count.

 

=={{header|Aikido}}==

Aikido supports threads and monitors natively (built in to the language). There is also support for closures, but this example will use threads:

monitor Queue {

var items = []

 

join (r)

 

=={{header|ALGOL 68}}==

STRING line;

INT count := 0, errno;

);

print((count))

 

=={{header|BCPL}}==

// concurrency test using BCPL coroutines and a coroutine implementation

// of a Occum-style channels.

}

}

 

=={{header|C}}==

{{libheader|libco}}

#include <stdio.h> /* fopen(), fgetc(), fwrite(), printf() */

 

co_delete(reader);

return 0;

 

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

{{works with|C++11}}

#include <iostream>

#include <fstream>

std::thread t2(printer, std::move(promise), std::ref(queue));

t1.join(); t2.join();

{{out}}

<pre>

 

Printed 9 lines

</pre>

 

The ''state'' argument is the agent's state at the start of the call, and the last expression becomes the agent's new state.

 

 

(def writer (agent 0))

(defn write-line [state line]

(println line)

 

The reader executes on the main thread. It passes each line to the writer -- the ''send'' call returns immediately, waits until the writer has finished writing all the lines, gets the line count & prints it, and terminates the writer.

 

(doseq [line (line-seq r)]

(send writer write-line line)))

(await writer)

(println "lines written:" @writer)

That's it!

 

First, implement message-passing:

 

 

(defclass queue ()

`(receive-message (,message . ,body)))

 

 

Should be easy from now on:

 

(with-open-file (stream pathname)

(loop for line = (read-line stream nil)

(thread reader (reader pathname writer))

(thread writer (writer *standard-output* reader)))

 

And now an example:

 

foo

bar

xenu 666

line count: 4

 

Note that to run the example from the SLIME REPL you need to put:

 

in your <code>~/.swank.lisp</code>

 

=={{header|D}}==

 

void main() {

);

}

 

=={{header|Delphi}}==

program Project2;

 

end;

end.

 

=={{header|E}}==

var count := 0

def printer {

 

# Stream IO in E is not finished yet, so this example just uses a list.

 

=={{header|EchoLisp}}==

This task uses two processes, reader/writer, synchronized by a semaphore S and its queue of messages. The reader sends write or reader-count-please messages. The writer responds with ack or count messages.

(require 'sequences)

(require 'tasks)

count)

 

{{out}}

<pre>

=={{header|Elixir}}==

{{trans|Erlang}}

def start do

my_pid = self

end

 

 

=={{header|Erlang}}==

 

 

-export([start/0]).

io:fwrite("~s", [Any]),

reader(Pid, C+1)

 

=={{header|Euphoria}}==

sequence count

lines = {}

while length(task_list()) > 1 do

task_yield()

 

=={{header|F_Sharp|F#}}==

threads from the ThreadPool, using a MailboxProcessor for lock-free communication between threads and tracking the line count without the use of mutable state.

 

open System.IO

 

 

reader printer @"c:\temp\input.txt"

 

=={{header|Go}}==

 

import (

close(lines)

fmt.Println("Number of lines:", <-count)

 

=={{header|Haskell}}==

The code below defines various signals in terms of takeMVar and putMVar and then passes those to the parts of the code which should be permitted to use them. Note that this way, it is impossible for the reader process to take the current line, for example.

 

import Control.Concurrent.MVar

 

 

forkIO $ writer takeLine putCount

 

The reader simply reads the file lazily, applying putLine to each of the lines in turn, which blocks until the writer has taken the line. It then signals that it is finished with putEOF, and then takes the count and prints it.

 

do ls <- fmap lines (readFile "input.txt")

mapM putLine ls

putEOF

n <- takeCount

 

The writer gets the lines in a loop with takeLine until it receives Nothing, at which point it uses putCount to tell the reader how many lines there were.

 

where loop n = do l <- takeLine

case l of

Just x -> do putStrLn x

loop (n+1)

 

=={{header|Icon}} and {{header|Unicon}}==

Unicon supports concurrent programming. A Unicon solution is:

 

fName := A[1]|"index.txt"

p := thread produce(fName)

# (blocking until each message arrives)

i@>>p # put row count into p's inbox

 

Sample output:

 

Implementation:

nlines=: 0

u;._2@fread 'input.txt'

nlines=: nlines+1

smoutput y

 

Usage:

 

 

Note that we are not using OS threads here, but instead - as specified by this task - are using two concurrent activities which share data synchronously.

=={{header|Java}}==

 

import java.io.FileReader;

import java.util.concurrent.atomic.AtomicBoolean;

}

}

 

=={{header|Logtalk}}==

The following example can be found in the Logtalk distribution and is used here with permission. The "input.txt" file contains the terms "a(0)..a(9)", one per line. Works when using SWI-Prolog, XSB, or YAP as the backend compiler.

:- object(team).

 

 

:- end_object.

 

Output:

 

| ?- ?- team::start.

a(0)

Number of lines: 10

true.

 

=={{header|Lua}}==

local fp = io.open( "input.txt" )

assert( fp ~= nil )

print( val )

end

 

=={{header|Mercury}}==

:- interface.

:- import_module io.

LineOrStop = stop,

mvar.put(MVar, Count, !IO)

 

=={{header|OCaml}}==

===Using only the standard library===

We use the built-in Event module to provide communication channels between threads.

 

The reader is a simple loop. It starts by opening a file, then reads lines from that file until there is nothing left to read. After each line, it sends <tt>Some v</tt> on channel <tt>lines_dest</tt>, where <tt>v</tt> is the contents of the line. Once there are no lines anymore,

exception <tt>End_of_file</tt> is raised and we send <tt>None</tt> on that same channel. After that, it's just the matter of waiting for one message on <tt>count_source</tt>, closing the file and printing the result:

let file = open_in "input.txt" in

let rec aux () =

Printf.printf "The task wrote %i strings\n" printed;

close_in file

 

The printer is also a simple loop. It keeps receiving messages on <tt>lines_source</tt>. If a message has structure <tt>Some v</tt>, then <tt>v</tt> is a line, print it and increment the counter. Otherwise, the message has structure <tt>None</tt>, which means that we're done, just send the number of lines on <tt>count_dest</tt>:

let rec aux i =

match sync (receive lines_source) with

| Some line -> print_endline line; aux ( i + 1 )

| None -> sync (send count_target i)

 

Finally, our main program creates both communication channels and backgrounds treatment of <tt>printer</tt>:

let count = new_channel ()

and lines = new_channel ()

in

let _ = Thread.create (printer lines) count

 

Note that, had we decided to background treatment of <tt>reader</tt> instead, an additional synchronization would have been necessary to prevent the program from leaving once the main thread is over.

When the file has been read and sent to channel chPrint, this channel is closed. All tasks waiting for an object in this channel are released and receive null. Then printing task returns number of printed lines into chCount channel for the main task.

 

 

: printing(chPrint, chCount)

aFileName File new forEach: line [ chPrint send(line) drop ]

chPrint close

 

=={{header|ooRexx}}==

queue = .workqueue~new

input = .stream~new("jabberwocky.txt")

stream~lineout(item)

end

 

=={{header|Oz}}==

%% Helper function to read a file lazily.

%% Returns a lazy list of lines.

%% Sync on Count and print its value.

{Wait Count}

 

=={{header|Perl}}==

use Thread::Queue qw();

$reader->join;

 

=={{header|Phix}}==

 

=={{header|PicoLisp}}==

within the same process and not even the same machine. "input.txt" would rather

be a device (like a named pipe or socket) than a plain file.

(task (open "input.txt")

(let Fd @

(sigio (close "Sock")) ) # and stop the task

(println X) # Else print line to stdout

If the two cases of 'sigio' in the printing task are replaced with 'task',

that task would also be synchronous. The resulting behavior is the same.

===Using coroutines===

Coroutines are available only in the 64-bit version.

(yield) # Allow 'unit2' to start

(in "input.txt" # Read the file

(println @) # Print it

(inc 'Cnt) ) # Increment count

 

=={{header|PureBasic}}==

PureBasic uses Semaphores and Mutex's to coordinate threads.

#Write

#Done

WaitThread(Thread1) And WaitThread(Thread2)

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

 

=={{header|Python}}==

Notes: instead of hardcoding the input and output files in the units, each unit is created with a file and read or write the given file.

 

from Queue import Queue

from threading import Thread

writer.start()

reader.join()

 

===Using generators===

{{trans|Ruby}}

 

def reader():

for line in r:

print(line)

 

Note that the above accesses a variable from both paths of execution. To be more in the spirit of the task, we can actually communicate the count from the printer to the reader:

{{works with|Python|2.5+}}

for line in open('input.txt'):

yield line.rstrip()

r.send(count)

except StopIteration:

 

=={{header|Racket}}==

 

 

 

=={{header|Raven}}==

 

class Queue

 

reader as r

 

=={{header|Ruby}}==

The task is to refactor this program to use two concurrent units.

 

IO.foreach("input.txt") { |line| print line; count += 1 }

 

The above program has no concurrency, because the printer is a block <tt>{ |line| print line; count += 1 }</tt> that can print only one line. (The reader calls block more than one time.) After the refactoring, the printer will print all lines, and the program will jump between the reader and the printer. For the refactoring, Fibers might be best, Continuations work if our Ruby is too old for Fibers, and Threads are another alternative.

 

{{works with|Ruby|1.9}}

reader = Fiber.new do

IO.foreach("input.txt") { |line| Fiber.yield line }

print line

count += 1

 

* We create a fiber as the reader, and we use the current fiber as the printer.

 

{{libheader|continuation}}

 

count = 0

print line

count += 1

 

* The above program uses continuations almost like fibers. The reader continues the printer, and the printer continues the reader.

 

{{libheader|thread}}

 

counts = Queue.new

end

end

 

* We create a thread as the reader, and use the current thread as the writer.

=={{header|Rust}}==

 

 

use std::io::BufRead;

 

 

fn main() {

 

let reader = BufReader::new(file);

 

for line in reader.lines() {

match line {

}

}

 

 

 

 

Ok(msg) => {

println!("{}", msg);

}

}

 

 

=={{header|Scala}}==

[[Category:Scala examples needing attention]]

A possible implementation using Actors

 

val printer = actor {

}

 

 

=={{header|Swift}}==

Using Grand Central Dispatch and NSNotificationCenter<br><br>

'''Reader.swift'''

// Reader.swift

//

}

}

'''Printer.swift'''

// Printer.swift

//

}

}

'''main.swift'''

// main.swift

//

 

CFRunLoopRun()

 

=={{header|SystemVerilog}}==

 

 

mailbox#(bit) p2c_cmd = new;

end

 

 

=={{header|Tcl}}==

Uses the Thread package.

 

# Define the input thread

 

# Connect everything together and start the processing

 

=={{header|TXR}}==

To get things going, we resume the producer via <code>pro.(resume)</code>, because we started that in a suspended state. This is actually not necessary; if we remove the <code>suspended t</code> from the <code>new</code> expression which instantiates the producer, we can remove this line. However, this means that the body of the <code>let</code> doesn't receive control. The producer finishes producing and then the <code>pro</code> variable is bound, and the final <code>(put-line ...)</code> expression evaluates. Starting the producer suspended lets us insert some logic prior to dispatching the producer. We implicitly start the consumer, though, because it immediately suspends to wait for an item, saving its context in a continuation and relinquishing control.

 

suspended

cont

(pro (new producer suspended t consumer con)))

pro.(resume)

 

=={{header|UnixPipes}}==

forks.

 

 

=={{header|Visual Basic .NET}}==

This can be improved by adding a blocking Dequeue instead of spinning on TryDequeue.

 

 

Module Module1

End SyncLock

End Sub

 

=={{header|zkl}}==

n:=0; foreach line in (File(fileName)) { out.write(line); n+=1; }

out.close(); // signal done

p:=Thread.Pipe(); // NOT Unix pipes, thread safe channel between threads

reader.launch("input.txt",p);

Light on error handling, easily fixed.

{{out}}