diff --git a/README b/README --- a/README +++ b/README @@ -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. diff --git a/epoll.cabal b/epoll.cabal --- a/epoll.cabal +++ b/epoll.cabal @@ -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 diff --git a/examples/Server.hs b/examples/Server.hs --- a/examples/Server.hs +++ b/examples/Server.hs @@ -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 ") 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 diff --git a/examples/Stdin.hs b/examples/Stdin.hs --- a/examples/Stdin.hs +++ b/examples/Stdin.hs @@ -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 diff --git a/examples/StdinM.hs b/examples/StdinM.hs deleted file mode 100644 --- a/examples/StdinM.hs +++ /dev/null @@ -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 - diff --git a/src/BChan.hs b/src/BChan.hs new file mode 100644 --- /dev/null +++ b/src/BChan.hs @@ -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 () + diff --git a/src/System/Linux/Epoll.hs b/src/System/Linux/Epoll.hs --- a/src/System/Linux/Epoll.hs +++ b/src/System/Linux/Epoll.hs @@ -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 diff --git a/src/System/Linux/Epoll/Base.hsc b/src/System/Linux/Epoll/Base.hsc --- a/src/System/Linux/Epoll/Base.hsc +++ b/src/System/Linux/Epoll/Base.hsc @@ -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 -- diff --git a/src/System/Linux/Epoll/Buffer.hs b/src/System/Linux/Epoll/Buffer.hs new file mode 100644 --- /dev/null +++ b/src/System/Linux/Epoll/Buffer.hs @@ -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] + diff --git a/src/System/Linux/Epoll/EventLoop.hs b/src/System/Linux/Epoll/EventLoop.hs new file mode 100644 --- /dev/null +++ b/src/System/Linux/Epoll/EventLoop.hs @@ -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] + diff --git a/src/System/Linux/EpollM.hs b/src/System/Linux/EpollM.hs deleted file mode 100644 --- a/src/System/Linux/EpollM.hs +++ /dev/null @@ -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 () - diff --git a/src/Util.hs b/src/Util.hs --- a/src/Util.hs +++ b/src/Util.hs @@ -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)