How to detect a timeout when using asynchronous Socket.BeginReceive?

Question

Writing an asynchronous Ping using Raw Sockets in F#, to enable parallel requests using as few threads as possible. Not using "System.Net.NetworkInformation.Ping", because it appears to allocate one thread per request. Am also interested in using F# async workflows.

The synchronous version below correctly times out when the target host does not exist/respond, but the asynchronous version hangs. Both work when the host does respond. Not sure if this is a .NET issue, or an F# one...

Any ideas?

(note: the process must run as Admin to allow Raw Socket access)

This throws a timeout:

let result = Ping.Ping ( IPAddress.Parse( "192.168.33.22" ), 1000 )

However, this hangs:

let result = Ping.AsyncPing ( IPAddress.Parse( "192.168.33.22" ), 1000 )
             |> Async.RunSynchronously

Here's the code...

module Ping

open System
open System.Net
open System.Net.Sockets
open System.Threading

//---- ICMP Packet Classes

type IcmpMessage (t : byte) =
    let mutable m_type = t
    let mutable m_code = 0uy
    let mutable m_checksum = 0us

    member this.Type
        with get() = m_type

    member this.Code
        with get() = m_code

    member this.Checksum = m_checksum

    abstract Bytes : byte array

    default this.Bytes
        with get() =
            [|
                m_type
                m_code
                byte(m_checksum)
                byte(m_checksum >>> 8)
            |]

    member this.GetChecksum() =
        let mutable sum = 0ul
        let bytes = this.Bytes
        let mutable i = 0

        // Sum up uint16s
        while i < bytes.Length - 1 do
            sum <- sum + uint32(BitConverter.ToUInt16( bytes, i ))
            i <- i + 2

        // Add in last byte, if an odd size buffer
        if i <> bytes.Length then
            sum <- sum + uint32(bytes.[i])

        // Shuffle the bits
        sum <- (sum >>> 16) + (sum &&& 0xFFFFul)
        sum <- sum + (sum >>> 16)
        sum <- ~~~sum
        uint16(sum)

    member this.UpdateChecksum() =
        m_checksum <- this.GetChecksum()


type InformationMessage (t : byte) =
    inherit IcmpMessage(t)

    let mutable m_identifier = 0us
    let mutable m_sequenceNumber = 0us

    member this.Identifier = m_identifier
    member this.SequenceNumber = m_sequenceNumber

    override this.Bytes
        with get() =
            Array.append (base.Bytes)
                         [|
                            byte(m_identifier)
                            byte(m_identifier >>> 8)
                            byte(m_sequenceNumber)
                            byte(m_sequenceNumber >>> 8)
                         |]

type EchoMessage() =
    inherit InformationMessage( 8uy )
    let mutable m_data = Array.create 32 32uy
    do base.UpdateChecksum()

    member this.Data
        with get()  = m_data
        and  set(d) = m_data <- d
                      this.UpdateChecksum()

    override this.Bytes
        with get() =
            Array.append (base.Bytes)
                         (this.Data)

//---- Synchronous Ping

let Ping (host : IPAddress, timeout : int ) =
    let mutable ep = new IPEndPoint( host, 0 )
    let socket = new Socket( AddressFamily.InterNetwork, SocketType.Raw, ProtocolType.Icmp )
    socket.SetSocketOption( SocketOptionLevel.Socket, SocketOptionName.SendTimeout, timeout )
    socket.SetSocketOption( SocketOptionLevel.Socket, SocketOptionName.ReceiveTimeout, timeout )
    let packet = EchoMessage()
    let mutable buffer = packet.Bytes

    try
        if socket.SendTo( buffer, ep ) <= 0 then
            raise (SocketException())
        buffer <- Array.create (buffer.Length + 20) 0uy

        let mutable epr = ep :> EndPoint
        if socket.ReceiveFrom( buffer, &epr ) <= 0 then
            raise (SocketException())
    finally
        socket.Close()

    buffer

//---- Entensions to the F# Async class to allow up to 5 paramters (not just 3)

type Async with
    static member FromBeginEnd(arg1,arg2,arg3,arg4,beginAction,endAction,?cancelAction): Async<'T> =
        Async.FromBeginEnd((fun (iar,state) -> beginAction(arg1,arg2,arg3,arg4,iar,state)), endAction, ?cancelAction=cancelAction)
    static member FromBeginEnd(arg1,arg2,arg3,arg4,arg5,beginAction,endAction,?cancelAction): Async<'T> =
        Async.FromBeginEnd((fun (iar,state) -> beginAction(arg1,arg2,arg3,arg4,arg5,iar,state)), endAction, ?cancelAction=cancelAction)

//---- Extensions to the Socket class to provide async SendTo and ReceiveFrom

type System.Net.Sockets.Socket with

    member this.AsyncSendTo( buffer, offset, size, socketFlags, remoteEP ) =
        Async.FromBeginEnd( buffer, offset, size, socketFlags, remoteEP,
                            this.BeginSendTo,
                            this.EndSendTo )
    member this.AsyncReceiveFrom( buffer, offset, size, socketFlags, remoteEP ) =
        Async.FromBeginEnd( buffer, offset, size, socketFlags, remoteEP,
                            this.BeginReceiveFrom,
                            (fun asyncResult -> this.EndReceiveFrom(asyncResult, remoteEP) ) )

//---- Asynchronous Ping

let AsyncPing (host : IPAddress, timeout : int ) =  
    async {
        let ep = IPEndPoint( host, 0 )
        use socket = new Socket( AddressFamily.InterNetwork, SocketType.Raw, ProtocolType.Icmp )
        socket.SetSocketOption( SocketOptionLevel.Socket, SocketOptionName.SendTimeout, timeout )
        socket.SetSocketOption( SocketOptionLevel.Socket, SocketOptionName.ReceiveTimeout, timeout )

        let packet = EchoMessage()
        let outbuffer = packet.Bytes

        try
            let! result = socket.AsyncSendTo( outbuffer, 0, outbuffer.Length, SocketFlags.None, ep )
            if result <= 0 then
                raise (SocketException())

            let epr = ref (ep :> EndPoint)
            let inbuffer = Array.create (outbuffer.Length + 256) 0uy 
            let! result = socket.AsyncReceiveFrom( inbuffer, 0, inbuffer.Length, SocketFlags.None, epr )
            if result <= 0 then
                raise (SocketException())
            return inbuffer
        finally
            socket.Close()
    }

Answer 1

James, your own accepted answer has a problem I wanted to point out. You only allocate one timer, which makes the async object returned by AsyncReceiveEx a stateful one-time-use object. Here's a similar example that I trimmed down:

let b,e,c = Async.AsBeginEnd(Async.Sleep)

type Example() =
    member this.Close() = ()
    member this.AsyncReceiveEx( sleepTime, (timeoutMS:int) ) = 
        let timedOut = ref false 
        let completed = ref false 
        let timer = new System.Timers.Timer(double(timeoutMS), AutoReset=false)
        timer.Elapsed.Add( fun _ -> 
            lock timedOut (fun () -> 
                timedOut := true 
                if not !completed 
                then this.Close() 
                ) 
            ) 
        let complete() = 
            lock timedOut (fun () -> 
                timer.Stop() 
                timer.Dispose() 
                completed := true 
                ) 
        Async.FromBeginEnd( sleepTime,
                            (fun st -> 
                                let result = b(st)
                                timer.Start() 
                                result 
                            ), 
                            (fun result -> 
                                complete() 
                                if !timedOut 
                                then printfn "err"; () 
                                else e(result)
                            ), 
                            (fun () -> 
                                complete() 
                                this.Close() 
                                ) 
                            ) 

let ex = new Example()
let a = ex.AsyncReceiveEx(3000, 1000)
Async.RunSynchronously a
printfn "ok..."
// below throws ODE, because only allocated one Timer
Async.RunSynchronously a

Ideally you want every 'run' of the async returned by AsyncReceiveEx to behave the same, which means each run needs its own timer and set of ref flags. This is easy to fix thusly:

let b,e,c = Async.AsBeginEnd(Async.Sleep)

type Example() =
    member this.Close() = ()
    member this.AsyncReceiveEx( sleepTime, (timeoutMS:int) ) = 
        async {
        let timedOut = ref false 
        let completed = ref false 
        let timer = new System.Timers.Timer(double(timeoutMS), AutoReset=false)
        timer.Elapsed.Add( fun _ -> 
            lock timedOut (fun () -> 
                timedOut := true 
                if not !completed 
                then this.Close() 
                ) 
            ) 
        let complete() = 
            lock timedOut (fun () -> 
                timer.Stop() 
                timer.Dispose() 
                completed := true 
                ) 
        return! Async.FromBeginEnd( sleepTime,
                            (fun st -> 
                                let result = b(st)
                                timer.Start() 
                                result 
                            ), 
                            (fun result -> 
                                complete() 
                                if !timedOut 
                                then printfn "err"; () 
                                else e(result)
                            ), 
                            (fun () -> 
                                complete() 
                                this.Close() 
                                ) 
                            ) 
        }
let ex = new Example()
let a = ex.AsyncReceiveEx(3000, 1000)
Async.RunSynchronously a
printfn "ok..."
Async.RunSynchronously a

The only change is to put the body of AsyncReceiveEx inside async{...} and have the last line return!.