# LANGUAGE BangPatterns           #
{-# LANGUAGE FlexibleContexts      #-}
# LANGUAGE FlexibleInstances      #
{-# LANGUAGE LambdaCase            #-}
# LANGUAGE MultiParamTypeClasses #
{-# LANGUAGE RankNTypes            #-}
# LANGUAGE RecordWildCards        #
{-# LANGUAGE ScopedTypeVariables   #-}
# LANGUAGE TypeApplications       #

module Main where

import           Control.Applicative (empty, liftA2)
import           Control.Concurrent (threadDelay)
import           Control.Concurrent.Async (concurrently, forConcurrently)
import           Control.Exception (throwIO)
import           Control.Exception.Safe (throwString)
import           Control.Monad (forM, forM_)

import           Data.Foldable (foldlM)
import           Data.Functor.Contravariant (contramap)
import           Data.Time.Clock.POSIX (getPOSIXTime)
import           Data.Time.Units (Microsecond)
import           GHC.IO.Encoding (setLocaleEncoding, utf8)
import qualified Network.Transport.TCP as TCP
import qualified Network.Transport.TCP.Internal as TCP (encodeEndPointAddress)
import           Options.Applicative.Simple (simpleOptions)
import           System.Random (mkStdGen, randomR)

import qualified Network.Transport as NT
import           Node (Conversation (..), ConversationActions (..), Node (Node),
                     NodeAction (..), converseWith, defaultNodeEnvironment,
                     noReceiveDelay, node, simpleNodeEndPoint)
import           Node.Internal (NodeId (..))
import           Node.Message.Binary (binaryPacking)
import           Pos.Util.Trace (Severity (..), wlogTrace)
import           Pos.Util.Wlog (removeAllHandlers)

import           Bench.Network.Commons (MeasureEvent (..), Payload (..),
                     Ping (..), Pong (..), loadLogConfig, logMeasure)
import           SenderOptions (Args (..), argsParser)

data PingState = PingState
    { _lastResetMcs    :: !Microsecond
    , _currentMessages :: !Word
    }

oneSecondMcs :: Int
oneSecondMcs = 1000000

main :: IO ()
main = do
    (Args {..}, ()) <-
        simpleOptions
            "bench-sender"
            "Sender utility for benches"
            "Use it!"
            argsParser
            empty

    lh <- loadLogConfig logsPrefix logConfig
    setLocaleEncoding utf8

    transport <- do
        transportOrError <-
            TCP.createTransport (TCP.defaultTCPAddr "127.0.0.1" "3432")
            TCP.defaultTCPParameters
        either throwIO return transportOrError

    let prngNode = mkStdGen 0
        prngWork = mkStdGen 1
        nodeIds  = [ NodeId $ TCP.encodeEndPointAddress host (show port) 0
                   | (host, port) <- recipients ]
        tasksIds = [[tid, tid + threadNum .. msgNum] | tid <- [1..threadNum]]

        action :: IO ()
        action = do
            startTime <- round . (* 1000000) <$> getPOSIXTime

            -- TODO: is it good idea to start (recipients number * thread number) threads?
            let pingWorkers = liftA2 (pingSender prngWork payloadBound startTime msgRate)
                                     tasksIds
                                     (zip [0, msgNum..] nodeIds)

            node logTrace (simpleNodeEndPoint transport) (const noReceiveDelay) (const noReceiveDelay) prngNode binaryPacking () defaultNodeEnvironment $ \node' ->
                NodeAction (const []) $ \converse -> () <$ do
                    drones <- forM nodeIds (startDrone node')
                    forConcurrently pingWorkers ($ converse) `concurrently` do
                        threadDelay (duration * 1000000)
                        forM_ drones stopDrone

    action
    removeAllHandlers lh

  where

    logTrace = wlogTrace "sender"
    logTrace' = contramap ((,) Info) logTrace

    pingSender gen payloadBound startTimeMcs msgRate msgIds (msgStartId, peerId) converse =
        foldlM (pingSenderOnce payloadBound msgRate msgStartId peerId converse)
               (gen, PingState startTimeMcs 0)
               msgIds

    pingSenderOnce payloadBound msgRate msgStartId peerId converse (gen, !pingState) msgId = do
        let sMsgId = msgStartId + msgId
            (i, gen') = randomR (0, payloadBound) gen
            payload = Payload i
        logMeasure logTrace' PingSent sMsgId payload
        converseWith converse peerId $ \_ -> Conversation $ \cactions -> do
            send cactions (Ping sMsgId payload)
            recv cactions maxBound >>= \case
                Just (Pong _ _) -> pure ()
                _ -> throwString "Expected a pong"
            pure ()
        curTime <- round . (* 1000000) <$> getPOSIXTime
        let fromLastReset = curTime - fromIntegral (_lastResetMcs pingState)
        pingState' <-
            if fromLastReset >= oneSecondMcs
            then pure (PingState (fromIntegral curTime) 0)
            else if _currentMessages pingState >= msgRate
                then do
                    let waitToNextSecond = oneSecondMcs - fromLastReset
                    threadDelay waitToNextSecond
                    freshTime <- round .(* 1000000) <$> getPOSIXTime
                    pure $ PingState freshTime 0
                else pure $ pingState { _currentMessages = _currentMessages pingState + 1 }
        pure (gen', pingState')

    startDrone
        :: Node
        -> NodeId
        -> IO NT.Connection
    startDrone (Node _ endPoint _) (NodeId peer) = do
        connOrErr <- NT.connect endPoint peer NT.ReliableOrdered (NT.ConnectHints Nothing)
        either throwIO return connOrErr

    stopDrone = NT.close

# LANGUAGE FlexibleContexts      #
# LANGUAGE LambdaCase            #
# LANGUAGE RankNTypes            #
# LANGUAGE ScopedTypeVariables   #
 TODO: is it good idea to start (recipients number * thread number) threads?
# LANGUAGE BangPatterns           #
# LANGUAGE FlexibleInstances      #
# LANGUAGE MultiParamTypeClasses #
# LANGUAGE RecordWildCards        #
# LANGUAGE TypeApplications       #

module Main where

import           Control.Applicative (empty, liftA2)
import           Control.Concurrent (threadDelay)
import           Control.Concurrent.Async (concurrently, forConcurrently)
import           Control.Exception (throwIO)
import           Control.Exception.Safe (throwString)
import           Control.Monad (forM, forM_)

import           Data.Foldable (foldlM)
import           Data.Functor.Contravariant (contramap)
import           Data.Time.Clock.POSIX (getPOSIXTime)
import           Data.Time.Units (Microsecond)
import           GHC.IO.Encoding (setLocaleEncoding, utf8)
import qualified Network.Transport.TCP as TCP
import qualified Network.Transport.TCP.Internal as TCP (encodeEndPointAddress)
import           Options.Applicative.Simple (simpleOptions)
import           System.Random (mkStdGen, randomR)

import qualified Network.Transport as NT
import           Node (Conversation (..), ConversationActions (..), Node (Node),
                     NodeAction (..), converseWith, defaultNodeEnvironment,
                     noReceiveDelay, node, simpleNodeEndPoint)
import           Node.Internal (NodeId (..))
import           Node.Message.Binary (binaryPacking)
import           Pos.Util.Trace (Severity (..), wlogTrace)
import           Pos.Util.Wlog (removeAllHandlers)

import           Bench.Network.Commons (MeasureEvent (..), Payload (..),
                     Ping (..), Pong (..), loadLogConfig, logMeasure)
import           SenderOptions (Args (..), argsParser)

data PingState = PingState
    { _lastResetMcs    :: !Microsecond
    , _currentMessages :: !Word
    }

oneSecondMcs :: Int
oneSecondMcs = 1000000

main :: IO ()
main = do
    (Args {..}, ()) <-
        simpleOptions
            "bench-sender"
            "Sender utility for benches"
            "Use it!"
            argsParser
            empty

    lh <- loadLogConfig logsPrefix logConfig
    setLocaleEncoding utf8

    transport <- do
        transportOrError <-
            TCP.createTransport (TCP.defaultTCPAddr "127.0.0.1" "3432")
            TCP.defaultTCPParameters
        either throwIO return transportOrError

    let prngNode = mkStdGen 0
        prngWork = mkStdGen 1
        nodeIds  = [ NodeId $ TCP.encodeEndPointAddress host (show port) 0
                   | (host, port) <- recipients ]
        tasksIds = [[tid, tid + threadNum .. msgNum] | tid <- [1..threadNum]]

        action :: IO ()
        action = do
            startTime <- round . (* 1000000) <$> getPOSIXTime

            let pingWorkers = liftA2 (pingSender prngWork payloadBound startTime msgRate)
                                     tasksIds
                                     (zip [0, msgNum..] nodeIds)

            node logTrace (simpleNodeEndPoint transport) (const noReceiveDelay) (const noReceiveDelay) prngNode binaryPacking () defaultNodeEnvironment $ \node' ->
                NodeAction (const []) $ \converse -> () <$ do
                    drones <- forM nodeIds (startDrone node')
                    forConcurrently pingWorkers ($ converse) `concurrently` do
                        threadDelay (duration * 1000000)
                        forM_ drones stopDrone

    action
    removeAllHandlers lh

  where

    logTrace = wlogTrace "sender"
    logTrace' = contramap ((,) Info) logTrace

    pingSender gen payloadBound startTimeMcs msgRate msgIds (msgStartId, peerId) converse =
        foldlM (pingSenderOnce payloadBound msgRate msgStartId peerId converse)
               (gen, PingState startTimeMcs 0)
               msgIds

    pingSenderOnce payloadBound msgRate msgStartId peerId converse (gen, !pingState) msgId = do
        let sMsgId = msgStartId + msgId
            (i, gen') = randomR (0, payloadBound) gen
            payload = Payload i
        logMeasure logTrace' PingSent sMsgId payload
        converseWith converse peerId $ \_ -> Conversation $ \cactions -> do
            send cactions (Ping sMsgId payload)
            recv cactions maxBound >>= \case
                Just (Pong _ _) -> pure ()
                _ -> throwString "Expected a pong"
            pure ()
        curTime <- round . (* 1000000) <$> getPOSIXTime
        let fromLastReset = curTime - fromIntegral (_lastResetMcs pingState)
        pingState' <-
            if fromLastReset >= oneSecondMcs
            then pure (PingState (fromIntegral curTime) 0)
            else if _currentMessages pingState >= msgRate
                then do
                    let waitToNextSecond = oneSecondMcs - fromLastReset
                    threadDelay waitToNextSecond
                    freshTime <- round .(* 1000000) <$> getPOSIXTime
                    pure $ PingState freshTime 0
                else pure $ pingState { _currentMessages = _currentMessages pingState + 1 }
        pure (gen', pingState')

    startDrone
        :: Node
        -> NodeId
        -> IO NT.Connection
    startDrone (Node _ endPoint _) (NodeId peer) = do
        connOrErr <- NT.connect endPoint peer NT.ReliableOrdered (NT.ConnectHints Nothing)
        either throwIO return connOrErr

    stopDrone = NT.close
