diff --git a/.github/workflows/applications.yml b/.github/workflows/applications.yml index e017571e50..8f553999d3 100644 --- a/.github/workflows/applications.yml +++ b/.github/workflows/applications.yml @@ -477,7 +477,7 @@ jobs: ghc: ${{ matrix.ghc }} use-freeze-file: ${{ matrix.use-freeze-file }} - name: Run compaction-tests - timeout-minutes: 7 + timeout-minutes: 10 run: | ulimit -n 10000 compaction-tests @@ -505,7 +505,7 @@ jobs: ghc: ${{ matrix.ghc }} use-freeze-file: ${{ matrix.use-freeze-file }} - name: Run multi-node-network-tests - timeout-minutes: 10 + timeout-minutes: 15 run: | ulimit -n 10000 multi-node-network-tests diff --git a/src/Chainweb/BlockWeight.hs b/src/Chainweb/BlockWeight.hs index 59604a3085..6094748c12 100644 --- a/src/Chainweb/BlockWeight.hs +++ b/src/Chainweb/BlockWeight.hs @@ -57,7 +57,7 @@ newtype BlockWeight = BlockWeight HashDifficulty ( Hashable , ToJSON, FromJSON, ToJSONKey, FromJSONKey , AdditiveSemigroup, AdditiveAbelianSemigroup - , Num + , Num, Bounded ) instance MerkleHashAlgorithm a => IsMerkleLogEntry a ChainwebHashTag BlockWeight where diff --git a/src/Chainweb/Chainweb/MinerResources.hs b/src/Chainweb/Chainweb/MinerResources.hs index af646a56bf..5cb9a76ebc 100644 --- a/src/Chainweb/Chainweb/MinerResources.hs +++ b/src/Chainweb/Chainweb/MinerResources.hs @@ -105,24 +105,23 @@ withMiningCoordination logger conf cdb inner c503 <- newIORef 0 c403 <- newIORef 0 l <- newIORef (_coordinationUpdateStreamLimit coordConf) - fmap thd . runConcurrently $ (,,) - <$> Concurrently (prune t m c503 c403) - <*> Concurrently (mapConcurrently_ (primeWork m) cids) - <*> Concurrently (inner . Just $ MiningCoordination - { _coordLogger = logger - , _coordCutDb = cdb - , _coordState = t - , _coordLimit = _coordinationReqLimit coordConf - , _coord503s = c503 - , _coord403s = c403 - , _coordConf = coordConf - , _coordUpdateStreamCount = l - , _coordPrimedWork = m - , _coordTargetFork = - if _coordinationTargetForkOverride coordConf - then pred $ max 1 (_versionForkNumber v) - else _versionForkNumber v - }) + withAsync (prune t m c503 c403) (\_ -> do + withAsync (mapConcurrently_ (primeWork m) cids) (\_ -> do + inner (Just MiningCoordination + { _coordLogger = logger + , _coordCutDb = cdb + , _coordState = t + , _coordLimit = _coordinationReqLimit coordConf + , _coord503s = c503 + , _coord403s = c403 + , _coordConf = coordConf + , _coordUpdateStreamCount = l + , _coordPrimedWork = m + , _coordTargetFork = + if _coordinationTargetForkOverride coordConf + then pred $ max 1 (_versionForkNumber v) + else _versionForkNumber v + }))) where coordConf = _miningCoordination conf inNodeConf = _miningInNode conf diff --git a/src/Chainweb/CutDB.hs b/src/Chainweb/CutDB.hs index 9960da5921..8bc8a8e390 100644 --- a/src/Chainweb/CutDB.hs +++ b/src/Chainweb/CutDB.hs @@ -241,15 +241,6 @@ cutHashesTable rdb = Casify $ newTable rdb valueCodec keyCodec ["CutHashes"] (runGetS $ (,,) <$> decodeCutHeightBe <*> decodeBlockWeightBe <*> decodeCutId) valueCodec = Codec encodeToByteString decodeStrictOrThrow' --- -------------------------------------------------------------------------- -- --- Exceptions - -data CutDbStopped = CutDbStopped - deriving (Eq, Show, Generic) - -instance Exception CutDbStopped where - fromException = asyncExceptionFromException - toException = asyncExceptionToException -- -------------------------------------------------------------------------- -- -- Cut DB @@ -518,10 +509,12 @@ fastForwardCutDb cutDb = do -- stopCutDb :: CutDb tbl -> IO () stopCutDb db = do + pQueueEnd (_cutDbQueue db) + void $ waitCatch (_cutDbAsync db) + currentCut <- readTVarIO (_cutDbCut db) unless (_cutDbReadOnly db) $ casInsert (_cutDbCutStore db) (cutToCutHashes Nothing currentCut) - cancelWith (_cutDbAsync db) CutDbStopped -- | Lookup the BlockHeaders for a CutHashes structure. Throws an exception if -- the lookup for some BlockHash in the input CutHashes. diff --git a/src/Data/PQueue.hs b/src/Data/PQueue.hs index fd0de8cd32..e4a95f4e1d 100644 --- a/src/Data/PQueue.hs +++ b/src/Data/PQueue.hs @@ -1,6 +1,8 @@ {-# LANGUAGE BangPatterns #-} {-# LANGUAGE ExistentialQuantification #-} {-# LANGUAGE ScopedTypeVariables #-} +{-# LANGUAGE LambdaCase #-} + -- | -- Module: Data.PQueue @@ -18,10 +20,14 @@ module Data.PQueue , pQueueRemove , pQueueIsEmpty , pQueueSize +, pQueueEnd +, pQueueEndNice ) where import Control.Concurrent.STM +import Control.Concurrent.Async (AsyncCancelled(..)) import Control.Monad +import Control.Exception import Data.Ord import qualified Data.Map as M @@ -38,11 +44,14 @@ import Numeric.Natural -- items in the queue. An item of low priority my starve in the queue if higher -- priority items are added at a rate at least as high as items are removed. -- + +data PQueueElement a = PQueueEOF | PQueueData a + data PQueue a = - forall p k. (Ord p, Ord k) => - PQueue (TVar (M.Map (Down p, k) a)) (TVar (S.Set k)) (a -> p) (a -> k) (Maybe Natural) + forall p k. (Ord p, Bounded p, Ord k, Bounded k) => + PQueue (TVar (M.Map (Down p, k) (PQueueElement a))) (TVar (S.Set k)) (a -> p) (a -> k) (Maybe Natural) -newEmptyPQueue :: (Ord p, Ord k) => (a -> p) -> (a -> k) -> Maybe Natural -> IO (PQueue a) +newEmptyPQueue :: (Ord p, Bounded p, Ord k, Bounded k) => (a -> p) -> (a -> k) -> Maybe Natural -> IO (PQueue a) newEmptyPQueue getPrio getKey maybeMaxLen = PQueue <$> newTVarIO mempty <*> newTVarIO mempty @@ -60,7 +69,7 @@ pQueueInsert (PQueue mv sv getPrio getKey maybeMaxLen) a = then return () else do let s' = S.insert k s - let m' = M.insert (Down $ getPrio a, k) a m + let m' = M.insert (Down $ getPrio a, k) (PQueueData a) m let fixup (maxlen :: Natural) = if M.size m' > fromIntegral (2 * maxlen) then let (keep, dontkeep) = M.splitAt (fromIntegral maxlen) m' in (foldl' (flip (S.delete . snd)) s' (M.keys dontkeep), keep) @@ -75,10 +84,21 @@ pQueueIsEmpty (PQueue mv _ _ _ _) = M.null <$!> readTVarIO mv pQueueSize :: PQueue a -> IO Natural pQueueSize (PQueue mv _ _ _ _) = fromIntegral . M.size <$!> readTVarIO mv +-- End the queue by pushing a top priority EOF +pQueueEnd :: PQueue a -> IO () +pQueueEnd (PQueue mv _ _ _ _) = atomically $ modifyTVar mv $ M.insert (Down maxBound, minBound) PQueueEOF + +-- End the queue by pushing a Lowest priority EOF +pQueueEndNice :: PQueue a -> IO () +pQueueEndNice (PQueue mv _ _ _ _) = atomically $ modifyTVar mv $ M.insert (Down minBound, maxBound) PQueueEOF + -- | If the queue is empty it blocks and races for new items -- pQueueRemove :: PQueue a -> IO a -pQueueRemove (PQueue mv sv _getPrio _getKey _) = atomically run +pQueueRemove (PQueue mv sv _getPrio _getKey _) = + atomically run >>= \case + PQueueEOF -> throwIO AsyncCancelled + PQueueData a -> return a where run = do m <- readTVar mv diff --git a/src/P2P/TaskQueue.hs b/src/P2P/TaskQueue.hs index d7ad02dae4..6a3da536e1 100644 --- a/src/P2P/TaskQueue.hs +++ b/src/P2P/TaskQueue.hs @@ -91,9 +91,14 @@ newtype TaskId = TaskId T.Text deriving (Show, Eq, Ord, Generic) deriving newtype (IsString) +instance Bounded TaskId where + minBound = TaskId T.empty + maxBound = TaskId $ T.singleton maxBound + newtype Priority = Priority Int deriving (Show, Eq, Ord, Generic) deriving anyclass (NFData, Hashable) + deriving newtype (Bounded) newtype AttemptsCount = AttemptsCount Natural deriving (Show, Eq, Ord, Generic) diff --git a/test/lib/Chainweb/Test/MultiNode.hs b/test/lib/Chainweb/Test/MultiNode.hs index 5c77e78034..8037d2f504 100644 --- a/test/lib/Chainweb/Test/MultiNode.hs +++ b/test/lib/Chainweb/Test/MultiNode.hs @@ -63,6 +63,7 @@ import Data.ByteString.Base16 qualified as Base16 import Chainweb.Pact.Backend.PactState.EmbeddedSnapshot (Snapshot(..)) import Data.Aeson (ToJSON) import Data.Foldable +import Data.Maybe import Data.Hashable import qualified Data.HashMap.Strict as HM import qualified Data.HashSet as HS @@ -87,7 +88,6 @@ import System.Directory (createDirectoryIfMissing) import System.FilePath import System.IO.Temp import System.LogLevel -import System.Timeout import Test.Tasty.HUnit @@ -245,18 +245,30 @@ harvestConsensusState _ _ _ (Replayed _ _) = error "harvestConsensusState: doesn't work when replaying, replays don't do consensus" harvestConsensusState logger stateVar nid (StartedChainweb cw) = do runChainweb cw (\_ -> return ()) `finally` do - logFunctionText logger Info "write sample data" + logFunctionText logger' Info "Node main threads ended" + + -- At this point, Warp/Servant server is supposed to be closed. + -- But Warp doesn't kill existing connections. As such, other nodes + -- can continue to push new Cuts, despite we would like to freeze a final Cut. + + -- A workaround is to early stop the CutDB, and not wait for the node to do it + -- naturally during unwinding. + stopCutDb (cw ^. chainwebCutResources . cutsCutDb) + + logFunctionText logger' Info "write sample data" modifyMVar_ stateVar $ sampleConsensusState nid (view (chainwebCutResources . cutsCutDb . cutDbWebBlockHeaderDb) cw) (view (chainwebCutResources . cutsCutDb) cw) - logFunctionText logger Info "shutdown node" + logFunctionText logger' Info "shutdown node" + where + logger' = addLabel ("node", toText nid) logger multiNode :: LogLevel -> (T.Text -> IO ()) - -> MVar PeerInfo + -> MVar (Maybe PeerInfo) -> ChainwebConfiguration -> RocksDb -> FilePath @@ -269,9 +281,9 @@ multiNode loglevel write bootstrapPeerInfoVar conf rdb pactDbDir nid inner = do withChainweb conf logger namespacedNodeRocksDb (pactDbDir show nid) backupTmpDir False $ \cw -> do case cw of StartedChainweb cw' -> - when (nid == bootstrapNodeId) $ putMVar bootstrapPeerInfoVar + when (nid == bootstrapNodeId) $ putMVar bootstrapPeerInfoVar $ Just $ view (chainwebPeer . peerResPeer . peerInfo) cw' - Replayed _ _ -> return () + Replayed _ _ -> when (nid == bootstrapNodeId) $ putMVar bootstrapPeerInfoVar Nothing inner nid cw where logger :: GenericLogger @@ -313,7 +325,7 @@ runNodes loglevel write v confBuilders rdb pactDbDir inner = do | i == 0 -> return $ multiBootstrapConfig baseConf | otherwise -> - setBootstrapPeerInfo <$> readMVar bootstrapPortVar <*> pure baseConf + maybe baseConf (`setBootstrapPeerInfo` baseConf) <$> readMVar bootstrapPortVar multiNode loglevel write bootstrapPortVar (confBuilder conf) rdb pactDbDir (NodeId i) inner @@ -330,9 +342,11 @@ runNodesForSeconds -> FilePath -> (forall logger. NodeId -> StartedChainweb logger -> IO ()) -> IO () -runNodesForSeconds loglevel write v confBuilders (Seconds seconds) rdb pactDbDir inner = do - void $ timeout (int seconds * 1_000_000) - $ runNodes loglevel write v confBuilders rdb pactDbDir inner +runNodesForSeconds loglevel write v confBuilders (Seconds seconds) rdb pactDbDir inner = + runNodes loglevel write v confBuilders rdb pactDbDir innerWithTimeout + where + innerWithTimeout:: NodeId -> StartedChainweb a -> IO() + innerWithTimeout nid cw = void $ race (inner nid cw) $ threadDelay (int seconds * 1_000_000) -- | Ensure that we can compact a live node(s). -- @@ -617,19 +631,18 @@ replayTest loglevel v n rdb pactDbDir step = do tastylog $ "phase 3... replaying" let replayInitialHeight = 5 firstReplayCompleteRef <- newIORef False - runNodesForSeconds loglevel logFun v + runNodes loglevel logFun v (replicate n $ multiConfig n & mapped . configCuts . cutInitialBlockHeightLimit .~ Just replayInitialHeight & mapped . configOnlySyncPact .~ True) - (Seconds 20) rdb pactDbDir $ \nid cw -> case cw of + rdb pactDbDir $ \nid cw -> case cw of Replayed l (Just u) -> do writeIORef firstReplayCompleteRef True _ <- flip HM.traverseWithKey (_cutMap l) $ \cid bh -> assertEqual ("lower chain " <> sshow cid) replayInitialHeight (view blockHeight bh) - -- TODO: this is flaky, presumably because a node's cutdb - -- is not being cancelled synchronously enough + assertEqual "upper cut" (_stateCutMap state2 HM.! nid) u _ <- flip HM.traverseWithKey (_cutMap u) $ \cid bh -> assertGe ("upper chain " <> sshow cid) (Actual $ view blockHeight bh) (Expected replayInitialHeight) @@ -640,7 +653,7 @@ replayTest loglevel v n rdb pactDbDir step = do let fastForwardHeight = 10 tastylog $ "phase 4... replaying with fast-forward limit" secondReplayCompleteRef <- newIORef False - runNodesForSeconds loglevel logFun v + runNodes loglevel logFun v (replicate n $ multiConfig n & mapped . configCuts . cutInitialBlockHeightLimit @@ -649,7 +662,7 @@ replayTest loglevel v n rdb pactDbDir step = do .~ Just fastForwardHeight & mapped . configOnlySyncPact .~ True ) - (Seconds 20) rdb pactDbDir $ \_ cw -> case cw of + rdb pactDbDir $ \_ cw -> case cw of Replayed l (Just u) -> do writeIORef secondReplayCompleteRef True _ <- flip HM.traverseWithKey (_cutMap l) $ \cid bh ->