<commit_before>{-# LANGUAGE BangPatterns             #-}
{-# LANGUAGE CPP                      #-}
{-# LANGUAGE DeriveDataTypeable       #-}
{-# LANGUAGE ForeignFunctionInterface #-}
{-# LANGUAGE OverloadedStrings        #-}
{-# LANGUAGE PackageImports           #-}
{-# LANGUAGE Rank2Types               #-}
{-# LANGUAGE ScopedTypeVariables      #-}

module Snap.Internal.Http.Server.SimpleBackend
  ( simpleEventLoop
  ) where


------------------------------------------------------------------------------
import           Control.Monad.Trans

import           Control.Concurrent hiding (yield)
import           Control.Concurrent.Extended (forkOnLabeledWithUnmaskBs)
import           Control.Exception
import           Control.Monad
import           Data.ByteString (ByteString)
import qualified Data.ByteString as S
import qualified Data.ByteString.Char8 as SC
import           Data.ByteString.Internal (c2w)
import           Data.Maybe
import           Foreign hiding (new)
import           Foreign.C
import           GHC.Conc (labelThread)
import           Network.Socket
#if !MIN_VERSION_base(4,6,0)
import           Prelude hiding (catch)
#endif
------------------------------------------------------------------------------
import           Snap.Internal.Debug
import           Snap.Internal.Http.Server.Date
import qualified Snap.Internal.Http.Server.TimeoutManager as TM
import           Snap.Internal.Http.Server.TimeoutManager (TimeoutManager)
import           Snap.Internal.Http.Server.Backend
import           Snap.Internal.Http.Server.Address
import qualified Snap.Internal.Http.Server.ListenHelpers as Listen
import           Snap.Iteratee hiding (map)

#if defined(HAS_SENDFILE)
import qualified System.SendFile as SF
import           System.Posix.IO
import           System.Posix.Types (Fd(..))
#endif


------------------------------------------------------------------------------
-- | For each cpu, we store:
--    * A list of accept threads, one per port.
--    * A TimeoutManager
--    * An mvar to signal when the timeout thread is shutdown
data EventLoopCpu = EventLoopCpu
    { _boundCpu        :: Int
    , _acceptThreads   :: [ThreadId]
    , _timeoutManager  :: TimeoutManager
    , _exitMVar        :: !(MVar ())
    }


------------------------------------------------------------------------------
simpleEventLoop :: EventLoop
simpleEventLoop defaultTimeout sockets cap elog initial handler = do
    loops <- Prelude.mapM (newLoop defaultTimeout sockets handler elog)
                          [0..(cap-1)]

    initial
    debug "simpleEventLoop: waiting for mvars"

    --wait for all threads to exit
    Prelude.mapM_ (takeMVar . _exitMVar) loops `finally` do
        debug "simpleEventLoop: killing all threads"
        _ <- mapM_ stopLoop loops
        mapM_ Listen.closeSocket sockets


------------------------------------------------------------------------------
newLoop :: Int
        -> [ListenSocket]
        -> SessionHandler
        -> (S.ByteString -> IO ())
        -> Int
        -> IO EventLoopCpu
newLoop defaultTimeout sockets handler elog cpu = do
    tmgr       <- TM.initialize defaultTimeout getCurrentDateTime
    exit       <- newEmptyMVar
    accThreads <- forM sockets $ \p -> do
      let label = S.concat
                  [ "snap-server: ",    SC.pack (show p)
                  , " on capability: ", SC.pack (show cpu)
                  ]
      forkOnLabeledWithUnmaskBs label cpu $ \unmask ->
        unmask $ acceptThread defaultTimeout handler tmgr elog cpu p exit

    return $! EventLoopCpu cpu accThreads tmgr exit


------------------------------------------------------------------------------
stopLoop :: EventLoopCpu -> IO ()
stopLoop loop = block $ do
    TM.stop $ _timeoutManager loop
    Prelude.mapM_ killThread $ _acceptThreads loop


------------------------------------------------------------------------------
acceptThread :: Int
             -> SessionHandler
             -> TimeoutManager
             -> (S.ByteString -> IO ())
             -> Int
             -> ListenSocket
             -> MVar ()
             -> IO ()
acceptThread defaultTimeout handler tmgr elog cpu sock exitMVar =
    loop `finally` (tryPutMVar exitMVar () >> return ())
  where
    acceptAndFork = do
        debug $ "acceptThread: calling accept() on socket " ++ show sock
        (s,addr) <- accept $ Listen.listenSocket sock
        setSocketOption s NoDelay 1
        debug $ "acceptThread: accepted connection from remote: " ++ show addr
        let label = S.concat
                    [ "snap-server: connection from: "
                    , SC.pack (show addr)
                    , " on socket: "
                    , SC.pack (show (fdSocket s))
                    , "\0"
                    ]
        _ <- forkOnLabeledWithUnmaskBs label cpu $ \unmask ->
               unmask $ go s addr `catches` cleanup
        return ()

    loop = do
        acceptAndFork `catches` acceptHandler
        loop

    go = runSession defaultTimeout handler tmgr sock

    acceptHandler =
        [ Handler $ \(e :: AsyncException) -> throwIO e
        , Handler $ \(e :: SomeException) -> do
              elog $ S.concat [ "SimpleBackend.acceptThread: accept threw: "
                              , S.pack . map c2w $ show e ]
              -- we're out of file descriptors, and it isn't likely to get
              -- better immediately; sleep for 10ms to avoid spamming the error
              -- log.
              threadDelay $ 10000
        ]

    cleanup =
        [
          Handler $ \(_ :: AsyncException) -> return ()
        , Handler $ \(e :: SomeException) -> elog
                  $ S.concat [ "SimpleBackend.acceptThread: "
                             , S.pack . map c2w $ show e]
        ]
<commit_msg>Install the 'cleanup' exception handler before unmasking asynchronous exceptions<commit_after>{-# LANGUAGE BangPatterns             #-}
{-# LANGUAGE CPP                      #-}
{-# LANGUAGE DeriveDataTypeable       #-}
{-# LANGUAGE ForeignFunctionInterface #-}
{-# LANGUAGE OverloadedStrings        #-}
{-# LANGUAGE PackageImports           #-}
{-# LANGUAGE Rank2Types               #-}
{-# LANGUAGE ScopedTypeVariables      #-}

module Snap.Internal.Http.Server.SimpleBackend
  ( simpleEventLoop
  ) where


------------------------------------------------------------------------------
import           Control.Monad.Trans

import           Control.Concurrent hiding (yield)
import           Control.Concurrent.Extended (forkOnLabeledWithUnmaskBs)
import           Control.Exception
import           Control.Monad
import           Data.ByteString (ByteString)
import qualified Data.ByteString as S
import qualified Data.ByteString.Char8 as SC
import           Data.ByteString.Internal (c2w)
import           Data.Maybe
import           Foreign hiding (new)
import           Foreign.C
import           GHC.Conc (labelThread)
import           Network.Socket
#if !MIN_VERSION_base(4,6,0)
import           Prelude hiding (catch)
#endif
------------------------------------------------------------------------------
import           Snap.Internal.Debug
import           Snap.Internal.Http.Server.Date
import qualified Snap.Internal.Http.Server.TimeoutManager as TM
import           Snap.Internal.Http.Server.TimeoutManager (TimeoutManager)
import           Snap.Internal.Http.Server.Backend
import           Snap.Internal.Http.Server.Address
import qualified Snap.Internal.Http.Server.ListenHelpers as Listen
import           Snap.Iteratee hiding (map)

#if defined(HAS_SENDFILE)
import qualified System.SendFile as SF
import           System.Posix.IO
import           System.Posix.Types (Fd(..))
#endif


------------------------------------------------------------------------------
-- | For each cpu, we store:
--    * A list of accept threads, one per port.
--    * A TimeoutManager
--    * An mvar to signal when the timeout thread is shutdown
data EventLoopCpu = EventLoopCpu
    { _boundCpu        :: Int
    , _acceptThreads   :: [ThreadId]
    , _timeoutManager  :: TimeoutManager
    , _exitMVar        :: !(MVar ())
    }


------------------------------------------------------------------------------
simpleEventLoop :: EventLoop
simpleEventLoop defaultTimeout sockets cap elog initial handler = do
    loops <- Prelude.mapM (newLoop defaultTimeout sockets handler elog)
                          [0..(cap-1)]

    initial
    debug "simpleEventLoop: waiting for mvars"

    --wait for all threads to exit
    Prelude.mapM_ (takeMVar . _exitMVar) loops `finally` do
        debug "simpleEventLoop: killing all threads"
        _ <- mapM_ stopLoop loops
        mapM_ Listen.closeSocket sockets


------------------------------------------------------------------------------
newLoop :: Int
        -> [ListenSocket]
        -> SessionHandler
        -> (S.ByteString -> IO ())
        -> Int
        -> IO EventLoopCpu
newLoop defaultTimeout sockets handler elog cpu = do
    tmgr       <- TM.initialize defaultTimeout getCurrentDateTime
    exit       <- newEmptyMVar
    accThreads <- forM sockets $ \p -> do
      let label = S.concat
                  [ "snap-server: ",    SC.pack (show p)
                  , " on capability: ", SC.pack (show cpu)
                  ]
      forkOnLabeledWithUnmaskBs label cpu $ \unmask ->
        unmask $ acceptThread defaultTimeout handler tmgr elog cpu p exit

    return $! EventLoopCpu cpu accThreads tmgr exit


------------------------------------------------------------------------------
stopLoop :: EventLoopCpu -> IO ()
stopLoop loop = block $ do
    TM.stop $ _timeoutManager loop
    Prelude.mapM_ killThread $ _acceptThreads loop


------------------------------------------------------------------------------
acceptThread :: Int
             -> SessionHandler
             -> TimeoutManager
             -> (S.ByteString -> IO ())
             -> Int
             -> ListenSocket
             -> MVar ()
             -> IO ()
acceptThread defaultTimeout handler tmgr elog cpu sock exitMVar =
    loop `finally` (tryPutMVar exitMVar () >> return ())
  where
    acceptAndFork = do
        debug $ "acceptThread: calling accept() on socket " ++ show sock
        (s,addr) <- accept $ Listen.listenSocket sock
        setSocketOption s NoDelay 1
        debug $ "acceptThread: accepted connection from remote: " ++ show addr
        let label = S.concat
                    [ "snap-server: connection from: "
                    , SC.pack (show addr)
                    , " on socket: "
                    , SC.pack (show (fdSocket s))
                    , "\0"
                    ]
        _ <- forkOnLabeledWithUnmaskBs label cpu $ \unmask ->
               unmask (go s addr) `catches` cleanup
        return ()

    loop = do
        acceptAndFork `catches` acceptHandler
        loop

    go = runSession defaultTimeout handler tmgr sock

    acceptHandler =
        [ Handler $ \(e :: AsyncException) -> throwIO e
        , Handler $ \(e :: SomeException) -> do
              elog $ S.concat [ "SimpleBackend.acceptThread: accept threw: "
                              , S.pack . map c2w $ show e ]
              -- we're out of file descriptors, and it isn't likely to get
              -- better immediately; sleep for 10ms to avoid spamming the error
              -- log.
              threadDelay $ 10000
        ]

    cleanup =
        [
          Handler $ \(_ :: AsyncException) -> return ()
        , Handler $ \(e :: SomeException) -> elog
                  $ S.concat [ "SimpleBackend.acceptThread: "
                             , S.pack . map c2w $ show e]
        ]

{'fraction_non_alphanumeric': 0.17375246929485527, 'fraction_numerical': 0.003263763634802027, 'mean_word_length': 1.4380234505862646, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 51, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2227349', 'n_tokens_mistral': 2992, 'n_tokens_neox': 2883, 'n_words': 830}