batching.fsx

#if !INTERACTIVE
module Program
#endif

open BenchmarkDotNet.Diagnostics.Windows
open BenchmarkDotNet.Attributes
open BenchmarkDotNet
open BenchmarkDotNet.Running
open Hyperion.ValueSerializers

type 'a Sized =
    { MaxSize : int64
      mutable currPos : int
      elements : 'a array }


type TestDU = 
| A
| B of int * string

let b<'a> = 
    { new Hyperion.SerializerFactories.ValueSerializerFactory () with
        override __.CanSerialize (_, type') = type' = typeof<'a Sized>
        override x.CanDeserialize (s, type') = x.CanSerialize(s, type')
        override __.BuildSerializer (serializer, type', serializers) = 
            let s = Hyperion.ValueSerializers.ObjectSerializer (type')
            let r = Hyperion.ObjectReader (fun a b -> serializer.GetSerializerByType(typeof<'a array>).ReadValue (a,b))
            s.Initialize(r, Hyperion.ObjectWriter(fun stream o session ->
                    let o = o:?> 'a Sized
                    let writeSize (l:int) = 
                        let b = session.GetBuffer 4
                        Hyperion.NoAllocBitConverter.GetBytes(l, b)
                        stream.Write (b, 0, 4)
                    let sizePos = stream.Position
                    stream.Position <- stream.Position + 4L

                    let rec writeAsMuchAsAllowed el (e: 'a Sized) overflow =
                        if e.currPos = e.elements.Length || stream.Position >= e.MaxSize || overflow then 
                            let currPos = stream.Position
                            stream.Position <- sizePos
                            writeSize el
                            stream.Position <- currPos

                        else
                            let i = o.elements.[e.currPos]
                            let startPosition = stream.Position
                            let duValue = serializer.GetSerializerByType (i.GetType())
                            duValue.WriteManifest(stream, session)
                            duValue.WriteValue(stream, i, session)
                            if stream.Position >= e.MaxSize && el > 0 then
                                printfn "overflow"
                                stream.SetLength startPosition
                                stream.Position <- startPosition
                                writeAsMuchAsAllowed el e true
                            else
                                e.currPos <- e.currPos + 1
                                writeAsMuchAsAllowed (el + 1) e false
                    writeAsMuchAsAllowed 0 o false
                    
                ))

            s :> ValueSerializer }

let settings = Hyperion.SerializerOptions(serializerFactories = [ b<TestDU> ], preserveObjectReferences = false)


let serializer = Hyperion.Serializer(settings)


let s (d:TestDU array) = 
    let d = { elements = d; currPos = 0; MaxSize = 223L }
    use mem = new System.IO.MemoryStream ()
    [| while d.currPos < d.elements.Length do
        mem.SetLength 0L
        serializer.Serialize (d, mem)
        yield mem.ToArray() |]


let d (d:byte array array) = 
    [|
        for d in d do
            use mem = new System.IO.MemoryStream (d)
            yield serializer.Deserialize<TestDU array> (mem) |]
    |> Array.concat



s [| A; A; 
     B (1, "fsdfsdfsdfsdf")
     B (1, "fsdfsdfsdfsdf")
     B (1, "fsdfsdfsdfsdf")
     B (1, "fsdfsdfsdfsdf")
     B (1, "fsdfsdfsdfsdf")
     B (1, "fsdfsdfsdfsdf")
     B (1, "fsdfsdfsdfsdf")
     B (1, "fsdfsdfsdfsdf")
     A
    
    |] |> d




//[<MemoryDiagnoser>]
//type Test () =

//    [<Benchmark>] member __.oneConnection  () = oneConnection ()
//    [<Benchmark>] member __.multipleThreads2  () = multipleThreadsN 2
//    [<Benchmark>] member __.multipleThreads4  () = multipleThreadsN 4
//    [<Benchmark>] member __.multipleThreads8  () = multipleThreadsN 8
//    [<Benchmark>] member __.multipleThreadsMultipleConn  () = multipleThreadsMultipleConn ()
        


//[<EntryPoint>]
//let Main args =
//    BenchmarkRunner.Run<Test>() |> ignore
//    0