diff --git a/.github/workflows/applications.yml b/.github/workflows/applications.yml index e017571e50..db51cc37a7 100644 --- a/.github/workflows/applications.yml +++ b/.github/workflows/applications.yml @@ -606,7 +606,7 @@ jobs: timeout-minutes: 10 run: | ulimit -n 10000 - chainweb-tests --hide-successes --results-json test-results.json + chainweb-tests # Publish test results to S3 - name: Upload test results diff --git a/src/Chainweb/Chainweb/MinerResources.hs b/src/Chainweb/Chainweb/MinerResources.hs index af646a56bf..6dfa3abcc0 100644 --- a/src/Chainweb/Chainweb/MinerResources.hs +++ b/src/Chainweb/Chainweb/MinerResources.hs @@ -8,6 +8,7 @@ {-# LANGUAGE TupleSections #-} {-# LANGUAGE TypeApplications #-} {-# LANGUAGE TypeFamilies #-} +{-# LANGUAGE TemplateHaskell #-} -- | -- Module: Chainweb.Chainweb.MinerResources @@ -35,15 +36,16 @@ import Control.Concurrent (threadDelay) import Control.Concurrent.Async import Control.Concurrent.STM (atomically) import Control.Concurrent.STM.TVar +import Control.Exception (finally) import Control.Lens import Control.Monad import Data.HashMap.Strict (HashMap) +import Data.Coerce (coerce) import qualified Data.HashMap.Strict as HM import qualified Data.HashSet as HS import Data.IORef (IORef, atomicWriteIORef, newIORef, readIORef) import qualified Data.Map.Strict as M -import Data.Maybe import qualified Data.Set as S import qualified Data.Vector as V @@ -54,14 +56,14 @@ import qualified System.Random.MWC as MWC import Chainweb.BlockHeader import Chainweb.ChainId +import Chainweb.BlockHash import Chainweb.Chainweb.ChainResources -import Chainweb.Cut (_cutMap) -import Chainweb.CutDB (CutDb, awaitNewBlock, cutDbPactService, _cut) +import Chainweb.CutDB (CutDb, awaitNewBlock, cutDbPactService) import Chainweb.Logger import Chainweb.Miner.Config import Chainweb.Miner.Coordinator import Chainweb.Miner.Miners -import Chainweb.Miner.Pact (Miner(..), minerId) +import Chainweb.Miner.Pact (Miner(..), minerId, MinerId(..)) import Chainweb.Pact.Types import Chainweb.Pact.Utils import Chainweb.Payload @@ -80,6 +82,31 @@ import Numeric.AffineSpace -- -------------------------------------------------------------------------- -- -- Miner +data MinerLastSeen = NeverExpire | LastSeen (Time Micros) + -- ^ ^-- For Dynamic miners + -- \ For static mineres + deriving (Eq) + +data ActiveMiner = ActiveMiner + { _lastSeen :: MinerLastSeen + , _handles :: [Async ()] + } + +makeLenses ''ActiveMiner + +isMinerExpired:: Time Micros -> ActiveMiner -> Bool +isMinerExpired exp_t miner = case (miner ^. lastSeen) of + NeverExpire -> False + LastSeen t | t < exp_t -> True + LastSeen _ -> False + + +type ActiveMiners = HM.HashMap MinerId ActiveMiner + +-- Return all Handles from an ActiveMiners HashMap +allHandles :: ActiveMiners -> [Async ()] +allHandles = mconcat . fmap (^. handles) . HM.elems + withMiningCoordination :: Logger logger => logger @@ -90,25 +117,21 @@ withMiningCoordination withMiningCoordination logger conf cdb inner | not (_coordinationEnabled coordConf) = inner Nothing | otherwise = do - cut <- _cut cdb + t <- newTVarIO mempty - initialPw <- fmap (PrimedWork . HM.fromList) $ - forM miners $ \miner -> - let mid = view minerId miner - in fmap ((mid,) . HM.fromList) $ - forM cids $ \cid -> do - let bh = fromMaybe (genesisBlockHeader v cid) (HM.lookup cid (_cutMap cut)) - newBlock <- throwIfNoHistory =<< getPayload cid miner (ParentHeader bh) - return (cid, WorkReady newBlock) - - m <- newTVarIO initialPw + tpw <- newTVarIO (PrimedWork HM.empty) + tam <- newTVarIO HM.empty + + let miners = S.toList (_coordinationMiners coordConf) + <> [ _nodeMiner inNodeConf | _nodeMiningEnabled inNodeConf ] + + forM_ miners $ initMiner tam tpw + 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 + withAsync (prune t tam tpw c503 c403) $ \_ -> do + inner (Just $ MiningCoordination { _coordLogger = logger , _coordCutDb = cdb , _coordState = t @@ -117,12 +140,17 @@ withMiningCoordination logger conf cdb inner , _coord403s = c403 , _coordConf = coordConf , _coordUpdateStreamCount = l - , _coordPrimedWork = m + , _coordRefreshMiner = updateDynamicMiner tam tpw + , _coordPrimedWork = tpw , _coordTargetFork = if _coordinationTargetForkOverride coordConf then pred $ max 1 (_versionForkNumber v) else _versionForkNumber v }) + + `finally` do + am <- readTVarIO tam + cancelMany $ allHandles am where coordConf = _miningCoordination conf inNodeConf = _miningInNode conf @@ -131,11 +159,61 @@ withMiningCoordination logger conf cdb inner cids :: [ChainId] cids = HS.toList (chainIds v) - !miners = S.toList (_coordinationMiners coordConf) - <> [ _nodeMiner inNodeConf | _nodeMiningEnabled inNodeConf ] - chainLogger cid = addLabel ("chain", toText cid) + minerIdLogger:: Logger l => MinerId -> (l -> l) + minerIdLogger mid = addLabel ("miner", _minerId mid) + + updateDynamicMiner :: TVar ActiveMiners -> TVar PrimedWork -> Miner -> IO (Either String ()) + updateDynamicMiner tam tpm m = do + let mid = view minerId m + ct <- getCurrentTimeIntegral + isNewMiner <- atomically $ do + am <- readTVar tam + case HM.lookup mid am of + -- Static Miner + Just currentMiner | currentMiner ^. lastSeen == NeverExpire -> return (Right False) + + -- Dynamic miner already registered, update the time + Just _ -> do + modifyTVar tam (HM.adjust (lastSeen .~ LastSeen ct) mid) + return (Right False) + + -- Dynamic miners no atllowed + Nothing | not (_coordinationDynamicMinersEnabled coordConf) -> return ( Left "Unknown miners not allowed") + + -- Miner doesn't exist but we are full + Nothing | HM.size am > (_coordinationMinersLimit coordConf) -> return ( Left "Miners Pool Full") + + -- Miner doesn't exist but there is some room for it + Nothing -> do + -- Insert into active miners list + modifyTVar tam (HM.insert mid (ActiveMiner (LastSeen ct) [])) + -- And prepare a PrimeWork structure with evrything Stale, and let updatePrimeWork take care of it + modifyTVar tpm $ coerce $ HM.insert mid (HM.fromList $ fmap (\c -> (c, WorkStale)) cids) + return $ Right True + + -- In case the miner is new => Start the updatePrimeWork threads + forM isNewMiner $ \isNew -> + when isNew $ do + logFunctionText (minerIdLogger mid logger) Info "Added" + -- Create the threads + h <- mapM (async . updatePrimeWork tpm m) cids + -- And store their handles to cancel them in case the miner stops. + atomically $ modifyTVar tam (HM.adjust (handles .~ h) mid) + + -- Only used for static miners + -- A smplified version of updateDynamic miner + initMiner :: TVar ActiveMiners -> TVar PrimedWork -> Miner -> IO() + initMiner tam tpm m = do + let mid = view minerId m + atomically $ do + modifyTVar tam (HM.insert mid (ActiveMiner NeverExpire [])) + modifyTVar tpm $ coerce $ HM.insert mid (HM.fromList $ fmap (\c -> (c, WorkStale)) cids) + + h <- mapM (async . updatePrimeWork tpm m) cids + atomically $ modifyTVar tam (HM.adjust (handles .~ h) mid) + -- we assume that this path always exists in PrimedWork and never delete it. workForMiner :: Miner -> ChainId -> Traversal' PrimedWork WorkState workForMiner miner cid = _Wrapped' . ix (view minerId miner) . ix cid @@ -182,23 +260,25 @@ withMiningCoordination logger conf cdb inner -- that when they request new work, the block can be instantly constructed -- without interacting with the Pact Queue. -- - primeWork :: TVar PrimedWork -> ChainId -> IO () - primeWork tpw cid = - forConcurrently_ miners $ \miner -> - runForever (logFunction (chainLogger cid logger)) "primeWork" (go miner) + updatePrimeWork :: TVar PrimedWork -> Miner -> ChainId -> IO () + updatePrimeWork tpw miner cid = runForever (logFunction (chainLogger cid logger)) "primeWork" go where - go :: Miner -> IO () - go miner = do - pw <- readTVarIO tpw - let - -- we assume that this path always exists in PrimedWork and never delete it. - ourMiner :: Traversal' PrimedWork WorkState + go :: IO () + go = do + let ourMiner :: Traversal' PrimedWork WorkState ourMiner = workForMiner miner cid - let !outdatedPayload = fromJuste $ pw ^? ourMiner - let outdatedParentHash = case outdatedPayload of - WorkReady outdatedBlock -> view _1 (newBlockParent outdatedBlock) - WorkAlreadyMined outdatedBlockHash -> outdatedBlockHash - WorkStale -> error "primeWork loop: Invariant Violation: Stale work should be an impossibility" + + pw <- readTVarIO tpw + + outdatedParentHash <- case pw ^? ourMiner of + Just (WorkReady outdatedBlock) -> return $ view _1 (newBlockParent outdatedBlock) + Just (WorkAlreadyMined outdatedBlockHash) -> return outdatedBlockHash + -- Stale here is not an usual state (newly registered miners, or failure during previous iteration) + -- => using nullBlockHash will uncondionally trigger awaitNewBlock + Just WorkStale -> return nullBlockHash + -- Having no primerwork data is not normal. It could only happen in a race condition when the thread is going to be cancelled + -- Better we can do is to wait for our death, it should happen very soon. + Nothing -> forever (threadDelay 10_000_000) >> return nullBlockHash newParent <- either ParentHeader id <$> race -- wait for a block different from what we've got primed work for @@ -216,7 +296,6 @@ withMiningCoordination logger conf cdb inner logFunctionText (addLabel ("chain", toText cid) logger) Warn "current block is not in the checkpointer; halting primed work loop temporarily" approximateThreadDelay 1_000_000 - atomically $ modifyTVar' tpw (ourMiner .~ outdatedPayload) Historical newBlock -> atomically $ modifyTVar' tpw (ourMiner .~ WorkReady newBlock) @@ -241,8 +320,8 @@ withMiningCoordination logger conf cdb inner -- | THREAD: Periodically clear out the cached payloads kept for Mining -- Coordination. -- - prune :: TVar MiningState -> TVar PrimedWork -> IORef Int -> IORef Int -> IO () - prune t tpw c503 c403 = runForever (logFunction logger) "MinerResources.prune" $ do + prune :: TVar MiningState -> TVar ActiveMiners -> TVar PrimedWork -> IORef Int -> IORef Int -> IO () + prune t tam tpw c503 c403 = runForever (logFunction logger) "MinerResources.prune" $ do let !d = 30_000_000 -- 30 seconds let !maxAge = (5 :: Int) `scaleTimeSpan` minute -- 5 minutes threadDelay d @@ -251,6 +330,20 @@ withMiningCoordination logger conf cdb inner ms <- readTVar t modifyTVar' t . over miningState $ M.filter (f ago) pure ms + + -- Remove dynamic miners not seen since more than 5 minutes + removedMiners <- atomically $ do + am <- readTVar tam + let expiredMiners = HM.filter (isMinerExpired ago) am + forM_ (HM.keys expiredMiners) $ \mid -> do + modifyTVar tam $ HM.delete mid + modifyTVar tpw $ coerce $ PrimedWork . HM.delete mid + return expiredMiners + + cancelMany $ allHandles removedMiners + + forM_ (HM.keys removedMiners) $ \mid -> logFunctionText (minerIdLogger mid logger) Info "Nos seen since a while => Removed" + count503 <- readIORef c503 count403 <- readIORef c403 PrimedWork pw <- readTVarIO tpw diff --git a/src/Chainweb/Miner/Config.hs b/src/Chainweb/Miner/Config.hs index 97fc76fbd4..69835aec4a 100644 --- a/src/Chainweb/Miner/Config.hs +++ b/src/Chainweb/Miner/Config.hs @@ -158,6 +158,10 @@ data CoordinationConfig = CoordinationConfig , _coordinationReqLimit :: !Int -- ^ The number of @/mining/work/@ requests that can be made to this node -- in a 5 minute period. + , _coordinationDynamicMinersEnabled :: !Bool + -- ^ Is it possible to add miners dynamically + , _coordinationMinersLimit :: !Int + -- ^ The maximum number of miners allowed to mine simultanously , _coordinationUpdateStreamLimit :: !Int -- ^ the maximum number of concurrent update streams that is supported , _coordinationUpdateStreamTimeout :: !Seconds @@ -193,6 +197,14 @@ coordinationTargetForkOverride :: Lens' CoordinationConfig Bool coordinationTargetForkOverride = lens _coordinationTargetForkOverride (\m c -> m { _coordinationTargetForkOverride = c }) +coordinationDynamicMinersEnabled :: Lens' CoordinationConfig Bool +coordinationDynamicMinersEnabled = + lens _coordinationDynamicMinersEnabled (\m c -> m { _coordinationDynamicMinersEnabled = c }) + +coordinationMinersLimit :: Lens' CoordinationConfig Int +coordinationMinersLimit = + lens _coordinationMinersLimit (\m c -> m { _coordinationMinersLimit = c }) + instance ToJSON CoordinationConfig where toJSON o = object [ "enabled" .= _coordinationEnabled o @@ -201,6 +213,8 @@ instance ToJSON CoordinationConfig where , "updateStreamLimit" .= _coordinationUpdateStreamLimit o , "updateStreamTimeout" .= _coordinationUpdateStreamTimeout o , "payloadRefreshDelay" .= _coordinationPayloadRefreshDelay o + , "dynamicMinersEnabled" .= _coordinationDynamicMinersEnabled o + , "coordinationMinersLimit" .= _coordinationMinersLimit o , "targetForkOverride" .= _coordinationTargetForkOverride o ] @@ -213,12 +227,16 @@ instance FromJSON (CoordinationConfig -> CoordinationConfig) where <*< coordinationUpdateStreamTimeout ..: "updateStreamTimeout" % o <*< coordinationPayloadRefreshDelay ..: "payloadRefreshDelay" % o <*< coordinationTargetForkOverride ..: "targetForkOverride" % o + <*< coordinationDynamicMinersEnabled ..: "dynamicMinersEnabled" % o + <*< coordinationMinersLimit ..: "coordinationMinersLimit" % o defaultCoordination :: CoordinationConfig defaultCoordination = CoordinationConfig { _coordinationEnabled = False , _coordinationMiners = mempty , _coordinationReqLimit = 1200 + , _coordinationDynamicMinersEnabled = False + , _coordinationMinersLimit = 10 , _coordinationUpdateStreamLimit = 2000 , _coordinationUpdateStreamTimeout = 240 , _coordinationPayloadRefreshDelay = TimeSpan (Micros 15_000_000) diff --git a/src/Chainweb/Miner/Coordinator.hs b/src/Chainweb/Miner/Coordinator.hs index a5a2fc210d..923f9fe8d3 100644 --- a/src/Chainweb/Miner/Coordinator.hs +++ b/src/Chainweb/Miner/Coordinator.hs @@ -129,6 +129,7 @@ data MiningCoordination logger tbl = MiningCoordination , _coord503s :: !(IORef Int) , _coord403s :: !(IORef Int) , _coordConf :: !CoordinationConfig + , _coordRefreshMiner :: Miner -> IO (Either String ()) , _coordUpdateStreamCount :: !(IORef Int) , _coordPrimedWork :: !(TVar PrimedWork) , _coordTargetFork :: !ForkNumber diff --git a/src/Chainweb/Miner/RestAPI/Server.hs b/src/Chainweb/Miner/RestAPI/Server.hs index 26d11ac250..dbedffc075 100644 --- a/src/Chainweb/Miner/RestAPI/Server.hs +++ b/src/Chainweb/Miner/RestAPI/Server.hs @@ -27,7 +27,7 @@ module Chainweb.Miner.RestAPI.Server import Control.Concurrent.STM.TVar (TVar, readTVar, readTVarIO, registerDelay) import Control.Lens -import Control.Monad (when, unless) +import Control.Monad (when) import Control.Monad.Catch (bracket, try, catches) import qualified Control.Monad.Catch as E import Control.Monad.Except (throwError) @@ -39,7 +39,6 @@ import qualified Data.HashMap.Strict as HM import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef, writeIORef) import qualified Data.Map.Strict as M import Data.Proxy (Proxy(..)) -import qualified Data.Set as S import qualified Data.Vector as V import Network.HTTP.Types.Status @@ -83,11 +82,13 @@ workHandler mr mcid m@(Miner (MinerId mid) _) = do when (M.size ms > _coordLimit mr) $ do liftIO $ atomicModifyIORef' (_coord503s mr) (\c -> (c + 1, ())) throwError $ setErrText "Too many work requests" err503 - let !conf = _coordConf mr - !primed = S.member m $ _coordinationMiners conf - unless primed $ do - liftIO $ atomicModifyIORef' (_coord403s mr) (\c -> (c + 1, ())) - throwError $ setErrText ("Unauthorized Miner: " <> mid) err403 + + (liftIO $ _coordRefreshMiner mr m) >>= \case + Right _ -> return () + Left x -> do + liftIO $ atomicModifyIORef' (_coord403s mr) (\c -> (c + 1, ())) + throwError $ setErrText ("Unauthorized Miner/" <> toText x <> ": " <> mid) err403 + wh <- liftIO $ work mr mcid m return $ WorkBytes $ runPutS $ encodeWorkHeader wh diff --git a/test/lib/Chainweb/Test/Orphans/Internal.hs b/test/lib/Chainweb/Test/Orphans/Internal.hs index a442a651d8..75d7c8f5b2 100644 --- a/test/lib/Chainweb/Test/Orphans/Internal.hs +++ b/test/lib/Chainweb/Test/Orphans/Internal.hs @@ -841,6 +841,8 @@ instance Arbitrary CoordinationConfig where <*> arbitrary <*> arbitrary <*> arbitrary + <*> arbitrary + <*> arbitrary instance Arbitrary NodeMiningConfig where arbitrary = NodeMiningConfig