squash

axmmisaka · axmmisaka · commit 3f7a21f4036e · 2023-08-22T13:10:47.000-07:00
diff --git a/src/benchmark/sucoverrelax/actor.ts b/src/benchmark/sucoverrelax/actor.ts
@@ -1,6 +1,7 @@
 /* eslint-disable @typescript-eslint/prefer-readonly */
 // The SOR (Re)actor
-import { Reactor, State } from "../../core/internal";
+import { InPort, Reactor, State } from "../../core/internal";
+import { Message } from "./sorutils";
 
 export class SORActor extends Reactor {
     constructor(
@@ -22,6 +23,8 @@ export class SORActor extends Reactor {
         const omegaOverFour = 0.25 * omega;
         const oneMinusOmega = 1.0 - omega;
 
+        this.portFromRunner = new InPort<Message>(this);
+
         const neighbours = (() => {
             const calPos = (x: number, y: number): number => (x * ny + y);
 
@@ -60,5 +63,5 @@ export class SORActor extends Reactor {
     private msgRcv = new State(0);
     private sorActors = new State<Reactor[]>([]);
     
-    protected 
+    public portFromRunner: InPort<Message>;
 }
diff --git a/src/benchmark/sucoverrelax/peer.ts b/src/benchmark/sucoverrelax/peer.ts
@@ -1,18 +1,71 @@
-import { Reactor, State } from "../../core/internal";
+import { InPort, MutationSandbox, OutPort, Reactor, State } from "../../core/internal";
+import { SORActor } from "./actor";
 import { SORRunner } from "./runner";
-import { SorBorder } from "./sorutils";
+import { Message, MessageTypes, SorBorder as SORBorder, SorBorder } from "./sorutils";
 
 export class SORPeer extends Reactor {
+
+    s: State<number>;
+    partStart: State<number>;
+    matrixPart: State<number[][]>;
+    border: State<SORBorder>;
+    sorSource: State<SORRunner>
+
+
     sorActors: State<Reactor[]>;
+    public portFromSORRunner: InPort<Message>;
+
     constructor(
         parent: Reactor,
         s: number,
         partStart: number,
         matrixPart: number[][],
-        border: SorBorder,
+        border: SORBorder,
         sorSource: SORRunner
     ) {
         super(parent, "SORPeer");
         this.sorActors = new State([]);
+        this.s = new State(s);
+        this.partStart = new State(partStart);
+        this.matrixPart = new State(matrixPart);
+        this.border = new State(border);
+        this.sorSource = new State(sorSour ce);
+
+        this.portFromSORRunner = new InPort(this);
     }
-}
+
+    static boot(this: MutationSandbox, args: BootProcessingArgs) {
+        const myBorder: SORActor[] = [];
+        const {_s, border, sorActors, partStart} = args;
+        const s = _s.get();
+        const partStartVal = partStart.get();
+        
+        const sorActorsValue = sorActors.get();
+        const borderValue = border.get();
+        for (let i = 0; i < s; ++i) {
+            sorActorsValue[i * (s - partStartVal + 1)] = borderValue.borderActors[i];
+        }
+        sorActors.set(sorActorsValue);
+
+        for ()
+    }
+}
+
+interface BootProcessingArgs {
+    type: MessageTypes.sorBootMessage,
+    _s: State<number>,
+    border: State<SORBorder>,
+    sorActors: State<SORActor[]>,
+    partStart: State<number>,
+}
+interface ResultProcessingArgs {
+    type: MessageTypes.sorResultMessage,
+    expectingBoot: State<boolean>,
+    totalMsgRcv: State<number>,
+    returned: State<number>,
+    gTotal: State<number>,
+    s: State<number>,
+    part: State<number>
+}
+
+type ProcessingArgs = BootProcessingArgs | ResultProcessingArgs;
diff --git a/src/benchmark/sucoverrelax/runner.ts b/src/benchmark/sucoverrelax/runner.ts
@@ -1,25 +1,35 @@
-import { InMultiPort, InPort, OutMultiPort, OutPort, Parameter, Reactor, State, } from "../../core/internal";
+import { InMultiPort, InPort, MutationSandbox, OutMultiPort, OutPort, Parameter, Reactor, State, Variable, } from "../../core/internal";
 import { SORActor } from "./actor";
 import { SORPeer } from "./peer";
-import { Message, SorBorder, omega } from "./sorutils";
+import { Message, MessageTypes, SORBootMessage, SORResultMessage, SorBorder, jacobi, omega } from "./sorutils";
 
 export class SORRunner extends Reactor {
-    protected sorActors: State<Reactor[]>;
-    protected sorPeer: State<Reactor | undefined>;
+    protected s: State<number>;
+    protected part: State<number>;
+
+    protected sorActors: State<SORActor[]>;
+    protected sorPeer: State<SORPeer | undefined>;
 
     protected portToSORActors: OutPort<Message>;
     protected portToSORPeer: OutPort<Message>;
     // Unsure if this would work, let's just try......
     protected portFromSORActor: InPort<Message>;
     protected portFromSORPeer: InPort<Message>;
 
+    protected gTotal = new State(0.0);
+    protected returned = new State(0);
+    protected totalMsgRcv = new State(0);
+    protected expectingBoot = new State(true);
+
+    // In Savina, randoms is accessed directly from SucOverRelaxConfig.
+    // Here we pass it in to the closure.
     constructor(parent: Reactor, size: number, _randoms: number[][]) {
         super(parent, "SORRunner");
         // These are in SorRunner; 
-        const s = size;
+        this.s = new State(size);
         // In the scala implementation a simple /2 was used.
         // In JS we might need to enforce some sort of guarantee as it was used to calculate position
-        const part = Math.floor(s / 2);
+        this.part = new State(Math.floor(size / 2));
         /** These are from Savina. They should be rather irrelevant, actually. */
         this.sorActors = new State([]);
         this.sorPeer = new State(undefined);
@@ -28,16 +38,105 @@ export class SORRunner extends Reactor {
         // This creates a bunch of ports.
         this.portToSORActors = new OutPort(this);
         this.portToSORPeer = new OutPort(this);
-
-        // SorRunner::boot()
+        this.portFromSORActor = new InPort(this);
+        this.portFromSORPeer = new InPort(this);
+        
         this.addMutation(
             [this.startup],
-            [this.sorActors, this.sorPeer],
-            function (this, sorActors, sorPeer) {
-                const myBorder: Reactor[] = [];
-                const randoms = _randoms;
+            [this.sorActors, this.sorPeer, this.portToSORActors, this.portToSORPeer],
+            function (this, sorActors, sorPeer, portToSORActors, portToSORPeer) {
+                // TODO: Add actual stuff
+                ;
+            }
+        );
+    }
+
+    // This is to be used WITHIN mutation react functions.
+    static process(this: MutationSandbox, message: Message, args: ProcessingArgs): void {
+        switch (message.messageType) {
+            case MessageTypes.sorBootMessage: {
+                if (args.type !== MessageTypes.sorBootMessage) {
+                    throw new Error("Wrong type of arguments passed.");
+                }
+                // expectingBoot is args[0]
+                args.expectingBoot.set(false);
+                SORRunner.boot.apply(this, [args]);
+                break;
+            }
+            case MessageTypes.sorResultMessage: {
+                if (args.type !== MessageTypes.sorResultMessage) {
+                    throw new Error("Wrong type of arguments passed.");
+                }
 
-                // In scala, (i <- 0 until s) is loop excluding s.
+                const {mv, msgRcv} = message;
+                const {expectingBoot, totalMsgRcv, returned, gTotal, s, part} = args;
+                
+                if (expectingBoot.get()) {
+                    throw new Error("SORRunner not booted yet!");
+                } 
+
+                totalMsgRcv.set(totalMsgRcv.get() + msgRcv);
+                returned.set(returned.get() + 1);
+                gTotal.set(gTotal.get() + mv);
+
+                if (returned.get() === (s.get() * part.get()) + 1) {
+                    // TODO: validate
+                    // TODO: exit
+                    ;
+                }
+                break;
+            }
+            case MessageTypes.sorBorderMessage: {
+                if (args.type !== MessageTypes.sorBorderMessage) {
+                    throw new Error("Wrong type of arguments passed.");
+                }
+
+                const {mBorder} = message;
+                const {expectingBoot, s, part, sorActors, portToSORActors} = args;
+
+                if (expectingBoot.get()) {
+                    throw new Error("SORRunner not booted yet!");
+                } 
+                const sorActorsValue = sorActors.get();
+                for (let i = 0; i <= s.get(); ++i) {
+                    sorActorsValue[(i + 1) * (part.get() + 1) - 1] = mBorder.borderActors[i];
+                }
+                sorActors.set(sorActorsValue);
+                for (let i = 0; i <= s.get(); ++i) {
+                    for (let j = 0; j <= part.get(); ++j) {
+                        const pos = (i * (part.get() + 1)) + j;
+                        // Ibidem, connect then disconnect to simulate 
+                        // "fire and forget" in scala.
+                        this.connect(portToSORActors, sorActorsValue[pos].portFromRunner);
+                        this.getReactor().writable(portToSORActors).set(
+                            {
+                                messageType: MessageTypes.sorStartMessage,
+                                mi: jacobi,
+                                mActors: sorActorsValue
+                            }
+                        );
+                        this.disconnect(portToSORActors, sorActorsValue[pos].portFromRunner);
+                    }   
+                }
+                break;
+            }
+            default: {
+                throw new Error("Received wrong message from port");
+            }
+        }
+    }
+
+    // SorRunner::boot()
+    static boot(this: MutationSandbox,
+            args: BootProcessingArgs
+            ): void {
+                const {_randoms, _s, _part, sorActors, sorPeer, portToSORPeer} = args;
+
+                const myBorder: SORActor[] = [];
+                const randoms = _randoms;
+                const s = _s.get();
+                const part = _part.get();
+                // In scala, (i <- 0 until s) is a loop excluding s.
                 const sorActorsValue = sorActors.get();
                 for (let i = 0; i < s; ++i) {
                     let c = i % 2;
@@ -51,7 +150,7 @@ export class SORRunner extends Reactor {
                             SORActor,
                             pos, randoms[i][j], c, s, part + 1, omega, this.getReactor(), false
                         );
-                        // TODO: Make connections
+
                         if (j === (part - 1)) {
                             myBorder[i] = sorActorsValue[pos];
                         }
@@ -69,14 +168,50 @@ export class SORRunner extends Reactor {
 
                 const sorPeerValue = this.getReactor()._uncheckedAddSibling(
                     SORPeer,
-                    s, part, partialMatrix, new SorBorder(myBorder), 
+                    s, part, partialMatrix, new SorBorder(myBorder),
                     // A dirty hack. Maybe this will be removed as ports get added.
                     this.getReactor() as SORRunner
-                    );
+                );
                 sorPeer.set(sorPeerValue);
-                    // TODO: Add connections.
-
-            }
-        );
+                // Pass message.
+                // This is similar to Scala's !; but it looks pretty...... interesting in LF.
+                // If node is concurrent or parallel, this might be a problem, so direct copy-pastaing to C++ runtime might not work.
+                this.connect(portToSORPeer, sorPeerValue.portFromSORRunner);
+                this.getReactor().writable(portToSORPeer).set({messageType: MessageTypes.sorBootMessage});
+                // Disconnect immediately.
+                this.disconnect(portToSORPeer, sorPeerValue.portFromSORRunner);
     }
 }
+
+
+interface BootProcessingArgs {
+    type: MessageTypes.sorBootMessage,
+    expectingBoot: State<boolean>,            
+     _randoms: number[][],
+    _s: State<number>,
+    _part: State<number>,
+    sorActors: State<SORActor[]>,
+    sorPeer: State<SORPeer>,
+    portToSORPeer: OutPort<Message>
+}
+
+interface ResultProcessingArgs {
+    type: MessageTypes.sorResultMessage,
+    expectingBoot: State<boolean>,
+    totalMsgRcv: State<number>,
+    returned: State<number>,
+    gTotal: State<number>,
+    s: State<number>,
+    part: State<number>
+}
+
+interface BorderProcessingArgs {
+    type: MessageTypes.sorBorderMessage,
+    expectingBoot: State<boolean>,
+    s: State<number>,
+    part: State<number>,
+    sorActors: State<SORActor[]>,
+    portToSORActors: OutPort<Message>
+}
+
+type ProcessingArgs = BootProcessingArgs | ResultProcessingArgs | BorderProcessingArgs;
diff --git a/src/benchmark/sucoverrelax/sorutils.ts b/src/benchmark/sucoverrelax/sorutils.ts
@@ -1,4 +1,5 @@
 import { Reactor } from "../../core/reactor";
+import { SORActor } from "./actor";
 
 // Savina implementation of PRNG
 export class SavinaPRNG {
@@ -62,21 +63,47 @@ export function jgfValidate(gTotal: number, size: number): void {
 }
 
 export enum MessageTypes {
+    sorBootMessage,
+    sorResultMessage,
     sorBorderMessage,
     sorStartMessage,
     sorValueMessage,
-    sorBootMessage,
-    sorResuleMessage
 }
 
 export class SorBorder {
-    borderActors: Reactor[];
-    constructor(borderActors: Reactor[]) {
+    borderActors: SORActor[];
+    constructor(borderActors: SORActor[]) {
         this.borderActors = borderActors;
     }
 }
 
-export interface Message {
-    messageType: MessageTypes;
+export interface SORBootMessage {
+    messageType: MessageTypes.sorBootMessage;
+}
+
+export interface SORResultMessage {
+    messageType: MessageTypes.sorResultMessage;
+    mx: number;
+    my: number;
+    mv: number;
+    msgRcv: number;
+}
+
+export interface SORBorderMessage {
+    messageType: MessageTypes.sorBorderMessage;
+
+    mBorder: SorBorder;
+}
+
+export interface SORStartMessage {
+    messageType: MessageTypes.sorStartMessage;
+    mi: number;
+    mActors: SORActor[];
 }
 
+export interface SORValueMessage {
+    messageType: MessageTypes.sorValueMessage;
+}
+
+export type Message = SORBootMessage | SORResultMessage | SORBorderMessage | SORStartMessage | SORValueMessage;
+
diff --git a/src/benchmark/sucoverrelax/test.ts b/src/benchmark/sucoverrelax/test.ts
diff --git a/src/core/reactor.ts b/src/core/reactor.ts
