diff --git a/README.md b/README.md --- a/README.md +++ b/README.md @@ -179,11 +179,12 @@ ### Running the tests -There are three test cases provided in `tests`: +There are four test cases provided in `tests`: ``` stack build --test --no-run-tests -mpirun -np 3 --mca btl self,vader --oversubscribe stack exec $(stack path --dist-dir)/build/mpi-test/mpi-test && echo SUCCESS || echo FAILURE -mpirun -np 3 --mca btl self,vader --oversubscribe stack exec $(stack path --dist-dir)/build/mpi-test-packing/mpi-test-packing && echo SUCCESS || echo FAILURE -mpirun -np 3 --mca btl self,vader --oversubscribe stack exec $(stack path --dist-dir)/build/mpi-test-store/mpi-test-store && echo SUCCESS || echo FAILURE +mpirun-openmpi-mp -np 3 --mca btl self,vader --oversubscribe stack exec $(stack path --dist-dir)/build/mpi-test/mpi-test && echo SUCCESS || echo FAILURE +mpirun-openmpi-mp -np 3 --mca btl self,vader --oversubscribe stack exec $(stack path --dist-dir)/build/mpi-test-binary/mpi-test-binary && echo SUCCESS || echo FAILURE +mpirun-openmpi-mp -np 3 --mca btl self,vader --oversubscribe stack exec $(stack path --dist-dir)/build/mpi-test-serialize/mpi-test-serialize && echo SUCCESS || echo FAILURE +mpirun-openmpi-mp -np 3 --mca btl self,vader --oversubscribe stack exec $(stack path --dist-dir)/build/mpi-test-store/mpi-test-store && echo SUCCESS || echo FAILURE ``` diff --git a/lib/Control/Distributed/MPI/Binary.hs b/lib/Control/Distributed/MPI/Binary.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Distributed/MPI/Binary.hs @@ -0,0 +1,403 @@ +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE TypeApplications #-} + +-- | Module: Control.Distributed.MPI.Binary +-- Description: Simplified MPI bindings with automatic serialization +-- based on Data.Binary +-- Copyright: (C) 2018 Erik Schnetter +-- License: Apache-2.0 +-- Maintainer: Erik Schnetter +-- Stability: experimental +-- Portability: Requires an externally installed MPI library + +module Control.Distributed.MPI.Binary + ( -- * Types, and associated functions and constants + MPIException(..) + + -- ** Communicators + , Comm(..) + , commSelf + , commWorld + + -- ** Message sizes + , Count(..) + , fromCount + , toCount + + -- ** Process ranks + , Rank(..) + , anySource + , commRank + , commSize + , fromRank + , rootRank + , toRank + + -- ** Message status + , Status(..) + + -- ** Message tags + , Tag(..) + , anyTag + , fromTag + , toTag + , unitTag + + , Request + + -- * Functions + + -- ** Initialization and shutdown + , abort + , mainMPI + + -- ** Point-to-point (blocking) + , recv + , recv_ + , send + , sendrecv + , sendrecv_ + + -- ** Point-to-point (non-blocking) + , irecv + , isend + , test + , test_ + , wait + , wait_ + + -- ** Collective (blocking) + , barrier + , bcastRecv + , bcastSend + + -- ** Collective (non-blocking) + , ibarrier + , ibcastRecv + , ibcastSend + ) where + +import Prelude hiding (init) + +import Control.Concurrent +import Control.Exception +import Control.Monad +import Control.Monad.Loops +import qualified Data.Binary as Binary +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as BL +import qualified Data.ByteString.Unsafe as B +import Data.Coerce +import Data.Typeable +import Foreign +import Foreign.C.Types + +import qualified Control.Distributed.MPI as MPI +import Control.Distributed.MPI + ( Comm(..) + , commSelf + , commWorld + , Count(..) + , fromCount + , toCount + , Rank(..) + , anySource + , commRank + , commSize + , fromRank + , rootRank + , toRank + , Tag(..) + , anyTag + , fromTag + , toTag + , unitTag + , abort + , barrier + ) + + + +-- Serialization, based on Data.Binary +type CanSerialize a = Binary.Binary a +serialize :: CanSerialize a => a -> IO B.ByteString +serialize = return . BL.toStrict . Binary.encode +deserialize :: CanSerialize a => B.ByteString -> IO a +deserialize = return . Binary.decode . BL.fromStrict + +-- Why is this necessary? Shouldn't 'Data.Binary' already contain +-- this? +instance Binary.Binary CInt where + put x = Binary.put @Int32 (coerce x) + get = coerce <$> (Binary.get @Int32) +instance Binary.Binary Rank + + + +-- | Run the supplied Maybe computation repeatedly while it returns +-- Nothing. If it returns a value, then returns that value. +whileNothing :: Monad m => m (Maybe a) -> m () -> m a +whileNothing cond loop = go + where go = do mx <- cond + case mx of + Nothing -> do loop + go + Just x -> return x + + + +-- | Exception type indicating an error in a call to MPI +newtype MPIException = MPIException String + deriving (Eq, Ord, Read, Show, Typeable) +instance Exception MPIException + +mpiAssert :: Bool -> String -> IO () +mpiAssert cond msg = + do when (not cond) $ throw (MPIException msg) + return () + + + +data DidInit = DidInit | DidNotInit + +initMPI :: IO DidInit +initMPI = + do isInit <- MPI.initialized + if isInit + then return DidNotInit + else do ts <- MPI.initThread MPI.ThreadMultiple + mpiAssert (ts >= MPI.ThreadMultiple) + ("MPI.init: Insufficient thread support: requiring " ++ + show MPI.ThreadMultiple ++ + ", but MPI library provided only " ++ show ts) + return DidInit + +finalizeMPI :: DidInit -> IO () +finalizeMPI DidInit = + do isFinalized <- MPI.finalized + if isFinalized + then return () + else do MPI.finalize +finalizeMPI DidNotInit = return () + +-- | Convenience function to initialize and finalize MPI. This +-- initializes MPI with 'ThreadMultiple' thread support. +mainMPI :: IO () -- ^ action to run with MPI, typically the whole program + -> IO () +mainMPI action = bracket initMPI finalizeMPI (\_ -> action) + + + +-- | A communication request, usually created by a non-blocking +-- communication function. +newtype Request a = Request (MVar (Status, a)) + +-- | The status of a finished communication, indicating rank and tag +-- of the other communication end point. +data Status = Status { msgRank :: !Rank + , msgTag :: !Tag + } + deriving (Eq, Ord, Read, Show) + + + +-- | Receive an object. +recv :: CanSerialize a + => Rank -- ^ Source rank + -> Tag -- ^ Source tag + -> Comm -- ^ Communicator + -> IO (Status, a) -- ^ Message status and received object +recv recvrank recvtag comm = + do status <- whileNothing (MPI.iprobe recvrank recvtag comm) yield + source <- MPI.getSource status + tag <- MPI.getTag status + count <- MPI.getCount status MPI.datatypeByte + let len = MPI.fromCount count + ptr <- mallocBytes len + buffer <- B.unsafePackMallocCStringLen (ptr, len) + req <- MPI.irecv buffer source tag comm + whileM_ (not <$> MPI.test_ req) yield + recvobj <- deserialize buffer + return (Status source tag, recvobj) + +-- | Receive an object without returning a status. +recv_ :: CanSerialize a + => Rank -- ^ Source rank + -> Tag -- ^ Source tag + -> Comm -- ^ Communicator + -> IO a -- ^ Received object +recv_ recvrank recvtag comm = + snd <$> recv recvrank recvtag comm + +-- | Send an object. +send :: CanSerialize a + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Message tag + -> Comm -- ^ Communicator + -> IO () +send sendobj sendrank sendtag comm = + do sendbuf <- serialize sendobj + -- Use 'unsafeUseAsCStringLen' to ensure 'sendbuf' is not freed + -- too early + B.unsafeUseAsCStringLen sendbuf $ \_ -> + do req <- MPI.isend sendbuf sendrank sendtag comm + whileM_ (not <$> MPI.test_ req) yield + +-- | Send and receive objects simultaneously. +sendrecv :: (CanSerialize a, CanSerialize b) + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Send message tag + -> Rank -- ^ Source rank + -> Tag -- ^ Receive message tag + -> Comm -- ^ Communicator + -> IO (Status, b) -- ^ Message status and received object +sendrecv sendobj sendrank sendtag recvrank recvtag comm = + do recvreq <- irecv recvrank recvtag comm + send sendobj sendrank sendtag comm + wait recvreq + +-- | Send and receive objects simultaneously, without returning a +-- status for the received message. +sendrecv_ :: (CanSerialize a, CanSerialize b) + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Send message tag + -> Rank -- ^ Source rank + -> Tag -- ^ Receive message tag + -> Comm -- ^ Communicator + -> IO b -- ^ Received object +sendrecv_ sendobj sendrank sendtag recvrank recvtag comm = + snd <$> sendrecv sendobj sendrank sendtag recvrank recvtag comm + +-- | Begin to receive an object. Call `test` or `wait` to finish the +-- communication, and to obtain the received object. +irecv :: CanSerialize a + => Rank -- ^ Source rank + -> Tag -- ^ Source tag + -> Comm -- ^ Communicator + -> IO (Request a) -- ^ Communication request +irecv recvrank recvtag comm = + do result <- newEmptyMVar + _ <- forkIO $ + do res <- recv recvrank recvtag comm + putMVar result res + return (Request result) + +-- | Begin to send an object. Call 'test' or 'wait' to finish the +-- communication. +isend :: CanSerialize a + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Message tag + -> Comm -- ^ Communicator + -> IO (Request ()) -- ^ Communication request +isend sendobj sendrank sendtag comm = + do result <- newEmptyMVar + _ <- forkIO $ do send sendobj sendrank sendtag comm + putMVar result (Status sendrank sendtag, ()) + return (Request result) + +-- | Check whether a communication has finished, and return the +-- communication result if so. +test :: Request a -- ^ Communication request + -> IO (Maybe (Status, a)) -- ^ 'Just' communication result, if + -- communication has finished, else 'Nothing' +test (Request result) = tryTakeMVar result + +-- | Check whether a communication has finished, and return the +-- communication result if so, without returning a message status. +test_ :: Request a -- ^ Communication request + -> IO (Maybe a) -- ^ 'Just' communication result, if + -- communication has finished, else 'Nothing' +test_ req = fmap snd <$> test req + +-- | Wait for a communication to finish and return the communication +-- result. +wait :: Request a -- ^ Communication request + -> IO (Status, a) -- ^ Message status and communication result +wait (Request result) = takeMVar result + +-- | Wait for a communication to finish and return the communication +-- result, without returning a message status. +wait_ :: Request a -- ^ Communication request + -> IO a -- ^ Communication result +wait_ req = snd <$> wait req + + + +-- | Broadcast a message from one process (the "root") to all other +-- processes in the communicator. Call this function on all non-root +-- processes. Call 'bcastSend' instead on the root process. +bcastRecv :: CanSerialize a + => Rank + -> Comm + -> IO a +bcastRecv root comm = + do rank <- MPI.commRank comm + mpiAssert (rank /= root) "bcastRecv: expected rank /= root" + lenbuf <- mallocForeignPtr @CLong + lenreq <- MPI.ibcast (lenbuf, 1::Int) root comm + whileM_ (not <$> MPI.test_ lenreq) yield + len <- withForeignPtr lenbuf peek + ptr <- mallocBytes (fromIntegral len) + recvbuf <- B.unsafePackMallocCStringLen (ptr, fromIntegral len) + req <- MPI.ibcast recvbuf root comm + whileM_ (not <$> MPI.test_ req) yield + recvobj <- deserialize recvbuf + return recvobj + +-- | Broadcast a message from one process (the "root") to all other +-- processes in the communicator. Call this function on the root +-- process. Call 'bcastRecv' instead on all non-root processes. +bcastSend :: CanSerialize a + => a + -> Rank + -> Comm + -> IO () +bcastSend sendobj root comm = + do rank <- MPI.commRank comm + mpiAssert (rank == root) "bcastSend: expected rank == root" + sendbuf <- serialize sendobj + lenbuf <- mallocForeignPtr @CLong + withForeignPtr lenbuf $ \ptr -> poke ptr (fromIntegral (B.length sendbuf)) + lenreq <- MPI.ibcast (lenbuf, 1::Int) root comm + whileM_ (not <$> MPI.test_ lenreq) yield + req <- MPI.ibcast sendbuf root comm + whileM_ (not <$> MPI.test_ req) yield + +ibcastRecv :: CanSerialize a + => Rank + -> Comm + -> IO (Request a) +ibcastRecv root comm = + do result <- newEmptyMVar + _ <- forkIO $ + do recvobj <- bcastRecv root comm + putMVar result (Status root MPI.anyTag, recvobj) + return (Request result) + +ibcastSend :: CanSerialize a + => a + -> Rank + -> Comm + -> IO (Request ()) +ibcastSend sendobj root comm = + do result <- newEmptyMVar + _ <- forkIO $ + do bcastSend sendobj root comm + putMVar result (Status root MPI.anyTag, ()) + return (Request result) + +-- | Begin a barrier. Call 'test' or 'wait' to finish the +-- communication. +ibarrier :: Comm + -> IO (Request ()) +ibarrier comm = + do result <- newEmptyMVar + _ <- forkIO $ + do req <- MPI.ibarrier comm + whileM_ (not <$> MPI.test_ req) yield + putMVar result (Status MPI.anySource MPI.anyTag, ()) + return (Request result) diff --git a/lib/Control/Distributed/MPI/Packing.hs b/lib/Control/Distributed/MPI/Packing.hs deleted file mode 100644 --- a/lib/Control/Distributed/MPI/Packing.hs +++ /dev/null @@ -1,396 +0,0 @@ -{-# LANGUAGE ConstraintKinds #-} -{-# LANGUAGE TypeApplications #-} - --- | Module: Control.Distributed.MPI.Packing --- Description: Simplified MPI bindings with automatic serialization --- based on GHC.Packing --- Copyright: (C) 2018 Erik Schnetter --- License: Apache-2.0 --- Maintainer: Erik Schnetter --- Stability: experimental --- Portability: Requires an externally installed MPI library - -module Control.Distributed.MPI.Packing - ( -- * Types, and associated functions and constants - MPIException(..) - - -- ** Communicators - , Comm(..) - , commSelf - , commWorld - - -- ** Message sizes - , Count(..) - , fromCount - , toCount - - -- ** Process ranks - , Rank(..) - , anySource - , commRank - , commSize - , fromRank - , rootRank - , toRank - - -- ** Message status - , Status(..) - - -- ** Message tags - , Tag(..) - , anyTag - , fromTag - , toTag - , unitTag - - , Request - - -- * Functions - - -- ** Initialization and shutdown - , abort - , mainMPI - - -- ** Point-to-point (blocking) - , recv - , recv_ - , send - , sendrecv - , sendrecv_ - - -- ** Point-to-point (non-blocking) - , irecv - , isend - , test - , test_ - , wait - , wait_ - - -- ** Collective (blocking) - , barrier - , bcastRecv - , bcastSend - - -- ** Collective (non-blocking) - , ibarrier - , ibcastRecv - , ibcastSend - ) where - -import Prelude hiding (init) - -import Control.Concurrent -import Control.Exception -import Control.Monad -import Control.Monad.Loops -import qualified Data.ByteString.Lazy as BL -import qualified Data.ByteString as B -import qualified Data.ByteString.Unsafe as B -import qualified Data.Binary as Binary -import Data.Typeable -import Foreign -import Foreign.C.Types -import qualified GHC.Packing as Packing - -import qualified Control.Distributed.MPI as MPI -import Control.Distributed.MPI - ( Comm(..) - , commSelf - , commWorld - , Count(..) - , fromCount - , toCount - , Rank(..) - , anySource - , commRank - , commSize - , fromRank - , rootRank - , toRank - , Tag(..) - , anyTag - , fromTag - , toTag - , unitTag - , abort - , barrier - ) - - - --- Serialization, based on GHC.Packing -type CanSerialize a = Typeable a -serialize :: CanSerialize a => a -> IO B.ByteString -serialize = fmap (BL.toStrict . Binary.encode) . Packing.trySerialize -deserialize :: CanSerialize a => B.ByteString -> IO a -deserialize = Packing.deserialize . Binary.decode . BL.fromStrict - - - --- | Run the supplied Maybe computation repeatedly while it returns --- Nothing. If it returns a value, then returns that value. -whileNothing :: Monad m => m (Maybe a) -> m () -> m a -whileNothing cond loop = go - where go = do mx <- cond - case mx of - Nothing -> do loop - go - Just x -> return x - - - --- | Exception type indicating an error in a call to MPI -newtype MPIException = MPIException String - deriving (Eq, Ord, Read, Show, Typeable) -instance Exception MPIException - -mpiAssert :: Bool -> String -> IO () -mpiAssert cond msg = - do when (not cond) $ throw (MPIException msg) - return () - - - -data DidInit = DidInit | DidNotInit - -initMPI :: IO DidInit -initMPI = - do isInit <- MPI.initialized - if isInit - then return DidNotInit - else do ts <- MPI.initThread MPI.ThreadMultiple - mpiAssert (ts >= MPI.ThreadMultiple) - ("MPI.init: Insufficient thread support: requiring " ++ - show MPI.ThreadMultiple ++ - ", but MPI library provided only " ++ show ts) - return DidInit - -finalizeMPI :: DidInit -> IO () -finalizeMPI DidInit = - do isFinalized <- MPI.finalized - if isFinalized - then return () - else do MPI.finalize -finalizeMPI DidNotInit = return () - --- | Convenience function to initialize and finalize MPI. This --- initializes MPI with 'ThreadMultiple' thread support. -mainMPI :: IO () -- ^ action to run with MPI, typically the whole program - -> IO () -mainMPI action = bracket initMPI finalizeMPI (\_ -> action) - - - --- | A communication request, usually created by a non-blocking --- communication function. -newtype Request a = Request (MVar (Status, a)) - --- | The status of a finished communication, indicating rank and tag --- of the other communication end point. -data Status = Status { msgRank :: !Rank - , msgTag :: !Tag - } - deriving (Eq, Ord, Read, Show) - - - --- | Receive an object. -recv :: CanSerialize a - => Rank -- ^ Source rank - -> Tag -- ^ Source tag - -> Comm -- ^ Communicator - -> IO (Status, a) -- ^ Message status and received object -recv recvrank recvtag comm = - do status <- whileNothing (MPI.iprobe recvrank recvtag comm) yield - source <- MPI.getSource status - tag <- MPI.getTag status - count <- MPI.getCount status MPI.datatypeByte - let len = MPI.fromCount count - ptr <- mallocBytes len - buffer <- B.unsafePackMallocCStringLen (ptr, len) - req <- MPI.irecv buffer source tag comm - whileM_ (not <$> MPI.test_ req) yield - recvobj <- deserialize buffer - return (Status source tag, recvobj) - --- | Receive an object without returning a status. -recv_ :: CanSerialize a - => Rank -- ^ Source rank - -> Tag -- ^ Source tag - -> Comm -- ^ Communicator - -> IO a -- ^ Received object -recv_ recvrank recvtag comm = - snd <$> recv recvrank recvtag comm - --- | Send an object. -send :: CanSerialize a - => a -- ^ Object to send - -> Rank -- ^ Destination rank - -> Tag -- ^ Message tag - -> Comm -- ^ Communicator - -> IO () -send sendobj sendrank sendtag comm = - do sendbuf <- serialize sendobj - -- Use 'unsafeUseAsCStringLen' to ensure 'sendbuf' is not freed - -- too early - B.unsafeUseAsCStringLen sendbuf $ \_ -> - do req <- MPI.isend sendbuf sendrank sendtag comm - whileM_ (not <$> MPI.test_ req) yield - --- | Send and receive objects simultaneously. -sendrecv :: (CanSerialize a, CanSerialize b) - => a -- ^ Object to send - -> Rank -- ^ Destination rank - -> Tag -- ^ Send message tag - -> Rank -- ^ Source rank - -> Tag -- ^ Receive message tag - -> Comm -- ^ Communicator - -> IO (Status, b) -- ^ Message status and received object -sendrecv sendobj sendrank sendtag recvrank recvtag comm = - do recvreq <- irecv recvrank recvtag comm - send sendobj sendrank sendtag comm - wait recvreq - --- | Send and receive objects simultaneously, without returning a --- status for the received message. -sendrecv_ :: (CanSerialize a, CanSerialize b) - => a -- ^ Object to send - -> Rank -- ^ Destination rank - -> Tag -- ^ Send message tag - -> Rank -- ^ Source rank - -> Tag -- ^ Receive message tag - -> Comm -- ^ Communicator - -> IO b -- ^ Received object -sendrecv_ sendobj sendrank sendtag recvrank recvtag comm = - snd <$> sendrecv sendobj sendrank sendtag recvrank recvtag comm - --- | Begin to receive an object. Call `test` or `wait` to finish the --- communication, and to obtain the received object. -irecv :: CanSerialize a - => Rank -- ^ Source rank - -> Tag -- ^ Source tag - -> Comm -- ^ Communicator - -> IO (Request a) -- ^ Communication request -irecv recvrank recvtag comm = - do result <- newEmptyMVar - _ <- forkIO $ - do res <- recv recvrank recvtag comm - putMVar result res - return (Request result) - --- | Begin to send an object. Call 'test' or 'wait' to finish the --- communication. -isend :: CanSerialize a - => a -- ^ Object to send - -> Rank -- ^ Destination rank - -> Tag -- ^ Message tag - -> Comm -- ^ Communicator - -> IO (Request ()) -- ^ Communication request -isend sendobj sendrank sendtag comm = - do result <- newEmptyMVar - _ <- forkIO $ do send sendobj sendrank sendtag comm - putMVar result (Status sendrank sendtag, ()) - return (Request result) - --- | Check whether a communication has finished, and return the --- communication result if so. -test :: Request a -- ^ Communication request - -> IO (Maybe (Status, a)) -- ^ 'Just' communication result, if - -- communication has finished, else 'Nothing' -test (Request result) = tryTakeMVar result - --- | Check whether a communication has finished, and return the --- communication result if so, without returning a message status. -test_ :: Request a -- ^ Communication request - -> IO (Maybe a) -- ^ 'Just' communication result, if - -- communication has finished, else 'Nothing' -test_ req = fmap snd <$> test req - --- | Wait for a communication to finish and return the communication --- result. -wait :: Request a -- ^ Communication request - -> IO (Status, a) -- ^ Message status and communication result -wait (Request result) = takeMVar result - --- | Wait for a communication to finish and return the communication --- result, without returning a message status. -wait_ :: Request a -- ^ Communication request - -> IO a -- ^ Communication result -wait_ req = snd <$> wait req - - - --- | Broadcast a message from one process (the "root") to all other --- processes in the communicator. Call this function on all non-root --- processes. Call 'bcastSend' instead on the root process. -bcastRecv :: CanSerialize a - => Rank - -> Comm - -> IO a -bcastRecv root comm = - do rank <- MPI.commRank comm - mpiAssert (rank /= root) "bcastRecv: expected rank /= root" - lenbuf <- mallocForeignPtr @CLong - lenreq <- MPI.ibcast (lenbuf, 1::Int) root comm - whileM_ (not <$> MPI.test_ lenreq) yield - len <- withForeignPtr lenbuf peek - ptr <- mallocBytes (fromIntegral len) - recvbuf <- B.unsafePackMallocCStringLen (ptr, fromIntegral len) - req <- MPI.ibcast recvbuf root comm - whileM_ (not <$> MPI.test_ req) yield - recvobj <- deserialize recvbuf - return recvobj - --- | Broadcast a message from one process (the "root") to all other --- processes in the communicator. Call this function on the root --- process. Call 'bcastRecv' instead on all non-root processes. -bcastSend :: CanSerialize a - => a - -> Rank - -> Comm - -> IO () -bcastSend sendobj root comm = - do rank <- MPI.commRank comm - mpiAssert (rank == root) "bcastSend: expected rank == root" - sendbuf <- serialize sendobj - lenbuf <- mallocForeignPtr @CLong - withForeignPtr lenbuf $ \ptr -> poke ptr (fromIntegral (B.length sendbuf)) - lenreq <- MPI.ibcast (lenbuf, 1::Int) root comm - whileM_ (not <$> MPI.test_ lenreq) yield - req <- MPI.ibcast sendbuf root comm - whileM_ (not <$> MPI.test_ req) yield - -ibcastRecv :: CanSerialize a - => Rank - -> Comm - -> IO (Request a) -ibcastRecv root comm = - do result <- newEmptyMVar - _ <- forkIO $ - do recvobj <- bcastRecv root comm - putMVar result (Status root MPI.anyTag, recvobj) - return (Request result) - -ibcastSend :: CanSerialize a - => a - -> Rank - -> Comm - -> IO (Request ()) -ibcastSend sendobj root comm = - do result <- newEmptyMVar - _ <- forkIO $ - do bcastSend sendobj root comm - putMVar result (Status root MPI.anyTag, ()) - return (Request result) - --- | Begin a barrier. Call 'test' or 'wait' to finish the --- communication. -ibarrier :: Comm - -> IO (Request ()) -ibarrier comm = - do result <- newEmptyMVar - _ <- forkIO $ - do req <- MPI.ibarrier comm - whileM_ (not <$> MPI.test_ req) yield - putMVar result (Status MPI.anySource MPI.anyTag, ()) - return (Request result) diff --git a/lib/Control/Distributed/MPI/Serialize.hs b/lib/Control/Distributed/MPI/Serialize.hs new file mode 100644 --- /dev/null +++ b/lib/Control/Distributed/MPI/Serialize.hs @@ -0,0 +1,399 @@ +{-# LANGUAGE ConstraintKinds #-} +{-# LANGUAGE TypeApplications #-} + +-- | Module: Control.Distributed.MPI.Serialize +-- Description: Simplified MPI bindings with automatic serialization +-- based on Data.Serialize +-- Copyright: (C) 2018 Erik Schnetter +-- License: Apache-2.0 +-- Maintainer: Erik Schnetter +-- Stability: experimental +-- Portability: Requires an externally installed MPI library + +module Control.Distributed.MPI.Serialize + ( -- * Types, and associated functions and constants + MPIException(..) + + -- ** Communicators + , Comm(..) + , commSelf + , commWorld + + -- ** Message sizes + , Count(..) + , fromCount + , toCount + + -- ** Process ranks + , Rank(..) + , anySource + , commRank + , commSize + , fromRank + , rootRank + , toRank + + -- ** Message status + , Status(..) + + -- ** Message tags + , Tag(..) + , anyTag + , fromTag + , toTag + , unitTag + + , Request + + -- * Functions + + -- ** Initialization and shutdown + , abort + , mainMPI + + -- ** Point-to-point (blocking) + , recv + , recv_ + , send + , sendrecv + , sendrecv_ + + -- ** Point-to-point (non-blocking) + , irecv + , isend + , test + , test_ + , wait + , wait_ + + -- ** Collective (blocking) + , barrier + , bcastRecv + , bcastSend + + -- ** Collective (non-blocking) + , ibarrier + , ibcastRecv + , ibcastSend + ) where + +import Prelude hiding (init) + +import Control.Concurrent +import Control.Exception +import Control.Monad +import Control.Monad.Loops +import qualified Data.ByteString as B +import qualified Data.ByteString.Unsafe as B +import qualified Data.Serialize as Serialize +import Data.Typeable +import Foreign +import Foreign.C.Types + +import qualified Control.Distributed.MPI as MPI +import Control.Distributed.MPI + ( Comm(..) + , commSelf + , commWorld + , Count(..) + , fromCount + , toCount + , Rank(..) + , anySource + , commRank + , commSize + , fromRank + , rootRank + , toRank + , Tag(..) + , anyTag + , fromTag + , toTag + , unitTag + , abort + , barrier + ) + + + +-- Serialization, based on Data.Store +type CanSerialize a = Serialize.Serialize a +serialize :: CanSerialize a => a -> IO B.ByteString +serialize = return . Serialize.encode +deserialize :: CanSerialize a => B.ByteString -> IO a +deserialize buf = + do let obj = Serialize.decode buf + case obj of + Left str -> throwIO $ + MPIException ("Data.Serialize.decode failed: " ++ str) + Right x -> return x + + + +-- | Run the supplied Maybe computation repeatedly while it returns +-- Nothing. If it returns a value, then returns that value. +whileNothing :: Monad m => m (Maybe a) -> m () -> m a +whileNothing cond loop = go + where go = do mx <- cond + case mx of + Nothing -> do loop + go + Just x -> return x + + + +-- | Exception type indicating an error in a call to MPI +newtype MPIException = MPIException String + deriving (Eq, Ord, Read, Show, Typeable) +instance Exception MPIException + +mpiAssert :: Bool -> String -> IO () +mpiAssert cond msg = + do when (not cond) $ throw (MPIException msg) + return () + + + +data DidInit = DidInit | DidNotInit + +initMPI :: IO DidInit +initMPI = + do isInit <- MPI.initialized + if isInit + then return DidNotInit + else do ts <- MPI.initThread MPI.ThreadMultiple + mpiAssert (ts >= MPI.ThreadMultiple) + ("MPI.init: Insufficient thread support: requiring " ++ + show MPI.ThreadMultiple ++ + ", but MPI library provided only " ++ show ts) + return DidInit + +finalizeMPI :: DidInit -> IO () +finalizeMPI DidInit = + do isFinalized <- MPI.finalized + if isFinalized + then return () + else do MPI.finalize +finalizeMPI DidNotInit = return () + +-- | Convenience function to initialize and finalize MPI. This +-- initializes MPI with 'ThreadMultiple' thread support. +mainMPI :: IO () -- ^ action to run with MPI, typically the whole program + -> IO () +mainMPI action = bracket initMPI finalizeMPI (\_ -> action) + + + +-- | A communication request, usually created by a non-blocking +-- communication function. +newtype Request a = Request (MVar (Status, a)) + +-- | The status of a finished communication, indicating rank and tag +-- of the other communication end point. +data Status = Status { msgRank :: !Rank + , msgTag :: !Tag + } + deriving (Eq, Ord, Read, Show) + + + +-- | Receive an object. +recv :: CanSerialize a + => Rank -- ^ Source rank + -> Tag -- ^ Source tag + -> Comm -- ^ Communicator + -> IO (Status, a) -- ^ Message status and received object +recv recvrank recvtag comm = + do status <- whileNothing (MPI.iprobe recvrank recvtag comm) yield + source <- MPI.getSource status + tag <- MPI.getTag status + count <- MPI.getCount status MPI.datatypeByte + let len = MPI.fromCount count + ptr <- mallocBytes len + buffer <- B.unsafePackMallocCStringLen (ptr, len) + req <- MPI.irecv buffer source tag comm + whileM_ (not <$> MPI.test_ req) yield + recvobj <- deserialize buffer + return (Status source tag, recvobj) + +-- | Receive an object without returning a status. +recv_ :: CanSerialize a + => Rank -- ^ Source rank + -> Tag -- ^ Source tag + -> Comm -- ^ Communicator + -> IO a -- ^ Received object +recv_ recvrank recvtag comm = + snd <$> recv recvrank recvtag comm + +-- | Send an object. +send :: CanSerialize a + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Message tag + -> Comm -- ^ Communicator + -> IO () +send sendobj sendrank sendtag comm = + do sendbuf <- serialize sendobj + -- Use 'unsafeUseAsCStringLen' to ensure 'sendbuf' is not freed + -- too early + B.unsafeUseAsCStringLen sendbuf $ \_ -> + do req <- MPI.isend sendbuf sendrank sendtag comm + whileM_ (not <$> MPI.test_ req) yield + +-- | Send and receive objects simultaneously. +sendrecv :: (CanSerialize a, CanSerialize b) + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Send message tag + -> Rank -- ^ Source rank + -> Tag -- ^ Receive message tag + -> Comm -- ^ Communicator + -> IO (Status, b) -- ^ Message status and received object +sendrecv sendobj sendrank sendtag recvrank recvtag comm = + do recvreq <- irecv recvrank recvtag comm + send sendobj sendrank sendtag comm + wait recvreq + +-- | Send and receive objects simultaneously, without returning a +-- status for the received message. +sendrecv_ :: (CanSerialize a, CanSerialize b) + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Send message tag + -> Rank -- ^ Source rank + -> Tag -- ^ Receive message tag + -> Comm -- ^ Communicator + -> IO b -- ^ Received object +sendrecv_ sendobj sendrank sendtag recvrank recvtag comm = + snd <$> sendrecv sendobj sendrank sendtag recvrank recvtag comm + +-- | Begin to receive an object. Call `test` or `wait` to finish the +-- communication, and to obtain the received object. +irecv :: CanSerialize a + => Rank -- ^ Source rank + -> Tag -- ^ Source tag + -> Comm -- ^ Communicator + -> IO (Request a) -- ^ Communication request +irecv recvrank recvtag comm = + do result <- newEmptyMVar + _ <- forkIO $ + do res <- recv recvrank recvtag comm + putMVar result res + return (Request result) + +-- | Begin to send an object. Call 'test' or 'wait' to finish the +-- communication. +isend :: CanSerialize a + => a -- ^ Object to send + -> Rank -- ^ Destination rank + -> Tag -- ^ Message tag + -> Comm -- ^ Communicator + -> IO (Request ()) -- ^ Communication request +isend sendobj sendrank sendtag comm = + do result <- newEmptyMVar + _ <- forkIO $ do send sendobj sendrank sendtag comm + putMVar result (Status sendrank sendtag, ()) + return (Request result) + +-- | Check whether a communication has finished, and return the +-- communication result if so. +test :: Request a -- ^ Communication request + -> IO (Maybe (Status, a)) -- ^ 'Just' communication result, if + -- communication has finished, else 'Nothing' +test (Request result) = tryTakeMVar result + +-- | Check whether a communication has finished, and return the +-- communication result if so, without returning a message status. +test_ :: Request a -- ^ Communication request + -> IO (Maybe a) -- ^ 'Just' communication result, if + -- communication has finished, else 'Nothing' +test_ req = fmap snd <$> test req + +-- | Wait for a communication to finish and return the communication +-- result. +wait :: Request a -- ^ Communication request + -> IO (Status, a) -- ^ Message status and communication result +wait (Request result) = takeMVar result + +-- | Wait for a communication to finish and return the communication +-- result, without returning a message status. +wait_ :: Request a -- ^ Communication request + -> IO a -- ^ Communication result +wait_ req = snd <$> wait req + + + +-- | Broadcast a message from one process (the "root") to all other +-- processes in the communicator. Call this function on all non-root +-- processes. Call 'bcastSend' instead on the root process. +bcastRecv :: CanSerialize a + => Rank + -> Comm + -> IO a +bcastRecv root comm = + do rank <- MPI.commRank comm + mpiAssert (rank /= root) "bcastRecv: expected rank /= root" + lenbuf <- mallocForeignPtr @CLong + lenreq <- MPI.ibcast (lenbuf, 1::Int) root comm + whileM_ (not <$> MPI.test_ lenreq) yield + len <- withForeignPtr lenbuf peek + ptr <- mallocBytes (fromIntegral len) + recvbuf <- B.unsafePackMallocCStringLen (ptr, fromIntegral len) + req <- MPI.ibcast recvbuf root comm + whileM_ (not <$> MPI.test_ req) yield + recvobj <- deserialize recvbuf + return recvobj + +-- | Broadcast a message from one process (the "root") to all other +-- processes in the communicator. Call this function on the root +-- process. Call 'bcastRecv' instead on all non-root processes. +bcastSend :: CanSerialize a + => a + -> Rank + -> Comm + -> IO () +bcastSend sendobj root comm = + do rank <- MPI.commRank comm + mpiAssert (rank == root) "bcastSend: expected rank == root" + sendbuf <- serialize sendobj + lenbuf <- mallocForeignPtr @CLong + withForeignPtr lenbuf $ \ptr -> poke ptr (fromIntegral (B.length sendbuf)) + lenreq <- MPI.ibcast (lenbuf, 1::Int) root comm + whileM_ (not <$> MPI.test_ lenreq) yield + req <- MPI.ibcast sendbuf root comm + whileM_ (not <$> MPI.test_ req) yield + +ibcastRecv :: CanSerialize a + => Rank + -> Comm + -> IO (Request a) +ibcastRecv root comm = + do result <- newEmptyMVar + _ <- forkIO $ + do recvobj <- bcastRecv root comm + putMVar result (Status root MPI.anyTag, recvobj) + return (Request result) + +ibcastSend :: CanSerialize a + => a + -> Rank + -> Comm + -> IO (Request ()) +ibcastSend sendobj root comm = + do result <- newEmptyMVar + _ <- forkIO $ + do bcastSend sendobj root comm + putMVar result (Status root MPI.anyTag, ()) + return (Request result) + +-- | Begin a barrier. Call 'test' or 'wait' to finish the +-- communication. +ibarrier :: Comm + -> IO (Request ()) +ibarrier comm = + do result <- newEmptyMVar + _ <- forkIO $ + do req <- MPI.ibarrier comm + whileM_ (not <$> MPI.test_ req) yield + putMVar result (Status MPI.anySource MPI.anyTag, ()) + return (Request result) diff --git a/mpi-hs.cabal b/mpi-hs.cabal --- a/mpi-hs.cabal +++ b/mpi-hs.cabal @@ -1,13 +1,13 @@ cabal-version: 1.12 --- This file has been generated from package.yaml by hpack version 0.31.0. +-- This file has been generated from package.yaml by hpack version 0.31.1. -- -- see: https://github.com/sol/hpack -- --- hash: 78a9591a1cd036f2ecd9afed2b5c9b1a094791c057255536650d00de3174f23c +-- hash: 51db71059fcdd89fa4df2f6b20866b5abb0c7a517f8fe02a9d9687bb1139aa86 name: mpi-hs -version: 0.4.1.0 +version: 0.5.1.1 synopsis: MPI bindings for Haskell description: MPI (the [Message Passing Interface](https://www.mpi-forum.org)) is widely used standard for distributed-memory programming on HPC (High @@ -59,7 +59,8 @@ library exposed-modules: Control.Distributed.MPI - Control.Distributed.MPI.Packing + Control.Distributed.MPI.Binary + Control.Distributed.MPI.Serialize Control.Distributed.MPI.Store other-modules: Paths_mpi_hs @@ -76,8 +77,8 @@ base >=4 && <5 , binary , bytestring + , cereal , monad-loops - , packman , store build-tools: c2hs @@ -92,6 +93,7 @@ ghc-options: -Wall -rtsopts -threaded -with-rtsopts=-N build-depends: base + , binary , mpi-hs default-language: Haskell2010 @@ -109,13 +111,26 @@ , mpi-hs default-language: Haskell2010 -test-suite mpi-test-packing +test-suite mpi-test-binary type: exitcode-stdio-1.0 main-is: Main.hs other-modules: Paths_mpi_hs hs-source-dirs: - tests/packing + tests/binary + ghc-options: -Wall -rtsopts -threaded -with-rtsopts=-N + build-depends: + base + , mpi-hs + default-language: Haskell2010 + +test-suite mpi-test-serialize + type: exitcode-stdio-1.0 + main-is: Main.hs + other-modules: + Paths_mpi_hs + hs-source-dirs: + tests/serialize ghc-options: -Wall -rtsopts -threaded -with-rtsopts=-N build-depends: base diff --git a/package.yaml b/package.yaml --- a/package.yaml +++ b/package.yaml @@ -1,5 +1,5 @@ name: mpi-hs -version: '0.4.1.0' +version: '0.5.1.1' github: "eschnett/mpi-hs" license: Apache-2.0 author: "Erik Schnetter " @@ -51,8 +51,8 @@ - base >=4 && <5 # tested with 4.11 and 4.12 - binary - bytestring + - cereal - monad-loops - - packman - store build-tools: - c2hs @@ -70,6 +70,7 @@ main: Main.hs dependencies: - base + - binary - mpi-hs ghc-options: - -rtsopts @@ -105,8 +106,22 @@ - -rtsopts - -threaded - -with-rtsopts=-N - mpi-test-packing: - source-dirs: tests/packing + mpi-test-binary: + source-dirs: tests/binary + main: Main.hs + dependencies: + - base + - mpi-hs + # - tasty + # - tasty-hunit + # - tasty-hspec + # - unix + ghc-options: + - -rtsopts + - -threaded + - -with-rtsopts=-N + mpi-test-serialize: + source-dirs: tests/serialize main: Main.hs dependencies: - base diff --git a/stack.yaml b/stack.yaml --- a/stack.yaml +++ b/stack.yaml @@ -1,12 +1,12 @@ # Resolver to choose a 'specific' stackage snapshot or a compiler version. -resolver: lts-12.16 +resolver: lts-13.4 # User packages to be built. packages: - . -extra-deps: - - packman-0.5.0 +# extra-deps: +# - packman-0.5.0 # Extra directories used by stack for building extra-include-dirs: diff --git a/tests/binary/Main.hs b/tests/binary/Main.hs new file mode 100644 --- /dev/null +++ b/tests/binary/Main.hs @@ -0,0 +1,166 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +import System.IO +import System.Exit + +import qualified Control.Distributed.MPI.Binary as MPI + +default (Int) + + + +-------------------------------------------------------------------------------- + +infix 1 @? +(@?) :: Bool -> String -> IO () +x @? msg = if not x then die msg else return () + +infix 1 @?= +(@?=) :: Eq a => a -> a -> IO () +x @?= y = x == y @? "test failed" + + + +type TestTree = IO () + +testCase :: String -> IO () -> TestTree +testCase name test = + do rank <- MPI.commRank MPI.commWorld + if rank == 0 + then do putStrLn $ " " ++ name ++ "..." + hFlush stdout + else return () + MPI.barrier MPI.commWorld + test + MPI.barrier MPI.commWorld + + + +testGroup :: String -> [TestTree] -> TestTree +testGroup name cases = + do rank <- MPI.commRank MPI.commWorld + if rank == 0 + then do putStrLn $ name ++ ":" + hFlush stdout + else return () + sequence_ cases + + + +defaultMain :: TestTree -> IO () +defaultMain tree = + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + if rank == 0 + then do putStrLn $ "MPI Tests: running on " ++ show size ++ " processes" + hFlush stdout + else return () + tree + + + +-------------------------------------------------------------------------------- + + + +main :: IO () +main = MPI.mainMPI $ defaultMain tests + +tests :: TestTree +tests = testGroup "MPI" + [ rankSize + , pointToPoint + , pointToPointNonBlocking + , collective + , collectiveNonBlocking + ] + + + +rankSize :: TestTree +rankSize = testGroup "rank and size" + [ testCase "commSelf" $ + do rank <- MPI.commRank MPI.commSelf + size <- MPI.commSize MPI.commSelf + rank == 0 && size == 1 @? "" + , testCase "commWorld" $ + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + rank >= 0 && rank < size @? "" + ] + + + +pointToPoint :: TestTree +pointToPoint = testGroup "point-to-point" + [ testCase "sendrecv" $ + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + let sendmsg :: String = "Hello, World!" + let sendrank = (rank + 1) `mod` size + let recvrank = (rank - 1) `mod` size + (status, recvmsg :: String) <- + MPI.sendrecv sendmsg sendrank MPI.unitTag recvrank MPI.unitTag + MPI.commWorld + (recvmsg == sendmsg && + MPI.msgRank status == recvrank && + MPI.msgTag status == MPI.unitTag) @? "" + ] + + + +pointToPointNonBlocking :: TestTree +pointToPointNonBlocking = testGroup "point-to-point non-blocking" + [ testCase "send and recv" $ + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + let sendmsg :: String = "Hello, World!" + let sendrank = (rank + 1) `mod` size + sendreq <- MPI.isend sendmsg sendrank MPI.unitTag MPI.commWorld + let recvrank = (rank - 1) `mod` size + recvreq <- MPI.irecv recvrank MPI.unitTag MPI.commWorld + (sendstatus, ()) <- MPI.wait sendreq + (recvstatus, recvmsg :: String) <- MPI.wait recvreq + (recvmsg == sendmsg && + MPI.msgRank sendstatus == sendrank && + MPI.msgTag sendstatus == MPI.unitTag && + MPI.msgRank recvstatus == recvrank && + MPI.msgTag recvstatus == MPI.unitTag) @? "" + ] + + + +collective :: TestTree +collective = testGroup "collective" + [ testCase "barrier" $ + do MPI.barrier MPI.commWorld + , testCase "bcast" $ + do rank <- MPI.commRank MPI.commWorld + let sendmsg :: String = "Hello, World!" + recvmsg :: String <- + if rank == MPI.rootRank + then do MPI.bcastSend sendmsg MPI.rootRank MPI.commWorld + return sendmsg + else do MPI.bcastRecv MPI.rootRank MPI.commWorld + recvmsg == sendmsg @? "" + ] + + + +collectiveNonBlocking :: TestTree +collectiveNonBlocking = testGroup "collective non-blocking" + [ testCase "barrier" $ + do req <- MPI.ibarrier MPI.commWorld + MPI.wait_ req + , testCase "bcast" $ + do rank <- MPI.commRank MPI.commWorld + let sendmsg :: String = "Hello, World!" + recvmsg :: String <- + if rank == MPI.rootRank + then do req <- MPI.ibcastSend sendmsg MPI.rootRank MPI.commWorld + MPI.wait_ req + return sendmsg + else do req <- MPI.ibcastRecv MPI.rootRank MPI.commWorld + MPI.wait_ req + recvmsg == sendmsg @? "" + ] diff --git a/tests/packing/Main.hs b/tests/packing/Main.hs deleted file mode 100644 --- a/tests/packing/Main.hs +++ /dev/null @@ -1,166 +0,0 @@ -{-# LANGUAGE ScopedTypeVariables #-} - -import System.IO -import System.Exit - -import qualified Control.Distributed.MPI.Packing as MPI - -default (Int) - - - --------------------------------------------------------------------------------- - -infix 1 @? -(@?) :: Bool -> String -> IO () -x @? msg = if not x then die msg else return () - -infix 1 @?= -(@?=) :: Eq a => a -> a -> IO () -x @?= y = x == y @? "test failed" - - - -type TestTree = IO () - -testCase :: String -> IO () -> TestTree -testCase name test = - do rank <- MPI.commRank MPI.commWorld - if rank == 0 - then do putStrLn $ " " ++ name ++ "..." - hFlush stdout - else return () - MPI.barrier MPI.commWorld - test - MPI.barrier MPI.commWorld - - - -testGroup :: String -> [TestTree] -> TestTree -testGroup name cases = - do rank <- MPI.commRank MPI.commWorld - if rank == 0 - then do putStrLn $ name ++ ":" - hFlush stdout - else return () - sequence_ cases - - - -defaultMain :: TestTree -> IO () -defaultMain tree = - do rank <- MPI.commRank MPI.commWorld - size <- MPI.commSize MPI.commWorld - if rank == 0 - then do putStrLn $ "MPI Tests: running on " ++ show size ++ " processes" - hFlush stdout - else return () - tree - - - --------------------------------------------------------------------------------- - - - -main :: IO () -main = MPI.mainMPI $ defaultMain tests - -tests :: TestTree -tests = testGroup "MPI" - [ rankSize - , pointToPoint - , pointToPointNonBlocking - , collective - , collectiveNonBlocking - ] - - - -rankSize :: TestTree -rankSize = testGroup "rank and size" - [ testCase "commSelf" $ - do rank <- MPI.commRank MPI.commSelf - size <- MPI.commSize MPI.commSelf - rank == 0 && size == 1 @? "" - , testCase "commWorld" $ - do rank <- MPI.commRank MPI.commWorld - size <- MPI.commSize MPI.commWorld - rank >= 0 && rank < size @? "" - ] - - - -pointToPoint :: TestTree -pointToPoint = testGroup "point-to-point" - [ testCase "sendrecv" $ - do rank <- MPI.commRank MPI.commWorld - size <- MPI.commSize MPI.commWorld - let sendmsg :: String = "Hello, World!" - let sendrank = (rank + 1) `mod` size - let recvrank = (rank - 1) `mod` size - (status, recvmsg :: String) <- - MPI.sendrecv sendmsg sendrank MPI.unitTag recvrank MPI.unitTag - MPI.commWorld - (recvmsg == sendmsg && - MPI.msgRank status == recvrank && - MPI.msgTag status == MPI.unitTag) @? "" - ] - - - -pointToPointNonBlocking :: TestTree -pointToPointNonBlocking = testGroup "point-to-point non-blocking" - [ testCase "send and recv" $ - do rank <- MPI.commRank MPI.commWorld - size <- MPI.commSize MPI.commWorld - let sendmsg :: String = "Hello, World!" - let sendrank = (rank + 1) `mod` size - sendreq <- MPI.isend sendmsg sendrank MPI.unitTag MPI.commWorld - let recvrank = (rank - 1) `mod` size - recvreq <- MPI.irecv recvrank MPI.unitTag MPI.commWorld - (sendstatus, ()) <- MPI.wait sendreq - (recvstatus, recvmsg :: String) <- MPI.wait recvreq - (recvmsg == sendmsg && - MPI.msgRank sendstatus == sendrank && - MPI.msgTag sendstatus == MPI.unitTag && - MPI.msgRank recvstatus == recvrank && - MPI.msgTag recvstatus == MPI.unitTag) @? "" - ] - - - -collective :: TestTree -collective = testGroup "collective" - [ testCase "barrier" $ - do MPI.barrier MPI.commWorld - , testCase "bcast" $ - do rank <- MPI.commRank MPI.commWorld - let sendmsg :: String = "Hello, World!" - recvmsg :: String <- - if rank == MPI.rootRank - then do MPI.bcastSend sendmsg MPI.rootRank MPI.commWorld - return sendmsg - else do MPI.bcastRecv MPI.rootRank MPI.commWorld - recvmsg == sendmsg @? "" - ] - - - -collectiveNonBlocking :: TestTree -collectiveNonBlocking = testGroup "collective non-blocking" - [ testCase "barrier" $ - do req <- MPI.ibarrier MPI.commWorld - MPI.wait_ req - , testCase "bcast" $ - do rank <- MPI.commRank MPI.commWorld - let sendmsg :: String = "Hello, World!" - recvmsg :: String <- - if rank == MPI.rootRank - then do req <- MPI.ibcastSend sendmsg MPI.rootRank MPI.commWorld - MPI.wait_ req - return sendmsg - else do req <- MPI.ibcastRecv MPI.rootRank MPI.commWorld - MPI.wait_ req - recvmsg == sendmsg @? "" - ] diff --git a/tests/serialize/Main.hs b/tests/serialize/Main.hs new file mode 100644 --- /dev/null +++ b/tests/serialize/Main.hs @@ -0,0 +1,166 @@ +{-# LANGUAGE ScopedTypeVariables #-} + +import System.IO +import System.Exit + +import qualified Control.Distributed.MPI.Serialize as MPI + +default (Int) + + + +-------------------------------------------------------------------------------- + +infix 1 @? +(@?) :: Bool -> String -> IO () +x @? msg = if not x then die msg else return () + +infix 1 @?= +(@?=) :: Eq a => a -> a -> IO () +x @?= y = x == y @? "test failed" + + + +type TestTree = IO () + +testCase :: String -> IO () -> TestTree +testCase name test = + do rank <- MPI.commRank MPI.commWorld + if rank == 0 + then do putStrLn $ " " ++ name ++ "..." + hFlush stdout + else return () + MPI.barrier MPI.commWorld + test + MPI.barrier MPI.commWorld + + + +testGroup :: String -> [TestTree] -> TestTree +testGroup name cases = + do rank <- MPI.commRank MPI.commWorld + if rank == 0 + then do putStrLn $ name ++ ":" + hFlush stdout + else return () + sequence_ cases + + + +defaultMain :: TestTree -> IO () +defaultMain tree = + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + if rank == 0 + then do putStrLn $ "MPI Tests: running on " ++ show size ++ " processes" + hFlush stdout + else return () + tree + + + +-------------------------------------------------------------------------------- + + + +main :: IO () +main = MPI.mainMPI $ defaultMain tests + +tests :: TestTree +tests = testGroup "MPI" + [ rankSize + , pointToPoint + , pointToPointNonBlocking + , collective + , collectiveNonBlocking + ] + + + +rankSize :: TestTree +rankSize = testGroup "rank and size" + [ testCase "commSelf" $ + do rank <- MPI.commRank MPI.commSelf + size <- MPI.commSize MPI.commSelf + rank == 0 && size == 1 @? "" + , testCase "commWorld" $ + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + rank >= 0 && rank < size @? "" + ] + + + +pointToPoint :: TestTree +pointToPoint = testGroup "point-to-point" + [ testCase "sendrecv" $ + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + let sendmsg :: String = "Hello, World!" + let sendrank = (rank + 1) `mod` size + let recvrank = (rank - 1) `mod` size + (status, recvmsg :: String) <- + MPI.sendrecv sendmsg sendrank MPI.unitTag recvrank MPI.unitTag + MPI.commWorld + (recvmsg == sendmsg && + MPI.msgRank status == recvrank && + MPI.msgTag status == MPI.unitTag) @? "" + ] + + + +pointToPointNonBlocking :: TestTree +pointToPointNonBlocking = testGroup "point-to-point non-blocking" + [ testCase "send and recv" $ + do rank <- MPI.commRank MPI.commWorld + size <- MPI.commSize MPI.commWorld + let sendmsg :: String = "Hello, World!" + let sendrank = (rank + 1) `mod` size + sendreq <- MPI.isend sendmsg sendrank MPI.unitTag MPI.commWorld + let recvrank = (rank - 1) `mod` size + recvreq <- MPI.irecv recvrank MPI.unitTag MPI.commWorld + (sendstatus, ()) <- MPI.wait sendreq + (recvstatus, recvmsg :: String) <- MPI.wait recvreq + (recvmsg == sendmsg && + MPI.msgRank sendstatus == sendrank && + MPI.msgTag sendstatus == MPI.unitTag && + MPI.msgRank recvstatus == recvrank && + MPI.msgTag recvstatus == MPI.unitTag) @? "" + ] + + + +collective :: TestTree +collective = testGroup "collective" + [ testCase "barrier" $ + do MPI.barrier MPI.commWorld + , testCase "bcast" $ + do rank <- MPI.commRank MPI.commWorld + let sendmsg :: String = "Hello, World!" + recvmsg :: String <- + if rank == MPI.rootRank + then do MPI.bcastSend sendmsg MPI.rootRank MPI.commWorld + return sendmsg + else do MPI.bcastRecv MPI.rootRank MPI.commWorld + recvmsg == sendmsg @? "" + ] + + + +collectiveNonBlocking :: TestTree +collectiveNonBlocking = testGroup "collective non-blocking" + [ testCase "barrier" $ + do req <- MPI.ibarrier MPI.commWorld + MPI.wait_ req + , testCase "bcast" $ + do rank <- MPI.commRank MPI.commWorld + let sendmsg :: String = "Hello, World!" + recvmsg :: String <- + if rank == MPI.rootRank + then do req <- MPI.ibcastSend sendmsg MPI.rootRank MPI.commWorld + MPI.wait_ req + return sendmsg + else do req <- MPI.ibcastRecv MPI.rootRank MPI.commWorld + MPI.wait_ req + recvmsg == sendmsg @? "" + ]