diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,33 @@ +Copyright (C) 2012 Parallel Scientific Labs. +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 names of the copyright holders nor the names of its + contributors author may be used to endorse or promote products derived + from this software without specific prior written permission. + +The copyright holders provide no reassurances that the source code +provided does not infringe any patent, copyright, or any other +intellectual property rights of third parties. The copyright holders +disclaim any liability to any recipient for claims brought against +recipient by any third party for infringement of that parties +intellectual property rights. + +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 +OWNER 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. diff --git a/Network/CCI.hsc b/Network/CCI.hsc new file mode 100644 --- /dev/null +++ b/Network/CCI.hsc @@ -0,0 +1,1574 @@ +{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE EmptyDataDecls #-} +{-# LANGUAGE GeneralizedNewtypeDeriving #-} +{-# LANGUAGE TypeSynonymInstances #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ForeignFunctionInterface #-} +{-# LANGUAGE GADTs #-} + +-- | +-- Copyright : (C) 2012-2013 Parallel Scientific Labs, LLC. +-- License : BSD3 +-- +-- Haskell bindings for CCI. +-- +-- See . +-- Most of the comments in this module has been taken and reedited from there. +-- + +#let alignment t = "%lu", (unsigned long)offsetof(struct {char x__; t (y__);}, y__) + +module Network.CCI + ( initCCI + , finalizeCCI + , withCCI + -- * Devices + , Device + , getDeviceInfo + , DeviceInfo(..) + , getDevices + -- * Endpoints + , Endpoint + , createBlockingEndpoint + , createPollingEndpoint + , destroyEndpoint + , withBlockingEndpoint + , withPollingEndpoint + -- * Connections + , Connection + , connMaxSendSize + , connEndpoint + , connAttributes + , connContext + , accept + , reject + , connect + , disconnect + , ConnectionAttributes(..) + , WordPtr + -- * Data transfers + -- $dt + + -- ** Active Messages + -- $amsg + , send + , sendBlocking + , sendSilent + , sendNoCopy + , sendNoCopySilent + , sendv + , sendvBlocking + , sendvSilent + , sendvNoCopy + , sendvNoCopySilent + -- ** RMA + -- $rma + , rmaRegister + , rmaDeregister + , withRMALocalHandle + , RMA_MODE(..) + , rmaRead + , rmaWrite + , RMA_FLAG(..) + , RMALocalHandle + , RMARemoteHandle + , rmaHandle2ByteString + , createRMARemoteHandle + -- * Event handling + , getEvent + , returnEvent + , tryWithEventData + , pollWithEventData + , withEventData + , Event + , getEventData + , EventBytes + , packEventBytes + , unsafePackEventBytes + , EventData(..) + -- * Endpoint options + , getEndpt_SendTimeout + , setEndpt_SendTimeout + , getEndpt_SendBufCount + , setEndpt_SendBufCount + , getEndpt_RecvBufCount + , setEndpt_RecvBufCount + , getEndpt_KeepAliveTimeout + , setEndpt_KeepAliveTimeout + , getEndpt_RMAAlign + , getEndpt_URI + , RMAAlignments(..) + -- * Connection options + , getConn_SendTimeout + , setConn_SendTimeout + -- * Error handling + , strError + , CCIException(..) + , Status(..) + ) where + +import Control.Concurrent ( threadWaitRead, yield ) +import Control.Exception ( Exception, throwIO, bracket, bracket_, onException, mask_ ) +import Control.Monad ( liftM2, liftM3, join ) +import Data.Bits ( (.|.) ) +import Data.ByteString as B ( ByteString, packCStringLen, length ) +import Data.ByteString.Unsafe ( unsafePackCStringLen, unsafeUseAsCStringLen, unsafeUseAsCString ) +import Data.Dynamic ( Typeable ) +import Data.Word ( Word32, Word64 ) +import Foreign.C.Types ( CInt(..), CChar ) +import Foreign.C.String ( CString, peekCString, CStringLen, withCString ) +import Foreign.Ptr ( Ptr, nullPtr, WordPtr, wordPtrToPtr, plusPtr, ptrToWordPtr, castPtr ) +import Foreign.Marshal.Alloc ( alloca, allocaBytesAligned ) +import Foreign.Storable ( Storable(..) ) +import System.Posix.Types ( Fd(Fd) ) +import System.IO.Unsafe ( unsafePerformIO ) + + +#include + + +-- | Initializes CCI. If called more than once, only the +-- first call has any effects. +-- +-- This function will attempt to load the configuration file pointed by +-- the CCI_CONFIG environment variable. +-- A summary of the format of this configuration file follows. +-- +-- > # Comments are anything after the # symbols. +-- > # Sections in this file are denoted by [section name]. Each section +-- > # denotes a single CCI device. +-- > +-- > [bob0] +-- > # The only mandated field in each section is "driver". It indicates +-- > # which CCI driver should be applied to this device. +-- > driver = psm +-- > +-- > # The priority field determines the ordering of devices returned by +-- > # getDevices. 100 is the highest priority; 0 is the lowest priority. +-- > # If not specified, the priority value is 50. +-- > priority = 10 +-- > +-- > # The last field understood by the CCI core is the "default" field. +-- > # Only one device is allowed to have a "true" value for default. All +-- > # others must be set to 0 (or unset, which is assumed to be 0). If +-- > # one device is marked as the default, then this device will be used +-- > # when NULL is passed as the device when creating an endpoint. If no +-- > # device is marked as the default, it is undefined as to which device +-- > # will be used when NULL is passed as the device when creating an +-- > # endpoint. +-- > default = 1 +-- > +-- > # All other fields are uninterpreted by the CCI core; they're just +-- > # passed to the driver. The driver can do whatever it wants with +-- > # these values (e.g., system admins can set values to configure the +-- > # driver). Driver documentation should specify what parameters are +-- > # available, what each parameter is/does, and what its legal values +-- > # are. +-- > +-- > # This example shows a bonded PSM device that uses both the ipath0 and +-- > # ipath1 devices. Some other parameters are also passed to the PSM +-- > # driver; it assumedly knows how to handle them. +-- > +-- > device = ipath0,ipath1 +-- > capabilities = bonded,failover,age_of_captain:52 +-- > qos_stuff = fast +-- > +-- > # bob2 is another PSM device, but it only uses the ipath0 device. +-- > [bob2] +-- > driver = psm +-- > device = ipath0 +-- > +-- > # bob3 is another PSM device, but it only uses the ipath1 device. +-- > [bob3] +-- > driver = psm +-- > device = ipath1 +-- > sl = 3 # IB service level (if applicable) +-- > +-- > # storage is a device that uses the sock driver (udp sockets). Note +-- > # that this driver allows specifying which device to use by specifying +-- > # its IP address and MAC address -- assumedly it's an error if there is +-- > # no single device that matches both the specified IP address and MAC +-- > # (vs. specifying a specific device name). +-- > [storage] +-- > driver = sock +-- > priority = 5 +-- > ip = 172.31.194.1 +-- > mac = 01:12:23:34:45 +-- +-- On a debian PC this is the configuration which worked for us: +-- +-- > [eth0] +-- > driver = sock +-- > # replace with your device ip address +-- > ip = 192.168.0.1 +-- > # replace with your device mac address +-- > mac = 00:e0:7d:ad:95:5e +-- > default = 1 +-- +-- May throw: +-- +-- * 'ENOMEM' Not enough memory to complete. +-- +-- * 'ERR_NOT_FOUND' No driver or CCI_CONFIG. +-- +-- * 'ERROR' Unable to parse CCI_CONFIG. +-- +-- * errno errors if fopen() fails. +-- +-- * driver-specific errors. +initCCI :: IO () +initCCI = alloca$ \p -> poke p 0 >> cci_init #{const CCI_ABI_VERSION} 0 p >>= cci_check_exception + +foreign import ccall unsafe cci_init :: Word32 -> Word32 -> Ptr Word32 -> IO CInt + + +-- | This is the last CCI function that must be called; no other +-- CCI functions can be invoked after this function. +-- +-- Returns 'SUCCESS' iff CCI has been properly finalized. +-- +-- If initCCI was invoked multiple times, finalizeCCI should be +-- called as many times, and only the last one will not be a no-op. +-- +finalizeCCI :: IO () +finalizeCCI = cci_finalize >>= cci_check_exception + +foreign import ccall unsafe cci_finalize :: IO CInt + + +-- | Wraps an IO action between calls to 'initCCI' and 'finalizeCCI'. +withCCI :: IO a -> IO a +withCCI = bracket_ initCCI finalizeCCI + + +------------------------------------------ +-- Devices +------------------------------------------ + +-- | A device represents a network interface card (NIC) or host channel adapter +-- (HCA) that connects the host and network. A device may provide multiple +-- endpoints (typically one per core). +-- +data Device = Device + { devPtr :: Ptr Device + , getDeviceInfo :: DeviceInfo -- ^ Device information + } + +-- | Device information. +data DeviceInfo = DeviceInfo + { devName :: String -- ^ Name of the device from the config file, e.g. @"bob0"@ + + , devTransport :: String -- ^ Name of the device driver, e.g., @"sock"@ or @"verbs"@ + + , devUp :: Bool -- ^ Is this device actually up and running? + + , devInfo :: String -- ^ Human readable description string (to include newlines); should + -- contain debugging info, probably the network address of the + -- device at a bare minimum. + + , devConfArgs :: [String] -- ^ @"key=value"@ strings from the config file for this device. + + , devMaxSendSize :: Word32 -- ^ Maximum send size supported by the device + , devRate :: Word64 -- ^ Data rate per specification: data bits per second (not the + -- signaling rate). @0@ if unknown. + + , devPCI :: (Word32,Word32,Word32,Word32) -- ^ The PCI ID of this device as reported by the OS/hardware. All + -- values will be @0xFFFF@ for non-PCI devices (e.g. shared memory) + -- (domain, bus, dev, func) + } + deriving Show + + +-- | Returns a list of \"up\" devices. +-- +-- The order of devices returned corresponds to the priority fields in +-- the devices. If two devices share the same priority, their +-- ordering in the return array is arbitrary. +-- +getDevices :: IO [Device] +getDevices = do + pdev <- alloca$ \ppdev -> cci_get_devices ppdev >>= cci_check_exception >> peek ppdev + devices <- peekNullTerminated pdev 0 + mapM mkDevice devices + where + peekNullTerminated :: Ptr (Ptr a) -> Int -> IO [Ptr a] + peekNullTerminated p i = do + d <- peekElemOff p i + if d == nullPtr then return [] + else fmap (d:)$ peekNullTerminated p (i+1) + + mkDevice :: Ptr Device -> IO Device + mkDevice p = do + name <- (#{peek cci_device_t, name} p) >>= peekCString + transport <- #{peek cci_device_t, transport} p >>= peekCString + up <- #{peek cci_device_t, up} p + info <- #{peek cci_device_t, info} p >>= \x -> if x == nullPtr then return "" else peekCString x + conf_args <- (#{peek cci_device_t, conf_argv} p) >>= flip peekNullTerminated 0 >>= mapM peekCString + max_send_size <- #{peek cci_device_t, max_send_size} p + rate <- #{peek cci_device_t, rate} p + pci_domain <- #{peek cci_device_t, pci.domain} p + pci_bus <- #{peek cci_device_t, pci.bus} p + pci_dev <- #{peek cci_device_t, pci.dev} p + pci_func <- #{peek cci_device_t, pci.func} p + return$ Device p DeviceInfo + { devName = name + , devTransport = transport + , devUp = up + , devInfo = info + , devConfArgs = conf_args + , devMaxSendSize = max_send_size + , devRate = rate + , devPCI = (pci_domain, pci_bus, pci_dev, pci_func) + } + + +foreign import ccall unsafe cci_get_devices :: Ptr (Ptr (Ptr Device)) -> IO CInt + + + +------------------------------------------ +-- +-- Endpoints +-- +------------------------------------------ + + +-- | In CCI, an endpoint is the process virtual instance of a device. The endpoint +-- is the container of all the communication resources needed by the process +-- including queues and buffers including shared send and receive buffers. A +-- single endpoint may communicate with any number of peers and provides a single +-- completion queue regardless of the number of peers. CCI achieves better +-- scalability on very large HPC and data center deployments since the endpoint +-- manages buffers independent of how many peers it is communicating with. +-- +-- CCI uses an event model in which an application may either poll or wait for the +-- next event. Events include communication (e.g. send completions) as well as +-- connection handling (e.g. incoming client connection requests). The application +-- should return the events to CCI when it no longer needs them. CCI +-- achieves better scalability in time versus Sockets since all events are managed +-- by a single completion queue. +-- +-- Because an endpoint serves many connections, some events carry a so-called +-- context value which is user-defined and helps identifying with which connection +-- and transfer operation the event is related. +-- +-- In this realization of the API, all events related to an endpoint use the same type +-- for context values, which is expressed as the @ctx@ type parameter of 'Endpoint's, +-- 'Connection's and 'Event's. +-- +newtype Endpoint = Endpoint (Ptr EndpointV) + deriving (Storable, Show) + +data EndpointV + + +-- | This function creates a CCI endpoint. +-- An endpoint is associated with a device that performs the +-- actual communication (see 'getDevices'). +-- +-- If it is desirable to bind the CCI endpoint to a specific set of +-- resources (e.g., a NUMA node), you should bind the calling thread +-- before calling 'createBlockingEndpoint'. This is probably done by starting +-- the thread with 'Control.Concurrent.forkOn'. +-- +-- The OS handle returned by 'createBlockingEndpoint' can be manipulated with +-- 'Control.Concurrent.threadWaitRead' and 'Control.Concurrent.threadWaitWrite'. +-- +-- May throw: +-- +-- * 'ENODEV' if the device is not \"up\" +-- +-- * 'ENOMEM' if not enough memory is available +-- +-- * driver-specific errors +-- +createBlockingEndpoint :: Maybe Device -- ^ The device to use or Nothing to use the system-default device. + -> IO (Endpoint,Fd) -- ^ The endpoint and an operating system handle that can be used + -- to block for progress on this endpoint. +createBlockingEndpoint mdev = alloca$ \ppend -> + alloca$ \pfd -> do + cci_create_endpoint (maybe nullPtr devPtr mdev) 0 ppend pfd >>= cci_check_exception + liftM2 ((,)) (peek ppend) (fmap Fd$ peek pfd) + + +-- | Like 'createBlockingEndpoint' but avoids creation of an OS handle. Use this function +-- only if you intend to poll for events without blocking. +createPollingEndpoint :: Maybe Device -- ^ The device to use or Nothing to use the system-default device + -> IO Endpoint -- ^ The endpoint +createPollingEndpoint mdev = alloca$ \ppend -> do + cci_create_endpoint (maybe nullPtr devPtr mdev) 0 ppend nullPtr >>= cci_check_exception + peek ppend + + +foreign import ccall unsafe cci_create_endpoint :: Ptr Device -> CInt -> Ptr Endpoint -> Ptr CInt -> IO CInt + + +-- | Frees resources associated with the endpoint. All open connections +-- are closed immediately. It is exactly as if 'disconnect' was +-- invoked on every open connection on this endpoint. +-- +-- May throw driver-specific errors. +destroyEndpoint :: Endpoint -> IO () +destroyEndpoint (Endpoint pend) = cci_destroy_endpoint pend >>= cci_check_exception + +foreign import ccall unsafe cci_destroy_endpoint :: Ptr EndpointV -> IO CInt + + +-- | Wraps an IO action with calls to 'createBlockingEndpoint' and 'destroyEndpoint'. +-- +-- Makes sure that 'destroyEndpoint' is called in the presence of errors. +-- +withBlockingEndpoint :: Maybe Device -> ((Endpoint,Fd) -> IO a) -> IO a +withBlockingEndpoint mdev = bracket (createBlockingEndpoint mdev) (destroyEndpoint . fst) + +-- | Wraps an IO action with calls to 'createPollingEndpoint' and 'destroyEndpoint'. +-- +-- Makes sure that 'destroyEndpoint' is called in the presence of errors. +-- +withPollingEndpoint :: Maybe Device -> (Endpoint -> IO a) -> IO a +withPollingEndpoint mdev = bracket (createPollingEndpoint mdev) destroyEndpoint + + +------------------------------------------ +-- +-- Connections +-- +------------------------------------------ + + +-- | CCI uses connections to allow an application to choose the level +-- of service that best fits its needs and to provide fault isolation +-- should a peer stop responding. CCI does not, however, allocate +-- buffers per peer which reduces scalability. +-- +-- CCI offers choices of reliability and ordering. Some applications +-- such as distributed filesystems need to know that data has arrived +-- at the peer. A health monitoring application, on the other hand, +-- may want to send data that has a very short lifetime and does not +-- want to expend any efforts on retransmission. CCI can accommodate both. +-- +-- CCI provides Unreliable-Unordered (UU), Reliable-Unordered (RU), +-- and Reliable-Ordered (RO) connections as well as UU with send +-- multicast and UU with receive multicast. Most RPC style applications +-- do not require ordering and can take advantage of RU connections. +-- When ordering is not required, CCI can take better advantage of +-- multiple network paths, etc. +-- +-- This datatype represents connections for endpoints with contexts +-- of type @ctx@. +-- +newtype Connection = Connection (Ptr ConnectionV) + deriving (Storable, Show, Ord, Eq) + +data ConnectionV + + +-- | Maximum size of the messages the connection can send. +connMaxSendSize :: Connection -> Word32 +connMaxSendSize (Connection pconn) = unsafePerformIO$ #{peek cci_connection_t, max_send_size} pconn + +-- | Local endpoint on which the connection was created +connEndpoint :: Connection -> Endpoint +connEndpoint (Connection pconn) = Endpoint$ unsafePerformIO$ #{peek cci_connection_t, endpoint} pconn + +-- | Attributes of the connection +connAttributes :: Connection -> ConnectionAttributes +connAttributes (Connection pconn) = toEnum$ unsafePerformIO$ #{peek cci_connection_t, attribute} pconn + +-- | Context provided when the connection was created +connContext :: Connection -> WordPtr +connContext (Connection pconn) = unsafePerformIO$ #{peek cci_connection_t, context} pconn + + +-- | Accepts a connection request and establish a connection with a +-- specific endpoint. +-- +-- May throw: +-- +-- * 'EINVAL' if the Event was not a EvConnectRequest +-- +-- * driver-specific errors +-- +accept :: Event s -- ^ A connection request event previously returned by 'getEvent' + -> WordPtr -- ^ A context value which is passed in an 'EvAccept' event. + -> IO () +accept (Event penv) pctx = + cci_accept penv (wordPtrToPtr pctx) >>= cci_check_exception + +foreign import ccall unsafe cci_accept :: Ptr EventV -> Ptr () -> IO CInt + + + +-- | Rejects a connection request. +-- +-- May throw: +-- +-- * 'EINVAL' if the Event was not a 'EvConnectRequest' +-- +-- * driver-specific errors +-- +reject :: Event s -> IO () +reject (Event penv) = cci_reject penv >>= cci_check_exception + +foreign import ccall unsafe cci_reject :: Ptr EventV -> IO CInt + + +-- | Initiate a connection request (client side). +-- +-- Request a connection from a specific endpoint. The server endpoint's address +-- is described by a Uniform Resource Identifier. The use of an URI allows for +-- flexible description (IP address, hostname, etc). +-- +-- The connection request can carry limited amount of data to be passed to the +-- server for application-specific usage (identification, authentication, etc). +-- Implementations must support lengths <= 1,024 bytes. +-- +-- The connect call is always non-blocking, reliable and requires a decision +-- by the server (accept or reject), even for an unreliable connection, except +-- for multicast. +-- +-- Multicast connections don't necessarily involve a discrete connection +-- server, they may be handled by IGMP or other distributed framework. +-- +-- endpoint Local endpoint to use for requested connection. +-- +-- May throw device-specific errors. +-- +connect :: Endpoint -- ^ Local endpoint to use for requested connection. + -> String -- ^ Uniform Resource Identifier of the server. It is + -- accessible when the server's endpoint is created. + -- + -- Examples: + -- + -- * IP address: \"ip://172.31.194.2\" + -- + -- * Resolvable name: \"ip://foo.bar.com\" + -- + -- * IB LID or GID: \"ib://TBD\" + -- + -- * Blah id: \"blah://crap0123\" + -- + -- * With arguments: \"ip://foo.bar.com:eth1,eth3\" + + -> ByteString -- ^ Connection data to be send in the connection request + -- (for authentication, etc). + -> ConnectionAttributes -- ^ Attributes of the requested connection (reliability, + -- ordering, multicast, etc). + -> WordPtr -- ^ Pointer-sized integer used as context to identify the connection later. + -> Maybe Word64 -- ^ Timeout to wait for a connection in microseconds. Nothing means \"forever\". + -> IO () +connect (Endpoint pend) uri bdata ca pctx mtimeout = withCString uri$ \curi -> + unsafeUseAsCStringLen bdata$ \(cdata,clen) -> do + let fconnect ptv = cci_connect pend curi cdata (fromIntegral clen) + (fromIntegral$ fromEnum ca) (wordPtrToPtr pctx) 0 ptv + >>= cci_check_exception + case mtimeout of + Nothing -> fconnect nullPtr + Just ts -> + let (sec,usec) = divMod ts (10^(6::Int)) + in allocaBytesAligned #{size struct timeval} #{alignment struct timeval}$ \ptv -> do + #{poke struct timeval, tv_sec} ptv sec + #{poke struct timeval, tv_usec} ptv usec + fconnect ptv + +foreign import ccall unsafe cci_connect :: Ptr EndpointV -> CString -> Ptr CChar -> Word32 -> CInt -> Ptr () -> CInt -> Ptr () -> IO CInt + + +-- | Tears down an existing connection. +-- +-- Operation is local, remote side is not notified. From that point, +-- both local and remote side will get a 'ERR_DISCONNECTED' communication error +-- if sends are initiated on this connection. +-- +-- May throw driver-specific errors. +disconnect :: Connection -> IO () +disconnect (Connection pconn) = cci_disconnect pconn >>= cci_check_exception + +foreign import ccall unsafe cci_disconnect :: Ptr ConnectionV -> IO CInt + + +-- | Connection characteristics. +data ConnectionAttributes = + -- | Reliable ordered. Means that both completions and delivery are + -- in the same order that they were issued. + CONN_ATTR_RO + -- | Reliable unordered. Means that delivery is guaranteed, but both + -- delivery and completion may be in a different order than they were + -- issued. + | CONN_ATTR_RU + -- | Unreliable unordered (RMA forbidden). Delivery is not guaranteed, + -- and both delivery and completions may be in a different order + -- than they were issued. + | CONN_ATTR_UU + -- Multicast send (RMA forbidden). + | CONN_ATTR_UU_MC_TX + -- Multicast rcv (RMA forbidden). + | CONN_ATTR_UU_MC_RX + + deriving Show + + +instance Enum ConnectionAttributes where + + fromEnum CONN_ATTR_RO = #const CCI_CONN_ATTR_RO + fromEnum CONN_ATTR_RU = #const CCI_CONN_ATTR_RU + fromEnum CONN_ATTR_UU = #const CCI_CONN_ATTR_UU + fromEnum CONN_ATTR_UU_MC_TX = #const CCI_CONN_ATTR_UU_MC_TX + fromEnum CONN_ATTR_UU_MC_RX = #const CCI_CONN_ATTR_UU_MC_RX + + toEnum #{const CCI_CONN_ATTR_RO} = CONN_ATTR_RO + toEnum #{const CCI_CONN_ATTR_RU} = CONN_ATTR_RU + toEnum #{const CCI_CONN_ATTR_UU} = CONN_ATTR_UU + toEnum #{const CCI_CONN_ATTR_UU_MC_TX} = CONN_ATTR_UU_MC_TX + toEnum #{const CCI_CONN_ATTR_UU_MC_RX} = CONN_ATTR_UU_MC_RX + toEnum i = error$ "ConnectionAttributes toEnum: unknown value: "++show i + + +------------------------------------------ +-- +-- Data transfers +-- +------------------------------------------ + +-- $dt +-- CCI has two modes of communication: active messages (AM) and +-- remote memory access (RMA). + + +--------- +-- Active Messages +--------- + +-- $amsg +-- Loosely based on Berkeley's Active Messages, CCI's messages are small +-- (typically MTU sized) messages. Unlike Berkeley's AM, the message header +-- does not include a handler address. Instead, the receiver's CCI library will +-- generate a receive event that includes a pointer to the data within the CCI +-- library's buffers. The application may inspect the data, copy it out if needed +-- longer term, or even forward the data by passing it to the send function. +-- By using events instead of handlers, the application does not block further +-- communication progress. + + +-- | Send a short message. +-- +-- A short message limited to the size of 'connMaxSendSize'. +-- +-- If the application needs to send a message larger than +-- 'connMaxSendSize', the application is responsible for +-- segmenting and reassembly or it should use 'rmaWrite'. +-- +-- When send returns, the provided ByteString is disposable. By +-- default, CCI will buffer the data internally. +-- +-- The send will complete differently in reliable and unreliable +-- connections: +-- +-- * Reliable: only when remote side ACKs complete delivery -- but not +-- necessary consumption (i.e., remote completion). +-- +-- * Unreliable: when the buffer is re-usable (i.e., local completion). +-- +-- May throw 'ERR_DISCONNECTED', 'ERR_RNR', or driver-specific errors. +-- +send :: Connection -> ByteString -> WordPtr -> IO () +send conn msg pctx = unsafeUseAsCStringLen msg$ \cmsg -> send' conn cmsg pctx 0 + +-- | Like 'send' but will also block until the send completion has occurred. +-- +-- In this case, there is no event returned for this send via 'getEvent', the +-- send completion status is delivered in a 'CCIException' if sending fails. +-- A safe foreign call is made when using this flag. +-- +sendBlocking :: Connection -> ByteString -> WordPtr -> IO () +sendBlocking (Connection pconn) msg pctx = unsafeUseAsCStringLen msg$ \(cmsg,clen) -> + safe_cci_send pconn cmsg (fromIntegral clen) (wordPtrToPtr pctx) #{const CCI_FLAG_BLOCKING} + >>= cci_check_exception + +-- | Like 'send' but no send completion will be generated. +-- +-- For reliable ordered connections, since completions are issued in order, the +-- completion of any non-SILENT send directly implies the completion of any +-- previous SILENT sends. For unordered connections, completion ordering is +-- not guaranteed -- it is not safe to assume that application protocol +-- semantics imply specific unordered SILENT send completions. The +-- only ways to know when unordered SILENT sends have completed is +-- either to close the connection or issue a non-SILENT send. The +-- completion of a non-SILENT send guarantees the completion of all +-- previous SILENT sends. +-- +sendSilent :: Connection -> ByteString -> WordPtr -> IO () +sendSilent conn msg pctx = unsafeUseAsCStringLen msg$ \cmsg -> send' conn cmsg pctx #{const CCI_FLAG_SILENT} + +-- | Like 'send' but the message buffer remains in use until the send completion occurs. +-- +-- The CCI implementation is therefore free to use \"zero copy\" types +-- of transmission with the buffer if it wants to. +-- +sendNoCopy :: Connection -> CStringLen -> WordPtr -> IO () +sendNoCopy conn msg pctx = send' conn msg pctx #{const CCI_FLAG_NO_COPY} + +-- | Like 'sendNoCopy' and 'sendSilent'. +sendNoCopySilent :: Connection -> CStringLen -> WordPtr -> IO () +sendNoCopySilent conn msg pctx = send' conn msg pctx (#{const CCI_FLAG_NO_COPY} .|. #{const CCI_FLAG_SILENT}) + + +send' :: Connection -> CStringLen -> WordPtr -> CInt -> IO () +send' (Connection pconn) (cmsg,clen) pctx flags = + cci_send pconn cmsg (fromIntegral clen) (wordPtrToPtr pctx) flags + >>= cci_check_exception + +foreign import ccall unsafe cci_send :: Ptr ConnectionV -> Ptr CChar -> Word32 -> Ptr () -> CInt -> IO CInt +foreign import ccall safe "cci_send" safe_cci_send :: Ptr ConnectionV -> Ptr CChar -> Word32 -> Ptr () -> CInt -> IO CInt + + + +-- | Send a short vectored (gather) message. +-- +-- Like 'send', 'sendv' sends a short message bound by +-- 'connMaxSendSize'. Instead of a single data buffer, +-- 'sendv' allows the application to gather a list of +-- ByteStrings. +-- +-- May throw driver-specific errors. +-- +sendv :: Connection -> [ByteString] -> WordPtr -> IO () +sendv conn msgs pctx = unsafeUseAsCStringLens msgs [] 0$ \cmsgs clen -> + sendv' conn cmsgs clen pctx 0 + +-- | Like 'sendv' and 'sendBlocking'. +sendvBlocking :: Connection -> [ByteString] -> WordPtr -> IO () +sendvBlocking (Connection pconn) msgs pctx = unsafeUseAsCStringLens msgs [] 0$ \cmsgs clen -> + allocaBytesAligned (clen * #{size struct iovec}) #{alignment struct iovec}$ \piovecs -> do + sequence_ (zipWith (write_iovec piovecs) [clen-1,clen-2..0] cmsgs) + safe_cci_sendv pconn piovecs (fromIntegral clen) (wordPtrToPtr pctx) #{const CCI_FLAG_BLOCKING} + >>= cci_check_exception + +-- | Like 'sendv' and 'sendSilent'. +sendvSilent :: Connection -> [ByteString] -> WordPtr -> IO () +sendvSilent conn msgs pctx = unsafeUseAsCStringLens msgs [] 0$ \cmsgs clen -> + sendv' conn cmsgs clen pctx #{const CCI_FLAG_SILENT} + +-- | Like 'sendv' and 'sendNoCopySilent'. +sendvNoCopy :: Connection -> [CStringLen] -> WordPtr -> IO () +sendvNoCopy conn cmsgs pctx = sendv' conn cmsgs (Prelude.length cmsgs) pctx #{const CCI_FLAG_NO_COPY} + +-- | Like 'sendv' and 'sendNoCopySilent'. +sendvNoCopySilent :: Connection -> [CStringLen] -> WordPtr -> IO () +sendvNoCopySilent conn cmsgs pctx = + sendv' conn cmsgs (Prelude.length cmsgs) pctx (#{const CCI_FLAG_NO_COPY} .|. #{const CCI_FLAG_SILENT}) + +sendv' :: Connection -> [CStringLen] -> Int -> WordPtr -> CInt -> IO () +sendv' (Connection pconn) cmsgs clen pctx flags = + allocaBytesAligned (clen * #{size struct iovec}) #{alignment struct iovec}$ \piovecs -> do + sequence_ (zipWith (write_iovec piovecs) [clen-1,clen-2..0] cmsgs) + cci_sendv pconn piovecs (fromIntegral clen) (wordPtrToPtr pctx) flags + >>= cci_check_exception + +unsafeUseAsCStringLens :: [ByteString] -> [CStringLen] -> Int -> ([CStringLen] -> Int -> IO a) -> IO a +unsafeUseAsCStringLens (x:xs) acc len f + | seq len True = unsafeUseAsCStringLen x$ \cx -> unsafeUseAsCStringLens xs (cx:acc) (len+1) f +unsafeUseAsCStringLens _ acc len f = f acc len + +write_iovec :: Ptr a -> Int -> CStringLen -> IO () +write_iovec piovecs offs (cmsg,clen) = do + let p = plusPtr piovecs (offs * #{size struct iovec}) + #{poke struct iovec, iov_base} p cmsg + #{poke struct iovec, iov_len} p clen + + +foreign import ccall unsafe cci_sendv :: Ptr ConnectionV -> Ptr () -> Word32 -> Ptr () -> CInt -> IO CInt +foreign import ccall safe "cci_sendv" safe_cci_sendv :: Ptr ConnectionV -> Ptr () -> Word32 -> Ptr () -> CInt -> IO CInt + +enumFlags :: (Enum a, Num b) => [a] -> b +enumFlags = fromIntegral . foldl (\f-> (f.|.) . fromEnum) (0::Int) + + +--------- +-- RMA +--------- + +-- $rma +-- When an application needs to move bulk data, CCI provides RMA. To +-- use RMA, the application will explicitly register memory with CCI +-- and receive a handle. The application can pass the handle to a peer +-- using an active message and then perform a RMA Read or Write. The +-- RMA may also include a Fence. RMA requires a reliable connection +-- (ordered or unordered). An RMA may optionally include a remote +-- completion message that will be delivered to the peer after the RMA +-- completes. The completion message may be as large as a full active +-- message. + + + +-- | Registers memory for RMA operations on the connections of an endpoint. +-- +-- The intent is that this function is invoked frequently -- \"just +-- register everything\" before invoking RMA operations. +-- +-- In the best case, the implementation is cheap/fast enough that the +-- invocation time doesn't noticeably affect performance (e.g., MX and +-- PSM). If the implementation is slow (e.g., IB/iWARP), this function +-- should probably have a registration cache so that at least repeated +-- registrations are fast. +-- +-- It is allowable to have overlapping registerations. +-- +-- May throw driver-specific errors. +-- +rmaRegister :: Endpoint -> CStringLen -> RMA_MODE -> IO RMALocalHandle +rmaRegister (Endpoint pend) (cbuf,clen) m = alloca$ \p -> + cci_rma_register pend cbuf (fromIntegral clen) (fromIntegral (fromEnum m)) p + >>= cci_check_exception >> fmap RMALocalHandle (peek p) + +foreign import ccall unsafe cci_rma_register :: Ptr EndpointV + -> Ptr CChar -> Word64 -> CInt -> Ptr (Ptr CChar) -> IO CInt + +-- | Mode for registered handles. +data RMA_MODE = + RMA_READ -- ^ Handle can be read by other endpoints. + | RMA_WRITE -- ^ Handle can be written by other endpoints. + | RMA_READ_WRITE -- ^ Handle can be both read and written by other endpoints. + + +instance Enum RMA_MODE where + + fromEnum RMA_READ = #const CCI_FLAG_READ + fromEnum RMA_WRITE = #const CCI_FLAG_WRITE + fromEnum RMA_READ_WRITE = #{const CCI_FLAG_READ} .|. #{const CCI_FLAG_WRITE} + + toEnum #{const CCI_FLAG_READ} = RMA_READ + toEnum #{const CCI_FLAG_WRITE} = RMA_WRITE + toEnum i | i == #{const CCI_FLAG_READ} .|. #{const CCI_FLAG_WRITE} = RMA_READ_WRITE + toEnum i = error$ "RMA_MODE_FLAG toEnum: unknown value: "++show i + + + + +-- | Deregisters memory. +-- +-- If an RMA is in progress that uses this handle, the RMA may abort or +-- the deregisteration may fail. +-- +-- May throw driver-specific errors. +-- +rmaDeregister :: Endpoint -> RMALocalHandle -> IO () +rmaDeregister (Endpoint ep) (RMALocalHandle p) = cci_rma_deregister ep p >>= cci_check_exception + +foreign import ccall unsafe cci_rma_deregister :: Ptr EndpointV -> Ptr CChar -> IO CInt + + +-- | Wraps an IO operation with calls to 'rmaRegister' and 'rmaDeregister'. +-- +-- This function makes sure to call 'rmaDeregister' even in the presence of errors. +-- +withRMALocalHandle :: Endpoint -> CStringLen -> RMA_MODE -> (RMALocalHandle -> IO a) -> IO a +withRMALocalHandle e cs m = bracket (rmaRegister e cs m) (rmaDeregister e) + + +-- | Transfers data in remote memory to local memory. +-- +-- An RMA operation is a data transfer between local and remote buffers. +-- In order to obtain an RMA handle to local buffers, you should use +-- 'rmaRegister'. To obtain an RMA handle to a remote buffer, the handle +-- should be transmitted (see 'rmaHandle2ByteString' and +-- 'createRMARemoteHandle'). +-- +-- Adding the 'RMA_FENCE' flag ensures all previous operations are guaranteed to complete +-- remotely prior to this operation and all subsequent operations. Remote +-- completion does not imply a remote completion event, merely a successful +-- RMA operation. +-- +-- Optionally, sends a remote completion event to the target. If a 'ByteString' +-- message is provided, send a completion event to the target after +-- the RMA has completed. It is guaranteed to arrive after the RMA operation +-- has finished. +-- +-- CCI makes no guarantees about the data delivery within the RMA operation +-- (e.g., no last-byte-written-last). +-- +-- Only a local completion 'EvSend' will be generated. +-- +-- The following flags might be specified: +-- +-- * 'RMA_BLOCKING': Blocking call (see 'send' for details). +-- +-- * 'RMA_FENCE': All previous operations are guaranteed to +-- complete remotely prior to this operation +-- and all subsequent operations. +-- +-- * 'RMA_SILENT': Generates no local completion event (see 'send' +-- for details). +-- +-- May throw: +-- +-- * 'EINVAL' if the connection is unreliable or the data length is 0. +-- +-- * driver-specific errors. +-- +rmaRead :: Connection -- ^ Connection used for the RMA transfer. + -> Maybe ByteString -- ^ If @Just bs@, sends @bs@ as completion event to the peer. + -> RMALocalHandle -- ^ Handle to the transfer destination. + -> Word64 -- ^ Offset inside the destination buffer. + -> RMARemoteHandle -- ^ Handle to the transfer source. + -> Word64 -- ^ Offset inside the source buffer. + -> Word64 -- ^ Length of the data to transfer. + -> WordPtr -- ^ Context to deliver in the local 'EvSend' event. + -> [RMA_FLAG] -- ^ Flags specifying the transfer. + -> IO () +rmaRead (Connection pconn) mb (RMALocalHandle lhp) lo (RMARemoteHandle rh) ro dlen pctx flags = + unsafeUseAsCString rh$ \rhp -> + let crma (cb,clen) = cci_rma pconn cb (fromIntegral clen) lhp lo rhp ro dlen + (wordPtrToPtr pctx) (#{const CCI_FLAG_READ} .|. enumFlags flags) + >>= cci_check_exception + in case mb of + Just b -> unsafeUseAsCStringLen b$ crma + Nothing -> crma (nullPtr,0::Int) + +foreign import ccall unsafe cci_rma :: Ptr ConnectionV -> Ptr CChar -> Word32 + -> Ptr CChar -> Word64 -> Ptr CChar -> Word64 -> Word64 + -> Ptr () -> CInt -> IO CInt + + +-- | Transfers data in local memory to remote memory. +-- +-- Flags are set the same than for 'rmaRead'. +-- +rmaWrite :: Connection -- ^ Connection used for the RMA transfer. + -> Maybe ByteString -- ^ If @Just bs@, sends @bs@ as completion event to the peer. + -> RMARemoteHandle -- ^ Handle to the transfer destination. + -> Word64 -- ^ Offset inside the destination buffer. + -> RMALocalHandle -- ^ Handle to the transfer source. + -> Word64 -- ^ Offset inside the source buffer. + -> Word64 -- ^ Length of the data to transfer. + -> WordPtr -- ^ Context to deliver in the local 'EvSend' event. + -> [RMA_FLAG] -- ^ Flags specifying the transfer. + -> IO () +rmaWrite (Connection pconn) mb (RMARemoteHandle rh) ro (RMALocalHandle lhp) lo dlen pctx flags = + unsafeUseAsCString rh$ \rhp -> + let crma (cb,clen) = cci_rma pconn cb (fromIntegral clen) lhp lo rhp ro dlen + (wordPtrToPtr pctx) (#{const CCI_FLAG_WRITE} .|. enumFlags flags) + >>= cci_check_exception + in case mb of + Just b -> unsafeUseAsCStringLen b$ crma + Nothing -> crma (nullPtr,0::Int) + + +-- | Flags for 'rmaRead' and 'rmaWrite'. +data RMA_FLAG = + RMA_BLOCKING + | RMA_NO_COPY + | RMA_SILENT + | RMA_FENCE + + +instance Enum RMA_FLAG where + + fromEnum RMA_BLOCKING = #const CCI_FLAG_BLOCKING + fromEnum RMA_NO_COPY = #const CCI_FLAG_NO_COPY + fromEnum RMA_SILENT = #const CCI_FLAG_SILENT + fromEnum RMA_FENCE = #const CCI_FLAG_FENCE + + toEnum #{const CCI_FLAG_BLOCKING} = RMA_BLOCKING + toEnum #{const CCI_FLAG_NO_COPY} = RMA_NO_COPY + toEnum #{const CCI_FLAG_SILENT} = RMA_SILENT + toEnum #{const CCI_FLAG_FENCE} = RMA_FENCE + toEnum i = error$ "RMA_FLAG toEnum: unknown value: "++show i + + + +-- | RMA local handles have an associated buffer in local memory +-- which is read or written during RMA operations. +-- +newtype RMALocalHandle = RMALocalHandle (Ptr CChar) + deriving Eq + +-- | RMA remote handles have an associated buffer in a remote location +-- which is read or written during RMA operations. +-- +newtype RMARemoteHandle = RMARemoteHandle ByteString + + +-- | Gets a ByteString representation of the handle which can be sent to a peer. +rmaHandle2ByteString :: RMALocalHandle -> IO ByteString +rmaHandle2ByteString (RMALocalHandle p) = packCStringLen (p,32) + +-- | Creates a remote handle from a ByteString representation of a remote +-- handle. It is the inverse 'rmaHandle2ByteString'. +createRMARemoteHandle :: ByteString -> Maybe RMARemoteHandle +createRMARemoteHandle b = case B.length b of + 32 -> Just$ RMARemoteHandle b + _ -> Nothing + + +------------------------------------------ +-- +-- Event handling +-- +------------------------------------------ + + +-- | Get the next available CCI event. +-- +-- This function never blocks; it polls instantly to see if there is +-- any pending event of any type (send completion, receive, or other +-- events -- errors, incoming connection requests, etc.). If you want to +-- block, use the OS handle to use your OS's native blocking mechanism +-- (e.g., select/poll on the POSIX fd). This also allows the app to +-- busy poll for a while and then OS block if nothing interesting is +-- happening. The default OS handle returned when creating the +-- endpoint will return the equivalent of a POLL_IN when any event is +-- available. +-- +-- This function borrows the buffer associated with the event; it must +-- be explicitly returned later via returnEvent. +-- +-- May throw driver-specific errors. +-- +getEvent :: Endpoint -> IO (Maybe (Event s)) +getEvent (Endpoint pend) = alloca$ \pev -> do + st <- cci_get_event pend pev + case st of + #{const CCI_EAGAIN} -> return Nothing + _ -> cci_check_exception st >> fmap Just (peek pev) + + +foreign import ccall unsafe cci_get_event :: Ptr EndpointV -> Ptr (Event s) -> IO CInt + + +-- | This function gives back to the CCI implementation the +-- buffer associated with an event that was previously obtained +-- via 'getEvent'. The data buffer associated with the event will +-- immediately become stale to the application. +-- +-- Events may be returned in any order; they do not need to be returned +-- in the same order that 'getEvent' issued them. All events +-- must be returned, even send completions and \"other\" events, not +-- just receive events. However, it is possible (likely) that +-- returning send completion and \"other\" events will be no-ops. +-- +-- May throw driver-specific errors. +-- +returnEvent :: Event s -> IO () +returnEvent (Event pev) = cci_return_event pev >>= cci_check_exception + +foreign import ccall unsafe cci_return_event :: Ptr EventV -> IO CInt + + +-- | Wraps an IO action with calls to 'getEvent', 'getEventData' and 'returnEvent'. +-- +-- Makes sure the event is returned to the CCI implementation +-- even in the presence of errors. +-- +-- The callback IO action is performed with asynchronous exceptions blocked. Otherwise +-- a message taken out of the queue would be lost before it can be adequately processed. +-- If this does not fit your needs consider using one of 'Control.Exception.bracket' or +-- 'Control.Exception.mask' in combination with 'getEvent'. +-- +tryWithEventData :: + Endpoint -- ^ Endpoint on which to listen for events. + -> IO a -- ^ Action to perform in case no event is available. + -> (forall s. EventData s -> IO a) -- ^ Callback for the case when an event is available. + -> IO a -- ^ Yields the callback or the no-event action result. +tryWithEventData endp noEvent f = mask_$ do + mev <- getEvent endp + case mev of + Nothing -> noEvent + Just ev -> do + evd <- getEventData ev + r <- f evd `onException` returnEvent ev + returnEvent ev + return r + +-- | Like 'tryWithEventData' but blocks if no events are available. +withEventData :: + Endpoint -- ^ Endpoint on which to listen for events. + -> Fd -- ^ OS handle to block onto. + -> (forall s. EventData s -> IO a) + -> IO a -- ^ Yields the callback result. +withEventData endp fd f = go + where + go = tryWithEventData endp (threadWaitRead fd >> go) f + + +-- | Like 'tryWithEventData' but loops polling until events are available. +pollWithEventData :: + Endpoint -- ^ Endpoint on which to listen for events. + -> (forall s. EventData s -> IO a) + -> IO a -- ^ Yields the callback result. +pollWithEventData endp f = go + where + go = tryWithEventData endp (yield >> go) f + + +-- | Event representation. +-- +-- The @s@ parameter encodes the scope in which the event is valid. +-- It prevents the Event from escaping the 'withEventData'-like functions +-- which clearly define its lifetime. The protection is limited though. +-- Launching threads from these functions or using existential types +-- would sidestep the constraint. +-- +newtype Event s = Event (Ptr EventV) + deriving (Storable, Show) + +data EventV + + +-- | Retrieves the public data contained in an event. +getEventData :: Event s -> IO (EventData s) +getEventData ev@(Event pev) = do + t <- #{peek cci_event_t, type} pev + case t::CInt of + #{const CCI_EVENT_SEND} -> + liftM3 EvSend (#{peek cci_event_send_t, context} pev) + (fmap ctoEnum$ #{peek cci_event_send_t, status} pev) + (#{peek cci_event_send_t, connection} pev) + #{const CCI_EVENT_RECV} -> + liftM2 EvRecv (join$ liftM2 mkEventBytes + (#{peek cci_event_recv_t, ptr} pev) + (#{peek cci_event_recv_t, len} pev)) + (#{peek cci_event_recv_t, connection} pev) + #{const CCI_EVENT_CONNECT} -> do + st <- #{peek cci_event_connect_t, status} pev + liftM2 EvConnect (fmap ptrToWordPtr$ #{peek cci_event_connect_t, context} pev) + (toEither st$ #{peek cci_event_connect_t, connection} pev) + #{const CCI_EVENT_ACCEPT} -> do + st <- #{peek cci_event_accept_t, status} pev + liftM2 EvAccept (fmap ptrToWordPtr$ #{peek cci_event_accept_t, context} pev) + (toEither st$ #{peek cci_event_accept_t, connection} pev) + #{const CCI_EVENT_CONNECT_REQUEST} -> + liftM2 (EvConnectRequest ev) (join$ liftM2 mkEventBytes + (#{peek cci_event_connect_request_t, data_ptr} pev) + (#{peek cci_event_connect_request_t, data_len} pev)) + (fmap ctoEnum$ #{peek cci_event_connect_request_t, attribute} pev) + #{const CCI_EVENT_KEEPALIVE_TIMEDOUT} -> + fmap EvKeepAliveTimedOut (#{peek cci_event_keepalive_timedout_t, connection} pev) + #{const CCI_EVENT_ENDPOINT_DEVICE_FAILED} -> + fmap EvEndpointDeviceFailed (#{peek cci_event_endpoint_device_failed_t, endpoint} pev) + + _ -> error$ "getEventData: unexpected event type: "++show t + where + mkEventBytes p len = return$ EventBytes (p,fromIntegral (len::CInt)) + ctoEnum :: Enum a => CInt -> a + ctoEnum = toEnum . fromIntegral + toEither :: CInt -> IO Connection -> IO (Either Status Connection) + toEither #{const CCI_SUCCESS} c = fmap Right c + toEither st _ = return$ Left$ ctoEnum st + + + +-- | Bytes owned by an 'Event'. They will be released when the event is returned. +-- +-- The @s@ parameter encodes the scope in which the event is valid. +-- It prevents the EventBytes from escaping the 'withEventData'-like functions +-- which clearly define their lifetime. The protection is limited though. +-- Launching threads from these functions or using existential types +-- would sidestep the constraint. +-- +newtype EventBytes s = EventBytes CStringLen + deriving Show + + +-- | Copies the event bytes into a ByteString. +packEventBytes :: EventBytes s -> IO ByteString +packEventBytes (EventBytes bs) = packCStringLen bs + + +-- | Wraps the event bytes into a ByteString. The ByteString is valid for as long +-- as the given event bytes are valid. +unsafePackEventBytes :: EventBytes s -> IO ByteString +unsafePackEventBytes (EventBytes bs) = unsafePackCStringLen bs + + +-- | Representation of data contained in events. +-- +-- The @s@ parameter encodes the scope in which the event is valid. +-- It prevents the EventData from escaping the 'withEventData'-like functions +-- which clearly define their lifetime. The protection is limited though. +-- Launching threads from these functions or using existential types +-- would sidestep the constraint. +-- +data EventData s = + + -- | A 'send' or 'rmaRead' has completed. + -- + -- On a reliable connection, a sender will generally complete a send + -- when the receiver replies for that message. Additionally, an error + -- status may be returned ('ERR_DISCONNECTED', 'ERR_RNR'). + -- + -- On an unreliable connection, a sender will return 'SUCCESS' upon + -- local completion (i.e., the message has been queued up to some lower + -- layer -- there is no guarantee that it is \"on the wire\", etc.). + -- Other send statuses will only be returned for local errors. + -- + -- Contains the context provided to 'send', the result of the + -- send and the connection. + -- + EvSend WordPtr Status Connection + + -- | An active message has been received. + -- + -- A completion event is returned for each message received. + -- + -- Contains the transmitted data which is valid as long as the event + -- is not returned ('returnEvent'). + -- + | EvRecv (EventBytes s) Connection + + -- | A new outgoing connection was successfully accepted at the + -- peer; a connection is now available for data transfer. + -- If a connection request is rejected or fails otherwise, + -- an error code will be provided instead of a connection. + -- + -- Contains the context given to 'connect'. + -- + | EvConnect WordPtr (Either Status Connection) + + -- | A new incoming connection was successfully accepted + -- provided that the status field is 'SUCCESS'; + -- a connection is now available for data transfer. + -- If accepting the connection fails for whatever reason, + -- an error code will be provided instead of a connection. + -- + -- Contains the context given to 'accept'. + -- + | EvAccept WordPtr (Either Status Connection) + + -- | An incoming connection request from a client. + -- + -- Contains the data transmitted with the request, the + -- connection attributes and a reference to the event + -- for convenience to call 'accept' and 'reject'. + -- + | EvConnectRequest (Event s) (EventBytes s) ConnectionAttributes + + -- | The keepalive timeout has expired, i.e. no data from the + -- peer has been received during that time period + -- (see'OPT_ENDPT_KEEPALIVE_TIMEOUT' for more details). + -- + | EvKeepAliveTimedOut Connection + + -- | A device on this endpoint has failed. + -- + -- Contains the endpoint on the device that failed. + | EvEndpointDeviceFailed Endpoint + + deriving Show + + + + +------------------------------------------ +-- +-- Setting options +-- +------------------------------------------ + + +-- | The \"keepalive\" timeout is to prevent a client from connecting +-- to a server and then the client disappears without the server +-- noticing. If the server never sends anything on the connection, +-- it'll never realize that the client is gone, but the connection +-- is still consuming resources. But note that keepalive timers +-- apply to both clients and servers. +-- +-- The keepalive timeout is expressed in microseconds. If the +-- keepalive timeout value is set: +-- +-- * If no traffic at all is received on a connection within the +-- keepalive timeout, the EVENT_KEEPALIVE_TIMEOUT event is +-- raised on that connection. +-- +-- * The CCI implementation will automatically send control +-- hearbeats across an inactive (but still alive) connection to +-- reset the peer's keepalive timer before it times out. +-- +-- If a keepalive event is raised, the keepalive timeout is set to +-- 0 (i.e., it must be \"re-armed\" before it will timeout again), +-- but the connection is *not* disconnected. Recovery decisions +-- are up to the application; it may choose to 'disconnect' the +-- connection, re-arm the keepalive timeout, etc. +getEndpt_KeepAliveTimeout :: Endpoint -> IO Word32 +getEndpt_KeepAliveTimeout = getEndpointOpt' #{const CCI_OPT_ENDPT_KEEPALIVE_TIMEOUT} peek + +-- | See 'getEndpt_KeepAliveTimeout'. +setEndpt_KeepAliveTimeout :: Endpoint -> Word32 -> IO () +setEndpt_KeepAliveTimeout = setEndpointOpt' #{const CCI_OPT_ENDPT_KEEPALIVE_TIMEOUT} poke + + +-- | Default send timeout for all new connections. +getEndpt_SendTimeout :: Endpoint -> IO Word32 +getEndpt_SendTimeout = getEndpointOpt' #{const CCI_OPT_ENDPT_SEND_TIMEOUT} peek + +-- | See 'getEndpt_SendTimeout'. +setEndpt_SendTimeout :: Endpoint -> Word32 -> IO () +setEndpt_SendTimeout = setEndpointOpt' #{const CCI_OPT_ENDPT_SEND_TIMEOUT} poke + +-- | How many send buffers on the endpoint. It is the max number of +-- pending messages the CCI layer can buffer before failing or +-- blocking (depending on reliability mode). +getEndpt_SendBufCount :: Endpoint -> IO Word32 +getEndpt_SendBufCount = getEndpointOpt' #{const CCI_OPT_ENDPT_SEND_BUF_COUNT} peek + +-- | See 'getEndpt_SendBufCount'. +setEndpt_SendBufCount :: Endpoint -> Word32 -> IO () +setEndpt_SendBufCount = setEndpointOpt' #{const CCI_OPT_ENDPT_SEND_BUF_COUNT} poke + +-- | How many receiver buffers on the endpoint. It is the max +-- number of messages the CCI layer can receive without dropping. +getEndpt_RecvBufCount :: Endpoint -> IO Word32 +getEndpt_RecvBufCount = getEndpointOpt' #{const CCI_OPT_ENDPT_RECV_BUF_COUNT} peek + +-- | See 'getEndpt_RecvBufCount'. +setEndpt_RecvBufCount :: Endpoint -> Word32 -> IO () +setEndpt_RecvBufCount = setEndpointOpt' #{const CCI_OPT_ENDPT_RECV_BUF_COUNT} poke + +-- | RMA registration alignment requirements can be retrieved for an endpoint. +-- +-- The CTP will yield the minimal alignment needed for different operations. +-- A value of 0 indicates that there are no alignment requirements for that +-- operation. A value of 4, for example, indicates that that member must be 4-byte +-- aligned. +-- +-- If the CTP requires RMA alignment and the application passes in an +-- un-aligned parameter, the CTP may need to allocate a temporary +-- buffer, register it, and use it instead. This will also require a +-- copy of the data to the correct location. This will decrease +-- performance for these cases. +-- +getEndpt_RMAAlign :: Endpoint -> IO RMAAlignments +getEndpt_RMAAlign = getEndpointOpt' #{const CCI_OPT_ENDPT_RMA_ALIGN} peek + +-- | Retrieve the endpoint's URI used for listening for connection +-- requests. The application should never need to parse this URI. +getEndpt_URI :: Endpoint -> IO String +getEndpt_URI = getEndpointOpt' #{const CCI_OPT_ENDPT_URI} ((peekCString =<<) . peek) + + +setEndpointOpt' :: Storable a => CInt -> (Ptr a -> a -> IO ()) -> Endpoint -> a -> IO () +setEndpointOpt' opt pk (Endpoint pend) v = withValueO pk v$ \pv -> + cci_set_opt (castPtr pend) opt (castPtr pv) >>= cci_check_exception + +getEndpointOpt' :: Storable a => CInt -> (Ptr a -> IO b) -> Endpoint -> IO b +getEndpointOpt' opt pk (Endpoint pend) = + alloca$ \pv -> do + cci_get_opt (castPtr pend) opt (castPtr pv) >>= cci_check_exception + pk pv + + +foreign import ccall unsafe cci_set_opt :: Ptr () -> CInt -> Ptr () -> IO CInt +foreign import ccall unsafe cci_get_opt :: Ptr () -> CInt -> Ptr () -> IO CInt + + +data RMAAlignments = RMAAlignments + { rmaWriteLocalAddr :: Word32 + , rmaWriteRemoteAddr :: Word32 + , rmaWriteLength :: Word32 + , rmaReadLocalAddr :: Word32 + , rmaReadRemoteAddr :: Word32 + , rmaReadLength :: Word32 + } + deriving Show + +instance Storable RMAAlignments where + sizeOf _ = #{size cci_alignment_t} + alignment _ = #{alignment cci_alignment_t} + peek p = do + rwla <- #{peek cci_alignment_t, rma_write_local_addr} p + rwra <- #{peek cci_alignment_t, rma_write_remote_addr} p + rwl <- #{peek cci_alignment_t, rma_write_length} p + rrla <- #{peek cci_alignment_t, rma_read_local_addr} p + rrra <- #{peek cci_alignment_t, rma_read_remote_addr} p + rrl <- #{peek cci_alignment_t, rma_read_length} p + return RMAAlignments + { rmaWriteLocalAddr = rwla + , rmaWriteRemoteAddr = rwra + , rmaWriteLength = rwl + , rmaReadLocalAddr = rrla + , rmaReadRemoteAddr = rrra + , rmaReadLength = rrl + } + poke = error "poke RMAAlignments: unimplemented" + +withValueO :: Storable a => (Ptr a -> a -> IO ()) -> a -> (Ptr a -> IO b) -> IO b +withValueO pk v f = alloca$ \pv -> pk pv v >> f pv + + +-------------------------- +-- Connection options +-------------------------- + +-- | Reliable send timeout in microseconds. +getConn_SendTimeout :: Connection -> IO Word32 +getConn_SendTimeout = getConnectionOpt' #{const CCI_OPT_CONN_SEND_TIMEOUT} peek + +-- | See 'getConn_SendTimeout'. +setConn_SendTimeout :: Connection -> Word32 -> IO () +setConn_SendTimeout = setConnectionOpt' #{const CCI_OPT_CONN_SEND_TIMEOUT} poke + + +setConnectionOpt' :: Storable a => CInt -> (Ptr a -> a -> IO ()) -> Connection -> a -> IO () +setConnectionOpt' opt pk (Connection pconn) v = withValueO pk v$ \pv -> + cci_set_opt (castPtr pconn) opt (castPtr pv) >>= cci_check_exception + + +getConnectionOpt' :: Storable a => CInt -> (Ptr a -> IO b) -> Connection -> IO b +getConnectionOpt' opt pk (Connection pconn) = + alloca$ \pv -> do + cci_get_opt (castPtr pconn) opt (castPtr pv) >>= cci_check_exception + pk pv + + + +------------------------------------------ +-- +-- Error handling +-- +------------------------------------------ + + +-- | Returns a human readable description of a 'Status' value. +strError :: Maybe Endpoint -> Status -> IO String +strError me st = do + cs <- cci_strerror (maybe nullPtr (\(Endpoint ep) -> ep) me)$ fromIntegral$ fromEnum st + if nullPtr == cs then return "" + else peekCString cs + +foreign import ccall unsafe cci_strerror :: Ptr EndpointV -> CInt -> IO CString + + +-- | Type of exceptions that can arise when calling cci functions. +data CCIException = CCIException Status + deriving (Show,Typeable) + +instance Exception CCIException + + +-- Error codes resulting from CCI operations. +data Status = + SUCCESS -- ^ Returned from most functions when they succeed. + | ERROR -- ^ Generic error + + -- Send completion status codes + + -- | For both reliable and unreliable sends, this error code means + -- that 'disconnect' has been invoked on the send side (in + -- which case this is an application error), or the receiver + -- replied that the receiver invoked 'disconnect'. + | ERR_DISCONNECTED + + -- | For a reliable send, this error code means that a receiver + -- is reachable, the connection is connected but the receiver + -- could not receive the incoming message during the timeout + -- period. If a receiver cannot receive an incoming message for + -- transient reasons (most likely out of resources), it returns + -- an Receiver-Not-Ready NACK and drops the message. The sender + -- keeps retrying to send the message until the timeout expires, + -- + -- If the timeout expires and the last control message received + -- from the receiver was an RNR NACK, then this message is + -- completed with the RNR status. If the connection is both + -- reliable and ordered, then all successive sends are also + -- completed in the order in which they were issued with the RNR + -- status. + -- + -- This error code will not be returned for unreliable sends. + | ERR_RNR + | ERR_DEVICE_DEAD -- ^ The local device is gone, not coming back + + -- | Error returned from remote peer indicating that the address was + -- either invalid or unable to be used for access / permissions + -- reasons. + | ERR_RMA_HANDLE + + -- | Error returned from remote peer indicating that it does not support + -- the operation that was requested. + | ERR_RMA_OP + | ERR_NOT_IMPLEMENTED -- ^ Not yet implemented + | ERR_NOT_FOUND -- ^ Not found + + -- Errno.h error codes + + | EINVAL -- ^ Invalid parameter passed to CCI function call + + -- | For a reliable send, this error code means that the sender did + -- not get anything back from the receiver within a timeout (no + -- ACK, no NACK, etc.). It is unknown whether the receiver + -- actually received the message or not. + -- + -- This error code won't occur for unreliable sends. + -- + -- One other place where @ETIMEDOUT@ appears is when an outgoing + -- connection request expires (in an 'EvConnect' event). + | ETIMEDOUT + | ECONNREFUSED + | ENOMEM -- ^ No more memory + | ENODEV -- ^ No device available + | EBUSY -- ^ Resource busy (e.g. port in use) + | ERANGE -- ^ Value out of range (e.g. no port available) + | EAGAIN -- ^ Resource temporarily unavailable + | ENOBUFS -- ^ The output queue for a network interface is full + | EMSGSIZE -- ^ Message too long + | ENOMSG -- ^ No message of desired type + | EADDRNOTAVAIL -- ^ Address not available + | OTHER Int + deriving (Show,Read,Eq) + + +cci_check_exception :: CInt -> IO () +cci_check_exception #{const CCI_SUCCESS} = return () +cci_check_exception c = throwIO$ CCIException$ toEnum$ fromIntegral c + + +instance Enum Status where + + fromEnum c = + case c of + SUCCESS -> #{const CCI_SUCCESS} + ERROR -> #{const CCI_ERROR} + ERR_DISCONNECTED -> #{const CCI_ERR_DISCONNECTED} + ERR_RNR -> #{const CCI_ERR_RNR} + ERR_DEVICE_DEAD -> #{const CCI_ERR_DEVICE_DEAD} + ERR_RMA_HANDLE -> #{const CCI_ERR_RMA_HANDLE} + ERR_RMA_OP -> #{const CCI_ERR_RMA_OP} + ERR_NOT_IMPLEMENTED -> #{const CCI_ERR_NOT_IMPLEMENTED} + ERR_NOT_FOUND -> #{const CCI_ERR_NOT_FOUND} + EINVAL -> #{const CCI_EINVAL} + ETIMEDOUT -> #{const CCI_ETIMEDOUT} + ECONNREFUSED -> #{const CCI_ECONNREFUSED} + ENOMEM -> #{const CCI_ENOMEM} + ENODEV -> #{const CCI_ENODEV} + EBUSY -> #{const CCI_EBUSY} + ERANGE -> #{const CCI_ERANGE} + EAGAIN -> #{const CCI_EAGAIN} + ENOBUFS -> #{const CCI_ENOBUFS} + EMSGSIZE -> #{const CCI_EMSGSIZE} + ENOMSG -> #{const CCI_ENOMSG} + EADDRNOTAVAIL -> #{const CCI_EADDRNOTAVAIL} + OTHER co -> co + + toEnum c = + case c of + #{const CCI_SUCCESS} -> SUCCESS + #{const CCI_ERROR} -> ERROR + #{const CCI_ERR_DISCONNECTED} -> ERR_DISCONNECTED + #{const CCI_ERR_RNR} -> ERR_RNR + #{const CCI_ERR_DEVICE_DEAD} -> ERR_DEVICE_DEAD + #{const CCI_ERR_RMA_HANDLE} -> ERR_RMA_HANDLE + #{const CCI_ERR_RMA_OP} -> ERR_RMA_OP + #{const CCI_ERR_NOT_IMPLEMENTED} -> ERR_NOT_IMPLEMENTED + #{const CCI_ERR_NOT_FOUND} -> ERR_NOT_FOUND + #{const CCI_EINVAL} -> EINVAL + #{const CCI_ETIMEDOUT} -> ETIMEDOUT + #{const CCI_ECONNREFUSED} -> ECONNREFUSED + #{const CCI_ENOMEM} -> ENOMEM + #{const CCI_ENODEV} -> ENODEV + #{const CCI_EBUSY} -> EBUSY + #{const CCI_ERANGE} -> ERANGE + #{const CCI_EAGAIN} -> EAGAIN + #{const CCI_ENOBUFS} -> ENOBUFS + #{const CCI_EMSGSIZE} -> EMSGSIZE + #{const CCI_ENOMSG} -> ENOMSG + #{const CCI_EADDRNOTAVAIL} -> EADDRNOTAVAIL + _ -> OTHER c diff --git a/Setup.lhs b/Setup.lhs new file mode 100644 --- /dev/null +++ b/Setup.lhs @@ -0,0 +1,4 @@ +#!/usr/bin/env runhaskell + +> import Distribution.Simple +> main = defaultMain diff --git a/cci.cabal b/cci.cabal new file mode 100644 --- /dev/null +++ b/cci.cabal @@ -0,0 +1,86 @@ +Name: cci +Version: 0.3.1 +Synopsis: Bindings for the CCI networking library +Description: The CCI project is an open source communication interface that + aims to provide a simple and portable API, high performance, + scalability for the largest deployments, and robustness in the + presence of faults. It is developed and maintained by a + partnership of research, academic, and industry members. + . + See for details. +Copyright: (C) 2012-2013 Parallel Scientific Labs, LLC +Maintainer: Facundo Domínguez +License: BSD3 +License-file: LICENSE +Category: Network +Build-Type: Simple +Cabal-Version: >= 1.10 + +source-repository head + type: git + location: git://github.com/ps-labs/haskell-cci.git + +Library + default-language: Haskell2010 + exposed-modules: Network.CCI + build-depends: base >= 4.3 && < 5, + bytestring >= 0.9 + ghc-options: -Wall + extra-libraries: cci + +Flag examples + Description: Build examples + Default: False + +Executable cci-example-client + default-language: Haskell2010 + main-is: examples/Client.hs + ghc-options: -Wall -idist/build + if flag(examples) + build-depends: base, bytestring, cci + else + buildable: False + +Executable cci-example-pingpong + default-language: Haskell2010 + main-is: examples/pingpong.hs + ghc-options: -Wall -idist/build -threaded + if flag(examples) + build-depends: base, bytestring, cci, time + else + buildable: False + +Executable cci-example-server + default-language: Haskell2010 + main-is: examples/Server.hs + ghc-options: -Wall -idist/build + if flag(examples) + build-depends: base, bytestring, cci + else + buildable: False + +Executable cci-example-registration-benchmark + default-language: Haskell2010 + main-is: examples/registration-benchmark.hs + ghc-options: -Wall -idist/build + if flag(examples) + build-depends: base, bytestring, cci, time + else + buildable: False + +Test-Suite test-cci + default-language: Haskell2010 + type: exitcode-stdio-1.0 + main-is: test_cci.hs + hs-source-dirs: test + ghc-options: -Wall -fno-warn-name-shadowing -idist/build + build-depends: base, bytestring, cci, containers, binary, process, + filepath, mtl, random, cmdargs < 0.11, pretty + +Test-Suite test-worker + default-language: Haskell2010 + type: exitcode-stdio-1.0 + main-is: Worker.hs + ghc-options: -Wall -idist/build -threaded + hs-source-dirs: test + build-depends: base, bytestring, cci, time, containers, binary diff --git a/examples/Client.hs b/examples/Client.hs new file mode 100644 --- /dev/null +++ b/examples/Client.hs @@ -0,0 +1,45 @@ +-- +-- Copyright (C) 2012 Parallel Scientific. All rights reserved. +-- +-- See the accompanying LICENSE file for license information. +-- + +import Control.Exception ( finally ) +import Data.ByteString as B ( empty ) +import Data.ByteString.Char8 as B ( pack, putStrLn ) +import System.Environment ( getArgs ) + +import Network.CCI ( withCCI, withPollingEndpoint, connect, ConnectionAttributes(..) + , pollWithEventData, EventData(..), disconnect, send + , unsafePackEventBytes + ) + + +main :: IO () +main = do + [uri] <- getArgs + withCCI$ + withPollingEndpoint Nothing$ \ep -> do + connect ep uri empty CONN_ATTR_UU 0 Nothing + _ <- loopWhileM id$ pollWithEventData ep$ \ev -> + case ev of + + EvConnect ctx (Right conn) -> + do print ctx + send conn (pack "ping!") 1 + return True + + EvRecv ebs conn -> flip finally (disconnect conn)$ + do unsafePackEventBytes ebs >>= B.putStrLn + return False + + _ -> print ev >> return True + + return () + + +-- | @loopWhileM p io@ performs @io@ repeteadly while its result satisfies @p@. +-- Yields the first offending result. +loopWhileM :: (a -> Bool) -> IO a -> IO a +loopWhileM p io = io >>= \a -> if p a then loopWhileM p io else return a + diff --git a/examples/Server.hs b/examples/Server.hs new file mode 100644 --- /dev/null +++ b/examples/Server.hs @@ -0,0 +1,34 @@ +-- +-- Copyright (C) 2012 Parallel Scientific. All rights reserved. +-- +-- See the accompanying LICENSE file for license information. +-- + +import Control.Exception ( finally ) +import Control.Monad ( forever, void ) +import qualified Data.ByteString.Char8 as B ( putStrLn, pack ) +import Network.CCI ( withCCI, withPollingEndpoint, getEndpt_URI, accept + , pollWithEventData, EventData(..), disconnect, send + , unsafePackEventBytes + ) + + +main :: IO () +main = + withCCI$ + withPollingEndpoint Nothing$ \ep -> do + getEndpt_URI ep >>= putStrLn + _ <- forever$ pollWithEventData ep$ \evd -> + case evd of + + EvConnectRequest ev _bs _cattr -> void$ accept ev 0 + + EvRecv ebs conn -> flip finally (disconnect conn)$ + do unsafePackEventBytes ebs >>= B.putStrLn + send conn (B.pack "pong!") 1 + + _ -> print evd + + return () + + diff --git a/examples/pingpong.hs b/examples/pingpong.hs new file mode 100644 --- /dev/null +++ b/examples/pingpong.hs @@ -0,0 +1,205 @@ +-- +-- Copyright (C) 2012 Parallel Scientific. All rights reserved. +-- +-- See the accompanying LICENSE file for license information. +-- + + +import Control.Monad ( forever, void, forM_, replicateM_, when ) +import qualified Data.ByteString as B ( empty, take, length, head, singleton, null ) +import Data.ByteString.Char8 ( pack, unpack ) +import Data.ByteString.Unsafe ( unsafePackCStringLen ) +import Data.Maybe ( isNothing, fromJust ) +import Data.Time ( getCurrentTime, diffUTCTime ) +import Data.Word ( Word64 ) +import Foreign.Storable ( poke ) +import Foreign.Marshal.Alloc ( allocaBytesAligned ) +import Network.CCI ( withCCI, withPollingEndpoint, getEndpt_URI, accept, Endpoint + , pollWithEventData, EventData(..), send, Connection + , connect, ConnectionAttributes(..), connMaxSendSize + , createRMARemoteHandle, RMARemoteHandle, rmaWrite + , withRMALocalHandle, rmaHandle2ByteString, RMA_MODE(..) + , unsafePackEventBytes, packEventBytes, disconnect + ) +import Numeric ( fromRat ) +import Text.Printf ( printf ) +import System.Console.GetOpt ( OptDescr(..), ArgDescr(..), getOpt, ArgOrder(..), usageInfo ) +import System.Environment ( getArgs ) + +data Options = Options + { oServerURI :: Maybe String + , oIsServer :: Bool + , oWarmup :: Int + , oIters :: Int + , oRMA :: Maybe Word64 + } + +defaultOptions :: Options +defaultOptions = Options + { oServerURI = Nothing + , oIsServer = False + , oWarmup = 1024 + , oIters = 16*1024 + , oRMA = Nothing + } + +data ConnectionSpecs = AM | RMA Word64 + deriving (Show, Read) + +options :: [ OptDescr (Options -> Options) ] +options = + [ Option "h" [] (ReqArg (\uri o -> o { oServerURI = Just uri }) "URI") "server URI" + , Option "s" [] (NoArg (\o -> o { oIsServer = True })) "run as server" + , Option "r" [] (ReqArg (\sz o -> o { oRMA = Just$ read sz }) "BUFSIZE") "use rma with the given buffer size" + ] + +main :: IO () +main = do + args <- getArgs + let (fns,nonOptions,errors) = getOpt Permute options args + o = foldl (flip ($)) defaultOptions fns + if null nonOptions && null errors && isNothing (oServerURI o) == oIsServer o then + withCCI$ + withPollingEndpoint Nothing$ \ep -> do + getEndpt_URI ep >>= putStrLn + if oIsServer o then forever$ goServer ep o + else goClient ep o + else + putStrLn$ unlines$ errors ++ [ usageInfo header options ] + where + header = "Usage: pingpong [OPTION...]" + + +goServer :: Endpoint -> Options -> IO () +goServer ep _o = do + pollWithEventData ep$ \evd -> + case evd of + EvConnectRequest ev _bs _cattr -> void$ accept ev 0 + _ -> print evd + + pollWithEventData ep$ \evd -> + case evd of + EvAccept 0 (Right _conn) -> return () + _ -> print evd + + (conn,cs) <- pollWithEventData ep$ \evd -> + case evd of + EvRecv ebs conn -> packEventBytes ebs >>= return . (,) conn . read . unpack + _ -> error ("goServer: unexpected event"++show evd) + case cs of + AM -> goServer'$ connMaxSendSize conn + RMA sz -> do + allocaBytesAligned (fromIntegral sz) 4096 $ \cbuf -> + withRMALocalHandle ep (cbuf,fromIntegral sz) RMA_READ_WRITE$ \lh -> do + bh <- rmaHandle2ByteString lh + send conn bh 0 + goServer' sz + where + goServer' sz = allocaBytesAligned (fromIntegral sz) 4096 $ \cbuf -> + unsafePackCStringLen (cbuf,fromIntegral sz) >>= \sbuf -> do + Just conn <- loopWhileM isNothing$ pollWithEventData ep$ \evd -> + case evd of + EvRecv ebs conn -> do + bs <- unsafePackEventBytes ebs + if not (B.null bs) && B.head bs == 1 then + return$ Just conn + else + send conn (B.take (B.length bs) sbuf) 0 >> return Nothing + EvSend _ _ _ -> return Nothing + _ -> print evd >> return Nothing + + disconnect conn + + +goClient :: Endpoint -> Options -> IO () +goClient ep o = do + connect ep (fromJust$ oServerURI o) B.empty CONN_ATTR_RO 0 Nothing + conn <- pollWithEventData ep$ \evd -> + case evd of + EvConnect _ctx (Right conn) -> return conn + _ -> print evd >> return (error "goClient: unexpected event.") + mrmah <- exchangeConnectionSpecs ep o conn + let sz = if isNothing mrmah then fromIntegral$ connMaxSendSize conn + else fromIntegral$ fromJust$ oRMA o + allocaBytesAligned sz 4096 $ \cbuf -> do + when (sz>0)$ poke cbuf 0 + unsafePackCStringLen (cbuf,sz) >>= \sbuf -> + case mrmah of + Nothing -> goClient' conn sbuf Nothing + Just rh -> withRMALocalHandle ep (cbuf,sz) RMA_READ_WRITE (goClient' conn sbuf . Just . flip (,) rh) + where + goClient' conn sbuf mrmahs = do + putStrLn "Bytes\t\tLatency (one-way)\tThroughput" + let app = if isNothing mrmahs then (0:) else id + bszs = takeWhile (<=B.length sbuf)$ app$ iterate (*2) 1 + its = replicate (length$ takeWhile (<1024*(64::Int)) bszs) 0 ++ [0::Int ..] + forM_ (zip bszs its)$ \(current_size,i) -> do + + let warmup = max 2$ oWarmup o `div` 2^i + let iters = max 16$ oIters o `div` 2^i + + replicateM_ warmup$ testRoundTrip conn sbuf mrmahs current_size + + start <- getCurrentTime + replicateM_ iters$ testRoundTrip conn sbuf mrmahs current_size + end <- getCurrentTime + + let lat = toRational (diffUTCTime end start) * 1000000 / toRational iters + / if isNothing mrmahs then 2 else 1 + bw = fromIntegral current_size / lat; + printf "%8d\t%8.2f us\t\t%8.2f MB/s\n" current_size (fromRat lat :: Double) (fromRat bw :: Double) + + -- say goodbye + send conn (B.singleton 1) 0 + void$ loopWhileM id$ pollWithEventData ep$ \evd -> + case evd of + EvSend _ _ _ -> return False + _ -> error ("goClient': unexpected event: "++show evd) + disconnect conn + + + testRoundTrip conn _sbuf (Just (lh,rh)) msg_size = do + rmaWrite conn Nothing rh 0 lh 0 (fromIntegral msg_size) 0 [] + void$ loopWhileM id$ pollWithEventData ep$ \evd -> + case evd of + EvSend _ _ _ -> return False + _ -> print evd >> return True + + testRoundTrip conn sbuff Nothing msg_size = do + send conn (B.take (fromIntegral msg_size) sbuff) 0 + void$ loopWhileM id$ pollWithEventData ep$ \evd -> + case evd of + EvRecv _ _conn -> return False + EvSend _ _ _ -> return True + _ -> print evd >> return True + +exchangeConnectionSpecs :: Endpoint -> Options -> Connection -> IO (Maybe RMARemoteHandle) +exchangeConnectionSpecs ep o conn = do + let cs = connectionSpecs o + send conn (pack$ show cs) 0 + case cs of + RMA _ -> loopWhileM isNothing$ pollWithEventData ep$ \evd -> + case evd of + EvRecv ebs _conn -> packEventBytes ebs >>= return . createRMARemoteHandle + EvSend _ _ _ -> return Nothing + _ -> print evd >> return Nothing + _ -> return Nothing + +connectionSpecs :: Options -> ConnectionSpecs +connectionSpecs = maybe AM RMA . oRMA + + +-- foreign import ccall unsafe posix_memalign :: Ptr (Ptr ()) -> Word32 -> Word32 -> IO Int + + +-- | @loopWhileM p io@ performs @io@ repeteadly while its result satisfies @p@. +-- Yields the first offending result. +loopWhileM :: Monad m => (a -> Bool) -> m a -> m a +loopWhileM p m = go + where go = m >>= \a -> if p a then go else return a + +-- | @foreverSt s f@ performs @f@ repeteadly which modifies a state initialized with @s@. +-- Never finishes unless there is an exception. +-- foreverSt :: Monad m => a -> (a -> m a) -> m b +-- foreverSt a fm = fm a >>= \a' -> foreverSt a' fm + diff --git a/examples/registration-benchmark.hs b/examples/registration-benchmark.hs new file mode 100644 --- /dev/null +++ b/examples/registration-benchmark.hs @@ -0,0 +1,106 @@ +-- +-- Copyright (C) 2013 Parallel Scientific Labs, LLC. All rights reserved. +-- +-- This is a benchmark comparing copying time and RMA buffer registration time. +-- + +{-# LANGUAGE ForeignFunctionInterface #-} +import Control.Concurrent ( forkIO, newEmptyMVar, takeMVar, putMVar ) +import Control.Exception ( evaluate ) +import Control.Monad ( replicateM_, void, forM_, when, replicateM, forM ) +import qualified Data.ByteString as B ( copy, replicate ) +import Data.Time ( getCurrentTime, diffUTCTime ) +import Data.Word ( Word8 ) +import Foreign.C.Types ( CSize(..), CInt(..) ) +import Foreign.Marshal.Alloc ( mallocBytes, free, alloca ) +import Foreign.Ptr ( Ptr ) +import Foreign.Storable ( pokeByteOff, peek ) +import Network.CCI ( withCCI, withPollingEndpoint, rmaRegister + , rmaDeregister, RMA_MODE(..) + ) +import Numeric ( fromRat ) +import Text.Printf ( printf ) +import System.Environment ( getArgs ) + +foreign import ccall unsafe posix_memalign :: Ptr (Ptr a) -> CSize -> CSize -> IO CInt + +allocAligned :: Int -> Int -> IO (Ptr a) +allocAligned align size = alloca $ \ptr -> + do ret <- posix_memalign ptr (fromIntegral align) (fromIntegral size) + when (ret /= 0) $ error $ "allocAligned: " ++ show ret + peek ptr + +main :: IO () +main = do + [szLogArg] <- getArgs + let sz = 2 ^ (read szLogArg :: Int) + + iters :: Int + iters = 100 + + measureAndReport label action = do + start <- getCurrentTime + res <- action + end <- getCurrentTime + let t = toRational (diffUTCTime end start) * 1000000 / toRational iters + printf (label ++ " time: %8.2f us\n") (fromRat t :: Double) :: IO () + return res + + printf "Benchmarking buffers of %d bytes ...\n" sz :: IO () + + -- measure registration time with posix_memalign + do + withCCI $ withPollingEndpoint Nothing $ \ep -> do + + cbufs <- replicateM iters $ allocAligned 8 sz + hs <- measureAndReport "registration" $ + forM cbufs $ \cbuf -> rmaRegister ep (cbuf,sz) RMA_READ_WRITE + forM_ hs $ rmaDeregister ep + forM_ cbufs free + + -- measure registration time when reusing addresses + do + withCCI $ withPollingEndpoint Nothing $ \ep -> do + + cbuf <- allocAligned 8 sz + measureAndReport "registration+deregistration reusing address" $ + replicateM_ iters $ + rmaRegister ep (cbuf,sz) RMA_READ_WRITE >>= rmaDeregister ep + free cbuf + + -- measure deregistration time + do + withCCI $ withPollingEndpoint Nothing $ \ep -> do + + cbufs <- replicateM iters $ allocAligned 8 sz + hs <- forM cbufs $ \cbuf -> rmaRegister ep (cbuf,sz) RMA_READ_WRITE + measureAndReport "deregistration" $ + forM_ hs $ rmaDeregister ep + forM_ cbufs free + + -- measure copying time + do + buf <- evaluate $ B.replicate sz 0 + measureAndReport "copying" $ + let go _ 0 = return () + go x n = evaluate (B.copy x) >> go x (n-1) + in go buf iters + + -- measure allocation time + do + let idxs = takeWhile ( pokeByteOff buf i (0 :: Word8) + free buf + + -- measure mvar time + do + mv0 <- newEmptyMVar + mv1 <- newEmptyMVar + void $ forkIO $ replicateM_ iters $ takeMVar mv0 >> putMVar mv1 () + measureAndReport "mvar" $ + replicateM_ iters $ do + putMVar mv0 () + takeMVar mv1 diff --git a/test/Worker.hs b/test/Worker.hs new file mode 100644 --- /dev/null +++ b/test/Worker.hs @@ -0,0 +1,464 @@ +-- +-- Copyright (C) 2012 Parallel Scientific. All rights reserved. +-- +-- See the accompanying LICENSE file for license information. +-- + +-- This file implements worker processes. +-- +-- A worker process runs a loop which takes commands from a driver process and executes them. +-- The commands are CCI operations that the worker executes upon command reception. The +-- driver communicates with the workers through the standard input and output. The workers +-- communicate among themselves through CCI. +-- +-- The driver process must issue commands to indicate to the workers which connection requests +-- to accept and reject before the connection requests events arrive. For every command, the +-- worker process sends and Idle response to the driver, so the driver can synchronize all the +-- workers and make bugs reproducible. +-- +-- Worker processes are spawned by the driver process. +-- +-- See test/test_cci.hs for an example of driver process. +{-# LANGUAGE ForeignFunctionInterface #-} + +import Control.Exception ( catch, SomeException ) +import Control.Monad ( when ) +import Data.Binary ( decode, encode ) +import Data.ByteString ( ByteString ) +import qualified Data.ByteString as B ( concat, length, null, drop ) +import Data.ByteString.Lazy ( toChunks, fromChunks ) +import qualified Data.ByteString.Char8 as B8 ( unpack, pack, break ) +import Data.Char ( isDigit, isSpace ) +import Data.IORef ( newIORef, IORef, atomicModifyIORef, readIORef, writeIORef ) +import qualified Data.Map as M ( empty, lookup, insert, delete ) +import Data.Map ( Map ) +import Data.Maybe ( isNothing ) +import qualified Data.Set as S ( empty, insert, member, delete ) +import Data.Set ( Set ) +import qualified Data.IntSet as IS ( empty, IntSet, insert, member, delete ) +import Data.Word ( Word64 ) +import Foreign.Ptr ( WordPtr, Ptr ) +import Foreign.C.String ( castCharToCChar, castCCharToChar ) +import Foreign.C.Types ( CInt(..), CChar ) +import Foreign.Storable ( peek ) +import Foreign.Marshal.Alloc ( alloca ) +import Foreign.Marshal.Array ( pokeArray, peekArray ) +import System.IO ( hPutStrLn, stderr ) + +import Network.CCI ( withCCI, withPollingEndpoint, connect, ConnectionAttributes(..) + , EventData(..), disconnect, send, Connection + , accept, reject, Event, Status(..), unsafePackEventBytes + , getEndpt_URI, pollWithEventData, RMALocalHandle, RMARemoteHandle + , RMA_MODE(..), RMALocalHandle, RMARemoteHandle, rmaRegister + , rmaHandle2ByteString, createRMARemoteHandle, Endpoint, rmaWrite + , rmaRead, packEventBytes + ) + +import Commands ( initCommands,readCommand, msgToString + , Command(..), Msg(..) + , Response( Error,Recv,ReqAccepted,ReqRejected,ReqIgnored,ConnectAccepted + , SendCompletion, Rejected, TimedOut, KeepAliveTimedOut + , EndpointDeviceFailed, Idle + ) + ) +import qualified Commands as C ( sendResponse ) + +sendResponse :: Response -> IO () +sendResponse r = hPutStrLn stderr ("response: "++show r) >> C.sendResponse r + +main :: IO () +main = flip catch (\e -> sendResponse$ Error$ "Exception: "++show (e :: SomeException))$ do + initCommands + withCCI$ do + rcm <- emptyConnMap + rcrs <- emptyConnReq + rmar <- emptyRMAState + withPollingEndpoint Nothing$ \ep -> do + getEndpt_URI ep >>= putStrLn + getEndpt_URI ep >>= hPutStrLn stderr + processCommands rcm rcrs rmar ep + + where + + processCommands rcm rcrs rmar ep = + readCommand >>= \cm -> do + let go = processCommands rcm rcrs rmar ep + hPutStrLn stderr$ " command: "++show cm + case cm of + + ConnectTo uri _ i mt -> do + connect ep uri (B.concat$ toChunks$ encode (fromIntegral i :: Word64)) CONN_ATTR_RO i mt + sendResponse Idle >> go + + Accept i -> markAccept i rcrs >> sendResponse Idle >> go + + Reject i -> markReject i rcrs >> sendResponse Idle >> go + + Disconnect i -> do + c <- getConn' i rcm + disconnect c >> sendResponse Idle >> go + + Send i ctx bs -> do + c <- getConn' i rcm + send c (B8.pack$ msgToString bs) ctx + sendResponse Idle >> go + + WaitConnection cid -> do + waitConnection rmar rcm rcrs ep cid + sendResponse Idle + go + + WaitSendCompletion cid sid -> do + waitSendCompletion rmar rcm rcrs ep cid sid + sendResponse Idle + go + + WaitRecv cid rid -> do + waitRecv rmar rcm rcrs ep cid rid + sendResponse Idle + go + + RMAReuseRMAHandle cid -> do + markReuseRMAH cid rmar + sendResponse Idle + go + + RMAHandleExchange cid sid -> do + h <- createRMALocalHandle ep cid rmar + c <- getConn' cid rcm + bh <- rmaHandle2ByteString h + send c (B8.pack$ msgToString$ MsgRMAH$ bh) sid + insertRMAHandleSendId cid sid rmar + sendResponse Idle + go + + RMAWaitExchange cid -> do + waitRMAExchange rmar rcm rcrs ep cid + sendResponse Idle + go + + RMAPrepareRead cid ctx -> do + (_,ptr,n) <- getRMALocalHandle cid rmar + pokeArray ptr (map castCharToCChar$ take n$ cycle$ show ctx) + sendResponse Idle >> go + + RMAWrite cid ctx -> do + c <- getConn' cid rcm + (lh,ptr,n) <- getRMALocalHandle cid rmar + Just rh <- getRMARemoteHandle cid rmar + pokeArray ptr (map castCharToCChar$ take n$ cycle$ show ctx) + rmaWrite c (Just (B8.pack$ "rmaWrite "++show ctx)) rh 0 lh 0 (fromIntegral n) ctx [] + sendResponse Idle >> go + + RMARead cid ctx -> do + ci <- getConnInfo' cid rcm + (lh,_,n) <- getRMALocalHandle cid rmar + Just rh <- getRMARemoteHandle cid rmar + insertConnInfo cid ci { rmaReadIds = IS.insert (fromIntegral ctx) (rmaReadIds ci) } rcm + rmaRead (connection ci) (Just (B8.pack$ "rmaRead "++show ctx)) lh 0 rh 0 (fromIntegral n) ctx [] + sendResponse Idle >> go + + RMAWaitWrite cid ctx -> do + waitRMAWrite rmar rcm rcrs ep cid ctx + sendResponse Idle + go + + RMAWaitRead cid ctx -> do + waitRMARead rmar rcm rcrs ep cid ctx + sendResponse Idle + go + + RMAFreeHandles cid -> do + freeRMALocalHandle cid rmar + sendResponse Idle >> go + + Quit -> sendResponse Idle + + + waitConnection rmar rcm rcrs ep cid = do + mc <- getConn cid rcm + case mc of + Nothing -> do + pollWithEventData ep$ handleEvent rmar rcm rcrs + waitConnection rmar rcm rcrs ep cid + + _ -> return () + + waitRecv rmar rcm rcrs ep cid ri = do + ci <- getConnInfo' cid rcm + when (not$ IS.member (fromIntegral ri)$ recvs ci)$ do + pollWithEventData ep$ handleEvent rmar rcm rcrs + waitRecv rmar rcm rcrs ep cid ri + + waitSendCompletion rmar rcm rcrs ep cid si = do + ci <- getConnInfo' cid rcm + when (not$ IS.member (fromIntegral si)$ sendCompletions ci)$ do + pollWithEventData ep$ handleEvent rmar rcm rcrs + waitSendCompletion rmar rcm rcrs ep cid si + + waitRMAWrite rmar rcm rcrs ep cid si = do + ci <- getConnInfo' cid rcm + written <- testRMAWriteId cid rmar + when (not written && not (IS.member (fromIntegral si)$ sendCompletions ci))$ do + pollWithEventData ep$ handleEvent rmar rcm rcrs + waitRMAWrite rmar rcm rcrs ep cid si + + waitRMARead rmar rcm rcrs ep cid si = do + ci <- getConnInfo' cid rcm + when (not (IS.member (fromIntegral si)$ recvs ci) + && not (IS.member (fromIntegral si)$ sendCompletions ci))$ do + pollWithEventData ep$ handleEvent rmar rcm rcrs + waitRMARead rmar rcm rcrs ep cid si + + waitRMAExchange rmar rcm rcrs ep cid = do + mr <- getRMARemoteHandle cid rmar + Just si <- getRMAHandleSendId cid rmar + ci <- getConnInfo' cid rcm + when (isNothing mr || (not$ IS.member (fromIntegral si)$ sendCompletions ci))$ do + pollWithEventData ep$ handleEvent rmar rcm rcrs + waitRMAExchange rmar rcm rcrs ep cid + + handleEvent rmar rcm rcrs ev = do + hPutStrLn stderr$ " event: "++show ev + case ev of + EvAccept ctx (Right conn) -> insertConn ctx conn rcm >> sendResponse (ConnectAccepted ctx) + + EvConnect ctx (Right conn) -> insertConn ctx conn rcm >> sendResponse (ConnectAccepted ctx) + + EvConnect ctx (Left ETIMEDOUT) -> sendResponse (TimedOut ctx) + + EvConnect ctx (Left ECONNREFUSED) -> sendResponse (Rejected ctx) + + EvSend ctx st conn -> do + cid <- getConnId conn rcm + ci <- getConnInfo' cid rcm + when (ctx/=0)$ insertConnInfo cid ci { sendCompletions = IS.insert (fromIntegral ctx) (sendCompletions ci) } rcm + when (IS.member (fromIntegral ctx) (rmaReadIds ci))$ do + (_,ptr,n) <- getRMALocalHandle cid rmar + checkRMABuffer ptr n ctx + sendResponse (SendCompletion cid ctx st) + + EvRecv bs conn -> do + cid <- getConnId conn rcm + ci <- getConnInfo' cid rcm + bs' <- packEventBytes bs + let m = byteStringToMsg bs' + case m of + Msg ctx _ -> seq ctx$ insertConnInfo cid (ci { recvs = IS.insert (fromIntegral ctx) (recvs ci) }) rcm + MsgRMARead ctx -> seq ctx$ insertConnInfo cid (ci { recvs = IS.insert (fromIntegral ctx) (recvs ci) }) rcm + MsgRMAH rh -> insertRMARemoteHandle cid (maybe (error "handleEvent: MsgRMAH") id$ createRMARemoteHandle rh) rmar + MsgRMAWrite ctx -> do + (_,ptr,n) <- getRMALocalHandle cid rmar + checkRMABuffer ptr n ctx + insertRMAWriteId cid rmar + sendResponse (Recv cid m) + + EvConnectRequest e bs cattrs -> do + bs' <- unsafePackEventBytes bs + handleConnectionRequest rcrs e bs' cattrs + + EvKeepAliveTimedOut conn -> getConnId conn rcm >>= \c -> sendResponse (KeepAliveTimedOut c) + + EvEndpointDeviceFailed _ -> sendResponse EndpointDeviceFailed + + _ -> sendResponse$ Error$ "unhandled event: " ++ show ev + +-- | @loopWhileM p io@ performs @io@ repeteadly while its result satisfies @p@. +-- Yields the first offending result. +-- loopWhileM :: (a -> Bool) -> IO a -> IO a +-- loopWhileM p io = io >>= \a -> if p a then loopWhileM p io else return a + +byteStringToMsg :: ByteString -> Msg +byteStringToMsg bs = let (ctxs,rest) = B8.break isSpace bs + in if not (B.null ctxs) && wellFormed (B8.unpack ctxs) (B8.unpack rest) + then case B8.unpack ctxs of + "rmaH" -> MsgRMAH$ B.drop 1 rest + "rmaRead" -> MsgRMARead$ read$ drop 1$ B8.unpack rest + "rmaWrite" -> MsgRMAWrite$ read$ drop 1$ B8.unpack rest + _ -> Msg (read$ B8.unpack ctxs) (B.length bs) + else error$ "error parsing message "++show ctxs++" (length: "++show (B.length bs)++"): "++show rest + where + wellFormed "rmaH" (' ':rs) = length rs==32 + wellFormed "rmaRead" (' ':rs) = all isDigit rs + wellFormed "rmaWrite" (' ':rs) = all isDigit rs + wellFormed ctx (' ':rs) | all isDigit ctx = and$ zipWith (==) (cycle ctx) rs + wellFormed _ _ = False + +checkRMABuffer :: Ptr CChar -> Int -> WordPtr -> IO () +checkRMABuffer ptr n ctx = do + msg <- fmap (map castCCharToChar)$ peekArray n ptr + when (msg/= (take n$ cycle$ show ctx))$ ioError$ userError$ "checkRMABuffer: "++msg + +-- Map of connection requests + +type ConnReqs = IORef ConnReqsD + +-- | This is a map that specifies whether a connection request with a specific identifier +-- should be accepted or rejected upon reception. +data ConnReqsD = ConnReqsD + { connAccept :: Set WordPtr -- ^ Requests with these identifiers should be accepted. + , connReject :: Set WordPtr -- ^ Requests with these identifiers should be rejected. + } + +emptyConnReq :: IO ConnReqs +emptyConnReq = newIORef$ ConnReqsD S.empty S.empty + +markAccept :: WordPtr -> ConnReqs -> IO () +markAccept i rcrs = atomicModifyIORef rcrs$ + \cr -> (cr { connAccept = S.insert i (connAccept cr) } , ()) + +markReject :: WordPtr -> ConnReqs -> IO () +markReject i rcrs = atomicModifyIORef rcrs$ + \cr -> (cr { connReject = S.insert i (connReject cr) } , ()) + +handleConnectionRequest :: ConnReqs -> Event s -> ByteString -> ConnectionAttributes -> IO () +handleConnectionRequest rcrs ev bs _cattrs = do + r <- atomicModifyIORef rcrs$ \cr -> + if S.member w (connAccept cr) then (cr { connAccept = S.delete w (connAccept cr) } , ReqAccepted w) + else if S.member w (connReject cr) then (cr { connReject = S.delete w (connReject cr) } , ReqRejected w) + else (cr,ReqIgnored w) + case r of + ReqAccepted _ -> accept ev w >> sendResponse r + ReqRejected _ -> reject ev >> sendResponse r + _ -> sendResponse r + where + w = (fromIntegral :: Word64 -> WordPtr)$ decode$ fromChunks [bs] + + +-- Map of connections + +type ConnMap = IORef (Map WordPtr ConnectionInfo,Map Connection WordPtr) +data ConnectionInfo = ConnInfo + { connection :: Connection + , sendCompletions :: IS.IntSet + , rmaReadIds :: IS.IntSet + , recvs :: IS.IntSet + } + + +emptyConnMap :: IO ConnMap +emptyConnMap = newIORef (M.empty,M.empty) + +getConnInfo :: WordPtr -> ConnMap -> IO (Maybe ConnectionInfo) +getConnInfo w rcm = readIORef rcm >>= return . M.lookup w . fst + +getConnInfo' :: WordPtr -> ConnMap -> IO ConnectionInfo +getConnInfo' w rcm = getConnInfo w rcm >>= maybe (do + sendResponse (Error$ "unknown connection: "++show w) + ioError$ userError$ "unknown connection: "++show w + ) return + + +getConn :: WordPtr -> ConnMap -> IO (Maybe Connection) +getConn w rcm = getConnInfo w rcm >>= return . fmap connection + +-- | Fails with an error if the connection is not in the map. +getConn' :: WordPtr -> ConnMap -> IO Connection +getConn' w rcm = getConn w rcm >>= maybe (do + sendResponse (Error$ "unknown connection: "++show w) + ioError$ userError$ "unknown connection: "++show w + ) return + +getConnId :: Connection -> ConnMap -> IO WordPtr +getConnId c rcm = + readIORef rcm >>= + maybe (sendResponse (Error$ "The given connection was not found in the connection map.") + >> ioError (userError "Cannot find word.") + ) + return + . M.lookup c . snd + +insertConn :: WordPtr -> Connection -> ConnMap -> IO () +insertConn w c rcm = insertConnInfo w (ConnInfo c IS.empty IS.empty IS.empty) rcm + +insertConnInfo :: WordPtr -> ConnectionInfo -> ConnMap -> IO () +insertConnInfo w ci rcm = atomicModifyIORef rcm $ \(wc,cw) -> ((M.insert w ci wc, M.insert (connection ci) w cw),()) + +---------------------- +-- RMA state +---------------------- + +data RMAState = RMAState + { reused :: Set WordPtr + , reservedLocalHandles :: Map WordPtr (RMALocalHandle,Ptr CChar,Int) + , availableHandles :: [(RMALocalHandle,Ptr CChar,Int)] + , remoteHandles :: Map WordPtr RMARemoteHandle + , localHandlesSendIds :: Map WordPtr WordPtr + , rmaWritesIds :: IS.IntSet + } + +emptyRMAState :: IO (IORef RMAState) +emptyRMAState = newIORef RMAState + { reused = S.empty + , availableHandles = [] + , remoteHandles = M.empty + , localHandlesSendIds = M.empty + , reservedLocalHandles = M.empty + , rmaWritesIds = IS.empty + } + +markReuseRMAH :: WordPtr -> IORef RMAState -> IO () +markReuseRMAH w r = atomicModifyIORef r (\rmas -> (rmas { reused = S.insert w (reused rmas) },())) + + +createRMALocalHandle :: Endpoint -> WordPtr -> IORef RMAState -> IO RMALocalHandle +createRMALocalHandle ep cid rmar = do + rmas <- readIORef rmar + let h@(lh',_,_):hss = availableHandles rmas + lh <- if S.member cid (reused rmas) && not (null$ availableHandles rmas) + then do + writeIORef rmar + rmas { reused = S.delete cid (reused rmas) + , availableHandles = hss + , reservedLocalHandles = M.insert cid h (reservedLocalHandles rmas) + } + return lh' + + else do + ptr <- alloca$ \pptr -> posix_memalign pptr 4096 4096 >> peek pptr + lh <- rmaRegister ep (ptr,4096) RMA_READ_WRITE + writeIORef rmar + rmas { reservedLocalHandles = M.insert cid (lh,ptr,4096) (reservedLocalHandles rmas) + } + return lh + + return lh + +insertRMARemoteHandle :: WordPtr -> RMARemoteHandle -> IORef RMAState -> IO () +insertRMARemoteHandle cid rh rmar = do + atomicModifyIORef rmar$ \rmas -> ( rmas { remoteHandles = M.insert cid rh (remoteHandles rmas) } , ()) + +getRMARemoteHandle :: WordPtr -> IORef RMAState -> IO (Maybe RMARemoteHandle) +getRMARemoteHandle cid rmar = readIORef rmar >>= return . M.lookup cid . remoteHandles + +getRMAHandleSendId :: WordPtr -> IORef RMAState -> IO (Maybe WordPtr) +getRMAHandleSendId cid rmar = readIORef rmar >>= return . M.lookup cid . localHandlesSendIds + +getRMALocalHandle :: WordPtr -> IORef RMAState -> IO (RMALocalHandle,Ptr CChar,Int) +getRMALocalHandle cid rmar = readIORef rmar >>= return . maybe (error "getRMALocalHandle") id . M.lookup cid . reservedLocalHandles + +freeRMALocalHandle :: WordPtr -> IORef RMAState -> IO () +freeRMALocalHandle cid rmar = atomicModifyIORef rmar$ \rmas -> + let h = maybe (error "freeRMALocalHandle") id$ M.lookup cid (reservedLocalHandles rmas) + in ( rmas { availableHandles = availableHandles rmas ++ [ h ] + , reservedLocalHandles = M.delete cid (reservedLocalHandles rmas) + , remoteHandles = M.delete cid (remoteHandles rmas) + } + , () + ) + +insertRMAHandleSendId :: WordPtr -> WordPtr -> IORef RMAState -> IO () +insertRMAHandleSendId cid sid rmar = atomicModifyIORef rmar (\rmas -> (rmas { localHandlesSendIds = M.insert cid sid (localHandlesSendIds rmas) }, ())) + +insertRMAWriteId :: WordPtr -> IORef RMAState -> IO () +insertRMAWriteId cid rmar = atomicModifyIORef rmar (\rmas -> (rmas { rmaWritesIds = IS.insert (fromIntegral cid) (rmaWritesIds rmas) }, ())) + +testRMAWriteId :: WordPtr -> IORef RMAState -> IO Bool +testRMAWriteId cid rmar = atomicModifyIORef rmar (\rmas -> (rmas { rmaWritesIds = IS.delete (fromIntegral cid) (rmaWritesIds rmas) } + , IS.member (fromIntegral cid)$ rmaWritesIds rmas + ) + ) + +-- int posix_memalign(void **memptr, size_t alignment, size_t size); +foreign import ccall "static stdlib.h" posix_memalign :: Ptr (Ptr CChar) -> CInt -> CInt -> IO CInt + diff --git a/test/test_cci.hs b/test/test_cci.hs new file mode 100644 --- /dev/null +++ b/test/test_cci.hs @@ -0,0 +1,130 @@ +-- +-- Copyright (C) 2012 Parallel Scientific. All rights reserved. +-- +-- See the accompanying LICENSE file for license information. +-- +-- +-- This file implements an executable which runs automatically +-- generated tests for the CCI haskell bindings. +-- +import Control.Monad ( when ) +import qualified Data.Map as M( empty, insert, lookup ) +import GHC.Exts ( groupWith ) +import System.Exit ( exitFailure ) + +import TestGen ( testProp, defaultTestConfig, runCommands, Command(..), Msg(..) + , ProcCommand, Response(..), TestConfig(..), generateCTest ) +import Network.CCI ( Status(..) ) +import System.Console.CmdArgs + +main :: IO () +main = props + +props :: IO () +props = do + tc <- cmdArgs (TestConfig { nSends= intopt "Amount of messages sent per interaction" (nSends defaultTestConfig) + , nTries= intopt "Amount of command sequences to generate" (nTries defaultTestConfig) + , nProcesses= intopt "Amount of processes to test" (nProcesses defaultTestConfig) + , nMinMsgLen= intopt "Minimum size of messages to test" (nMinMsgLen defaultTestConfig) + , nMaxMsgLen= intopt "Maximum size of messages to test" (nMaxMsgLen defaultTestConfig) + , nPerProcessInteractions= + intopt "Amount of interactions each process initiates during a test" (nPerProcessInteractions defaultTestConfig) + &= name "nPerProcRuns" &= explicit + , nErrors= intopt "Amount of errors to collect before stopping testing" (nErrors defaultTestConfig) + , withValgrind = boolopt "Run tests with valgrind" (withValgrind defaultTestConfig) &= name "with-valgrind" &= explicit + , testRMA = boolopt "Test rma operations" (testRMA defaultTestConfig) &= name "test-rma" &= explicit + } + &= summary "CCI test generator v0.0.1, (C) Parallel Scientific 2012" + &= details ["See test/README for details."] + &= program "test_cci" + ) + errs <- testProp tc printErrorAndGenerateTest$ \cmds rs -> + [ ( "sends equal completions" + , testRMA tc || length (filter (isSendCommand . snd) cmds) == length (concatMap (filter isSendCompletion) rs) + ) + , ( "messages arrive sorted" + , testRMA tc || matchSendRecvs (collectSends cmds) (collectRecvs rs) + ) + ] + when (not (null errs))$ exitFailure + where + isSendCompletion (SendCompletion _ mid _) = mid>0 + isSendCompletion _ = False + intopt :: (Default a,Data a,Show a) => String -> a -> a + intopt s i = i &= help (s++" [default: "++show i++"]") &= opt i + boolopt :: (Default a,Data a,Show a) => String -> a -> a + boolopt s i = i &= help (s++" [default: "++show i++"]") + + printErrorAndGenerateTest e@(s,_,_) i = do + print e + let cfilename :: Int -> String + cfilename i = "t"++show i++".c" + writeFile (cfilename i)$ generateCTest s + putStrLn$ "A test program has been written to " ++ show (cfilename i) + + +isSendCommand :: Command -> Bool +isSendCommand (Send _ _ _) = True +isSendCommand _ = False + +collectSends :: [ProcCommand] -> [(Int,[[Command]])] +collectSends = map toPair' + . groupWith fst + . map toPair + . groupWith destinationAndConnection + . filter (isSendCommand . snd . snd) + . attachProcDestination + where + destinationAndConnection (i,(_,Send cid _ _)) = (i,cid) + destinationAndConnection _ = error "Props.destinationAndConnection: unexpected value" + + toPair ls@((Just i,_):_) = (i,map (snd . snd) ls) + toPair _ = error "Props.collectSends: unexpected value" + + toPair' ls@((i,_):_) = (i,map snd ls) + toPair' _ = error "Props.collectSends toPair': unexpected value" + +attachProcDestination :: [ProcCommand] -> [(Maybe Int,ProcCommand)] +attachProcDestination = go M.empty + where go m (c@(tprocs,ConnectTo _ pid cid _):cms) = + (Nothing,c) : go (M.insert cid (pid,tprocs) m) cms + go m (c@(tprocs,Send cid _ _):cms) = + ( fmap (\(pid,tps) -> if elem pid tprocs then head tps else pid) + $ M.lookup cid m + , c + ) : go m cms + go m (c:cms) = (Nothing,c) : go m cms + go _ [] = [] + +collectRecvs :: [[Response]] -> [[[Response]]] +collectRecvs = map (groupWith getConn) . map (filter isRecvResp) + where + isRecvResp (Recv _ (Msg _ _)) = True + isRecvResp _ = False + getConn (Recv cid _) = cid + getConn _ = error "Props.collectRecvs: unexpected value." + + +matchSendRecvs :: [(Int,[[Command]])] -> [[[Response]]] -> Bool +matchSendRecvs [] _ = True +matchSendRecvs ((i,ss):cms) rs = match' ss (rs!!i) && matchSendRecvs cms rs + where match' ss' rs' = length ss' == length rs' && and (zipWith match'' ss' rs') + match'' ss'' rs'' = length ss'' == length rs'' && and (zipWith matchCR ss'' rs'') + matchCR (Send cid _ msg) (Recv cid' msg') = cid == cid' && msg == msg' + matchCR _ _ = error "Props.matchSendRecvs: unexpected values" + + + +singleTest :: IO () +singleTest = runCommands test2 >>= print + + +test :: [ProcCommand] +test = [([0],Accept 5),([1],ConnectTo "" 0 5 Nothing),([1,0],WaitConnection 5),([1],Send 5 1 (Msg 1 185)),([0],WaitRecv 5 1),([1],Send 5 2 (Msg 2 969)),([1],WaitSendCompletion 5 1),([1],WaitSendCompletion 5 2),([1],Send 5 3 (Msg 3 495)),([1],WaitSendCompletion 5 3),([0],WaitRecv 5 2),([0],WaitRecv 5 3),([1],Disconnect 5)] + +testR :: [[Response]] +testR = [[ReqAccepted 5,ConnectAccepted 5,Recv 5 (Msg 0 185),Recv 5 (Msg 2 495),Recv 5 (Msg 1 969)],[ConnectAccepted 5,SendCompletion 5 0 SUCCESS,SendCompletion 5 1 SUCCESS,SendCompletion 5 2 SUCCESS]] + +test2 :: [ProcCommand] +test2 = [] +