chainweb-mining-client (empty) → 0.2
raw patch · 11 files changed
+1658/−0 lines, 11 filesdep +aesondep +asyncdep +basebuild-type:Customsetup-changed
Dependencies added: aeson, async, base, bytes, bytestring, configuration-tools, connection, containers, cryptonite, exceptions, hashable, hostaddress, http-client, http-client-tls, http-types, lens, loglevel, memory, mwc-random, process, retry, stm, streaming, streaming-events, text, time, unordered-containers, wai-extra
Files
- CHANGELOG.md +9/−0
- LICENSE +29/−0
- README.md +10/−0
- Setup.hs +4/−0
- chainweb-mining-client.cabal +88/−0
- main/Main.hs +788/−0
- src/Logger.hs +227/−0
- src/Worker.hs +106/−0
- src/Worker/CPU.hs +181/−0
- src/Worker/External.hs +131/−0
- src/Worker/Simulation.hs +85/−0
+ CHANGELOG.md view
@@ -0,0 +1,9 @@+# Revision history for chainweb-mining-client++## 0.2 -- 2020-08-20++* Rename package into chainweb-mining-client++## 0.1 -- 2020-08-20++* First version. Released on an unsuspecting world.
+ LICENSE view
@@ -0,0 +1,29 @@+Copyright (c) 2019 Kadena LLC+All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are+met:++1. Redistributions of source code must retain the above copyright+notice, this list of conditions and the following disclaimer.++2. Redistributions in binary form must reproduce the above copyright+notice, this list of conditions and the following disclaimer in the+documentation and/or other materials provided with the distribution.++3. Neither the name of the copyright holder nor the names of its+contributors may be used to endorse or promote products derived from+this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,10 @@+A mining client for Kadena Chainweb. It supports++* multi threaded CPU mining,+* external mining workers (e.g. a GPU),+* simulated mining for testing.++Competitive mining on the Kadena Chainweb Mainnet requires special mining+hardware, which usually comes with its own mining client and mining pool+support implementations. This generic mining client is intended mostly for+testing.
+ Setup.hs view
@@ -0,0 +1,4 @@+module Main (main) where++import Configuration.Utils.Setup+
+ chainweb-mining-client.cabal view
@@ -0,0 +1,88 @@+cabal-version: 3.0+name: chainweb-mining-client+description:+ A mining client for Kadena Chainweb. It supports++ * multi threaded CPU mining,+ * external mining workers (e.g. a GPU),+ * simulated mining for testing.++ Competitive mining on the Kadena Chainweb Mainnet requires special mining+ hardware, which usually comes with its own mining client and mining pool+ support implementations. This generic mining client is intended mostly for+ testing.++version: 0.2+synopsis: Mining Client for Kadena Chainweb+homepage: https://github.com/kadena-io/chainweb-mining-client+bug-reports: https://github.com/kadena-io/chainweb-mining-client/issues+license: BSD-3-Clause+license-file: LICENSE+author: Lars Kuhtz+maintainer: lars@kadena.io+copyright: Copyright (c) 2019 - 2020, Kadena LLC+category: Data, Mathematics+build-type: Custom+tested-with:+ GHC==8.10.2+ , GHC==8.8.4+extra-source-files:+ README.md+ CHANGELOG.md++source-repository head+ type: git+ location: https://github.com/kadena-io/chainweb-mining-client.git++custom-setup+ setup-depends:+ base >=4.10 && <5+ , Cabal >=3.0+ , configuration-tools >=0.5++executable chainweb-mining-client+ hs-source-dirs: main, src+ main-is: Main.hs+ other-modules:+ PkgInfo+ Logger+ Worker+ Worker.CPU+ Worker.External+ Worker.Simulation+ autogen-modules:+ PkgInfo+ default-language: Haskell2010+ ghc-options:+ -Wall+ -threaded+ -with-rtsopts=-N+ build-depends:+ base >=4.10 && <4.15+ , aeson >=1.5+ , async >=2.2+ , bytes >=0.17+ , bytestring >=0.10+ , configuration-tools >=0.5+ , connection >=0.3+ , containers >=0.5+ , cryptonite >=0.27+ , exceptions >=0.10+ , hashable >=1.3+ , hostaddress >=0.1+ , http-client >=0.7+ , http-client-tls >=0.3+ , http-types >=0.12+ , lens >=4.19+ , loglevel >=0.1+ , memory >=0.15+ , mwc-random >=0.14+ , process >=1.6+ , retry >=0.8+ , stm >=2.5+ , streaming >=0.2+ , streaming-events >=1.0+ , text >=1.2+ , time >=1.9+ , unordered-containers >=0.2+ , wai-extra >=3.0
+ main/Main.hs view
@@ -0,0 +1,788 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE DeriveAnyClass #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE FlexibleInstances #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TupleSections #-}+{-# LANGUAGE TypeApplications #-}++{-# OPTIONS_GHC -fno-warn-orphans #-}++-- |+-- Module: Main+-- Copyright: Copyright © 2020 Kadena LLC.+-- License: MIT+-- Maintainer: Lars Kuhtz <lars@kadena.io>+-- Stability: experimental+--+module Main+( main+) where++import Configuration.Utils hiding (Error)++import Control.Concurrent+import Control.Concurrent.Async+import Control.Concurrent.STM+import Control.Exception (IOException, SomeAsyncException)+import Control.Lens hiding ((.=))+import Control.Monad+import Control.Monad.Catch+import Control.Monad.IO.Class+import Control.Retry++import Crypto.Hash.Algorithms (Blake2s_256)+import qualified Crypto.PubKey.Ed25519 as C++import Data.Bifunctor+import qualified Data.ByteArray.Encoding as BA+import Data.Bytes.Get+import Data.Bytes.Put+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import qualified Data.ByteString.Lazy as BL+import qualified Data.ByteString.Short as BS+import Data.Hashable+import qualified Data.HashMap.Strict as HM+import Data.String+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import Data.Word++import GHC.Generics++import qualified Network.Connection as HTTP+import Network.HostAddress+import qualified Network.HTTP.Client as HTTP+import qualified Network.HTTP.Client.TLS as HTTP+import qualified Network.HTTP.Types.Status as HTTP+import Network.Wai.EventSource.EventStream+import Network.Wai.EventSource.Streaming++import Numeric.Natural++import PkgInfo hiding (tag)++import qualified Streaming.Prelude as SP++import System.LogLevel+import qualified System.Random.MWC as MWC++import qualified Text.ParserCombinators.ReadP as R+import qualified Text.ParserCombinators.ReadPrec as P+import Text.Printf+import Text.Read (Read(..), readListPrecDefault)++-- internal modules++import Logger+import Worker+import Worker.CPU+import Worker.External+import Worker.Simulation++-- -------------------------------------------------------------------------- --+-- Orphans++instance ToJSON HostAddress where+ toJSON = toJSON . hostAddressToText+ {-# INLINE toJSON #-}++instance FromJSON HostAddress where+ parseJSON = withText "HostAddress"+ $ either (fail . show) return . hostAddressFromText+ {-# INLINE parseJSON #-}++-- -------------------------------------------------------------------------- --+-- Utils++textReader :: (T.Text -> Either SomeException a) -> ReadM a+textReader p = eitherReader $ first show . p . T.pack++sshow :: Show a => IsString b => a -> b+sshow = fromString . show+{-# INLINE sshow #-}++-- -------------------------------------------------------------------------- --+-- Integral Unit Prefixes++-- | TODO: make this type roundtripable and add support for fractional unit+-- prefixes.+--+newtype UnitPrefixed a = UnitPrefixed { _getUnitPrefixed :: a }+ deriving newtype+ ( Show+ , Eq+ , Ord+ , Enum+ , Bounded+ , Num+ , Integral+ , Fractional+ , Floating+ , Real+ , ToJSON+ , FromJSON+ )++instance (Num a, Read a) => Read (UnitPrefixed a) where+ readPrec = UnitPrefixed <$> readWithUnit+ readListPrec = readListPrecDefault+ {-# INLINE readPrec #-}+ {-# INLINE readListPrec #-}++-- | Read number with Unit Prefixes. The implementation supports integral SI+-- units prefixes. Binary prefixes are supported according to ISO/IEC 80000.+--+-- "Kilo" is supported, both in upper and lower case.+--+readWithUnit :: forall a . Num a => Read a => P.ReadPrec a+readWithUnit = do+ n <- readPrec+ p <- P.lift $ noPrefix <|> siPrefix <|> binaryPrefix+ return $! n * p+ where+ noPrefix :: R.ReadP a+ noPrefix = 1 <$ R.eof++ siPrefix :: R.ReadP a+ siPrefix+ = 10^(1 :: Int) <$ R.string "da"+ <|> 10^(2 :: Int) <$ R.char 'h'+ <|> 10^(3 :: Int) <$ (R.char 'K' <|> R.char 'k')+ <|> 10^(6 :: Int) <$ R.char 'M'+ <|> 10^(9 :: Int) <$ R.char 'G'+ <|> 10^(12 :: Int) <$ R.char 'T'+ <|> 10^(15 :: Int) <$ R.char 'P'+ <|> 10^(18 :: Int) <$ R.char 'E'+ <|> 10^(21 :: Int) <$ R.char 'Z'+ <|> 10^(24 :: Int) <$ R.char 'Y'++ binaryPrefix :: R.ReadP a+ binaryPrefix+ = 1024^(1 :: Int) <$ (R.string "Ki" <|> R.string "ki")+ <|> 1024^(2 :: Int) <$ R.string "Mi"+ <|> 1024^(3 :: Int) <$ R.string "Gi"+ <|> 1024^(4 :: Int) <$ R.string "Ti"+ <|> 1024^(5 :: Int) <$ R.string "Pi"+ <|> 1024^(6 :: Int) <$ R.string "Ei"+ <|> 1024^(7 :: Int) <$ R.string "Zi"+ <|> 1024^(8 :: Int) <$ R.string "Yi"++-- -------------------------------------------------------------------------- --+-- Miner++-- | The account name is that same as the public key. Different account names+-- are not supported.+--+-- Only a single base64UrlWithoutPadding encoded key may be used and the keyset+-- is "keys-all".+--+newtype MinerPublicKey = MinerPublicKey T.Text+ deriving (Show, Eq, Ord, Generic)+ deriving newtype (Hashable)++instance ToJSON MinerPublicKey where+ toJSON (MinerPublicKey k) = toJSON k+ {-# INLINE toJSON #-}++instance FromJSON MinerPublicKey where+ parseJSON = withText "MinerPublicKey" $ return . MinerPublicKey+ {-# INLINE parseJSON #-}+ -- TODO perform well-formedness checks++newtype Miner = Miner MinerPublicKey+ deriving (Show, Eq, Ord, Generic)+ deriving newtype (Hashable)++instance ToJSON Miner where+ toJSON (Miner (MinerPublicKey k)) = object+ [ "account" .= k+ , "public-keys" .= [ k ]+ , "predicate" .= ("keys-all" :: T.Text)+ ]++-- -------------------------------------------------------------------------- --+-- Worker Configuration++data WorkerConfig+ = CpuWorker+ | ExternalWorker+ | SimulationWorker+ deriving (Show, Eq, Ord, Generic)+ deriving anyclass (Hashable)++instance ToJSON WorkerConfig where+ toJSON = toJSON . workerConfigToText+ {-# INLINE toJSON #-}++instance FromJSON WorkerConfig where+ parseJSON = withText "WorkerConfig" $+ either (fail . show) return . workerConfigFromText+ {-# INLINE parseJSON #-}++workerConfigToText :: WorkerConfig -> T.Text+workerConfigToText CpuWorker = "cpu"+workerConfigToText ExternalWorker = "external"+workerConfigToText SimulationWorker = "simulation"++workerConfigFromText :: MonadThrow m => T.Text -> m WorkerConfig+workerConfigFromText t = case T.toCaseFold t of+ "cpu" -> return CpuWorker+ "external" -> return ExternalWorker+ "simulation" -> return SimulationWorker+ _ -> error $ "unknown worker configuraton: " <> T.unpack t++-- -------------------------------------------------------------------------- --+-- Configuration++newtype ChainwebVersion = ChainwebVersion T.Text+ deriving (Show, Read, Eq, Ord, Generic)+ deriving newtype (Hashable, ToJSON, FromJSON)++data Config = Config+ { _configHashRate :: !(UnitPrefixed HashRate)+ , _configNode :: !HostAddress+ , _configUseTls :: !Bool+ , _configInsecure :: !Bool+ , _configPublicKey :: !MinerPublicKey+ , _configThreadCount :: !Natural+ , _configGenerateKey :: !Bool+ , _configLogLevel :: !LogLevel+ , _configWorker :: !WorkerConfig+ , _configExternalWorkerCommand :: !String+ }+ deriving (Show, Eq, Ord, Generic)++makeLenses ''Config++defaultConfig :: Config+defaultConfig = Config+ { _configHashRate = UnitPrefixed defaultHashRate+ , _configNode = unsafeHostAddressFromText "localhost:1789"+ , _configUseTls = True+ , _configInsecure = True+ , _configPublicKey = MinerPublicKey ""+ , _configThreadCount = 10+ , _configGenerateKey = False+ , _configLogLevel = Info+ , _configWorker = CpuWorker+ , _configExternalWorkerCommand = "echo 'no external worker command configured' && /bin/false"+ }++instance ToJSON Config where+ toJSON c = object+ [ "hashRate" .= _configHashRate c+ , "node" .= _configNode c+ , "useTls" .= _configUseTls c+ , "insecure" .= _configInsecure c+ , "publicKey" .= _configPublicKey c+ , "threadCount" .= _configThreadCount c+ , "generateKey" .= _configGenerateKey c+ , "logLevel" .= logLevelToText @T.Text (_configLogLevel c)+ , "worker" .= _configWorker c+ , "externalWorkerCommand" .= _configExternalWorkerCommand c+ ]++instance FromJSON (Config -> Config) where+ parseJSON = withObject "Config" $ \o -> id+ <$< configHashRate ..: "hashRate" % o+ <*< configNode ..: "node" % o+ <*< configUseTls ..: "useTls" % o+ <*< configInsecure ..: "insecure" % o+ <*< configPublicKey ..: "publicKey" % o+ <*< configThreadCount ..: "threadCount" % o+ <*< configGenerateKey ..: "generateKey" % o+ <*< setProperty configLogLevel "logLevel" parseLogLevel o+ <*< configWorker ..: "worker" % o+ <*< configExternalWorkerCommand ..: "externalWorkerCommand" % o+ where+ parseLogLevel = withText "LogLevel" $ return . logLevelFromText++parseConfig :: MParser Config+parseConfig = id+ <$< configHashRate .:: option auto+ % short 'r'+ <> long "hash-rate"+ <> help "hashes per second (only relevant for mining simulation, ignored by the cpu worker)"+ <*< configNode .:: option (textReader hostAddressFromText)+ % short 'n'+ <> long "node"+ <> help "node to which to connect"+ <> metavar "DOMAIN:PORT"+ <*< configUseTls .:: boolOption_+ % short 't'+ <> long "tls"+ <> help "use TLS to connect to node"+ <*< configInsecure .:: boolOption_+ % short 'x'+ <> long "insecure"+ <> help "accept self-signed TLS certificates"+ <*< configPublicKey .:: fmap MinerPublicKey . strOption+ % short 'k'+ <> long "public-key"+ <> help "the public-key for the mining rewards account"+ <*< configThreadCount .:: option auto+ % short 'c'+ <> long "thread-count"+ <> help "number of concurrent mining threads"+ <*< configGenerateKey .:: boolOption_+ % long "generate-key"+ <> help "Generate a new key pair and exit"+ <*< configLogLevel .:: option (textReader $ Right . logLevelFromText)+ % short 'l'+ <> long "log-level"+ <> help "Level at which log messages are written to the console"+ <> metavar "error|warn|info|debug"+ <*< configWorker .:: option (textReader workerConfigFromText)+ % short 'w'+ <> long "worker"+ <> help "The type of mining worker that is used"+ <> metavar "cpu|external|simulation"+ <*< configExternalWorkerCommand .:: option (textReader $ Right . T.unpack)+ % long "external-worker-cmd"+ <> help "command that is used to call an external worker. When the command is called the target value is added as last parameter to the command line."++-- -------------------------------------------------------------------------- --+-- HTTP Retry Logic++-- | We don't limit retries. The maximum delay between retries is 5 seconds.+--+-- TODO: add configuration option for limitRetriesByCumulativeDelay+--+retryHttp :: Logger -> IO a -> IO a+retryHttp logger = recovering policy (httpRetryHandler logger) . const+ where+ policy = capDelay 5000000 $ fullJitterBackoff 100++httpRetryHandler :: Logger -> [RetryStatus -> Handler IO Bool]+httpRetryHandler logger = skipAsyncExceptions <>+ [ logRetries (return . httpRetries) f+ , logRetries (\(_ :: SomeException) -> return True) logRetry+ ]+ where+ logRetry True reason s = writeLog logger Warn+ $ "Http request failed: " <> sshow reason+ <> ". Retrying attempt " <> sshow (rsIterNumber s)+ logRetry False reason s = writeLog logger Warn+ $ "Http request finally failed after " <> sshow (rsIterNumber s)+ <> " retries: " <> sshow reason++ f True (HTTP.HttpExceptionRequest _req reason) s = logRetry True reason s+ f False (HTTP.HttpExceptionRequest _req reason) s = logRetry False reason s+ f _ e _ = throwM e+++-- | HTTP Exceptions for which a retry may result in subsequent succes.+--+-- This retries rather aggressively on any server or network related failure+-- condition.+--+httpRetries :: HTTP.HttpException -> Bool+httpRetries (HTTP.HttpExceptionRequest _req reason) = case reason of+ HTTP.StatusCodeException resp _body+ | HTTP.statusIsServerError (HTTP.responseStatus resp) -> True+ HTTP.ResponseTimeout -> True+ HTTP.ConnectionTimeout -> True+ HTTP.ConnectionFailure _e -> True+ HTTP.InvalidStatusLine _bs -> True+ HTTP.InvalidHeader _bs -> True+ HTTP.InternalException _e -> True+ HTTP.ProxyConnectException _host _port status+ | HTTP.statusIsServerError status -> True+ HTTP.NoResponseDataReceived -> True+ HTTP.ResponseBodyTooShort _expected _actual -> True+ HTTP.InvalidChunkHeaders -> True+ HTTP.IncompleteHeaders -> True+ HTTP.HttpZlibException _e -> True+ HTTP.ConnectionClosed -> True+ _ -> False+httpRetries (HTTP.InvalidUrlException _url _reason) = False++-- -------------------------------------------------------------------------- --+-- Chainweb Mining API Types++-- | ChainId+--+newtype ChainId = ChainId Word32+ deriving (Show, Eq, Ord, Generic)+ deriving anyclass (Hashable)++decodeChainId :: MonadGet m => m ChainId+decodeChainId = ChainId <$> getWord32le++encodeChainId :: MonadPut m => ChainId -> m ()+encodeChainId (ChainId w32) = putWord32le w32++-- -------------------------------------------------------------------------- --+-- Chainweb Mining API Requetss++newtype GetWorkFailure = GetWorkFailure T.Text+ deriving (Show, Eq, Ord)++instance Exception GetWorkFailure++-- | Make an HTTP request with an JSON response+--+getJson :: FromJSON a => HTTP.Manager -> HTTP.Request -> IO a+getJson mgr req = (eitherDecode . HTTP.responseBody <$> HTTP.httpLbs req mgr) >>= \case+ Left e -> error $ "Failed to decode json response: " <> show e+ Right r -> return r++-- | Base request type for chainweb queries+--+baseReq :: Config -> ChainwebVersion -> B.ByteString -> HTTP.Request+baseReq conf (ChainwebVersion v) pathSuffix = HTTP.defaultRequest+ { HTTP.host = T.encodeUtf8 $ hostnameToText $ _hostAddressHost node+ , HTTP.path = "chainweb/0.0/" <> T.encodeUtf8 v <> "/" <> pathSuffix+ , HTTP.port = fromIntegral $ _hostAddressPort node+ , HTTP.secure = _configUseTls conf+ , HTTP.method = "GET"+ , HTTP.checkResponse = HTTP.throwErrorStatusCodes+ }+ where+ node = _configNode conf++-- | Query node info+--+getInfo :: Config -> HTTP.Manager -> IO (HM.HashMap T.Text Value)+getInfo conf mgr = getJson mgr req+ where+ req = HTTP.defaultRequest+ { HTTP.host = T.encodeUtf8 $ hostnameToText $ _hostAddressHost node+ , HTTP.path = "info"+ , HTTP.port = fromIntegral $ _hostAddressPort node+ , HTTP.secure = _configUseTls conf+ , HTTP.method = "GET"+ , HTTP.checkResponse = HTTP.throwErrorStatusCodes+ }+ node = _configNode conf++-- | Obtain chainweb version of the chainweb node+--+-- No retry here. This is use at startup and we want to fail fast if the node+-- isn't available.+--+getNodeVersion :: Config -> HTTP.Manager -> IO ChainwebVersion+getNodeVersion conf mgr = do+ i <- getInfo conf mgr+ case HM.lookup "nodeVersion" i of+ Just (String x) -> return $ ChainwebVersion x+ _ -> error "failed to parse chainweb version from node info"++-- | Get new work from the chainweb node (for some available chain)+--+-- We don't retry here. If this fails, we loop around.+--+getJob :: Config -> ChainwebVersion -> HTTP.Manager -> IO (ChainId, Target, Work)+getJob conf ver mgr = do+ bytes <- HTTP.httpLbs req mgr+ case runGetS decodeJob (BL.toStrict $ HTTP.responseBody $ bytes) of+ Left e -> error $ "failed to decode work: " <> sshow e+ Right (a,b,c) -> return (a, b, c)+ where+ req = (baseReq conf ver "mining/work")+ { HTTP.requestBody = HTTP.RequestBodyLBS $ encode $ Miner $ _configPublicKey conf+ , HTTP.requestHeaders = [("content-type", "application/json")]+ }++ decodeJob :: MonadGet m => m (ChainId, Target, Work)+ decodeJob = (,,)+ <$> decodeChainId+ <*> decodeTarget+ <*> decodeWork++-- | Post solved work to the chainweb node+--+-- No timeout is used and in case of failure we retry aggressively. If the+-- solvedwork becomes stale, the thread will be preempted and cancled.+--+postSolved :: Config -> ChainwebVersion -> Logger -> HTTP.Manager -> Work -> IO ()+postSolved conf ver logger mgr (Work bytes) = retryHttp logger $ do+ logg Info "post solved worked"+ (void $ HTTP.httpLbs req mgr)+ `catch` \(e@(HTTP.HttpExceptionRequest _ _)) -> do+ logg Error $ "failed to submit solved work: " <> sshow e+ return ()+ where+ logg = writeLog logger+ req = (baseReq conf ver "mining/solved")+ { HTTP.requestBody = HTTP.RequestBodyBS $ BS.fromShort $ bytes+ , HTTP.method = "POST"+ }++-- | Automatically restarts the stream when the response status is 2** and throws+-- and exception otherwise.+--+-- No+-- No retry is used. Retrying is handled by the outer logic.+--+updateStream+ :: Config+ -> ChainwebVersion+ -> Logger+ -> HTTP.Manager+ -> ChainId+ -> TVar Int+ -> IO ()+updateStream conf v logger mgr cid var =+ liftIO $ withEvents req mgr $ \updates -> updates+ & SP.filter realEvent+ & SP.chain (\_ -> logg Info $ "got update on chain " <> sshow cid)+ & SP.mapM_ (\_ -> atomically $ modifyTVar' var (+ 1))+ where+ logg = writeLog logger++ realEvent ServerEvent{} = True+ realEvent _ = False++ req = (baseReq conf v "mining/updates")+ { HTTP.requestBody = HTTP.RequestBodyBS $ runPutS $ encodeChainId cid+ , HTTP.responseTimeout = HTTP.responseTimeoutNone+ }++-- -------------------------------------------------------------------------- --+-- Trigger Type++data Reason = Timeout | Update | StreamClosed | StreamFailed SomeException+ deriving (Show)++-- | A trigger is used to preempt a worker thread.+--+newtype Trigger = Trigger (STM Reason)++awaitTrigger :: Trigger -> IO Reason+awaitTrigger (Trigger t) = atomically t++-- -------------------------------------------------------------------------- --+-- Update Map++newtype UpdateFailure = UpdateFailure T.Text+ deriving (Show, Eq, Ord)++instance Exception UpdateFailure++-- | This keeps track of the current work for the respective chain. It is shared+-- among all worker threads. If an update for some chain is received the worker+-- threads are preempted and query new work.+--+-- It also keeps track of the long-polling update streams.+--+newtype UpdateMap = UpdateMap+ { _updateMap :: MVar (HM.HashMap ChainId (TVar Int, Async ()))+ }++-- | Creates a map that maintains one upstream for each chain+--+newUpdateMap :: IO UpdateMap+newUpdateMap = UpdateMap <$> newMVar mempty++-- | Obtain a trigger that is used to preempt a worker threads. It notifies the+-- thread if an update is available.+--+getTrigger+ :: Config+ -> ChainwebVersion+ -> Logger+ -> HTTP.Manager+ -> UpdateMap+ -> ChainId+ -> IO Trigger+getTrigger conf ver logger mgr (UpdateMap v) k = modifyMVar v $ \m -> case HM.lookup k m of++ -- If there exists already an update stream, check that it's live, and+ -- restart if necessary.+ --+ Just s -> do+ logg Debug "use existing update stream"+ n@(!var, !a) <- checkStream s+ !t <- newTrigger var a+ let !x = HM.insert k n m+ return (x, t)++ -- If there isn't an update stream in the map, create a new one.+ --+ Nothing -> do+ logg Debug "create new update stream"+ n@(!var, !a) <- newTVarIO 0 >>= newUpdateStream+ !t <- newTrigger var a+ let !x = HM.insert k n m+ return (x, t)+ where+ logg = writeLog logger++ checkStream :: (TVar Int, Async ()) -> IO (TVar Int, Async ())+ checkStream (!var, !a) = poll a >>= \case+ Nothing -> return (var, a)+ Just (Left _) -> newUpdateStream var -- TODO logging, throttling+ Just (Right _) -> newUpdateStream var++ newUpdateStream :: TVar Int -> IO (TVar Int, Async ())+ newUpdateStream var = (var,)+ <$> async (updateStream conf ver logger mgr k var)++ -- There are three possible outcomes+ --+ newTrigger :: TVar Int -> Async () -> IO Trigger+ newTrigger var a = do+ cur <- readTVarIO var+ timeoutVar <- registerDelay (5 * 30_000_000)+ -- 5 times the block time ~ 0.7% of all blocks. This for detecting if+ -- a stream gets stale without failing.++ return $ Trigger $ pollSTM a >>= \case+ Just (Right ()) -> return StreamClosed+ Just (Left e) -> return $ StreamFailed e+ Nothing -> do+ isTimeout <- readTVar timeoutVar+ isUpdate <- (/= cur) <$> readTVar var+ unless (isTimeout || isUpdate) retry+ return Update++-- | Run an operation that is preempted if an update event occurs.+--+-- Streams are restarted automatically, when they got closed by the server. We+-- don't restart streams automatically in case of a failure, but instead throw+-- an exception. Failures are supposed to be handled in the outer mining+-- functions.+--+-- There is risk that a stream stalls without explicitely failing. We solve this+-- by preempting the loop if we haven't seen an update after 5 times the block+-- time (which will affect about 0.7% of all blocks).+--+withPreemption+ :: Config+ -> ChainwebVersion+ -> Logger+ -> HTTP.Manager+ -> UpdateMap+ -> ChainId+ -> IO a+ -> IO (Either () a)+withPreemption conf ver logger mgr m k = race awaitChange+ where+ awaitChange = do+ trigger <- getTrigger conf ver logger mgr m k+ awaitTrigger trigger >>= \case+ StreamClosed -> awaitChange+ StreamFailed e -> throwM $ UpdateFailure $ "update stream failed: " <> errMsg e+ Timeout -> throwM $ UpdateFailure "timeout of update stream"+ Update -> return ()++ errMsg e = case fromException e of+ Just (HTTP.HttpExceptionRequest _ ex) -> sshow ex+ _ -> sshow e++-- -------------------------------------------------------------------------- --+-- Mining Loop++data Recovery = Irrecoverable | Recoverable++-- | The outer mining loop.+--+miningLoop+ :: Config+ -> ChainwebVersion+ -> Logger+ -> HTTP.Manager+ -> UpdateMap+ -> Worker+ -> IO ()+miningLoop conf ver logger mgr umap worker = go+ where+ nonce = Nonce 0+ logg = writeLog logger+ go = (forever loopBody `catches` handlers) >>= \case+ Irrecoverable -> return ()+ Recoverable -> threadDelay 500_000 >> go++ handlers =+ [ Handler $ \(e :: IOException) -> do+ logg Error $ T.pack $ displayException e+ return Irrecoverable+ , Handler $ \(e :: GetWorkFailure) -> do+ logg Error $ T.pack $ displayException e+ return Irrecoverable -- FIXME we want proper retry logic for all of this!+ , Handler $ \(e :: UpdateFailure) -> do+ logg Error $ T.pack $ displayException e+ return Recoverable+ , Handler $ \(e :: SomeAsyncException) -> do+ logg Warn $ "Mining Loop terminated: " <> sshow e+ throwM e+ , Handler $ \(e :: SomeException) -> do+ logg Warn "Some general error in mining loop. Trying again..."+ logg Info $ "Exception: " <> T.pack (displayException e)+ return Recoverable+ ]++ loopBody = do+ (cid, target, work) <- getJob conf ver mgr+ logg Info $ "got new work for chain " <> sshow cid+ withPreemption conf ver logger mgr umap cid (worker nonce target work) >>= \case+ Right solved -> do+ -- TODO: we should do this asynchronously, however, preemption+ -- should still apply. So, ideally, we would kick of a new+ -- asynchronous loop interation while continuing this loop+ -- iteration here.+ postSolved conf ver logger mgr solved+ logg Debug "submitted work"+ Left () ->+ logg Info "Mining loop was preempted. Getting updated work ..."++-- -------------------------------------------------------------------------- --+-- Key generation++genKeys :: IO ()+genKeys = do+ sk <- C.generateSecretKey+ let !pk = C.toPublic sk+ printf "public: %s\n" (B8.unpack $ BA.convertToBase BA.Base16 pk)+ printf "private: %s\n" (B8.unpack $ BA.convertToBase BA.Base16 sk)++-- -------------------------------------------------------------------------- --+-- Main++mainInfo :: ProgramInfo Config+mainInfo = programInfo "Kadena Chainweb Mining Client" parseConfig defaultConfig++-- | TODO: validate the configuration:+--+-- * MinerPublicKey must be present+-- * node must be present+--+main :: IO ()+main = runWithPkgInfoConfiguration mainInfo pkgInfo $ \conf ->+ if _configGenerateKey conf+ then genKeys+ else withLogger (_configLogLevel conf) $ run conf++run :: Config -> Logger -> IO ()+run conf logger = do+ mgr <- HTTP.newManager (HTTP.mkManagerSettings tlsSettings Nothing)+ { HTTP.managerResponseTimeout = HTTP.responseTimeoutMicro 1000000 }+ -- We don't want to wait too long, because latencies matter in+ -- mining. NOTE, however, that for large blocks it can take a while+ -- to get new work. This can be an issue with public mining mode.+ -- For private mining that is done asynchronously. Public mining is+ -- considered deprecated.+ ver <- getNodeVersion conf mgr+ rng <- MWC.createSystemRandom+ updateMap <- newUpdateMap+ forConcurrently_ [0 .. (_configThreadCount conf) - 1] $ \i ->+ withLogTag logger ("Thread " <> sshow i) $ \taggedLogger ->+ miningLoop conf ver taggedLogger mgr updateMap $+ case _configWorker conf of+ SimulationWorker -> simulationWorker taggedLogger rng workerRate+ ExternalWorker -> externalWorker taggedLogger (_configExternalWorkerCommand conf)+ CpuWorker -> cpuWorker @Blake2s_256 taggedLogger+ where+ tlsSettings = HTTP.TLSSettingsSimple (_configInsecure conf) False False++ workerRate = _getUnitPrefixed (_configHashRate conf) / fromIntegral (_configThreadCount conf)+
+ src/Logger.hs view
@@ -0,0 +1,227 @@+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- |+-- Module: Logger+-- Copyright: Copyright © 2020 Kadena LLC.+-- License: MIT+-- Maintainer: Lars Kuhtz <lars@kadena.io>+-- Stability: experimental+--+-- A Lightweight Logging System+--+module Logger+( maxLoggerQueueSize+, Logger+, withLogger+, withLogTag+, writeLog+) where++import Control.Concurrent+import Control.Concurrent.Async+import Control.Exception+import Control.Monad++import Data.IORef+import qualified Data.Text as T+import qualified Data.Text.IO as T+import Data.Time.Clock.System+import Data.Time.Format.ISO8601++import System.IO+import System.IO.Unsafe+import System.LogLevel++-- -------------------------------------------------------------------------- --+-- Constants++maxLoggerQueueSize :: Int+maxLoggerQueueSize = 10000++-- -------------------------------------------------------------------------- --+-- Utils++incrementCounter :: IORef Int -> IO ()+incrementCounter ref = atomicModifyIORef' ref $ \x -> (x + 1, ())++decrementCounter :: IORef Int -> IO ()+decrementCounter ref = atomicModifyIORef' ref $ \x -> (x - 1, ())++resetCounter :: IORef Int -> IO Int+resetCounter ref = atomicModifyIORef' ref $ \x -> (0, x)++-- -------------------------------------------------------------------------- --+-- Terminal Colors++useColor :: Bool+useColor = unsafePerformIO $ hIsTerminalDevice stdout++data Color = Black | Red | Green | Yellow | Blue | Magenta | Cyan | White++colorCode :: Color -> Int+colorCode Black = 0+colorCode Red = 1+colorCode Green = 2+colorCode Yellow = 3+colorCode Blue = 4+colorCode Magenta = 5+colorCode Cyan = 6+colorCode White = 7++asDull, asVivid :: Color -> T.Text -> T.Text+asDull c t = setCode (30 + colorCode c) <> t <> setCode 0+asVivid c t = setCode (90 + colorCode c) <> t <> setCode 0++setCode :: Int -> T.Text+setCode c+ | useColor = "\ESC[" <> (T.pack . show) c <> "m"+ | otherwise = ""++-- -------------------------------------------------------------------------- --+-- Log Messages++-- | Log Message that is emitted by the code and enqueued in the logging queue.+--+-- The goal is to produce messages with very low latency in order to not delay+-- production logic. The most expensive part obtaining the system time, which is+-- stored raw without formatting.+--+-- When the LogMessage is written to the queue it must be fully evaluted to+-- normal form.+--+data LogMessage = LogMessage+ { _logMessageText :: !T.Text+ , _logMessageLevel :: !LogLevel+ , _logMessageTime :: !SystemTime+ , _logMessageTags :: ![T.Text]+ }++-- | Format Log message.+--+-- This is done in the logging backend asynchronously.+--+-- TODO: implement Chunk formatting.+--+formatLogMessage :: LogMessage -> T.Text+formatLogMessage !msg =+ asDull Cyan (padTime . T.pack . iso8601Show . systemToUTCTime $ _logMessageTime msg)+ <> " "+ <> formatLogLevel (_logMessageLevel msg)+ <> " "+ <> bracketed (formatTags (_logMessageTags msg))+ <> " "+ <> _logMessageText msg+ where+ formatTags tags = T.intercalate "|" $ asDull Green <$> reverse tags+ bracketed t = "[" <> t <> "]"++ padTime t+ | T.length t == 28 = t+ | otherwise = T.take 27 (T.drop 1 t <> "000000") <> "Z"++ formatLogLevel Quiet = "Quiet"+ formatLogLevel Debug = asVivid Blue "Debug"+ formatLogLevel Info = asVivid Yellow "Info "+ formatLogLevel Warn = asVivid Magenta "Warn "+ formatLogLevel Error = asVivid Red "Error"+ formatLogLevel (Other x) = asVivid Blue x++-- -------------------------------------------------------------------------- --+-- Logger Context++data Logger = Logger+ { _loggerTags :: [T.Text]+ , _loggerLevel :: !LogLevel+ , _loggerQueue :: !(Chan LogMessage)+ , _loggerApproxQueueSize :: !(IORef Int)+ , _loggerSkipped :: !(IORef Int)+ , _loggerMaxQueueSize :: !Int+ }++-- | Locally push a tag to the stack of log message tags.+--+withLogTag :: Logger -> T.Text -> (Logger -> IO a) -> IO a+withLogTag !logger !tag inner = inner $! logger+ { _loggerTags = tag : _loggerTags logger+ }++-- | Write a log message.+--+-- A common pattern is to define a local helper function:+--+-- @+-- where+-- logg = writeLog logger+-- @+--+writeLog :: Logger -> LogLevel -> T.Text -> IO ()+writeLog !logger !level !msg+ | level <= _loggerLevel logger = mask_ $ do+ -- Nothing in here is expected to block or throw. So the mask should be+ -- sufficient. (TODO what about the getSystemTime?)+ c <- readIORef (_loggerApproxQueueSize logger)+ if (c > _loggerMaxQueueSize logger)+ then incrementCounter (_loggerSkipped logger)+ else do+ skipped <- resetCounter (_loggerSkipped logger)+ now <- getSystemTime+ when (skipped > 0) $ do+ writeChan (_loggerQueue logger) $ skippedMessage now skipped+ incrementCounter (_loggerApproxQueueSize logger)+ writeChan (_loggerQueue logger) $ LogMessage+ { _logMessageText = msg+ , _logMessageLevel = level+ , _logMessageTime = now+ , _logMessageTags = _loggerTags logger+ }+ incrementCounter (_loggerApproxQueueSize logger)++ | otherwise = return ()+ where+ skippedMessage now skipped = LogMessage+ { _logMessageText = "Skipped " <> T.pack (show skipped) <> " log messages."+ , _logMessageLevel = Error+ , _logMessageTime = now+ , _logMessageTags = _loggerTags logger+ }++-- -------------------------------------------------------------------------- --+-- Logging Backend Worker++-- | Initialize a logger context and start a background worker for writing+-- log messages to stdout.+--+withLogger :: LogLevel -> (Logger -> IO a) -> IO a+withLogger !level inner = do+ queue <- newChan+ sizeRef <- newIORef 0+ skippedRef <- newIORef 0+ r <- race (backend queue sizeRef) $ inner $ Logger+ { _loggerQueue = queue+ , _loggerLevel = level+ , _loggerTags = []+ , _loggerApproxQueueSize = sizeRef+ , _loggerSkipped = skippedRef+ , _loggerMaxQueueSize = maxLoggerQueueSize+ }+ case r of+ Left () -> error "logger existed unexpectedly"+ Right a -> return a+ where+ -- TODO: implement batch processing+ --+ backend queue sizeRef = forever $ do+ -- No need for masking here. If anything throws in here we exit the+ -- application anyways.+ --+ -- TODO: there is no way to check if Chan is empty. Use a different+ -- queue that allows to check that sizeRef and the Queue are+ -- approximately in sync.+ --+ msg <- readChan queue+ decrementCounter sizeRef+ T.putStrLn $ formatLogMessage msg+
+ src/Worker.hs view
@@ -0,0 +1,106 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- |+-- Module: Worker+-- Copyright: Copyright © 2020 Kadena LLC.+-- License: MIT+-- Maintainer: Lars Kuhtz <lars@kadena.io>+-- Stability: experimental+--+-- The type of a mining worker+--+module Worker+(+-- * Hash Target+ Target(..)+, encodeTarget+, decodeTarget+, targetToText16++-- * Mining Work+, Work(..)+, encodeWork+, decodeWork++-- * Nonce+, Nonce(..)++-- * Mining Worker+, Worker+) where++import qualified Data.ByteArray.Encoding as BA+import Data.Bytes.Get+import Data.Bytes.Put+import qualified Data.ByteString.Short as BS+import Data.Hashable+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import Data.Word++import GHC.Generics++-- -------------------------------------------------------------------------- --+-- Hash Target++-- | Hash target. A little endian encoded 256 bit (unsigned) word.+--+-- Cf. https://github.com/kadena-io/chainweb-node/wiki/Block-Header-Binary-Encoding#work-header-binary-format+--+newtype Target = Target { _targetBytes :: BS.ShortByteString }+ deriving (Show, Eq, Ord, Generic)+ deriving newtype (Hashable)++decodeTarget :: MonadGet m => m Target+decodeTarget = Target . BS.toShort <$> getBytes 32+{-# INLINE decodeTarget #-}++encodeTarget :: MonadPut m => Target -> m ()+encodeTarget (Target b) = putByteString $ BS.fromShort b+{-# INLINE encodeTarget #-}++-- | Represent target bytes in hexadecimal base+--+targetToText16 :: Target -> T.Text+targetToText16 = T.decodeUtf8 . BA.convertToBase BA.Base16 . BS.fromShort . _targetBytes+{-# INLINE targetToText16 #-}++-- -------------------------------------------------------------------------- --+-- Work++-- | Work bytes. The last 8 bytes are the nonce that is updated by the miner+-- while solving the work.+--+-- Cf. https://github.com/kadena-io/chainweb-node/wiki/Block-Header-Binary-Encoding#work-header-binary-format+--+newtype Work = Work BS.ShortByteString+ deriving (Show, Eq, Ord, Generic)+ deriving newtype (Hashable)++decodeWork :: MonadGet m => m Work+decodeWork = Work . BS.toShort <$> getBytes 286+{-# INLINE decodeWork #-}++encodeWork :: MonadPut m => Work -> m ()+encodeWork (Work b) = putByteString $ BS.fromShort b+{-# INLINE encodeWork #-}++-- -------------------------------------------------------------------------- --+-- Nonce++-- | POW Nonce+--+newtype Nonce = Nonce Word64+ deriving (Show, Eq, Ord, Generic)+ deriving newtype (Hashable)++-- -------------------------------------------------------------------------- --+-- Mining Worker++type Worker = Nonce -> Target -> Work -> IO Work+
+ src/Worker/CPU.hs view
@@ -0,0 +1,181 @@+{-# LANGUAGE AllowAmbiguousTypes #-}+{-# LANGUAGE BangPatterns #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE TypeApplications #-}++-- |+-- Module: Worker.CPU+-- Copyright: Copyright © 2020 Kadena LLC.+-- License: MIT+-- Maintainer: Lars Kuhtz <lars@kadena.io>+-- Stability: experimental+--+-- TODO+--+module Worker.CPU+( cpuWorker+) where++import Crypto.Hash.IO++import qualified Data.ByteArray as BA+import Data.Bytes.Signed+import qualified Data.ByteString as B+import qualified Data.ByteString.Short as BS+import Data.Int+import Data.IORef+import qualified Data.Memory.Endian as BA+import qualified Data.Text as T+import Data.Time.Clock.System+import Data.Word++import Foreign.Marshal.Alloc (allocaBytes)+import Foreign.Ptr (Ptr, castPtr)+import Foreign.Storable (peekElemOff, pokeByteOff)++import System.LogLevel++-- internal modules++import Logger+import Worker++-- -------------------------------------------------------------------------- --+-- Utils++int :: Integral a => Num b => a -> b+int = fromIntegral++getCurrentTimeMicros :: IO Int64+getCurrentTimeMicros = do+ MkSystemTime secs nanos <- getSystemTime+ return $! secs * 1000000 + (int nanos `div` 1000)++-- | `injectNonce` makes low-level assumptions about the byte layout of a+-- hashed `BlockHeader`. If that layout changes, this functions need to be+-- updated. The assumption allows us to iterate on new nonces quickly.+--+-- Recall: `Nonce` contains a `Word64`, and is thus 8 bytes long.+--+-- See also: https://github.com/kadena-io/chainweb-node/wiki/Block-Header-Binary-Encoding+--+injectNonce :: Nonce -> Ptr Word8 -> IO ()+injectNonce (Nonce n) buf = pokeByteOff buf 278 n+{-# INLINE injectNonce #-}++injectTime :: Int64 -> Ptr Word8 -> IO ()+injectTime t buf = pokeByteOff buf 8 $ encodeTimeToWord64 t+{-# INLINE injectTime #-}++encodeTimeToWord64 :: Int64 -> Word64+encodeTimeToWord64 t = BA.unLE . BA.toLE $ unsigned t+{-# INLINE encodeTimeToWord64 #-}++-- | `PowHashNat` interprets POW hashes as unsigned 256 bit integral numbers in+-- little endian encoding, hence we compare against the target from the end of+-- the bytes first, then move toward the front 8 bytes at a time.+--+fastCheckTarget :: Ptr Word64 -> Ptr Word64 -> IO Bool+fastCheckTarget !trgPtr !powPtr =+ fastCheckTargetN 3 trgPtr powPtr >>= \case+ LT -> return False+ GT -> return True+ EQ -> fastCheckTargetN 2 trgPtr powPtr >>= \case+ LT -> return False+ GT -> return True+ EQ -> fastCheckTargetN 1 trgPtr powPtr >>= \case+ LT -> return False+ GT -> return True+ EQ -> fastCheckTargetN 0 trgPtr powPtr >>= \case+ LT -> return False+ GT -> return True+ EQ -> return True+{-# INLINE fastCheckTarget #-}++-- | Recall that `peekElemOff` acts like `drop` for the size of the type in+-- question. Here, this is `Word64`. Since our hash is treated as a `Word256`,+-- each @n@ knocks off a `Word64`'s worth of bytes, and there would be 4 such+-- sections (64 * 4 = 256).+--+-- This must never be called for @n >= 4@.+--+fastCheckTargetN :: Int -> Ptr Word64 -> Ptr Word64 -> IO Ordering+fastCheckTargetN n trgPtr powPtr = compare+ <$> peekElemOff trgPtr n+ <*> peekElemOff powPtr n+{-# INLINE fastCheckTargetN #-}++-- -------------------------------------------------------------------------- --+-- Worker++-- | Single threaded CPU mining worker for Chainweb.+--+-- TODO: Check the chainweb version to make sure this function can handle the+-- respective version.+--+cpuWorker+ :: forall a+ . HashAlgorithm a+ => Logger+ -> Worker+cpuWorker logger orig@(Nonce o) target work = do+ nonces <- newIORef 0+ BA.withByteArray tbytes $ \trgPtr -> do+ !ctx <- hashMutableInit @a+ new <- BA.copy hbytes $ \buf ->+ allocaBytes (powSize :: Int) $ \pow -> do++ -- inner mining loop+ --+ let go1 0 n = return (Just n)+ go1 !i !n@(Nonce nv) = do+ -- Compute POW hash for the nonce+ injectNonce n buf+ hash ctx buf pow++ -- check whether the nonce meets the target+ fastCheckTarget trgPtr (castPtr pow) >>= \case+ True -> Nothing <$ writeIORef nonces (nv - o)+ False -> go1 (i - 1) (Nonce $! nv + 1)++ -- outer loop+ -- Estimates how many iterations of the inner loop run in one second. It runs the inner loop+ -- that many times and injects an updated creation time in each cycle.+ let go0 :: Int -> Int64 {- microseconds -} -> Nonce -> IO ()+ go0 x t !n = do+ injectTime t buf+ go1 x n >>= \case+ Nothing -> return ()+ Just n' -> do+ t' <- getCurrentTimeMicros+ let td = t' - t+ x' = round @Double (int x * 1000000 / int td) -- target 1 second+ go0 x' t' n'++ -- Start outer mining loop+ t <- getCurrentTimeMicros+ go0 100000 t orig+ attempts <- readIORef nonces+ writeLog logger Info $ "Solved header with " <> T.pack (show attempts) <> " attempts."+ return (Work $ BS.toShort new)+ where+ tbytes = let (Target b) = target in BS.fromShort b+ hbytes = let (Work b) = work in BS.fromShort b++ bufSize :: Int+ !bufSize = B.length hbytes++ powSize :: Int+ !powSize = hashDigestSize @a undefined++ -- Compute POW hash+ hash :: MutableContext a -> Ptr Word8 -> Ptr Word8 -> IO ()+ hash ctx buf pow = do+ hashMutableReset ctx+ BA.withByteArray ctx $ \ctxPtr -> do+ hashInternalUpdate @a ctxPtr buf $ fromIntegral bufSize+ hashInternalFinalize ctxPtr $ castPtr pow+ {-# INLINE hash #-}+
+ src/Worker/External.hs view
@@ -0,0 +1,131 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- |+-- Module: Worker.External+-- Copyright: Copyright © 2020 Kadena LLC.+-- License: MIT+-- Maintainer: Lars Kuhtz <lars@kadena.io>+-- Stability: experimental+--+-- TODO+--+module Worker.External+( externalWorker+) where++import Control.Concurrent.Async+import Control.Monad.Catch++import qualified Data.ByteArray.Encoding as BA+import qualified Data.ByteString as B+import qualified Data.ByteString.Char8 as B8+import qualified Data.ByteString.Short as BS+import Data.Char+import qualified Data.Text as T+import qualified Data.Text.Encoding as T+import qualified Data.Text.IO as T++import GHC.Generics++import System.Exit+import System.IO+import System.LogLevel+import qualified System.Process as P++-- internal modules++import Logger+import Worker++-- -------------------------------------------------------------------------- --+-- Exceptions++newtype ExternalWorkerException = ExternalWorkerException T.Text+ deriving (Show, Eq, Ord, Generic)++instance Exception ExternalWorkerException++-- -------------------------------------------------------------------------- --+--++-- | Run an external worker:+--+-- External workers is an external operating system process that is provided by+-- it's file system path.+--+-- It is invoked with+--+-- * the target as first parameter (hex bytes in little endian encoding),+-- followed by+-- * what ever extra arguments are configured by the user.+--+-- The work bytes are provided to stdin as raw bytes.+--+-- On finding a solution it is expected to print the nonce (encoded in hex in+-- big endian byte order) and to exit with an exit code of 0.+--+-- In case of an exit code other than zero any outputs are logged and discarded.+--+externalWorker+ :: Logger+ -> String+ -- ^ worker command+ -> Worker+externalWorker logger cmd _nonce target (Work work) =+ withLogTag logger "Worker" $ \workerLogger ->+ P.withCreateProcess workerProc (go workerLogger)+ where+ targetArg = T.unpack $ targetToText16 target++ workerProc = (P.shell $ cmd <> " " <> targetArg)+ { P.std_in = P.CreatePipe+ , P.std_out = P.CreatePipe+ , P.std_err = P.CreatePipe+ }++ go workerLogger (Just hstdin) (Just hstdout) (Just hstderr) ph = do+ B.hPut hstdin $ BS.fromShort work+ hClose hstdin+ writeLog logger Info "send command to external worker"+ writeLog logger Debug $ "external worker command: " <> T.pack (cmd <> " " <> targetArg)+ withAsync (B.hGetContents hstdout) $ \stdoutThread ->+ withAsync (errThread workerLogger hstderr) $ \stderrThread -> do+ code <- P.waitForProcess ph+ (outbytes, _) <- (,) <$> wait stdoutThread <*> wait stderrThread+ writeLog logger Info "received nonce for solved work from external worker"+ if (code /= ExitSuccess)+ then+ throwM $ ExternalWorkerException $+ "External worker failed with code: " <> (T.pack . show) code+ -- FIXME: handle non-printable characters+ else do+ nonceB16 <- case B8.splitWith isSpace outbytes of+ [] -> throwM $ ExternalWorkerException $+ "expected nonce from miner, got: " <> T.decodeUtf8 outbytes+ -- FIXME: handle non-printable characters+ (a:_) -> return a++ -- reverse -- we want little-endian+ case BA.convertFromBase BA.Base16 nonceB16 of+ Left e -> throwM $ ExternalWorkerException $+ "failed to decode nonce bytes: " <> (T.pack . show) e+ Right bs+ | B.length bs /= 8 ->+ throwM $ ExternalWorkerException "process returned short nonce"+ | otherwise -> return $ Work $!+ BS.toShort $ B.take 278 (BS.fromShort work) <> B.reverse bs++ go _ _ _ _ _ = throwM $ ExternalWorkerException $+ "impossible: process is opened with CreatePipe in/out/err"++ errThread l hstderr = next+ where+ next = hIsEOF hstderr >>= \case+ True -> return ()+ False -> do+ T.hGetLine hstderr >>= writeLog l Debug+ next+
+ src/Worker/Simulation.hs view
@@ -0,0 +1,85 @@+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DerivingStrategies #-}+{-# LANGUAGE GeneralizedNewtypeDeriving #-}+{-# LANGUAGE LambdaCase #-}+{-# LANGUAGE NumericUnderscores #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE ScopedTypeVariables #-}++-- |+-- Module: Worker.Simulation+-- Copyright: Copyright © 2020 Kadena LLC.+-- License: MIT+-- Maintainer: Lars Kuhtz <lars@kadena.io>+-- Stability: experimental+--+-- Simulation Mining Worker+--+-- A fake mining worker that is not actually doing any work. It calculates the+-- solve time base on the assumed hash power of the worker thread and returns+-- the work bytes unchanged after that time has passed.+--+module Worker.Simulation+( HashRate(..)+, defaultHashRate+, simulationWorker+) where++import Control.Concurrent (threadDelay)++import Data.Aeson+import qualified Data.ByteString.Short as BS+import Data.Hashable+import qualified Data.Text as T++import GHC.Generics++import System.LogLevel+import qualified System.Random.MWC as MWC+import qualified System.Random.MWC.Distributions as MWC++-- internal modules++import Logger+import Worker++-- -------------------------------------------------------------------------- --+-- HashRate++newtype HashRate = HashRate Double+ deriving (Show, Eq, Ord, Generic)+ deriving newtype (Read, Num, Fractional, Floating, Real, Enum, Hashable, ToJSON, FromJSON)++-- | Default is 1MH+defaultHashRate :: HashRate+defaultHashRate = 1_000_000++-- -------------------------------------------------------------------------- --+-- Simulation Mining Worker++-- | A fake mining worker that is not actually doing any work. It calculates the+-- solve time base on the assumed hash power of the worker thread and returns+-- the work bytes unchanged after that time has passed.+--+simulationWorker :: Logger -> MWC.GenIO -> HashRate -> Worker+simulationWorker logger rng rate _nonce (Target targetBytes) work = do+ delay <- round <$> MWC.exponential scale rng+ logg Info $ "solve time (microseconds): " <> T.pack (show delay)+ threadDelay delay+ return work+ where+ logg = writeLog logger++ -- expectedMicros = 1_000_000 * difficulty / rate++ -- MWC.exponential is parameterized by the rate, i.e. 1 / expected_time+ scale = realToFrac $ realToFrac rate / (difficulty * 1_000_000)++ -- the expected number of attempts for solving a target is the difficulty.+ difficulty :: Rational+ difficulty = 2^(256 :: Integer) / targetNum++ -- Target is an little endian encoded (unsigned) 256 bit word.+ targetNum :: Rational+ targetNum = foldr (\b a -> fromIntegral b + 256 * a) 0 $ BS.unpack $ targetBytes+