Negli esempi finora proposti, la realizzazione degli schemi dataflow è stata ottenuta con una duplice trasformazione dell'originale grafo di dipendenza dei dati. In una prima fase si identificano un certo numero di cicli computazionali paralleli ciascuno dei quali viene implementato con una specifica coroutine. Successivamente si impiega una rete di comunicazione fra coroutine, con la quale trasferire i necessari dati, utilizzando il già noto meccanismo dei token.
Si tratta di un parallelismo a grana grossa, che si riferisce a nodi istruzione contenenti un qualsivoglia numero di istruzioni semplici, la cui l'esecuzione è realizzata estendendo l'usuale macchina sequenziale di Von Neumann in modo da permettere la coesistenza di un certo numero di macchine virtuali sull'unica macchina fisica di Von Neumann disponibile.
La realizzazione del modello multisequenziale della macchina di Von Neumann richiede l'implementazione di un'istruzioni fork, responsabile della creazione delle coroutine che costituiscono i flussi sequenziali componenti, e della coppia di istruzioni send e recv, da utilizzarsi nello scambio di dati fra le componenti sequenziali.
Si tratta, in effetti, del nucleo di un sistema operativo dove lo scheduler e il dispatcher, nel realizzare l'istruzione fork, si accompagnano sempre ad una qualche forma di gestione della memoria centrale, per un trattamento corretto dei dati privati a ciascun flusso sequenziale.
Per quanto riguarda, invece, l'interazione fra i processi sequenziali va ricordato che, affinchè una cooperazione sia effettivamente possibile, è necessario che tali processi possano comunicare e sincronizzarsi. Il meccanismo dei token è una forma effettiva di comunicazione che, però, limita il trasferimento dell'informazione ad un token per volta. Per sfruttare al meglio le caratteristiche di asincronicità dei processi conviene riferirsi a canali con una maggiore larghezza di banda.
Consideriamo, allora, una coppia di processi P e Q che vogliono comunicare. Questo significa che deve essere presente un canale entro il quale P introduce dati, utilizzando l'istruzione send e Q li estrae con l'omologa receive. Analogamente al caso dei token, utilizzati nei sistemi dataflow, separiamo nel messaggio scambiato la parte che contiene il dato da quella che si riferisce al controllo.
Non è difficile rendersi conto che la tipica struttura asimmetrica del canale si trasforma in uno schema simmetrico di controllo d'accesso alla risorsa che realizza fisicamente il transito dei dati. In effetti, il processo Q non può estrarre dati dal canale, se questo è vuoto. Pertanto, deve attendere dal processo P il segnale NotEmpty di canale non vuoto.
Analogamente, quest'ultimo non può inserire dati, in quantità arbitraria nel canale, altrimenti eccederebbe quella che si chiama la capacità di canale, provocando distorsione che, nel nostro caso, si traduce in perdita di informazione. Dunque, il processo Q deve attendere il segnale NotFull di canale non pieno.
In conclusione, le operazioni di send e receive ammettono l'implementazione riportata di seguito
procedure send( var Mbx : Channel; Msg : Message); begin wait(Mbx.NotFull); Put(Mbx.Bu, Msg); signal(Mbx.NotEmpty) end; procedure receive( var Mbx : Channel; var Msg : Message); begin wait(Mbx.NotEmpty); Get(Mbx.Bu, Msg); signal(Mbx.NotFull) end;dove la struttura dati Bu dipende dal tipo di messaggio in transito e dalla gestione del flusso di informazioni all'interno del canale.
Nel caso del Concurrent Pascal Bu è un buffer circolare mentre NotFull e NotEmpty sono due semafori, il primo inizializzato alla capacità del canale, ossia, la dimensione del buffer, il secondo a zero perchè non possono essere presenti informazioni nel canale prima che qualcuno le produca.
Le operazioni Put e Get dipendono anch'esse dalla struttura del supporto informativo e dalla sua gestione. Il segmento di codice riportato più oltre permette di realizzare l'appropriato ambiente di comunicazione fra le coroutine attivate nel programma principale con l'istruzione cobegin..coend.
const Max0 = 5;
Max1 = 6;
type Semaphore = integer;
Buffer = array[0..Max0] of integer;
Timing = array[0..Max0] of boolean;
Channel = record
NotFull : Semaphore;
NotEmpty : Semaphore;
lf : integer;
ld : integer;
Bu : Buffer
end;
Event = record
NotFull : Semaphore;
NotEmpty : Semaphore;
lf : integer;
ld : integer;
Bu : Timing
end;
{*********************** Trattamento degli Interi *************************}
procedure InitChannel( var Mbx : Channel );
begin
Mbx.NotFull := Max1;
Mbx.NotEmpty := 0;
Mbx.lf := 0;
Mbx.ld := 0
end;
procedure Send( var Mbx : Channel; Msg : integer );
begin
wait( Mbx.NotFull );
Mbx.Bu[ Mbx.ld ] := Msg;
Mbx.ld := ( Mbx.ld + 1 ) mod Max1;
signal( Mbx.NotEmpty )
end;
procedure Receive( var Mbx : Channel; var Msg : integer );
begin
wait( Mbx.NotEmpty );
Msg:= Mbx.Bu[ Mbx.lf ];
Mbx.lf := ( Mbx.lf + 1 ) mod Max1;
signal( Mbx.NotFull )
end;
{*********************** Trattamento degli Eventi *************************}
procedure StartEvent( var Mbx : Event );
begin
Mbx.NotFull := Max1;
Mbx.NotEmpty := 0;
Mbx.lf := 0;
Mbx.ld := 0
end;
procedure Cause( var Mbx : Event; T : boolean );
begin
wait( Mbx.NotFull );
Mbx.Bu[ Mbx.ld ] := T;
Mbx.ld := ( Mbx.ld + 1 ) mod Max1;
signal( Mbx.NotEmpty )
end;
procedure Await( var Mbx : Event; var T : boolean );
begin
wait( Mbx.NotEmpty );
T := Mbx.Bu[ Mbx.lf ];
Mbx.lf := ( Mbx.lf + 1 ) mod Max1;
signal( Mbx.NotFull )
end;
Come si può facilmente notare lo schema implementativo precedente
propone due ambienti di comunicazione, uno per i valori booleani (eventi),
che possono essere utilizzati per la sincronizzazione fra coroutine,
l'altro per trasferire valori interi da una coroutine all'altra.
Naturalmente, nel caso della programmazione concorrente, una grande cura va riservata alla definizione dei canali di comunicazione che permettono alle diverse coroutine di interagire correttamente.
Viene presentato di seguito l'esempio di un semplice programma concorrente del quale, però, non viene fornita la corrispondente forma sequenziale. Lo scopo è quello di realizzarlo direttamente tenendo conto del possibile parallelismo presente.
Si tratta di un semplice esempio di interazione fra l'algoritmo che calcola il valore di un polinomio, riferito ad particolare valore assegnato, e l'insieme dei suoi coefficienti passati come input in corso di esecuzione. La computazione termina quando l'ultimo coefficiente fornito è nullo.
Si noti come, in questo caso, compaiono sole le dichiarazioni relative ai canali di comunicazione per il trasferimento di numeri interi.
program Horner( input, output );
{$V+}
const Max0 = 3;
Max1 = 4;
type Semaphore = integer;
Buffer = array[0..Max0] of integer;
Channel = record
NotFull : Semaphore;
NotEmpty : Semaphore;
lf : integer;
ld : integer;
Bu : Buffer
end;
var eC : Channel;
X : integer;
R : integer;
{****************** Trattamento degli Interi ********************}
procedure InitChannel( var Mbx : Channel );
begin
Mbx.NotFull := Max1;
Mbx.NotEmpty := 0;
Mbx.lf := 0;
Mbx.ld := 0
end;
procedure Send( var Mbx : Channel; Msg : integer );
begin
wait( Mbx.NotFull );
Mbx.Bu[ Mbx.ld ] := Msg;
Mbx.ld := ( Mbx.ld + 1 ) mod Max1;
signal( Mbx.NotEmpty )
end;
procedure Receive( var Mbx : Channel; var Msg : integer );
begin
wait( Mbx.NotEmpty );
Msg:= Mbx.Bu[ Mbx.lf ];
Mbx.lf := ( Mbx.lf + 1 ) mod Max1;
signal( Mbx.NotFull )
end;
{****************** Implementazione ********************}
procedure Generate;
var T : integer;
begin
writeln;
writeln('Polynomial Coefficients');
writeln;
repeat
write('Enter parm : ');
readln( T );
Send( eC, T )
until T = 0;
end;
procedure Evaluate( X : integer; var Y : integer );
var P : integer;
A : integer;
begin
P := 0;
repeat
Receive( eC, A );
P := P * X + A
until A = 0;
Y := P div X
end;
begin
InitChannel( eC );
writeln;
writeln('Valore di un polinomio in X');
writeln;
write(' Enter Point [X] : ');
readln( X );
cobegin
Generate;
Evaluate( X, R )
coend;
writeln;
writeln( R )
end.