Metered concurrency

Metered concurrency
You are encouraged to solve this task according to the task description, using any language you may know.

The goal of this task is to create a counting semaphore used to control the execution of a set of concurrent units. This task intends to demonstrate coordination of active concurrent units through the use of a passive concurrent unit. The operations for a counting semaphore are acquire, release, and count. Each active concurrent unit should attempt to acquire the counting semaphore before executing its assigned duties. In this case the active concurrent unit should report that it has acquired the semaphore. It should sleep for 2 seconds and then release the semaphore.

Ada

Works with: GNAT version GPL 2006

The interface for the counting semaphore is defined in an Ada package specification: <lang ada>package Semaphores is

  protected type Counting_Semaphore(Max : Positive) is
     entry Acquire;
     procedure Release;
     function Count return Natural;
  private
     Lock_Count : Natural := 0;
  end Counting_Semaphore;

end Semaphores;</lang> The Acquire entry has a condition associated with it. A task can only execute the Acquire entry when Lock_Count is less than Max. This is the key to making this structure behave as a counting semaphore. This condition, and all the other aspects of Counting_Semaphore are contained in the package body. <lang ada>package body Semaphores is

  ------------------------
  -- Counting_Semaphore --
  ------------------------ 

  protected body Counting_Semaphore is

     -------------
     -- Acquire --
     -------------

     entry Acquire when Lock_Count < Max is
     begin
        Lock_Count := Lock_Count + 1;
     end Acquire;

     -----------
     -- Count --
     -----------

     function Count return Natural is
     begin
        return Lock_Count;
     end Count;

     -------------
     -- Release --
     -------------

     procedure Release is
     begin
        if Lock_Count > 0 then
           Lock_Count := Lock_Count - 1;
        end if;
     end Release;

  end Counting_Semaphore;

end Semaphores;</lang> We now need a set of tasks to properly call an instance of Counting_Semaphore. <lang ada>with Semaphores; with Ada.Text_Io; use Ada.Text_Io;

procedure Semaphores_Main is

  -- Create an instance of a Counting_Semaphore with Max set to 3
  Lock : Semaphores.Counting_Semaphore(3);

  -- Define a task type to interact with the Lock object declared above
  task type Worker is
     entry Start (Sleep : in Duration; Id : in Positive);
  end Worker;

  task body Worker is
     Sleep_Time : Duration;
     My_Id : Positive;
  begin
     accept Start(Sleep : in Duration; Id : in Positive) do
        My_Id := Id;
        Sleep_Time := Sleep;
     end Start;
     --Acquire the lock. The task will suspend until the Acquire call completes
     Lock.Acquire;
     Put_Line("Task #" & Positive'Image(My_Id) & " acquired the lock.");
     -- Suspend the task for Sleep_Time seconds
     delay Sleep_Time;
     -- Release the lock. Release is unconditional and happens without suspension
     Lock.Release;
  end Worker;

  -- Create an array of 5 Workers
  type Staff is array(Positive range 1..5) of Worker;
  Crew : Staff;

begin

  for I in Crew'range loop
     Crew(I).Start(2.0, I);
  end loop;

end Semaphores_Main;</lang>

ALGOL 68

Works with: ALGOL 68 version Revision 1 - no extensions to language used

Works with: ALGOL 68G version Any - tested with release 1.18.0-9h.tiny

<lang algol68>SEMA sem = LEVEL 1;

PROC job = (INT n)VOID: (

  printf(($" Job "d" acquired Semaphore ..."$,n));
  TO 10000000 DO SKIP OD;
  printf(($" Job "d" releasing Semaphore"l$,n))

);

PAR (

 ( DOWN sem ; job(1) ; UP sem ) ,
 ( DOWN sem ; job(2) ; UP sem ) ,
 ( DOWN sem ; job(3) ; UP sem )

)</lang> Output:

 Job 3 acquired Semaphore ... Job 3 releasing Semaphore
 Job 1 acquired Semaphore ... Job 1 releasing Semaphore
 Job 2 acquired Semaphore ... Job 2 releasing Semaphore

C

Works with: POSIX

<lang c>#include <semaphore.h>

1. include <pthread.h>
2. include <stdlib.h>
3. include <stdio.h>
4. include <unistd.h>

sem_t sem; int count = 3;

1. define getcount() count

void acquire() { sem_wait(&sem); count--; }

void release() { count++; sem_post(&sem); }

void* work(void * id) { int i = 10; while (i--) { acquire(); printf("#%d acquired sema at %d\n", *(int*)id, getcount()); usleep(rand() % 4000000); /* sleep 2 sec on average */ release(); usleep(0); /* effectively yield */ } return 0; }

int main() { pthread_t th[4]; int i, ids[] = {1, 2, 3, 4};

sem_init(&sem, 0, count);

for (i = 4; i--;) pthread_create(th + i, 0, work, ids + i); for (i = 4; i--;) pthread_join(th[i], 0); printf("all workers done\n");

return sem_destroy(&sem); }</lang>

C#

C# has built in semaphore system where acquire is called via Wait(), release with Release() and count with semaphore.CurrentCount. <lang csharp>using System; using System.Threading; using System.Threading.Tasks;

namespace RosettaCode {

 internal sealed class Program
 {
   private static void Worker(object arg, int id)
   {
     var sem = arg as SemaphoreSlim;
     sem.Wait();
     Console.WriteLine("Thread {0} has a semaphore & is now working.", id);
     Thread.Sleep(2*1000);
     Console.WriteLine("#{0} done.", id);
     sem.Release();
   }

   private static void Main()
   {
     var semaphore = new SemaphoreSlim(Environment.ProcessorCount*2, int.MaxValue);

     Console.WriteLine("You have {0} processors availiabe", Environment.ProcessorCount);
     Console.WriteLine("This program will use {0} semaphores.\n", semaphore.CurrentCount);

     Parallel.For(0, Environment.ProcessorCount*3, y => Worker(semaphore, y));
   }
 }

}</lang>

D

<lang d>module meteredconcurrency ; import std.stdio ; import std.thread ; import std.c.time ;

class Semaphore {

 private int lockCnt, maxCnt ;
 this(int count) { maxCnt = lockCnt = count ;}
 void acquire() {
   if(lockCnt < 0 || maxCnt <= 0)
     throw new Exception("Negative Lock or Zero init. Lock") ;
   while(lockCnt == 0)
     Thread.getThis.yield ; // let other threads release lock
   synchronized lockCnt-- ;  
 }
 void release() {
   synchronized 
     if (lockCnt < maxCnt)
       lockCnt++ ;
     else
       throw new Exception("Release lock before acquire") ;    
 }
 int getCnt() { synchronized return lockCnt ; }

}

class Worker : Thread {

 private static int Id = 0 ;
 private Semaphore lock ;
 private int myId ;
 this (Semaphore l) { super() ; lock = l ; myId = Id++ ; }
 override int run() {
   lock.acquire ;  
   writefln("Worker %d got a lock(%d left).", myId, lock.getCnt) ;
   msleep(2000) ;  // wait 2.0 sec
   lock.release ; 
   writefln("Worker %d released a lock(%d left).", myId, lock.getCnt) ;
   return 0 ;
 } 

}

void main() {

 Worker[10] crew ;
 Semaphore lock = new Semaphore(4) ;
 
 foreach(inout c ; crew)
   (c = new Worker(lock)).start ;
 foreach(inout c ; crew)
   c.wait ;

}</lang>

Phobos with tools

Using the scrapple.tools extension library for Phobos .. <lang d>module metered;

import tools.threads, tools.log, tools.time, tools.threadpool;

void main() {

 log_threads = false;
 auto done = new Semaphore, lock = new Semaphore(4);
 auto tp = new Threadpool(10);
 for (int i = 0; i < 10; ++i) {
   tp.addTask(i /apply/ (int i) {
     scope(exit) done.release;
     lock.acquire;
     logln(i, ": lock acquired");
     sleep(2.0);
     lock.release;
     logln(i, ": lock released");
   });
 }
 for (int i = 0; i < 10; ++i)
   done.acquire;

}</lang>

E

This semaphore slightly differs from the task description; the release operation is not on the semaphore itself but given out with each acquisition, and cannot be invoked too many times.

<lang e>def makeSemaphore(maximum :(int > 0)) {

   var current := 0
   def waiters := <elib:vat.makeQueue>()
   def notify() {
       while (current < maximum && waiters.hasMoreElements()) {
           current += 1
           waiters.optDequeue().resolve(def released)
           when (released) -> {
               current -= 1
               notify()
           }
       }
   }
   def semaphore {
       to acquire() {
           waiters.enqueue(def response)
           notify()
           return response
       }
       to count() { return current }
   }
   return semaphore

}

def work(label, interval, semaphore, timer, println) {

   when (def releaser := semaphore <- acquire()) -> {
       println(`$label: I have acquired the lock.`)
       releaser.resolve(
           timer.whenPast(timer.now() + interval, fn {
               println(`$label: I will have released the lock.`)
           })
       )
   }

}

def semaphore := makeSemaphore(3) for i in 1..5 {

   work(i, 2000, semaphore, timer, println)

}</lang>

Euphoria

<lang euphoria>sequence sems sems = {} constant COUNTER = 1, QUEUE = 2

function semaphore(integer n)

   if n > 0 then
       sems = append(sems,{n,{}})
       return length(sems)
   else
       return 0
   end if

end function

procedure acquire(integer id)

   if sems[id][COUNTER] = 0 then
       task_suspend(task_self())
       sems[id][QUEUE] &= task_self()
       task_yield()
   end if
   sems[id][COUNTER] -= 1

end procedure

procedure release(integer id)

   sems[id][COUNTER] += 1
   if length(sems[id][QUEUE])>0 then
       task_schedule(sems[id][QUEUE][1],1)
       sems[id][QUEUE] = sems[id][QUEUE][2..$]
   end if

end procedure

function count(integer id)

   return sems[id][COUNTER]

end function

procedure delay(atom delaytime)

   atom t
   t = time()
   while time() - t < delaytime do
       task_yield()
   end while

end procedure

integer sem

procedure worker()

   acquire(sem)
       printf(1,"- Task %d acquired semaphore.\n",task_self())
       delay(2)
   release(sem)
   printf(1,"+ Task %d released semaphore.\n",task_self())

end procedure

integer task

sem = semaphore(4)

for i = 1 to 10 do

   task = task_create(routine_id("worker"),{})
   task_schedule(task,1)
   task_yield()

end for

while length(task_list())>1 do

   task_yield()

end while</lang>

Output:

- Task 1 acquired semaphore.
- Task 2 acquired semaphore.
- Task 3 acquired semaphore.
- Task 4 acquired semaphore.
+ Task 1 released semaphore.
- Task 5 acquired semaphore.
+ Task 4 released semaphore.
- Task 6 acquired semaphore.
+ Task 3 released semaphore.
- Task 7 acquired semaphore.
+ Task 2 released semaphore.
- Task 8 acquired semaphore.
+ Task 7 released semaphore.
- Task 9 acquired semaphore.
+ Task 6 released semaphore.
- Task 10 acquired semaphore.
+ Task 5 released semaphore.
+ Task 8 released semaphore.
+ Task 10 released semaphore.
+ Task 9 released semaphore.

Go

Counting semaphore implemented here with Go channels. Also WaitGroup used as a completion checkpoint. <lang go>package main

import (

   "log"
   "os"
   "time"
   "sync"

)

var l = log.New(os.Stdout, "", 0)

func main() {

   // three operations per task description
   acquire := make(chan int)
   release := make(chan int)
   count := make(chan chan int)

   // library analogy per WP article
   go librarian(acquire, release, count, 10)
   nStudents := 20
   var studied sync.WaitGroup
   studied.Add(nStudents)
   for i := 0; i < nStudents; i++ {
       go student(acquire, release, &studied)
   }
   // wait until all students have studied before terminating program
   studied.Wait()

}

func librarian(a, r chan int, c chan chan int, count int) {

   p := a // accquire operation is served or not depending on count
   for {
       select {
       case <-p: // accquire/p/wait operation
           count--
           if count == 0 {
               p = nil
           }
       case <-r: // release/v operation
           count++
           p = a
       case cc := <-c: // count operation
           cc <- count
       }
   }

}

func student(a, r chan int, studied *sync.WaitGroup) {

   a <- 0                   // accquire
   l.Println("Studying...") // report per task descrption
   time.Sleep(2e9)          // sleep per task description
   r <- 0                   // release
   studied.Done()           // signal done

}</lang> Output shows 10 students studying immediately, about a 2 second pause, 10 more students studying, then another pause of about 2 seconds before returning to the command prompt.

Haskell

The QSem (quantity semaphore) waitQSem and signalQSem functions are the Haskell acquire and release equivalents, and the MVar (synchronizing variable) functions are used to put the workers statuses on the main thread for printing. Note that this code is likely only compatible with GHC due to the use of "threadDelay" from Control.Concurrent.

<lang Haskell>import Control.Concurrent import Control.Monad

worker :: QSem -> MVar String -> Int -> IO () worker q m n = do

   waitQSem q
   putMVar m $ "Worker " ++ show n ++ " has acquired the lock."
   threadDelay 2000000 -- microseconds!
   signalQSem q
   putMVar m $ "Worker " ++ show n ++ " has released the lock."

main :: IO () main = do

   q <- newQSem 3
   m <- newEmptyMVar
   let workers = 5
       prints  = 2 * workers
   mapM_ (forkIO . worker q m) [1..workers]
   replicateM_ prints $ takeMVar m >>= print</lang>

Java

<lang java>public class CountingSemaphore{

  private int lockCount = 0;
  private int maxCount;

  CountingSemaphore(int Max){
     maxCount = Max;
  }
 
  public synchronized void acquire() throws InterruptedException{
     while( lockCount >= maxCount){
        wait();
     }
     lockCount++;
  }
  public synchronized void release(){
     if (lockCount > 0)
     {
        lockCount--;
        notifyAll();
     }
  }
  public synchronized int getCount(){
     return lockCount;
  }

}

public class Worker extends Thread{

  private CountingSemaphore lock;
  private int id;

  Worker(CountingSemaphore coordinator, int num){
     lock = coordinator;
     id = num;
  }
  Worker(){
  }
  public void run(){
     try{
        lock.acquire();
        System.out.println("Worker " + id + " has acquired the lock.");
        sleep(2000);
     }
     catch (InterruptedException e){
     }
     finally{
        lock.release();
     }
  }
  public static void main(String[] args){
     CountingSemaphore lock = new CountingSemaphore(3);
     Worker crew[];
     crew = new Worker[5];
     for (int i = 0; i < 5; i++){
        crew[i] = new Worker(lock, i);
        crew[i].start();
     }

  }

}</lang>

Oz

Counting semaphores can be implemented in terms of mutexes (called "locks" in Oz) and dataflow variables (used as condition variables here). The mutex protects both the counter and the mutable reference to the dataflow variable. <lang oz>declare

 fun {NewSemaphore N}
    sem(max:N count:{NewCell 0} 'lock':{NewLock} sync:{NewCell _})
 end

 proc {Acquire Sem=sem(max:N count:C 'lock':L sync:S)}
    Sync
    Acquired
 in
    lock L then
       if @C < N then
        C := @C + 1
        Acquired = true
       else
        Sync = @S
        Acquired = false
       end
    end
    if {Not Acquired} then
       {Wait Sync}
       {Acquire Sem}
    end
 end

 proc {Release sem(count:C 'lock':L sync:S ...)}
    lock L then
       C := @C - 1
       @S = unit %% wake up waiting threads
       S := _ %% prepare for new waiters
    end
 end

 proc {WithSemaphore Sem Proc}
    {Acquire Sem}
    try
       {Proc}
    finally
       {Release Sem}
    end
 end

 S = {NewSemaphore 4}

 proc {StartWorker Name}
    thread

for do {WithSemaphore S proc {$} {System.showInfo Name#" acquired semaphore"} {Delay 2000} end } {Delay 100} end

    end
 end

in

 for I in 1..10 do
    {StartWorker I}
 end</lang>

Perl

See Coro::Semaphore.

PicoLisp

<lang PicoLisp>(let Sem (tmp "sem")

  (for U 4  # Create 4 concurrent units
     (unless (fork)
        (ctl Sem
           (prinl "Unit " U " aquired the semaphore")
           (wait 2000)
           (prinl "Unit " U " releasing the semaphore") )
        (bye) ) ) )</lang>

PureBasic

This launches a few threads in parallel, but restricted by the counter. After a thread has completed it releases the Semaphore and a new thread will be able to start. <lang PureBasic>#Threads=10

1. Parallels=3

Global Semaphore=CreateSemaphore(#Parallels)

Procedure Worker(*arg.i)

 WaitSemaphore(Semaphore)
 Debug "Thread #"+Str(*arg)+" active."
 Delay(Random(2000))
 SignalSemaphore(Semaphore)

EndProcedure

Start a multi-thread based work

Dim thread(#Threads) For i=0 To #Threads

 thread(i)=CreateThread(@Worker(),i)

Next Debug "Launcher done."

Wait for all threads to finish before closing down

For i=0 To #Threads

 If IsThread(i)
   WaitThread(i)
 EndIf

Next</lang> Sample output

Thread #0 active.
Thread #2 active.
Thread #4 active.
Launcher done.
Thread #1 active.
Thread #3 active.
Thread #5 active.
Thread #7 active.
Thread #9 active.
Thread #6 active.
Thread #8 active.
Thread #10 active.

Python

Python threading module includes a semaphore implementation. This code show how to use it.

<lang python>import time import threading

1. Only 4 workers can run in the same time

sem = threading.Semaphore(4)

workers = [] running = 1

def worker():

   me = threading.currentThread()
   while 1:
       sem.acquire()
       try:
           if not running:
               break
           print '%s acquired semaphore' % me.getName()
           time.sleep(2.0)
       finally:
           sem.release()
       time.sleep(0.01) # Let others acquire

1. Start 10 workers

for i in range(10):

   t = threading.Thread(name=str(i), target=worker)
   workers.append(t)
   t.start()

1. Main loop

try:

   while 1:
       time.sleep(0.1)

except KeyboardInterrupt:

   running = 0
   for t in workers:
       t.join()</lang>

Raven

Counting semaphores are built in:

<lang raven># four workers may be concurrent 4 semaphore as sem

thread worker

   5 each as i
       sem acquire
       # tid is thread id
       tid "%d acquired semaphore\n" print
       2000 ms
       sem release
       # let others acquire
       100 ms

1. start 10 threads

group

   10 each drop worker

list as workers</lang>

Thread joining is automatic by default.

Tcl

Works with: Tcl version 8.6

Uses the Thread package, which is expected to form part of the overall Tcl 8.6 release. <lang tcl>package require Tcl 8.6 package require Thread

1. Create the global shared state of the semaphore

set handle semaphore0 tsv::set $handle mutex [thread::mutex create] tsv::set $handle cv [thread::cond create] tsv::set $handle count 0 tsv::set $handle max 3

1. Make five worker tasks

for {set i 0} {$i<5} {incr i} {

   lappend threads [thread::create -preserved {

# Not bothering to wrap this in an object for demonstration proc init {handle} { global mutex cv count max set mutex [tsv::object $handle mutex] set cv [tsv::object $handle cv] set count [tsv::object $handle count] set max [tsv::get $handle max] } proc acquire {} { global mutex cv count max thread::mutex lock [$mutex get] while {[$count get] >= $max} { thread::cond wait [$cv get] [$mutex get] } $count incr thread::mutex unlock [$mutex get] } proc release {} { global mutex cv count max thread::mutex lock [$mutex get] if {[$count get] > 0} { $count incr -1 thread::cond notify [$cv get] } thread::mutex unlock [$mutex get] }

       # The core task of the worker

proc run {handle id} { init $handle acquire puts "worker $id has acquired the lock" after 2000 release puts "worker $id is done" }

       # Wait for further instructions from the main thread

thread::wait

   }]

}

1. Start the workers doing useful work, giving each a unique id for pretty printing

set i 0 foreach t $threads {

   puts "starting thread [incr i]"
   thread::send -async $t [list run $handle $i]

}

1. Wait for all the workers to finish

foreach t $threads {

   thread::release -wait $t

}</lang>

UnixPipes

The number of concurrent jobs can be set by issuing that many echo '1s at the begining to sem.

<lang bash>rm -f sem ; mkfifo sem

acquire() {

  x=;while test -z "$x"; do read x; done;

}

release() {

  echo '1'

}

job() {

  n=$1; echo "Job $n acquired Semaphore">&2 ; sleep 2; echo "Job $n released Semaphore">&2 ;

}

( acquire < sem ; job 1 ; release > sem ) & ( acquire < sem ; job 2 ; release > sem ) & ( acquire < sem ; job 3 ; release > sem ) &

echo 'Initialize Jobs' >&2 ; echo '1' > sem</lang>

Visual Basic .NET

This code shows using a local semaphore. Semaphores can also be named, in which case they will be shared system wide.

<lang vbnet>Dim sem As New Semaphore(5, 5) 'Indicates that up to 5 resources can be aquired sem.WaitOne() 'Blocks until a resouce can be aquired Dim oldCount = sem.Release() 'Returns a resource to the pool 'oldCount has the Semaphore's count before Release was called</lang>