Rendezvous

From Rosetta Code
Revision as of 22:49, 7 May 2011 by rosettacode>CRGreathouse ({{omit from|PARI/GP}})
Task
Rendezvous
You are encouraged to solve this task according to the task description, using any language you may know.

Demonstrate the “rendezvous” communications technique by implementing a printer monitor.

Detailed Description of Programming Task

Rendezvous is a synchronization mechanism based on procedural decomposition. Rendezvous is similar to a procedure call with the difference that the caller and the callee belong to different tasks. The called procedure is usually called an entry point of the corresponding task. A call to an entry point is synchronous, i.e. the caller is blocked until completion. For the caller a call to the entry point is indivisible. Internally it consists of:

  • Waiting for the callee ready to accept the rendezvous;
  • Engaging the rendezvous (servicing the entry point).

The caller may limit the waiting time to the callee to accept the rendezvous. I.e. a rendezvous request can be aborted if not yet accepted by the callee. When accepted the rendezvous is processed until its completion. During this time the caller and the callee tasks stay synchronized. Which context is used to process the rendezvous depends on the implementation which may wish to minimize context switching.

The callee task may accept several rendezvous requests:

  • Rendezvous to the same entry point from different tasks;
  • Rendezvous to different entry points.

The callee accepts one rendezvous at a time.

Language mechanism of exceptions (if any) has to be consistent with the rendezvous. In particular when an exception is propagated out of a rendezvous it shall do in both tasks. The exception propagation is synchronous within the rendezvous and asynchronous outside it.

An engaged rendezvous can be requeued by the callee to another entry point of its task or to another task, transparently to the caller.

Differently to messages which are usually asynchronous, rendezvous are synchronous, as it was stated before. Therefore a rendezvous does not require marshaling the parameters and a buffer to keep them. Further, rendezvous can be implemented without context switch. This makes rendezvous a more efficient than messaging.

Rendezvous can be used to implement monitor synchronization objects. A monitor guards a shared resource. All users of the resource request a rendezvous to the monitor in order to get access to the resource. Access is granted by accepting the rendezvous for the time while the rendezvous is serviced.

Language task

Show how rendezvous are supported by the language. If the language does not have rendezvous, provide an implementation of them based on other primitives.

Use case task

Implement a printer monitor. The monitor guards a printer. There are two printers main and reserve. Each has a monitor that accepts a rendezvous Print with a text line to print of the printer. The standard output may serve for printing purpose. Each character of the line is printed separately in order to illustrate that lines are printed indivisibly. Each printer has ink for only 5 lines of text. When the main printer runs out of ink it redirects its requests to the reserve printer. When that runs out of ink too, Out_Of_Ink exception propagates back to the caller. Create two writer tasks which print their plagiarisms on the printer. One does Humpty Dumpty, another Mother Goose.

Ada

Ada has integrated rendezvous support. The caller calls to a rendezvous using the name of the task suffixed by the entry point name and the parameters. An entry point can be called using timed entry call statement which allow limit waiting time: <lang ada>select

  Server.Wake_Up (Parameters);

or delay 5.0;

  -- No response, try something else
  ...

end select;</lang> The task accepts a rendezvous using accept statement. The statement can contain body which implements the rendezvous. When several rendezvous need to be accepted a selective accept statement can be used. For example: <lang ada>select

  accept Wake_Up (Parameters : Work_Item) do
     Current_Work_Item := Parameters;
  end;
  Process (Current_Work_Item);

or accept Shut_Down;

  exit;       -- Shut down requested

end select;</lang> Entry points in the selective accept can be guarded by Boolean expressions which close the entry point when the expression yield false.

A task may requeue rendezvous request from the body of an accept statement to an entry point of the same or another task if the parameter profile of the entry point is compatible. The requeue statement may contain clause 'with abort which allows the caller to abort the request when it waits for other task to accept it. Without the clause the request is protected from abortion. This might be useful when the first task initiates processing of the request and the side effect of this action need to be removed when processing is completed.

The task

<lang ada>with Ada.Text_IO; use Ada.Text_IO;

procedure Rendezvous is

  Out_Of_Ink : exception;
  type Printer;
  type Printer_Ptr is access all Printer;
  task type Printer (ID : Natural; Backup : Printer_Ptr) is
     entry Print (Line : String);
  end Printer;
  task body Printer is
     Ink : Natural := 5;
  begin
     loop
        begin
           select
              accept Print (Line : String) do
                 if Ink = 0 then
                    if Backup = null then
                       raise Out_Of_Ink;
                    else
                       requeue Backup.Print with abort;
                    end if;
                 else
                    Put (Integer'Image (ID) & ": ");
                    for I in Line'Range loop
                       Put (Line (I));
                    end loop;
                    New_Line;
                    Ink := Ink - 1;
                 end if;
              end Print;
           or terminate;
           end select;
        exception
           when Out_Of_Ink =>
              null;
        end;
     end loop;
  end Printer;
  Reserve : aliased Printer (2, null);
  Main    : Printer (1, Reserve'Access);
  
  task Humpty_Dumpty;
  task Mother_Goose;
  
  task body Humpty_Dumpty is
  begin
     Main.Print ("Humpty Dumpty sat on a wall.");
     Main.Print ("Humpty Dumpty had a great fall.");
     Main.Print ("All the king's horses and all the king's men");
     Main.Print ("Couldn't put Humpty together again.");
  exception
     when Out_Of_Ink =>
        Put_Line ("      Humpty Dumpty out of ink!");
  end Humpty_Dumpty;
  task body Mother_Goose is
  begin
     Main.Print ("Old Mother Goose");
     Main.Print ("When she wanted to wander,");
     Main.Print ("Would ride through the air");
     Main.Print ("On a very fine gander.");
     Main.Print ("Jack's mother came in,");
     Main.Print ("And caught the goose soon,");
     Main.Print ("And mounting its back,");
     Main.Print ("Flew up to the moon.");
  exception
     when Out_Of_Ink =>
        Put_Line ("      Mother Goose out of ink!");
  end Mother_Goose;

begin

  null;   

end Rendezvous;</lang> Sample output:

 1: Old Mother Goose
 1: Humpty Dumpty sat on a wall.
 1: When she wanted to wander,
 1: Humpty Dumpty had a great fall.
 1: Would ride through the air
 2: All the king's horses and all the king's men
 2: On a very fine gander.
 2: Couldn't put Humpty together again.
 2: Jack's mother came in,
 2: And caught the goose soon,
      Mother Goose out of ink!

AutoHotkey

<lang AutoHotkey>OnMessage(0x4a, "PrintMonitor") SetTimer, print2, 400

print1:

 print("Old Mother Goose")
 print("When she wanted to wander,")
 print("Would ride through the air")
 print("On a very fine gander.")
 print("Jack's mother came in,")
 print("And caught the goose soon,")
 print("And mounting its back,")
 print("Flew up to the moon.")

Return

print2:

 SetTimer, print2, Off
 print("Humpty Dumpty sat on a wall.")
 print("Humpty Dumpty had a great fall.")
 print("All the king's horses and all the king's men")
 print("Couldn't put Humpty together again.")

Return

print(message) {

 Static StringToSend
 StringToSend := message
 Gui +LastFound
 VarSetCapacity(CopyDataStruct, 12, 0) 
 NumPut(StrLen(StringToSend) + 1, CopyDataStruct, 4)
 NumPut(&StringToSend, CopyDataStruct, 8) 
 SendMessage, 0x4a, 0, &CopyDataStruct
 If ErrorLevel
   MsgBox out of ink
 Sleep, 200
 Return

}

PrintMonitor(wParam, lParam, msg) {

 Static ink = 5
 Global printed
 Critical
 If ink
 {
   StringAddress := NumGet(lParam + 8)   
   StringLength := DllCall("lstrlen", UInt, StringAddress)
   VarSetCapacity(CopyOfData, StringLength)
   DllCall("lstrcpy", "str", CopyOfData, "uint", StringAddress)   
   printed .= "primaryprinter: " . CopyOfData . "`n"
   ToolTip, primary printer`n: %printed%
   ink--
 }
 Else
 {
   OnMessage(0x4a, "Reserve")
   print(CopyOfData)
 }

}

Reserve(wParam, lParam, msg) {

 Static ink = 5
 Global printed
 Critical
 If ink
 {
   StringAddress := NumGet(lParam + 8)   
   StringLength := DllCall("lstrlen", UInt, StringAddress)
   VarSetCapacity(CopyOfData, StringLength)
   DllCall("lstrcpy", "str", CopyOfData, "uint", StringAddress)   
   printed .= "reserveprinter: " . CopyOfData . "`n"
   ToolTip, Reserve printer`n: %printed%
   ink--
 }
 Else
   Return -1

}</lang>

F#

The rendezvous mechanism is realized by using F#'s mailbox processors to implement active objects.

It is possible to extract the boilerplate code into a reusable helper class which should be considered when using active objects a lot.

<lang fsharp>open System

type PrinterCommand = Print of string

// a message is a command and a facility to return an exception type Message = Message of PrinterCommand * AsyncReplyChannel<Exception option>

// thrown if we have no more ink (and neither has our possible backup printer) exception OutOfInk

type Printer(id, ?backup:Printer) =

  let mutable ink = 5
  
  // the actual printing logic as a private function
  let print line =
     if ink > 0 then
        printf "%d: " id
        Seq.iter (printf "%c") line
        printf "\n" 
        ink <- ink - 1
     else
        match backup with
        | Some p -> p.Print line
        | None -> raise OutOfInk
  // use a MailboxProcessor to process commands asynchronously;
  // if an exception occurs, we return it to the calling thread
  let agent = MailboxProcessor.Start( fun inbox ->
     async {
        while true do
           let! Message (command, replyChannel) = inbox.Receive()
           try
              match command with
              | Print line -> print line
              replyChannel.Reply None
           with
              | ex -> replyChannel.Reply (Some ex)
     })
  // public printing method:
  // send Print command and propagate exception if one occurs
  member x.Print line = 
     match agent.PostAndReply( fun replyChannel -> Message (Print line, replyChannel) ) with
     | None -> ()
     | Some ex -> raise ex


open System.Threading

do

 let main = new Printer(id=1, backup=new Printer(id=2))
 (new Thread(fun () -> 
     try
       main.Print "Humpty Dumpty sat on a wall."
       main.Print "Humpty Dumpty had a great fall."
       main.Print "All the king's horses and all the king's men"
       main.Print "Couldn't put Humpty together again."
     with
       | OutOfInk -> printfn "      Humpty Dumpty out of ink!"
 )).Start()
 (new Thread(fun () -> 
     try
       main.Print "Old Mother Goose"
       main.Print "Would ride through the air"
       main.Print "On a very fine gander."
       main.Print "Jack's mother came in,"
       main.Print "And caught the goose soon,"
       main.Print "And mounting its back,"
       main.Print "Flew up to the moon."
     with
       | OutOfInk -> printfn "      Mother Goose out of ink!"
 )).Start()
 Console.ReadLine() |> ignore</lang>

Example output:

1: Old Mother Goose
1: Humpty Dumpty sat on a wall.
1: Would ride through the air
1: Humpty Dumpty had a great fall.
1: On a very fine gander.
2: All the king's horses and all the king's men
2: Jack's mother came in,
2: Couldn't put Humpty together again.
2: And caught the goose soon,
2: And mounting its back,
      Mother Goose out of ink!


Go

Pretty sure printing could be done better ways in Go, but I stayed as close to the task description as I could.

Back-to-back synchronous send and receive, for example, achieves the rendezvous property of one task suspending while the other executes. Passing an interface means only a reference to data is passed, the underlying data is not passed on a channel or moved in any way. rSync and it's rendezvous method are defined in a way that is completely general and not specific to this task description. The only nod to exception handling here is the rendezvous response being an interface, which can reference a value or object of any type. <lang go>package main

import (

   "fmt"
   "strings"
   "sync"

)

var hdText = `Humpty Dumpty sat on a wall. Humpty Dumpty had a great fall. All the king's horses and all the king's men, Couldn't put Humpty together again.`

var mgText = `Old Mother Goose, When she wanted to wander, Would ride through the air, On a very fine gander. Jack's mother came in, And caught the goose soon, And mounting its back, Flew up to the moon.`

func main() {

   reservePrinter := startMonitor(newPrinter(5), nil)
   mainPrinter := startMonitor(newPrinter(5), reservePrinter)
   var busy sync.WaitGroup
   busy.Add(2)
   go writer(mainPrinter, "hd", hdText, &busy)
   go writer(mainPrinter, "mg", mgText, &busy)
   busy.Wait()

}

func newPrinter(ink int) func(string) string {

   return func(line string) string {
       if ink == 0 {
           return "out of ink"
       }
       for _, c := range line {
           fmt.Printf("%c", c)
       }
       fmt.Println("")
       ink--
       return ""
   }

}

type rSync struct {

   call     chan interface{}
   response chan interface{}

}

func (r *rSync) rendezvous(data interface{}) interface{} {

   r.call <- data
   return <-r.response

}

func startMonitor(printer func(string) string, reservePrinter *rSync) *rSync {

   r := &rSync{make(chan interface{}), make(chan interface{})}
   go monitor(printer, r, reservePrinter)
   return r

}

func monitor(print func(string) string, entry, reserve *rSync) {

   for {
       data := <-entry.call
       err := print(data.(string))
       switch {
       case err == "":
           entry.response <- nil
       case err == "out of ink" && reserve != nil:
           entry.response <- reserve.rendezvous(data)
       default:
           entry.response <- err
       }
   }

}

func writer(printer *rSync, id, text string, busy *sync.WaitGroup) {

   for _, line := range strings.Split(text, "\n", -1) {
       if e := printer.rendezvous(line); e != nil {
           fmt.Println("**** print job", id, "terminated:", e, "****")
           break
       }
   }
   busy.Done()

}</lang> Output:

Old Mother Goose,
Humpty Dumpty sat on a wall.
Humpty Dumpty had a great fall.
When she wanted to wander,
All the king's horses and all the king's men,
Would ride through the air,
Couldn't put Humpty together again.
On a very fine gander.
Jack's mother came in,
And caught the goose soon,
**** print job mg terminated: out of ink ****

Oz

Oz does not have a rendezvous mechanism, but the task description lends itself to synchronous active objects. We show how to implement this in Oz and then discuss the differences to the rendezvous model.

First a simple printer class whose definition is completely orthogonal to multithreading issues: <lang oz>declare

 class Printer
    attr ink:5
 
    feat id backup
 
    meth init(id:ID backup:Backup<=unit)
       self.id = ID
       self.backup = Backup
    end
 
    meth print(Line)=Msg
       if @ink == 0 then
          if self.backup == unit then
             raise outOfInk end
          else
             {self.backup Msg}
          end
       else
          {System.printInfo self.id#": "}
          for C in Line do
             {System.printInfo [C]}
          end
          {System.printInfo "\n"}
          ink := @ink - 1
       end
    end
 end</lang>

Note how requeuing the task simply becomes delegation to a different object.

Active object are not a predefined abstraction in Oz. But due to Oz' first-class object messages, we can easily define it using ports and streams (many-to-one message passing): <lang oz> fun {NewActiveSync Class Init}

    Obj = {New Class Init}
    MsgPort
 in
    thread MsgStream in
       {NewPort ?MsgStream ?MsgPort}
       for Msg#Sync in MsgStream do
          try
             {Obj Msg}
             Sync = unit
          catch E then
             Sync = {Value.failed E}
          end
       end
    end
    proc {$ Msg}
       Sync = {Port.sendRecv MsgPort Msg}
    in
       {Wait Sync}
    end
 end</lang>

This functions takes a class and an initialization message and returns a procedure. When called, this procedure will send messages to the new object in a new thread and then wait for the Sync variable to become bound. Exceptions are propagated using failed values.

This works because a unary procedure is syntactically indistinguishable from an object in Oz.

With this new abstraction we can create the two printers and execute both print tasks in their own thread: <lang oz> Main = {NewActiveSync Printer init(id:1 backup:Reserve)}

 Reserve = {NewActiveSync Printer init(id:2)}

in

 %% task Humpty Dumpty
 thread
    try

{Main print("Humpty Dumpty sat on a wall.")} {Main print("Humpty Dumpty had a great fall.")} {Main print("All the king's horses and all the king's men")} {Main print("Couldn't put Humpty together again.")}

    catch outOfInk then

{System.showInfo " Humpty Dumpty out of ink!"}

    end
 end
 %% task Mother Goose
 thread
    try

{Main print("Old Mother Goose")} {Main print("When she wanted to wander,")} {Main print("Would ride through the air")} {Main print("On a very fine gander.")} {Main print("Jack's mother came in,")} {Main print("And caught the goose soon,")} {Main print("And mounting its back,")} {Main print("Flew up to the moon.")}

    catch outOfInk then

{System.showInfo " Mother Goose out of ink!"}

    end
 end</lang>

Example output:

1: Humpty Dumpty sat on a wall.
1: Old Mother Goose
1: Humpty Dumpty had a great fall.
1: When she wanted to wander,
1: All the king's horses and all the king's men
2: Would ride through the air
2: Couldn't put Humpty together again.
2: On a very fine gander.
2: Jack's mother came in,
2: And caught the goose soon,
      Mother Goose out of ink!

Comparison to Ada

What is called an "entry point" in Ada, is a method in our implementation.

We cannot limit the waiting time. This could be implemented in the NewActiveSync function, but it is not needed in the example task.

The callee task accepts rendezvous requests to the same entry point from multiple caller tasks. Similarly, the active object in Oz is designed to accept messages from different threads. To implement "rendezvous to different entry points", we simply add public methods to the Printer class.

The callee in ADA accepts one rendezvous at a time. The active object in Oz reads one message at a time from the stream.

Messages cannot be requeued in the given implementation. But we can delegate to a different active object which has the same effect, at least for the example given.

Like in the rendezvous mechanism, parameters are not marshalled. This is because sharing immutable data between threads is safe. In contrast to ADA, the parameters are buffered until the printer becomes ready. But with a synchronous communication mechanism, this should not cause problems.

PicoLisp

Rendezvous can be implemented in PicoLisp via the following function: <lang PicoLisp>(de rendezvous (Pid . Exe)

  (when
     (catch '(NIL)
        (tell Pid 'setq 'Rendezvous (lit (eval Exe)))
        NIL )
     (tell Pid 'quit @) ) )  # Raise caught error in caller</lang>

The caller invokes it in the callee via the 'tell' interprocess communication, and it uses 'tell' in turn to communicate results (and possible errors) back to the caller.

Use case task: <lang PicoLisp>(de printLine (Str)

  (cond
     ((gt0 *Ink) (prinl *ID ": " Str) (dec '*Ink))
     (*Backup (rendezvousPrint @ Str) T)
     (T (quit "Out of Ink")) ) )

(de rendezvousPrint (Printer Str)

  (let Rendezvous NIL
     (tell Printer 'rendezvous *Pid 'printLine Str)  # Call entry point
     (unless (wait 6000 Rendezvous)                  # Block max. 1 minute
        (quit "Rendezvous timed out") ) ) )
  1. Start RESERVE printer process

(unless (setq *ReservePrinter (fork))

  (setq *ID 2  *Ink 5)
  (wait) )  # Run forever
  1. Start MAIN printer process

(unless (setq *MainPrinter (fork))

  (setq *ID 1  *Ink 5  *Backup *ReservePrinter)
  (wait) )
  1. Start Humpty Dumpty process

(unless (fork)

  (when
     (catch '(NIL)
        (for Line
           (quote
              "Humpty Dumpty sat on a wall."
              "Humpty Dumpty had a great fall."
              "All the king's horses and all the king's men"
              "Couldn't put Humpty together again." )
           (rendezvousPrint *MainPrinter Line) ) )
     (prinl "      Humpty Dumpty: " @ "!") )
  (bye) )
  1. Start Mother Goose process

(unless (fork)

  (when
     (catch '(NIL)
        (for Line
           (quote
              "Old Mother Goose"
              "When she wanted to wander,"
              "Would ride through the air"
              "On a very fine gander."
              "Jack's mother came in,"
              "And caught the goose soon,"
              "And mounting its back,"
              "Flew up to the moon." )
           (rendezvousPrint *MainPrinter Line) ) )
     (prinl "      Mother Goose: " @ "!") )
  (bye) )
  1. Prepare to terminate all processes upon exit

(push '*Bye '(tell 'bye))</lang> Output:

1: Old Mother Goose
1: Humpty Dumpty sat on a wall.
1: When she wanted to wander,
1: Humpty Dumpty had a great fall.
1: Would ride through the air
2: All the king's horses and all the king's men
2: On a very fine gander.
2: Jack's mother came in,
2: And caught the goose soon,
2: And mounting its back,
      Humpty Dumpty: Out of Ink!

Tcl

Tcl does not have a rendezvous operation, but it does have the ability to send a script to another thread to be evaluated and the results passed back. Combined with coroutines (so that the code is not too ugly), this can make something that works very much like a rendezvous operation.

Works with: Tcl version 8.6

<lang tcl>package require Tcl 8.6 package require Thread

  1. Really ought to go in a package

eval [set rendezvousEngine { array set Select {w {} c 0}

  1. Turns the task into a coroutine, making it easier to write in "Ada style".
  2. The real thread ids are stored in shared variables.

proc task {id script} {

   global rendezvousEngine
   set task [list coroutine RTask eval "$script;thread::exit"]
   tsv::set tasks $id [thread::create \

"$rendezvousEngine;$task;thread::wait"] }

  1. A simple yielding pause.

proc pause t {

   after $t [info coroutine]
   yield

}

  1. Wait for a message. Note that this is *not* pretty code and doesn't do
  2. everything that the Ada rendezvous does.

proc select args {

   global Select
   set var [namespace which -variable Select](m[incr Select(c)])
   set messages {}
   foreach {message vars body} $args {

dict set messages $message $body dict set bindings $message $vars

   }
   lappend Select(w) [list $var [dict keys $messages]]
   try {

set Master "" while {$Master eq ""} { set Master [yield] } lassign $Master message responder payload foreach vbl [dict get $bindings $message] value $payload { upvar 1 $vbl v set v $value } set body [dict get $messages $message] set code [uplevel 1 [list catch $body ::Select(em) ::Select(op)]] set opts $Select(op) if {$code == 1} { dict append opts -errorinfo \ "\n while processing message\n$message $payload" } set $responder [list $code $Select(em) $opts]

   } finally {

catch {unset $var} set Select(w) [lrange $Select(w) 0 end-1]

   }

}

  1. This acts as a receiver for messages, feeding them into the waiting
  2. [select]. It is incomplete as it should (but doesn't) queue messages that
  3. can't be received currently.

proc receive {message args} {

   global Select
   lassign [lindex $Select(w) end] var messages
   if {$message ni $messages} {

throw BAD_MESSAGE "don't know message $message"

   }
   set responder [namespace which -variable Select](r[incr Select(c)])
   set $responder ""
   RTask [list $message $responder $args]
   set response [set $responder]
   unset responder
   after 1
   return $response

}

  1. This dispatches a message to a task in another thread.

proc send {target message args} {

   after 1
   set t [tsv::get tasks $target]
   if {![thread::send $t [list receive $message {*}$args] response]} {

lassign $response code msg opts return -options $opts $msg

   } else {

return -code error $response

   }

} }]

  1. The backup printer task.

task BackupPrinter {

   set n 5
   while {$n >= 0} {

select Print msg { if {$n > 0} { incr n -1 puts Backup:$msg } else { throw OUT_OF_INK "out of ink" } }

   }

}

  1. The main printer task.

task MainPrinter {

   set n 5
   set Backup BackupPrinter
   while 1 {

select Print msg { try { if {$n > 0} { incr n -1 puts Main:$msg } elseif {$Backup ne ""} { send $Backup Print $msg } else { throw OUT_OF_INK "out of ink" } } trap OUT_OF_INK {} { set Backup "" throw OUT_OF_INK "out of ink" } }

   }

}

  1. Tasks that generate messages to print.

task HumptyDumpty {

   pause 100
   try {

send MainPrinter Print "Humpty Dumpty sat on a wall." send MainPrinter Print "Humpty Dumpty had a great fall." send MainPrinter Print "All the King's horses and all the King's men" send MainPrinter Print "Couldn't put Humpty together again."

   } trap OUT_OF_INK {} {

puts "Humpty Dumpty out of ink!"

   }

} task MotherGoose {

   pause 100
   try {

send MainPrinter Print "Old Mother Goose" send MainPrinter Print "When she wanted to wander," send MainPrinter Print "Would ride through the air" send MainPrinter Print "On a very fine gander." send MainPrinter Print "Jack's mother came in," send MainPrinter Print "And caught the goose soon," send MainPrinter Print "And mounting its back," send MainPrinter Print "Flew up to the moon."

   } trap OUT_OF_INK {} {

puts "Mother Goose out of ink!"

   }

}

  1. Wait enough time for the example to run and then finish

after 1000 thread::broadcast thread::exit</lang>

See also