scalable-server 0.1.2 → 0.2.0

raw patch · 3 files changed

+15/−13 lines, 3 filesdep +BoundedChandep −stm

Dependencies added: BoundedChan

Dependencies removed: stm

Files

• example.hs +1/−1
• scalable-server.cabal +3/−3
• src/Network/Server/ScalableServer.hs +11/−9

@@ -32,6 +32,6 @@ main = testServer  testServer = do-    let definition = RequestPipeline parseRequest handler+    let definition = RequestPipeline parseRequest handler 10     void $ forkIO $ runServer definition 6004     forever $ threadDelay (1000000 * 60)

@@ -3,7 +3,7 @@ Synopsis:            Library for writing fast/scalable TCP-based services Description:         Library for writing fast/scalable TCP-based services -Version:             0.1.2+Version:             0.2.0  License:             BSD3 @@ -29,7 +29,6 @@   hs-source-dirs:      src    Build-depends: base >=4 && <5,-                 stm>=2.2,                  bytestring >=0.9,                  attoparsec >=0.9.1 && < 0.10,                  enumerator >= 0.4 && < 0.5,@@ -37,6 +36,7 @@                  network-enumerator >= 0.1 && < 0.2,                  network>=2.3,                  mtl>=2 && <3,-                 blaze-builder>=0.3 && <0.4+                 blaze-builder>=0.3 && <0.4,+                 BoundedChan >=1.0 && < 1.1    ghc-options:   -O2

@@ -14,7 +14,7 @@ import Control.Monad (forever, liftM, replicateM, void) import Control.Monad.Trans (liftIO) import Control.Concurrent (forkIO, threadDelay)-import Control.Concurrent.STM (newTChanIO, writeTChan, readTChan, atomically, TChan)+import Control.Concurrent.BoundedChan (newBoundedChan, writeChan, readChan, BoundedChan) import Data.Enumerator (($$), run_) import qualified Data.Enumerator as E import Data.Attoparsec.Enumerator (iterParser)@@ -48,7 +48,7 @@ -- indicating both a parser/request object generator, the RequestCreator, -- and the processor of these requests, one that ultimately generates a -- response expressed by a blaze 'Builder'-data RequestPipeline a = RequestPipeline (RequestCreator a) (RequestProcessor a)+data RequestPipeline a = RequestPipeline (RequestCreator a) (RequestProcessor a) PipelineSize  -- |The RequestCreator is an Attoparsec parser that yields some request -- object 'a'@@ -57,8 +57,11 @@ -- |The RequestProcessor is a function in the IO monad (for DB access, etc) -- that returns a builder that can generate the response type RequestProcessor a = a -> IO Builder-type RequestChannel a = TChan (Maybe a) +type RequestChannel a  = BoundedChan (Maybe a)++type PipelineSize = Int+ -- |Given a pipeline specification and a port, run TCP traffic using the -- pipeline for parsing, processing and response. --@@ -80,32 +83,31 @@         forkIO $ connHandler pipe c  connHandler :: RequestPipeline a -> Socket -> IO ()-connHandler (RequestPipeline reqParse reqProc) s = do-    chan <- newTChanIO+connHandler (RequestPipeline reqParse reqProc size) s = do+    chan <- newBoundedChan size     (do         let enum = enumSocket 4096 s         let parser = iterParser reqParse         void $ forkIO $ processRequests chan reqProc s         void $ run_ (enum $$ E.sequence parser $$ requestHandler chan s)-        ) `finally` ( (atomically $ writeTChan chan Nothing) >> sClose s )+        ) `finally` ( (writeChan chan Nothing) >> sClose s )  requestHandler :: RequestChannel a -> Socket -> Iteratee a IO () requestHandler chan s = do     continue requestConsume   where     requestConsume (Chunks mrs) = do-        liftIO $ mapM_ (\m -> atomically $ writeTChan chan $ Just m) mrs+        liftIO $ mapM_ (\m -> writeChan chan $ Just m) mrs         continue requestConsume     requestConsume EOF = do         yield () EOF  processRequests :: RequestChannel a -> RequestProcessor a -> Socket -> IO () processRequests chan proc s = do-    next <- atomically $ readTChan chan+    next <- readChan chan     case next of         Just a -> do             resp <- proc a -- XXX handle exceptions?-            -- XXX eventually, pipeline to enumerator?             toByteStringIO (BinSock.sendAll s) $ resp             processRequests chan proc s         Nothing -> return ()