reliable-io (empty) → 0.0.1
raw patch · 11 files changed
+3279/−0 lines, 11 filesdep +basedep +bindings-DSLdep +reliable-io
Dependencies added: base, bindings-DSL, reliable-io
Files
- Bindings/Reliable/IO.hsc +121/−0
- LICENSE +29/−0
- README.md +2/−0
- Reliable/IO.hs +518/−0
- examples/RunCUnitTests.hs +15/−0
- examples/Soak.hs +143/−0
- reliable-io.cabal +75/−0
- reliable.io/LICENCE +19/−0
- reliable.io/README.md +57/−0
- reliable.io/reliable.c +2135/−0
- reliable.io/reliable.h +165/−0
+ Bindings/Reliable/IO.hsc view
@@ -0,0 +1,121 @@+{-| + Module : Bindings.Reliable.IO + Description : Low-level bindings to the reliable.io library. + Copyright : (c) Pavel Krajcevski, 2020 + License : BSD-3 + Maintainer : krajcevski@gmail.com + Stability : experimental + Portability : Portable + + This module contains the low-level bindings that represent the direct + interface between Haskell and the C library + <https://github.com/networkprotocol/reliable.io reliable.io>. + + The bindings here are meant for advanced usage, as they are not particularly + idiomatic Haskell, and largely represent the types that we get from the + <https://hackage.haskell.org/package/bindings-DSL bindings-DSL> library. For + high level bindings (recommended), please refer to the "Reliable.IO" module. +-} + +{-# LANGUAGE NoImplicitPrelude #-} +-------------------------------------------------------------------------------- + +#include <reliable.h> +#include <bindings.dsl.h> + +-------------------------------------------------------------------------------- + +module Bindings.Reliable.IO where + +import Data.Word (Word8, Word16, Word64) +import Foreign.C.String (CString) +import Foreign.C.Types (CChar(..), CInt(..), CFloat(..), CDouble(..)) +import Foreign.Marshal.Array (peekArray, pokeArray) +import Foreign.Ptr (Ptr, FunPtr, plusPtr) +import Foreign.Storable (Storable(..)) +import Prelude ( IO, Eq, Show, Num + , ($) + , div, undefined, return, take + ) + +-------------------------------------------------------------------------------- + +#num RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_SENT +#num RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_RECEIVED +#num RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_ACKED +#num RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_STALE +#num RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_INVALID +#num RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_SEND +#num RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_RECEIVE +#num RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_SENT +#num RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_RECEIVED +#num RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_INVALID +#num RELIABLE_ENDPOINT_NUM_COUNTERS + +#num RELIABLE_LOG_LEVEL_NONE +#num RELIABLE_LOG_LEVEL_ERROR +#num RELIABLE_LOG_LEVEL_INFO +#num RELIABLE_LOG_LEVEL_DEBUG + +#num RELIABLE_OK +#num RELIABLE_ERROR + +#ccall reliable_init, IO CInt +#ccall reliable_term, IO () + +#callback_t allocate_function_t, Ptr () -> Word64 -> IO (Ptr ()) +#callback_t free_function_t, Ptr () -> Ptr () -> IO () +#callback_t transmit_packet_function_t, Ptr () -> CInt -> Word16 -> Ptr Word8 -> CInt -> IO () +#callback_t process_packet_function_t, Ptr () -> CInt -> Word16 -> Ptr Word8 -> CInt -> IO CInt + +#starttype struct reliable_config_t +#array_field name, CChar +#field context, Ptr () +#field index, CInt +#field max_packet_size, CInt +#field fragment_above, CInt +#field max_fragments, CInt +#field fragment_size, CInt +#field ack_buffer_size, CInt +#field sent_packets_buffer_size, CInt +#field received_packets_buffer_size, CInt +#field fragment_reassembly_buffer_size, CInt +#field rtt_smoothing_factor, CFloat +#field packet_loss_smoothing_factor, CFloat +#field bandwidth_smoothing_factor, CFloat +#field packet_header_size, CInt +#field transmit_packet_function, <transmit_packet_function_t> +#field process_packet_function, <process_packet_function_t> +#field allocator_context, Ptr () +#field allocate_function, <allocate_function_t> +#field free_function, <free_function_t> +#stoptype + +#opaque_t reliable_endpoint_t + +#ccall reliable_default_config, Ptr <reliable_config_t> -> IO () +#ccall reliable_endpoint_create, Ptr <reliable_config_t> -> CDouble -> IO (Ptr <reliable_endpoint_t>) +#ccall reliable_endpoint_next_packet_sequence, Ptr <reliable_endpoint_t> -> IO Word16 +#ccall reliable_endpoint_send_packet, Ptr <reliable_endpoint_t> -> Ptr Word8 -> CInt -> IO () +#ccall reliable_endpoint_receive_packet, Ptr <reliable_endpoint_t> -> Ptr Word8 -> CInt -> IO () +#ccall reliable_endpoint_get_acks, Ptr <reliable_endpoint_t> -> Ptr CInt -> IO (Ptr Word16) +#ccall reliable_endpoint_clear_acks, Ptr <reliable_endpoint_t> -> IO () +#ccall reliable_endpoint_reset, Ptr <reliable_endpoint_t> -> IO () +#ccall reliable_endpoint_update, Ptr <reliable_endpoint_t> -> CDouble -> IO () +#ccall reliable_endpoint_rtt, Ptr <reliable_endpoint_t> -> IO CFloat +#ccall reliable_endpoint_packet_loss, Ptr <reliable_endpoint_t> -> IO CFloat +#ccall reliable_endpoint_bandwidth, Ptr <reliable_endpoint_t> -> Ptr CFloat -> Ptr CFloat -> Ptr CFloat -> IO () +#ccall reliable_endpoint_counters, Ptr <reliable_endpoint_t> -> IO (Ptr Word64) +#ccall reliable_endpoint_destroy, Ptr <reliable_endpoint_t> -> IO () +#ccall reliable_log_level, CInt -> IO () + +-- | Generally not useful -- just calls 'free_function' as used in the config. +#ccall reliable_endpoint_free_packet, Ptr <reliable_endpoint_t> -> Ptr () -> IO () + +-- Only available via low level bindings. In order to use these, the cc-flags +-- for this package need to remove NDEBUG as part of the preprocessor options. +#callback_t assert_function_t, CString -> CString -> CString -> CInt -> IO () +#ccall reliable_set_assert_function, <assert_function_t> -> IO () + +-- | For testing only. +#ccall reliable_test, IO ()
+ LICENSE view
@@ -0,0 +1,29 @@+BSD 3-Clause License + +Copyright (c) 2020, Pavel Krajcevski +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions are met: + +1. Redistributions of source code must retain the above copyright notice, this + list of conditions and the following disclaimer. + +2. Redistributions in binary form must reproduce the above copyright notice, + this list of conditions and the following disclaimer in the documentation + and/or other materials provided with the distribution. + +3. Neither the name of the copyright holder nor the names of its + contributors may be used to endorse or promote products derived from + this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" +AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE +IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL +DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER +CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, +OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE +OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,2 @@+# reliable-io +Haskell bindings to the reliable.io library: https://github.com/networkprotocol/reliable.io
+ Reliable/IO.hs view
@@ -0,0 +1,518 @@+{-| + +Module : Reliable.IO +Description : High-level bindings to the reliable.io library. +Copyright : (c) Pavel Krajcevski, 2020 +License : BSD-3 +Maintainer : krajcevski@gmail.com +Stability : experimental +Portability : Portable + +This module contains the high-level bindings on top of the module +"Bindings.Netcode.IO". These provide a cleaner interface to the +<https://github.com/networkprotocol/reliable.io reliable.io> C library and are +the recommended interface for application developers. + +These bindings have some limitations. Namely, they are not as performant as +the "close to the metal" bindings provided in "Bindings.Reliable.IO". In the +event that you need more performance, that module is available for use. + +This library is intended to be used with a way to send and receive fixed size +packets over an unreliable channel. If such an interface exists, then, assuming +that the two parties are in constant communication, this library will do the +following for you: + + 1. Break a packet down into a sequence of fixed size fragments to match + your data channel size. + 2. Determine whether or not a sent packet has been acked by the receiver. + 3. Reassemble a packet once all fragments have been received. + +With this in mind, the singular datatype provided by this library is an +'Endpoint'. Each endpoint requires the following: + + * How to send packet fragments ('TransmitPacketFunction') + * What to do with reassembled packets ('ProcessPacketFunction') + +Once you have an 'Endpoint', the two main operations that you would do with it +are to send a (possibly very large) packet ('sendPacket'), and provide it with +(possibly just one) packet fragments ('receivePacket'). On top of this library, +if a user would like to create a fully-reliable data channel (a la TCP), it +is that user's responsibility to identify when a packet has been dropped or has +arrived out of order to resend the appropriate packet. +-} +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE DeriveGeneric #-} +module Reliable.IO ( + -- * Initialization + initialize + , terminate + + -- * Utilities + , LogLevel(..) + , logLevel + + -- * Endpoint Configuration + , EndpointConfig + , defaultConfig + , setName + , setMaxPacketSize + , setPacketFragmentationLimit + , setPacketFragmentSize + , setMaxNumFragments + , setAckBufferSize + , setSentPacketsBufferSize + , setReceivedPacketsBufferSize + , setFragmentReassemblyBufferSize + , setRTTSmoothingFactor + , setPacketLossSmoothingFactor + , setBandwidthSmoothingFactor + , PacketType(..) + , setPacketType + + -- * Callbacks + , TransmitPacketFunction + , ProcessPacketFunction + + -- * Endpoints + , Endpoint + , createEndpoint + , destroyEndpoint + , withEndpoint + , nextPacketSequence + , sendPacket + , receivePacket + , getAcks + , clearAcks + , reset + , update + + -- * Analytics + , roundTripTime + , packetLoss + + , BandwidthMeasurements(..) + , bandwidth + + , Counter(..) + , getCounter + +) where + +------------------------------------------------------------------------------- + +import Control.Monad (when, unless) +import Data.Bool (bool) +import Data.Data (Data) +import Data.Typeable (Typeable) +import Data.Word (Word8, Word16, Word64) +import GHC.Generics (Generic) +import GHC.Ptr (Ptr) +import Foreign.C.String (withCStringLen) +import Foreign.C.Types (CInt, CDouble(..), CFloat(..)) +import Foreign.Marshal.Alloc (alloca) +import Foreign.Marshal.Array (peekArray) +import Foreign.Ptr (freeHaskellFunPtr) +import Foreign.Storable (poke, Storable(..)) + +import Bindings.Reliable.IO + +------------------------------------------------------------------------------- + +-- | Initializes the @reliable.io@ library runtime. This should be called before +-- any additional functions in this library. Throws an +-- t'Control.Exception.IOException' on failure. +initialize :: IO () +initialize = do + result <- c'reliable_init + unless (result == c'RELIABLE_OK) $ + fail $ "Failed to initialize reliable.io. Result: " <> show result + return () + +-- | Terminates the @reliable.io@ library runtime. This should be called only +-- after all other library functions terminate. +terminate :: IO () +terminate = c'reliable_term + +-- | Specifies the logging behavior of @reliable.io@. Note, this logging behavior +-- is called from C calls to @printf@ and therefore might interfere with the +-- Haskell runtime (such as 'putStrLn'). +data LogLevel = LogLevel'None + | LogLevel'Info + | LogLevel'Error + | LogLevel'Debug + deriving (Eq, Ord, Show, Read, Bounded, Enum, Generic, Data, Typeable) + +-- | Set the @reliable.io@ 'LogLevel'. The default is 'LogLevel'None'. +logLevel :: LogLevel -> IO () +logLevel LogLevel'None = c'reliable_log_level c'RELIABLE_LOG_LEVEL_NONE +logLevel LogLevel'Info = c'reliable_log_level c'RELIABLE_LOG_LEVEL_INFO +logLevel LogLevel'Error = c'reliable_log_level c'RELIABLE_LOG_LEVEL_ERROR +logLevel LogLevel'Debug = c'reliable_log_level c'RELIABLE_LOG_LEVEL_DEBUG + +-- | An 'EndpointConfig' is a write-only opaque datatype that is used to define +-- the settings for creating an 'Endpoint'. +newtype EndpointConfig = EndpointConfig { + generateConfig :: Ptr C'reliable_config_t -> IO () + } + +-- | The default 'EndpointConfig'. This uses sensible defaults for the library +-- (as opposed to being zero-initialized, for example). +defaultConfig :: EndpointConfig +defaultConfig = EndpointConfig c'reliable_default_config + +-- | Sets the name of the endpoint. This is usually not relevant, except when +-- setting the log level to be more than 'LogLevel'None'. +setName :: String -> EndpointConfig -> EndpointConfig +setName s (EndpointConfig fn) = EndpointConfig $ \cfgPtr -> do + fn cfgPtr + when (length s >= 256) $ fail "Endpoint config name too long" + withCStringLen s $ \(csPtr, csLen) -> do + config <- peek cfgPtr + cs <- peekArray csLen csPtr + poke cfgPtr $ config { c'reliable_config_t'name = (cs <> [0]) } + +-- Helper function to convert transforms on C structs into transforms on +-- Haskell datatypes. +setConfig :: (C'reliable_config_t -> C'reliable_config_t) + -> EndpointConfig -> EndpointConfig +setConfig fn (EndpointConfig mkCfg) = EndpointConfig $ \cfgPtr -> + mkCfg cfgPtr >> peek cfgPtr >>= poke cfgPtr . fn + +-- Calls 'setConfig' with an Int value +setConfigIntVal :: (CInt -> C'reliable_config_t -> C'reliable_config_t) + -> Int -> EndpointConfig -> EndpointConfig +setConfigIntVal fn x = setConfig $ fn (fromIntegral x) + +-- Calls 'setConfig' with a Float value +setConfigFloatVal :: (CFloat -> C'reliable_config_t -> C'reliable_config_t) + -> Float -> EndpointConfig -> EndpointConfig +setConfigFloatVal fn x = setConfig $ fn (CFloat x) + +-- | Sets the maximum packet size for the endpoint. This will allow the API to +-- know when to throw an error when the packet being sent is too big. The +-- packet size is purely application specific, but may be useful for making +-- sure that your data sizes don't grow too large during development. The +-- default value for this is 16KB. +setMaxPacketSize :: Int -> EndpointConfig -> EndpointConfig +setMaxPacketSize = setConfigIntVal $ \sz config -> + config { c'reliable_config_t'max_packet_size = sz } + +-- | Sets the fragmentation limit for this endpoint. The fragmentation limit is +-- the size in bytes for a packet where it will be split into multiple +-- fragments. This need not be @maxPacketSize / maxNumFragments@, but that is +-- usually a sensible choice. The default value is 1KB. +setPacketFragmentationLimit :: Int -> EndpointConfig -> EndpointConfig +setPacketFragmentationLimit = setConfigIntVal $ \l config -> + config { c'reliable_config_t'fragment_above = l } + +-- | Sets the fragment size for this endpoint. The fragment size determines the +-- size in bytes of each fragment. This need not be the same as the +-- fragmentation limit, although that is certainly a sensible choice. The +-- default for this value is 1KB. +setPacketFragmentSize :: Int -> EndpointConfig -> EndpointConfig +setPacketFragmentSize = setConfigIntVal $ \sz config -> + config { c'reliable_config_t'fragment_size = sz } + +-- | Sets the number of fragments per packet in this endpoint. This is only to +-- make sure that the endpoint has enough buffer space provisioned for incoming +-- packets. Default for this value is 16, and the maximum value is 256. +setMaxNumFragments :: Int -> EndpointConfig -> EndpointConfig +setMaxNumFragments = setConfigIntVal $ \n config -> + config { c'reliable_config_t'max_fragments = n } + +-- | Sets the number of packets for which to store received sequence numbers. +-- The default value is 256. +setAckBufferSize :: Int -> EndpointConfig -> EndpointConfig +setAckBufferSize = setConfigIntVal $ \sz config -> + config { c'reliable_config_t'ack_buffer_size = sz } + +-- | Sets the maximum number of packets for which to store sent packet info. +-- This number reflects the largest number of packets we expect to be in flight +-- at any given time, in order to properly ack them upon receipt of some other +-- endpoint's packets. Also useful for properly computing bandwidth of the +-- endpoint. Default value is 256. +setSentPacketsBufferSize :: Int -> EndpointConfig -> EndpointConfig +setSentPacketsBufferSize = setConfigIntVal $ \sz config -> + config { c'reliable_config_t'sent_packets_buffer_size = sz } + +-- | Sets the maximum number of packets for which to store received packet +-- info. Useful for properly acking packets and for accurately computing +-- bandwidth of the endpoint. Default value is 256. +setReceivedPacketsBufferSize :: Int -> EndpointConfig -> EndpointConfig +setReceivedPacketsBufferSize = setConfigIntVal $ \sz config -> + config { c'reliable_config_t'received_packets_buffer_size = sz } + +-- | Sets the maximum number of in flight packet fragments that we can store +-- in order to properly recreate the packets upon receipt. This buffer is used +-- to process out of order and dropped packets, as fragments from a packet may +-- not arrive contiguously. Default value is 64. +setFragmentReassemblyBufferSize :: Int -> EndpointConfig -> EndpointConfig +setFragmentReassemblyBufferSize = setConfigIntVal $ \sz config -> + config { c'reliable_config_t'fragment_reassembly_buffer_size = sz } + +-- | Sets the round trip time smoothing factor. This is purely for diagnostic +-- purposes when determining what the round trip time is for this endpoint. +-- Smaller numbers will vary the RTT measurement more slowly. Default value is +-- 0.0025f. +setRTTSmoothingFactor :: Float -> EndpointConfig -> EndpointConfig +setRTTSmoothingFactor = setConfigFloatVal $ \factor config -> + config { c'reliable_config_t'rtt_smoothing_factor = factor } + +-- | Sets the packet loss smoothing factor. This is purely for diagnostic +-- purposes when determining what the packet loss rate is for this endpoint. +-- Smaller numbers will vary the packet loss measurement more slowly. +-- Default value is 0.1f. +setPacketLossSmoothingFactor :: Float -> EndpointConfig -> EndpointConfig +setPacketLossSmoothingFactor = setConfigFloatVal $ \factor config -> + config { c'reliable_config_t'packet_loss_smoothing_factor = factor } + +-- | Sets the bandwidth smoothing factor. This is purely for diagnostic +-- purposes when determining what the bandwidth is from this endpoint. Smaller +-- numbers will vary the bandwidth measurement more slowly. Default value is +-- 0.1f. +setBandwidthSmoothingFactor :: Float -> EndpointConfig -> EndpointConfig +setBandwidthSmoothingFactor = setConfigFloatVal $ \factor config -> + config { c'reliable_config_t'bandwidth_smoothing_factor = factor } + +-- | Endpoints support two packet types, either IPV4 or IPV6. +data PacketType = PacketType'IPV4 | PacketType'IPV6 + deriving (Eq, Ord, Show, Read, Bounded, Enum, Generic, Data, Typeable) + +-- | Sets the packet type for this endpoint, which determines the header size +-- that the library needs to allocate in order to properly keep track of the +-- packets. +setPacketType :: PacketType -> EndpointConfig -> EndpointConfig +setPacketType PacketType'IPV4 = setConfig $ \config -> + config { c'reliable_config_t'packet_header_size = 28 } +setPacketType PacketType'IPV6 = setConfig $ \config -> + config { c'reliable_config_t'packet_header_size = 48 } + +-- Utility structure to know what to free when destroying endpoints. +data EndpointCallbacks = EndpointCallbacks + { _endpointCallbacksXmit :: C'transmit_packet_function_t + , _endpointCallbacksRecv :: C'process_packet_function_t + } + +-- | An 'Endpoint' is the main datatype of the reliable.io library. Two +-- endpoints (usually, but not exclusively) on separate hosts represent a +-- connection over an unreliable network, such as the UDP protocol over the +-- internet. The function of an endpoint is to provide a way to administer +-- traffic to the corresponding receiver. It is not responsible for performing +-- the actual sending and receiving of data. +-- +-- Endpoints provide two main services: +-- +-- 1. Breaking down a large packet into fragments, each of which is a +-- predetermined size. +-- 2. Assembling a sequence of fragments from a corresponding endpoint. +-- 3. Notifying the user when a packet has been received (ack'd) by the +-- corresponding endpoint. +-- +-- Packets to be disassembled into fragments and transmitted are passed to the +-- endpoint via the 'sendPacket' function. Fragments that are received from the +-- corresponding endpoint and should be reassmbled are passed to the endpoint +-- via the 'receivePacket' function. These functions only queue the data for +-- processing, but the actual processing of packets only takes place during a +-- call to 'update'. +-- +-- Additionally, each 'Endpoint' keeps track of the metrics associated with it, +-- providing the user with ways to measure the round trip time for each packet, +-- the bandwidth of the connection, and a measurement of the packet loss. +data Endpoint = Endpoint + { _endpointPtr :: Ptr C'reliable_endpoint_t + , _endpointCallbacks :: EndpointCallbacks + } + +-- | Function used by an 'Endpoint' to send packet fragments over the +-- unreliable data channel. One use case would be to have the given data sent +-- to a UDP socket. +type TransmitPacketFunction + = Word16 -- ^ Sequence number of the packet being sent + -> Ptr Word8 -- ^ Pointer to memory containing the packet data + -> Int -- ^ Size of the data in bytes + -> IO () + +-- Utility function for converting TransmitPacketFunctions to the C callback +-- type. +mkTransmitPacketFunction :: TransmitPacketFunction + -> IO C'transmit_packet_function_t +mkTransmitPacketFunction fn = mk'transmit_packet_function_t $ + \_ _ seqNo ptr -> fn seqNo ptr . fromIntegral + +-- | A user function supplied to an 'Endpoint' that handles reassembled packets +-- once they've been received. +type ProcessPacketFunction + = Word16 -- ^ Sequence number of the packet received. + -> Ptr Word8 -- ^ Pointer to the memory containing the packet data + -> Int -- ^ Size of the data in bytes. + -> IO Bool -- ^ Returns true if the packet was successfully processed + +-- Utility function for converting ProcessPacketFunctions to the C callback type. +mkProcessPacketFunction :: ProcessPacketFunction -> IO C'process_packet_function_t +mkProcessPacketFunction fn = mk'process_packet_function_t $ + \_ _ seqNo ptr -> fmap (bool 0 1) . fn seqNo ptr . fromIntegral + +-- | Creates an 'Endpoint'. The two main callbacks required for each endpoint: +-- +-- 1. A 'TransmitPacketFunction' that is able to send packet fragments to a +-- corresponding 'Endpoint' +-- 2. A 'ProcessPacketFunction' that administers the reassmbled packet from a +-- collection of fragments. +-- +-- The 'Double' parameter corresponds to the time (in seconds) at which the +-- endpoint is created. This time value is needed to be in the same domain to +-- subsequent calls to 'update' in order to properly calculate metrics. +createEndpoint :: EndpointConfig + -> Double + -> TransmitPacketFunction + -> ProcessPacketFunction + -> IO Endpoint +createEndpoint cfg t xmitFn recvFn = alloca $ \ptr -> do + generateConfig cfg ptr + config <- peek ptr + xmitCFn <- mkTransmitPacketFunction xmitFn + recvCFn <- mkProcessPacketFunction recvFn + poke ptr $ config { + c'reliable_config_t'transmit_packet_function = xmitCFn, + c'reliable_config_t'process_packet_function = recvCFn + } + endpoint <- c'reliable_endpoint_create ptr (CDouble t) + return $ Endpoint endpoint (EndpointCallbacks xmitCFn recvCFn) + +-- | Destroys an 'Endpoint' and any associated callbacks. +destroyEndpoint :: Endpoint -> IO () +destroyEndpoint (Endpoint ptr cbs) = do + c'reliable_endpoint_destroy ptr + freeHaskellFunPtr $ _endpointCallbacksXmit cbs + freeHaskellFunPtr $ _endpointCallbacksRecv cbs + return () + +-- | Convenience function that follows the +-- <https://wiki.haskell.org/Bracket_pattern Bracket pattern> for encapsulating +-- the resource management associated with interfacing with an 'Endpoint'. +withEndpoint :: EndpointConfig + -> Double + -> TransmitPacketFunction + -> ProcessPacketFunction + -> (Endpoint -> IO a) + -> IO a +withEndpoint cfg t xmit recv fn = do + endpoint <- createEndpoint cfg t xmit recv + result <- fn endpoint + destroyEndpoint endpoint + return result + +-- | Returns the sequence number of the next packet that will be sent from this +-- 'Endpoint'. +nextPacketSequence :: Endpoint -> IO Word16 +nextPacketSequence (Endpoint ptr _) = + c'reliable_endpoint_next_packet_sequence ptr + +-- | @sendPacket e p sz@ will send a packet from 'Endpoint' @e@ with @sz@ bytes +-- whose data resides in the memory pointed to by @p@. If @sz@ is larger than +-- the fragment limit, the packet will be split into multiple fragments. Each +-- fragment will then be sent via the 'TransmitPacketFunction' passed to +-- 'createEndpoint'. Note, this function does not actually send the packet, and +-- rather queues it for sending during the next call to 'update'. +sendPacket :: Endpoint -> Ptr Word8 -> Int -> IO () +sendPacket (Endpoint epPtr _) pktPtr = + c'reliable_endpoint_send_packet epPtr pktPtr . fromIntegral + +-- | @receivePacket e p sz@ will add a packet fragment to 'Endpoint' @e@ with +-- @sz@ bytes whose data resides in the memory pointed to by @p@. Once all of +-- the fragments of a given packet have been received via this function, the +-- 'Endpoint' will pass the reassembled packet to the 'ProcessPacketFunction' +-- passed to 'createEndpoint'. Note, this function does not actually reassemble +-- the packet, and rather queues it for processing during the next call to +-- 'update'. +receivePacket :: Endpoint -> Ptr Word8 -> Int -> IO () +receivePacket (Endpoint epPtr _) pktPtr = + c'reliable_endpoint_receive_packet epPtr pktPtr . fromIntegral + +-- | Returns the list of sequence numbers for the most recently ack'd packets +-- that have been sent from this 'Endpoint'. +getAcks :: Endpoint -> IO [Word16] +getAcks (Endpoint epPtr _) = alloca $ \numAcksPtr -> do + acksPtr <- c'reliable_endpoint_get_acks epPtr numAcksPtr + numAcks <- peek acksPtr + peekArray (fromIntegral numAcks) acksPtr + +-- | Clears the list of sequence numbers for the most recently ack'd packets. +clearAcks :: Endpoint -> IO () +clearAcks (Endpoint ptr _) = c'reliable_endpoint_clear_acks ptr + +-- | Resets the endpoint, including all metrics about network traffic and any +-- information about ack'd packets. +reset :: Endpoint -> IO () +reset (Endpoint ptr _) = c'reliable_endpoint_reset ptr + +-- | Performs the work of updating the endpoint. This sends packets, +-- reassembles packets, and identifies any acks received from the corresponding +-- 'Endpoint'. The time passed to this function must be measured in seconds and +-- correspond to the same time domain as 'createEndpoint'. +update :: Endpoint -> Double -> IO () +update (Endpoint ptr _) = c'reliable_endpoint_update ptr . CDouble + +-- | Returns the measured round trip time for packets sent from this +-- 'Endpoint'. +roundTripTime :: Endpoint -> IO Float +roundTripTime (Endpoint ptr _) = do + (CFloat f) <- c'reliable_endpoint_rtt ptr + return f + +-- | Returns the measured packet loss for packets sent from this 'Endpoint'. +packetLoss :: Endpoint -> IO Float +packetLoss (Endpoint ptr _) = do + (CFloat f) <- c'reliable_endpoint_packet_loss ptr + return f + +-- | Bandwidth measurements taken for each 'Endpoint'. +data BandwidthMeasurements = BandwidthMeasurements + { bandwidthSentKbps :: Float + , bandwidthReceivedKbps :: Float + , bandwidthAckedKbps :: Float + } deriving (Show, Read, Generic) + +-- | Returns the measured bandwidth for data on this 'Endpoint'. +bandwidth :: Endpoint -> IO BandwidthMeasurements +bandwidth (Endpoint ptr _) = + alloca $ \sentPtr -> + alloca $ \receivedPtr -> + alloca $ \ackedPtr -> do + c'reliable_endpoint_bandwidth ptr sentPtr receivedPtr ackedPtr + (CFloat sent) <- peek sentPtr + (CFloat received) <- peek receivedPtr + (CFloat acked) <- peek ackedPtr + return $ BandwidthMeasurements sent received acked + +-- | Counters for metrics that are collected for each 'Endpoint'. These are +-- reset upon calling 'reset' for the given 'Endpoint'. +data Counter + = Counter'NumPacketsSent + | Counter'NumPacketsReceived + | Counter'NumPacketsAcked + | Counter'NumPacketsStale + | Counter'NumPacketsInvalid + | Counter'NumPacketsTooLargeToSend + | Counter'NumPacketsTooLargeToReceive + | Counter'NumFragmentsSent + | Counter'NumFragmentsReceived + | Counter'NumFragmentsInvalid + deriving (Eq, Ord, Show, Read, Enum, Bounded, Generic, Data, Typeable) + +-- | Returns the counter value associated with the 'Counter' for the given +-- 'Endpoint'. +getCounter :: Endpoint -> Counter -> IO Word64 +getCounter (Endpoint ptr _) ctr = + let ctrIdx = case ctr of + Counter'NumPacketsSent -> c'RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_SENT + Counter'NumPacketsReceived -> c'RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_RECEIVED + Counter'NumPacketsAcked -> c'RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_ACKED + Counter'NumPacketsStale -> c'RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_STALE + Counter'NumPacketsInvalid -> c'RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_INVALID + Counter'NumPacketsTooLargeToSend -> c'RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_SEND + Counter'NumPacketsTooLargeToReceive -> c'RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_RECEIVE + Counter'NumFragmentsSent -> c'RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_SENT + Counter'NumFragmentsReceived -> c'RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_RECEIVED + Counter'NumFragmentsInvalid -> c'RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_INVALID + in c'reliable_endpoint_counters ptr >>= flip peekElemOff ctrIdx
+ examples/RunCUnitTests.hs view
@@ -0,0 +1,15 @@+module Main (main) where + +import Bindings.Reliable.IO ( c'RELIABLE_OK + , c'reliable_init, c'reliable_test, c'reliable_term + ) + +main :: IO () +main = do + initResult <- c'reliable_init + if initResult == c'RELIABLE_OK + then return () + else fail "Failed to initialize reliable.io" + c'reliable_test + c'reliable_term + putStrLn "Success!"
+ examples/Soak.hs view
@@ -0,0 +1,143 @@+{-# LANGUAGE RecursiveDo #-} +{-# LANGUAGE TupleSections #-} +module Main (main) where + +-------------------------------------------------------------------------------- + +import Control.Exception (try, AsyncException(..)) +import Control.Monad (forM_, when, unless) +import Data.Unique (hashUnique, newUnique) +import Data.Word (Word8, Word16) +import Foreign.Marshal.Array (peekArray, pokeArray, allocaArray) +import Foreign.Ptr (Ptr, nullPtr) +import System.Environment (getArgs) +import System.IO (hSetBuffering, stdin, BufferMode(..)) + +import Reliable.IO (Endpoint) +import qualified Reliable.IO as Reliable + +-------------------------------------------------------------------------------- + +gMaxPacketBytes :: Num a => a +gMaxPacketBytes = 16 * 1024 + +randomInt :: Int -> Int -> IO Int +randomInt low high = do + randInt <- hashUnique <$> newUnique + return $ low + (randInt `mod` (high - low + 1)) + +seqNoToPktSz :: Word16 -> Int +seqNoToPktSz seqNo = + ((fromIntegral seqNo * 1023) `mod` (gMaxPacketBytes - 2)) + 2 + +initializeSoak :: Double -> IO (Endpoint, Endpoint) +initializeSoak startTime = do + putStrLn "initializing" + Reliable.initialize + Reliable.logLevel Reliable.LogLevel'Debug + + let clientConfig = + Reliable.setPacketFragmentationLimit 500 $ + Reliable.setName "client" $ + Reliable.defaultConfig + + serverConfig = + Reliable.setPacketFragmentationLimit 500 $ + Reliable.setName "server" $ + Reliable.defaultConfig + + recvFn :: Word16 -> Ptr Word8 -> Int -> IO Bool + recvFn _ ptr sz = do + when (ptr == nullPtr) $ fail "Received null pointer" + when (sz <= 0) $ fail "Received empty packet" + when (sz <= 2) $ fail "Received empty header" + when (sz > gMaxPacketBytes) $ fail "Packet too large." + + (s0 : s1 : arr) <- peekArray sz ptr + let seqNo :: Word16 + seqNo = (fromIntegral s0) + (256 * fromIntegral s1) + + unless (sz == seqNoToPktSz seqNo) $ + fail "Sequence number doesn't match packet size" + + forM_ (zip [(2 :: Int)..] arr) $ \(i, dat) -> + unless (fromIntegral dat == (i + fromIntegral seqNo) `mod` 256) $ + fail "Malformed packet" + + return True + + rec clientEndpoint <- Reliable.createEndpoint clientConfig startTime + (xmitFn True) recvFn + serverEndpoint <- Reliable.createEndpoint serverConfig startTime + (xmitFn False) recvFn + + let xmitFn :: Bool -> Word16 -> Ptr Word8 -> Int -> IO () + xmitFn toServer _ ptr sz = do + rand <- randomInt 0 100 + let ep = if toServer then serverEndpoint else clientEndpoint + unless (rand < 5) $ Reliable.receivePacket ep ptr sz + + return (clientEndpoint, serverEndpoint) + +shutdownSoak :: Endpoint -> Endpoint -> IO () +shutdownSoak client server = do + putStrLn "shutdown" + + Reliable.destroyEndpoint client + Reliable.destroyEndpoint server + Reliable.terminate + +iterateSoak :: Endpoint -> Endpoint -> Double -> IO () +iterateSoak client server t = + let generatePacketData :: Word16 -> [Word8] + generatePacketData seqNo = + (fromIntegral $ seqNo `mod` 256) : + (fromIntegral $ (seqNo `div` 256) `mod` 256) : + map (fromIntegral . (+ fromIntegral seqNo)) + [2..(seqNoToPktSz seqNo - 1)] + + sendNextPacket ep ptr = do + seqNo <- Reliable.nextPacketSequence ep + let pkt = generatePacketData seqNo + pokeArray ptr pkt + Reliable.sendPacket ep ptr (length pkt) + + in allocaArray gMaxPacketBytes $ \clientPktPtr -> + allocaArray gMaxPacketBytes $ \serverPktPtr -> do + sendNextPacket client clientPktPtr + sendNextPacket server serverPktPtr + + Reliable.update client t + Reliable.update server t + + Reliable.clearAcks client + Reliable.clearAcks server + +main :: IO () +main = do + hSetBuffering stdin LineBuffering + args <- getArgs + let startTime :: Double + startTime = 100.0 + + numIters :: Maybe Int + numIters = case length args of + 1 -> Just $ read (head args) + _ -> Nothing + + times = case numIters of + Just x -> take x $ map (+startTime) [0.0,0.1..] + Nothing -> map (+startTime) [0.0,0.1..] + + putStrLn "[soak]" + putStrLn $ "num_iterations = " <> show numIters + + (client, server) <- initializeSoak startTime + soakResult <- try $ forM_ times $ iterateSoak client server + + case soakResult of + (Left UserInterrupt) -> return () + (Left e) -> fail (show e) + (Right ()) -> return () + + shutdownSoak client server
+ reliable-io.cabal view
@@ -0,0 +1,75 @@+cabal-version: >= 1.10 +name: reliable-io +version: 0.0.1 +synopsis: Bindings to the low-level reliable.io library. +description: Bindings to the low-level reliable.io library, which come + in two flavors: c-level bindings and a high-level + interface to the library. + + For the low level interface, refer to the + Bindings.Reliable.IO module. + + For the high-level interface, which is a bit closer to + idiomatic Haskell, refer to the Reliable.IO module. +homepage: http://www.github.com/Mokosha/reliable-io +bug-reports: http://www.github.com/Mokosha/reliable-io/issues +license: BSD3 +license-file: LICENSE +author: Pavel Krajcevski +maintainer: krajcevski@gmail.com +copyright: Pavel Krajcevski, 2020 +category: Network +build-type: Simple +extra-source-files: README.md + , reliable.io/reliable.h + , reliable.io/reliable.c + , reliable.io/README.md + , reliable.io/LICENCE + +source-repository head + type: git + location: https://www.github.com/Mokosha/reliable-io.git + +library + ghc-options: -Wall -fPIC + include-dirs: reliable.io + c-sources: reliable.io/reliable.c + cc-options: -std=c99 -DNDEBUG -DRELIABLE_ENABLE_TESTS=1 + + exposed-modules: Bindings.Reliable.IO + Reliable.IO + + build-tools: hsc2hs + build-depends: base >= 4.12 && < 5 + , bindings-DSL + default-language: Haskell2010 + +flag examples + description: If true, build the examples + default: False + +executable reliable-io-c-unit-tests + default-language: Haskell2010 + main-is: RunCUnitTests.hs + hs-source-dirs: examples + ghc-options: -Wall -static -fPIC + build-depends: base > 4 + , reliable-io + + if flag(examples) + buildable: True + else + buildable: False + +executable reliable-io-soak + default-language: Haskell2010 + main-is: Soak.hs + hs-source-dirs: examples + ghc-options: -Wall -static -fPIC + build-depends: base > 4 + , reliable-io + + if flag(examples) + buildable: True + else + buildable: False
+ reliable.io/LICENCE view
@@ -0,0 +1,19 @@+Copyright © 2017 - 2019, The Network Protocol Company, Inc. + +Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: + + 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. + + 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer + in the documentation and/or other materials provided with the distribution. + + 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, +INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE +DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, +SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR +SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, +WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE +USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ reliable.io/README.md view
@@ -0,0 +1,57 @@+[](https://travis-ci.org/networkprotocol/reliable.io) + +# Introduction + +**reliable.io** is a packet acknowlegement system for UDP protocols. + +It has the following features: + +1. Identifies which packets are received by the other side +2. Packet fragmentation and reassembly +3. RTT and packet loss estimates + +reliable.io is stable and well tested having been used in AAA game projects for over 2 years now. + +# Author + +The author of this library is [Glenn Fiedler](https://www.linkedin.com/in/glennfiedler). + +Glenn wrote an article series about the development of this library called [Building a Game Network Protocol](http://gafferongames.com/2016/05/10/building-a-game-network-protocol/). + +Open source libraries by the same author include: [netcode.io](http://netcode.io) and [yojimbo](http://libyojimbo.com) + +# Source Code + +This repository holds the implementation of reliable.io in C. + +Other reliable.io repositories include: + +* [reliable.io Rust implementation](https://github.com/jaynus/reliable.io) + +# Contributors + +These people are awesome: + +* [Walter Pearce](https://github.com/jaynus) - Rust Implementation + +# Sponsors + +**reliable.io** was generously sponsored by: + +* **Gold Sponsors** + * [Remedy Entertainment](http://www.remedygames.com/) + * [Cloud Imperium Games](https://cloudimperiumgames.com) + +* **Silver Sponsors** + * [Moon Studios](http://www.oriblindforest.com/#!moon-3/) + * [The Network Protocol Company](http://www.thenetworkprotocolcompany.com) + +* **Bronze Sponsors** + * [Kite & Lightning](http://kiteandlightning.la/) + * [Data Realms](http://datarealms.com) + +And by individual supporters on Patreon. Thank you. You made this possible! + +# License + +[BSD 3-Clause license](https://opensource.org/licenses/BSD-3-Clause).
+ reliable.io/reliable.c view
@@ -0,0 +1,2135 @@+/* + reliable.io + + Copyright © 2017 - 2019, The Network Protocol Company, Inc. + + Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: + + 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. + + 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer + in the documentation and/or other materials provided with the distribution. + + 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + + THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, + INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR + SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, + WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE + USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +#include "reliable.h" +#include <stdlib.h> +#include <memory.h> +#include <stdio.h> +#include <string.h> +#include <stdarg.h> +#include <inttypes.h> +#include <float.h> +#include <math.h> + +#ifndef RELIABLE_ENABLE_TESTS +#define RELIABLE_ENABLE_TESTS 0 +#endif // #ifndef RELIABLE_ENABLE_TESTS + +#ifndef RELIABLE_ENABLE_LOGGING +#define RELIABLE_ENABLE_LOGGING 1 +#endif // #ifndef RELIABLE_ENABLE_LOGGING + +// ------------------------------------------------------------------ + +static void default_assert_handler( RELIABLE_CONST char * condition, RELIABLE_CONST char * function, RELIABLE_CONST char * file, int line ) +{ + printf( "assert failed: ( %s ), function %s, file %s, line %d\n", condition, function, file, line ); + #if defined( __GNUC__ ) + __builtin_trap(); + #elif defined( _MSC_VER ) + __debugbreak(); + #endif + exit( 1 ); +} + +static int log_level = 0; +static int (*printf_function)( RELIABLE_CONST char *, ... ) = ( int (*)( RELIABLE_CONST char *, ... ) ) printf; +void (*reliable_assert_function)( RELIABLE_CONST char *, RELIABLE_CONST char *, RELIABLE_CONST char * file, int line ) = default_assert_handler; + +void reliable_log_level( int level ) +{ + log_level = level; +} + +void reliable_set_printf_function( int (*function)( RELIABLE_CONST char *, ... ) ) +{ + reliable_assert( function ); + printf_function = function; +} + +void reliable_set_assert_function( void (*function)( RELIABLE_CONST char *, RELIABLE_CONST char *, RELIABLE_CONST char * file, int line ) ) +{ + reliable_assert_function = function; +} + +#if RELIABLE_ENABLE_LOGGING + +void reliable_printf( int level, RELIABLE_CONST char * format, ... ) +{ + if ( level > log_level ) + return; + va_list args; + va_start( args, format ); + char buffer[4*1024]; + vsprintf( buffer, format, args ); + printf_function( "%s", buffer ); + va_end( args ); +} + +#else // #if RELIABLE_ENABLE_LOGGING + +void reliable_printf( int level, RELIABLE_CONST char * format, ... ) +{ + (void) level; + (void) format; +} + +#endif // #if RELIABLE_ENABLE_LOGGING + +void * reliable_default_allocate_function( void * context, uint64_t bytes ) +{ + (void) context; + return malloc( bytes ); +} + +void reliable_default_free_function( void * context, void * pointer ) +{ + (void) context; + free( pointer ); +} + +// ------------------------------------------------------------------ + +int reliable_init(void) +{ + return RELIABLE_OK; +} + +void reliable_term(void) +{ +} + +// --------------------------------------------------------------- + +int reliable_sequence_greater_than( uint16_t s1, uint16_t s2 ) +{ + return ( ( s1 > s2 ) && ( s1 - s2 <= 32768 ) ) || + ( ( s1 < s2 ) && ( s2 - s1 > 32768 ) ); +} + +int reliable_sequence_less_than( uint16_t s1, uint16_t s2 ) +{ + return reliable_sequence_greater_than( s2, s1 ); +} + +// --------------------------------------------------------------- + +struct reliable_sequence_buffer_t +{ + void * allocator_context; + void * (*allocate_function)(void*,uint64_t); + void (*free_function)(void*,void*); + uint16_t sequence; + int num_entries; + int entry_stride; + uint32_t * entry_sequence; + uint8_t * entry_data; +}; + +struct reliable_sequence_buffer_t * reliable_sequence_buffer_create( int num_entries, + int entry_stride, + void * allocator_context, + void * (*allocate_function)(void*,uint64_t), + void (*free_function)(void*,void*) ) +{ + reliable_assert( num_entries > 0 ); + reliable_assert( entry_stride > 0 ); + + if ( allocate_function == NULL ) + { + allocate_function = reliable_default_allocate_function; + } + + if ( free_function == NULL ) + { + free_function = reliable_default_free_function; + } + + struct reliable_sequence_buffer_t * sequence_buffer = (struct reliable_sequence_buffer_t*) + allocate_function( allocator_context, sizeof( struct reliable_sequence_buffer_t ) ); + + sequence_buffer->allocator_context = allocator_context; + sequence_buffer->allocate_function = allocate_function; + sequence_buffer->free_function = free_function; + sequence_buffer->sequence = 0; + sequence_buffer->num_entries = num_entries; + sequence_buffer->entry_stride = entry_stride; + sequence_buffer->entry_sequence = (uint32_t*) allocate_function( allocator_context, num_entries * sizeof( uint32_t ) ); + sequence_buffer->entry_data = (uint8_t*) allocate_function( allocator_context, num_entries * entry_stride ); + reliable_assert( sequence_buffer->entry_sequence ); + reliable_assert( sequence_buffer->entry_data ); + memset( sequence_buffer->entry_sequence, 0xFF, sizeof( uint32_t) * sequence_buffer->num_entries ); + memset( sequence_buffer->entry_data, 0, num_entries * entry_stride ); + + return sequence_buffer; +} + +void reliable_sequence_buffer_destroy( struct reliable_sequence_buffer_t * sequence_buffer ) +{ + reliable_assert( sequence_buffer ); + sequence_buffer->free_function( sequence_buffer->allocator_context, sequence_buffer->entry_sequence ); + sequence_buffer->free_function( sequence_buffer->allocator_context, sequence_buffer->entry_data ); + sequence_buffer->free_function( sequence_buffer->allocator_context, sequence_buffer ); +} + +void reliable_sequence_buffer_reset( struct reliable_sequence_buffer_t * sequence_buffer ) +{ + reliable_assert( sequence_buffer ); + sequence_buffer->sequence = 0; + memset( sequence_buffer->entry_sequence, 0xFF, sizeof( uint32_t) * sequence_buffer->num_entries ); +} + +void reliable_sequence_buffer_remove_entries( struct reliable_sequence_buffer_t * sequence_buffer, + int start_sequence, + int finish_sequence, + void (*cleanup_function)(void*,void*,void(*free_function)(void*,void*)) ) +{ + reliable_assert( sequence_buffer ); + if ( finish_sequence < start_sequence ) + { + finish_sequence += 65536; + } + if ( finish_sequence - start_sequence < sequence_buffer->num_entries ) + { + int sequence; + for ( sequence = start_sequence; sequence <= finish_sequence; ++sequence ) + { + if ( cleanup_function ) + { + cleanup_function( sequence_buffer->entry_data + sequence_buffer->entry_stride * ( sequence % sequence_buffer->num_entries ), + sequence_buffer->allocator_context, + sequence_buffer->free_function ); + } + sequence_buffer->entry_sequence[ sequence % sequence_buffer->num_entries ] = 0xFFFFFFFF; + } + } + else + { + int i; + for ( i = 0; i < sequence_buffer->num_entries; ++i ) + { + if ( cleanup_function ) + { + cleanup_function( sequence_buffer->entry_data + sequence_buffer->entry_stride * i, + sequence_buffer->allocator_context, + sequence_buffer->free_function ); + } + sequence_buffer->entry_sequence[i] = 0xFFFFFFFF; + } + } +} + +int reliable_sequence_buffer_test_insert( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t sequence ) +{ + return reliable_sequence_less_than( sequence, sequence_buffer->sequence - ((uint16_t)sequence_buffer->num_entries) ) ? ((uint16_t)0) : ((uint16_t)1); +} + +void * reliable_sequence_buffer_insert( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t sequence ) +{ + reliable_assert( sequence_buffer ); + if ( reliable_sequence_less_than( sequence, sequence_buffer->sequence - ((uint16_t)sequence_buffer->num_entries) ) ) + { + return NULL; + } + if ( reliable_sequence_greater_than( sequence + 1, sequence_buffer->sequence ) ) + { + reliable_sequence_buffer_remove_entries( sequence_buffer, sequence_buffer->sequence, sequence, NULL ); + sequence_buffer->sequence = sequence + 1; + } + int index = sequence % sequence_buffer->num_entries; + sequence_buffer->entry_sequence[index] = sequence; + return sequence_buffer->entry_data + index * sequence_buffer->entry_stride; +} + +void reliable_sequence_buffer_advance( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t sequence ) +{ + reliable_assert( sequence_buffer ); + if ( reliable_sequence_greater_than( sequence + 1, sequence_buffer->sequence ) ) + { + reliable_sequence_buffer_remove_entries( sequence_buffer, sequence_buffer->sequence, sequence, NULL ); + sequence_buffer->sequence = sequence + 1; + } +} + +void * reliable_sequence_buffer_insert_with_cleanup( struct reliable_sequence_buffer_t * sequence_buffer, + uint16_t sequence, + void (*cleanup_function)(void*,void*,void(*free_function)(void*,void*)) ) +{ + reliable_assert( sequence_buffer ); + if ( reliable_sequence_greater_than( sequence + 1, sequence_buffer->sequence ) ) + { + reliable_sequence_buffer_remove_entries( sequence_buffer, sequence_buffer->sequence, sequence, cleanup_function ); + sequence_buffer->sequence = sequence + 1; + } + else if ( reliable_sequence_less_than( sequence, sequence_buffer->sequence - ((uint16_t)sequence_buffer->num_entries) ) ) + { + return NULL; + } + int index = sequence % sequence_buffer->num_entries; + if ( sequence_buffer->entry_sequence[index] != 0xFFFFFFFF ) + { + cleanup_function( sequence_buffer->entry_data + sequence_buffer->entry_stride * ( sequence % sequence_buffer->num_entries ), + sequence_buffer->allocator_context, + sequence_buffer->free_function ); + } + sequence_buffer->entry_sequence[index] = sequence; + return sequence_buffer->entry_data + index * sequence_buffer->entry_stride; +} + +void reliable_sequence_buffer_remove( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t sequence ) +{ + reliable_assert( sequence_buffer ); + sequence_buffer->entry_sequence[ sequence % sequence_buffer->num_entries ] = 0xFFFFFFFF; +} + +void reliable_sequence_buffer_remove_with_cleanup( struct reliable_sequence_buffer_t * sequence_buffer, + uint16_t sequence, + void (*cleanup_function)(void*,void*,void(*free_function)(void*,void*)) ) +{ + reliable_assert( sequence_buffer ); + int index = sequence % sequence_buffer->num_entries; + if ( sequence_buffer->entry_sequence[index] != 0xFFFFFFFF ) + { + sequence_buffer->entry_sequence[index] = 0xFFFFFFFF; + cleanup_function( sequence_buffer->entry_data + sequence_buffer->entry_stride * index, sequence_buffer->allocator_context, sequence_buffer->free_function ); + } +} + +int reliable_sequence_buffer_available( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t sequence ) +{ + reliable_assert( sequence_buffer ); + return sequence_buffer->entry_sequence[ sequence % sequence_buffer->num_entries ] == 0xFFFFFFFF; +} + +int reliable_sequence_buffer_exists( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t sequence ) +{ + reliable_assert( sequence_buffer ); + return sequence_buffer->entry_sequence[ sequence % sequence_buffer->num_entries ] == (uint32_t) sequence; +} + +void * reliable_sequence_buffer_find( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t sequence ) +{ + reliable_assert( sequence_buffer ); + int index = sequence % sequence_buffer->num_entries; + return ( ( sequence_buffer->entry_sequence[index] == (uint32_t) sequence ) ) ? ( sequence_buffer->entry_data + index * sequence_buffer->entry_stride ) : NULL; + +} + +void * reliable_sequence_buffer_at_index( struct reliable_sequence_buffer_t * sequence_buffer, int index ) +{ + reliable_assert( sequence_buffer ); + reliable_assert( index >= 0 ); + reliable_assert( index < sequence_buffer->num_entries ); + return sequence_buffer->entry_sequence[index] != 0xFFFFFFFF ? ( sequence_buffer->entry_data + index * sequence_buffer->entry_stride ) : NULL; +} + +void reliable_sequence_buffer_generate_ack_bits( struct reliable_sequence_buffer_t * sequence_buffer, uint16_t * ack, uint32_t * ack_bits ) +{ + reliable_assert( sequence_buffer ); + reliable_assert( ack ); + reliable_assert( ack_bits ); + *ack = sequence_buffer->sequence - 1; + *ack_bits = 0; + uint32_t mask = 1; + int i; + for ( i = 0; i < 32; ++i ) + { + uint16_t sequence = *ack - ((uint16_t)i); + if ( reliable_sequence_buffer_exists( sequence_buffer, sequence ) ) + *ack_bits |= mask; + mask <<= 1; + } +} + +// --------------------------------------------------------------- + +void reliable_write_uint8( uint8_t ** p, uint8_t value ) +{ + **p = value; + ++(*p); +} + +void reliable_write_uint16( uint8_t ** p, uint16_t value ) +{ + (*p)[0] = value & 0xFF; + (*p)[1] = value >> 8; + *p += 2; +} + +void reliable_write_uint32( uint8_t ** p, uint32_t value ) +{ + (*p)[0] = value & 0xFF; + (*p)[1] = ( value >> 8 ) & 0xFF; + (*p)[2] = ( value >> 16 ) & 0xFF; + (*p)[3] = value >> 24; + *p += 4; +} + +void reliable_write_uint64( uint8_t ** p, uint64_t value ) +{ + (*p)[0] = value & 0xFF; + (*p)[1] = ( value >> 8 ) & 0xFF; + (*p)[2] = ( value >> 16 ) & 0xFF; + (*p)[3] = ( value >> 24 ) & 0xFF; + (*p)[4] = ( value >> 32 ) & 0xFF; + (*p)[5] = ( value >> 40 ) & 0xFF; + (*p)[6] = ( value >> 48 ) & 0xFF; + (*p)[7] = value >> 56; + *p += 8; +} + +void reliable_write_bytes( uint8_t ** p, uint8_t * byte_array, int num_bytes ) +{ + int i; + for ( i = 0; i < num_bytes; ++i ) + { + reliable_write_uint8( p, byte_array[i] ); + } +} + +uint8_t reliable_read_uint8( uint8_t ** p ) +{ + uint8_t value = **p; + ++(*p); + return value; +} + +uint16_t reliable_read_uint16( uint8_t ** p ) +{ + uint16_t value; + value = (*p)[0]; + value |= ( ( (uint16_t)( (*p)[1] ) ) << 8 ); + *p += 2; + return value; +} + +uint32_t reliable_read_uint32( uint8_t ** p ) +{ + uint32_t value; + value = (*p)[0]; + value |= ( ( (uint32_t)( (*p)[1] ) ) << 8 ); + value |= ( ( (uint32_t)( (*p)[2] ) ) << 16 ); + value |= ( ( (uint32_t)( (*p)[3] ) ) << 24 ); + *p += 4; + return value; +} + +uint64_t reliable_read_uint64( uint8_t ** p ) +{ + uint64_t value; + value = (*p)[0]; + value |= ( ( (uint64_t)( (*p)[1] ) ) << 8 ); + value |= ( ( (uint64_t)( (*p)[2] ) ) << 16 ); + value |= ( ( (uint64_t)( (*p)[3] ) ) << 24 ); + value |= ( ( (uint64_t)( (*p)[4] ) ) << 32 ); + value |= ( ( (uint64_t)( (*p)[5] ) ) << 40 ); + value |= ( ( (uint64_t)( (*p)[6] ) ) << 48 ); + value |= ( ( (uint64_t)( (*p)[7] ) ) << 56 ); + *p += 8; + return value; +} + +void reliable_read_bytes( uint8_t ** p, uint8_t * byte_array, int num_bytes ) +{ + int i; + for ( i = 0; i < num_bytes; ++i ) + { + byte_array[i] = reliable_read_uint8( p ); + } +} + +// --------------------------------------------------------------- + +struct reliable_fragment_reassembly_data_t +{ + uint16_t sequence; + uint16_t ack; + uint32_t ack_bits; + int num_fragments_received; + int num_fragments_total; + uint8_t * packet_data; + int packet_bytes; + int packet_header_bytes; + uint8_t fragment_received[256]; +}; + +void reliable_fragment_reassembly_data_cleanup( void * data, void * allocator_context, void (*free_function)(void*,void*) ) + +{ + reliable_assert( free_function ); + struct reliable_fragment_reassembly_data_t * reassembly_data = (struct reliable_fragment_reassembly_data_t*) data; + if ( reassembly_data->packet_data ) + { + free_function( allocator_context, reassembly_data->packet_data ); + reassembly_data->packet_data = NULL; + } +} + +// --------------------------------------------------------------- + +struct reliable_endpoint_t +{ + void * allocator_context; + void * (*allocate_function)(void*,uint64_t); + void (*free_function)(void*,void*); + struct reliable_config_t config; + double time; + float rtt; + float packet_loss; + float sent_bandwidth_kbps; + float received_bandwidth_kbps; + float acked_bandwidth_kbps; + int num_acks; + uint16_t * acks; + uint16_t sequence; + struct reliable_sequence_buffer_t * sent_packets; + struct reliable_sequence_buffer_t * received_packets; + struct reliable_sequence_buffer_t * fragment_reassembly; + uint64_t counters[RELIABLE_ENDPOINT_NUM_COUNTERS]; +}; + +struct reliable_sent_packet_data_t +{ + double time; + uint32_t acked : 1; + uint32_t packet_bytes : 31; +}; + +struct reliable_received_packet_data_t +{ + double time; + uint32_t packet_bytes; +}; + +void reliable_default_config( struct reliable_config_t * config ) +{ + reliable_assert( config ); + memset( config, 0, sizeof( struct reliable_config_t ) ); + config->name[0] = 'e'; + config->name[1] = 'n'; + config->name[2] = 'd'; + config->name[3] = 'p'; + config->name[4] = 'o'; + config->name[5] = 'i'; + config->name[6] = 'n'; + config->name[7] = 't'; + config->name[8] = '\0'; + config->max_packet_size = 16 * 1024; + config->fragment_above = 1024; + config->max_fragments = 16; + config->fragment_size = 1024; + config->ack_buffer_size = 256; + config->sent_packets_buffer_size = 256; + config->received_packets_buffer_size = 256; + config->fragment_reassembly_buffer_size = 64; + config->rtt_smoothing_factor = 0.0025f; + config->packet_loss_smoothing_factor = 0.1f; + config->bandwidth_smoothing_factor = 0.1f; + config->packet_header_size = 28; // note: UDP over IPv4 = 20 + 8 bytes, UDP over IPv6 = 40 + 8 bytes +} + +struct reliable_endpoint_t * reliable_endpoint_create( struct reliable_config_t * config, double time ) +{ + reliable_assert( config ); + reliable_assert( config->max_packet_size > 0 ); + reliable_assert( config->fragment_above > 0 ); + reliable_assert( config->max_fragments > 0 ); + reliable_assert( config->max_fragments <= 256 ); + reliable_assert( config->fragment_size > 0 ); + reliable_assert( config->ack_buffer_size > 0 ); + reliable_assert( config->sent_packets_buffer_size > 0 ); + reliable_assert( config->received_packets_buffer_size > 0 ); + reliable_assert( config->transmit_packet_function != NULL ); + reliable_assert( config->process_packet_function != NULL ); + + void * allocator_context = config->allocator_context; + void * (*allocate_function)(void*,uint64_t) = config->allocate_function; + void (*free_function)(void*,void*) = config->free_function; + + if ( allocate_function == NULL ) + { + allocate_function = reliable_default_allocate_function; + } + + if ( free_function == NULL ) + { + free_function = reliable_default_free_function; + } + + struct reliable_endpoint_t * endpoint = (struct reliable_endpoint_t*) allocate_function( allocator_context, sizeof( struct reliable_endpoint_t ) ); + + reliable_assert( endpoint ); + + memset( endpoint, 0, sizeof( struct reliable_endpoint_t ) ); + + endpoint->allocator_context = allocator_context; + endpoint->allocate_function = allocate_function; + endpoint->free_function = free_function; + endpoint->config = *config; + endpoint->time = time; + + endpoint->acks = (uint16_t*) allocate_function( allocator_context, config->ack_buffer_size * sizeof( uint16_t ) ); + + endpoint->sent_packets = reliable_sequence_buffer_create( config->sent_packets_buffer_size, + sizeof( struct reliable_sent_packet_data_t ), + allocator_context, + allocate_function, + free_function ); + + endpoint->received_packets = reliable_sequence_buffer_create( config->received_packets_buffer_size, + sizeof( struct reliable_received_packet_data_t ), + allocator_context, + allocate_function, + free_function ); + + endpoint->fragment_reassembly = reliable_sequence_buffer_create( config->fragment_reassembly_buffer_size, + sizeof( struct reliable_fragment_reassembly_data_t ), + allocator_context, + allocate_function, + free_function ); + + memset( endpoint->acks, 0, config->ack_buffer_size * sizeof( uint16_t ) ); + + return endpoint; +} + +void reliable_endpoint_destroy( struct reliable_endpoint_t * endpoint ) +{ + reliable_assert( endpoint ); + reliable_assert( endpoint->acks ); + reliable_assert( endpoint->sent_packets ); + reliable_assert( endpoint->received_packets ); + + int i; + for ( i = 0; i < endpoint->config.fragment_reassembly_buffer_size; ++i ) + { + struct reliable_fragment_reassembly_data_t * reassembly_data = (struct reliable_fragment_reassembly_data_t*) + reliable_sequence_buffer_at_index( endpoint->fragment_reassembly, i ); + + if ( reassembly_data && reassembly_data->packet_data ) + { + endpoint->free_function( endpoint->allocator_context, reassembly_data->packet_data ); + reassembly_data->packet_data = NULL; + } + } + + endpoint->free_function( endpoint->allocator_context, endpoint->acks ); + + reliable_sequence_buffer_destroy( endpoint->sent_packets ); + reliable_sequence_buffer_destroy( endpoint->received_packets ); + reliable_sequence_buffer_destroy( endpoint->fragment_reassembly ); + + endpoint->free_function( endpoint->allocator_context, endpoint ); +} + +uint16_t reliable_endpoint_next_packet_sequence( struct reliable_endpoint_t * endpoint ) +{ + reliable_assert( endpoint ); + return endpoint->sequence; +} + +int reliable_write_packet_header( uint8_t * packet_data, uint16_t sequence, uint16_t ack, uint32_t ack_bits ) +{ + uint8_t * p = packet_data; + + uint8_t prefix_byte = 0; + + if ( ( ack_bits & 0x000000FF ) != 0x000000FF ) + { + prefix_byte |= (1<<1); + } + + if ( ( ack_bits & 0x0000FF00 ) != 0x0000FF00 ) + { + prefix_byte |= (1<<2); + } + + if ( ( ack_bits & 0x00FF0000 ) != 0x00FF0000 ) + { + prefix_byte |= (1<<3); + } + + if ( ( ack_bits & 0xFF000000 ) != 0xFF000000 ) + { + prefix_byte |= (1<<4); + } + + int sequence_difference = sequence - ack; + if ( sequence_difference < 0 ) + sequence_difference += 65536; + if ( sequence_difference <= 255 ) + prefix_byte |= (1<<5); + + reliable_write_uint8( &p, prefix_byte ); + + reliable_write_uint16( &p, sequence ); + + if ( sequence_difference <= 255 ) + { + reliable_write_uint8( &p, (uint8_t) sequence_difference ); + } + else + { + reliable_write_uint16( &p, ack ); + } + + if ( ( ack_bits & 0x000000FF ) != 0x000000FF ) + { + reliable_write_uint8( &p, (uint8_t) ( ack_bits & 0x000000FF ) ); + } + + if ( ( ack_bits & 0x0000FF00 ) != 0x0000FF00 ) + { + reliable_write_uint8( &p, (uint8_t) ( ( ack_bits & 0x0000FF00 ) >> 8 ) ); + } + + if ( ( ack_bits & 0x00FF0000 ) != 0x00FF0000 ) + { + reliable_write_uint8( &p, (uint8_t) ( ( ack_bits & 0x00FF0000 ) >> 16 ) ); + } + + if ( ( ack_bits & 0xFF000000 ) != 0xFF000000 ) + { + reliable_write_uint8( &p, (uint8_t) ( ( ack_bits & 0xFF000000 ) >> 24 ) ); + } + + reliable_assert( p - packet_data <= RELIABLE_MAX_PACKET_HEADER_BYTES ); + + return (int) ( p - packet_data ); +} + +void reliable_endpoint_send_packet( struct reliable_endpoint_t * endpoint, uint8_t * packet_data, int packet_bytes ) +{ + reliable_assert( endpoint ); + reliable_assert( packet_data ); + reliable_assert( packet_bytes > 0 ); + + if ( packet_bytes > endpoint->config.max_packet_size ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] packet too large to send. packet is %d bytes, maximum is %d\n", + endpoint->config.name, packet_bytes, endpoint->config.max_packet_size ); + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_SEND]++; + return; + } + + uint16_t sequence = endpoint->sequence++; + uint16_t ack; + uint32_t ack_bits; + + reliable_sequence_buffer_generate_ack_bits( endpoint->received_packets, &ack, &ack_bits ); + + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] sending packet %d\n", endpoint->config.name, sequence ); + + struct reliable_sent_packet_data_t * sent_packet_data = (struct reliable_sent_packet_data_t*) reliable_sequence_buffer_insert( endpoint->sent_packets, sequence ); + + reliable_assert( sent_packet_data ); + + sent_packet_data->time = endpoint->time; + sent_packet_data->packet_bytes = endpoint->config.packet_header_size + packet_bytes; + sent_packet_data->acked = 0; + + if ( packet_bytes <= endpoint->config.fragment_above ) + { + // regular packet + + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] sending packet %d without fragmentation\n", endpoint->config.name, sequence ); + + uint8_t * transmit_packet_data = (uint8_t*) endpoint->allocate_function( endpoint->allocator_context, packet_bytes + RELIABLE_MAX_PACKET_HEADER_BYTES ); + + int packet_header_bytes = reliable_write_packet_header( transmit_packet_data, sequence, ack, ack_bits ); + + memcpy( transmit_packet_data + packet_header_bytes, packet_data, packet_bytes ); + + endpoint->config.transmit_packet_function( endpoint->config.context, endpoint->config.index, sequence, transmit_packet_data, packet_header_bytes + packet_bytes ); + + endpoint->free_function( endpoint->allocator_context, transmit_packet_data ); + } + else + { + // fragmented packet + + uint8_t packet_header[RELIABLE_MAX_PACKET_HEADER_BYTES]; + + memset( packet_header, 0, RELIABLE_MAX_PACKET_HEADER_BYTES ); + + int packet_header_bytes = reliable_write_packet_header( packet_header, sequence, ack, ack_bits ); + + int num_fragments = ( packet_bytes / endpoint->config.fragment_size ) + ( ( packet_bytes % endpoint->config.fragment_size ) != 0 ? 1 : 0 ); + + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] sending packet %d as %d fragments\n", endpoint->config.name, sequence, num_fragments ); + + reliable_assert( num_fragments >= 1 ); + reliable_assert( num_fragments <= endpoint->config.max_fragments ); + + int fragment_buffer_size = RELIABLE_FRAGMENT_HEADER_BYTES + RELIABLE_MAX_PACKET_HEADER_BYTES + endpoint->config.fragment_size; + + uint8_t * fragment_packet_data = (uint8_t*) endpoint->allocate_function( endpoint->allocator_context, fragment_buffer_size ); + + uint8_t * q = packet_data; + + uint8_t * end = q + packet_bytes; + + int fragment_id; + for ( fragment_id = 0; fragment_id < num_fragments; ++fragment_id ) + { + uint8_t * p = fragment_packet_data; + + reliable_write_uint8( &p, 1 ); + reliable_write_uint16( &p, sequence ); + reliable_write_uint8( &p, (uint8_t) fragment_id ); + reliable_write_uint8( &p, (uint8_t) ( num_fragments - 1 ) ); + + if ( fragment_id == 0 ) + { + memcpy( p, packet_header, packet_header_bytes ); + p += packet_header_bytes; + } + + int bytes_to_copy = endpoint->config.fragment_size; + if ( q + bytes_to_copy > end ) + { + bytes_to_copy = (int) ( end - q ); + } + + memcpy( p, q, bytes_to_copy ); + + p += bytes_to_copy; + q += bytes_to_copy; + + int fragment_packet_bytes = (int) ( p - fragment_packet_data ); + + endpoint->config.transmit_packet_function( endpoint->config.context, endpoint->config.index, sequence, fragment_packet_data, fragment_packet_bytes ); + + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_SENT]++; + } + + endpoint->free_function( endpoint->allocator_context, fragment_packet_data ); + } + + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_SENT]++; +} + +int reliable_read_packet_header( RELIABLE_CONST char * name, uint8_t * packet_data, int packet_bytes, uint16_t * sequence, uint16_t * ack, uint32_t * ack_bits ) +{ + if ( packet_bytes < 3 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] packet too small for packet header (1)\n", name ); + return -1; + } + + uint8_t * p = packet_data; + + uint8_t prefix_byte = reliable_read_uint8( &p ); + + if ( ( prefix_byte & 1 ) != 0 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] prefix byte does not indicate a regular packet\n", name ); + return -1; + } + + *sequence = reliable_read_uint16( &p ); + + if ( prefix_byte & (1<<5) ) + { + if ( packet_bytes < 3 + 1 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] packet too small for packet header (2)\n", name ); + return -1; + } + uint8_t sequence_difference = reliable_read_uint8( &p ); + *ack = *sequence - sequence_difference; + } + else + { + if ( packet_bytes < 3 + 2 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] packet too small for packet header (3)\n", name ); + return -1; + } + *ack = reliable_read_uint16( &p ); + } + + int expected_bytes = 0; + int i; + for ( i = 1; i <= 4; ++i ) + { + if ( prefix_byte & (1<<i) ) + { + expected_bytes++; + } + } + if ( packet_bytes < ( p - packet_data ) + expected_bytes ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] packet too small for packet header (4)\n", name ); + return -1; + } + + *ack_bits = 0xFFFFFFFF; + + if ( prefix_byte & (1<<1) ) + { + *ack_bits &= 0xFFFFFF00; + *ack_bits |= (uint32_t) ( reliable_read_uint8( &p ) ); + } + + if ( prefix_byte & (1<<2) ) + { + *ack_bits &= 0xFFFF00FF; + *ack_bits |= (uint32_t) ( reliable_read_uint8( &p ) ) << 8; + } + + if ( prefix_byte & (1<<3) ) + { + *ack_bits &= 0xFF00FFFF; + *ack_bits |= (uint32_t) ( reliable_read_uint8( &p ) ) << 16; + } + + if ( prefix_byte & (1<<4) ) + { + *ack_bits &= 0x00FFFFFF; + *ack_bits |= (uint32_t) ( reliable_read_uint8( &p ) ) << 24; + } + + return (int) ( p - packet_data ); +} + +int reliable_read_fragment_header( char * name, + uint8_t * packet_data, + int packet_bytes, + int max_fragments, + int fragment_size, + int * fragment_id, + int * num_fragments, + int * fragment_bytes, + uint16_t * sequence, + uint16_t * ack, + uint32_t * ack_bits ) +{ + if ( packet_bytes < RELIABLE_FRAGMENT_HEADER_BYTES ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] packet is too small to read fragment header\n", name ); + return -1; + } + + uint8_t * p = packet_data; + + uint8_t prefix_byte =reliable_read_uint8( &p ); + if ( prefix_byte != 1 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] prefix byte is not a fragment\n", name ); + return -1; + } + + *sequence = reliable_read_uint16( &p ); + *fragment_id = (int) reliable_read_uint8( &p ); + *num_fragments = ( (int) reliable_read_uint8( &p ) ) + 1; + + if ( *num_fragments > max_fragments ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] num fragments %d outside of range of max fragments %d\n", name, *num_fragments, max_fragments ); + return -1; + } + + if ( *fragment_id >= *num_fragments ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] fragment id %d outside of range of num fragments %d\n", name, *fragment_id, *num_fragments ); + return -1; + } + + *fragment_bytes = packet_bytes - RELIABLE_FRAGMENT_HEADER_BYTES; + + uint16_t packet_sequence = 0; + uint16_t packet_ack = 0; + uint32_t packet_ack_bits = 0; + + if ( *fragment_id == 0 ) + { + int packet_header_bytes = reliable_read_packet_header( name, + packet_data + RELIABLE_FRAGMENT_HEADER_BYTES, + packet_bytes, + &packet_sequence, + &packet_ack, + &packet_ack_bits ); + + if ( packet_header_bytes < 0 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] bad packet header in fragment\n", name ); + return -1; + } + + if ( packet_sequence != *sequence ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] bad packet sequence in fragment. expected %d, got %d\n", name, *sequence, packet_sequence ); + return -1; + } + + *fragment_bytes = packet_bytes - packet_header_bytes - RELIABLE_FRAGMENT_HEADER_BYTES; + } + + *ack = packet_ack; + *ack_bits = packet_ack_bits; + + if ( *fragment_bytes > fragment_size ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] fragment bytes %d > fragment size %d\n", name, *fragment_bytes, fragment_size ); + return - 1; + } + + if ( *fragment_id != *num_fragments - 1 && *fragment_bytes != fragment_size ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] fragment %d is %d bytes, which is not the expected fragment size %d\n", + name, *fragment_id, *fragment_bytes, fragment_size ); + return -1; + } + + return (int) ( p - packet_data ); +} + +void reliable_store_fragment_data( struct reliable_fragment_reassembly_data_t * reassembly_data, + uint16_t sequence, + uint16_t ack, + uint32_t ack_bits, + int fragment_id, + int fragment_size, + uint8_t * fragment_data, + int fragment_bytes ) +{ + if ( fragment_id == 0 ) + { + uint8_t packet_header[RELIABLE_MAX_PACKET_HEADER_BYTES]; + + memset( packet_header, 0, RELIABLE_MAX_PACKET_HEADER_BYTES ); + + reassembly_data->packet_header_bytes = reliable_write_packet_header( packet_header, sequence, ack, ack_bits ); + + memcpy( reassembly_data->packet_data + RELIABLE_MAX_PACKET_HEADER_BYTES - reassembly_data->packet_header_bytes, + packet_header, + reassembly_data->packet_header_bytes ); + + fragment_data += reassembly_data->packet_header_bytes; + fragment_bytes -= reassembly_data->packet_header_bytes; + } + + if ( fragment_id == reassembly_data->num_fragments_total - 1 ) + { + reassembly_data->packet_bytes = ( reassembly_data->num_fragments_total - 1 ) * fragment_size + fragment_bytes; + } + + memcpy( reassembly_data->packet_data + RELIABLE_MAX_PACKET_HEADER_BYTES + fragment_id * fragment_size, fragment_data, fragment_bytes ); +} + +void reliable_endpoint_receive_packet( struct reliable_endpoint_t * endpoint, uint8_t * packet_data, int packet_bytes ) +{ + reliable_assert( endpoint ); + reliable_assert( packet_data ); + reliable_assert( packet_bytes > 0 ); + + if ( packet_bytes > endpoint->config.max_packet_size ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] packet too large to receive. packet is %d bytes, maximum is %d\n", + endpoint->config.name, packet_bytes, endpoint->config.max_packet_size ); + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_RECEIVE]++; + return; + } + + uint8_t prefix_byte = packet_data[0]; + + if ( ( prefix_byte & 1 ) == 0 ) + { + // regular packet + + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_RECEIVED]++; + + uint16_t sequence; + uint16_t ack; + uint32_t ack_bits; + + int packet_header_bytes = reliable_read_packet_header( endpoint->config.name, packet_data, packet_bytes, &sequence, &ack, &ack_bits ); + if ( packet_header_bytes < 0 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] ignoring invalid packet. could not read packet header\n", endpoint->config.name ); + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_INVALID]++; + return; + } + + if ( !reliable_sequence_buffer_test_insert( endpoint->received_packets, sequence ) ) + { + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] ignoring stale packet %d\n", endpoint->config.name, sequence ); + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_STALE]++; + return; + } + + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] processing packet %d\n", endpoint->config.name, sequence ); + + if ( endpoint->config.process_packet_function( endpoint->config.context, + endpoint->config.index, + sequence, + packet_data + packet_header_bytes, + packet_bytes - packet_header_bytes ) ) + { + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] process packet %d successful\n", endpoint->config.name, sequence ); + + struct reliable_received_packet_data_t * received_packet_data = (struct reliable_received_packet_data_t*) + reliable_sequence_buffer_insert( endpoint->received_packets, sequence ); + + reliable_sequence_buffer_advance( endpoint->fragment_reassembly, sequence ); + + reliable_assert( received_packet_data ); + + received_packet_data->time = endpoint->time; + received_packet_data->packet_bytes = endpoint->config.packet_header_size + packet_bytes; + + int i; + for ( i = 0; i < 32; ++i ) + { + if ( ack_bits & 1 ) + { + uint16_t ack_sequence = ack - ((uint16_t)i); + + struct reliable_sent_packet_data_t * sent_packet_data = (struct reliable_sent_packet_data_t*) + reliable_sequence_buffer_find( endpoint->sent_packets, ack_sequence ); + + if ( sent_packet_data && !sent_packet_data->acked && endpoint->num_acks < endpoint->config.ack_buffer_size ) + { + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] acked packet %d\n", endpoint->config.name, ack_sequence ); + endpoint->acks[endpoint->num_acks++] = ack_sequence; + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_ACKED]++; + sent_packet_data->acked = 1; + + float rtt = (float) ( endpoint->time - sent_packet_data->time ) * 1000.0f; + reliable_assert( rtt >= 0.0 ); + if ( ( endpoint->rtt == 0.0f && rtt > 0.0f ) || fabs( endpoint->rtt - rtt ) < 0.00001 ) + { + endpoint->rtt = rtt; + } + else + { + endpoint->rtt += ( rtt - endpoint->rtt ) * endpoint->config.rtt_smoothing_factor; + } + } + } + ack_bits >>= 1; + } + } + else + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] process packet failed\n", endpoint->config.name ); + } + } + else + { + // fragment packet + + int fragment_id; + int num_fragments; + int fragment_bytes; + + uint16_t sequence; + uint16_t ack; + uint32_t ack_bits; + + int fragment_header_bytes = reliable_read_fragment_header( endpoint->config.name, + packet_data, + packet_bytes, + endpoint->config.max_fragments, + endpoint->config.fragment_size, + &fragment_id, + &num_fragments, + &fragment_bytes, + &sequence, + &ack, + &ack_bits ); + + if ( fragment_header_bytes < 0 ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] ignoring invalid fragment. could not read fragment header\n", endpoint->config.name ); + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_INVALID]++; + return; + } + + struct reliable_fragment_reassembly_data_t * reassembly_data = (struct reliable_fragment_reassembly_data_t*) + reliable_sequence_buffer_find( endpoint->fragment_reassembly, sequence ); + + if ( !reassembly_data ) + { + reassembly_data = (struct reliable_fragment_reassembly_data_t*) + reliable_sequence_buffer_insert_with_cleanup( endpoint->fragment_reassembly, sequence, reliable_fragment_reassembly_data_cleanup ); + + if ( !reassembly_data ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] ignoring invalid fragment. could not insert in reassembly buffer (stale)\n", endpoint->config.name ); + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_INVALID]++; + return; + } + + reliable_sequence_buffer_advance( endpoint->received_packets, sequence ); + + int packet_buffer_size = RELIABLE_MAX_PACKET_HEADER_BYTES + num_fragments * endpoint->config.fragment_size; + + reassembly_data->sequence = sequence; + reassembly_data->ack = 0; + reassembly_data->ack_bits = 0; + reassembly_data->num_fragments_received = 0; + reassembly_data->num_fragments_total = num_fragments; + reassembly_data->packet_data = (uint8_t*) endpoint->allocate_function( endpoint->allocator_context, packet_buffer_size ); + reassembly_data->packet_bytes = 0; + memset( reassembly_data->fragment_received, 0, sizeof( reassembly_data->fragment_received ) ); + } + + if ( num_fragments != (int) reassembly_data->num_fragments_total ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] ignoring invalid fragment. fragment count mismatch. expected %d, got %d\n", + endpoint->config.name, (int) reassembly_data->num_fragments_total, num_fragments ); + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_INVALID]++; + return; + } + + if ( reassembly_data->fragment_received[fragment_id] ) + { + reliable_printf( RELIABLE_LOG_LEVEL_ERROR, "[%s] ignoring fragment %d of packet %d. fragment already received\n", + endpoint->config.name, fragment_id, sequence ); + return; + } + + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] received fragment %d of packet %d (%d/%d)\n", + endpoint->config.name, fragment_id, sequence, reassembly_data->num_fragments_received+1, num_fragments ); + + reassembly_data->num_fragments_received++; + reassembly_data->fragment_received[fragment_id] = 1; + + reliable_store_fragment_data( reassembly_data, + sequence, + ack, + ack_bits, + fragment_id, + endpoint->config.fragment_size, + packet_data + fragment_header_bytes, + packet_bytes - fragment_header_bytes ); + + if ( reassembly_data->num_fragments_received == reassembly_data->num_fragments_total ) + { + reliable_printf( RELIABLE_LOG_LEVEL_DEBUG, "[%s] completed reassembly of packet %d\n", endpoint->config.name, sequence ); + + reliable_endpoint_receive_packet( endpoint, + reassembly_data->packet_data + RELIABLE_MAX_PACKET_HEADER_BYTES - reassembly_data->packet_header_bytes, + reassembly_data->packet_header_bytes + reassembly_data->packet_bytes ); + + reliable_sequence_buffer_remove_with_cleanup( endpoint->fragment_reassembly, sequence, reliable_fragment_reassembly_data_cleanup ); + } + + endpoint->counters[RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_RECEIVED]++; + } +} + +void reliable_endpoint_free_packet( struct reliable_endpoint_t * endpoint, void * packet ) +{ + reliable_assert( endpoint ); + reliable_assert( packet ); + endpoint->free_function( endpoint->allocator_context, packet ); +} + +uint16_t * reliable_endpoint_get_acks( struct reliable_endpoint_t * endpoint, int * num_acks ) +{ + reliable_assert( endpoint ); + reliable_assert( num_acks ); + *num_acks = endpoint->num_acks; + return endpoint->acks; +} + +void reliable_endpoint_clear_acks( struct reliable_endpoint_t * endpoint ) +{ + reliable_assert( endpoint ); + endpoint->num_acks = 0; +} + +void reliable_endpoint_reset( struct reliable_endpoint_t * endpoint ) +{ + reliable_assert( endpoint ); + + endpoint->num_acks = 0; + endpoint->sequence = 0; + + memset( endpoint->acks, 0, endpoint->config.ack_buffer_size * sizeof( uint16_t ) ); + memset( endpoint->counters, 0, RELIABLE_ENDPOINT_NUM_COUNTERS * sizeof( uint64_t ) ); + + int i; + for ( i = 0; i < endpoint->config.fragment_reassembly_buffer_size; ++i ) + { + struct reliable_fragment_reassembly_data_t * reassembly_data = (struct reliable_fragment_reassembly_data_t*) + reliable_sequence_buffer_at_index( endpoint->fragment_reassembly, i ); + + if ( reassembly_data && reassembly_data->packet_data ) + { + endpoint->free_function( endpoint->allocator_context, reassembly_data->packet_data ); + reassembly_data->packet_data = NULL; + } + } + + reliable_sequence_buffer_reset( endpoint->sent_packets ); + reliable_sequence_buffer_reset( endpoint->received_packets ); + reliable_sequence_buffer_reset( endpoint->fragment_reassembly ); +} + +void reliable_endpoint_update( struct reliable_endpoint_t * endpoint, double time ) +{ + reliable_assert( endpoint ); + + endpoint->time = time; + + // calculate packet loss + { + uint32_t base_sequence = ( endpoint->sent_packets->sequence - endpoint->config.sent_packets_buffer_size + 1 ) + 0xFFFF; + int i; + int num_dropped = 0; + int num_samples = endpoint->config.sent_packets_buffer_size / 2; + for ( i = 0; i < num_samples; ++i ) + { + uint16_t sequence = (uint16_t) ( base_sequence + i ); + struct reliable_sent_packet_data_t * sent_packet_data = (struct reliable_sent_packet_data_t*) + reliable_sequence_buffer_find( endpoint->sent_packets, sequence ); + if ( sent_packet_data && !sent_packet_data->acked ) + { + num_dropped++; + } + } + float packet_loss = ( (float) num_dropped ) / ( (float) num_samples ) * 100.0f; + if ( fabs( endpoint->packet_loss - packet_loss ) > 0.00001 ) + { + endpoint->packet_loss += ( packet_loss - endpoint->packet_loss ) * endpoint->config.packet_loss_smoothing_factor; + } + else + { + endpoint->packet_loss = packet_loss; + } + } + + // calculate sent bandwidth + { + uint32_t base_sequence = ( endpoint->sent_packets->sequence - endpoint->config.sent_packets_buffer_size + 1 ) + 0xFFFF; + int i; + int bytes_sent = 0; + double start_time = FLT_MAX; + double finish_time = 0.0; + int num_samples = endpoint->config.sent_packets_buffer_size / 2; + for ( i = 0; i < num_samples; ++i ) + { + uint16_t sequence = (uint16_t) ( base_sequence + i ); + struct reliable_sent_packet_data_t * sent_packet_data = (struct reliable_sent_packet_data_t*) + reliable_sequence_buffer_find( endpoint->sent_packets, sequence ); + if ( !sent_packet_data ) + { + continue; + } + bytes_sent += sent_packet_data->packet_bytes; + if ( sent_packet_data->time < start_time ) + { + start_time = sent_packet_data->time; + } + if ( sent_packet_data->time > finish_time ) + { + finish_time = sent_packet_data->time; + } + } + if ( start_time != FLT_MAX && finish_time != 0.0 ) + { + float sent_bandwidth_kbps = (float) ( ( (double) bytes_sent ) / ( finish_time - start_time ) * 8.0f / 1000.0f ); + if ( fabs( endpoint->sent_bandwidth_kbps - sent_bandwidth_kbps ) > 0.00001 ) + { + endpoint->sent_bandwidth_kbps += ( sent_bandwidth_kbps - endpoint->sent_bandwidth_kbps ) * endpoint->config.bandwidth_smoothing_factor; + } + else + { + endpoint->sent_bandwidth_kbps = sent_bandwidth_kbps; + } + } + } + + // calculate received bandwidth + { + uint32_t base_sequence = ( endpoint->received_packets->sequence - endpoint->config.received_packets_buffer_size + 1 ) + 0xFFFF; + int i; + int bytes_sent = 0; + double start_time = FLT_MAX; + double finish_time = 0.0; + int num_samples = endpoint->config.received_packets_buffer_size / 2; + for ( i = 0; i < num_samples; ++i ) + { + uint16_t sequence = (uint16_t) ( base_sequence + i ); + struct reliable_received_packet_data_t * received_packet_data = (struct reliable_received_packet_data_t*) + reliable_sequence_buffer_find( endpoint->received_packets, sequence ); + if ( !received_packet_data ) + { + continue; + } + bytes_sent += received_packet_data->packet_bytes; + if ( received_packet_data->time < start_time ) + { + start_time = received_packet_data->time; + } + if ( received_packet_data->time > finish_time ) + { + finish_time = received_packet_data->time; + } + } + if ( start_time != FLT_MAX && finish_time != 0.0 ) + { + float received_bandwidth_kbps = (float) ( ( (double) bytes_sent ) / ( finish_time - start_time ) * 8.0f / 1000.0f ); + if ( fabs( endpoint->received_bandwidth_kbps - received_bandwidth_kbps ) > 0.00001 ) + { + endpoint->received_bandwidth_kbps += ( received_bandwidth_kbps - endpoint->received_bandwidth_kbps ) * endpoint->config.bandwidth_smoothing_factor; + } + else + { + endpoint->received_bandwidth_kbps = received_bandwidth_kbps; + } + } + } + + // calculate acked bandwidth + { + uint32_t base_sequence = ( endpoint->sent_packets->sequence - endpoint->config.sent_packets_buffer_size + 1 ) + 0xFFFF; + int i; + int bytes_sent = 0; + double start_time = FLT_MAX; + double finish_time = 0.0; + int num_samples = endpoint->config.sent_packets_buffer_size / 2; + for ( i = 0; i < num_samples; ++i ) + { + uint16_t sequence = (uint16_t) ( base_sequence + i ); + struct reliable_sent_packet_data_t * sent_packet_data = (struct reliable_sent_packet_data_t*) + reliable_sequence_buffer_find( endpoint->sent_packets, sequence ); + if ( !sent_packet_data || !sent_packet_data->acked ) + { + continue; + } + bytes_sent += sent_packet_data->packet_bytes; + if ( sent_packet_data->time < start_time ) + { + start_time = sent_packet_data->time; + } + if ( sent_packet_data->time > finish_time ) + { + finish_time = sent_packet_data->time; + } + } + if ( start_time != FLT_MAX && finish_time != 0.0 ) + { + float acked_bandwidth_kbps = (float) ( ( (double) bytes_sent ) / ( finish_time - start_time ) * 8.0f / 1000.0f ); + if ( fabs( endpoint->acked_bandwidth_kbps - acked_bandwidth_kbps ) > 0.00001 ) + { + endpoint->acked_bandwidth_kbps += ( acked_bandwidth_kbps - endpoint->acked_bandwidth_kbps ) * endpoint->config.bandwidth_smoothing_factor; + } + else + { + endpoint->acked_bandwidth_kbps = acked_bandwidth_kbps; + } + } + } +} + +float reliable_endpoint_rtt( struct reliable_endpoint_t * endpoint ) +{ + reliable_assert( endpoint ); + return endpoint->rtt; +} + +float reliable_endpoint_packet_loss( struct reliable_endpoint_t * endpoint ) +{ + reliable_assert( endpoint ); + return endpoint->packet_loss; +} + +void reliable_endpoint_bandwidth( struct reliable_endpoint_t * endpoint, float * sent_bandwidth_kbps, float * received_bandwidth_kbps, float * acked_bandwidth_kbps ) +{ + reliable_assert( endpoint ); + reliable_assert( sent_bandwidth_kbps ); + reliable_assert( acked_bandwidth_kbps ); + reliable_assert( received_bandwidth_kbps ); + *sent_bandwidth_kbps = endpoint->sent_bandwidth_kbps; + *received_bandwidth_kbps = endpoint->received_bandwidth_kbps; + *acked_bandwidth_kbps = endpoint->acked_bandwidth_kbps; +} + +RELIABLE_CONST uint64_t * reliable_endpoint_counters( struct reliable_endpoint_t * endpoint ) +{ + reliable_assert( endpoint ); + return endpoint->counters; +} + +// --------------------------------------------------------------- + +#if RELIABLE_ENABLE_TESTS + +#include <stdio.h> +#include <stdlib.h> +#include <memory.h> + +static void check_handler( RELIABLE_CONST char * condition, + RELIABLE_CONST char * function, + RELIABLE_CONST char * file, + int line ) +{ + printf( "check failed: ( %s ), function %s, file %s, line %d\n", condition, function, file, line ); +#ifndef NDEBUG + #if defined( __GNUC__ ) + __builtin_trap(); + #elif defined( _MSC_VER ) + __debugbreak(); + #endif +#endif + exit( 1 ); +} + +#define check( condition ) \ +do \ +{ \ + if ( !(condition) ) \ + { \ + check_handler( #condition, (RELIABLE_CONST char*) __FUNCTION__, __FILE__, __LINE__ ); \ + } \ +} while(0) + +static void test_endian() +{ + uint32_t value = 0x11223344; + + char * bytes = (char*) &value; + +#if RELIABLE_LITTLE_ENDIAN + + check( bytes[0] == 0x44 ); + check( bytes[1] == 0x33 ); + check( bytes[2] == 0x22 ); + check( bytes[3] == 0x11 ); + +#else // #if RELIABLE_LITTLE_ENDIAN + + check( bytes[3] == 0x44 ); + check( bytes[2] == 0x33 ); + check( bytes[1] == 0x22 ); + check( bytes[0] == 0x11 ); + +#endif // #if RELIABLE_LITTLE_ENDIAN +} + +struct test_sequence_data_t +{ + uint16_t sequence; +}; + +#define TEST_SEQUENCE_BUFFER_SIZE 256 + +static void test_sequence_buffer() +{ + struct reliable_sequence_buffer_t * sequence_buffer = reliable_sequence_buffer_create( TEST_SEQUENCE_BUFFER_SIZE, + sizeof( struct test_sequence_data_t ), + NULL, + NULL, + NULL ); + + check( sequence_buffer ); + check( sequence_buffer->sequence == 0 ); + check( sequence_buffer->num_entries == TEST_SEQUENCE_BUFFER_SIZE ); + check( sequence_buffer->entry_stride == sizeof( struct test_sequence_data_t ) ); + + int i; + for ( i = 0; i < TEST_SEQUENCE_BUFFER_SIZE; ++i ) + { + check( reliable_sequence_buffer_find( sequence_buffer, ((uint16_t)i) ) == NULL ); + } + + for ( i = 0; i <= TEST_SEQUENCE_BUFFER_SIZE*4; ++i ) + { + struct test_sequence_data_t * entry = (struct test_sequence_data_t*) reliable_sequence_buffer_insert( sequence_buffer, ((uint16_t)i) ); + check( entry ); + entry->sequence = (uint16_t) i; + check( sequence_buffer->sequence == i + 1 ); + } + + for ( i = 0; i <= TEST_SEQUENCE_BUFFER_SIZE; ++i ) + { + struct test_sequence_data_t * entry = (struct test_sequence_data_t*) reliable_sequence_buffer_insert( sequence_buffer, ((uint16_t)i) ); + check( entry == NULL ); + } + + int index = TEST_SEQUENCE_BUFFER_SIZE * 4; + for ( i = 0; i < TEST_SEQUENCE_BUFFER_SIZE; ++i ) + { + struct test_sequence_data_t * entry = (struct test_sequence_data_t*) reliable_sequence_buffer_find( sequence_buffer, (uint16_t) index ); + check( entry ); + check( entry->sequence == (uint32_t) index ); + index--; + } + + reliable_sequence_buffer_reset( sequence_buffer ); + + check( sequence_buffer ); + check( sequence_buffer->sequence == 0 ); + check( sequence_buffer->num_entries == TEST_SEQUENCE_BUFFER_SIZE ); + check( sequence_buffer->entry_stride == sizeof( struct test_sequence_data_t ) ); + + for ( i = 0; i < TEST_SEQUENCE_BUFFER_SIZE; ++i ) + { + check( reliable_sequence_buffer_find( sequence_buffer, (uint16_t) i ) == NULL ); + } + + reliable_sequence_buffer_destroy( sequence_buffer ); +} + +static void test_generate_ack_bits() +{ + struct reliable_sequence_buffer_t * sequence_buffer = reliable_sequence_buffer_create( TEST_SEQUENCE_BUFFER_SIZE, + sizeof( struct test_sequence_data_t ), + NULL, + NULL, + NULL ); + + uint16_t ack = 0; + uint32_t ack_bits = 0xFFFFFFFF; + + reliable_sequence_buffer_generate_ack_bits( sequence_buffer, &ack, &ack_bits ); + check( ack == 0xFFFF ); + check( ack_bits == 0 ); + + int i; + for ( i = 0; i <= TEST_SEQUENCE_BUFFER_SIZE; ++i ) + { + reliable_sequence_buffer_insert( sequence_buffer, (uint16_t) i ); + } + + reliable_sequence_buffer_generate_ack_bits( sequence_buffer, &ack, &ack_bits ); + check( ack == TEST_SEQUENCE_BUFFER_SIZE ); + check( ack_bits == 0xFFFFFFFF ); + + reliable_sequence_buffer_reset( sequence_buffer ); + + uint16_t input_acks[] = { 1, 5, 9, 11 }; + int input_num_acks = sizeof( input_acks ) / sizeof( uint16_t ); + for ( i = 0; i < input_num_acks; ++i ) + { + reliable_sequence_buffer_insert( sequence_buffer, input_acks[i] ); + } + + reliable_sequence_buffer_generate_ack_bits( sequence_buffer, &ack, &ack_bits ); + + check( ack == 11 ); + check( ack_bits == ( 1 | (1<<(11-9)) | (1<<(11-5)) | (1<<(11-1)) ) ); + + reliable_sequence_buffer_destroy( sequence_buffer ); +} + +static void test_packet_header() +{ + uint16_t write_sequence; + uint16_t write_ack; + uint32_t write_ack_bits; + + uint16_t read_sequence; + uint16_t read_ack; + uint32_t read_ack_bits; + + uint8_t packet_data[RELIABLE_MAX_PACKET_HEADER_BYTES]; + + // worst case, sequence and ack are far apart, no packets acked. + + write_sequence = 10000; + write_ack = 100; + write_ack_bits = 0; + + int bytes_written = reliable_write_packet_header( packet_data, write_sequence, write_ack, write_ack_bits ); + + check( bytes_written == RELIABLE_MAX_PACKET_HEADER_BYTES ); + + int bytes_read = reliable_read_packet_header( "test_packet_header", packet_data, bytes_written, &read_sequence, &read_ack, &read_ack_bits ); + + check( bytes_read == bytes_written ); + + check( read_sequence == write_sequence ); + check( read_ack == write_ack ); + check( read_ack_bits == write_ack_bits ); + + // rare case. sequence and ack are far apart, significant # of acks are missing + + write_sequence = 10000; + write_ack = 100; + write_ack_bits = 0xFEFEFFFE; + + bytes_written = reliable_write_packet_header( packet_data, write_sequence, write_ack, write_ack_bits ); + + check( bytes_written == 1 + 2 + 2 + 3 ); + + bytes_read = reliable_read_packet_header( "test_packet_header", packet_data, bytes_written, &read_sequence, &read_ack, &read_ack_bits ); + + check( bytes_read == bytes_written ); + + check( read_sequence == write_sequence ); + check( read_ack == write_ack ); + check( read_ack_bits == write_ack_bits ); + + // common case under packet loss. sequence and ack are close together, some acks are missing + + write_sequence = 200; + write_ack = 100; + write_ack_bits = 0xFFFEFFFF; + + bytes_written = reliable_write_packet_header( packet_data, write_sequence, write_ack, write_ack_bits ); + + check( bytes_written == 1 + 2 + 1 + 1 ); + + bytes_read = reliable_read_packet_header( "test_packet_header", packet_data, bytes_written, &read_sequence, &read_ack, &read_ack_bits ); + + check( bytes_read == bytes_written ); + + check( read_sequence == write_sequence ); + check( read_ack == write_ack ); + check( read_ack_bits == write_ack_bits ); + + // ideal case. no packet loss. + + write_sequence = 200; + write_ack = 100; + write_ack_bits = 0xFFFFFFFF; + + bytes_written = reliable_write_packet_header( packet_data, write_sequence, write_ack, write_ack_bits ); + + check( bytes_written == 1 + 2 + 1 ); + + bytes_read = reliable_read_packet_header( "test_packet_header", packet_data, bytes_written, &read_sequence, &read_ack, &read_ack_bits ); + + check( bytes_read == bytes_written ); + + check( read_sequence == write_sequence ); + check( read_ack == write_ack ); + check( read_ack_bits == write_ack_bits ); +} + +struct test_context_t +{ + int drop; + struct reliable_endpoint_t * sender; + struct reliable_endpoint_t * receiver; +}; + +static void test_transmit_packet_function( void * _context, int index, uint16_t sequence, uint8_t * packet_data, int packet_bytes ) +{ + (void) sequence; + + struct test_context_t * context = (struct test_context_t*) _context; + + if ( context->drop ) + { + return; + } + + if ( index == 0 ) + { + reliable_endpoint_receive_packet( context->receiver, packet_data, packet_bytes ); + } + else if ( index == 1 ) + { + reliable_endpoint_receive_packet( context->sender, packet_data, packet_bytes ); + } +} + +static int test_process_packet_function( void * _context, int index, uint16_t sequence, uint8_t * packet_data, int packet_bytes ) +{ + struct test_context_t * context = (struct test_context_t*) _context; + + (void) context; + (void) index; + (void) sequence; + (void) packet_data; + (void) packet_bytes; + + return 1; +} + +#define TEST_ACKS_NUM_ITERATIONS 256 + +static void test_acks() +{ + double time = 100.0; + + struct test_context_t context; + memset( &context, 0, sizeof( context ) ); + + struct reliable_config_t sender_config; + struct reliable_config_t receiver_config; + + reliable_default_config( &sender_config ); + reliable_default_config( &receiver_config ); + + sender_config.context = &context; + sender_config.index = 0; + sender_config.transmit_packet_function = &test_transmit_packet_function; + sender_config.process_packet_function = &test_process_packet_function; + + receiver_config.context = &context; + receiver_config.index = 1; + receiver_config.transmit_packet_function = &test_transmit_packet_function; + receiver_config.process_packet_function = &test_process_packet_function; + + context.sender = reliable_endpoint_create( &sender_config, time ); + context.receiver = reliable_endpoint_create( &receiver_config, time ); + + double delta_time = 0.01; + + int i; + for ( i = 0; i < TEST_ACKS_NUM_ITERATIONS; ++i ) + { + uint8_t dummy_packet[8]; + memset( dummy_packet, 0, sizeof( dummy_packet ) ); + + reliable_endpoint_send_packet( context.sender, dummy_packet, sizeof( dummy_packet ) ); + reliable_endpoint_send_packet( context.receiver, dummy_packet, sizeof( dummy_packet ) ); + + reliable_endpoint_update( context.sender, time ); + reliable_endpoint_update( context.receiver, time ); + + time += delta_time; + } + + uint8_t sender_acked_packet[TEST_ACKS_NUM_ITERATIONS]; + memset( sender_acked_packet, 0, sizeof( sender_acked_packet ) ); + int sender_num_acks; + uint16_t * sender_acks = reliable_endpoint_get_acks( context.sender, &sender_num_acks ); + for ( i = 0; i < sender_num_acks; ++i ) + { + if ( sender_acks[i] < TEST_ACKS_NUM_ITERATIONS ) + { + sender_acked_packet[sender_acks[i]] = 1; + } + } + for ( i = 0; i < TEST_ACKS_NUM_ITERATIONS / 2; ++i ) + { + check( sender_acked_packet[i] == 1 ); + } + + uint8_t receiver_acked_packet[TEST_ACKS_NUM_ITERATIONS]; + memset( receiver_acked_packet, 0, sizeof( receiver_acked_packet ) ); + int receiver_num_acks; + uint16_t * receiver_acks = reliable_endpoint_get_acks( context.receiver, &receiver_num_acks ); + for ( i = 0; i < receiver_num_acks; ++i ) + { + if ( receiver_acks[i] < TEST_ACKS_NUM_ITERATIONS ) + receiver_acked_packet[receiver_acks[i]] = 1; + } + for ( i = 0; i < TEST_ACKS_NUM_ITERATIONS / 2; ++i ) + { + check( receiver_acked_packet[i] == 1 ); + } + + reliable_endpoint_destroy( context.sender ); + reliable_endpoint_destroy( context.receiver ); +} + +static void test_acks_packet_loss() +{ + double time = 100.0; + + struct test_context_t context; + memset( &context, 0, sizeof( context ) ); + + struct reliable_config_t sender_config; + struct reliable_config_t receiver_config; + + reliable_default_config( &sender_config ); + reliable_default_config( &receiver_config ); + + sender_config.context = &context; + sender_config.index = 0; + sender_config.transmit_packet_function = &test_transmit_packet_function; + sender_config.process_packet_function = &test_process_packet_function; + + receiver_config.context = &context; + receiver_config.index = 1; + receiver_config.transmit_packet_function = &test_transmit_packet_function; + receiver_config.process_packet_function = &test_process_packet_function; + + context.sender = reliable_endpoint_create( &sender_config, time ); + context.receiver = reliable_endpoint_create( &receiver_config, time ); + + const double delta_time = 0.1f; + + int i; + for ( i = 0; i < TEST_ACKS_NUM_ITERATIONS; ++i ) + { + uint8_t dummy_packet[8]; + memset( dummy_packet, 0, sizeof( dummy_packet ) ); + + context.drop = ( i % 2 ); + + reliable_endpoint_send_packet( context.sender, dummy_packet, sizeof( dummy_packet ) ); + reliable_endpoint_send_packet( context.receiver, dummy_packet, sizeof( dummy_packet ) ); + + reliable_endpoint_update( context.sender, time ); + reliable_endpoint_update( context.receiver, time ); + + time += delta_time; + } + + uint8_t sender_acked_packet[TEST_ACKS_NUM_ITERATIONS]; + memset( sender_acked_packet, 0, sizeof( sender_acked_packet ) ); + int sender_num_acks; + uint16_t * sender_acks = reliable_endpoint_get_acks( context.sender, &sender_num_acks ); + for ( i = 0; i < sender_num_acks; ++i ) + { + if ( sender_acks[i] < TEST_ACKS_NUM_ITERATIONS ) + { + sender_acked_packet[sender_acks[i]] = 1; + } + } + for ( i = 0; i < TEST_ACKS_NUM_ITERATIONS / 2; ++i ) + { + check( sender_acked_packet[i] == (i+1) % 2 ); + } + + uint8_t receiver_acked_packet[TEST_ACKS_NUM_ITERATIONS]; + memset( receiver_acked_packet, 0, sizeof( receiver_acked_packet ) ); + int receiver_num_acks; + uint16_t * receiver_acks = reliable_endpoint_get_acks( context.sender, &receiver_num_acks ); + for ( i = 0; i < receiver_num_acks; ++i ) + { + if ( receiver_acks[i] < TEST_ACKS_NUM_ITERATIONS ) + { + receiver_acked_packet[receiver_acks[i]] = 1; + } + } + for ( i = 0; i < TEST_ACKS_NUM_ITERATIONS / 2; ++i ) + { + check( receiver_acked_packet[i] == (i+1) % 2 ); + } + + reliable_endpoint_destroy( context.sender ); + reliable_endpoint_destroy( context.receiver ); +} + +#define TEST_MAX_PACKET_BYTES (4*1024) + +static int generate_packet_data( uint16_t sequence, uint8_t * packet_data ) +{ + int packet_bytes = ( ( (int)sequence * 1023 ) % ( TEST_MAX_PACKET_BYTES - 2 ) ) + 2; + reliable_assert( packet_bytes >= 2 ); + reliable_assert( packet_bytes <= TEST_MAX_PACKET_BYTES ); + packet_data[0] = (uint8_t) ( sequence & 0xFF ); + packet_data[1] = (uint8_t) ( (sequence>>8) & 0xFF ); + int i; + for ( i = 2; i < packet_bytes; ++i ) + { + packet_data[i] = (uint8_t) ( ( (int)i + sequence ) % 256 ); + } + return packet_bytes; +} + +static void validate_packet_data( uint8_t * packet_data, int packet_bytes ) +{ + reliable_assert( packet_bytes >= 2 ); + reliable_assert( packet_bytes <= TEST_MAX_PACKET_BYTES ); + uint16_t sequence = 0; + sequence |= (uint16_t) packet_data[0]; + sequence |= ( (uint16_t) packet_data[1] ) << 8; + check( packet_bytes == ( ( (int)sequence * 1023 ) % ( TEST_MAX_PACKET_BYTES - 2 ) ) + 2 ); + int i; + for ( i = 2; i < packet_bytes; ++i ) + { + check( packet_data[i] == (uint8_t) ( ( (int)i + sequence ) % 256 ) ); + } +} + +static int test_process_packet_function_validate( void * context, int index, uint16_t sequence, uint8_t * packet_data, int packet_bytes ) +{ + reliable_assert( packet_data ); + reliable_assert( packet_bytes > 0 ); + reliable_assert( packet_bytes <= TEST_MAX_PACKET_BYTES ); + + (void) context; + (void) index; + (void) sequence; + + validate_packet_data( packet_data, packet_bytes ); + + return 1; +} + +void test_packets() +{ + double time = 100.0; + + struct test_context_t context; + memset( &context, 0, sizeof( context ) ); + + struct reliable_config_t sender_config; + struct reliable_config_t receiver_config; + + reliable_default_config( &sender_config ); + reliable_default_config( &receiver_config ); + + sender_config.fragment_above = 500; + receiver_config.fragment_above = 500; + +#if defined(_MSC_VER) + strcpy_s( sender_config.name, sizeof( sender_config.name ), "sender" ); +#else + strcpy( sender_config.name, "sender" ); +#endif + sender_config.context = &context; + sender_config.index = 0; + sender_config.transmit_packet_function = &test_transmit_packet_function; + sender_config.process_packet_function = &test_process_packet_function_validate; + +#if defined(_MSC_VER) + strcpy_s( receiver_config.name, sizeof( receiver_config.name ), "receiver" ); +#else + strcpy( receiver_config.name, "receiver" ); +#endif + receiver_config.context = &context; + receiver_config.index = 1; + receiver_config.transmit_packet_function = &test_transmit_packet_function; + receiver_config.process_packet_function = &test_process_packet_function_validate; + + context.sender = reliable_endpoint_create( &sender_config, time ); + context.receiver = reliable_endpoint_create( &receiver_config, time ); + + double delta_time = 0.1; + + int i; + for ( i = 0; i < 16; ++i ) + { + { + uint8_t packet_data[TEST_MAX_PACKET_BYTES]; + uint16_t sequence = reliable_endpoint_next_packet_sequence( context.sender ); + int packet_bytes = generate_packet_data( sequence, packet_data ); + reliable_endpoint_send_packet( context.sender, packet_data, packet_bytes ); + } + + { + uint8_t packet_data[TEST_MAX_PACKET_BYTES]; + uint16_t sequence = reliable_endpoint_next_packet_sequence( context.sender ); + int packet_bytes = generate_packet_data( sequence, packet_data ); + reliable_endpoint_send_packet( context.sender, packet_data, packet_bytes ); + } + + reliable_endpoint_update( context.sender, time ); + reliable_endpoint_update( context.receiver, time ); + + reliable_endpoint_clear_acks( context.sender ); + reliable_endpoint_clear_acks( context.receiver ); + + time += delta_time; + } + + reliable_endpoint_destroy( context.sender ); + reliable_endpoint_destroy( context.receiver ); +} + +void test_sequence_buffer_rollover() +{ + double time = 100.0; + + struct test_context_t context; + memset( &context, 0, sizeof( context ) ); + + struct reliable_config_t sender_config; + struct reliable_config_t receiver_config; + + reliable_default_config( &sender_config ); + reliable_default_config( &receiver_config ); + + sender_config.fragment_above = 500; + receiver_config.fragment_above = 500; + +#if defined(_MSC_VER) + strcpy_s( sender_config.name, sizeof( sender_config.name ), "sender" ); +#else + strcpy( sender_config.name, "sender" ); +#endif + sender_config.context = &context; + sender_config.index = 0; + sender_config.transmit_packet_function = &test_transmit_packet_function; + sender_config.process_packet_function = &test_process_packet_function; + +#if defined(_MSC_VER) + strcpy_s( receiver_config.name, sizeof( receiver_config.name ), "receiver" ); +#else + strcpy( receiver_config.name, "receiver" ); +#endif + receiver_config.context = &context; + receiver_config.index = 1; + receiver_config.transmit_packet_function = &test_transmit_packet_function; + receiver_config.process_packet_function = &test_process_packet_function; + + context.sender = reliable_endpoint_create( &sender_config, time ); + context.receiver = reliable_endpoint_create( &receiver_config, time ); + + int num_packets_sent = 0; + int i; + for (i = 0; i <= 32767; ++i) + { + uint8_t packet_data[16]; + int packet_bytes = sizeof( packet_data ) / sizeof( uint8_t ); + reliable_endpoint_next_packet_sequence( context.sender ); + reliable_endpoint_send_packet( context.sender, packet_data, packet_bytes ); + + ++num_packets_sent; + } + + uint8_t packet_data[TEST_MAX_PACKET_BYTES]; + int packet_bytes = sizeof( packet_data ) / sizeof( uint8_t ); + reliable_endpoint_next_packet_sequence( context.sender ); + reliable_endpoint_send_packet( context.sender, packet_data, packet_bytes ); + ++num_packets_sent; + + RELIABLE_CONST uint64_t * receiver_counters = reliable_endpoint_counters( context.receiver ); + + check( receiver_counters[RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_RECEIVED] == (uint16_t) num_packets_sent ); + check( receiver_counters[RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_INVALID] == 0 ); + + reliable_endpoint_destroy( context.sender ); + reliable_endpoint_destroy( context.receiver ); +} + +#define RUN_TEST( test_function ) \ + do \ + { \ + printf( #test_function "\n" ); \ + test_function(); \ + } \ + while (0) + +void reliable_test() +{ + //while ( 1 ) + { + RUN_TEST( test_endian ); + RUN_TEST( test_sequence_buffer ); + RUN_TEST( test_generate_ack_bits ); + RUN_TEST( test_packet_header ); + RUN_TEST( test_acks ); + RUN_TEST( test_acks_packet_loss ); + RUN_TEST( test_packets ); + RUN_TEST( test_sequence_buffer_rollover ); + } +} + +#endif // #if RELIABLE_ENABLE_TESTS
+ reliable.io/reliable.h view
@@ -0,0 +1,165 @@+/* + reliable.io + + Copyright © 2017 - 2019, The Network Protocol Company, Inc. + + Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: + + 1. Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. + + 2. Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer + in the documentation and/or other materials provided with the distribution. + + 3. Neither the name of the copyright holder nor the names of its contributors may be used to endorse or promote products derived + from this software without specific prior written permission. + + THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, + INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE + DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR + SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, + WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE + USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. +*/ + +#ifndef RELIABLE_H +#define RELIABLE_H + +#include <stdint.h> + +#if defined(__386__) || defined(i386) || defined(__i386__) \ + || defined(__X86) || defined(_M_IX86) \ + || defined(_M_X64) || defined(__x86_64__) \ + || defined(alpha) || defined(__alpha) || defined(__alpha__) \ + || defined(_M_ALPHA) \ + || defined(ARM) || defined(_ARM) || defined(__arm__) \ + || defined(WIN32) || defined(_WIN32) || defined(__WIN32__) \ + || defined(_WIN32_WCE) || defined(__NT__) \ + || defined(__MIPSEL__) + #define RELIABLE_LITTLE_ENDIAN 1 +#else + #define RELIABLE_BIG_ENDIAN 1 +#endif + +#define RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_SENT 0 +#define RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_RECEIVED 1 +#define RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_ACKED 2 +#define RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_STALE 3 +#define RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_INVALID 4 +#define RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_SEND 5 +#define RELIABLE_ENDPOINT_COUNTER_NUM_PACKETS_TOO_LARGE_TO_RECEIVE 6 +#define RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_SENT 7 +#define RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_RECEIVED 8 +#define RELIABLE_ENDPOINT_COUNTER_NUM_FRAGMENTS_INVALID 9 +#define RELIABLE_ENDPOINT_NUM_COUNTERS 10 + +#define RELIABLE_MAX_PACKET_HEADER_BYTES 9 +#define RELIABLE_FRAGMENT_HEADER_BYTES 5 + +#define RELIABLE_LOG_LEVEL_NONE 0 +#define RELIABLE_LOG_LEVEL_ERROR 1 +#define RELIABLE_LOG_LEVEL_INFO 2 +#define RELIABLE_LOG_LEVEL_DEBUG 3 + +#define RELIABLE_OK 1 +#define RELIABLE_ERROR 0 + +#ifdef __cplusplus +#define RELIABLE_CONST const +extern "C" { +#else +#if defined(__STDC__) +#define RELIABLE_CONST const +#else +#define RELIABLE_CONST +#endif +#endif + +int reliable_init(void); + +void reliable_term(void); + +struct reliable_config_t +{ + char name[256]; + void * context; + int index; + int max_packet_size; + int fragment_above; + int max_fragments; + int fragment_size; + int ack_buffer_size; + int sent_packets_buffer_size; + int received_packets_buffer_size; + int fragment_reassembly_buffer_size; + float rtt_smoothing_factor; + float packet_loss_smoothing_factor; + float bandwidth_smoothing_factor; + int packet_header_size; + void (*transmit_packet_function)(void*,int,uint16_t,uint8_t*,int); + int (*process_packet_function)(void*,int,uint16_t,uint8_t*,int); + void * allocator_context; + void * (*allocate_function)(void*,uint64_t); + void (*free_function)(void*,void*); +}; + +void reliable_default_config( struct reliable_config_t * config ); + +struct reliable_endpoint_t * reliable_endpoint_create( struct reliable_config_t * config, double time ); + +uint16_t reliable_endpoint_next_packet_sequence( struct reliable_endpoint_t * endpoint ); + +void reliable_endpoint_send_packet( struct reliable_endpoint_t * endpoint, uint8_t * packet_data, int packet_bytes ); + +void reliable_endpoint_receive_packet( struct reliable_endpoint_t * endpoint, uint8_t * packet_data, int packet_bytes ); + +void reliable_endpoint_free_packet( struct reliable_endpoint_t * endpoint, void * packet ); + +uint16_t * reliable_endpoint_get_acks( struct reliable_endpoint_t * endpoint, int * num_acks ); + +void reliable_endpoint_clear_acks( struct reliable_endpoint_t * endpoint ); + +void reliable_endpoint_reset( struct reliable_endpoint_t * endpoint ); + +void reliable_endpoint_update( struct reliable_endpoint_t * endpoint, double time ); + +float reliable_endpoint_rtt( struct reliable_endpoint_t * endpoint ); + +float reliable_endpoint_packet_loss( struct reliable_endpoint_t * endpoint ); + +void reliable_endpoint_bandwidth( struct reliable_endpoint_t * endpoint, float * sent_bandwidth_kbps, float * received_bandwidth_kbps, float * acked_bandwidth_kpbs ); + +RELIABLE_CONST uint64_t * reliable_endpoint_counters( struct reliable_endpoint_t * endpoint ); + +void reliable_endpoint_destroy( struct reliable_endpoint_t * endpoint ); + +void reliable_log_level( int level ); + +void reliable_set_printf_function( int (*function)( RELIABLE_CONST char *, ... ) ); + +extern void (*netcode_assert_function)( RELIABLE_CONST char *, RELIABLE_CONST char *, RELIABLE_CONST char * file, int line ); + +#ifndef NDEBUG +#define reliable_assert( condition ) \ +do \ +{ \ + if ( !(condition) ) \ + { \ + reliable_assert_function( #condition, __FUNCTION__, __FILE__, __LINE__ ); \ + exit(1); \ + } \ +} while(0) +#else +#define reliable_assert( ignore ) ((void)0) +#endif + +void reliable_set_assert_function( void (*function)( RELIABLE_CONST char * /*condition*/, + RELIABLE_CONST char * /*function*/, + RELIABLE_CONST char * /*file*/, + int /*line*/ ) ); + +#ifdef __cplusplus +} +#endif + +#endif // #ifndef RELIABLE_H