Initial commit of structure

PulsingServer.sln

@@ -0,0 +1,22 @@
+
+Microsoft Visual Studio Solution File, Format Version 12.00
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "PulsingServer", "PulsingServer\PulsingServer.fsproj", "{FF56F740-69E3-4072-B6B9-D69395D0E53A}"
+EndProject
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Test", "Test\Test.fsproj", "{6234E354-B2B1-4A45-8134-C306846E5A71}"
+EndProject
+Global
+	GlobalSection(SolutionConfigurationPlatforms) = preSolution
+		Debug|Any CPU = Debug|Any CPU
+		Release|Any CPU = Release|Any CPU
+	EndGlobalSection
+	GlobalSection(ProjectConfigurationPlatforms) = postSolution
+		{FF56F740-69E3-4072-B6B9-D69395D0E53A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{FF56F740-69E3-4072-B6B9-D69395D0E53A}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{FF56F740-69E3-4072-B6B9-D69395D0E53A}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{FF56F740-69E3-4072-B6B9-D69395D0E53A}.Release|Any CPU.Build.0 = Release|Any CPU
+		{6234E354-B2B1-4A45-8134-C306846E5A71}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{6234E354-B2B1-4A45-8134-C306846E5A71}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{6234E354-B2B1-4A45-8134-C306846E5A71}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{6234E354-B2B1-4A45-8134-C306846E5A71}.Release|Any CPU.Build.0 = Release|Any CPU
+	EndGlobalSection
+EndGlobal

PulsingServer/ExternalInfoProvider.fs

@@ -0,0 +1,77 @@
+namespace PulsingServer
+
+open System
+
+[<Measure>]
+type ms
+
+type private ExternalInfoProviderMessage<'info> =
+    | Get of AsyncReplyChannel<unit> option * int<ms>
+    | NewConsumers of AsyncReplyChannel<unit> * ServerAgent<'info> array
+
+/// An entity which periodically pulls information from some external source
+/// and pushes it out to a collection of ServerAgents.
+type ExternalInfoProvider<'info> =
+    private
+    | ExternalInfoProvider of MailboxProcessor<ExternalInfoProviderMessage<'info>>
+
+[<RequireQualifiedAccess>]
+module ExternalInfoProvider =
+
+    /// Create an ExternalInfoProvider which runs the `get` async every `timer` milliseconds.
+    /// When it gets a different `info`, it pings its `receivers` with that new info.
+    /// The async returns when the ExternalInfoProvider has constructed its first info
+    /// and has served that first info to its receivers.
+    let make<'info when 'info : equality>
+        (sleep : TimeSpan -> Async<unit>)
+        (get : Async<'info>)
+        (timer : int<ms>)
+        (receivers : ServerAgent<'info> array)
+        : Async<ExternalInfoProvider<'info>>
+        =
+        let rec loop (info : 'info option) (receivers : ServerAgent<'info> array) (mailbox : MailboxProcessor<ExternalInfoProviderMessage<'info>>) =
+            async {
+                match! mailbox.Receive () with
+                | Get (channel, timeout) ->
+                    let! newInfo = get
+                    match info with
+                    | Some info when newInfo = info ->
+                        ()
+                    | _ ->
+                        do!
+                            receivers
+                            |> Array.map (ServerAgent.post newInfo)
+                            |> Async.Parallel
+                            |> Async.Ignore
+
+                    match channel with
+                    | None -> ()
+                    | Some channel ->
+                        channel.Reply ()
+                    do! sleep (TimeSpan.FromMilliseconds (float timeout))
+                    // There's a small inaccuracy here. We actually will wait until the end
+                    // of a timeout cycle before we can process any new consumers. What we
+                    // should really do is to allow NewConsumers messages to be processed
+                    // during this downtime, by storing a "when did I start waiting" and
+                    // testing "has `timeout` elapsed since then?", rather than just waiting
+                    // for the timeout.
+                    mailbox.Post (Get (None, timeout))
+                    return! loop (Some newInfo) receivers mailbox
+                | NewConsumers (channel, receivers) ->
+                    channel.Reply ()
+                    return! loop info receivers mailbox
+            }
+
+        async {
+            let mailbox = MailboxProcessor.Start (loop None receivers)
+            do! mailbox.PostAndAsyncReply (fun channel -> Get (Some channel, timer))
+            return
+                mailbox
+                |> ExternalInfoProvider
+        }
+
+    /// Replace the collection of ServerAgents this ExternalInfoProvider is hooked up to.
+    /// The replacement may take place any time after function invocation, but it is
+    /// guaranteed to be complete once the Async returns.
+    let updateConsumers<'info> (arr : ServerAgent<'info> array) (ExternalInfoProvider prov) : Async<unit> =
+        prov.PostAndAsyncReply (fun channel -> NewConsumers (channel, arr))

PulsingServer/PulsingServer.fsproj

@@ -0,0 +1,12 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net5.0</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <Compile Include="ServerAgent.fs" />
+        <Compile Include="ExternalInfoProvider.fs" />
+    </ItemGroup>
+
+</Project>

PulsingServer/ServerAgent.fs

@@ -0,0 +1,46 @@
+namespace PulsingServer
+
+type private ServerAgentMessage<'info> =
+    | Read of AsyncReplyChannel<unit> * ('info -> unit)
+    | Write of AsyncReplyChannel<unit> * 'info
+
+type ServerAgent<'info> = private | ServerAgent of MailboxProcessor<ServerAgentMessage<'info>>
+
+[<RequireQualifiedAccess>]
+module ServerAgent =
+
+    /// Create a ServerAgent which is ready to take new information
+    /// and is ready to give responses.
+    let make<'info> (initialInfo : 'info) : ServerAgent<'info> =
+        let rec loop (info : 'info) (mailbox : MailboxProcessor<ServerAgentMessage<'info>>) =
+            async {
+                match! mailbox.Receive () with
+                | Read (channel, reply) ->
+                    reply info
+                    channel.Reply ()
+                    return! loop info mailbox
+                | Write (channel, info) ->
+                    channel.Reply ()
+                    return! loop info mailbox
+            }
+
+        loop initialInfo
+        |> MailboxProcessor.Start
+        |> ServerAgent
+
+    /// Write new information to this ServerAgent's internal store.
+    /// The write may take place any time after this function returns;
+    /// but the write is guaranteed to have been performed once the Async completes.
+    let post<'info> (info : 'info) (ServerAgent agent) : Async<unit> =
+        agent.PostAndAsyncReply (fun channel -> Write (channel, info))
+
+    /// Ask the ServerAgent to give back its info.
+    /// The function returns an async once it has submitted the request to the ServerAgent;
+    /// the async returns once the ServerAgent has finished responding.
+    let giveNextResponse<'info> (ServerAgent agent) : Async<'info> =
+        let mutable answer = Unchecked.defaultof<'info>
+        let result = agent.PostAndAsyncReply (fun channel -> Read (channel, fun info -> answer <- info))
+        async {
+            do! result
+            return answer
+        }

Test/PropertyBasedTests.fs

@@ -0,0 +1,36 @@
+namespace PulsingServer.Test
+
+open PulsingServer
+open NUnit.Framework
+open FsUnitTyped
+
+type AgentIndex = AgentIndex of int
+type ReadIndex = ReadIndex of int
+
+type Action<'info> =
+    | ChangeData of 'info
+    | BeginRead of AgentIndex
+    | AwaitRead of ReadIndex
+
+[<TestFixture>]
+module TestProperties =
+
+    let executeAction
+        (ext : ExternalInfoProvider<'info>)
+        (agents : ServerAgent<'info> array)
+        ((readNumber : int), (awaitingRead : Map<ReadIndex, Async<'info>>))
+        (action : Action<'info>)
+        =
+        match action with
+        | BeginRead (AgentIndex i) ->
+            let mutable answer = None
+            let result = ServerAgent.giveNextResponse (fun resp -> answer <- Some resp) agents.[i]
+            let output =
+                async {
+                    do! result
+                    return Option.get answer
+                }
+            ext, agents, (readNumber + 1, Map.add (ReadIndex readNumber) output awaitingRead)
+        | AwaitRead index ->
+            awaitingRead.[index]
+            |> Async.RunSynchronously

Test/Test.fsproj

@@ -0,0 +1,23 @@
+<Project Sdk="Microsoft.NET.Sdk">
+
+    <PropertyGroup>
+        <TargetFramework>net5.0</TargetFramework>
+    </PropertyGroup>
+
+    <ItemGroup>
+        <Compile Include="TestServer.fs" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <PackageReference Include="FsCheck" Version="2.14.4" />
+      <PackageReference Include="FsUnit" Version="4.0.4" />
+      <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.8.3" />
+      <PackageReference Include="NUnit3TestAdapter" Version="3.16.1" />
+      <PackageReference Include="NUnit" Version="3.13.1" />
+    </ItemGroup>
+
+    <ItemGroup>
+      <ProjectReference Include="..\PulsingServer\PulsingServer.fsproj" />
+    </ItemGroup>
+
+</Project>

Test/TestServer.fs

@@ -0,0 +1,172 @@
+namespace PulsingServer.Test
+
+open System
+open System.Diagnostics
+open PulsingServer
+open NUnit.Framework
+open FsUnitTyped
+
+[<TestFixture>]
+module TestPulsingServer =
+
+    [<Test>]
+    let ``Example test scenario`` () =
+        let responder1 = ServerAgent.make "hi"
+        let responder2 = ServerAgent.make "hi"
+
+        let mutable info = "original info"
+        let count = ref 0
+
+        let getInfo = async {
+            System.Threading.Interlocked.Increment count |> ignore
+            let info = lock info (fun () -> sprintf "%s" info)
+            return info
+        }
+        let dontSleep (_ : TimeSpan) = async { return () }
+
+        let infoProvider = ExternalInfoProvider.make dontSleep getInfo 10<ms> [| responder1 ; responder2 |]
+
+        // We're not getting new info, because we didn't await the construction of ExternalInfoProvider
+        count.Value |> shouldEqual 0
+
+        // The two responders are ready, but have not received anything yet.
+        do
+            let response = ServerAgent.giveNextResponse responder1
+            response
+            |> Async.RunSynchronously
+            |> shouldEqual "hi"
+
+        // Now start off the ExternalInfoProvider!
+        let _ =
+            infoProvider
+            |> Async.RunSynchronously
+
+        // Now we have definitely started pinging...
+        count.Value |> shouldBeGreaterThan 0
+
+        // ... and at some point soon, the first responder will act on the info it receives.
+        do
+            let response = ServerAgent.giveNextResponse responder1
+            response
+            |> Async.RunSynchronously
+            |> shouldEqual "original info"
+
+        // Update the info. responder1 is not going to fail on the `received` check, because that
+        // was one-shot.
+        lock info (fun () ->
+            info <- "new info!"
+        )
+
+        // Get responder2 ready to act in a couple of different ways.
+        let response2 = ServerAgent.giveNextResponse responder2
+        let response2' = ServerAgent.giveNextResponse responder2
+
+        // At some point soon, the infoProvider picks up the change and propagates it.
+
+        response2
+        |> Async.RunSynchronously
+        |> fun info ->
+            // By design, we can't distinguish between these two cases.
+            (info = "new info!" || info = "original info")
+            |> shouldEqual true
+        response2'
+        |> Async.RunSynchronously
+        |> fun info ->
+            // By design, we can't distinguish between these two cases.
+            (info = "new info!" || info = "original info")
+            |> shouldEqual true
+
+        // Eventually, responder2 does pick up the new info.
+        let rec go () =
+            let response = ServerAgent.giveNextResponse responder2 |> Async.RunSynchronously
+            if response <> "new info!" then go ()
+        go ()
+
+    [<TestCase (10000, 1)>]
+    [<TestCase (10000, 3)>]
+    let ``Stress test`` (n : int, queues : int) =
+        let responders = Array.init queues (fun _ -> ServerAgent.make "uninitialised")
+
+        let mutable data = ""
+        let getInfo =
+            async {
+                // Simulate a slow network call
+                do! Async.Sleep (TimeSpan.FromSeconds 1.)
+                let result = lock data (fun () -> sprintf "%s" data)
+                return result
+            }
+        let _infoProvider =
+            ExternalInfoProvider.make Async.Sleep getInfo 10<ms> responders
+            |> Async.RunSynchronously
+
+        let time = Stopwatch ()
+        // Restart it a couple of times to warm it up
+        time.Restart ()
+        time.Restart ()
+
+        // n requests come in - note that we don't start them off yet,
+        // because we want to time them separately
+        let requests =
+            Array.init n (fun i ->
+                async {
+                    let! answer = ServerAgent.giveNextResponse responders.[i % queues]
+                    if answer <> "" then failwith "unexpected response!"
+                    return ()
+                }
+            )
+            |> Async.Parallel
+            |> Async.Ignore
+
+        time.Stop ()
+        printfn "Time to construct requests: %i ms" time.ElapsedMilliseconds
+
+        time.Restart ()
+
+        requests
+        |> Async.RunSynchronously
+
+        time.Stop ()
+        printfn "Time to execute: %i ms" time.ElapsedMilliseconds
+
+        // Now prepare n more requests, but halfway through, we'll be changing the data.
+        // Again, don't kick them off right now; wait for the timer.
+        time.Restart ()
+        let requests =
+            Array.init n (fun i ->
+                if i = n / 2 then
+                    async {
+                        lock data (fun () -> data <- "new data")
+                        return None
+                    }
+                else
+                    async {
+                        do! Async.Sleep (TimeSpan.FromMilliseconds (float i))
+                        let! response = ServerAgent.giveNextResponse (responders.[i % queues])
+                        return Some response
+                    }
+            )
+            |> Async.Parallel
+        time.Stop ()
+
+        printfn "Time to construct requests: %i ms" time.ElapsedMilliseconds
+
+        time.Restart ()
+
+        let results =
+            requests
+            |> Async.RunSynchronously
+
+        time.Stop ()
+        printfn "Time to execute: %i ms" time.ElapsedMilliseconds
+
+        let grouped =
+            results
+            |> Array.countBy id
+            |> Map.ofArray
+
+        grouped.[None] |> shouldEqual 1
+        let pre = Map.tryFind (Some "") grouped |> Option.defaultValue 0
+        let post = Map.tryFind (Some "new data") grouped |> Option.defaultValue 0
+
+        pre + post |> shouldEqual (n - 1)
+        printfn "Got old data: %i. Got new data: %i." pre post