message_chain.cs

// C# implementation of a circular message chain of n nodes, which pass a message around m times. 
// Uses a basic scheduler instead of threads.

using System;
using System.Linq;
using System.Collections.Generic;

public class MessageChain
{
    public static void Main()
    {
       // create the message ring
       var Pid = ConnectMessageLoop(3,3);

       // trigger the message ring
       Scheduler.Message(Pid);

       // do everything.
       Scheduler.ProcessMessages();

       // everything clean up?
       Console.WriteLine("Remaining processes: {0}", Scheduler.Processes.Count);
       Console.WriteLine("Remaining messages: {0}", Scheduler.Messages.Count);       
    }

    public static Scheduler.Process MessageNode (State state) 
    {     
        Console.WriteLine("Received. Messaging {0}. My loops are {1}", state.Next, state.Loops);

        // async message next node.
        Scheduler.Message(state.Next);

        // returns null if we have passed the message around enough. process death.
        if (state.Loops == 0) return null;
        
        // returns function of s, but closes over state to be used.
        return (s) => MessageNode(new State() { Next = state.Next, Loops = state.Loops - 1 });
    }

    public static Scheduler.Process StartNode (State state) 
    {
        Console.WriteLine("Received: Updated state. Next node is {0}", state.Next);
        
        // returns function of s, but closes over the updated state.
        return (s) => MessageNode(state);
    } 

    public static Guid CreateChain(Guid Pid, int nodes, int messages) 
    {
        // gots to stop somewhere.
        if (nodes <= 0) return Pid;
        
        // spawn new node, passing in current Pid
        Guid newPid = Scheduler.Spawn((s) => MessageNode(new State() { Next = Pid, Loops = messages }));

        // recurse with new Pid.
        return CreateChain(newPid, nodes - 1, messages);
    }

    public static Guid ConnectMessageLoop(int nodes, int messages) 
    {
       // create a start node 
       var Pid = Scheduler.Spawn(StartNode);

       // create a chain, from now on everything indexed from 0
       var EndPid = CreateChain(Pid, nodes - 1, messages - 1); 

       // message start node, to link it to end node
       Scheduler.Message(Pid, new State() { Next = EndPid, Loops = messages - 1 } );

       // return a node
       return Pid;
    }
    
}

public struct State {
    public Guid Next;
    public int Loops;
}

public struct Message {
    public Guid Address;
    public State State;
}

public class Scheduler
{
    // all processes have this signature.
    public delegate Process Process(State state);

    public static Dictionary<Guid, Process> Processes = new Dictionary<Guid,Process>();
    public static List<Message> Messages = new List<Message>();

    public static Guid Spawn(Process process)
    {
       // create guid and add process to process dictionary
       Guid key = Guid.NewGuid();
       Processes.Add(key, process);
       Console.WriteLine("Spawned: {0}", key); 
       return key; 
    }
    
    public static void Message(Guid address)
    {
        Message(address, new State());
    }

    public static void Message(Guid address, State state)
    {
        Messages.Add(new Message() { Address = address, State = state });
    }

    public static void ProcessMessages() {
        if (Messages.Count == 0) return;
       
        // pop the first message off the queue 
        var message = Messages.First();
        Messages.RemoveAt(0);

        // only send message if we have a matching process
        if (Processes.ContainsKey(message.Address)) 
        {
            // call delegate and save return delegate.
            Process process = Processes[message.Address](message.State);

            if (process != null) {
                // update key with new process
                Processes[message.Address] = process;
            } else {
                // remove process from the process list.
                Processes.Remove(message.Address);
            }
        }

        // recurse
        ProcessMessages();
    }
}

output.txt

Spawned: 19f6f4b5-9760-4994-8ad8-b0b42bd3b904
Spawned: d2255fd8-0626-4a2c-ac71-107ab9b9c281
Spawned: 350a53bb-babf-460e-9bb5-6b000da6920a
Received: Updated state. Next node is 350a53bb-babf-460e-9bb5-6b000da6920a
Received. Messaging 350a53bb-babf-460e-9bb5-6b000da6920a. My loops are 2
Received. Messaging d2255fd8-0626-4a2c-ac71-107ab9b9c281. My loops are 2
Received. Messaging 19f6f4b5-9760-4994-8ad8-b0b42bd3b904. My loops are 2
Received. Messaging 350a53bb-babf-460e-9bb5-6b000da6920a. My loops are 1
Received. Messaging d2255fd8-0626-4a2c-ac71-107ab9b9c281. My loops are 1
Received. Messaging 19f6f4b5-9760-4994-8ad8-b0b42bd3b904. My loops are 1
Received. Messaging 350a53bb-babf-460e-9bb5-6b000da6920a. My loops are 0
Received. Messaging d2255fd8-0626-4a2c-ac71-107ab9b9c281. My loops are 0
Received. Messaging 19f6f4b5-9760-4994-8ad8-b0b42bd3b904. My loops are 0
Remaining processes: 0
Remaining messages: 0