{-# LANGUAGE NamedFieldPuns      #-}
{-# LANGUAGE RankNTypes          #-}
{-# LANGUAGE ScopedTypeVariables #-}

-- | Node-wide mux counters: how the muxes of one set have failed, counted
-- once per mux as it dies, and the bytes they sent and read, by
-- mini-protocol and direction; a remote set and a local one, traced
-- periodically.
--
module Network.Mux.Counters
  ( -- * Counters
    Counters
  , withCounters
  , readCounters
  , Counts (..)
  , Failures (..)
  , noFailures
  , noCounts
  , ByteCounts
    -- * Tracing
  , CountersTrace (..)
  , countersInterval
    -- * Internal, for the mux and the tests
  , newCounters
  , countersLoop
  , ByteSlots
  , withByteCells
  , newByteSlots
  , readSlots
  , countMuxerFailure
  , countDemuxerFailure
  , countMiniProtocolFailure
  , snapshotCountersWith
  ) where

import Control.Concurrent.Class.MonadSTM.Strict
import Control.Exception (SomeAsyncException, SomeException, fromException)
import Control.Monad (forM_, forever, unless, (<$!>))
import Control.Monad.Class.MonadAsync (MonadAsync, race)
import Control.Monad.Class.MonadFork (labelThisThread)
import Control.Monad.Class.MonadThrow (MonadMask, bracket)
import Control.Monad.Class.MonadTimer.SI
import Control.Tracer (Tracer, traceWith)

import Data.IntMap.Strict (IntMap)
import Data.IntMap.Strict qualified as IntMap
import Data.Map.Strict (Map)
import Data.Map.Strict qualified as Map
import Data.Maybe (isJust)
import Data.Void (absurd)
import Data.Word (Word64)
import Foreign.C.Error (Errno (..), eCONNRESET, ePIPE)
import GHC.IO.Exception (IOException (..))

import Network.Mux.Trace (Error (..))
import Network.Mux.Types (IngressQueue, IngressQueueState (..), MiniProtocolDir,
           MiniProtocolNum, RuntimeError)

-- | How the muxes have failed, each failure counted once, as the mux died
-- of it.
data Failures = Failures {
    -- The muxer.
    Failures -> Word64
mfWriteTimeouts  :: !Word64,
    -- ^ the peer took nothing within the SDU timeout
    Failures -> Word64
mfWriteErrors    :: !Word64,
    -- ^ the bearer's write failed otherwise: a timed-out connection, a
    -- host unreachable

    -- The demuxer.
    Failures -> Word64
mfReadTimeouts   :: !Word64,
    -- ^ an SDU header arrived, the rest did not in time
    Failures -> Word64
mfOverruns       :: !Word64,
    -- ^ the peer exceeded a mini-protocol's ingress limit
    Failures -> Word64
mfProtocolErrors :: !Word64,
    -- ^ undecodable SDUs, unknown mini-protocols, data an initiator-only
    -- mux cannot take
    Failures -> Word64
mfReadErrors     :: !Word64,
    -- ^ the bearer's read failed otherwise: a connection that timed out, a
    -- host unreachable

    -- Either bearer side.
    Failures -> Word64
mfPeerClosed     :: !Word64,
    -- ^ the peer closed the connection and the mux failed of it: read as
    -- its end while a mini-protocol was running, or met as a reset or a
    -- broken pipe by a read or a write
    Failures -> Word64
mfOtherErrors    :: !Word64,
    -- ^ the muxer or the demuxer died with an exception of a kind not listed,
    -- asynchronous ones aside: an encoding that would not force, a bug

    -- The mini-protocols.
    Failures -> Map (MiniProtocolNum, MiniProtocolDir) Word64
mfMiniProtocols  :: !(Map (MiniProtocolNum, MiniProtocolDir) Word64)
    -- ^ a mini-protocol's thread died with an exception of its own, by
    -- protocol and our side's direction: a violation it detected, its time
    -- limit
  }
  deriving (Failures -> Failures -> Bool
(Failures -> Failures -> Bool)
-> (Failures -> Failures -> Bool) -> Eq Failures
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Failures -> Failures -> Bool
== :: Failures -> Failures -> Bool
$c/= :: Failures -> Failures -> Bool
/= :: Failures -> Failures -> Bool
Eq, Int -> Failures -> ShowS
[Failures] -> ShowS
Failures -> String
(Int -> Failures -> ShowS)
-> (Failures -> String) -> ([Failures] -> ShowS) -> Show Failures
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Failures -> ShowS
showsPrec :: Int -> Failures -> ShowS
$cshow :: Failures -> String
show :: Failures -> String
$cshowList :: [Failures] -> ShowS
showList :: [Failures] -> ShowS
Show)

-- | No failure counted.
noFailures :: Failures
noFailures :: Failures
noFailures = Word64
-> Word64
-> Word64
-> Word64
-> Word64
-> Word64
-> Word64
-> Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Failures
Failures Word64
0 Word64
0 Word64
0 Word64
0 Word64
0 Word64
0 Word64
0 Word64
0 Map (MiniProtocolNum, MiniProtocolDir) Word64
forall k a. Map k a
Map.empty

-- | One side's counters at an instant.
data Counts = Counts {
    Counts -> Failures
mcFailures :: !Failures,
    Counts -> Map (MiniProtocolNum, MiniProtocolDir) Word64
mcSent     :: !ByteCounts,  -- ^ bytes written, by mini-protocol and direction
    Counts -> Map (MiniProtocolNum, MiniProtocolDir) Word64
mcReceived :: !ByteCounts   -- ^ bytes read, likewise
  }
  deriving (Counts -> Counts -> Bool
(Counts -> Counts -> Bool)
-> (Counts -> Counts -> Bool) -> Eq Counts
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: Counts -> Counts -> Bool
== :: Counts -> Counts -> Bool
$c/= :: Counts -> Counts -> Bool
/= :: Counts -> Counts -> Bool
Eq, Int -> Counts -> ShowS
[Counts] -> ShowS
Counts -> String
(Int -> Counts -> ShowS)
-> (Counts -> String) -> ([Counts] -> ShowS) -> Show Counts
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> Counts -> ShowS
showsPrec :: Int -> Counts -> ShowS
$cshow :: Counts -> String
show :: Counts -> String
$cshowList :: [Counts] -> ShowS
showList :: [Counts] -> ShowS
Show)

-- | Nothing counted.
noCounts :: Counts
noCounts :: Counts
noCounts = Failures
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Counts
Counts Failures
noFailures Map (MiniProtocolNum, MiniProtocolDir) Word64
forall k a. Map k a
Map.empty Map (MiniProtocolNum, MiniProtocolDir) Word64
forall k a. Map k a
Map.empty

-- | Bytes on the wire, SDU headers included, by mini-protocol number and the
-- direction of our side of it: what our initiators and our responders sent,
-- or what was read for them. The handshake's bytes, run on the bearer before
-- the mux, are not among them.
type ByteCounts = Map (MiniProtocolNum, MiniProtocolDir) Word64

-- | One mux's byte counters, one per mini-protocol and direction it runs,
-- written only by its own muxer or demuxer, so no two connections share a
-- variable on the data path; made by 'newByteSlots'.
type ByteSlots m = Map (MiniProtocolNum, MiniProtocolDir) (StrictTVar m Word64)

-- | What one mux sent and read: a slot per mini-protocol and direction the
-- muxer writes into, and the mini-protocols' ingress queues themselves,
-- each carrying what the demuxer read for its mini-protocol, counted in
-- the write that queued it so that a read costs nothing beyond that write.
-- The queues are only ever read here, for that count, never written.
data ByteCells m = ByteCells {
    forall (m :: * -> *). ByteCells m -> ByteSlots m
mbcSent :: !(ByteSlots m),
    -- ^ the muxer's slots
    forall (m :: * -> *).
ByteCells m
-> Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
mbcRecv :: !(Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m))
    -- ^ the ingress queues, read for their 'iqReceived'
  }

-- | The counts in the slots, leaving out the ones still at zero: a protocol
-- appears once it has carried something.
readSlots :: MonadSTM m => ByteSlots m -> STM m ByteCounts
readSlots :: forall (m :: * -> *).
MonadSTM m =>
ByteSlots m
-> STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64)
readSlots = (Map (MiniProtocolNum, MiniProtocolDir) Word64
 -> Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall a b. (a -> b) -> STM m a -> STM m b
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
fmap Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
nonZero (STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64)
 -> STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64))
-> (ByteSlots m
    -> STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64))
-> ByteSlots m
-> STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall b c a. (b -> c) -> (a -> b) -> a -> c
. (StrictTVar m Word64 -> STM m Word64)
-> ByteSlots m
-> STM m (Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b)
-> Map (MiniProtocolNum, MiniProtocolDir) a
-> f (Map (MiniProtocolNum, MiniProtocolDir) b)
traverse StrictTVar m Word64 -> STM m Word64
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar

-- | The counts that are not zero.
nonZero :: ByteCounts -> ByteCounts
nonZero :: Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
nonZero = (Word64 -> Bool)
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
forall a k. (a -> Bool) -> Map k a -> Map k a
Map.filter (Word64 -> Word64 -> Bool
forall a. Eq a => a -> a -> Bool
/= Word64
0)

-- | Counters shared by every mux of one set: the failures, and the byte
-- cells of the live muxes beside the totals of the ones that have ended, so
-- a total never goes down when a connection closes.
data Counters m = Counters {
    forall (m :: * -> *). Counters m -> StrictTVar m Failures
countersFailures :: !(StrictTVar m Failures),
    -- ^ how the muxes have failed, bumped by each as it dies
    forall (m :: * -> *).
Counters m -> StrictTVar m (IntMap (ByteCells m))
countersLive     :: !(StrictTVar m (IntMap (ByteCells m))),
    -- ^ the byte cells of the muxes running, each under the key it
    -- registered with
    forall (m :: * -> *). Counters m -> StrictTVar m Int
countersNext     :: !(StrictTVar m Int),
    -- ^ the key the next mux registers under, counted up by each
    forall (m :: * -> *). Counters m -> StrictTVar m Retired
countersRetired  :: !(StrictTVar m Retired)
    -- ^ sent and read by the muxes that have ended, summed in as each
    -- retired
  }

-- | What the muxes that have ended sent and read.
data Retired = Retired !ByteCounts !ByteCounts

-- | A fresh set, nothing counted.
newCounters :: MonadSTM m => m (Counters m)
newCounters :: forall (m :: * -> *). MonadSTM m => m (Counters m)
newCounters =
  StrictTVar m Failures
-> StrictTVar m (IntMap (ByteCells m))
-> StrictTVar m Int
-> StrictTVar m Retired
-> Counters m
forall (m :: * -> *).
StrictTVar m Failures
-> StrictTVar m (IntMap (ByteCells m))
-> StrictTVar m Int
-> StrictTVar m Retired
-> Counters m
Counters (StrictTVar m Failures
 -> StrictTVar m (IntMap (ByteCells m))
 -> StrictTVar m Int
 -> StrictTVar m Retired
 -> Counters m)
-> m (StrictTVar m Failures)
-> m (StrictTVar m (IntMap (ByteCells m))
      -> StrictTVar m Int -> StrictTVar m Retired -> Counters m)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Failures -> m (StrictTVar m Failures)
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
newTVarIO Failures
noFailures
              m (StrictTVar m (IntMap (ByteCells m))
   -> StrictTVar m Int -> StrictTVar m Retired -> Counters m)
-> m (StrictTVar m (IntMap (ByteCells m)))
-> m (StrictTVar m Int -> StrictTVar m Retired -> Counters m)
forall a b. m (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> IntMap (ByteCells m) -> m (StrictTVar m (IntMap (ByteCells m)))
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
newTVarIO IntMap (ByteCells m)
forall a. IntMap a
IntMap.empty
              m (StrictTVar m Int -> StrictTVar m Retired -> Counters m)
-> m (StrictTVar m Int) -> m (StrictTVar m Retired -> Counters m)
forall a b. m (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Int -> m (StrictTVar m Int)
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
newTVarIO Int
0
              m (StrictTVar m Retired -> Counters m)
-> m (StrictTVar m Retired) -> m (Counters m)
forall a b. m (a -> b) -> m a -> m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> Retired -> m (StrictTVar m Retired)
forall (m :: * -> *) a. MonadSTM m => a -> m (StrictTVar m a)
newTVarIO (Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64 -> Retired
Retired Map (MiniProtocolNum, MiniProtocolDir) Word64
forall k a. Map k a
Map.empty Map (MiniProtocolNum, MiniProtocolDir) Word64
forall k a. Map k a
Map.empty)

-- | The failures with the bytes of every mux, ended or live, summed in, in
-- one transaction: for tests and small sets, as a transaction over every
-- live mux's counters is restarted by any muxer or demuxer writing one it
-- has read, and reads them all over again; a node reads with the loop's
-- snapshot, a transaction per counter.
readCounters :: MonadSTM m => Counters m -> STM m Counts
readCounters :: forall (m :: * -> *). MonadSTM m => Counters m -> STM m Counts
readCounters = (forall a. STM m a -> STM m a) -> Counters m -> STM m Counts
forall (m :: * -> *) (n :: * -> *).
(MonadSTM m, Monad n) =>
(forall a. STM m a -> n a) -> Counters m -> n Counts
snapshotCountersWith STM m a -> STM m a
forall a. a -> a
forall a. STM m a -> STM m a
id

-- | The same, with each read run by @runTx@: in a transaction for the totals
-- and one per counter of each live mux, so no transaction spans more than
-- one variable a muxer or a demuxer writes, and a busy connection makes it
-- rerun at most the read of one. The totals and the live set go together,
-- before the cells: a mux that ends in between is in the live set and not
-- yet in the totals, and its cells, which ending leaves as they were, count
-- it once; a sum never goes down across snapshots.
snapshotCountersWith :: (MonadSTM m, Monad n)
                        => (forall a. STM m a -> n a)
                        -> Counters m
                        -> n Counts
snapshotCountersWith :: forall (m :: * -> *) (n :: * -> *).
(MonadSTM m, Monad n) =>
(forall a. STM m a -> n a) -> Counters m -> n Counts
snapshotCountersWith forall a. STM m a -> n a
runTx Counters { StrictTVar m Failures
countersFailures :: forall (m :: * -> *). Counters m -> StrictTVar m Failures
countersFailures :: StrictTVar m Failures
countersFailures, StrictTVar m (IntMap (ByteCells m))
countersLive :: forall (m :: * -> *).
Counters m -> StrictTVar m (IntMap (ByteCells m))
countersLive :: StrictTVar m (IntMap (ByteCells m))
countersLive, StrictTVar m Retired
countersRetired :: forall (m :: * -> *). Counters m -> StrictTVar m Retired
countersRetired :: StrictTVar m Retired
countersRetired } = do
  (failures, Retired sent received, live) <- STM m (Failures, Retired, [ByteCells m])
-> n (Failures, Retired, [ByteCells m])
forall a. STM m a -> n a
runTx (STM m (Failures, Retired, [ByteCells m])
 -> n (Failures, Retired, [ByteCells m]))
-> STM m (Failures, Retired, [ByteCells m])
-> n (Failures, Retired, [ByteCells m])
forall a b. (a -> b) -> a -> b
$
    (,,) (Failures
 -> Retired -> [ByteCells m] -> (Failures, Retired, [ByteCells m]))
-> STM m Failures
-> STM
     m (Retired -> [ByteCells m] -> (Failures, Retired, [ByteCells m]))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTVar m Failures -> STM m Failures
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m Failures
countersFailures
         STM
  m (Retired -> [ByteCells m] -> (Failures, Retired, [ByteCells m]))
-> STM m Retired
-> STM m ([ByteCells m] -> (Failures, Retired, [ByteCells m]))
forall a b. STM m (a -> b) -> STM m a -> STM m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> StrictTVar m Retired -> STM m Retired
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m Retired
countersRetired
         STM m ([ByteCells m] -> (Failures, Retired, [ByteCells m]))
-> STM m [ByteCells m] -> STM m (Failures, Retired, [ByteCells m])
forall a b. STM m (a -> b) -> STM m a -> STM m b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (IntMap (ByteCells m) -> [ByteCells m]
forall a. IntMap a -> [a]
IntMap.elems (IntMap (ByteCells m) -> [ByteCells m])
-> STM m (IntMap (ByteCells m)) -> STM m [ByteCells m]
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> StrictTVar m (IntMap (ByteCells m)) -> STM m (IntMap (ByteCells m))
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar StrictTVar m (IntMap (ByteCells m))
countersLive)
  cells <- mapM (readByteCellsWith runTx) live
  -- Built lazily: a tracer that drops the snapshot sums nothing, and one
  -- that renders it forces the whole, the fields being strict; the counts
  -- read above are forced already, so an unforced snapshot holds numbers,
  -- not queues.
  return Counts {
      mcFailures = failures,
      mcSent     = Map.unionsWith (+) (sent : map fst cells),
      mcReceived = Map.unionsWith (+) (received : map snd cells)
    }

-- | One mux's bytes sent and read, each counter read in a transaction of
-- its own by @runTx@: the slots, and the bytes read for each queue's
-- mini-protocol, leaving out the ones still at zero. A queue's count is
-- forced in its transaction, so a snapshot kept unforced holds the number,
-- not the queue.
readByteCellsWith :: (MonadSTM m, Monad n)
                 => (forall a. STM m a -> n a)
                 -> ByteCells m
                 -> n (ByteCounts, ByteCounts)
readByteCellsWith :: forall (m :: * -> *) (n :: * -> *).
(MonadSTM m, Monad n) =>
(forall a. STM m a -> n a)
-> ByteCells m
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64,
      Map (MiniProtocolNum, MiniProtocolDir) Word64)
readByteCellsWith forall a. STM m a -> n a
runTx ByteCells { ByteSlots m
mbcSent :: forall (m :: * -> *). ByteCells m -> ByteSlots m
mbcSent :: ByteSlots m
mbcSent, Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
mbcRecv :: forall (m :: * -> *).
ByteCells m
-> Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
mbcRecv :: Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
mbcRecv } =
  (,) (Map (MiniProtocolNum, MiniProtocolDir) Word64
 -> Map (MiniProtocolNum, MiniProtocolDir) Word64
 -> (Map (MiniProtocolNum, MiniProtocolDir) Word64,
     Map (MiniProtocolNum, MiniProtocolDir) Word64))
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64
      -> (Map (MiniProtocolNum, MiniProtocolDir) Word64,
          Map (MiniProtocolNum, MiniProtocolDir) Word64))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
nonZero (Map (MiniProtocolNum, MiniProtocolDir) Word64
 -> Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (StrictTVar m Word64 -> n Word64)
-> ByteSlots m -> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b)
-> Map (MiniProtocolNum, MiniProtocolDir) a
-> f (Map (MiniProtocolNum, MiniProtocolDir) b)
traverse (STM m Word64 -> n Word64
forall a. STM m a -> n a
runTx (STM m Word64 -> n Word64)
-> (StrictTVar m Word64 -> STM m Word64)
-> StrictTVar m Word64
-> n Word64
forall b c a. (b -> c) -> (a -> b) -> a -> c
. StrictTVar m Word64 -> STM m Word64
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar) ByteSlots m
mbcSent)
      n (Map (MiniProtocolNum, MiniProtocolDir) Word64
   -> (Map (MiniProtocolNum, MiniProtocolDir) Word64,
       Map (MiniProtocolNum, MiniProtocolDir) Word64))
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64,
      Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall a b. n (a -> b) -> n a -> n b
forall (f :: * -> *) a b. Applicative f => f (a -> b) -> f a -> f b
<*> (Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
nonZero (Map (MiniProtocolNum, MiniProtocolDir) Word64
 -> Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> (IngressQueue m -> n Word64)
-> Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall (t :: * -> *) (f :: * -> *) a b.
(Traversable t, Applicative f) =>
(a -> f b) -> t a -> f (t b)
forall (f :: * -> *) a b.
Applicative f =>
(a -> f b)
-> Map (MiniProtocolNum, MiniProtocolDir) a
-> f (Map (MiniProtocolNum, MiniProtocolDir) b)
traverse (\IngressQueue m
q -> STM m Word64 -> n Word64
forall a. STM m a -> n a
runTx (IngressQueueState -> Word64
iqReceived (IngressQueueState -> Word64)
-> STM m IngressQueueState -> STM m Word64
forall (m :: * -> *) a b. Monad m => (a -> b) -> m a -> m b
<$!> IngressQueue m -> STM m IngressQueueState
forall (m :: * -> *) a. MonadSTM m => StrictTVar m a -> STM m a
readTVar IngressQueue m
q)) Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
mbcRecv)

-- | Byte cells for a mux for the duration of @k@, which is given the muxer's
-- slots: one per mini-protocol and direction, beside the ingress queues
-- given, which count what the demuxer reads whether or not anyone looks.
-- With counters the slots are made and the cells registered as the mux
-- starts, and on the way out their counts move into the retired totals in
-- the same transaction that unregisters them; without, there are no slots,
-- so the muxer counts nothing.
withByteCells :: forall m a. (MonadSTM m, MonadMask m)
              => Maybe (Counters m)
              -> Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
              -> (ByteSlots m -> m a)
              -> m a
withByteCells :: forall (m :: * -> *) a.
(MonadSTM m, MonadMask m) =>
Maybe (Counters m)
-> Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
-> (ByteSlots m -> m a)
-> m a
withByteCells Maybe (Counters m)
counters Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
queues ByteSlots m -> m a
k = case Maybe (Counters m)
counters of
    Maybe (Counters m)
Nothing -> ByteSlots m -> m a
k ByteSlots m
forall k a. Map k a
Map.empty
    Just Counters m
c  -> do
      slots <- STM m (ByteSlots m) -> m (ByteSlots m)
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m (ByteSlots m) -> m (ByteSlots m))
-> STM m (ByteSlots m) -> m (ByteSlots m)
forall a b. (a -> b) -> a -> b
$ [(MiniProtocolNum, MiniProtocolDir)] -> STM m (ByteSlots m)
forall (m :: * -> *).
MonadSTM m =>
[(MiniProtocolNum, MiniProtocolDir)] -> STM m (ByteSlots m)
newByteSlots (Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
-> [(MiniProtocolNum, MiniProtocolDir)]
forall k a. Map k a -> [k]
Map.keys Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
queues)
      let cells = ByteCells { mbcSent :: ByteSlots m
mbcSent = ByteSlots m
slots, mbcRecv :: Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
mbcRecv = Map (MiniProtocolNum, MiniProtocolDir) (IngressQueue m)
queues }
      bracket (register c cells) (retire c cells) (const (k slots))
  where
    -- The cells among the live muxes' under the next key, which is returned.
    register :: Counters m -> ByteCells m -> m Int
    register :: Counters m -> ByteCells m -> m Int
register Counters { StrictTVar m (IntMap (ByteCells m))
countersLive :: forall (m :: * -> *).
Counters m -> StrictTVar m (IntMap (ByteCells m))
countersLive :: StrictTVar m (IntMap (ByteCells m))
countersLive, StrictTVar m Int
countersNext :: forall (m :: * -> *). Counters m -> StrictTVar m Int
countersNext :: StrictTVar m Int
countersNext } ByteCells m
cells = STM m Int -> m Int
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m Int -> m Int) -> STM m Int -> m Int
forall a b. (a -> b) -> a -> b
$ do
      i <- StrictTVar m Int -> (Int -> (Int, Int)) -> STM m Int
forall (m :: * -> *) s a.
MonadSTM m =>
StrictTVar m s -> (s -> (a, s)) -> STM m a
stateTVar StrictTVar m Int
countersNext (\Int
n -> (Int
n, Int
n Int -> Int -> Int
forall a. Num a => a -> a -> a
+ Int
1))
      modifyTVar countersLive (IntMap.insert i cells)
      return i

    -- The cells' counts into the retired totals and the cells out of the live
    -- set, in one transaction. The cells themselves are left as they were: a
    -- snapshot that read the live set before the mux ended reads them after,
    -- and counts the mux once. So the queues keep what the run read, and a
    -- 'Network.Mux.Mux' is run once, as its status has it.
    retire :: Counters m -> ByteCells m -> Int -> m ()
    retire :: Counters m -> ByteCells m -> Int -> m ()
retire Counters { StrictTVar m (IntMap (ByteCells m))
countersLive :: forall (m :: * -> *).
Counters m -> StrictTVar m (IntMap (ByteCells m))
countersLive :: StrictTVar m (IntMap (ByteCells m))
countersLive, StrictTVar m Retired
countersRetired :: forall (m :: * -> *). Counters m -> StrictTVar m Retired
countersRetired :: StrictTVar m Retired
countersRetired } ByteCells m
cells Int
i = STM m () -> m ()
forall a. HasCallStack => STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically (STM m () -> m ()) -> STM m () -> m ()
forall a b. (a -> b) -> a -> b
$ do
      (sent, received) <- (forall a. STM m a -> STM m a)
-> ByteCells m
-> STM
     m
     (Map (MiniProtocolNum, MiniProtocolDir) Word64,
      Map (MiniProtocolNum, MiniProtocolDir) Word64)
forall (m :: * -> *) (n :: * -> *).
(MonadSTM m, Monad n) =>
(forall a. STM m a -> n a)
-> ByteCells m
-> n (Map (MiniProtocolNum, MiniProtocolDir) Word64,
      Map (MiniProtocolNum, MiniProtocolDir) Word64)
readByteCellsWith STM m a -> STM m a
forall a. a -> a
forall a. STM m a -> STM m a
id ByteCells m
cells
      modifyTVar countersRetired $ \(Retired Map (MiniProtocolNum, MiniProtocolDir) Word64
s Map (MiniProtocolNum, MiniProtocolDir) Word64
r) ->
        Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64 -> Retired
Retired ((Word64 -> Word64 -> Word64)
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
forall k a. Ord k => (a -> a -> a) -> Map k a -> Map k a -> Map k a
Map.unionWith Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
(+) Map (MiniProtocolNum, MiniProtocolDir) Word64
s Map (MiniProtocolNum, MiniProtocolDir) Word64
sent) ((Word64 -> Word64 -> Word64)
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
-> Map (MiniProtocolNum, MiniProtocolDir) Word64
forall k a. Ord k => (a -> a -> a) -> Map k a -> Map k a -> Map k a
Map.unionWith Word64 -> Word64 -> Word64
forall a. Num a => a -> a -> a
(+) Map (MiniProtocolNum, MiniProtocolDir) Word64
r Map (MiniProtocolNum, MiniProtocolDir) Word64
received)
      modifyTVar countersLive (IntMap.delete i)

-- | A slot per mini-protocol and direction given, at zero. A mux makes its
-- muxer's for every mini-protocol it runs; an SDU for a key not among them
-- is not counted.
newByteSlots :: MonadSTM m => [(MiniProtocolNum, MiniProtocolDir)] -> STM m (ByteSlots m)
newByteSlots :: forall (m :: * -> *).
MonadSTM m =>
[(MiniProtocolNum, MiniProtocolDir)] -> STM m (ByteSlots m)
newByteSlots [(MiniProtocolNum, MiniProtocolDir)]
keys = [((MiniProtocolNum, MiniProtocolDir), StrictTVar m Word64)]
-> Map (MiniProtocolNum, MiniProtocolDir) (StrictTVar m Word64)
forall k a. Ord k => [(k, a)] -> Map k a
Map.fromList ([((MiniProtocolNum, MiniProtocolDir), StrictTVar m Word64)]
 -> Map (MiniProtocolNum, MiniProtocolDir) (StrictTVar m Word64))
-> STM
     m [((MiniProtocolNum, MiniProtocolDir), StrictTVar m Word64)]
-> STM
     m (Map (MiniProtocolNum, MiniProtocolDir) (StrictTVar m Word64))
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> [STM m ((MiniProtocolNum, MiniProtocolDir), StrictTVar m Word64)]
-> STM
     m [((MiniProtocolNum, MiniProtocolDir), StrictTVar m Word64)]
forall (t :: * -> *) (m :: * -> *) a.
(Traversable t, Monad m) =>
t (m a) -> m (t a)
forall (m :: * -> *) a. Monad m => [m a] -> m [a]
sequence [ (,) (MiniProtocolNum, MiniProtocolDir)
key (StrictTVar m Word64
 -> ((MiniProtocolNum, MiniProtocolDir), StrictTVar m Word64))
-> STM m (StrictTVar m Word64)
-> STM m ((MiniProtocolNum, MiniProtocolDir), StrictTVar m Word64)
forall (f :: * -> *) a b. Functor f => (a -> b) -> f a -> f b
<$> Word64 -> STM m (StrictTVar m Word64)
forall (m :: * -> *) a. MonadSTM m => a -> STM m (StrictTVar m a)
newTVar Word64
0 | (MiniProtocolNum, MiniProtocolDir)
key <- [(MiniProtocolNum, MiniProtocolDir)]
keys ]

-- | What a muxer's death counts as, if anything: the peer closing, which a
-- write meets as a reset or a broken pipe; an asynchronous exception is the
-- mux ending and counts nothing; a kind not listed counts as an error of
-- its own. The mux fails of every one but the asynchronous, so the counts
-- sum to the muxes dropped.
muxerFailure :: SomeException -> Maybe (Failures -> Failures)
muxerFailure :: SomeException -> Maybe (Failures -> Failures)
muxerFailure SomeException
e
  | SomeException -> Bool
isPeerClose SomeException
e = (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a. a -> Maybe a
Just Failures -> Failures
peerClosed
  | SomeException -> Bool
isIOError SomeException
e   = (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a. a -> Maybe a
Just (\Failures
f -> Failures
f { mfWriteErrors = mfWriteErrors f + 1 })
  | SomeException -> Bool
isAsync SomeException
e     = Maybe (Failures -> Failures)
forall a. Maybe a
Nothing
  | Bool
otherwise     = (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a. a -> Maybe a
Just ((Failures -> Failures) -> Maybe (Failures -> Failures))
-> (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a b. (a -> b) -> a -> b
$ case SomeException -> Maybe Error
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e of
      Just Error
SDUWriteTimeout -> \Failures
f -> Failures
f { mfWriteTimeouts = mfWriteTimeouts f + 1 }
      Maybe Error
_                    -> Failures -> Failures
otherError

-- | What a demuxer's death counts as, if anything: the peer closing, read as
-- the end of the stream or met as a reset; the rest as for the muxer. The
-- mux asks only when it fails of it: a close read after every
-- mini-protocol has finished ends it stopped, and counts nothing.
demuxerFailure :: SomeException -> Maybe (Failures -> Failures)
demuxerFailure :: SomeException -> Maybe (Failures -> Failures)
demuxerFailure SomeException
e
  | SomeException -> Bool
isPeerClose SomeException
e = (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a. a -> Maybe a
Just Failures -> Failures
peerClosed
  | SomeException -> Bool
isIOError SomeException
e   = (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a. a -> Maybe a
Just (\Failures
f -> Failures
f { mfReadErrors = mfReadErrors f + 1 })
  | SomeException -> Bool
isAsync SomeException
e     = Maybe (Failures -> Failures)
forall a. Maybe a
Nothing
  | Bool
otherwise     = (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a. a -> Maybe a
Just ((Failures -> Failures) -> Maybe (Failures -> Failures))
-> (Failures -> Failures) -> Maybe (Failures -> Failures)
forall a b. (a -> b) -> a -> b
$ case SomeException -> Maybe Error
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e of
      Just Error
SDUReadTimeout         -> \Failures
f -> Failures
f { mfReadTimeouts = mfReadTimeouts f + 1 }
      Just IngressQueueOverRun {} -> \Failures
f -> Failures
f { mfOverruns = mfOverruns f + 1 }
      Just UnknownMiniProtocol {} -> Failures -> Failures
protocolError
      Just InitiatorOnly {}       -> Failures -> Failures
protocolError
      Just SDUDecodeError {}      -> Failures -> Failures
protocolError
      Maybe Error
_                           -> Failures -> Failures
otherError
  where
    protocolError :: Failures -> Failures
    protocolError :: Failures -> Failures
protocolError Failures
f = Failures
f { mfProtocolErrors = mfProtocolErrors f + 1 }

-- | One more peer close.
peerClosed :: Failures -> Failures
peerClosed :: Failures -> Failures
peerClosed Failures
f = Failures
f { mfPeerClosed = mfPeerClosed f + 1 }

-- | One more death of a kind not listed.
otherError :: Failures -> Failures
otherError :: Failures -> Failures
otherError Failures
f = Failures
f { mfOtherErrors = mfOtherErrors f + 1 }

-- | A bearer's I/O error, wrapped in 'Error' as the socket bearer throws it,
-- or raw.
ioErrorOf :: SomeException -> Maybe IOException
ioErrorOf :: SomeException -> Maybe IOException
ioErrorOf SomeException
e = case SomeException -> Maybe Error
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e of
  Just (IOException IOException
ioe String
_) -> IOException -> Maybe IOException
forall a. a -> Maybe a
Just IOException
ioe
  Maybe Error
_                        -> SomeException -> Maybe IOException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e

-- | An I/O error, wrapped or raw.
isIOError :: SomeException -> Bool
isIOError :: SomeException -> Bool
isIOError = Maybe IOException -> Bool
forall a. Maybe a -> Bool
isJust (Maybe IOException -> Bool)
-> (SomeException -> Maybe IOException) -> SomeException -> Bool
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SomeException -> Maybe IOException
ioErrorOf

-- | The peer closed the connection: the bearer read its end, or an
-- operation found the connection reset or the pipe broken, which is how
-- a close arrives from a peer that closes without lingering, as the node
-- does. A socket's error is judged by its errno alone: one without, as the
-- simulation's channel throws for a connection gone or a fault it injects,
-- is an I/O error like any other, and so is a reset on Windows, which the
-- socket library reports under EINVAL, the errno it gives every Windows
-- code it has none for.
isPeerClose :: SomeException -> Bool
isPeerClose :: SomeException -> Bool
isPeerClose SomeException
e = case SomeException -> Maybe Error
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e of
  Just BearerClosed {} -> Bool
True
  Maybe Error
_                    -> Bool -> (IOException -> Bool) -> Maybe IOException -> Bool
forall b a. b -> (a -> b) -> Maybe a -> b
maybe Bool
False IOException -> Bool
peerGone (SomeException -> Maybe IOException
ioErrorOf SomeException
e)
  where
    peerGone :: IOException -> Bool
    peerGone :: IOException -> Bool
peerGone IOException
ioe = case IOException -> Maybe CInt
ioe_errno IOException
ioe of
      Just CInt
errno -> CInt -> Errno
Errno CInt
errno Errno -> [Errno] -> Bool
forall a. Eq a => a -> [a] -> Bool
forall (t :: * -> *) a. (Foldable t, Eq a) => a -> t a -> Bool
`elem` [Errno
eCONNRESET, Errno
ePIPE]
      Maybe CInt
Nothing    -> Bool
False

-- | An asynchronous exception: the mux ending, or the thread cancelled.
isAsync :: SomeException -> Bool
isAsync :: SomeException -> Bool
isAsync SomeException
e = Maybe SomeAsyncException -> Bool
forall a. Maybe a -> Bool
isJust (SomeException -> Maybe SomeAsyncException
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e :: Maybe SomeAsyncException)

-- | The mux's own errors, not a mini-protocol's. An 'Error' is counted by
-- the muxer or the demuxer it ended. A 'RuntimeError' is a start the mux
-- refuses, thrown to the caller's transaction, or a protocol's completion
-- var found full by its last transaction, which cannot be: neither ends a
-- muxer or a demuxer, so none is counted anywhere.
muxOwn :: SomeException -> Bool
muxOwn :: SomeException -> Bool
muxOwn SomeException
e = Maybe Error -> Bool
forall a. Maybe a -> Bool
isJust (SomeException -> Maybe Error
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e :: Maybe Error)
        Bool -> Bool -> Bool
|| Maybe RuntimeError -> Bool
forall a. Maybe a -> Bool
isJust (SomeException -> Maybe RuntimeError
forall e. Exception e => SomeException -> Maybe e
fromException SomeException
e :: Maybe RuntimeError)

-- | Count the exception a mux's muxer died with.
countMuxerFailure :: MonadSTM m => Maybe (Counters m) -> SomeException -> STM m ()
countMuxerFailure :: forall (m :: * -> *).
MonadSTM m =>
Maybe (Counters m) -> SomeException -> STM m ()
countMuxerFailure Maybe (Counters m)
counters = Maybe (Counters m) -> Maybe (Failures -> Failures) -> STM m ()
forall (m :: * -> *).
MonadSTM m =>
Maybe (Counters m) -> Maybe (Failures -> Failures) -> STM m ()
countFailure Maybe (Counters m)
counters (Maybe (Failures -> Failures) -> STM m ())
-> (SomeException -> Maybe (Failures -> Failures))
-> SomeException
-> STM m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SomeException -> Maybe (Failures -> Failures)
muxerFailure

-- | Count the exception a mux's demuxer died with, the mux failing of it.
countDemuxerFailure :: MonadSTM m => Maybe (Counters m) -> SomeException -> STM m ()
countDemuxerFailure :: forall (m :: * -> *).
MonadSTM m =>
Maybe (Counters m) -> SomeException -> STM m ()
countDemuxerFailure Maybe (Counters m)
counters = Maybe (Counters m) -> Maybe (Failures -> Failures) -> STM m ()
forall (m :: * -> *).
MonadSTM m =>
Maybe (Counters m) -> Maybe (Failures -> Failures) -> STM m ()
countFailure Maybe (Counters m)
counters (Maybe (Failures -> Failures) -> STM m ())
-> (SomeException -> Maybe (Failures -> Failures))
-> SomeException
-> STM m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. SomeException -> Maybe (Failures -> Failures)
demuxerFailure

-- | Count a failure, if there is one to count and a set to count it in.
countFailure :: MonadSTM m
             => Maybe (Counters m)
             -> Maybe (Failures -> Failures)
             -> STM m ()
countFailure :: forall (m :: * -> *).
MonadSTM m =>
Maybe (Counters m) -> Maybe (Failures -> Failures) -> STM m ()
countFailure Maybe (Counters m)
counters Maybe (Failures -> Failures)
failure =
  Maybe (Counters m) -> (Counters m -> STM m ()) -> STM m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Counters m)
counters ((Counters m -> STM m ()) -> STM m ())
-> (Counters m -> STM m ()) -> STM m ()
forall a b. (a -> b) -> a -> b
$ \Counters { StrictTVar m Failures
countersFailures :: forall (m :: * -> *). Counters m -> StrictTVar m Failures
countersFailures :: StrictTVar m Failures
countersFailures } ->
    Maybe (Failures -> Failures)
-> ((Failures -> Failures) -> STM m ()) -> STM m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Failures -> Failures)
failure (StrictTVar m Failures -> (Failures -> Failures) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m Failures
countersFailures)

-- | Count the exception a mini-protocol's thread died with, by its protocol
-- and our side's direction, unless it is the mux's own or asynchronous,
-- which is the mux ending.
countMiniProtocolFailure :: MonadSTM m
                         => Maybe (Counters m)
                         -> MiniProtocolNum
                         -> MiniProtocolDir
                         -> SomeException
                         -> STM m ()
countMiniProtocolFailure :: forall (m :: * -> *).
MonadSTM m =>
Maybe (Counters m)
-> MiniProtocolNum -> MiniProtocolDir -> SomeException -> STM m ()
countMiniProtocolFailure Maybe (Counters m)
counters MiniProtocolNum
num MiniProtocolDir
dir SomeException
e =
  Maybe (Counters m) -> (Counters m -> STM m ()) -> STM m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Counters m)
counters ((Counters m -> STM m ()) -> STM m ())
-> (Counters m -> STM m ()) -> STM m ()
forall a b. (a -> b) -> a -> b
$ \Counters { StrictTVar m Failures
countersFailures :: forall (m :: * -> *). Counters m -> StrictTVar m Failures
countersFailures :: StrictTVar m Failures
countersFailures } ->
    Bool -> STM m () -> STM m ()
forall (f :: * -> *). Applicative f => Bool -> f () -> f ()
unless (SomeException -> Bool
muxOwn SomeException
e Bool -> Bool -> Bool
|| SomeException -> Bool
isAsync SomeException
e) (STM m () -> STM m ()) -> STM m () -> STM m ()
forall a b. (a -> b) -> a -> b
$ StrictTVar m Failures -> (Failures -> Failures) -> STM m ()
forall (m :: * -> *) a.
MonadSTM m =>
StrictTVar m a -> (a -> a) -> STM m ()
modifyTVar StrictTVar m Failures
countersFailures ((Failures -> Failures) -> STM m ())
-> (Failures -> Failures) -> STM m ()
forall a b. (a -> b) -> a -> b
$ \Failures
f ->
      Failures
f { mfMiniProtocols = Map.insertWith (+) (num, dir) 1 (mfMiniProtocols f) }

-- | A snapshot of one set's counters.
data CountersTrace =
    TraceRemoteCounters Counts  -- ^ the remote muxes
  | TraceLocalCounters  Counts  -- ^ the local muxes
  deriving (CountersTrace -> CountersTrace -> Bool
(CountersTrace -> CountersTrace -> Bool)
-> (CountersTrace -> CountersTrace -> Bool) -> Eq CountersTrace
forall a. (a -> a -> Bool) -> (a -> a -> Bool) -> Eq a
$c== :: CountersTrace -> CountersTrace -> Bool
== :: CountersTrace -> CountersTrace -> Bool
$c/= :: CountersTrace -> CountersTrace -> Bool
/= :: CountersTrace -> CountersTrace -> Bool
Eq, Int -> CountersTrace -> ShowS
[CountersTrace] -> ShowS
CountersTrace -> String
(Int -> CountersTrace -> ShowS)
-> (CountersTrace -> String)
-> ([CountersTrace] -> ShowS)
-> Show CountersTrace
forall a.
(Int -> a -> ShowS) -> (a -> String) -> ([a] -> ShowS) -> Show a
$cshowsPrec :: Int -> CountersTrace -> ShowS
showsPrec :: Int -> CountersTrace -> ShowS
$cshow :: CountersTrace -> String
show :: CountersTrace -> String
$cshowList :: [CountersTrace] -> ShowS
showList :: [CountersTrace] -> ShowS
Show)

-- | A prime, so that snapshots do not keep step with other periodic work such
-- as keep-alive, and shorter than a 10 s scrape interval.
countersInterval :: DiffTime
countersInterval :: DiffTime
countersInterval = DiffTime
7

-- | Every @interval@, a second at least, snapshot the remote and the local
-- set, each if there is one, and trace each. 'withCounters' runs it for
-- the sets it makes.
countersLoop :: (MonadDelay m, MonadSTM m)
             => Maybe (Counters m)    -- ^ the remote set, if any
             -> Maybe (Counters m)    -- ^ the local set, if any
             -> DiffTime
             -> Tracer m CountersTrace
             -> m void
countersLoop :: forall (m :: * -> *) void.
(MonadDelay m, MonadSTM m) =>
Maybe (Counters m)
-> Maybe (Counters m)
-> DiffTime
-> Tracer m CountersTrace
-> m void
countersLoop Maybe (Counters m)
remote Maybe (Counters m)
local DiffTime
interval Tracer m CountersTrace
tracer = m () -> m void
forall (f :: * -> *) a b. Applicative f => f a -> f b
forever (m () -> m void) -> m () -> m void
forall a b. (a -> b) -> a -> b
$ do
  DiffTime -> m ()
forall (m :: * -> *). MonadDelay m => DiffTime -> m ()
threadDelay DiffTime
delay
  -- Small transactions, one for the totals and one per counter: the counters
  -- need no consistency across them, and one over everything the muxes write
  -- would be restarted by every batch.
  Maybe (Counters m) -> (Counters m -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Counters m)
remote ((Counters m -> m ()) -> m ()) -> (Counters m -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Counters m
c ->
    (forall a. STM m a -> m a) -> Counters m -> m Counts
forall (m :: * -> *) (n :: * -> *).
(MonadSTM m, Monad n) =>
(forall a. STM m a -> n a) -> Counters m -> n Counts
snapshotCountersWith STM m a -> m a
forall a. HasCallStack => STM m a -> m a
forall a. STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically Counters m
c m Counts -> (Counts -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Tracer m CountersTrace -> CountersTrace -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m CountersTrace
tracer (CountersTrace -> m ())
-> (Counts -> CountersTrace) -> Counts -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Counts -> CountersTrace
TraceRemoteCounters
  Maybe (Counters m) -> (Counters m -> m ()) -> m ()
forall (t :: * -> *) (m :: * -> *) a b.
(Foldable t, Monad m) =>
t a -> (a -> m b) -> m ()
forM_ Maybe (Counters m)
local  ((Counters m -> m ()) -> m ()) -> (Counters m -> m ()) -> m ()
forall a b. (a -> b) -> a -> b
$ \Counters m
c ->
    (forall a. STM m a -> m a) -> Counters m -> m Counts
forall (m :: * -> *) (n :: * -> *).
(MonadSTM m, Monad n) =>
(forall a. STM m a -> n a) -> Counters m -> n Counts
snapshotCountersWith STM m a -> m a
forall a. HasCallStack => STM m a -> m a
forall a. STM m a -> m a
forall (m :: * -> *) a.
(MonadSTM m, HasCallStack) =>
STM m a -> m a
atomically Counters m
c m Counts -> (Counts -> m ()) -> m ()
forall a b. m a -> (a -> m b) -> m b
forall (m :: * -> *) a b. Monad m => m a -> (a -> m b) -> m b
>>= Tracer m CountersTrace -> CountersTrace -> m ()
forall (m :: * -> *) a. Monad m => Tracer m a -> a -> m ()
traceWith Tracer m CountersTrace
tracer (CountersTrace -> m ())
-> (Counts -> CountersTrace) -> Counts -> m ()
forall b c a. (b -> c) -> (a -> b) -> a -> c
. Counts -> CountersTrace
TraceLocalCounters
  where
    -- A second at least: a zero or a negative interval would make the loop
    -- spin.
    delay :: DiffTime
    delay :: DiffTime
delay = DiffTime -> DiffTime -> DiffTime
forall a. Ord a => a -> a -> a
max DiffTime
1 DiffTime
interval

-- | Counter sets for as long as @k@ runs: one for the remote muxes and, if
-- asked, one for the local muxes, each traced every @interval@, a second at
-- least, by 'countersLoop' raced with @k@, so an exception in either ends
-- both. An interval of zero, or less, makes no set: the muxes count nothing
-- and nothing is traced. The only way to a set from "Network.Mux", so none
-- is kept without the loop that reads it; the internals below make one for
-- the mux's tests.
withCounters :: (MonadAsync m, MonadDelay m)
                => DiffTime
                -> Bool                 -- ^ a local set too
                -> Tracer m CountersTrace
                -> (Maybe (Counters m) -> Maybe (Counters m) -> m a)
                -> m a
withCounters :: forall (m :: * -> *) a.
(MonadAsync m, MonadDelay m) =>
DiffTime
-> Bool
-> Tracer m CountersTrace
-> (Maybe (Counters m) -> Maybe (Counters m) -> m a)
-> m a
withCounters DiffTime
interval Bool
localToo Tracer m CountersTrace
tracer Maybe (Counters m) -> Maybe (Counters m) -> m a
k
  | DiffTime
interval DiffTime -> DiffTime -> Bool
forall a. Ord a => a -> a -> Bool
<= DiffTime
0 = Maybe (Counters m) -> Maybe (Counters m) -> m a
k Maybe (Counters m)
forall a. Maybe a
Nothing Maybe (Counters m)
forall a. Maybe a
Nothing
  | Bool
otherwise = do
      remote <- m (Counters m)
forall (m :: * -> *). MonadSTM m => m (Counters m)
newCounters
      local  <- if localToo then Just <$> newCounters else return Nothing
      either absurd id <$>
        race (labelThisThread "mux-counters" >> countersLoop (Just remote) local interval tracer)
             (k (Just remote) local)