packages feed

scalable-server (empty) → 0.1

raw patch · 5 files changed

+211/−0 lines, 5 filesdep +attoparsecdep +attoparsec-enumeratordep +basesetup-changed

Dependencies added: attoparsec, attoparsec-enumerator, base, blaze-builder, bytestring, enumerator, mtl, network, stm

Files

+ LICENSE view
@@ -0,0 +1,25 @@+Copyright 2011 Jamie Turner. 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.++THIS SOFTWARE IS PROVIDED BY THE AUTHOR ``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 AUTHOR 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.++The views and conclusions contained in the software and documentation are those of the+authors and should not be interpreted as representing official policies, either expressed+or implied, of Jamie Turner.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ example.hs view
@@ -0,0 +1,37 @@+{-# LANGUAGE OverloadedStrings #-}++import qualified Data.Attoparsec as Atto+import qualified Data.ByteString.Char8 as S++import Control.Monad (void, forever)+import Control.Concurrent (forkIO, threadDelay)++import Blaze.ByteString.Builder.ByteString (copyByteString, fromByteString)+import Blaze.ByteString.Builder (Builder)++import Network.Server.ScalableServer++data PingPongRequest = Request S.ByteString++parseRequest :: Atto.Parser PingPongRequest+parseRequest = do+    w <- Atto.takeTill (==13)+    _ <- Atto.string "\r\n"+    return $ Request w++handler :: PingPongRequest -> IO Builder+handler req = return $ handle req++handle :: PingPongRequest -> Builder+handle (Request "PING") = do+    copyByteString "PONG\r\n"++handle (Request _) = do+    copyByteString "ERROR\r\n"++main = testServer++testServer = do+    let definition = RequestPipeline parseRequest handler+    void $ forkIO $ runServer definition 6004+    forever $ threadDelay (1000000 * 60)
+ scalable-server.cabal view
@@ -0,0 +1,39 @@+Name:                scalable-server++Synopsis:            Library for writing fast/scalable TCP-based services+Description:         Library for writing fast/scalable TCP-based services++Version:             0.1++License:             BSD3++License-file:        LICENSE++Author:              Jamie Turner++Maintainer:          jamie@bu.mp++Category:            Network+Build-type:          Simple++Extra-source-files: example.hs++Cabal-version:       >=1.2+++Library+  Exposed-modules:     Network.Server.ScalableServer++  hs-source-dirs:      src++  Build-depends: base >=4 && <5,+                 stm,+                 bytestring >=0.9,+                 attoparsec >=0.7.2,+                 enumerator,+                 attoparsec-enumerator,+                 network>=2.0,+                 mtl>=2,+                 blaze-builder>=0.3++  ghc-options:   -O2
+ src/Network/Server/ScalableServer.hs view
@@ -0,0 +1,108 @@+module Network.Server.ScalableServer (runServer,+    RequestPipeline(..), RequestCreator,+    RequestProcessor) where++import Network.Socket+import qualified Network.Socket.ByteString as BinSock+import Network.BSD+import Control.Exception (finally)+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 Data.Enumerator (($$), run_)+import qualified Data.Enumerator as E+import Data.Enumerator.Binary (enumHandle)+import Data.Attoparsec.Enumerator (iterParser)+import System.IO (hSetBuffering, hClose, BufferMode(..),+    IOMode(..), Handle)+import Blaze.ByteString.Builder (toByteStringIO, Builder)+import Data.Enumerator (yield, continue, Iteratee, Stream(..))+import qualified Data.Attoparsec as Atto+import qualified Data.Attoparsec.Char8 as AttoC+import qualified Data.ByteString.Char8 as S+import qualified Data.ByteString as WS+import qualified Data.ByteString.Lazy.Char8 as B++-- 'ScalableServer' is a library that attempts to capture current best+-- practices for writing fast/scalable socket servers in Haskell.+--+-- Currently, that involves providing the right glue for hooking up+-- to enumerator/attoparsec-enumerator/blaze-builder and network-bytestring+--+-- It provides a relatively simple parse/generate toolchain for+-- plugging into these engines+--+-- Servers written using this library support "pipelining"; that is, a client+-- can issue many requests serially before the server has responded to the+-- first+--+-- Server written using this library also can be invoked with +RTS -NX+-- invocation for multicore support++-- |The 'RequestPipeline' acts as a specification for your service,+-- 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)++-- |The RequestCreator is an Attoparsec parser that yields some request+-- object 'a'+type RequestCreator a = Atto.Parser a++-- |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)++-- |Given a pipeline specification and a port, run TCP traffic using the+-- pipeline for parsing, processing and response.+--+-- Note: there is currently no way for a server to specific the socket+-- should be disconnected+runServer :: RequestPipeline a -> PortNumber -> IO ()+runServer pipe port = do+    proto <- getProtocolNumber "tcp"+    s <- socket AF_INET Stream proto+    setSocketOption s ReuseAddr 1+    bindSocket s (SockAddrInet port iNADDR_ANY)+    serverListenLoop pipe s++serverListenLoop :: RequestPipeline a -> Socket -> IO ()+serverListenLoop pipe s = do+    listen s 100+    forever $ do+        (c, _) <- accept s+        h <- socketToHandle c ReadMode+        hSetBuffering h NoBuffering+        forkIO $ connHandler pipe c h++connHandler :: RequestPipeline a -> Socket -> Handle -> IO ()+connHandler (RequestPipeline reqParse reqProc) s h = do+    chan <- newTChanIO+    (do+        let enum = enumHandle 32768 h+        let parser = iterParser reqParse+        void $ forkIO $ processRequests chan reqProc s+        void $ run_ (enum $$ E.sequence parser $$ requestHandler chan s)+        ) `finally` ( (atomically $ writeTChan chan Nothing) >> hClose h )++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+        continue requestConsume+    requestConsume EOF = do+        yield () EOF++processRequests :: RequestChannel a -> RequestProcessor a -> Socket -> IO ()+processRequests chan proc s = do+    next <- atomically $ readTChan chan+    case next of+        Just a -> do+            resp <- proc a -- XXX handle exceptions?+            toByteStringIO (BinSock.sendAll s) $ resp+            processRequests chan proc s+        Nothing -> return ()