September 2017 Update

Task/Synchronous-concurrency/Nim/synchronous-concurrency.nim

@@ -0,0 +1,33 @@
+var msgs: Channel[string]
+var count: Channel[int]
+
+const FILE = "input.txt"
+
+proc read() {.thread.} =
+  var file = open(FILE)
+  for line in file.lines:
+    msgs.send(line)
+  msgs.send(nil)
+  file.close()
+  echo count.recv()
+  count.close()
+
+proc print() {.thread.} =
+  var n = 0
+  while true:
+    var msg = msgs.recv()
+    if msg == nil:
+      break
+    echo msg
+    n += 1
+  msgs.close()
+  count.send(n)
+
+var reader_thread = Thread[void]()
+var printer_thread = Thread[void]()
+
+msgs.open()
+count.open()
+createThread(reader_thread, read)
+createThread(printer_thread, print)
+joinThreads(reader_thread, printer_thread)

Task/Synchronous-concurrency/OoRexx/synchronous-concurrency.rexx

@@ -0,0 +1,85 @@
+queue = .workqueue~new
+input = .stream~new("jabberwocky.txt")
+output = .output
+
+reader = .filereader~new(input, queue)
+writer = .filewriter~new(output, queue)
+
+::class workQueue
+::method init
+  expose queue stopped actionpending
+  queue = .queue~new
+  stopped = .false
+  actionPending = .false
+
+-- add an item to the work queue.  This is a
+-- guarded method, which means this is a synchronized access
+::method addItem guarded
+  expose queue actionPending
+  use arg item
+  -- add the item to the queue
+  queue~queue(item)
+  -- indicate there's something new.  This is a condition variable
+  -- that any will wake up any thread that's waiting on access.  They'll
+  -- be able to get access once we exit
+  actionPending = .true
+
+-- another method for coordinating access with the other thread.  This indicates
+-- it is time to shut down
+::method stop guarded
+  expose actionPending stopped
+  -- indicate this has been stopped and also flip the condition variable to
+  -- wake up any waiters
+  stopped = .true
+  actionPending = .true
+
+-- read the next item off of the queue.  .nil indicates we've reached
+-- the last item on the queue.  This is also a guarded method, but we'll use
+-- the GUARD ON instruction to wait for work if the queue is currently empty
+::method nextItem
+  expose queue stopped actionPending
+  -- we might need to loop a little to get an item
+  do forever
+    -- if there's something on the queue, pull the front item and return
+    if \queue~isEmpty then return queue~pull
+    -- if the other thread says it is done sending is stuff, time to shut down
+    if stopped then return .nil
+    -- nothing on the queue, not stopped yet, so release the guard and wait until
+    -- there's something pending to work on.
+    guard on when actionPending
+  end
+
+-- one half of the synchronization effort.  This will read lines and
+-- add them to the work queue.  The thread will terminate once we hit end-of-file
+::class filereader
+::method init
+  -- accept a generic stream...the data source need not be a file
+  use arg stream, queue
+
+  reply   -- now multithreaded
+
+  signal on notready
+  loop forever
+     queue~addItem(stream~linein)
+  end
+  -- we come here on an EOF condition.  Indicate we're done and terminate
+  -- the thread
+  notready:
+  queue~stop
+
+-- the other end of this.  This class will read lines from a work queue
+-- and write it to a stream
+::class filewriter
+::method init
+  -- accept a generic stream...the data source need not be a file
+  use arg stream, queue
+
+  reply   -- now multithreaded
+
+  loop forever
+     item = queue~nextItem
+     -- .nil means last item received
+     if item == .nil then return
+     -- write to the stream
+     stream~lineout(item)
+  end

Task/Synchronous-concurrency/Zkl/synchronous-concurrency.zkl

@@ -0,0 +1,15 @@
+fcn reader(fileName,out){
+   n:=0; foreach line in (File(fileName)) { out.write(line); n+=1; }
+   out.close();	// signal done
+   Atomic.waitFor(out.Property("isOpen")); // wait for other thread to reopen Pipe
+   out.write(n);
+}
+fcn writer(in){
+   Utils.zipWith(fcn(n,line){ "%3d: %s".fmt(n,line).print() },[1..],in);
+   in.open();  // signal other thread to send num lines read
+   println("Other thread read ",in.read()," lines");
+}
+
+p:=Thread.Pipe();  // NOT Unix pipes, thread safe channel between threads
+reader.launch("input.txt",p);
+writer.future(p).noop();  // noop forces eval, ie sleep until writer finished