epoll 0.1.1 → 0.2

raw patch · 12 files changed

+616/−269 lines, 12 filesPVP ok

version bump matches the API change (PVP)

API changes (from Hackage documentation)

Files

• README +19/−1
• epoll.cabal +10/−6
• examples/Server.hs +46/−35
• examples/Stdin.hs +14/−10
• examples/StdinM.hs +0/−28
• src/BChan.hs +83/−0
• src/System/Linux/Epoll.hs +8/−14
• src/System/Linux/Epoll/Base.hsc +32/−20
• src/System/Linux/Epoll/Buffer.hs +236/−0
• src/System/Linux/Epoll/EventLoop.hs +159/−0
• src/System/Linux/EpollM.hs +0/−154
• src/Util.hs +9/−1

@@ -1,7 +1,25 @@-This is a low-level Haskell FFI interface to epoll, a Linux specific I/O+This is a Haskell FFI interface to epoll, a Linux specific I/O notification facility. For usage examples please have a look into the  examples directory (please compile with -threaded option).   This library has been developed and tested with GHC 6.10.4 and  Linux 2.6.28-15 (x86).++Changelog+---------++epoll-0.2++Added EventLoop and Buffer layer which abstract from the low level bindings+contained in Base.+++epoll-0.1.1++Changed category+++epoll-0.1++Initial version. Contains the basic epoll bindings. 

@@ -1,6 +1,6 @@ name:               epoll-version:            0.1.1-synopsis:           Low-level bindings to epoll.+version:            0.2+synopsis:           epoll bindings description:        Bindings to epoll, a Linux specific I/O                     event notication facility (cf. man epoll(7)). category:           System@@ -17,11 +17,15 @@  library   exposed-modules:  System.Linux.Epoll,-                    System.Linux.EpollM,-                    System.Linux.Epoll.Base-  other-modules:    Util+                    System.Linux.Epoll.Base,+                    System.Linux.Epoll.Buffer,+                    System.Linux.Epoll.EventLoop+  other-modules:    Util,+                    BChan   ghc-options:      -Wall -O2-  extensions:       CPP, ForeignFunctionInterface, GeneralizedNewtypeDeriving+  extensions:       CPP,+                    ForeignFunctionInterface,+                    TypeSynonymInstances   build-depends:    base >= 3 && < 5, unix, mtl   hs-source-dirs:   src 

@@ -1,71 +1,71 @@ module Main where -import System.Linux.EpollM-import System.Posix.IO+import System.Linux.Epoll import System.Posix.Types import System.Environment import System.IO import Data.Maybe import Text.Printf+import Control.Concurrent import Control.Monad-import Control.Monad.Trans import System.Posix.Signals import Network.Socket+import Network.HTTP.Base+import Network.HTTP.Headers+import System.Time  main :: IO () main = getAndVerifyArgs >>= start . head  start :: String -> IO ()-start p = runEpollBig_ $ do--    defaultDispatchLoop_ readRequest+start p = do+    epoll <- createRuntime (fromJust . toSize $ 8192) -    ai <- liftIO $ getAddrInfo (Just (defaultHints {addrFlags = [AI_PASSIVE]}))-                               Nothing -                               (Just p)+    ai <- getAddrInfo (Just (defaultHints {addrFlags = [AI_PASSIVE]}))+                      Nothing +                      (Just p)      let serveraddr = head ai      -- create server socket-    sock <- liftIO $ socket (addrFamily serveraddr) Stream defaultProtocol-    liftIO $ setSocketOption sock ReuseAddr 1+    sock <- socket (addrFamily serveraddr) Stream defaultProtocol+    setSocketOption sock ReuseAddr 1      -- bind it to the address we're listening to-    liftIO $ bindSocket sock (addrAddress serveraddr)+    bindSocket sock (addrAddress serveraddr) -    liftIO $ printf "Listening at port %s\n" p+    printf "Listening at port %s\n" p      -- start listening for connection requests max. connection requests waiting     -- to be accepted = 5 (max. queue size)-    liftIO $ listen sock 256+    listen sock 256      -- Ignore broken pipes-    liftIO $ installHandler sigPIPE Ignore Nothing+    installHandler sigPIPE Ignore Nothing      -- accept connection requests (CTRL-C to abort)     forever $ do-        (connsock, clientaddr) <- liftIO $ accept sock-        procRequest connsock clientaddr+        (connsock, clientaddr) <- accept sock+        procRequest epoll connsock clientaddr++    shutdownRuntime epoll  where-    procRequest :: Socket -> SockAddr -> Epoll ()-    procRequest sock' _ = do+    procRequest :: Runtime -> Socket -> SockAddr -> IO ThreadId+    procRequest r sock' _ = forkIO $ do         let fd = Fd $ fdSocket sock'-        liftIO $ setFdOption fd NonBlockingRead True-        add sock' [inEvent, peerCloseEvent, oneShotEvent] fd-        return ()--readRequest :: Event Socket -> Epoll ()-readRequest e = fork_ $ do-    let et = eventType e-        fd = eventFd e-    liftIO $ printf "event=%s, fd=%s\n" (show et) (show fd)-    if et =~ peerCloseEvent-        then do liftIO $ putStrLn "peer closed"-                delete (eventDesc e)-                liftIO $ sClose (eventRef e)-        else do (s, _) <- liftIO $ fdRead fd 2048-                modify [inEvent, peerCloseEvent, oneShotEvent] (eventDesc e)-                liftIO $ printf "readRequest: %s\n" s+        withIBuffer r fd $ \ib -> do+            header <- readBuffer ib >>=+                return . parseRequestHead . takeWhile ("" /=) . breakLines+            case header of+                Left  e -> print $ "caught error: " ++ (show e)+                Right _ -> do+                    now <- getClockTime+                    let ts = show . calendarTimeToString . toUTCTime $ now+                        rs = response ts+                    withOBuffer r fd $ \ob -> do+                        writeBuffer ob (show rs)+                        writeBuffer ob (rspBody rs)+        sClose sock'  getAndVerifyArgs :: IO [String] getAndVerifyArgs = do@@ -74,4 +74,15 @@          (error "Usage: Server <port>")     return args +breakLines :: String -> [String]+breakLines str = pre : case rest of+    '\r':'\n':suf -> breakLines suf+    '\r':suf      -> breakLines suf+    '\n':suf      -> breakLines suf+    _             -> []+ where+    (pre, rest) = break (flip elem "\n\r") str +response :: String -> Response String+response s = Response (2, 0, 0) "OK"+    [Header HdrContentLength (show . length $ s)] s

@@ -5,24 +5,28 @@ import System.IO import Data.Maybe import Text.Printf+import Control.Concurrent.MVar -eventTypes :: [EventType]-eventTypes = [inEvent, edgeTriggeredEvent, oneShotEvent]+eventTypes :: EventType+eventTypes = combineEvents [inEvent, edgeTriggeredEvent, oneShotEvent] -callback :: Device -> Event a -> IO ()-callback d e = do+callback :: MVar () -> Device -> Event Data -> IO ()+callback lck d e = do     let et = eventType e         fd = eventFd e-    setFdOption fd NonBlockingRead True     printf "event=%s, fd=%s\n" (show et) (show fd)-    (s, c) <- fdRead fd 16+    (s, c) <- fdRead fd 16 `catch` \er -> print er >> return ("", 0)     printf "read %s byte(s): %s\n" (show c) s-    modify d eventTypes (eventDesc e)+    if s == "quit\n"+        then putMVar lck ()+        else reEnableCallback d (eventRef e) (eventDesc e)  main :: IO () main = do     let s = fromJust $ toSize 32-    dev <- create s-    add dev () eventTypes stdInput-    defaultDispatchLoop (callback dev) dev+    lock <- newEmptyMVar+    loop <- createEventLoop s+    addCallback loop stdInput [(eventTypes, callback lock)]+    putStrLn "Enter 'quit' to exit."+    takeMVar lock 

@@ -1,28 +0,0 @@-module Main where--import Text.Printf-import Control.Monad.Reader-import System.IO-import System.Posix.IO-import System.Linux.EpollM--eventTypes :: [EventType]-eventTypes = [inEvent, edgeTriggeredEvent, oneShotEvent]--callback :: Event a -> Epoll ()-callback e = do-    let fd = eventFd e-        et = eventType e-        ds = eventDesc e-    liftIO $ do-        setFdOption fd NonBlockingRead True-        printf "event=%s, fd=%s\n" (show et) (show fd)-        (s, c) <- fdRead fd 16-        printf "read %s byte(s): %s\n" (show c) s-    modify eventTypes ds--main :: IO ()-main = runEpollSmall_ $ do-    add_ eventTypes stdInput-    defaultDispatchLoop callback-

@@ -0,0 +1,83 @@+-- |+-- Module      : BChan+-- Copyright   : (c) 2009 Toralf Wittner+-- License     : LGPL+-- Maintainer  : toralf.wittner@gmail.com+-- Stability   : internal+-- Portability : portable+--+-- Internal helper module. Similar to 'Chan' but with support for blocking waits+-- until a buffer is empty.++module BChan where++import Data.IORef+import Control.Monad+import Control.Concurrent.MVar+import Control.Concurrent.Chan+import System.IO.Unsafe(unsafeInterleaveIO)++data BChan a = BChan {+        channel :: Chan a,+        counter :: IORef Int,+        rdrLock :: MVar (),+        waiting :: MVar [MVar ()]+    }++newBChan :: IO (BChan a)+newBChan = do+    c <- newChan+    n <- newIORef 0+    m <- newMVar ()+    w <- newMVar []+    return $ BChan c n m w++readBChan :: BChan a -> IO a+readBChan ch = withMVar (rdrLock ch) $ \_ -> do+    x <- readChan (channel ch)+    j <- atomicModifyIORef (counter ch) $ \i -> (i - 1, i - 1)+    when (j == 0) $+         modifyMVar_ (waiting ch) $ unblock . reverse+    return x++writeBChan :: BChan a -> a -> IO ()+writeBChan ch x = do+    writeChan (channel ch) x+    atomicModifyIORef (counter ch) $ \i -> (i + 1, i)+    return ()++peekBChan :: BChan a -> IO a+peekBChan ch = withMVar (rdrLock ch) $ \_ -> do+    x <- readChan (channel ch)+    unGetChan (channel ch) x+    return x++unGetBChan :: BChan a -> a -> IO ()+unGetBChan ch x = do+    unGetChan (channel ch) x+    atomicModifyIORef (counter ch) $ \i -> (i + 1, i)+    return ()++getBChanContents :: BChan a -> IO [a]+getBChanContents ch = unsafeInterleaveIO $ do+    x  <- readBChan ch+    xs <- getBChanContents ch+    return (x:xs)++waitBChan :: BChan a -> IO ()+waitBChan ch = do+    n <- readIORef (counter ch)+    unless (n == 0) $ do+        lock <- newEmptyMVar+        modifyMVar_ (waiting ch) $ return . (lock:)+        takeMVar lock++unblock :: [MVar ()] -> IO [MVar ()]+unblock = foldr (\m -> (>>) (putMVar m ())) (return [])++isEmptyBChan :: BChan a -> IO Bool+isEmptyBChan = isEmptyChan . channel++dropBChan :: BChan a -> IO ()+dropBChan ch = readBChan ch >> return ()+

@@ -5,25 +5,19 @@ -- Maintainer  : toralf.wittner@gmail.com -- Stability   : experimental -- Portability : non-portable+--+-- Re-exports all the public interfaces of 'Base',+-- 'Buffer' and 'EventLoop'.  module System.Linux.Epoll (+         module System.Linux.Epoll.Base,--    dispatchLoop,-    defaultDispatchLoop+    module System.Linux.Epoll.Buffer,+    module System.Linux.Epoll.EventLoop  ) where -import Data.Maybe-import Control.Monad import System.Linux.Epoll.Base---- | Repeatedly waits for epoll events and invokes the given callback function--- with each event occured.-dispatchLoop :: Duration -> (Event a -> IO ()) -> Device -> IO ()-dispatchLoop dur f dev = forever $ wait dur dev >>= mapM_ (\e -> f e)---- | Like 'dispatchLoop' but uses a default 'Duration' of 500ms.-defaultDispatchLoop :: (Event a -> IO ()) -> Device -> IO ()-defaultDispatchLoop = dispatchLoop (fromJust $ toDuration 500)+import System.Linux.Epoll.Buffer+import System.Linux.Epoll.EventLoop 

@@ -30,6 +30,7 @@     add,     modify,     delete,+    freeDesc,      inEvent,     outEvent,@@ -38,7 +39,8 @@     errorEvent,     hangupEvent,     edgeTriggeredEvent,-    oneShotEvent+    oneShotEvent,+    combineEvents  ) where @@ -120,14 +122,14 @@ }  instance Show EventType where-    show e | e == inEvent = "EPOLLIN"-           | e == outEvent = "EPOLLOUT"-           | e == peerCloseEvent = "EPOLLRDHUP"-           | e == urgentEvent = "EPOLLPRI"-           | e == errorEvent = "EPOLLERR"-           | e == hangupEvent = "EPOLLHUP"-           | e == edgeTriggeredEvent = "EPOLLET"-           | e == oneShotEvent = "EPOLLONESHOT"+    show e | e == inEvent = "EPOLLIN (0x001)"+           | e == outEvent = "EPOLLOUT (0x004)"+           | e == peerCloseEvent = "EPOLLRDHUP (0x2000)"+           | e == urgentEvent = "EPOLLPRI (0x002)"+           | e == errorEvent = "EPOLLERR (0x008)"+           | e == hangupEvent = "EPOLLHUP (0x010)"+           | e == edgeTriggeredEvent = "EPOLLET (1 << 31)"+           | e == oneShotEvent = "EPOLLONESHOT (1 << 30)"            | otherwise = show $ fromEventType e  #{enum Operation, Operation,@@ -159,7 +161,8 @@ -- | Adds a filedescriptor to the epoll watch set using the specified -- EventTypes. User data might be passed in as well which will be returned on -- event occurence as part of the 'Event' type. Returns an event descriptor--- which must be deleted from the watch set with 'delete'.+-- which must be deleted from the watch set with 'delete' and passed to+-- 'freeDesc' exactly once. add :: Device -> a -> [EventType] -> Fd -> IO (Descriptor a) add dev dat evts fd = do     p <- newStablePtr (fd, dat)@@ -171,21 +174,29 @@ modify dev evts des = control dev modifyOp evts (descrPtr des)  -- | Removes the event descriptor from the epoll watch set.+-- and frees descriptor which must not be used afterwards. delete :: Device -> Descriptor a -> IO ()-delete dev des = do-    control dev deleteOp [] (descrPtr des)-    freeStablePtr (descrPtr des)+delete dev des = control dev deleteOp [] (descrPtr des) +-- | Frees the resources associated with this descriptor.+-- Must be called exactly once.+freeDesc :: Descriptor a -> IO ()+freeDesc = freeStablePtr . descrPtr+ -- | Representation of epoll_ctl. control :: Device -> Operation -> [EventType] -> StablePtr (Fd, a) -> IO () control dev op evts ptr = do     (fd, _) <- deRefStablePtr ptr-    throwErrnoIfMinus1_ ("control: c_epoll_ctl (" ++ (show op) ++ ")")-        (with (EventStruct (fromIntegral . fromEventType $ combEvents evts)+    throwErrnoIfMinus1_ (errMsg fd) $+        with (EventStruct (fromIntegral . fromEventType $ combineEvents evts)                            (castStablePtrToPtr ptr))-            (c_epoll_ctl (intToNum $ deviceFd dev)-                         (fromIntegral $ fromOp op)-                         (intToNum fd)))+             (c_epoll_ctl (intToNum $ deviceFd dev)+                          (fromIntegral $ fromOp op)+                          (intToNum fd))+ where+    errMsg fd = "control: c_epoll_ctl: fd=" ++ (show fd)+             ++ ", op=" ++ (show op)+             ++ ", events=" ++ (show evts)  -- | Waits for the specified duration on all event descriptors. Returns the -- list of events that occured.@@ -219,8 +230,9 @@ toDuration :: Int -> Maybe Duration toDuration i = toWord32 i >>= Just . Duration -combEvents :: [EventType] -> EventType-combEvents = EventType . foldr ((.|.) . fromEventType) 0+-- | Bitwise OR of the list of 'EventType's.+combineEvents :: [EventType] -> EventType+combineEvents = EventType . foldr ((.|.) . fromEventType) 0  -- Foreign Imports -- 

@@ -0,0 +1,236 @@+{-# LANGUAGE TypeSynonymInstances #-}+-- |+-- Module      : System.Linux.Epoll.Buffer+-- Copyright   : (c) 2009 Toralf Wittner+-- License     : LGPL+-- Maintainer  : toralf.wittner@gmail.com+-- Stability   : experimental+-- Portability : non-portable+--+-- Buffer layer above epoll. Implemented using 'EventLoop's.+-- The general usage is that first an instance of 'Runtime' is obtained, then+-- one creates as many buffers as needed. Once done with a buffer, it has +-- to be closed and finally the runtime should be shutdown, which kills +-- the event loop, e.g.+--+-- @+-- do r <- createRuntime (fromJust . toSize $ 4096)+--    withIBuffer r stdInput $ \\b ->+--        readBuffer b >>= mapM_ print . take 10 . lines+--    shutdownRuntime r+-- @+--+-- Please note that one has to close all buffers before calling shutdown on+-- the runtime.++module System.Linux.Epoll.Buffer (+    +    Runtime,+    createRuntime,+    shutdownRuntime,++    BufElem (..),++    IBuffer,+    createIBuffer,+    closeIBuffer,+    withIBuffer,+    +    OBuffer,+    createOBuffer,+    closeOBuffer,+    withOBuffer,++    readBuffer,+    readAvail,+    readChunk,+    writeBuffer,+    flushBuffer++) where++import BChan+import Prelude+import Data.Maybe+import Control.Monad+import Control.Exception (bracket)+import System.Posix.Types (Fd)+import System.IO (hPrint, stderr)+import System.Posix.IO (fdRead, fdWrite)+import System.Linux.Epoll.Base+import System.Linux.Epoll.EventLoop++-- | Buffer Element type class.+-- Any instance of this class can be used as a buffer element.+class BufElem a where+    -- | Zero element, e.g. empty string.+    beZero   :: a+    -- | Concatenates multiple elements.+    beConcat :: [a] -> a+    -- | The length of one element.+    beLength :: a -> Int+    -- | Returns element minus integer.+    beDrop   :: Int -> a -> a+    -- | Writes element to 'Fd', returns written length.+    beWrite :: Fd -> a -> IO Int+    -- | Reads element of given length, returns element and actual length.+    beRead  :: Fd -> Int -> IO (a, Int)++instance BufElem String where+    beZero   = ""+    beConcat = concat+    beLength = length+    beDrop   = drop++    beWrite fd = liftM fromIntegral . fdWrite fd+    beRead  fd n = do+        (s, k) <- fdRead fd (fromIntegral n)+        return (s, fromIntegral k)++data Buffer a = Buffer {+        bufferChan  :: BChan (Maybe a),+        bufferCBack :: Callback+    }++-- | Buffer for reading after 'inEvent'.+newtype IBuffer a = IBuffer (Buffer a)++-- | Buffer for writing after 'outEvent'.+newtype OBuffer a = OBuffer (Buffer a)++-- | Abstract data type for buffer runtime support.+data Runtime = Runtime {+        rtILoop :: EventLoop,+        rtOLoop :: EventLoop+    }++-- | Creates a runtime instance where size denotes the epoll+-- device size (cf. 'create').+createRuntime :: Size -> IO Runtime+createRuntime s = do+    iloop <- createEventLoop s+    oloop <- createEventLoop s+    return $ Runtime iloop oloop++-- | Stops event processing and closes this runtime (and the +-- underlying epoll device).+shutdownRuntime :: Runtime -> IO ()+shutdownRuntime rt = do+    stopEventLoop (rtILoop rt)+    stopEventLoop (rtOLoop rt)++-- | Create buffer for 'inEvent's.+createIBuffer :: BufElem a => Runtime -> Fd -> IO (IBuffer a)+createIBuffer rt fd = do+    chan <- newBChan+    let emap = [(inEvents, handleRead chan), (closeEvents, handleClose chan)]+    cb <- addCallback (rtILoop rt) fd emap+    return $ IBuffer (Buffer chan cb)++-- | Create Buffer for 'outEvent's.+createOBuffer :: BufElem a => Runtime -> Fd -> IO (OBuffer a)+createOBuffer rt fd = do+    chan <- newBChan+    let emap = [(outEvents, handleWrite chan), (closeEvents, handleClose chan)]+    cb <- addCallback (rtOLoop rt) fd emap+    return $ OBuffer (Buffer chan cb)++-- | Close an IBuffer. Must not be called after 'shutdownRuntime' has been+-- invoked.+closeIBuffer :: BufElem a => Runtime -> IBuffer a -> IO ()+closeIBuffer rt (IBuffer b) = do+    writeBChan (bufferChan b) Nothing+    removeCallback (rtILoop rt) (bufferCBack b)++-- | Close an OBuffer. Must not be called after 'shutdownRuntime' has been+-- invoked.+closeOBuffer :: BufElem a => Runtime -> OBuffer a -> IO ()+closeOBuffer rt (OBuffer b) = do+    writeBChan (bufferChan b) Nothing+    removeCallback (rtOLoop rt) (bufferCBack b)++-- | Exception safe wrapper which creates an IBuffer, passes it to the provided+-- function and closes it afterwards.+withIBuffer :: BufElem a => Runtime -> Fd -> (IBuffer a -> IO ()) -> IO ()+withIBuffer r fd = bracket (createIBuffer r fd) (closeIBuffer r)++-- | Exception safe wrapper which creates an OBuffer, passes it to the provided+-- function and flushes and closes it afterwards.+withOBuffer :: BufElem a => Runtime -> Fd -> (OBuffer a -> IO ()) -> IO ()+withOBuffer r fd f = bracket (createOBuffer r fd) (closeOBuffer r) $ \b ->+    f b >> flushBuffer b++-- | Blocking read. Lazily returns all available contents from 'IBuffer'.+readBuffer :: BufElem a => IBuffer a -> IO a+readBuffer (IBuffer b) = liftM (beConcat . map fromJust . takeWhile isJust) $+    getBChanContents (bufferChan b)++-- | Blocking read. Returns one chunk from 'IBuffer'.+readChunk :: BufElem a => IBuffer a -> IO a+readChunk (IBuffer b) = liftM (fromMaybe beZero) $ readBChan (bufferChan b)++-- | Non-Blocking read. Returns one chunk if available.+readAvail :: BufElem a => IBuffer a -> IO (Maybe a)+readAvail (IBuffer b) = do+    let ch = bufferChan b+    empty <- isEmptyBChan ch+    if empty then readBChan ch else return Nothing++-- | Non-Blocking write. Writes value to buffer which will asynchronously be+-- written to file descriptor.+writeBuffer :: BufElem a => OBuffer a -> a -> IO ()+writeBuffer (OBuffer b) = writeBChan (bufferChan b) . Just++-- | Blocks until buffer is emptied.+flushBuffer :: OBuffer a -> IO ()+flushBuffer (OBuffer b) = waitBChan (bufferChan b)++--+-- Event handling+--++handleClose :: BufElem a => BChan (Maybe a) -> Device -> Event Data -> IO ()+handleClose ch _ _ = writeBChan ch Nothing -- Ensure blocking ops finish.++handleRead :: BufElem a => BChan (Maybe a) -> Device -> Event Data -> IO ()+handleRead cha dev e = do+    doRead cha (eventFd e)+    reEnableCallback dev (eventRef e) (eventDesc e)+ where+    doRead :: BufElem a => BChan (Maybe a) -> Fd -> IO ()+    doRead ch fd = do+        (s, k) <- beRead fd defaultBlockSize `catch` \er ->+                    logErr er >> return (beZero, 0)+        unless (k == 0) $+            writeBChan ch (Just s)+        when (k == defaultBlockSize) $+            doRead ch fd++handleWrite :: BufElem a => BChan (Maybe a) -> Device -> Event Data -> IO ()+handleWrite cha dev e = doWrite cha (eventFd e)+ where+    doWrite :: BufElem a => BChan (Maybe a) -> Fd -> IO ()+    doWrite ch fd = do+        s <- peekBChan ch+        case s of+            Just s' -> do+                k <- beWrite fd s' `catch` \er -> logErr er >> return 0+                if k == beLength s'+                    then do dropBChan ch+                            doWrite ch fd+                    else do unGetBChan ch (Just (beDrop k s'))+                            reEnableCallback dev (eventRef e) (eventDesc e)+            Nothing -> return ()++defaultBlockSize :: Int+defaultBlockSize = 8192++logErr :: (Show a) => a -> IO ()+logErr = hPrint stderr++inEvents :: EventType+inEvents = combineEvents [inEvent, edgeTriggeredEvent, oneShotEvent]++outEvents :: EventType+outEvents = combineEvents [outEvent, edgeTriggeredEvent, oneShotEvent]+

@@ -0,0 +1,159 @@+-- |+-- Module      : System.Linux.Epoll.EventLoop+-- Copyright   : (c) 2009 Toralf Wittner+-- License     : LGPL+-- Maintainer  : toralf.wittner@gmail.com+-- Stability   : experimental+-- Portability : non-portable+--+-- EventLoop's can be used to get notified when certain events occur on a file+-- descriptor. One can add callback functions for any 'EventType' combination.++module System.Linux.Epoll.EventLoop (++    Data,+    Callback,+    EventLoop,+    CallbackFn,+    EventMap,++    createEventLoop,+    stopEventLoop,+    addCallback,+    removeCallback,+    reEnableCallback,++    closeEvents++) where++import Util+import Control.Monad+import Control.Concurrent+import Control.Concurrent.MVar+import Data.Maybe+import System.Posix.Types (Fd)+import System.Linux.Epoll.Base+import System.Posix.IO (setFdOption, FdOption (NonBlockingRead))++-- | Callback function type+type CallbackFn = Device -> Event Data -> IO ()+type EventMap = [(EventType, CallbackFn)]++-- Used to queue up deleted descriptors to pass them to freeDesc.+--+-- The reason is that freeing descriptors is only safe while no event processing+-- takes place, otherwise delayed handler thread might still use the descriptor+-- or when free directly in 'doClose', it might be part of the next event set+-- still which would cause a segfault.+type Garbage = MVar [Descriptor Data]++data Data = Data {+        cbMap   :: EventMap,+        cbDirty :: MVar Bool -- True when closed.+    }++-- | Abstract data type holding bookeeping info.+data Callback = Callback {+        cbData :: Data,+        cbDesc :: Descriptor Data+    }++-- | Abstract data type.+data EventLoop = EventLoop {+        elDevice   :: Device,   -- Epoll Device+        elLoop     :: ThreadId, -- Epoll wait loop thread+        elGarbage  :: Garbage   -- Epoll Descriptors to be deleted+    }++-- | Create one event loop which handles up to 'Size' events per call to epoll's+-- 'wait'. An event loop runs until 'stopEventLoop' is invoked, calling 'wait'+-- with a max timeout of 500ms before it waits again.+createEventLoop :: Size -> IO EventLoop+createEventLoop s = do+    dev <- create s+    bin <- newMVar []+    lop <- forkIO $ runLoop dev bin+    return $ EventLoop dev lop bin++-- | Terminates the event loop and cleans resources. Note that one can only+-- remove callbacks from an eventloop while it is running, so make sure you call+-- this function after all 'removeCallback' calls.+stopEventLoop :: EventLoop -> IO ()+stopEventLoop el = do+    killThread (elLoop el)+    close (elDevice el)++-- | Adds a callback for the given file descriptor to this event loop. The event+-- map specifies for each event type which function to call. event types might+-- be combined using 'combineEvents'.+addCallback :: EventLoop -> Fd -> EventMap -> IO Callback+addCallback el fd emp = do+    dirty <- newMVar False+    let ety = map fst emp+        dat = Data emp dirty+    setFdOption fd NonBlockingRead True+    desc <- add (elDevice el) dat ety fd+    return $ Callback dat desc++-- | Removes the callback obtained from 'addCallback' from this event loop. Note+-- that you must not call 'stopEventLoop' before invoking this function.+removeCallback :: EventLoop -> Callback -> IO ()+removeCallback el cb = do+    doClose (elDevice el) (elGarbage el) (cbDesc cb) (cbData cb)+    return ()++-- | In case you use 'oneShotEvent' you can re-enable a callback after the event+-- occured. Otherwise no further events will be reported. Cf. epoll(7) for+-- details.+reEnableCallback :: Device -> Data -> Descriptor Data -> IO ()+reEnableCallback dev dat des =+    withMVar (cbDirty dat) $ \dirty ->+        unless dirty $+            modify dev (map fst (cbMap dat)) des++-- The heart of the event loop. Runs forever, dispatching events.+runLoop :: Device -> Garbage -> IO ()+runLoop dev bin = do+    let dur = fromJust . toDuration $ 500+    forever $ wait dur dev >>= mapM_ dispatch >> gc bin >> yield+ where+    dispatch e = case eventType e of+                    t | t =~ closeEvents -> handleClose dev bin e+                      | otherwise        -> handleEvent dev e++    -- garbage collection, i.e. free old descriptors+    gc b = modifyMVar b (\v -> return ([], v)) >>= mapM freeDesc++-- Invokes 'doClose' and potentially 'handleEvent' with 'closeEvents',+-- if the callback has not been closed before.+handleClose :: Device -> Garbage -> Event Data -> IO ()+handleClose dev bin e = do+    isDirty <- doClose dev bin (eventDesc e) (eventRef e)+    unless isDirty $ handleEvent dev (e { eventType = closeEvents })++-- Marks callback data as dirty and deletes descriptor, i.e. removes it from+-- epoll and schedules it for freeDesc.+doClose :: Device -> Garbage -> Descriptor Data -> Data -> IO Bool+doClose dev bin des dat =+    modifyMVar (cbDirty dat) $ \dirty -> do+        unless dirty $ do+            delete dev des+            modifyMVar_ bin $ return . (des:)+        return (True, dirty)++-- Forks each callback function which is applicable for the current+-- event into its own thread.+handleEvent :: Device -> Event Data -> IO ()+handleEvent dev e = do+    let fs = lookupCB (cbMap . eventRef $ e) (eventType e)+    forM_ fs $ \f -> forkIO_ $ f dev e++-- Matches the given event type against the event map.+lookupCB :: EventMap -> EventType -> [CallbackFn]+lookupCB emap ety = map snd $ filter ((=~ ety) . fst) emap++-- | A combination of 'peerCloseEvent', 'errorEvent', 'hangupEvent'.+closeEvents :: EventType+closeEvents = combineEvents [peerCloseEvent, errorEvent, hangupEvent]+

@@ -1,154 +0,0 @@-{-# LANGUAGE GeneralizedNewtypeDeriving #-}---- |--- Module      : System.Linux.EpollM--- Copyright   : (c) 2009 Toralf Wittner--- License     : LGPL--- Maintainer  : toralf.wittner@gmail.com--- Stability   : experimental--- Portability : non-portable------ Monadic epoll interface. Similar to System.Linux.Epoll.Base but uses--- 'ReaderT' to hold a 'Device' instead of passing it to many functions.--module System.Linux.EpollM (--    Epoll,-    runEpoll,-    runEpoll_,-    runEpollSmall_,-    runEpollMed_,-    runEpollBig_,-    add,-    add_,-    modify,-    delete,-    wait,-    wait_,-    device,-    dispatchLoop,-    dispatchLoop_,-    defaultDispatchLoop,-    defaultDispatchLoop_,-    fork,-    fork_,--    E.EventType,-    E.Size,-    E.toSize,-    E.Duration,-    E.toDuration,-    E.Descriptor,-    E.Device,-    E.Event (eventFd, eventType, eventRef, eventDesc),-    (E.=~),--    E.create,-    E.close,--    E.inEvent,-    E.outEvent,-    E.peerCloseEvent,-    E.urgentEvent,-    E.errorEvent,-    E.hangupEvent,-    E.edgeTriggeredEvent,-    E.oneShotEvent--) where--import Data.Maybe (fromJust)-import Control.Concurrent-import Control.Monad.Reader-import Control.Exception (bracket)-import System.Posix.Types (Fd)-import qualified System.Linux.Epoll as E--newtype Epoll a = Epoll {-        runDev :: ReaderT E.Device IO a-    } deriving (Monad, MonadIO, MonadReader E.Device) ---- | Run Epoll monad.-runEpoll :: E.Device -> Epoll a -> IO a-runEpoll d e = runReaderT (runDev e) d---- | Run Epoll monad. Like 'runEpoll' but creates and closes--- 'Device' implcitely.-runEpoll_ :: E.Size -> Epoll a -> IO a-runEpoll_ s e = bracket (E.create s) E.close (flip runEpoll e)---- | Like 'runEpoll_' but with an implicit 'Size' value of 8.-runEpollSmall_ :: Epoll a -> IO a-runEpollSmall_ = runEpollN_ 8---- | Like 'runEpoll_' but with an implicit 'Size' value of 256.-runEpollMed_ :: Epoll a -> IO a-runEpollMed_ = runEpollN_ 256---- | Like 'runEpoll_' but with an implicit 'Size' value of 8192.-runEpollBig_ :: Epoll a -> IO a-runEpollBig_ = runEpollN_ 8192---- | Adds the given file descriptor with the specified event types to epoll.-add :: a -> [E.EventType] -> Fd -> Epoll (E.Descriptor a)-add d e f = do-    dev <- ask-    des <- liftIO $ E.add dev d e f-    return des---- | Like 'add' but without accepting custom data.-add_ :: [E.EventType] -> Fd -> Epoll (E.Descriptor ())-add_ = add () ---- | Modify the event type set of the given descriptor.-modify :: [E.EventType] -> E.Descriptor a -> Epoll ()-modify e d = do-    dev <- ask-    liftIO $ E.modify dev e d---- | Deletes the descriptor from epoll.-delete :: E.Descriptor a -> Epoll ()-delete d = do-    dev <- ask-    liftIO $ E.delete dev d---- | Waits up to the given duration for events on all descriptors.-wait :: E.Duration -> Epoll [E.Event a]-wait d = do-    dev <- ask-    liftIO $ E.wait d dev---- | Like 'wait' but uses a 'Duration' of 500ms.-wait_ :: Epoll [E.Event a]-wait_ = wait (fromJust $ E.toDuration 500)--runEpollN_ :: Int -> Epoll a -> IO a-runEpollN_ = runEpoll_ . fromJust . E.toSize--device :: Epoll E.Device-device = ask---- | Waits for events and calls the given function for each.-dispatchLoop :: E.Duration -> (E.Event a -> Epoll ()) -> Epoll ()-dispatchLoop dur f = forever $ wait dur >>= mapM_ f---- | Like 'dispatchLoop' but with predefined 'Duration' of 500ms.-defaultDispatchLoop :: (E.Event a -> Epoll ()) -> Epoll ()-defaultDispatchLoop = dispatchLoop (fromJust $ E.toDuration 500)---- | Like 'dispatchLoop' but forks itself into another thread-dispatchLoop_ :: E.Duration -> (E.Event a -> Epoll ()) -> Epoll ThreadId-dispatchLoop_ dur f = fork $ forever $ wait dur >>= mapM_ f---- | Like 'defaultDispatchLoop' but forks itself into another thread-defaultDispatchLoop_ :: (E.Event a -> Epoll ()) -> Epoll ThreadId-defaultDispatchLoop_ = dispatchLoop_ (fromJust $ E.toDuration 500)---- | Uses 'forkIO' to spark an epoll computation into another thread.-fork :: Epoll () -> Epoll ThreadId-fork (Epoll r) = Epoll $ mapReaderT forkIO r---- | Like 'fork' but swallows the ThreadId.-fork_ :: Epoll () -> Epoll ()-fork_ e = fork e >> return ()-

@@ -1,4 +1,12 @@-module Util (intToNum, toWord32, forkIO_) where+-- |+-- Module      : Util+-- Copyright   : (c) 2009 Toralf Wittner+-- License     : LGPL+-- Maintainer  : toralf.wittner@gmail.com+-- Stability   : internal+-- Portability : portable++module Util where  import Data.Word import Control.Concurrent (forkIO)