diff --git a/Control/Concurrent/Chan/Closeable.hs b/Control/Concurrent/Chan/Closeable.hs new file mode 100644 --- /dev/null +++ b/Control/Concurrent/Chan/Closeable.hs @@ -0,0 +1,164 @@ +{-# LANGUAGE EmptyDataDecls #-} +----------------------------------------------------------------------------- +-- | +-- Module : Control.Concurrent.Chan.Closeable +-- Copyright : Andrea Vezzosi (based on Control.Concurrent.Chan (c) The University of Glasgow 2001) +-- License : BSD-style +-- +-- Maintainer : sanzhiyan@gmail.com +-- Stability : experimental +-- Portability : non-portable (concurrency) +-- +-- Unbounded closeable channels. +-- +----------------------------------------------------------------------------- + +module Control.Concurrent.Chan.Closeable + ( + -- * The 'Chan' type + Chan,R,W, -- abstract + + -- * Operations + newChan, -- :: IO (Chan R a, Chan W a) + writeChan, -- :: Chan W a -> a -> IO Bool + closeChan, -- :: Chan W a -> IO Bool + isClosedChan, -- :: Chan t a -> IO Bool + readChan, -- :: Chan R a -> IO (Maybe a) + forkChan, -- :: Chan t a -> IO (Chan R a) + unGetChan, -- :: Chan R a -> a -> IO () + isEmptyChan, -- :: Chan R a -> IO Bool + + -- * Stream interface + getChanContents, -- :: Chan R a -> IO [a] + writeList2Chan, -- :: Chan W a -> [a] -> IO (Maybe [a]) + + ) + where + +import System.IO.Unsafe ( unsafeInterleaveIO ) +import Control.Concurrent.MVar + + + +-- A channel is represented by two @MVar@s keeping track of the two ends +-- of the channel contents,i.e., the read- and write ends. Empty @MVar@s +-- are used to handle consumers trying to read from an empty channel. + +-- |'Chan' is an abstract type representing an unbounded FIFO channel. +newtype Chan t a = Chan (MVar (Stream a)) +-- |'R' for ReadOnly +data R +-- |'W' for WriteOnly +data W + + +type Stream a = (MVar (ChItem a)) + +data ChItem a = ChItem a !(Stream a) | Closed + + + +-- @newChan@ sets up the read and write end of a channel by initialising +-- these two @MVar@s with an empty @MVar@. + +-- |Build and returns a pair of 'Chan', data written on the W end can be read from the R end. +newChan :: IO (Chan R a,Chan W a) +newChan = do + hole <- newEmptyMVar + read <- newMVar hole + write <- newMVar hole + return (Chan read, Chan write) + +-- To put an element on a channel, a new hole at the write end is created. +-- What was previously the empty @MVar@ at the back of the channel is then +-- filled in with a new stream element holding the entered value and the +-- new hole. + +-- |Write a value to a 'Chan'. +-- Returns True if successful, False if the channel is closed. +writeChan :: Chan W a -> a -> IO Bool +writeChan (Chan write) val = do + new_hole <- newEmptyMVar + modifyMVar write $ \old_hole -> do + wasempty <- tryPutMVar old_hole (ChItem val new_hole) + if wasempty then return (new_hole,True) + else return (old_hole,False) + +-- |Close the 'Chan', data can be no more written to it. +-- Returns True if the 'Chan' was already closed. +closeChan :: Chan W a -> IO Bool +closeChan (Chan write) = + withMVar write $ \s -> + not `fmap` tryPutMVar s Closed + +-- |Non-blocking check. +isClosedChan :: Chan t a -> IO Bool +isClosedChan (Chan var) = + withMVar var $ \s -> do + e <- isEmptyMVar s + -- readMVar won't block because a stream is never emptied + -- i.e. we never use *takeMVar on it + if e then return False else isClosed `fmap` readMVar s + where isClosed Closed = True + isClosed _ = False +-- |Read the next value from the 'Chan'. +-- |Nothing if the 'Chan' is closed. +readChan :: Chan R a -> IO (Maybe a) +readChan (Chan read) = do + modifyMVar read $ \read_end -> do + cItem <- readMVar read_end + -- Use readMVar here, not takeMVar, + -- else dupChan doesn't work + return $ case cItem of + (ChItem val new_read_end) -> (new_read_end, Just val) + Closed -> (read_end,Nothing) + +-- |Forks a 'Chan': data that will be written (W) +-- or is yet to be read (R) on the argument, will also be available on the returned channel. +forkChan :: Chan t a -> IO (Chan R a) +forkChan (Chan write) = do + hole <- readMVar write + new_read <- newMVar hole + return (Chan new_read) + + + +-- |Put a data item back onto a channel, where it will be the next item read. +unGetChan :: Chan R a -> a -> IO () +unGetChan (Chan read) val = do + new_read_end <- newEmptyMVar + modifyMVar_ read $ \read_end -> do + putMVar new_read_end (ChItem val read_end) + return (new_read_end) + +-- |Returns 'True' if the supplied 'Chan' is empty, i.e. readChan won't block. +isEmptyChan :: Chan R a -> IO Bool +isEmptyChan (Chan read) = do + withMVar read $ \r -> do + isEmptyMVar r + +-- Operators for interfacing with functional streams. + +-- |Return a lazy list representing the contents of the supplied +-- 'Chan', much like 'System.IO.hGetContents'. +getChanContents :: Chan R a -> IO [a] +getChanContents ch + = unsafeInterleaveIO (do + xm <- readChan ch + case xm of + Nothing -> return [] + Just x -> do xs <- getChanContents ch + return (x:xs) + ) + +-- |Write an entire list of items to a 'Chan'. +-- Returning the remainder if the channel has been closed meanwhile. +writeList2Chan :: Chan W a -> [a] -> IO (Maybe [a]) +writeList2Chan ch ls = aux ls + where write = writeChan ch + aux [] = return Nothing + aux l@(x:xs) = do b <- write x + if b + then aux xs + else return $ Just l + diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -0,0 +1,27 @@ +Copyright : (c) Andrea Vezzosi + +All rights reserved. + +Redistribution and use in source and binary forms, with or without +modification, are permitted provided that the following conditions +are met: +1. Redistributions of source code must retain the above copyright + notice, this list of conditions and the following disclaimer. +2. Redistributions in binary form must reproduce the above copyright + notice, this list of conditions and the following disclaimer in the + documentation and/or other materials provided with the distribution. +3. Neither the name of the author nor the names of his contributors + may be used to endorse or promote products derived from this software + without specific prior written permission. + +THIS SOFTWARE IS PROVIDED BY THE REGENTS 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 AUTHORS 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/README b/README new file mode 100644 --- /dev/null +++ b/README @@ -0,0 +1,33 @@ + hspread : a client library for spread + + hspread is a client library for the Spread toolkit[1]. + It is fully implemented in haskell and supports the most recent version + of the protocol. Its aim is to make easier to implement correct distribuited + applications by taking advantage of the guarantees granted by Spread: + such as reliable and total ordered messages. It's intended to be used + with a serialization library like binary[2], and a separate installation + of the spread deamon. + +Installing: + + It installs like every other cabal package: + runhaskell Setup configure -- this can fail with missing dependecies + runhaskell Setup build + runhaskell Setup install -- you may need superuser privileges here + + Haddock documentation can be generated with: + runhaskell Setup haddock + + You can then test it with the spflooder clone under examples/, + not forgetting to have a spread deamon running. + +------------------------------------------------------------------------------- + +The development version is kept in a darcs repository here: + + darcs get http://happs.org/repos/hspread + +------------------------------------------------------------------------------- + +[1] http://spread.org +[2] http://code.haskell.org/binary diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,3 @@ +module Main(main) where +import Distribution.Simple +main = defaultMain diff --git a/Spread/Client.hs b/Spread/Client.hs new file mode 100644 --- /dev/null +++ b/Spread/Client.hs @@ -0,0 +1,8 @@ +module Spread.Client + (module Spread.Client.Message + ,module Spread.Client.Connection) + where +import Spread.Client.Connection +import Spread.Client.Message hiding (putPadded,multicast_internal,receive_internal,KillMsg) + + diff --git a/Spread/Client/Connection.hs b/Spread/Client/Connection.hs new file mode 100644 --- /dev/null +++ b/Spread/Client/Connection.hs @@ -0,0 +1,148 @@ +module Spread.Client.Connection + ( + Connection,privateGroup + ,Conf(..),defaultConf + ,AuthName + ,mkAuthName,authname + ,AuthMethod + ,connect,disconnect,startReceive,stopReceive,getDupedChan + ,join,leave,send + ) where +import Network +import qualified Data.ByteString.Char8 as B +import qualified Data.ByteString as Bs +import Data.ByteString (ByteString) +import qualified Data.ByteString.Lazy as L +import Foreign +import System.IO.Error +import Control.Concurrent hiding (Chan) +import Data.Bits +import System.IO +import Data.Binary.Put +import Spread.Client.Message +import Network +import Control.Applicative +import Control.Exception +import Control.Monad hiding (join) +import Control.Concurrent.Chan.Closeable as C +import Data.Maybe +import Spread.Constants +-- | Abstract type representing a connection with a spread server. +data Connection = C {privateGroup :: !PrivateGroup -- ^ private name of this connection, useful for p2p messages. + , chan :: !(Chan W Message), getHandle :: !Handle, thread :: !ThreadId, sync :: !(MVar ()) } +-- | Configuration passed to 'connect' +data Conf = Conf { address :: !(Maybe HostName) -- ^ Server address, using localhost if 'Nothing'. + , port :: !(Maybe PortNumber) -- ^ Server port, uses the default spread port if 'Nothing'. + , desiredName :: !PrivateName -- ^ It will become part of the 'PrivateGroup' of the 'Connection' + , priority :: !Bool -- ^ Is this a priority connection? + , groupMembership :: !Bool -- ^ Should it receive Membership messages? + , authMethods :: ![AuthMethod] -- ^ Authentication methods to use when connecting. + } +-- | defaulConf = Conf Nothing Nothing (mkPrivateName (B.pack \"user\")) False True [] +defaultConf :: Conf +defaultConf = Conf Nothing Nothing (mkPrivateName (B.pack "user")) False True [] +-- | Name of an authentication method. +newtype AuthName = AN ByteString deriving Eq +nullAuthMethod :: (AuthName, t -> IO Bool) +nullAuthMethod = (mkAuthName (B.pack "NULL"), \_ -> return True :: IO Bool) +-- | The 'ByteString' will be truncated to the maximum allowed size. +mkAuthName :: ByteString -> AuthName +mkAuthName = AN . B.take mAX_AUTH_NAME +authname :: AuthName -> ByteString +authname (AN s) = s +-- | The action should return True if the authentication succeded. +type AuthMethod = (AuthName,(Handle -> IO Bool)) + +-- | Connects to the specified server, will use a \"NULL\" authentication method if the 'authMethods' list is empty. +-- A spread server will refuse the connection if another with the same PrivateName is still active. +connect :: Conf -> IO (Chan R Message,Connection) +connect c = + let addr = fromMaybe "localhost" . address $ c + port' = fromMaybe dEFAULT_SPREAD_PORT . port $ c + (authnames,authmethods) = unzip $ let l = authMethods c in if null l then [nullAuthMethod] else l + in bracketOnError (withSocketsDo $ connectTo addr (PortNumber port')) hClose $ \h -> do + let hget = Bs.hGet h + writePut (mkConnectMsg (desiredName c) (priority c) (groupMembership c)) h + authlistlen <- hGetByte h + checkLen authlistlen + authlist <- map AN . B.split ' ' <$> hget authlistlen + checkAuthNames authnames authlist + writePut (putAuthNames authnames) h + results <- mapM ($h) authmethods + checkAuths results + checkAccepted =<< hGetByte h + [major,minor,patch,glen] <- map fromIntegral . Bs.unpack <$> hget 4 + checkVersion major minor patch + mprvg <- mkPrivateGroup <$> hget glen + makeConnection mprvg h + where checkLen l = when ( l > (mAX_AUTH_NAME * mAX_AUTH_METHODS)) $ fail $ "connect: illegal value in authlistlen " ++ show l + checkAuthNames nms list = unless (and $ map (`elem` list) nms) $ fail "connect: chosen authentication method is not permitted by daemon" + checkAuths r = unless (and r) $ fail "connect: authentication of connection failed" + checkAccepted v = unless (v == aCCEPT_SESSION) $ fail "session rejected" + checkVersion maj min p = let val = maj*10000+min*100+p + in do when (val < 30100) $ fail "old spread version, not supported" + when (val < 30800 && priority c) $ fail "old spread version, priority not supported." + makeConnection prvg h = do (r,w) <- C.newChan + m <- newEmptyMVar + tid <- forkIO $ receiver w m h prvg + return $ (r,C prvg w h tid m) + +receiver :: Chan W Message -> MVar () -> Handle -> PrivateGroup -> IO () +receiver c m h prvg = let + recv = receive_internal h prvg + putc = C.writeChan c + p = readMVar m + loop = p >> recv >>= putc >> loop + in handle (\_ -> return ()) $ loop `finally` (hClose h >> C.closeChan c) + +hGetByte :: Handle -> IO Int +hGetByte h = alloca $ \p -> + do n <- hGetBuf h p 1; + if n /= 1 + then ioError $ mkIOError eofErrorType "hGetByte" (Just h) Nothing + else fromIntegral <$> (peek p :: IO Word8) + + +putAuthNames :: [AuthName] -> Put +putAuthNames xs = mapM_ (\(AN s) -> putPadded mAX_AUTH_NAME s) . take mAX_AUTH_METHODS $ (xs ++ repeat (AN Bs.empty)) + +mkConnectMsg :: PrivateName -> Bool -> Bool -> Put +mkConnectMsg p priority gmemb = do + mapM_ putWord8 [sP_MAJOR_VERSION,sP_MINOR_VERSION,sP_PATCH_VERSION] + let setIf b n = if b then (.|. n) else id + putWord8 (setIf gmemb 0x01 . setIf priority 0x10 $ 0) + let pn = privateName p + putWord8 . fromIntegral . B.length $ pn + putByteString pn + +writePut p h = L.hPut h (runPut p) >> hFlush h +-- | Sends a disconnection message to the server, which will close the connection. +disconnect :: Connection -> IO () +disconnect c = let p = privateGroup c in sendInternal (Kill p) c + +-- | Messages received from now on will be available on the returned 'Chan' +getDupedChan :: Connection -> IO (Chan R Message) +getDupedChan c = C.forkChan (chan c) + +-- | Start fetching messages from the network, returns True if it was stopped. +startReceive :: Connection -> IO Bool +startReceive c = tryPutMVar (sync c) () + +-- | Stop fetching messages from the network (at most one more message can be read) +-- , returns True if it was started. +stopReceive :: Connection -> IO Bool +stopReceive c = maybe False (const True) <$> tryTakeMVar (sync c) +-- | Joins a group, the server will send a 'Reg'. +join :: Group -> Connection -> IO () +join g = sendInternal (Joining g) +-- | Leaves a group, the server will send a 'SelfLeave'. +leave :: Group -> Connection -> IO () +leave g = sendInternal (Leaving g) + +-- | Send a regular message. +send :: OutMsg -> Connection -> IO () +send = sendInternal + +sendInternal m c = let p = privateGroup c in multicast_internal p m (getHandle c) >> return () + + diff --git a/Spread/Client/Message.hs b/Spread/Client/Message.hs new file mode 100644 --- /dev/null +++ b/Spread/Client/Message.hs @@ -0,0 +1,249 @@ +module Spread.Client.Message + ( + OutMsg(..),InMsg(..),Message(..),Group,PrivateGroup,PrivateName,GroupId,OrderingType(..),Cause(..),groupName,mkPrivateGroup,privateName,GroupMsg(..), + MembershipMsg(..),receive_internal,multicast_internal,mkGroup,mkPrivateName,putPadded,makeGroup,privateGroupName,KillMsg(..),RejectedMsg(..)) + where +import Data.ByteString (ByteString) +import qualified Data.ByteString as B +import qualified Data.ByteString.Lazy as L +import qualified Data.ByteString.Char8 as Ch +import Data.Bits +import Data.Word +import Data.Binary.Get +import Data.Binary.Put +import Data.Map (fromList,findWithDefault) +import Control.Monad +import Data.Maybe +import Data.List (elemIndex,find) +import System.IO (hFlush) +import Spread.Constants + +-- | Represents the orderings as specified by the Spread toolkit. +data OrderingType = Unreliable | Reliable | Fifo | Causal | Agreed | Safe deriving (Eq,Ord,Read,Show) +orderingTable = fromList orderingList +orderingList = zip [Unreliable,Reliable,Fifo,Causal,Agreed,Safe] + [uNRELIABLE_MESS,rELIABLE_MESS,fIFO_MESS,cAUSAL_MESS,aGREED_MESS,sAFE_MESS] +getOrdering t = fst . fromJust . find (flip isSet t . snd) $ orderingList +-- | Message to be sent. +data OutMsg = Outgoing { outOrdering :: !OrderingType + , outDiscard :: !Bool -- ^ If True you won't get a copy of this message back from the server. + , outData :: !ByteString -- ^ Message body. + , outGroups :: ![Group] -- ^ Recipients of the message + , outMsgType :: !Word16 -- ^ To be used by the application to identify the kind of message. + } + deriving (Show) +-- | Message received. +data InMsg = Incoming { inOrdering :: !OrderingType + , inSender :: !PrivateGroup + , inData :: !ByteString + , inGroups :: ![Group] + , inMsgType :: !Word16 + , inEndianMismatch :: !Bool -- ^ True if the message has been sent with a different endian order. + } deriving (Show) +-- | Union Type of messages that can be received from the server. +data Message = Regular !InMsg + | Membership !MembershipMsg + | Rejected !RejectedMsg deriving (Show) + +instance Sendable OutMsg where + getType m = outMsgType m + getData m = outData m + getGroups m = outGroups m + getServiceType m = (if outDiscard m then (sELF_DISCARD .|.) else id ) $ (findWithDefault 0 (outOrdering m) orderingTable) +instance Sendable GroupMsg where + getGroups m = [grp m] + getServiceType (Leaving _) = lEAVE_MESS + getServiceType (Joining _) = jOIN_MESS +instance Sendable KillMsg where + getGroups (Kill g) = [fromPrivateGroup g] + getServiceType (Kill _) = kILL_MESS + +--TODO Rejected Message + +-- | Message regarding changes in group membership. +data MembershipMsg = Transient { changingGroup :: !Group } + | Reg { changingGroup :: !Group, index :: !Int, numMembers :: !Int, members :: ![PrivateGroup], groupId :: !GroupId, cause :: !Cause } + | SelfLeave { changingGroup :: !Group } deriving (Show) + +-- | Messages used to join or leave a group. +data GroupMsg = Joining {grp :: !Group} | Leaving {grp :: !Group} deriving Show +data KillMsg = Kill PrivateGroup +-- | A 'Group' is a collection of clients identified by a name. +newtype Group = G {groupName :: ByteString } deriving (Eq,Show) +mkGroup :: B.ByteString -> Maybe Group +mkGroup s = if B.all (\b -> (b > 36) && (b < 126)) s then Just (G (B.take mAX_GROUP_NAME s)) else Nothing +makeGroup = mkGroup . Ch.pack + +-- | A 'PrivateGroup' identifies a connection. +newtype PrivateGroup = PrivateGroup {privateGroupName :: ByteString} deriving (Eq,Show) +mkPrivateGroup = PrivateGroup . B.take mAX_GROUP_NAME +toPrivateGroup = PrivateGroup . groupName + +-- | Initial part of a 'PrivateGroup' name that is chosen by the client when connecting. +newtype PrivateName = PrivateName {privateName :: ByteString} deriving (Eq,Show) +mkPrivateName = PrivateName . B.take mAX_PRIVATE_NAME +-- | Identifier for a membership message. +data GroupId = GId !Word32 !Word32 !Word32 deriving (Eq,Show) +-- | What caused a membership message. +data Cause = Join {joining :: !PrivateGroup} + | Leave {leaving :: !PrivateGroup} + | Disconnect { disconnecting :: !PrivateGroup} + | Network {sets :: ![[PrivateGroup]], localSet :: ![PrivateGroup] } deriving (Show) + + +sameEndian i = (i .&. eNDIAN_TYPE) == aRCH_ENDIAN +clearEndian i = i .&. (complement eNDIAN_TYPE) +setEndian i = (i .&. complement eNDIAN_TYPE ) .|. aRCH_ENDIAN +flip32 :: Word32 -> Word32 +flip32 i = ((i `shiftR` 24) .&. 0x000000ff) .|. ((i `shiftR` 8) .&. 0x0000ff00) .|. ((i `shiftL` 8) .&. 0x00ff0000) .|. ((i `shiftL` 24) .&. 0xff000000) + +data Raw = Raw { serviceType :: !Word32, isender :: !Group, igroups :: ![Group], + itype :: !Word16,daemonEndianMismatch :: !Bool, iendianMismatch :: !Bool, body :: !ByteString } deriving Show + +parseRaw :: Raw -> Message +parseRaw i@Raw{serviceType=t} | isSet rEJECT_MESS t = Rejected $ asRejected i + | isSet mEMBERSHIP_MESS t = Membership $ asMembership i + | otherwise = Regular $ asRegular i + +data RejectedMsg = WasGroup !GroupMsg | WasOut !OutMsg deriving Show +asGroupMsg i@Raw{serviceType=t,igroups=[g]} | isSet lEAVE_MESS t = Leaving g + | isSet jOIN_MESS t = Joining g +asOutMsg i = Outgoing {outOrdering = getOrdering $ serviceType i, + outDiscard = isSet sELF_DISCARD $ serviceType i, + outData = body i, + outGroups = igroups i, + outMsgType = itype i} + +asRejected :: Raw -> RejectedMsg +asRejected i@Raw{serviceType=t} | isSet rEGULAR_MESS t = WasOut $ asOutMsg i + | otherwise = WasGroup $ asGroupMsg i + +asRegular :: Raw -> InMsg +asRegular i = Incoming { inOrdering = getOrdering $ serviceType i, + inSender = toPrivateGroup (isender i), + inData = body i, + inGroups = igroups i, + inMsgType = itype i, + inEndianMismatch = iendianMismatch i + } +--receive_internal :: GHC.IOBase.Handle -> Group -> IO Message +receive_internal h prvg = liftM parseRaw . Control.Monad.join $ runGetNH (mAX_GROUP_NAME + 16) h getInternal + where getInternal :: Get (IO Raw) + getInternal = do srvT <- getInt + let (dEM,maybeFlip) = if sameEndian srvT then (False,id) else (True, flip32)-- deamonEndianMismatch + senderbs <- getGroup + [ng,hint,dl] <- replicateM 3 getInt + let [serviceType',numGroups,dataLen] = map maybeFlip [srvT,ng,dl] + eM = not (sameEndian hint) + hint' = if eM then flip32 hint else hint + type' = fromIntegral $ (clearEndian hint' `shiftR` 8) .&. 0x0000FFFF + getOldType :: Get Word32 + getOldType = do oldt <- maybeFlip `fmap` getInt + return $ rEJECT_MESS .|. oldt + serviceType <- fmap clearEndian $ if isSet rEJECT_MESS serviceType' + then getOldType + else return serviceType' + return $ do groups <- runGetNH (fromIntegral numGroups * mAX_GROUP_NAME) h (readGroups (fromIntegral numGroups)) + body <- B.hGet h (fromIntegral dataLen ) + return Raw { serviceType = serviceType, + isender = senderbs, + igroups = groups, + itype = if isSet mEMBERSHIP_MESS serviceType && isSet rEG_MEMB_MESS serviceType + then fromIntegral . fromJust . elemIndex (fromPrivateGroup prvg) $ groups + else type', + daemonEndianMismatch = dEM, + body = body, + iendianMismatch = eM + } + +asMembership i@Raw{serviceType=t} | isSet tRANSITION_MESS t = Transient (isender i) + | isSet rEG_MEMB_MESS t = Reg { changingGroup = isender i, + index = fromIntegral $ itype i, + members = map toPrivateGroup (igroups i), + numMembers = Prelude.length (igroups i), + groupId = gid, + cause = cause + } + | isSet cAUSED_BY_LEAVE t && + not (isSet rEG_MEMB_MESS t) = SelfLeave { changingGroup = isender i } + | otherwise = error "asMembership: unexpected message type" + where (gids,rest) = B.splitAt 12 (body i) + gid = flip runGet (L.fromChunks [gids]) $ + ((\[i1,i2,i3] -> GId i1 i2 i3)) `fmap` replicateM 3 getInt' + getInt' = (if daemonEndianMismatch i then flip32 else id) `fmap` getInt + getSet = fmap (map toPrivateGroup) . readGroups . fromIntegral =<< getInt' -- a set is n followed by n Groups + cause = flip runGet (L.fromChunks [rest]) $ do + numSets <- fmap fromIntegral getInt' + byteOffsetToLocalSet <- fmap fromIntegral getInt' + firstSet <- bytesRead + let localSetIndex = firstSet + byteOffsetToLocalSet + pairs <- replicateM numSets $ do + mark <- bytesRead + grps <- getSet + return (mark,grps) + let first = head . head $ sets + sets = map snd pairs + lSet = fromJust . lookup localSetIndex $ pairs + return $ case () of + _ | isSet cAUSED_BY_JOIN t -> Join first + | isSet cAUSED_BY_LEAVE t -> Leave first + | isSet cAUSED_BY_DISCONNECT t -> Disconnect first + | isSet cAUSED_BY_NETWORK t -> Network sets lSet + +readGroups :: Int -> Get [Group] +readGroups n = replicateM n $ getGroup + +getGroup :: Get Group +getGroup = (G . fst . B.spanEnd (==0)) `fmap` getBytes mAX_GROUP_NAME + + + + + +-- lifts (Get a) to (IO a) reading n bytes from the handle. +--runGetNH :: Int -> GHC.IOBase.Handle -> Get a -> IO a +runGetNH n h m = runGet m `fmap` L.hGet h n + +getInt :: Get Word32 +getInt = getWord32host +putInt :: Word32 -> Put +putInt = putWord32host +isSet :: (Bits a) => a -> a -> Bool +isSet f t = t .&. f /= 0 + +class Sendable a where + getGroups :: a -> [Group] + --getGroups _ = [] + getData :: a -> ByteString + getData _ = B.empty + getServiceType :: a -> Word32 + getType :: a -> Word16 + getType _ = 0 + +putGroup :: Group -> PutM () +putGroup (G b) = putPadded mAX_GROUP_NAME . B.take mAX_GROUP_NAME $ b + +putPadded :: Int -> ByteString -> PutM () +putPadded n s = let len = B.length s in putByteString s >> replicateM_ (n - len) (putWord8 0) + +-- multicast_internal :: (Sendable a) => PrivateGroup -> a -> Handle -> IO Bool +multicast_internal prvg s h = maybe (return False) ((>> (hFlush h >> return True)) . L.hPut h) . sendable prvg $ s + +fromPrivateGroup :: PrivateGroup -> Group +fromPrivateGroup (PrivateGroup g) = G g + +sendable :: (Sendable a) => PrivateGroup -> a -> Maybe L.ByteString +sendable prvg m = if numBytes > mAX_MESSAGE_LENGTH + then Nothing + else Just . runPut $ do + putInt (setEndian $ getServiceType m) + putGroup (fromPrivateGroup prvg) + putInt (fromIntegral $ numGroups) + putInt (setEndian $ fromIntegral (getType m) `shiftL` 8 .&. 0x00FFFF00) + putInt (fromIntegral $ B.length data') + mapM_ putGroup groups + putByteString data' + where groups = getGroups m + numGroups = (length groups) + data' = getData m + numBytes = 16 + mAX_GROUP_NAME * (numGroups + 1) + B.length data' diff --git a/Spread/Constants.hs b/Spread/Constants.hs new file mode 100644 --- /dev/null +++ b/Spread/Constants.hs @@ -0,0 +1,55 @@ +module Spread.Constants where +import Data.Word +import Data.Binary +import Data.Binary.Put +import Data.Binary.Get +import Network +uNRELIABLE_MESS = 0x00000001 ::Word32 +rELIABLE_MESS = 0x00000002::Word32 +fIFO_MESS = 0x00000004::Word32 +cAUSAL_MESS = 0x00000080::Word32 +aGREED_MESS = 0x00000010::Word32 +sAFE_MESS = 0x00000020::Word32 +rEGULAR_MESS = 0x0000003f::Word32 +sELF_DISCARD = 0x00000040::Word32 +rEG_MEMB_MESS = 0x00001000::Word32 +tRANSITION_MESS = 0x00002000::Word32 +cAUSED_BY_JOIN = 0x00000100::Word32 +cAUSED_BY_LEAVE = 0x00000200::Word32 +cAUSED_BY_DISCONNECT = 0x00000400::Word32 +cAUSED_BY_NETWORK = 0x00000800::Word32 +mEMBERSHIP_MESS = 0x00003f00::Word32 +rEJECT_MESS = 0x00400000 ::Word32 + +--detect the endianness of the architecture we're running on +aRCH_ENDIAN = if (runGet getWord16host (runPut $ putWord16be (0x0001::Word16))) == 0x0001 then + (0x00000000::Word32) + else + (0x80000080 ::Word32) +--aRCH_ENDIAN = 0x80000080 + + +eNDIAN_TYPE = 0x80000080 :: Word32 +---- Service-types used only within the package. +jOIN_MESS = 0x00010000 ::Word32 +lEAVE_MESS = 0x00020000::Word32 +kILL_MESS = 0x00040000::Word32 +gROUPS_MESS = 0x00080000::Word32 + +-- Lengths +mAX_MESSAGE_LENGTH = 140000 :: Int +mAX_GROUP_NAME = 32 :: Int +mAX_PRIVATE_NAME = 10 :: Int +mAX_AUTH_NAME :: Int +mAX_AUTH_NAME = 30 +mAX_AUTH_METHODS :: Int +mAX_AUTH_METHODS = 3 + +dEFAULT_SPREAD_PORT :: PortNumber +dEFAULT_SPREAD_PORT = 4803 + +sP_MAJOR_VERSION = 4 :: Word8 +sP_MINOR_VERSION = 0 :: Word8 +sP_PATCH_VERSION = 0 :: Word8 + +aCCEPT_SESSION = 1 :: Int diff --git a/examples/hspflooder.hs b/examples/hspflooder.hs new file mode 100644 --- /dev/null +++ b/examples/hspflooder.hs @@ -0,0 +1,107 @@ + +import qualified Spread.Client as SC +import qualified Data.ByteString.Char8 as B +import qualified Control.Concurrent.Chan.Closeable as C +import System.Environment (getArgs) +import Control.Monad +import Data.Maybe +import Text.Regex.Base +import Text.Regex.Posix +import Network +import Data.List + +makeConf p n u = (SC.Conf n p (SC.mkPrivateName $ B.pack u) True True []) + +argVal o args = + case args of + [] -> Nothing + a:v:tl -> if o == a then Just v else argVal o (v:tl) + _:tl -> argVal o tl + +argExists o args = any (\i -> i == o) args + +withReplyGroup f verbose g (m,c) = do + f g c + msg <- C.readChan m + when verbose (putStrLn $ show msg) + +withReply f verbose (m,c) = do + f c + msg <- C.readChan m + when verbose (putStrLn $ show msg) + +waitForMsg verbose (m,c) = do + msg <- C.readChan m + when verbose (putStrLn $ show msg) + +sendRecvNoDiscard verbose repjl g (m,c) msg wo ro = do + when repjl (withReplyGroup SC.join verbose g (m,c)) + when wo (SC.send msg c) + unless wo (do + --putStrLn "not writeOnly" + --when ro (putStrLn "waitForMsg") + when ro (waitForMsg verbose (m,c)) + --unless ro (putStrLn "withReply") + unless ro (withReply (SC.send msg) verbose (m,c))) + when repjl (withReplyGroup SC.leave verbose g (m,c)) + +parseSpreadName s = + case s of + Nothing -> (Nothing,Nothing) + Just a -> + let p = Just (fromIntegral (read (a =~ "^[0-9]+" :: String) :: Int) :: PortNumber) + h = Just (drop 1 (a =~ "@[a-zA-Z0-9\\.]+" :: String)) + in + (p,h) + + + + +main = do + args <- getArgs + if (argExists "-h" args) || (argExists "--help" args) || (argExists "-help" args) + then putStrLn usage + else do + let (port,host) = parseSpreadName $ argVal "-s" args; + conf = makeConf port host $ fromMaybe "user" (argVal "-u" args); + reps = read $ fromMaybe "10000" (argVal "-m" args); + gname = fromMaybe "flooder" (argVal "-g" args); + mbytes = read $ fromMaybe "1000" (argVal "-b" args); + sendRecvMode = fromMaybe "sendRecvNoDiscard" (argVal "-s" args); + checkMsg = argExists "-check" args; + verbose = argExists "-v" args; + repjl = argExists "-repjl" args; + writeOnly = argExists "-wo" args; + readOnly = argExists "-ro" args; + msgData = B.pack (replicate mbytes 't'); + g = (fromJust (SC.makeGroup gname)); + msg = SC.Outgoing SC.Reliable writeOnly msgData [g] 1; + f = case sendRecvMode of + "sendRecvNoDiscard" -> sendRecvNoDiscard + _ -> sendRecvNoDiscard + (m,c) <- SC.connect conf + putStrLn ("Starting test loop with count " ++ (show reps)) + SC.startReceive c + when (not repjl) $ withReplyGroup SC.join verbose g (m,c) + replicateM_ reps $ f verbose repjl g (m,c) msg writeOnly readOnly + when (not repjl) $ withReplyGroup SC.leave verbose g (m,c) + putStrLn "Finished test loop" + SC.stopReceive c + putStrLn "stopped receiving" + SC.disconnect c + putStrLn "Disconnected" + +usage = concat . intersperse "\n" $ + ["hspflooder usage:" + ,"hspflooder " + ,"\t[-s spreadname]\t\tport or port@machine, if unspecified, defaults to \"4803@localhost\"" + ,"\t[-u username]\t\tif unspecified, defaults to \"user\"" + ,"\t[-m msgcount]\t\tnumber of messages to send -> if unspecified, defaults to 10000" + ,"\t[-g groupname]\t\tname of group to send msgs to -> if unspecified, defaults to \"flooder\"" + ,"\t[-b msgbytes]\t\tbyte size of messages to send -> if unspecified, defaults to \"1000\"" + ,"\t[-v]\t\tverbose output" + ,"\t[-repjl]\t\tjoin before every message,leave after each message" + ,"\t[-wo]\t\twrite non-group messages with selfDiscard == true, " + ,"\t[-ro]\t\tonly receive messages, ignored if -wo is specified" + ,"\t[-check]\t\t(Currently ignored) check input match output, ignored if -wo or -ro is specified" + ] diff --git a/hspread.cabal b/hspread.cabal new file mode 100644 --- /dev/null +++ b/hspread.cabal @@ -0,0 +1,34 @@ +name: hspread +version: 0.2 +license: BSD3 +license-file: LICENSE +author: Andrea Vezzosi +maintainer: Andrea Vezzosi, sanzhiyan@gmail.com +synopsis: A client library for the spread toolkit +description: + hspread is a client library for the Spread toolkit. + It is fully implemented in haskell and supports the most recent version + of the protocol. It's intended to be used with a serialization library + like binary, and a separate installation of the spread deamon. +build-type: Simple +cabal-version: >= 1.2 +category: Network +extra-source-files: README, LICENSE, examples/hspflooder.hs + +flag small_base + description: Choose the new smaller, split-up base package. + +library + exposed-modules: Control.Concurrent.Chan.Closeable + Spread.Client + other-modules: + Spread.Client.Message + Spread.Client.Connection + Spread.Constants + + if flag(small_base) + build-depends: base >= 3, containers, bytestring + else + build-depends: base < 3 + build-depends: network, binary >= 0.3 + ghc-options: -O2