packages feed

ipc (empty) → 0.0.2

raw patch · 5 files changed

+244/−0 lines, 5 filesdep +basedep +binarydep +bytestringsetup-changed

Dependencies added: base, binary, bytestring, dlist, mtl, network, network-bytestring, stm

Files

+ Data/Queue.hs view
@@ -0,0 +1,65 @@+module Data.Queue+    ( Queue+    , Data.Queue.empty+    , Input+    , Output+    , isEmpty+    , snoc+    , pop+    , cons+    , send+    , recv+    , sendAtomic+    , recvAtomic) where++import qualified Data.DList as D+import Control.Concurrent.STM++-- |A buffer data type with the direction ('Input' or 'Output') and element type.+data Queue dir a = Q (D.DList a) Int++data Input+data Output++empty :: Queue d a+empty = Q D.empty 0++-- |Tests if the element count of a queue is zero.+isEmpty :: Queue d a -> Bool+isEmpty (Q _ 0) = True+isEmpty _ = False++-- |Places an element at the end of the queue. O(1)+snoc :: Queue d a -> a -> Queue d a+snoc (Q d n) e = Q (D.snoc d e) (n + 1)++-- |Extracts an element from the front of the queue. O(1)+pop :: Queue d a -> (Maybe a, Queue d a)+pop (Q d 0) = (Nothing, Q d 0)+pop (Q d n) = (Just (D.head d), Q (D.tail d) (n - 1))++-- |Places an element at the front of the queue. O(1)+cons :: a -> Queue d a -> Queue d a+cons e (Q d n) = Q (D.cons e d) (n + 1)++-- |IO action for snoc on a TVar queue+send :: TVar (Queue Output a) -> a -> IO ()+send tv a = atomically ( sendAtomic tv a )++-- |STM action to snoc on a TVar queue+sendAtomic :: TVar (Queue Output a) -> a -> STM ()+sendAtomic tv a = do+    q <- readTVar tv+    writeTVar tv (snoc q a)++-- |IO action for popping an element from a TVar queue+recv :: TVar (Queue Input a) -> IO a+recv tv = atomically (recvAtomic tv)++-- |STM action for popping an element from a TVar queue+recvAtomic :: TVar (Queue Input a) -> STM a+recvAtomic tv = do+    q <- readTVar tv+    case pop q of+        (Nothing, _) -> retry+        (Just e, qs) -> writeTVar tv qs >> return e
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) Thomas DuBuisson++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 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.
+ Setup.lhs view
@@ -0,0 +1,3 @@+#!/usr/bin/env runhaskell+> import Distribution.Simple+> main = defaultMain
+ System/IPC.hs view
@@ -0,0 +1,130 @@+-- |The Inter-Process Communication (IPC) library makes the task of setting+-- up datagram sockets between processes trivial.  The simplest method is:+--+-- > outputQueue <- channelConnectSimple "channelName"+--+-- > send outputQueue someInstanceOfBinary+--+--  and+--  +-- > inputQueue <- channelAcceptSimple "channelName"+--+-- > val <- recv inputQueue+--+-- The Queues are TVars from Data.Queue, which can be imported for fine-tuned+-- control.+--+-- If you find this useful then e-mail me and I'll be more likely to maintain+-- it and add features.  Potential new features include:+--+--     * Remove the static maximum message size ('maxBytes') limitation.+--     +--     * Fix connection capability (ex: if the sender closes, receiver rebinds and continues)+--+--     * Testing on other platforms+--+--     * Better exception handling+--+--     * Input buffer size limitations (for memory concious apps)++module System.IPC+    ( channelAccept+    , channelConnect+    , channelAcceptSimple+    , channelConnectSimple+    , D.send+    , D.recv+    , maxBytes+    , waitTillEmpty+    ) where++import qualified Data.Queue as D+import Network.Socket hiding (recv, send)+import Network.Socket.ByteString+import Data.Binary+import qualified Data.ByteString as S+import qualified Data.ByteString.Lazy as L+import Control.Concurrent.STM+import Control.Concurrent (forkIO, threadDelay)+import Control.Monad (forever)+import Control.Exception as X++-- |The maximum bytes per datagram.  Any data structure with an encoding greater than+-- this size stands to be truncated (and probably non-decodable).+maxBytes = 4096++-- |A wrapper for 'channelAccept'.  Creates a new queue and sets conversion == return.+channelAcceptSimple :: (Binary i) => String -> IO (TVar (D.Queue D.Input i))+channelAcceptSimple tag = do+    q <- newTVarIO D.empty+    forkIO $ channelAccept tag return q+    return q++-- |A wrapper for 'channelConnect'.  Creates a new queue and sets conversion == return.+channelConnectSimple :: (Binary o) => String -> IO (TVar (D.Queue D.Output o))+channelConnectSimple tag = withSocketsDo $ do+    q <- newTVarIO D.empty+    forkIO $ channelConnect tag return q+    return q++-- |Sets up an input channel to receive data.+-- Uses the provided queue.  Converts all data from the deserialized format before enqueuing.+channelAccept :: (Binary a) => String -> (a -> IO input) -> TVar (D.Queue D.Input input) -> IO ()+channelAccept tag convIn q = withSocketsDo $ do+    sock <- socket AF_UNIX Datagram defaultProtocol+    bindSocket sock (SockAddrUnix ('\0' : tag))+    receiver sock convIn q++-- |Connects to an open channel to send data.+-- Uses the provided queue.  Converts all data from the queue formate before serializing / sending.+channelConnect :: (Binary b) => String -> (output -> IO b) -> TVar (D.Queue D.Output output) -> IO ()+channelConnect tag convOut q = withSocketsDo $ do+    sock <- socket AF_UNIX Datagram defaultProtocol+    makeConnection sock 0+    sender sock convOut q+  where+  makeConnection s t = X.catchJust ioErrors (doConnect s)+                               (\_ -> threadDelay t >> makeConnection s (newDelay t))+  doConnect sock = connect sock (SockAddrUnix ('\0' : tag))+  newDelay 0 = 1000+  newDelay x = min (x * 2) (10^6)++-- |Receives data from the socket, deserializes, converts, enqueues.+receiver :: (Binary b) => +            Socket ->                     -- ^ input socket+            (b -> IO a) ->                -- ^ conversion function+            TVar (D.Queue D.Input a) ->   -- ^ buffer+            IO ()+receiver sock conv q = forever $ do+    bs  <- recv sock maxBytes+    val <- conv (decode (toLazy bs))+    atomically ( readTVar q >>= writeTVar q . (flip D.snoc) val)++-- |Removes data from the buffer, converts, serializes and sends.+sender :: (Binary b) =>+          Socket ->                      -- ^ output socket+          (a -> IO b) ->                 -- ^ conversion function+          TVar (D.Queue D.Output a) ->   -- ^ buffer+          IO ()+sender sock conv tv = forever $ do+    v <- atomically ( do+       q <- readTVar tv+       case D.pop q of+           (Just e,rest) -> writeTVar tv rest >> return e+           (Nothing,  _) -> retry )+    val <- conv v+    send sock (fromLazy (encode val))++-- |Will block until the queue is empty.  This is useful for delaying program+-- shutdown till _most_ of the data has been sent.  Items are removed just+-- before sending, so an immediate shutdown could fail to send the final item.+waitTillEmpty :: TVar (D.Queue d a) -> IO ()+waitTillEmpty tv = atomically (do+    q <- readTVar tv+    if D.isEmpty q then return () else retry )++toLazy :: S.ByteString -> L.ByteString+toLazy bs = L.fromChunks [bs]++fromLazy :: L.ByteString -> S.ByteString+fromLazy bs = S.concat $ L.toChunks bs
+ ipc.cabal view
@@ -0,0 +1,16 @@+name:                ipc+version:             0.0.2+synopsis:            High level inter-process communication library+description:         Provides inter-process communication at a high level+Copyright:           2008, Thomas DuBuisson+category:            System+license:             BSD3+license-file:        LICENSE+author:              Thomas DuBuisson+maintainer:          Thomas.DuBuisson@gmail.com+build-type:          Simple+build-Depends:       base, network, dlist, network-bytestring, binary, stm, mtl, bytestring+exposed-modules:     System.IPC, Data.Queue+stability:           alpha+tested-with:         GHC == 6.8.2+extensions:          EmptyDataDecls