<commit_before>{-# LANGUAGE DeriveDataTypeable, RecordWildCards, FlexibleContexts #-}
--------------------------------------------------------------------------------
{- |
  Module      :  Data.Acid.CenteredSlave.hs
  Copyright   :  ?

  Maintainer  :  <EMAIL>
  Portability :  non-portable (uses GHC extensions)

  The Slave part of a the Centered replication backend for acid state.

-}

--------------------------------------------------------------------------------
-- SLAVE part

module Data.Acid.Centered.Slave
    (
      enslaveState
    , enslaveStateFrom
    , enslaveRedState
    , enslaveRedStateFrom
    , SlaveState(..)
    )  where

import Data.Typeable
import Data.SafeCopy
import Data.Serialize (decode, encode, runPutLazy, runGetLazy)

import Data.Acid
import Data.Acid.Core
import Data.Acid.Abstract
import Data.Acid.Local
import Data.Acid.Log

import Data.Acid.Centered.Common

import System.ZMQ4 (Context, Socket, Dealer(..),
                    setReceiveHighWM, setSendHighWM, restrict,
                    poll, Poll(..), Event(..),
                    context, term, socket, close,
                    connect, disconnect, send, receive)
import System.FilePath ( (</>) )

import Control.Concurrent (forkIO, ThreadId, myThreadId, killThread, threadDelay)
import Control.Concurrent.MVar (MVar, newMVar, newEmptyMVar,
                                withMVar, modifyMVar, modifyMVar_,
                                takeMVar, putMVar)
import Data.IORef (writeIORef)
import Control.Monad (void, when, unless, liftM)
import Control.Monad.STM (atomically)
import Control.Concurrent.STM.TVar (readTVar, writeTVar)
import qualified Control.Concurrent.Event as Event
import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
import Control.Exception (handle, throwTo, SomeException, ErrorCall(..))

import Data.IntMap (IntMap)
import qualified Data.IntMap as IM
import Data.Maybe (fromMaybe)

import qualified Data.ByteString.Lazy.Char8 as CSL

--------------------------------------------------------------------------------

-- | Slave state structure, for internal use.
data SlaveState st
    = SlaveState { slaveLocalState :: AcidState st
                 , slaveStateIsRed :: Bool
                 , slaveRepFinalizers :: MVar (IntMap (IO ()))
                 , slaveRepChan :: Chan SlaveRepItem
                 , slaveSyncDone :: Event.Event
                 , slaveRevision :: MVar NodeRevision
                 , slaveRequests :: MVar SlaveRequests
                 , slaveLastRequestID :: MVar RequestID
                 , slaveRepThreadId :: MVar ThreadId
                 , slaveReqThreadId :: MVar ThreadId
                 , slaveParentThreadId :: ThreadId
                 , slaveZmqContext :: Context
                 , slaveZmqAddr :: String
                 , slaveZmqSocket :: MVar (Socket Dealer)
                 } deriving (Typeable)

-- | Memory of own Requests sent to Master.
type SlaveRequests = IntMap (IO (IO ()),ThreadId)

-- | One Update + Metainformation to replicate.
data SlaveRepItem =
      SRIEnd
    | SRICheckpoint Revision
    | SRIArchive Revision
    | SRIUpdate Revision (Maybe RequestID) (Tagged CSL.ByteString)

-- | Open a local State as Slave for a Master.
--
-- The directory for the local state files is the default one ("state/[typeOf
-- state]").
enslaveState :: (IsAcidic st, Typeable st) =>
            String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveState address port initialState =
    enslaveStateFrom ("state" </> show (typeOf initialState)) address port initialState

-- | Open a local State as Slave for a Master.
--
-- The directory for the local state files is the default one ("state/[typeOf
-- state]").
enslaveRedState :: (IsAcidic st, Typeable st) =>
            String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveRedState address port initialState =
    enslaveRedStateFrom ("state" </> show (typeOf initialState)) address port initialState

-- | Open a local State as Slave for a Master. The directory of the local state
-- files can be specified.
enslaveStateFrom :: (IsAcidic st, Typeable st) =>
            FilePath        -- ^ location of the local state files.
         -> String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveStateFrom = enslaveMayRedStateFrom False

-- | Open a local State as Slave for a _redundant_ Master. The directory of the local state
-- files can be specified.
enslaveRedStateFrom :: (IsAcidic st, Typeable st) =>
            FilePath        -- ^ location of the local state files.
         -> String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveRedStateFrom = enslaveMayRedStateFrom True

-- | Open a local State as Slave for a Master, redundant or not.
--   The directory of the local state files can be specified.
enslaveMayRedStateFrom :: (IsAcidic st, Typeable st) =>
            Bool            -- ^ is redundant
         -> FilePath        -- ^ location of the local state files.
         -> String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveMayRedStateFrom isRed directory address port initialState = do
        -- local
        lst <- openLocalStateFrom directory initialState
        let levs = localEvents $ downcast lst
        lrev <- atomically $ readTVar $ logNextEntryId levs
        rev <- newMVar lrev
        debug $ "Opening enslaved state at revision " ++ show lrev
        srs <- newMVar IM.empty
        lastReqId <- newMVar 0
        repChan <- newChan
        syncDone <- Event.new
        reqTid <- newEmptyMVar
        repTid <- newEmptyMVar
        parTid <- myThreadId
        repFin <- newMVar IM.empty
        -- remote
        let addr = "tcp://" ++ address ++ ":" ++ show port
        ctx <- context
        sock <- socket ctx Dealer
        setReceiveHighWM (restrict (100*1000 :: Int)) sock
        setSendHighWM (restrict (100*1000 :: Int)) sock
        connect sock addr
        msock <- newMVar sock
        sendToMaster msock $ NewSlave lrev
        let slaveState = SlaveState { slaveLocalState = lst
                                    , slaveStateIsRed = isRed
                                    , slaveRepFinalizers = repFin
                                    , slaveRepChan = repChan
                                    , slaveSyncDone = syncDone
                                    , slaveRevision = rev
                                    , slaveRequests = srs
                                    , slaveLastRequestID = lastReqId
                                    , slaveReqThreadId = reqTid
                                    , slaveRepThreadId = repTid
                                    , slaveParentThreadId = parTid
                                    , slaveZmqContext = ctx
                                    , slaveZmqAddr = addr
                                    , slaveZmqSocket = msock
                                    }
        void $ forkIO $ slaveRequestHandler slaveState
        void $ forkIO $ slaveReplicationHandler slaveState
        return $ slaveToAcidState slaveState

-- | Replication handler of the Slave.
slaveRequestHandler :: (IsAcidic st, Typeable st) => SlaveState st -> IO ()
slaveRequestHandler slaveState@SlaveState{..} = do
    mtid <- myThreadId
    putMVar slaveReqThreadId mtid
    let loop = handle (\e -> throwTo slaveParentThreadId (e :: SomeException)) $
          handle killHandler $ do
            --waitRead =<< readMVar slaveZmqSocket
            -- FIXME: we needn't poll if not for strange zmq behaviour
            re <- withMVar slaveZmqSocket $ \sock -> poll 100 [Sock sock [In] Nothing]
            unless (null $ head re) $ do
                msg <- withMVar slaveZmqSocket receive
                case decode msg of
                    Left str -> error $ "Data.Serialize.decode failed on MasterMessage: " ++ show str
                    Right mmsg -> do
                         debug $ "Received: " ++ show mmsg
                         case mmsg of
                            -- We are sent an Update to replicate.
                            DoRep r i d -> queueRepItem slaveState (SRIUpdate r i d)
                            -- We are sent a Checkpoint for synchronization.
                            DoSyncCheckpoint r d -> replicateSyncCp slaveState r d
                            -- We are sent an Update to replicate for synchronization.
                            DoSyncRep r d -> replicateSyncUpdate slaveState r d
                            -- Master done sending all synchronization Updates.<commit_msg>fix: enable for proper reqHandler shutdown<commit_after>{-# LANGUAGE DeriveDataTypeable, RecordWildCards, FlexibleContexts #-}
--------------------------------------------------------------------------------
{- |
  Module      :  Data.Acid.CenteredSlave.hs
  Copyright   :  ?

  Maintainer  :  <EMAIL>
  Portability :  non-portable (uses GHC extensions)

  The Slave part of a the Centered replication backend for acid state.

-}

--------------------------------------------------------------------------------
-- SLAVE part

module Data.Acid.Centered.Slave
    (
      enslaveState
    , enslaveStateFrom
    , enslaveRedState
    , enslaveRedStateFrom
    , SlaveState(..)
    )  where

import Data.Typeable
import Data.SafeCopy
import Data.Serialize (decode, encode, runPutLazy, runGetLazy)

import Data.Acid
import Data.Acid.Core
import Data.Acid.Abstract
import Data.Acid.Local
import Data.Acid.Log

import Data.Acid.Centered.Common

import System.ZMQ4 (Context, Socket, Dealer(..),
                    setReceiveHighWM, setSendHighWM, restrict,
                    poll, Poll(..), Event(..),
                    context, term, socket, close,
                    connect, disconnect, send, receive)
import System.FilePath ( (</>) )

import Control.Concurrent (forkIO, ThreadId, myThreadId, killThread, threadDelay, forkIOWithUnmask)
import Control.Concurrent.MVar (MVar, newMVar, newEmptyMVar,
                                withMVar, modifyMVar, modifyMVar_,
                                takeMVar, putMVar)
import Data.IORef (writeIORef)
import Control.Monad (void, when, unless, liftM)
import Control.Monad.STM (atomically)
import Control.Concurrent.STM.TVar (readTVar, writeTVar)
import qualified Control.Concurrent.Event as Event
import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
import Control.Exception (handle, throwTo, SomeException, ErrorCall(..))

import Data.IntMap (IntMap)
import qualified Data.IntMap as IM
import Data.Maybe (fromMaybe)

import qualified Data.ByteString.Lazy.Char8 as CSL

--------------------------------------------------------------------------------

-- | Slave state structure, for internal use.
data SlaveState st
    = SlaveState { slaveLocalState :: AcidState st
                 , slaveStateIsRed :: Bool
                 , slaveRepFinalizers :: MVar (IntMap (IO ()))
                 , slaveRepChan :: Chan SlaveRepItem
                 , slaveSyncDone :: Event.Event
                 , slaveRevision :: MVar NodeRevision
                 , slaveRequests :: MVar SlaveRequests
                 , slaveLastRequestID :: MVar RequestID
                 , slaveRepThreadId :: MVar ThreadId
                 , slaveReqThreadId :: MVar ThreadId
                 , slaveParentThreadId :: ThreadId
                 , slaveZmqContext :: Context
                 , slaveZmqAddr :: String
                 , slaveZmqSocket :: MVar (Socket Dealer)
                 } deriving (Typeable)

-- | Memory of own Requests sent to Master.
type SlaveRequests = IntMap (IO (IO ()),ThreadId)

-- | One Update + Metainformation to replicate.
data SlaveRepItem =
      SRIEnd
    | SRICheckpoint Revision
    | SRIArchive Revision
    | SRIUpdate Revision (Maybe RequestID) (Tagged CSL.ByteString)

-- | Open a local State as Slave for a Master.
--
-- The directory for the local state files is the default one ("state/[typeOf
-- state]").
enslaveState :: (IsAcidic st, Typeable st) =>
            String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveState address port initialState =
    enslaveStateFrom ("state" </> show (typeOf initialState)) address port initialState

-- | Open a local State as Slave for a Master.
--
-- The directory for the local state files is the default one ("state/[typeOf
-- state]").
enslaveRedState :: (IsAcidic st, Typeable st) =>
            String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveRedState address port initialState =
    enslaveRedStateFrom ("state" </> show (typeOf initialState)) address port initialState

-- | Open a local State as Slave for a Master. The directory of the local state
-- files can be specified.
enslaveStateFrom :: (IsAcidic st, Typeable st) =>
            FilePath        -- ^ location of the local state files.
         -> String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveStateFrom = enslaveMayRedStateFrom False

-- | Open a local State as Slave for a _redundant_ Master. The directory of the local state
-- files can be specified.
enslaveRedStateFrom :: (IsAcidic st, Typeable st) =>
            FilePath        -- ^ location of the local state files.
         -> String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveRedStateFrom = enslaveMayRedStateFrom True

-- | Open a local State as Slave for a Master, redundant or not.
--   The directory of the local state files can be specified.
enslaveMayRedStateFrom :: (IsAcidic st, Typeable st) =>
            Bool            -- ^ is redundant
         -> FilePath        -- ^ location of the local state files.
         -> String          -- ^ hostname of the Master
         -> PortNumber      -- ^ port to connect to
         -> st              -- ^ initial state
         -> IO (AcidState st)
enslaveMayRedStateFrom isRed directory address port initialState = do
        -- local
        lst <- openLocalStateFrom directory initialState
        let levs = localEvents $ downcast lst
        lrev <- atomically $ readTVar $ logNextEntryId levs
        rev <- newMVar lrev
        debug $ "Opening enslaved state at revision " ++ show lrev
        srs <- newMVar IM.empty
        lastReqId <- newMVar 0
        repChan <- newChan
        syncDone <- Event.new
        reqTid <- newEmptyMVar
        repTid <- newEmptyMVar
        parTid <- myThreadId
        repFin <- newMVar IM.empty
        -- remote
        let addr = "tcp://" ++ address ++ ":" ++ show port
        ctx <- context
        sock <- socket ctx Dealer
        setReceiveHighWM (restrict (100*1000 :: Int)) sock
        setSendHighWM (restrict (100*1000 :: Int)) sock
        connect sock addr
        msock <- newMVar sock
        sendToMaster msock $ NewSlave lrev
        let slaveState = SlaveState { slaveLocalState = lst
                                    , slaveStateIsRed = isRed
                                    , slaveRepFinalizers = repFin
                                    , slaveRepChan = repChan
                                    , slaveSyncDone = syncDone
                                    , slaveRevision = rev
                                    , slaveRequests = srs
                                    , slaveLastRequestID = lastReqId
                                    , slaveReqThreadId = reqTid
                                    , slaveRepThreadId = repTid
                                    , slaveParentThreadId = parTid
                                    , slaveZmqContext = ctx
                                    , slaveZmqAddr = addr
                                    , slaveZmqSocket = msock
                                    }
        void $ forkIOWithUnmask $ slaveRequestHandler slaveState
        void $ forkIO $ slaveReplicationHandler slaveState
        return $ slaveToAcidState slaveState

-- | Replication handler of the Slave.
slaveRequestHandler :: (IsAcidic st, Typeable st) => SlaveState st -> (IO () -> IO ()) -> IO ()
slaveRequestHandler slaveState@SlaveState{..} unmask = do
    mtid <- myThreadId
    putMVar slaveReqThreadId mtid
    let loop = handle (\e -> throwTo slaveParentThreadId (e :: SomeException)) $
            unmask $ handle killHandler $ do
            --waitRead =<< readMVar slaveZmqSocket
            -- FIXME: we needn't poll if not for strange zmq behaviour
            re <- withMVar slaveZmqSocket $ \sock -> poll 100 [Sock sock [In] Nothing]
            unless (null $ head re) $ do
                msg <- withMVar slaveZmqSocket receive
                case decode msg of
                    Left str -> error $ "Data.Serialize.decode failed on MasterMessage: " ++ show str
                    Right mmsg -> do
                         debug $ "Received: " ++ show mmsg
                         case mmsg of
                            -- We are sent an Update to replicate.
                            DoRep r i d -> queueRepItem slaveState (SRIUpdate r i d)
                            -- We are sent a Checkpoint for synchronization.
                            DoSyncCheckpoint r d -> replicateSyncCp slaveState r d
                            -- We are sent an Update to replicate for synchronization.
                            DoSyncRep r d -> replicateSyncUpdate slaveState r d
                            -- Master done sending all synchronization Updates.
{'fraction_non_alphanumeric': 0.101971707903217, 'fraction_numerical': 0.0021847178983013817, 'mean_word_length': 1.6172098341909662, 'pattern_counts': {'":': 2, '<': 49, '<?xml version=': 0, '>': 79, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 28, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7463847', 'n_tokens_mistral': 4780, 'n_tokens_neox': 4576, 'n_words': 1527}