lib/ack/libm2/CSP.mod

   1 (*$R-*)
   2 IMPLEMENTATION MODULE CSP;
   3 (*
   4   Module:       Communicating Sequential Processes
   5   From:         "A Modula-2 Implementation of CSP",
   6                 M. Collado, R. Morales, J.J. Moreno,
   7                 SIGPlan Notices, Volume 22, Number 6, June 1987.
   8                 Some modifications by Ceriel J.H. Jacobs
   9   Version:      $Header$
  10
  11    See this article for an explanation of the use of this module.
  12 *)
  13
  14   FROM random   IMPORT  Uniform;
  15   FROM SYSTEM   IMPORT  BYTE, ADDRESS, NEWPROCESS, TRANSFER;
  16   FROM Storage  IMPORT  Allocate, Deallocate;
  17   FROM Traps    IMPORT  Message;
  18
  19   CONST WorkSpaceSize = 2000;
  20
  21   TYPE  ByteAddress =   POINTER TO BYTE;
  22         Channel =       POINTER TO ChannelDescriptor;
  23         ProcessType =   POINTER TO ProcessDescriptor;
  24         ProcessDescriptor = RECORD
  25                                 next: ProcessType;
  26                                 father: ProcessType;
  27                                 cor: ADDRESS;
  28                                 wsp: ADDRESS;
  29                                 guardindex: INTEGER;
  30                                 guardno: CARDINAL;
  31                                 guardcount: CARDINAL;
  32                                 opened: Channel;
  33                                 sons: CARDINAL;
  34                                 msgadr: ADDRESS;
  35                                 msglen: CARDINAL;
  36                             END;
  37
  38         Queue = RECORD
  39                     head, tail: ProcessType;
  40                 END;
  41
  42         ChannelDescriptor = RECORD
  43                                 senders: Queue;
  44                                 owner: ProcessType;
  45                                 guardindex: INTEGER;
  46                                 next: Channel;
  47                             END;
  48
  49   VAR   cp: ProcessType;
  50         free, ready: Queue;
  51
  52 (* ------------ Private modules and procedures ------------- *)
  53
  54   MODULE ProcessQueue;
  55
  56     IMPORT      ProcessType, Queue;
  57     EXPORT      Push, Pop, InitQueue, IsEmpty;
  58
  59     PROCEDURE InitQueue(VAR q: Queue);
  60     BEGIN
  61         WITH q DO
  62                 head := NIL;
  63                 tail := NIL
  64         END
  65     END InitQueue;
  66
  67     PROCEDURE Push(p: ProcessType; VAR q: Queue);
  68     BEGIN
  69         p^.next := NIL;
  70         WITH q DO
  71                 IF head = NIL THEN
  72                         tail := p
  73                 ELSE
  74                         head^.next := p
  75                 END;
  76                 head := p
  77         END
  78     END Push;
  79
  80     PROCEDURE Pop(VAR q: Queue; VAR p: ProcessType);
  81     BEGIN
  82         WITH q DO
  83                 p := tail;
  84                 IF p # NIL THEN
  85                         tail := tail^.next;
  86                         IF head = p THEN
  87                                 head := NIL
  88                         END
  89                 END
  90         END
  91     END Pop;
  92
  93     PROCEDURE IsEmpty(q: Queue): BOOLEAN;
  94     BEGIN
  95         RETURN q.head = NIL
  96     END IsEmpty;
  97
  98   END ProcessQueue;
  99
 100
 101   PROCEDURE DoTransfer;
 102     VAR aux: ProcessType;
 103   BEGIN
 104         aux := cp;
 105         Pop(ready, cp);
 106         IF cp = NIL THEN
 107                 HALT
 108         ELSE
 109                 TRANSFER(aux^.cor, cp^.cor)
 110         END
 111   END DoTransfer;
 112
 113   PROCEDURE OpenChannel(ch: Channel; n: INTEGER);
 114   BEGIN
 115         WITH ch^ DO
 116                 IF guardindex = 0 THEN
 117                         guardindex := n;
 118                         next := cp^.opened;
 119                         cp^.opened := ch
 120                 END
 121         END
 122   END OpenChannel;
 123
 124   PROCEDURE CloseChannels(p: ProcessType);
 125   BEGIN
 126         WITH p^ DO
 127                 WHILE opened # NIL DO
 128                         opened^.guardindex := 0;
 129                         opened := opened^.next
 130                 END
 131         END
 132   END CloseChannels;
 133
 134   PROCEDURE ThereAreOpenChannels(): BOOLEAN;
 135   BEGIN
 136         RETURN cp^.opened # NIL;
 137   END ThereAreOpenChannels;
 138
 139   PROCEDURE Sending(ch: Channel): BOOLEAN;
 140   BEGIN
 141         RETURN NOT IsEmpty(ch^.senders)
 142   END Sending;
 143
 144 (* -------------- Public Procedures ----------------- *)
 145
 146   PROCEDURE COBEGIN;
 147   (* Beginning of a COBEGIN .. COEND structure *)
 148   BEGIN
 149   END COBEGIN;
 150
 151   PROCEDURE COEND;
 152   (* End of a COBEGIN .. COEND structure *)
 153     (* VAR      aux: ProcessType; *)
 154   BEGIN
 155         IF cp^.sons > 0 THEN
 156                 DoTransfer
 157         END
 158   END COEND;
 159
 160   PROCEDURE StartProcess(P: PROC);
 161   (* Start an anonimous process that executes the procedure P *)
 162     VAR newprocess: ProcessType;
 163   BEGIN
 164         Pop(free, newprocess);
 165         IF newprocess = NIL THEN
 166                 Allocate(newprocess,SIZE(ProcessDescriptor));
 167                 Allocate(newprocess^.wsp, WorkSpaceSize)
 168         END;
 169         WITH newprocess^ DO
 170                 father := cp;
 171                 sons := 0;
 172                 msglen := 0;
 173                 NEWPROCESS(P, wsp, WorkSpaceSize, cor)
 174         END;
 175         cp^.sons := cp^.sons + 1;
 176         Push(newprocess, ready)
 177   END StartProcess;
 178
 179   PROCEDURE StopProcess;
 180   (* Terminate a Process (itself) *)
 181     VAR aux: ProcessType;
 182   BEGIN
 183         aux := cp^.father;
 184         aux^.sons := aux^.sons - 1;
 185         IF aux^.sons = 0 THEN
 186                 Push(aux, ready)
 187         END;
 188         aux := cp;
 189         Push(aux, free);
 190         Pop(ready, cp);
 191         IF cp = NIL THEN
 192                 HALT
 193         ELSE
 194                 TRANSFER(aux^.cor, cp^.cor)
 195         END
 196   END StopProcess;
 197
 198   PROCEDURE InitChannel(VAR ch: Channel);
 199   (* Initialize the channel ch *)
 200   BEGIN
 201         Allocate(ch, SIZE(ChannelDescriptor));
 202         WITH ch^ DO
 203                 InitQueue(senders);
 204                 owner := NIL;
 205                 next := NIL;
 206                 guardindex := 0
 207         END
 208   END InitChannel;
 209
 210   PROCEDURE GetChannel(ch: Channel);
 211   (* Assign the channel ch to the process that gets it *)
 212   BEGIN
 213         WITH ch^ DO
 214                 IF owner # NIL THEN
 215                         Message("Channel already has an owner");
 216                         HALT
 217                 END;
 218                 owner := cp
 219         END
 220   END GetChannel;
 221
 222   PROCEDURE Send(data: ARRAY OF BYTE; VAR ch: Channel);
 223   (* Send a message with the data to the cvhannel ch *)
 224     VAR m: ByteAddress;
 225         (* aux: ProcessType; *)
 226         i: CARDINAL;
 227   BEGIN
 228         WITH ch^ DO
 229                 Push(cp, senders);
 230                 Allocate(cp^.msgadr, SIZE(data));
 231                 m := cp^.msgadr;
 232                 cp^.msglen := HIGH(data);
 233                 FOR i := 0 TO HIGH(data) DO
 234                         m^ := data[i];
 235                         m := ADDRESS(m) + 1
 236                 END;
 237                 IF guardindex # 0 THEN
 238                         owner^.guardindex := guardindex;
 239                         CloseChannels(owner);
 240                         Push(owner, ready)
 241                 END
 242         END;
 243         DoTransfer
 244   END Send;
 245
 246   PROCEDURE Receive(VAR ch: Channel; VAR dest: ARRAY OF BYTE);
 247   (* Receive a message from the channel ch into the dest variable *)
 248     VAR aux: ProcessType;
 249         m: ByteAddress;
 250         i: CARDINAL;
 251   BEGIN
 252         WITH ch^ DO
 253                 IF cp # owner THEN
 254                         Message("Only owner of channel can receive from it");
 255                         HALT
 256                 END;
 257                 IF Sending(ch) THEN
 258                         Pop(senders, aux);
 259                         m := aux^.msgadr;
 260                         FOR i := 0 TO aux^.msglen DO
 261                                 dest[i] := m^;
 262                                 m := ADDRESS(m) + 1
 263                         END;
 264                         Push(aux, ready);
 265                         Push(cp, ready);
 266                         CloseChannels(cp)
 267                 ELSE
 268                         OpenChannel(ch, -1);
 269                         DoTransfer;
 270                         Pop(senders, aux);
 271                         m := aux^.msgadr;
 272                         FOR i := 0 TO aux^.msglen DO
 273                                 dest[i] := m^;
 274                                 m := ADDRESS(m) + 1
 275                         END;
 276                         Push(cp, ready);
 277                         Push(aux, ready)
 278                 END;
 279                 Deallocate(aux^.msgadr, aux^.msglen+1);
 280                 DoTransfer
 281         END
 282   END Receive;
 283
 284   PROCEDURE SELECT(n: CARDINAL);
 285   (* Beginning of a SELECT structure with n guards *)
 286   BEGIN
 287         cp^.guardindex := Uniform(1,n);
 288         cp^.guardno := n;
 289         cp^.guardcount := n
 290   END SELECT;
 291
 292   PROCEDURE NEXTGUARD(): CARDINAL;
 293   (* Returns an index to the next guard to be evaluated in a SELECT *)
 294   BEGIN
 295         RETURN cp^.guardindex
 296   END NEXTGUARD;
 297
 298   PROCEDURE GUARD(cond: BOOLEAN; ch: Channel;
 299                   VAR dest: ARRAY OF BYTE): BOOLEAN;
 300   (* Evaluates a guard, including reception management *)
 301     (* VAR      aux: ProcessType; *)
 302   BEGIN
 303         IF NOT cond THEN
 304                 RETURN FALSE
 305         ELSIF ch = NIL THEN
 306                 CloseChannels(cp);
 307                 cp^.guardindex := 0;
 308                 RETURN TRUE
 309         ELSIF Sending(ch) THEN
 310                 Receive(ch, dest);
 311                 cp^.guardindex := 0;
 312                 RETURN TRUE
 313         ELSE
 314                 OpenChannel(ch, cp^.guardindex);
 315                 RETURN FALSE
 316         END
 317   END GUARD;
 318
 319   PROCEDURE ENDSELECT(): BOOLEAN;
 320   (* End of a SELECT structure *)
 321   BEGIN
 322         WITH cp^ DO
 323                 IF guardindex <= 0 THEN
 324                         RETURN TRUE
 325                 END;
 326                 guardcount := guardcount - 1;
 327                 IF guardcount # 0 THEN
 328                         guardindex := (guardindex MOD INTEGER(guardno)) + 1
 329                 ELSIF ThereAreOpenChannels() THEN
 330                         DoTransfer
 331                 ELSE
 332                         guardindex := 0
 333                 END
 334         END;
 335         RETURN FALSE
 336   END ENDSELECT;
 337
 338 BEGIN
 339         InitQueue(free);
 340         InitQueue(ready);
 341         Allocate(cp,SIZE(ProcessDescriptor));
 342         WITH cp^ DO
 343                 sons := 0;
 344                 father := NIL
 345         END
 346 END CSP.
 347