all tasks

Task/Synchronous-concurrency/0DESCRIPTION

@@ -0,0 +1,6 @@
+{{omit from|BBC BASIC}}
+The goal of this task is to create two concurrent activities ("[[Thread|Threads]]" or "Tasks", not [[Process|processes]].) that share data synchronously. Your language may provide syntax or libraries to perform concurrency. Different languages provide different implementations of concurrency, often with different names. Some languages use the term threads, others use the term tasks, while others use co-processes. This task should not be implemented using fork, spawn, or the [[Linux]]/[[UNIX]]/[[Windows|Win32]] pipe command, as communication should be between threads, not processes.
+
+One of the concurrent units will read from a file named "input.txt" and send the contents of that file, one line at a time, to the other concurrent unit, which will print the line it receives to standard output. The printing unit must count the number of lines it prints. After the concurrent unit reading the file sends its last line to the printing unit, the reading unit will request the number of lines printed by the printing unit. The reading unit will then print the number of lines printed by the printing unit.
+
+This task requires two-way communication between the concurrent units. All concurrent units must cleanly terminate at the end of the program.

Task/Synchronous-concurrency/1META.yaml

@@ -0,0 +1,4 @@
+---
+note: Concurrency
+requires:
+- Concurrency

Task/Synchronous-concurrency/ALGOL-68/synchronous-concurrency.alg

@@ -0,0 +1,32 @@
+(
+  STRING line;
+  INT count := 0, errno;
+  BOOL input complete := FALSE;
+  SEMA output throttle = LEVEL 0, input throttle = LEVEL 1;
+<EFBFBD>
+  FILE input txt;
+  errno := open(input txt, "input.txt", stand in channel);
+<EFBFBD>
+  PROC call back done = (REF FILE f) BOOL: ( input complete := TRUE );
+  on logical file end(input txt, call back done);
+<EFBFBD>
+  PAR (
+    WHILE
+      DOWN input throttle;
+      get(input txt,(line, new line));
+      UP output throttle;
+      NOT input complete
+    DO
+      count+:=1
+    OD
+  ,
+    WHILE
+      DOWN output throttle;
+      NOT input complete
+    DO
+      print((line, new line));
+      UP input throttle
+    OD
+  );
+  print((count))
+)

Task/Synchronous-concurrency/Ada/synchronous-concurrency-1.ada

@@ -0,0 +1,6 @@
+package Synchronous_Concurrent is
+   task Printer is
+      entry Put(Item : in String);
+      entry Get_Count(Count : out Natural);
+   end Printer;
+end Synchronous_Concurrent;

Task/Synchronous-concurrency/Ada/synchronous-concurrency-2.ada

@@ -0,0 +1,26 @@
+with Ada.Text_Io; use Ada.Text_Io;
+with Ada.Strings.Unbounded; use Ada.Strings.Unbounded;
+
+package body Synchronous_Concurrent is
+
+   task body Printer is
+      Num_Iter : Natural := 0;
+      Line     : Unbounded_String;
+   begin
+      loop
+         select
+            accept Put(Item : in String) do
+               Line := To_Unbounded_String(Item);
+            end Put;
+            Put_Line(To_String(Line));
+            Num_Iter := Num_Iter + 1;
+         or
+            accept Get_Count(Count : out Natural) do
+               Count := Num_Iter;
+            end Get_Count;
+         or terminate;
+         end select;
+      end loop;
+   end Printer;
+
+end Synchronous_Concurrent;

Task/Synchronous-concurrency/Ada/synchronous-concurrency-3.ada

@@ -0,0 +1,19 @@
+with Synchronous_Concurrent; use Synchronous_Concurrent;
+with Ada.Text_Io; use Ada.Text_Io;
+
+procedure Synchronous_Concurrent_Main is
+   Num_Strings : Natural;
+   The_File : File_Type;
+   Line : String(1..255);
+   Length : Natural;
+begin
+   Open(File => The_File, Mode => In_File, Name => "input.txt");
+   while not End_Of_File(The_File) loop
+      Get_Line(File => The_File, Item => Line, Last => Length);
+      Printer.Put(Line(1..Length));
+   end loop;
+   Close(The_File);
+   Printer.Get_Count(Num_Strings);
+   New_Line;
+   Put_Line("The task wrote" & Natural'Image(Num_Strings) & " strings.");
+end Synchronous_Concurrent_Main;

Task/Synchronous-concurrency/Aikido/synchronous-concurrency.aikido

@@ -0,0 +1,48 @@
+monitor Queue {
+    var items = []
+    public function put (item) {
+        items.append (item)
+        notify()
+    }
+
+    public function get() {
+        while (items.size() == 0) {
+            wait()
+        }
+        var item = items[0]
+        items <<= 1
+        return item
+    }
+
+    public function close {
+        items.append (none)
+    }
+}
+
+thread reader (queue) {
+    var numlines = 0
+    for (;;) {
+        var line = queue.get()
+        if (typeof(line) == "none") {
+            break
+        }
+        print (line)
+        numlines++
+    }
+    println ("Number of lines: " + numlines)
+}
+
+thread writer (queue, lines) {
+    foreach line lines {
+        queue.put (line)
+    }
+    queue.close()
+}
+
+var queue = new Queue()
+var lines = readfile ("input.txt")
+var r = reader(queue)
+var w = writer(queue, lines)
+
+join (r)
+join (w)

Task/Synchronous-concurrency/BCPL/synchronous-concurrency.bcpl

@@ -0,0 +1,139 @@
+// This is a BCPL implementation of the Rosettacode synchronous
+// concurrency test using BCPL coroutines and a coroutine implementation
+// of a Occum-style channels.
+// BCPL is freely available from www.cl.cam.ac.uk/users/mr10
+
+SECTION "coinout"
+
+GET "libhdr.h"
+
+GLOBAL {
+  tracing: ug
+}
+
+LET start() = VALOF
+{ LET argv = VEC 50
+  LET in_co, out_co = 0, 0
+  LET channel = 0
+  LET filename = "input.txt"
+
+  UNLESS rdargs("-f,-t/S", argv, 50) DO
+  { writef("Bad arguments for coinout*n")
+    RESULTIS 0
+  }
+
+  IF argv!0 DO filename := argv!0  // -f     the source file
+  tracing := argv!1                // -t/S   tracing option
+
+  in_co  := initco(infn,  500, @channel)
+  out_co := initco(outfn, 500, @channel)
+
+  UNLESS in_co & out_co DO
+  { writef("Trouble creating the coroutines*n")
+    GOTO fin
+  }
+
+  IF tracing DO writef("*nBoth in and out coroutines created*n*n")
+
+  callco(in_co, filename)
+
+fin:
+  IF in_co  DO deleteco(in_co)
+  IF out_co DO deleteco(out_co)
+
+  IF tracing DO writef("Both in and out coroutines deleted*n*n")
+
+  RESULTIS 0
+}
+
+AND readline(line) = VALOF
+{ LET ch, i = 0, 0
+  line%0 := 0
+
+  { ch := rdch()
+    IF ch=endstreamch RESULTIS FALSE
+    i := i+1
+    line%0, line%i := i, ch
+    IF ch='*n' | i=255 BREAK
+  } REPEAT
+
+  RESULTIS TRUE
+}
+
+AND infn(args) BE
+{ LET channelptr = args!0
+  LET name = cowait()  // Get the file name
+  LET instream = findinput(name)
+  LET line = VEC 256/bytesperword
+
+  UNLESS instream DO
+  { writef("*nTrouble with file: %s*n", name)
+    RETURN
+  }
+
+  selectinput(instream)
+
+  { LET ok = readline(line)
+    UNLESS ok BREAK
+    IF tracing DO
+      writef("inco:  Sending a line to outco*n")
+    cowrite(channelptr, line)
+  } REPEAT
+
+  IF tracing DO
+    writef("inco:  Sending zero to outco*n")
+
+  writef("*nNumber of lines written was %n*n", cowrite(channelptr, 0))
+
+  endstream(instream)
+}
+
+AND outfn(args) BE
+{ LET channelptr = args!0
+  LET linecount = 0
+
+  { LET line = coread(channelptr)
+    UNLESS line BREAK
+    IF tracing DO writef("outfn: Received a line*n")
+    writes(line)
+    linecount := linecount + 1
+  } REPEAT
+
+  IF tracing DO
+    writef("outfn: Received zero, so sent count=%n back to inco*n",
+            linecount)
+
+  cowait(linecount)
+}
+
+// The following functions are a  implementation of Occum-style channels
+// using coroutines.
+
+// The first coroutine to request a transfer through a channel becomes
+// suspended and the second causes the data to be transfers and then allows
+// both coroutines to resume (in some order). The channel word is either
+// zero or points to a suspended (read or write) cocoutine.
+
+// The use of resumeco in coread is somewhat subtle!
+
+AND coread(ptr) = VALOF
+{ LET cptr = !ptr
+  TEST cptr
+  THEN { !ptr := 0         // Clear the channel word
+         RESULTIS resumeco(cptr, currco)
+       }
+  ELSE { !ptr := currco    // Set channel word to this coroutine
+         RESULTIS cowait() // Wait for value from cowrite
+       }
+}
+
+AND cowrite(ptr, val) BE
+{ LET cptr = !ptr
+  TEST cptr
+  THEN { !ptr := 0
+         callco(cptr, val) // Send val to coread
+       }
+  ELSE { !ptr := currco
+          callco(cowait(), val)
+       }
+}

Task/Synchronous-concurrency/C/synchronous-concurrency.c

@@ -0,0 +1,112 @@
+#include <stdlib.h>	/* malloc(), realloc(), free() */
+#include <stdio.h>	/* fopen(), fgetc(), fwrite(), printf() */
+
+#include <libco.h>	/* co_create(), co_switch() */
+
+void
+fail(const char *message) {
+	perror(message);
+	exit(1);
+}
+
+/*
+ * These are global variables of this process. All cothreads of this
+ * process will share these global variables.
+ */
+cothread_t reader;
+cothread_t printer;
+struct {
+	char	*buf;	/* Not a C string. No terminating '\0'. */
+	size_t	len;	/* Length of line in buffer. */
+	size_t	cap;	/* Maximum capacity of buffer. */
+} line;
+size_t count;		/* Number of lines printed. */
+
+/*
+ * The reader cothread reads every line of an input file, passes each
+ * line to the printer cothread, and reports the number of lines.
+ */
+void
+reader_entry(void)
+{
+	FILE *input;
+	size_t newcap;
+	int c, eof, eol;
+	char *newbuf;
+
+	input = fopen("input.txt", "r");
+	if (input == NULL)
+		fail("fopen");
+
+	line.buf = malloc(line.cap = 4096);  /* New buffer. */
+	if (line.buf == NULL)
+		fail("malloc");
+	line.len = 0;  /* Start with an empty line. */
+
+	do {
+		c = fgetc(input);  /* Read next character. */
+		if (ferror(input))
+			fail("fgetc");
+
+		eof = (c == EOF);
+		if (eof) {
+			/*
+			 * End of file is also end of line,
+		`	 * unless the line would be empty.
+			 */
+			eol = (line.len > 0);
+		} else {
+			/* Append c to the buffer. */
+			if (line.len == line.cap) {
+				/* Need a bigger buffer! */
+				newcap = line.cap * 2;
+				newbuf = realloc(line.buf, newcap);
+				if (newbuf == NULL)
+					fail("realloc");
+				line.buf = newbuf;
+				line.cap = newcap;
+			}
+			line.buf[line.len++] = c;
+
+			/* '\n' is end of line. */
+			eol = (c == '\n');
+		}
+
+		if (eol) {
+			/* Pass our complete line to the printer. */
+			co_switch(printer);
+			line.len = 0;  /* Destroy our line. */
+		}
+	} while (!eof);
+
+	free(line.buf);
+	line.buf = NULL;  /* Stops a loop in the printer. */
+
+	printf("Printed %zu lines.\n", count);
+	co_switch(printer);
+}
+
+/*
+ * The printer cothread starts the reader cothread, prints every line
+ * line from the reader cothread, and counts the number of lines.
+ */
+int
+main()
+{
+	reader = co_create(4096, reader_entry);
+	printer = co_active();
+	count = 0;
+
+	for (;;) {
+		co_switch(reader);
+		if (line.buf == NULL)
+			break;
+
+		/* Print this line. Count it. */
+		fwrite(line.buf, 1, line.len, stdout);
+		count++;
+	}
+
+	co_delete(reader);
+	return 0;
+}

Task/Synchronous-concurrency/Clojure/synchronous-concurrency-1.clj

@@ -0,0 +1,7 @@
+(use '[clojure.java.io :as io])
+
+(def writer (agent 0))
+
+(defn write-line [state line]
+  (println line)
+  (inc state))

Task/Synchronous-concurrency/Clojure/synchronous-concurrency-2.clj

@@ -0,0 +1,6 @@
+(with-open [r (io/reader "input.txt")]
+  (doseq [line (line-seq r)]
+    (send writer write-line line)))
+(await writer)
+(println "lines written:" @writer)
+(shutdown-agents)

Task/Synchronous-concurrency/Common-Lisp/synchronous-concurrency-1.lisp

@@ -0,0 +1,59 @@
+(defvar *self*)
+
+(defclass queue ()
+  ((condition :initform (make-condition-variable)
+              :reader condition-of)
+   (mailbox :initform '()
+            :accessor mailbox-of)
+   (lock :initform (make-lock)
+         :reader lock-of)))
+
+(defun message (recipient name &rest message)
+  (with-lock-held ((lock-of recipient))
+    ;; it would have been better to implement tail-consing or a LIFO
+    (setf (mailbox-of recipient)
+          (nconc (mailbox-of recipient)
+                 (list (list* name message))))
+    (condition-notify (condition-of recipient)))
+  message)
+
+(defun mklist (x)
+  (if (listp x)
+      x
+      (list x)))
+
+(defun slurp-message ()
+  (with-lock-held ((lock-of *self*))
+    (if (not (endp (mailbox-of *self*)))
+        (pop (mailbox-of *self*))
+        (progn (condition-wait (condition-of *self*)
+                               (lock-of *self*))
+               (assert (not (endp (mailbox-of *self*))))
+               (pop (mailbox-of *self*))))))
+
+(defmacro receive-message (&body cases)
+  (let ((msg-name (gensym "MESSAGE"))
+        (block-name (gensym "BLOCK")))
+    `(let ((,msg-name (slurp-message)))
+       (block ,block-name
+         ,@(loop for i in cases
+                 for ((name . case) . body) = (cons (mklist (car i))
+                                                    (cdr i))
+                 when (typep i '(or (cons (eql quote)
+                                          t)
+                                    (cons (cons (eql quote) t)
+                                          t)))
+                   do (warn "~S is a quoted form" i)
+                 collect `(when ,(if (null name)
+                                     't
+                                     `(eql ',name (car ,msg-name)))
+                            (destructuring-bind ,case
+                                (cdr ,msg-name)
+                              (return-from ,block-name
+                                (progn ,@body)))))
+         (error "Unknown message: ~S" ,msg-name)))))
+
+(defmacro receive-one-message (message &body body)
+  `(receive-message (,message . ,body)))
+
+(defun queue () (make-instance 'queue))

Task/Synchronous-concurrency/Common-Lisp/synchronous-concurrency-2.lisp

@@ -0,0 +1,33 @@
+(defun reader (pathname writer)
+  (with-open-file (stream pathname)
+    (loop for line = (read-line stream nil)
+          while line
+          do (message writer '|here's a line for you| line)
+          finally
+       (message writer '|how many lines?|)
+       (receive-one-message (|line count| count)
+          (format t "line count: ~D~%" count))
+       (message writer '|looks like i've got no more lines|))))
+
+(defun writer (stream reader)
+  ;; that would work better with ITERATE
+  (loop with line-count = 0 do
+    (receive-message
+     ((|here's a line for you| line)
+      (write-line line stream)
+      (incf line-count))
+     (|looks like i've got no more lines|
+      (return))
+     (|how many lines?|
+      (message reader '|line count| line-count)))))
+
+(defmacro thread (queue &body body)
+  `(make-thread (lambda (&aux (*self* ,queue))
+                  ,@body)))
+
+(defun synchronous-concurrency (&key (pathname "input.txt"))
+  (let ((reader (queue))
+        (writer (queue)))
+    (thread reader (reader pathname writer))
+    (thread writer (writer *standard-output* reader)))
+  (values))

Task/Synchronous-concurrency/Common-Lisp/synchronous-concurrency-3.lisp

@@ -0,0 +1,7 @@
+CL-USER> (synchronous-concurrency :pathname "/tmp/input.txt")
+foo
+bar
+baz
+xenu 666
+line count: 4
+; No value

Task/Synchronous-concurrency/D/synchronous-concurrency.d

@@ -0,0 +1,32 @@
+import tools.threads, std.stdio, std.stream, std.thread;
+
+void main() {
+  // line or EOF
+  struct InputLine {
+    string data;
+    bool eof;
+    static InputLine opCall(string s) { InputLine res; res.data = s; return res; }
+    static InputLine EOF() { InputLine res; res.eof = true; return res; }
+  }
+  auto LineCh = new MessageChannel!(InputLine),
+    ResultCh = new MessageChannel!(int);
+  auto printer = new Thread({
+    int count;
+    while (true) {
+      auto line = LineCh.get();
+      if (line.eof) break;
+      count ++;
+      writefln(count, ": ", line.data);
+    }
+    ResultCh.put(count);
+    return 0;
+  });
+  printer.start;
+  auto file = new File("input.txt");
+  while (!file.eof()) {
+    auto line = file.readLine();
+    LineCh.put(InputLine(line));
+  }
+  LineCh.put(InputLine.EOF());
+  writefln("Count: ", ResultCh.get());
+}

Task/Synchronous-concurrency/Delphi/synchronous-concurrency.delphi

@@ -0,0 +1,168 @@
+program Project2;
+
+{$APPTYPE CONSOLE}
+
+uses
+  SysUtils, Classes, Windows;
+
+type
+  EThreadStackFinalized = class(Exception);
+
+  PLine = ^TLine;
+  TLine = record
+    Text: string;
+  end;
+
+  TThreadQueue = class
+  private
+    FFinalized: Boolean;
+    FQueue: THandle;
+  public
+    constructor Create;
+    destructor Destroy; override;
+    procedure Finalize;
+    procedure Push(Data: Pointer);
+    function Pop(var Data: Pointer): Boolean;
+    property Finalized: Boolean read FFinalized;
+  end;
+
+  TPrintThread = class(TThread)
+  private
+    FCount: Integer;
+    FTreminateEvent: THandle;
+    FDoneEvent: THandle;
+    FQueue: TThreadQueue;
+  public
+    constructor Create(aTreminateEvent, aDoneEvent: THandle; aQueue: TThreadQueue);
+    procedure Execute; override;
+
+    property Count: Integer read FCount;
+  end;
+
+{ TThreadQueue }
+
+constructor TThreadQueue.Create;
+begin
+  FQueue := CreateIOCompletionPort(INVALID_HANDLE_VALUE, 0, 0, 0);
+  FFinalized := False;
+end;
+
+destructor TThreadQueue.Destroy;
+begin
+  if FQueue <> 0 then
+    CloseHandle(FQueue);
+  inherited;
+end;
+
+procedure TThreadQueue.Finalize;
+begin
+  PostQueuedCompletionStatus(FQueue, 0, 0, Pointer($FFFFFFFF));
+  FFinalized := True;
+end;
+
+function TThreadQueue.Pop(var Data: Pointer): Boolean;
+var
+  A: Cardinal;
+  OL: POverLapped;
+begin
+  Result := True;
+  if not FFinalized then
+    GetQueuedCompletionStatus(FQueue, A, Cardinal(Data), OL, INFINITE);
+
+  if FFinalized or (OL = Pointer($FFFFFFFF)) then begin
+    Data := nil;
+    Result := False;
+    Finalize;
+  end;
+end;
+
+procedure TThreadQueue.Push(Data: Pointer);
+begin
+  if FFinalized then
+    raise EThreadStackFinalized.Create('Stack is finalized');
+
+  PostQueuedCompletionStatus(FQueue, 0, Cardinal(Data), nil);
+end;
+
+{ TPrintThread }
+
+constructor TPrintThread.Create(aTreminateEvent, aDoneEvent: THandle; aQueue: TThreadQueue);
+begin
+  inherited Create(True);
+  FCount := 0;
+  FreeOnTerminate := True;
+  FTreminateEvent := aTreminateEvent;
+  FDoneEvent := aDoneEvent;
+  FQueue := aQueue;
+end;
+
+procedure TPrintThread.Execute;
+var
+  data: Pointer;
+  line: PLine;
+begin
+  repeat
+    if FQueue.Pop(data) then begin
+      line := data;
+      try
+        Writeln(line^.Text);
+        if line^.Text = #0 then
+          SetEvent(FDoneEvent);
+        Inc(FCount);
+      finally
+        Dispose(line);
+      end;
+    end;
+
+  until False;
+  WaitForSingleObject(FTreminateEvent, INFINITE);
+end;
+
+var
+  PrintThread: TPrintThread;
+  Queue: TThreadQueue;
+  lines: TStrings;
+  i: Integer;
+  line: PLine;
+  TreminateEvent, DoneEvent: THandle;
+begin
+  Queue := TThreadQueue.Create;
+  try
+    TreminateEvent := CreateEvent(nil, False, False, 'TERMINATE_EVENT');
+    DoneEvent := CreateEvent(nil, False, False, 'DONE_EVENT');
+    try
+      PrintThread := TPrintThread.Create(TreminateEvent, DoneEvent, Queue);
+      PrintThread.Start;
+      lines := TStringList.Create;
+      try
+        lines.LoadFromFile('input.txt');
+        for i := 0 to lines.Count - 1 do begin
+          New(line);
+          line^.Text := lines[i];
+          Queue.Push(line);
+        end;
+
+        New(line);
+        line^.Text := #0;
+        Queue.Push(line);
+
+        WaitForSingleObject(DoneEvent, INFINITE);
+
+        New(line);
+        line^.Text := IntToStr(PrintThread.Count);
+        Queue.Push(line);
+
+        SetEvent(TreminateEvent);
+      finally
+        lines.Free;
+      end;
+    finally
+      CloseHandle(TreminateEvent);
+      CloseHandle(DoneEvent)
+    end;
+
+    Readln;
+  finally
+    Queue.Free;
+  end;
+end.

Task/Synchronous-concurrency/E/synchronous-concurrency.e

@@ -0,0 +1,30 @@
+def printer := {
+    var count := 0
+    def printer {
+        to run(item) {
+            count += 1
+            println(item)
+        }
+        to getCount() {
+            return count
+        }
+    }
+}
+
+def sender(lines) {
+    switch (lines) {
+        match [] {
+            when (def count := printer <- getCount()) -> {
+                println(`$count lines were printed.`)
+            }
+        }
+        match [line] + rest {
+            when (printer <- run(line)) -> {
+                sender(rest)
+            }
+        }
+    }
+}
+
+# Stream IO in E is not finished yet, so this example just uses a list.
+sender(<file:input.txt>.getText().split("\n"))

Task/Synchronous-concurrency/Erlang/synchronous-concurrency.erl

@@ -0,0 +1,35 @@
+-module(cc).
+-export([start/0, reader/2]).
+
+start() ->
+   Pid = spawn(cc,reader,[self(), 0]),
+   case file:open("input.txt", read) of
+       {error, Any} -> io:fwrite("Error ~p~n",[Any]);
+       {ok, Io} ->
+           process(Io, Pid),
+           file:close(Io)
+   end,
+   ok.
+
+process(Io, Pid) ->
+   case io:get_line(Io,"") of
+       eof ->
+           Pid ! count,
+           wait();
+       Any ->
+           Pid ! Any,
+           process(Io, Pid)
+   end.
+
+wait() ->
+   receive
+       I -> io:fwrite("Count:~p~n", [I])
+   end.
+
+reader(Pid, C) ->
+   receive
+       count -> Pid ! C;
+       Any ->
+           io:fwrite("~s", [Any]),
+           reader(Pid, C+1)
+   end.

Task/Synchronous-concurrency/Euphoria/synchronous-concurrency.euphoria

@@ -0,0 +1,58 @@
+sequence lines
+sequence count
+lines = {}
+count = {}
+
+procedure read(integer fn)
+    object line
+    while 1 do
+        line = gets(fn)
+        if atom(line) then
+            exit
+        else
+            lines = append(lines, line)
+            task_yield()
+        end if
+    end while
+
+    lines = append(lines,0)
+    while length(count) = 0 do
+        task_yield()
+    end while
+
+    printf(1,"Count: %d\n",count[1])
+end procedure
+
+procedure write(integer fn)
+    integer n
+    object line
+    n = 0
+    while 1 do
+        while length(lines) = 0 do
+            task_yield()
+        end while
+
+        line = lines[1]
+        lines = lines[2..$]
+        if atom(line) then
+            exit
+        else
+            puts(fn,line)
+            n += 1
+        end if
+    end while
+    count = append(count,n)
+end procedure
+
+integer fn
+atom reader, writer
+constant stdout = 1
+fn = open("input.txt","r")
+reader = task_create(routine_id("read"),{fn})
+writer = task_create(routine_id("write"),{stdout})
+task_schedule(writer,1)
+task_schedule(reader,1)
+
+while length(task_list()) > 1 do
+    task_yield()
+end while

Task/Synchronous-concurrency/Go/synchronous-concurrency.go

@@ -0,0 +1,56 @@
+package main
+
+import (
+    "fmt"
+    "bufio"
+    "io"
+    "os"
+)
+
+// main, one of two goroutines used, will function as the "reading unit"
+func main() {
+    // get file open first
+    f, err := os.Open("input.txt")
+    if err != nil {
+        fmt.Println(err)
+        return
+    }
+    defer f.Close()
+    lr := bufio.NewReader(f)
+
+    // that went ok, now create communication channels,
+    // and start second goroutine as the "printing unit"
+    lines := make(chan string)
+    count := make(chan int)
+    go printer(lines, count)
+
+    for {
+        line, prefix, err := lr.ReadLine()
+        switch {
+        case err == io.EOF:
+        case err != nil:
+            fmt.Println(err)
+        case prefix:
+            fmt.Println("unexpected long line")
+        default:
+            lines <- string(line)
+            continue
+        }
+        break
+    }
+    // this represents the request for the printer to send the count
+    close(lines)
+    // wait for the count from the printer, then print it, then exit
+    fmt.Println("Number of lines:", <-count)
+}
+
+func printer(in <-chan string, count chan<- int) {
+    c := 0
+    // loop as long as in channel stays open
+    for s := range in {
+        fmt.Println(s)
+        c++
+    }
+    // make count available on count channel, then return (terminate goroutine)
+    count <- c
+}

Task/Synchronous-concurrency/Haskell/synchronous-concurrency-1.hs

@@ -0,0 +1,15 @@
+import Control.Concurrent
+import Control.Concurrent.MVar
+
+main =
+    do lineVar <- newEmptyMVar
+       countVar <- newEmptyMVar
+
+       let takeLine  = takeMVar lineVar
+           putLine   = putMVar lineVar . Just
+           putEOF    = putMVar lineVar Nothing
+           takeCount = takeMVar countVar
+           putCount  = putMVar countVar
+
+       forkIO $ writer takeLine putCount
+       reader putLine putEOF takeCount

Task/Synchronous-concurrency/Haskell/synchronous-concurrency-2.hs

@@ -0,0 +1,6 @@
+reader putLine putEOF takeCount =
+    do ls <- fmap lines (readFile "input.txt")
+       mapM putLine ls
+       putEOF
+       n <- takeCount
+       print n

Task/Synchronous-concurrency/Haskell/synchronous-concurrency-3.hs

@@ -0,0 +1,6 @@
+writer takeLine putCount = loop 0
+  where loop n = do l <- takeLine
+                    case l of
+                       Just x  -> do putStrLn x
+                                     loop (n+1)
+                       Nothing -> putCount n

Task/Synchronous-concurrency/Icon/synchronous-concurrency.icon

@@ -0,0 +1,28 @@
+procedure main()
+   local prod, cons
+   prod := create producer("input.txt")
+   cons := create consumer(prod)
+   @cons
+end
+
+procedure producer(fname)
+   local f
+   f := open(fname) | stop("Unable to open ", fname)
+   # send what we read [read(f)] to the consumer (&source)
+   while read(f) @ &source
+   # send it 'null' which we use as a signal to request count
+   write("count = ", &null @ &source)
+end
+
+procedure consumer(p)
+   local value, i
+   i := 1
+   value := @p
+   while \value do {
+      write("=> ",value)
+      value := @ &source
+      i := i + 1
+   }
+   # send producer our count
+   i @ &source
+end

Task/Synchronous-concurrency/Java/synchronous-concurrency.java

@@ -0,0 +1,52 @@
+import java.io.BufferedReader;
+import java.io.FileReader;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+
+class SynchronousConcurrency
+{
+  public static void main(String[] args) throws Exception
+  {
+    final AtomicLong lineCount = new AtomicLong(0);
+    final BlockingQueue<String> queue = new LinkedBlockingQueue<String>();
+    final String EOF = new String();
+
+    final Thread writerThread = new Thread(new Runnable() {
+        public void run()
+        {
+          long linesWrote = 0;
+          while (true)
+          {
+            try
+            {
+              String line = queue.take();
+              // Reference equality
+              if (line == EOF)
+                break;
+              System.out.println(line);
+              linesWrote++;
+            }
+            catch (InterruptedException ie)
+            {  }
+          }
+          lineCount.set(linesWrote);
+        }
+      }
+    );
+    writerThread.start();
+
+    // No need to start a third thread for the reader, just use this thread
+    BufferedReader br = new BufferedReader(new FileReader("input.txt"));
+    String line;
+    while ((line = br.readLine()) != null)
+      queue.put(line);
+    br.close();
+    queue.put(EOF);
+    writerThread.join();
+    // AtomicLong is not needed here due to memory barrier created by thread join, but still need a mutable long since lineCount must be final to access it from an anonymous class
+    System.out.println("Line count: " + lineCount.get());
+    return;
+  }
+}

Task/Synchronous-concurrency/Lua/synchronous-concurrency.lua

@@ -0,0 +1,18 @@
+function ReadFile()
+    local fp = io.open( "input.txt" )
+    assert( fp ~= nil )
+
+    for line in fp:lines() do
+	coroutine.yield( line )
+    end
+
+    fp:close()
+end
+
+co = coroutine.create( ReadFile )
+
+while true do
+    local status, val = coroutine.resume( co )
+    if coroutine.status( co ) == "dead" then break end
+    print( val )
+end

Task/Synchronous-concurrency/OCaml/synchronous-concurrency-1.ocaml

@@ -0,0 +1 @@
+open Event

Task/Synchronous-concurrency/OCaml/synchronous-concurrency-2.ocaml

@@ -0,0 +1,12 @@
+let reader count_source lines_dest =
+  let file = open_in "input.txt" in
+  let rec aux () =
+    let line = try  Some (input_line file)
+               with End_of_file -> None    in
+      sync (send lines_dest line);
+      match line with
+	| Some _ -> aux ()
+	| None   -> let printed = sync (receive count_source) in
+	    Printf.printf "The task wrote %i strings\n" printed;
+	    close_in file
+  in aux ()

Task/Synchronous-concurrency/OCaml/synchronous-concurrency-3.ocaml

@@ -0,0 +1,6 @@
+let printer lines_source count_target =
+  let rec aux i =
+    match sync (receive lines_source) with
+      | Some line -> print_endline line; aux ( i + 1 )
+      | None      -> sync (send count_target i)
+  in aux 0

Task/Synchronous-concurrency/OCaml/synchronous-concurrency-4.ocaml

@@ -0,0 +1,6 @@
+let _ =
+  let count = new_channel ()
+  and lines = new_channel ()
+  in
+  let _ = Thread.create (printer lines) count
+  in reader count lines

Task/Synchronous-concurrency/Oz/synchronous-concurrency.oz

@@ -0,0 +1,49 @@
+declare
+  %% Helper function to read a file lazily.
+  %% Returns a lazy list of lines.
+  fun {ReadLines FN}
+     F = {New class $ from Open.file Open.text end init(name:FN)}
+     fun lazy {ReadNext}
+        case {F getS($)} of
+           false then nil
+        [] Line then
+           Line|{ReadNext}
+        end
+     end
+  in
+     %% close file when handle becomes unreachable
+     {Finalize.register F proc {$ F} {F close} end}
+     {ReadNext}
+  end
+
+  Count %% Will receive the number of lines
+  PrinterPort
+in
+  %% Printer thread
+  thread
+     Stream
+     Counter = {NewCell 0} %% mutable variable
+  in
+     PrinterPort = {NewPort ?Stream}
+     for Line in Stream do
+        case Line of eof then
+           Count = @Counter
+        else
+           {System.showInfo Line}
+           Counter := @Counter + 1
+        end
+     end
+  end
+
+  %% Send all lines to printer thread; make sure that eof is sent.
+  try
+     for Line in {ReadLines "input.txt"} do
+        {Send PrinterPort Line}
+     end
+  finally
+     {Send PrinterPort eof}
+  end
+
+  %% Sync on Count and print its value.
+  {Wait Count}
+  {Show Count}

Task/Synchronous-concurrency/Perl-6/synchronous-concurrency.pl6

@@ -0,0 +1,13 @@
+sub MAIN ($infile) {
+    $infile.IO.lines ==> printer() ==> my $count;
+    say "printed $count lines";
+}
+
+sub printer(*@lines) {
+    my $lines;
+    for @lines {
+	.say;
+	++$lines;
+    }
+    $lines;
+}

Task/Synchronous-concurrency/Perl/synchronous-concurrency.pl

@@ -0,0 +1,33 @@
+use threads;
+use Thread::Queue qw();
+
+my $q1 = Thread::Queue->new;
+my $q2 = Thread::Queue->new;
+
+my $reader = threads->create(sub {
+     my $q1 = shift;
+     my $q2 = shift;
+
+     open my $fh, '<', 'input.txt';
+     $q1->enqueue($_) while <$fh>;
+     close $fh;
+     $q1->enqueue(undef);
+
+     print $q2->dequeue;
+}, $q1, $q2);
+
+my $printer = threads->create(sub {
+     my $q1 = shift;
+     my $q2 = shift;
+
+     my $count;
+     while (my $line = $q1->dequeue) {
+         print $line;
+         $count++;
+     };
+
+     $q2->enqueue($count);
+}, $q1, $q2);
+
+$reader->join;
+$printer->join;

Task/Synchronous-concurrency/PicoLisp/synchronous-concurrency-1.l

@@ -0,0 +1,21 @@
+# Reading task (synchronous)
+(task (open "input.txt")
+   (let Fd @
+      (if (in Fd (line T))             # More lines?
+         (udp "localhost" 4444 @)      # Yes: Send next line
+         (task (port T 4445)           # Else install handler
+            (prinl (udp @) " lines")   # to receive and print count
+            (task (close @)) )
+         (udp "localhost" 4444 T)      # Send 'T' for "Done"
+         (task (close Fd)) ) ) )       # Stop the task
+
+# Printing task (asynchronous)
+(sigio (setq "Sock" (port T 4444))
+   (job '((Cnt . 0))
+      (let? X (udp "Sock")
+         (if (=T X)                    # Done?
+            (prog
+               (udp "localhost" 4445 Cnt) # Yes: Send count
+               (sigio (close "Sock")) )   # and stop the task
+            (println X)                # Else print line to stdout
+            (inc 'Cnt) ) ) ) )         # and increment count

Task/Synchronous-concurrency/PicoLisp/synchronous-concurrency-2.l

@@ -0,0 +1,15 @@
+(co 'unit1
+   (yield)                       # Allow 'unit2' to start
+   (in "input.txt"               # Read the file
+      (while (line T)            # Send each line
+         (yield @ 'unit2) ) )    # to 'unit2'
+   (prinl
+      (yield NIL 'unit2)         # Send 'NIL' for "Done", receive count
+      " lines" ) )
+
+(co 'unit2
+   (let Cnt 0                    # Init counter
+      (while (yield NIL 'unit1)  # Receive line
+         (println @)             # Print it
+         (inc 'Cnt) )            # Increment count
+      (yield Cnt 'unit1) ) )     # Send count to 'unit1'

Task/Synchronous-concurrency/PureBasic/synchronous-concurrency.purebasic

@@ -0,0 +1,45 @@
+Enumeration
+  #Write
+  #Done
+EndEnumeration
+
+Structure commblock
+  txtline.s
+  Order.i
+EndStructure
+
+Global MessageSent=CreateSemaphore()
+Global LineWritten=CreateSemaphore()
+Global LinesWritten, com.commblock
+
+Procedure Writer(arg)
+  Repeat
+    WaitSemaphore(MessageSent)
+    If com\Order=#Write
+      PrintN(com\txtline)
+      LinesWritten+1
+    EndIf
+    SignalSemaphore(LineWritten)
+  Until com\Order=#Done
+EndProcedure
+
+Procedure Reader(arg)
+  Protected File=ReadFile(#PB_Any,OpenFileRequester("","input.txt","",0))
+  While file And Not Eof(file)
+    com\txtline=ReadString(File)
+    com\Order=#Write
+    SignalSemaphore(MessageSent)
+    WaitSemaphore(LineWritten)
+  Wend
+  com\Order=#Done
+  SignalSemaphore(MessageSent)
+  WaitSemaphore(LineWritten)
+  PrintN(Str(LinesWritten)+" lines written.")
+EndProcedure
+
+If OpenConsole()
+  Define Thread1=CreateThread(@Reader(),0)
+  Define Thread2=CreateThread(@Writer(),0)
+  WaitThread(Thread1) And WaitThread(Thread2)
+  Print("Press Enter to exit"):Input()
+EndIf

Task/Synchronous-concurrency/Python/synchronous-concurrency-1.py

@@ -0,0 +1,31 @@
+import sys
+from Queue import Queue
+from threading import Thread
+
+lines = Queue(1)
+count = Queue(1)
+
+def read(file):
+    try:
+        for line in file:
+            lines.put(line)
+    finally:
+        lines.put(None)
+    print count.get()
+
+def write(file):
+    n = 0
+    while 1:
+        line = lines.get()
+        if line is None:
+            break
+        file.write(line)
+        n += 1
+    count.put(n)
+
+reader = Thread(target=read, args=(open('input.txt'),))
+writer = Thread(target=write, args=(sys.stdout,))
+reader.start()
+writer.start()
+reader.join()
+writer.join()

Task/Synchronous-concurrency/Python/synchronous-concurrency-2.py

@@ -0,0 +1,12 @@
+count = 0
+
+def reader():
+    for line in open('input.txt'):
+        yield line.rstrip()
+    print('Printed %d lines.' % count)
+
+r = reader()
+# printer
+for line in r:
+    print(line)
+    count += 1

Task/Synchronous-concurrency/Python/synchronous-concurrency-3.py

@@ -0,0 +1,16 @@
+def reader():
+    for line in open('input.txt'):
+        yield line.rstrip()
+    count = yield None
+    print('Printed %d lines.' % count)
+
+r = reader()
+# printer
+count = 0
+while True:
+    line = next(r)
+    if not line:
+        break
+    print(line)
+    count += 1
+r.send(count)

Task/Synchronous-concurrency/Racket/synchronous-concurrency.rkt

@@ -0,0 +1,29 @@
+(define reader->printer-channel (make-channel))
+(define printer->reader-channel (make-channel))
+
+(define (sync-line-counter filename)
+  (define (reader)
+    (define file-port (open-input-file filename))
+    (let loop ([line (read-line file-port)])
+      (when (not (eof-object? line))
+        (begin
+          (channel-put reader->printer-channel line)
+          (loop (read-line file-port)))))
+    (channel-put reader->printer-channel eof)
+    (let ([num-lines (channel-get printer->reader-channel)])
+      (printf "Number of lines printed = ~a~%" num-lines)))
+
+  (define (printer)
+    (define count 0)
+    (let loop ([line (channel-get reader->printer-channel)])
+      (when (not (eof-object? line))
+        (begin
+          (printf "~a~%" line)
+          (set! count (add1 count))
+          (loop (channel-get reader->printer-channel)))))
+    (channel-put printer->reader-channel count))
+
+  (thread reader)
+  (thread printer))
+
+(sync-line-counter "input.txt")

Task/Synchronous-concurrency/Raven/synchronous-concurrency.raven

@@ -0,0 +1,28 @@
+'input.txt' as src_file
+
+class Queue
+
+    new list  as items
+    condition as ready
+
+    define item_put
+        items push ready notify
+
+    define item_get
+        items empty if ready wait
+        items shift
+
+Queue as lines
+Queue as count
+
+thread reader
+    "file://r:%(src_file)s" open each lines.item_put
+    NULL lines.item_put count.item_get "reader: %d\n" print
+
+thread writer
+    0 repeat lines.item_get dup while
+        "writer: %s" print 1+
+    drop count.item_put
+
+reader as r
+writer as w

Task/Synchronous-concurrency/Ruby/synchronous-concurrency-1.rb

@@ -0,0 +1,3 @@
+count = 0
+IO.foreach("input.txt") { |line| print line; count += 1 }
+puts "Printed #{count} lines."

Task/Synchronous-concurrency/Ruby/synchronous-concurrency-2.rb

@@ -0,0 +1,12 @@
+count = 0
+reader = Fiber.new do
+  IO.foreach("input.txt") { |line| Fiber.yield line }
+  puts "Printed #{count} lines."
+  nil
+end
+
+# printer
+while line = reader.resume
+  print line
+  count += 1
+end

Task/Synchronous-concurrency/Ruby/synchronous-concurrency-3.rb

@@ -0,0 +1,15 @@
+require 'continuation' unless defined? Continuation
+
+count = 0
+reader = proc do |cont|
+  IO.foreach("input.txt") { |line| cont = callcc { |cc| cont[cc, line] }}
+  puts "Printed #{count} lines."
+  cont[nil]
+end
+
+# printer
+while array = callcc { |cc| reader[cc] }
+  reader, line = array
+  print line
+  count += 1
+end

Task/Synchronous-concurrency/Ruby/synchronous-concurrency-4.rb

@@ -0,0 +1,26 @@
+require 'thread'
+
+counts = Queue.new
+lines = Queue.new
+reader = Thread.new do
+  begin
+    File.foreach("input.txt") { |line| lines << line }
+    lines << :EOF
+    puts "Printed #{counts.pop} lines."
+  ensure
+    lines << nil
+  end
+end
+
+# writer
+count = 0
+while line = lines.pop
+  case line
+  when String
+    print line
+    count += 1
+  when :EOF
+    counts << count
+  end
+end
+reader.join

Task/Synchronous-concurrency/Scala/synchronous-concurrency.scala

@@ -0,0 +1,23 @@
+case class HowMany(asker:Actor)
+
+val printer = actor {
+  var count = 0
+    while(true){
+      receive{
+	case line:String => print(line); count = count + 1
+        case HowMany(asker:Actor) => asker ! count ; exit()
+      }
+    }
+}
+
+def reader(printer:Actor) {
+  scala.io.Source.fromFile("c:\\input.txt").getLines foreach {printer ! _ }
+  printer ! HowMany(
+    actor {
+      receive{
+	case count:Int => println("line count = "+count)
+      }
+    })
+}
+
+reader(printer)

Task/Synchronous-concurrency/SystemVerilog/synchronous-concurrency.v

@@ -0,0 +1,37 @@
+program main;
+
+ mailbox#(bit) p2c_cmd = new;
+ mailbox#(string) p2c_data = new;
+ mailbox#(int) c2p_data = new;
+
+ initial begin
+   int fh = $fopen("input.txt", "r");
+   string line;
+   int count;
+   while ($fgets(line, fh)) begin
+     p2c_cmd.put(0);
+     p2c_data.put(line);
+   end
+   p2c_cmd.put(1);
+   c2p_data.get(count);
+   $display( "COUNT: %0d", count );
+ end
+
+ initial begin
+   bit done;
+   int count;
+   while (!done) begin
+     p2c_cmd.get(done);
+     if (done) begin
+       c2p_data.put(count);
+     end
+     else begin
+       string line;
+       p2c_data.get(line);
+       $display( "LINE: %s", line);
+       count++;
+     end
+   end
+ end
+
+endprogram

Task/Synchronous-concurrency/Tcl/synchronous-concurrency.tcl

@@ -0,0 +1,28 @@
+package require Thread
+
+# Define the input thread
+set input [thread::create {
+    proc readFile {filename receiver} {
+	set f [open $filename]
+	while {[gets $f line] >= 0} {
+	    thread::send $receiver [list line $line]
+	}
+	close $f
+	thread::send $receiver lineCount lines
+	puts "got $lines lines"
+    }
+    thread::wait
+}]
+# Define the output thread
+set output [thread::create {
+    set lines 0
+    proc line {string} {
+	puts $string
+	incr ::lines
+    }
+    proc lineCount {} {return $::lines}
+    thread::wait
+}]
+
+# Connect everything together and start the processing
+thread::send $input [list readFile "input.txt" $output]

Task/Synchronous-concurrency/UnixPipes/synchronous-concurrency.unixpipes

@@ -0,0 +1,2 @@
+rm -f node ; mkfifo node
+cat file | tee >(wc -l > node ) | cat - node

Task/Synchronous-concurrency/Visual-Basic-.NET/synchronous-concurrency.visual

@@ -0,0 +1,65 @@
+Imports System.Threading
+
+Module Module1
+
+   Sub Main()
+       Dim t1 As New Thread(AddressOf Reader)
+       Dim t2 As New Thread(AddressOf Writer)
+       t1.Start()
+       t2.Start()
+       t1.Join()
+       t2.Join()
+   End Sub
+
+   Sub Reader()
+       For Each line In IO.File.ReadAllLines("input.txt")
+           m_WriterQueue.Enqueue(line)
+       Next
+       m_WriterQueue.Enqueue(Nothing)
+
+       Dim result As Integer
+       Do Until m_ReaderQueue.TryDequeue(result)
+           Thread.Sleep(10)
+       Loop
+
+       Console.WriteLine(result)
+
+   End Sub
+
+   Sub Writer()
+       Dim count = 0
+       Dim line As String = Nothing
+       Do
+           Do Until m_WriterQueue.TryDequeue(line)
+               Thread.Sleep(10)
+           Loop
+           If line IsNot Nothing Then
+               Console.WriteLine(line)
+               count += 1
+           Else
+               m_ReaderQueue.Enqueue(count)
+               Exit Do
+           End If
+       Loop
+   End Sub
+
+   Private m_WriterQueue As New SafeQueue(Of String)
+   Private m_ReaderQueue As New SafeQueue(Of Integer)
+
+End Module
+
+Class SafeQueue(Of T)
+   Private m_list As New Queue(Of T)
+   Public Function TryDequeue(ByRef result As T) As Boolean
+       SyncLock m_list
+           If m_list.Count = 0 Then Return False
+           result = m_list.Dequeue
+           Return True
+       End SyncLock
+   End Function
+   Public Sub Enqueue(ByVal value As T)
+       SyncLock m_list
+           m_list.Enqueue(value)
+       End SyncLock
+   End Sub
+End Class