hspread (empty) → 0.2
raw patch · 10 files changed
+828/−0 lines, 10 filesdep +basedep +binarydep +bytestringsetup-changed
Dependencies added: base, binary, bytestring, containers, network
Files
- Control/Concurrent/Chan/Closeable.hs +164/−0
- LICENSE +27/−0
- README +33/−0
- Setup.hs +3/−0
- Spread/Client.hs +8/−0
- Spread/Client/Connection.hs +148/−0
- Spread/Client/Message.hs +249/−0
- Spread/Constants.hs +55/−0
- examples/hspflooder.hs +107/−0
- hspread.cabal +34/−0
+ Control/Concurrent/Chan/Closeable.hs view
@@ -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+
+ LICENSE view
@@ -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.
+ README view
@@ -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
+ Setup.hs view
@@ -0,0 +1,3 @@+module Main(main) where+import Distribution.Simple+main = defaultMain
+ Spread/Client.hs view
@@ -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)+ +
+ Spread/Client/Connection.hs view
@@ -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 ()++
+ Spread/Client/Message.hs view
@@ -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'
+ Spread/Constants.hs view
@@ -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
+ examples/hspflooder.hs view
@@ -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 <options>" + ,"\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" + ]
+ hspread.cabal view
@@ -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