From 70145bc5446570bf245fb6a108612b7606073593 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 20 Aug 2020 15:58:35 +0200 Subject: [PATCH 01/43] bump nixpkgs revision --- default.nix | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/default.nix b/default.nix index 4e77a05..cea4aa3 100644 --- a/default.nix +++ b/default.nix @@ -14,7 +14,7 @@ let name = "nixpkgs-pinned"; url = https://github.com/NixOS/nixpkgs/; ref = "refs/heads/release-20.03"; - rev = "076c67fdea6d0529a568c7d0e0a72e6bc161ecf5"; + rev = "de3780b937d2984f9b5e20d191f23be4f857b3aa"; }) { # Pass no config for purity config = {}; From 1aee41db88945ec37afebb37c78b288812c4e570 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 31 Aug 2020 13:37:40 +0200 Subject: [PATCH 02/43] enable compiler optimisation --- Hash2Pub.cabal | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index 2d195e3..2953d97 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -47,7 +47,7 @@ extra-source-files: CHANGELOG.md common deps build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays - ghc-options: -Wall -Wpartial-fields + ghc-options: -Wall -Wpartial-fields -O2 From 59beb3441f10623ea614c48777e676aefba40382 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 2 Sep 2020 21:37:01 +0200 Subject: [PATCH 03/43] instrumentation script executes the prepared schedule - reads CSV schedule from file - sends the given schedule of post events - not thoroughly tested yet implements #59 --- app/Experiment.hs | 71 ++++++++++++++++++++++++++--------------------- 1 file changed, 39 insertions(+), 32 deletions(-) diff --git a/app/Experiment.hs b/app/Experiment.hs index deb4cae..ffa8869 100644 --- a/app/Experiment.hs +++ b/app/Experiment.hs @@ -3,42 +3,49 @@ module Main where import Control.Concurrent -import Control.Monad (forM_) -import Control.Monad.IO.Class -import Control.Monad.State.Class -import Control.Monad.State.Strict (evalStateT) -import qualified Network.HTTP.Client as HTTP -import System.Random +import Control.Monad (forM_) +import qualified Data.Text.Lazy as Txt +import qualified Data.Text.Lazy.IO as TxtI +import qualified Network.HTTP.Client as HTTP +import System.Environment (getArgs) -import Hash2Pub.PostService (Hashtag, clientPublishPost) +import Hash2Pub.PostService (Hashtag, clientPublishPost) --- placeholder post data definition - -tagsToPostTo = [ "JustSomeTag", "WantAnotherTag234", "HereWeGoAgain", "Oyä", "通信端末" ] - -knownRelays :: [(String, Int)] -knownRelays = - [ ("animalliberation.social", 3342) - , ("hostux.social", 3343) - , ("social.diskseven.com", 3344) - , ("social.imirhil.fr", 3345) - ] +-- configuration constants +timelineFile = "../simulationData/inputs/generated/timeline_sample.csv" main :: IO () main = do + -- read CLI parameters + speedupStr : _ <- getArgs + -- read and parse timeline schedule + -- relying on lazyness of HaskellIO, hoping it does not introduce too strong delays + postEvents <- parseSchedule <$> TxtI.readFile timelineFile + -- actually schedule and send the post events + executeSchedule (read speedupStr) postEvents + pure () + + + +parseSchedule :: Txt.Text + -> [(Int, Hashtag, (String, Int))] -- ^ [(delay in microseconds, hashtag, (hostname, port))] +parseSchedule = fmap (parseEntry . Txt.split (== ';')) . Txt.lines + where + parseEntry [delayT, contactT, tag] = + (read $ Txt.unpack delayT, tag, read $ Txt.unpack contactT) + parseEntry _ = error "invalid schedule input format" + +executeSchedule :: Int -- ^ speedup factor + -> [(Int, Hashtag, (String, Int))] -- ^ [(delay in microseconds, hashtag, (hostname, port))] + -> IO () +executeSchedule speedup events = do -- initialise HTTP manager httpMan <- HTTP.newManager HTTP.defaultManagerSettings - -- initialise RNG - let initRGen = mkStdGen 12 - -- cycle through tags and post to a random instance - evalStateT (forM_ (cycle tagsToPostTo) $ publishPostRandom httpMan) initRGen - -- wait for a specified time - -publishPostRandom :: (RandomGen g, MonadIO m, MonadState g m) => HTTP.Manager -> Hashtag -> m () -publishPostRandom httpman tag = do - index <- state $ randomR (0, length knownRelays - 1) - let (pubHost, pubPort) = knownRelays !! index - _ <- liftIO . forkIO $ do - postResult <- liftIO $ clientPublishPost httpman pubHost pubPort ("foobar #" <> tag) - either putStrLn (const $ pure ()) postResult - liftIO $ threadDelay 500 + forM_ events $ \(delay, tag, (pubHost, pubPort)) -> do + _ <- forkIO $ + clientPublishPost httpMan pubHost pubPort ("foobar #" <> tag) + >>= either putStrLn (const $ pure ()) + -- while threadDelay gives only minimum delay guarantees, let's hope the + -- additional delays are negligible + -- otherwise: evaluate usage of https://hackage.haskell.org/package/schedule-0.3.0.0/docs/Data-Schedule.html + threadDelay $ delay `div` speedup From 20050654bce352e21520995d8524d4060b486f65 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 3 Sep 2020 11:20:38 +0200 Subject: [PATCH 04/43] make passing bootstrap information optional reason: allow the first node to start without having to wait for a timeout part of #58 --- app/Main.hs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/app/Main.hs b/app/Main.hs index 3bdb4d4..4810fb0 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -45,13 +45,19 @@ main = do readConfig :: IO (FediChordConf, ServiceConf) readConfig = do - confDomainString : ipString : portString : bootstrapHost : bootstrapPortString : servicePortString : speedup : _ <- getArgs + confDomainString : ipString : portString : servicePortString : speedup : remainingArgs <- getArgs + -- allow starting the initial node without bootstrapping info to avoid + -- waiting for timeout let + confBootstrapNodes' = case remainingArgs of + bootstrapHost : bootstrapPortString : _ -> + [(bootstrapHost, read bootstrapPortString)] + _ -> [] fConf = FediChordConf { confDomain = confDomainString , confIP = toHostAddress6 . read $ ipString , confDhtPort = read portString - , confBootstrapNodes = [(bootstrapHost, read bootstrapPortString)] + , confBootstrapNodes = confBootstrapNodes' --, confStabiliseInterval = 60 , confBootstrapSamplingInterval = 180 , confMaxLookupCacheAge = 300 From 4f08d33d2eb78d576308b4e3d0d2ac9de690f256 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Fri, 4 Sep 2020 11:08:40 +0200 Subject: [PATCH 05/43] make all delays configurable and scale them according to a speedup factor --- app/Main.hs | 26 +++++++++++--------- src/Hash2Pub/FediChord.hs | 43 ++++++++++++++-------------------- src/Hash2Pub/FediChordTypes.hs | 12 ++++++++-- 3 files changed, 43 insertions(+), 38 deletions(-) diff --git a/app/Main.hs b/app/Main.hs index 4810fb0..5f42f09 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -45,27 +45,31 @@ main = do readConfig :: IO (FediChordConf, ServiceConf) readConfig = do - confDomainString : ipString : portString : servicePortString : speedup : remainingArgs <- getArgs + confDomainString : ipString : portString : servicePortString : speedupString : remainingArgs <- getArgs -- allow starting the initial node without bootstrapping info to avoid -- waiting for timeout let + speedup = read speedupString confBootstrapNodes' = case remainingArgs of bootstrapHost : bootstrapPortString : _ -> [(bootstrapHost, read bootstrapPortString)] _ -> [] fConf = FediChordConf { - confDomain = confDomainString - , confIP = toHostAddress6 . read $ ipString - , confDhtPort = read portString - , confBootstrapNodes = confBootstrapNodes' - --, confStabiliseInterval = 60 - , confBootstrapSamplingInterval = 180 - , confMaxLookupCacheAge = 300 + confDomain = confDomainString + , confIP = toHostAddress6 . read $ ipString + , confDhtPort = read portString + , confBootstrapNodes = confBootstrapNodes' + , confStabiliseInterval = 60 * 10^6 + , confBootstrapSamplingInterval = 180 * 10^6 `div` speedup + , confMaxLookupCacheAge = 300 / fromIntegral speedup + , confJoinAttemptsInterval = 60 * 10^6 `div` speedup + , confMaxNodeCacheAge = 600 / fromIntegral speedup + , confResponsePurgeAge = 60 / fromIntegral speedup } sConf = ServiceConf { - confSubscriptionExpiryTime = fromIntegral $ 2*3600 `div` (read speedup :: Integer) - , confServicePort = read servicePortString - , confServiceHost = confDomainString + confSubscriptionExpiryTime = fromIntegral $ 2*3600 `div` speedup + , confServicePort = read servicePortString + , confServiceHost = confDomainString } pure (fConf, sConf) diff --git a/src/Hash2Pub/FediChord.hs b/src/Hash2Pub/FediChord.hs index 45d0bf9..d1568e4 100644 --- a/src/Hash2Pub/FediChord.hs +++ b/src/Hash2Pub/FediChord.hs @@ -199,7 +199,7 @@ convergenceSampleThread nsSTM = forever $ do -- unjoined node: try joining through all bootstrapping nodes else tryBootstrapJoining nsSTM >> pure () let delaySecs = confBootstrapSamplingInterval . nodeConfig $ parentNode - threadDelay $ delaySecs * 10^6 + threadDelay delaySecs -- | Try joining the DHT through any of the bootstrapping nodes until it succeeds. @@ -310,12 +310,13 @@ joinOnNewEntriesThread nsSTM = loop where loop = do nsSnap <- readTVarIO nsSTM - (lookupResult, cache) <- atomically $ do + (lookupResult, parentNode) <- atomically $ do cache <- readTVar $ nodeCacheSTM nsSnap + parentNode <- readTVar $ parentRealNode nsSnap case queryLocalCache nsSnap cache 1 (getNid nsSnap) of -- empty cache, block until cache changes and then retry (FORWARD s) | Set.null s -> retry - result -> pure (result, cache) + result -> pure (result, parentNode) case lookupResult of -- already joined FOUND _ -> @@ -325,8 +326,7 @@ joinOnNewEntriesThread nsSTM = loop joinResult <- runExceptT $ fediChordVserverJoin nsSTM either -- on join failure, sleep and retry - -- TODO: make delay configurable - (const $ threadDelay (30 * 10^6) >> loop) + (const $ threadDelay (confJoinAttemptsInterval . nodeConfig $ parentNode) >> loop) (const $ pure ()) joinResult @@ -341,20 +341,16 @@ nodeCacheWriter nsSTM = modifyTVar' (nodeCacheSTM ns) cacheModifier --- TODO: make max entry age configurable -maxEntryAge :: POSIXTime -maxEntryAge = 600 - - -- | Periodically iterate through cache, clean up expired entries and verify unverified ones nodeCacheVerifyThread :: LocalNodeStateSTM s -> IO () nodeCacheVerifyThread nsSTM = forever $ do putStrLn "cache verify run: begin" -- get cache - (ns, cache) <- atomically $ do + (ns, cache, maxEntryAge) <- atomically $ do ns <- readTVar nsSTM cache <- readTVar $ nodeCacheSTM ns - pure (ns, cache) + maxEntryAge <- confMaxNodeCacheAge . nodeConfig <$> readTVar (parentRealNode ns) + pure (ns, cache, maxEntryAge) -- iterate entries: -- for avoiding too many time syscalls, get current time before iterating. now <- getPOSIXTime @@ -402,7 +398,7 @@ nodeCacheVerifyThread nsSTM = forever $ do ) putStrLn "cache verify run: end" - threadDelay $ 10^6 * round maxEntryAge `div` 20 + threadDelay $ fromEnum (maxEntryAge / 20) `div` 10^6 -- convert from pico to milliseconds -- | Checks the invariant of at least @jEntries@ per cache slice. @@ -548,8 +544,8 @@ stabiliseThread nsSTM = forever $ do newPredecessor putStrLn "stabilise run: end" - -- TODO: make delay configurable - threadDelay (60 * 10^6) + stabiliseDelay <- confStabiliseInterval . nodeConfig <$> readTVarIO (parentRealNode newNs) + threadDelay stabiliseDelay where -- | send a stabilise request to the n-th neighbour -- (specified by the provided getter function) and on failure retry @@ -636,19 +632,15 @@ type RequestMap = Map.Map (SockAddr, Integer) RequestMapEntry data RequestMapEntry = RequestMapEntry (Set.Set FediChordMessage) (Maybe Integer) POSIXTime --- TODO: make purge age configurable --- | periodically clean up old request parts -responsePurgeAge :: POSIXTime -responsePurgeAge = 60 -- seconds -requestMapPurge :: MVar RequestMap -> IO () -requestMapPurge mapVar = forever $ do +requestMapPurge :: POSIXTime -> MVar RequestMap -> IO () +requestMapPurge purgeAge mapVar = forever $ do rMapState <- takeMVar mapVar now <- getPOSIXTime putMVar mapVar $ Map.filter (\(RequestMapEntry _ _ ts) -> - now - ts < responsePurgeAge + now - ts < purgeAge ) rMapState - threadDelay $ round responsePurgeAge * 2 * 10^6 + threadDelay $ (fromEnum purgeAge * 2) `div` 10^6 -- | Wait for messages, deserialise them, manage parts and acknowledgement status, @@ -663,12 +655,13 @@ fediMessageHandler sendQ recvQ nsSTM = do -- not change. -- Other functions are passed the nsSTM reference and thus can get the latest state. nsSnap <- readTVarIO nsSTM + nodeConf <- nodeConfig <$> readTVarIO (parentRealNode nsSnap) -- handling multipart messages: -- Request parts can be insert into a map (key: (sender IP against spoofing, request ID), value: timestamp + set of message parts, handle all of them when size of set == parts) before being handled. This map needs to be purged periodically by a separate thread and can be protected by an MVar for fairness. requestMap <- newMVar (Map.empty :: RequestMap) -- run receive loop and requestMapPurge concurrently, so that an exception makes -- both of them fail - concurrently_ (requestMapPurge requestMap) $ forever $ do + concurrently_ (requestMapPurge (confResponsePurgeAge nodeConf) requestMap) $ forever $ do -- wait for incoming messages (rawMsg, sourceAddr) <- atomically $ readTQueue recvQ let aMsg = deserialiseMessage rawMsg @@ -807,4 +800,4 @@ lookupCacheCleanup nodeSTM = do now - ts < confMaxLookupCacheAge (nodeConfig node) ) ) - threadDelay $ round (confMaxLookupCacheAge $ nodeConfig node) * (10^5) + threadDelay $ fromEnum (2 * confMaxLookupCacheAge (nodeConfig node)) `div` 10^6 diff --git a/src/Hash2Pub/FediChordTypes.hs b/src/Hash2Pub/FediChordTypes.hs index cbd3a58..46668ee 100644 --- a/src/Hash2Pub/FediChordTypes.hs +++ b/src/Hash2Pub/FediChordTypes.hs @@ -411,10 +411,18 @@ data FediChordConf = FediChordConf -- ^ listening port for the FediChord DHT , confBootstrapNodes :: [(String, PortNumber)] -- ^ list of potential bootstrapping nodes + , confStabiliseInterval :: Int + -- ^ pause between stabilise runs, in milliseconds , confBootstrapSamplingInterval :: Int - -- ^ pause between sampling the own ID through bootstrap nodes, in seconds + -- ^ pause between sampling the own ID through bootstrap nodes, in milliseconds , confMaxLookupCacheAge :: POSIXTime - -- ^ maximum age of lookup cache entries in seconds + -- ^ maximum age of key lookup cache entries in seconds + , confJoinAttemptsInterval :: Int + -- ^ interval between join attempts on newly learned nodes, in milliseconds + , confMaxNodeCacheAge :: POSIXTime + -- ^ maximum age of entries in the node cache, in milliseconds + , confResponsePurgeAge :: POSIXTime + -- ^ maximum age of message parts in response part cache, in seconds } deriving (Show, Eq) From c9b0e6611061af4231f691d6a03da19d9e27a38d Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Fri, 4 Sep 2020 16:53:48 +0200 Subject: [PATCH 06/43] scale request timeout with speedup and pass it directly to function --- app/Main.hs | 2 ++ src/Hash2Pub/DHTProtocol.hs | 49 ++++++++++++++++------------------ src/Hash2Pub/FediChord.hs | 5 ++-- src/Hash2Pub/FediChordTypes.hs | 4 +++ 4 files changed, 32 insertions(+), 28 deletions(-) diff --git a/app/Main.hs b/app/Main.hs index 5f42f09..c08cd3c 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -65,6 +65,8 @@ readConfig = do , confJoinAttemptsInterval = 60 * 10^6 `div` speedup , confMaxNodeCacheAge = 600 / fromIntegral speedup , confResponsePurgeAge = 60 / fromIntegral speedup + , confRequestTimeout = 5 * 10^6 `div` speedup + , confRequestRetries = 3 } sConf = ServiceConf { confSubscriptionExpiryTime = fromIntegral $ 2*3600 `div` speedup diff --git a/src/Hash2Pub/DHTProtocol.hs b/src/Hash2Pub/DHTProtocol.hs index fa5a54a..8258ca3 100644 --- a/src/Hash2Pub/DHTProtocol.hs +++ b/src/Hash2Pub/DHTProtocol.hs @@ -488,10 +488,11 @@ requestJoin :: (NodeState a, Service s (RealNodeSTM s)) => a -- ^ cu requestJoin toJoinOn ownStateSTM = do ownState <- readTVarIO ownStateSTM prn <- readTVarIO $ parentRealNode ownState - srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ownState) + nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ownState) + let srcAddr = confIP nodeConf bracket (mkSendSocket srcAddr (getDomain toJoinOn) (getDhtPort toJoinOn)) close (\sock -> do -- extract own state for getting request information - responses <- sendRequestTo (\rid -> Request rid (toRemoteNodeState ownState) 1 True Join (Just JoinRequestPayload)) sock + responses <- sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid -> Request rid (toRemoteNodeState ownState) 1 True Join (Just JoinRequestPayload)) sock (cacheInsertQ, joinedState) <- atomically $ do stateSnap <- readTVar ownStateSTM let @@ -584,10 +585,11 @@ sendQueryIdMessages targetID ns lParam targets = do -- create connected sockets to all query targets and use them for request handling - srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns) + nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns) + let srcAddr = confIP nodeConf -- ToDo: make attempts and timeout configurable queryThreads <- mapM (\resultNode -> async $ bracket (mkSendSocket srcAddr (getDomain resultNode) (getDhtPort resultNode)) close ( - sendRequestTo (lookupMessage targetID ns Nothing) + sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (lookupMessage targetID ns Nothing) )) targets -- ToDo: process results immediately instead of waiting for the last one to finish, see https://stackoverflow.com/a/38815224/9198613 -- ToDo: exception handling, maybe log them @@ -635,8 +637,9 @@ requestStabilise :: LocalNodeState s -- ^ sending node -> RemoteNodeState -- ^ neighbour node to send to -> IO (Either String ([RemoteNodeState], [RemoteNodeState])) -- ^ (predecessors, successors) of responding node requestStabilise ns neighbour = do - srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns) - responses <- bracket (mkSendSocket srcAddr (getDomain neighbour) (getDhtPort neighbour)) close (fmap Right . sendRequestTo (\rid -> + nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns) + let srcAddr = confIP nodeConf + responses <- bracket (mkSendSocket srcAddr (getDomain neighbour) (getDhtPort neighbour)) close (fmap Right . sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid -> Request { requestID = rid , sender = toRemoteNodeState ns @@ -673,13 +676,14 @@ requestLeave :: LocalNodeState s -> RemoteNodeState -- target node -> IO (Either String ()) -- error or success requestLeave ns doMigration target = do - srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns) - let leavePayload = LeaveRequestPayload { + nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns) + let srcAddr = confIP nodeConf + leavePayload = LeaveRequestPayload { leaveSuccessors = successors ns , leavePredecessors = predecessors ns , leaveDoMigration = doMigration } - responses <- bracket (mkSendSocket srcAddr (getDomain target) (getDhtPort target)) close (fmap Right . sendRequestTo (\rid -> + responses <- bracket (mkSendSocket srcAddr (getDomain target) (getDhtPort target)) close (fmap Right . sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid -> Request { requestID = rid , sender = toRemoteNodeState ns @@ -701,10 +705,11 @@ requestPing :: LocalNodeState s -- ^ sending node -> RemoteNodeState -- ^ node to be PINGed -> IO (Either String [RemoteNodeState]) -- ^ all active vServers of the pinged node requestPing ns target = do - srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns) + nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns) + let srcAddr = confIP nodeConf responses <- bracket (mkSendSocket srcAddr (getDomain target) (getDhtPort target)) close (\sock -> do - resp <- sendRequestTo (\rid -> + resp <- sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (\rid -> Request { requestID = rid , sender = toRemoteNodeState ns @@ -740,22 +745,14 @@ requestPing ns target = do ) responses --- | 'sendRequestToWithParams' with default timeout and retries already specified. --- Generic function for sending a request over a connected socket and collecting the response. --- Serialises the message and tries to deliver its parts for a number of attempts within a default timeout. -sendRequestTo :: (Integer -> FediChordMessage) -- ^ the message to be sent, still needing a requestID - -> Socket -- ^ connected socket to use for sending - -> IO (Set.Set FediChordMessage) -- ^ responses -sendRequestTo = sendRequestToWithParams 5000 3 - -- | Generic function for sending a request over a connected socket and collecting the response. -- Serialises the message and tries to deliver its parts for a number of attempts within a specified timeout. -sendRequestToWithParams :: Int -- ^ timeout in milliseconds - -> Int -- ^ number of retries - -> (Integer -> FediChordMessage) -- ^ the message to be sent, still needing a requestID - -> Socket -- ^ connected socket to use for sending - -> IO (Set.Set FediChordMessage) -- ^ responses -sendRequestToWithParams timeoutMillis numAttempts msgIncomplete sock = do +sendRequestTo :: Int -- ^ timeout in milliseconds + -> Int -- ^ number of retries + -> (Integer -> FediChordMessage) -- ^ the message to be sent, still needing a requestID + -> Socket -- ^ connected socket to use for sending + -> IO (Set.Set FediChordMessage) -- ^ responses +sendRequestTo timeoutMillis numAttempts msgIncomplete sock = do -- give the message a random request ID randomID <- randomRIO (0, 2^32-1) let @@ -764,7 +761,7 @@ sendRequestToWithParams timeoutMillis numAttempts msgIncomplete sock = do -- create a queue for passing received response messages back, even after a timeout responseQ <- newTBQueueIO $ 2*maximumParts -- keep room for duplicate packets -- start sendAndAck with timeout - attempts numAttempts . timeout (timeoutMillis*1000) $ sendAndAck responseQ sock requests + _ <- attempts numAttempts . timeout (timeoutMillis*1000) $ sendAndAck responseQ sock requests -- after timeout, check received responses, delete them from unacked message set/ map and rerun senAndAck with that if necessary. recvdParts <- atomically $ flushTBQueue responseQ pure $ Set.fromList recvdParts diff --git a/src/Hash2Pub/FediChord.hs b/src/Hash2Pub/FediChord.hs index d1568e4..33044fa 100644 --- a/src/Hash2Pub/FediChord.hs +++ b/src/Hash2Pub/FediChord.hs @@ -223,10 +223,11 @@ tryBootstrapJoining nsSTM = do bootstrapQueryId :: LocalNodeStateSTM s -> (String, PortNumber) -> NodeID -> IO (Either String RemoteNodeState) bootstrapQueryId nsSTM (bootstrapHost, bootstrapPort) targetID = do ns <- readTVarIO nsSTM - srcAddr <- confIP . nodeConfig <$> readTVarIO (parentRealNode ns) + nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns) + let srcAddr = confIP nodeConf bootstrapResponse <- bracket (mkSendSocket srcAddr bootstrapHost bootstrapPort) close ( -- Initialise an empty cache only with the responses from a bootstrapping node - fmap Right . sendRequestTo (lookupMessage targetID ns Nothing) + fmap Right . sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (lookupMessage targetID ns Nothing) ) `catch` (\e -> pure . Left $ "Error at bootstrap QueryId: " <> displayException (e :: IOException)) diff --git a/src/Hash2Pub/FediChordTypes.hs b/src/Hash2Pub/FediChordTypes.hs index 46668ee..3b563e6 100644 --- a/src/Hash2Pub/FediChordTypes.hs +++ b/src/Hash2Pub/FediChordTypes.hs @@ -423,6 +423,10 @@ data FediChordConf = FediChordConf -- ^ maximum age of entries in the node cache, in milliseconds , confResponsePurgeAge :: POSIXTime -- ^ maximum age of message parts in response part cache, in seconds + , confRequestTimeout :: Int + -- ^ how long to wait until response has arrived, in milliseconds + , confRequestRetries :: Int + -- ^ how often re-sending a timed-out request can be retried } deriving (Show, Eq) From d3e5eac5c5b62c84a8e18989d39ad873a51cf95a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9cate?= Date: Sat, 5 Sep 2020 12:41:18 +0200 Subject: [PATCH 07/43] Unsused imports and syntax error --- src/Hash2Pub/PostService.hs | 7 +++---- src/Hash2Pub/RingMap.hs | 4 ++-- 2 files changed, 5 insertions(+), 6 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 81cf552..7bf0c3c 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -12,15 +12,14 @@ import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM import Control.Exception (Exception (..), try) -import Control.Monad (foldM, forM, forM_, forever, void, - when) +import Control.Monad (foldM, forM_, forever, void, when) import Control.Monad.IO.Class (liftIO) import Data.Bifunctor import qualified Data.ByteString.Lazy.UTF8 as BSUL import qualified Data.ByteString.UTF8 as BSU import qualified Data.HashMap.Strict as HMap import qualified Data.HashSet as HSet -import Data.Maybe (fromMaybe, isJust) +import Data.Maybe (isJust) import Data.String (fromString) import qualified Data.Text.Lazy as Txt import Data.Text.Normalize (NormalizationMode (NFC), normalize) @@ -601,7 +600,7 @@ fetchTagPosts serv = forever $ do -- TODO: batching, retry -- TODO: process multiple in parallel pIdUri <- atomically . readTQueue $ postFetchQueue serv - fetchReq <- HTTP.parseRequest . Txt.unpack $pIdUri + fetchReq <- HTTP.parseRequest . Txt.unpack $ pIdUri resp <- try $ HTTP.httpLbs fetchReq (httpMan serv) :: IO (Either HTTP.HttpException (HTTP.Response BSUL.ByteString)) case resp of Right response -> diff --git a/src/Hash2Pub/RingMap.hs b/src/Hash2Pub/RingMap.hs index e99f8b2..ae1ec15 100644 --- a/src/Hash2Pub/RingMap.hs +++ b/src/Hash2Pub/RingMap.hs @@ -23,7 +23,7 @@ instance (Bounded k, Ord k, Eq a) => Eq (RingMap k a) where a == b = getRingMap a == getRingMap b instance (Bounded k, Ord k, Show k, Show a) => Show (RingMap k a) where - show rmap = shows "RingMap " (show $ getRingMap rmap) + show rmap = shows ("RingMap " :: String) (show $ getRingMap rmap) -- | entry of a 'RingMap' that holds a value and can also -- wrap around the lookup direction at the edges of the name space. @@ -207,7 +207,7 @@ takeEntriesUntil_ :: (Integral i, Bounded k, Ord k) -> Maybe i -- possible number limit -> [a] -> [a] -takeEntriesUntil_ rmap' getterFunc' havingReached previousEntry (Just remaining) takeAcc +takeEntriesUntil_ _rmap' _getterFunc' _havingReached _previousEntry (Just remaining) takeAcc -- length limit reached | remaining <= 0 = takeAcc takeEntriesUntil_ rmap' getterFunc' havingReached previousEntry numLimit takeAcc = From fa78c6fc430d6903e2a777642ac646722f372c96 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Sat, 5 Sep 2020 15:01:14 +0200 Subject: [PATCH 08/43] clarify different nix-shell environments in readme --- Readme.md | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/Readme.md b/Readme.md index 3c7dbe5..daf9e38 100644 --- a/Readme.md +++ b/Readme.md @@ -14,4 +14,7 @@ The ASN.1 module schema used for DHT messages can be found in `FediChord.asn1`. The project and its developent environment are built with [Nix](https://nixos.org/nix/). -The development environment can be entered with `nix-shell`. Then the project can be built with `cabal build` from within the environment, or using `nix-shell --command "cabal build"` to do both steps at once. +The development environment can be entered with `nix-shell shell-minimal.nix`. Then the project can be built with `cabal build` from within the environment, or using `nix-shell --command "cabal build" shell-minimal.nix` to do both steps at once. + +While the `shell-minimal.nix` environment contains everything necessary for building and testing this project, the `shell.nix` additionally contains the Haskell IDE engine *hie* and the documentation for all used Haskell packages for more convenient development. +Be aware that these need to be build from source and can take a very long time to build. From 7d833e064ba414b47f92a89cbd16227e50485c45 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?H=C3=A9cate?= Date: Sat, 5 Sep 2020 13:10:15 +0200 Subject: [PATCH 09/43] Improve readability --- src/Hash2Pub/PostService.hs | 70 +++++++++++++-------------------- src/Hash2Pub/PostService/API.hs | 37 +++++++++++++++++ src/Hash2Pub/ProtocolTypes.hs | 2 - 3 files changed, 64 insertions(+), 45 deletions(-) create mode 100644 src/Hash2Pub/PostService/API.hs diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 7bf0c3c..89c14d2 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -21,6 +21,7 @@ import qualified Data.HashMap.Strict as HMap import qualified Data.HashSet as HSet import Data.Maybe (isJust) import Data.String (fromString) +import Data.Text.Lazy (Text) import qualified Data.Text.Lazy as Txt import Data.Text.Normalize (NormalizationMode (NFC), normalize) import Data.Time.Clock.POSIX @@ -35,6 +36,7 @@ import Servant import Servant.Client import Hash2Pub.FediChordTypes +import Hash2Pub.PostService.API import Hash2Pub.RingMap @@ -47,7 +49,7 @@ data PostService d = PostService -- ^ for each tag store the subscribers + their queue , ownSubscriptions :: TVar (HMap.HashMap NodeID POSIXTime) -- ^ tags subscribed by the own node have an assigned lease time - , ownPosts :: TVar (HSet.HashSet Txt.Text) + , ownPosts :: TVar (HSet.HashSet Text) -- ^ just store the existence of posts for saving memory, , relayInQueue :: TQueue (Hashtag, PostID, PostContent) -- ^ Queue for processing incoming posts of own instance asynchronously @@ -57,9 +59,9 @@ data PostService d = PostService } deriving (Typeable) -type Hashtag = Txt.Text -type PostID = Txt.Text -type PostContent = Txt.Text +type Hashtag = Text +type PostID = Text +type PostContent = Text -- | For each handled tag, store its subscribers and provide a -- broadcast 'TChan' for enqueuing posts type RelayTags = RingMap NodeID (TagSubscribersSTM, TChan PostID, Hashtag) @@ -130,38 +132,13 @@ postServiceApplication :: DHT d => PostService d -> Application postServiceApplication serv = serve exposedPostServiceAPI $ postServer serv --- | needed for guiding type inference -exposedPostServiceAPI :: Proxy PostServiceAPI -exposedPostServiceAPI = Proxy - -- ========= constants =========== -placeholderPost :: Txt.Text +placeholderPost :: Text placeholderPost = Txt.take 5120 . Txt.repeat $ 'O' -- size 5KiB -- ========= HTTP API and handlers ============= -type PostServiceAPI = "relay" :> "inbox" :> Capture "hashtag" Txt.Text :> ReqBody '[PlainText] Txt.Text :> PutCreated '[PlainText] NoContent - -- delivery endpoint at responsible relay for delivering posts of $tag for distribution - :<|> "relay" :> "subscribers" :> Capture "senderID" Integer :> ReqBody '[PlainText] Txt.Text :> PostNoContent '[PlainText] Txt.Text - -- endpoint for delivering the subscriptions and outstanding queue - :<|> "post" :> Capture "postid" Txt.Text :> Get '[PlainText] Txt.Text - -- fetch endpoint for posts, full post ID is http://$domain/post/$postid - :<|> "posts" :> ReqBody '[PlainText] Txt.Text :> Post '[PlainText] Txt.Text - -- endpoint for fetching multiple posts at once - :<|> "posts" :> "inbox" :> ReqBody '[PlainText] Txt.Text :> PutCreated '[PlainText] NoContent - -- delivery endpoint of newly published posts of the relay's instance - :<|> "tags" :> Capture "hashtag" Txt.Text :> ReqBody '[PlainText] Txt.Text :> PostCreated '[PlainText] Txt.Text - -- delivery endpoint for posts of $tag at subscribing instance - :<|> "tags" :> Capture "hashtag" Txt.Text :> "subscribe" :> Header "Origin" Txt.Text :> Get '[PlainText] Integer - -- endpoint for subscribing the instance specified in - -- the Origin header to $hashtag. - -- Returns subscription lease time in seconds. - :<|> "tags" :> Capture "hashtag" Txt.Text :> "unsubscribe" :> Header "Origin" Txt.Text :> Get '[PlainText] Txt.Text - -- endpoint for unsubscribing the instance specified in - -- the Origin header to $hashtag - - postServer :: DHT d => PostService d -> Server PostServiceAPI postServer service = relayInbox service :<|> subscriptionDelivery service @@ -173,7 +150,7 @@ postServer service = relayInbox service :<|> tagUnsubscribe service -relayInbox :: DHT d => PostService d -> Hashtag -> Txt.Text -> Handler NoContent +relayInbox :: DHT d => PostService d -> Hashtag -> Text -> Handler NoContent relayInbox serv tag posts = do let -- skip checking whether the post actually contains the tag, just drop full post @@ -201,7 +178,7 @@ newtype UnhandledTagException = UnhandledTagException String instance Exception UnhandledTagException -subscriptionDelivery :: DHT d => PostService d -> Integer -> Txt.Text -> Handler Txt.Text +subscriptionDelivery :: DHT d => PostService d -> Integer -> Text -> Handler Text subscriptionDelivery serv senderID subList = do let tagSubs = Txt.lines subList @@ -235,7 +212,7 @@ subscriptionDelivery serv senderID subList = do Right _ -> pure "" -- TODO: check and only accept tags in own (future?) responsibility where - processTag :: TVar RelayTags -> Txt.Text -> STM () + processTag :: TVar RelayTags -> Text -> STM () processTag subscriberSTM tagData = do let tag:subText:lease:posts:_ = Txt.splitOn "," tagData @@ -246,7 +223,7 @@ subscriptionDelivery serv senderID subList = do enqueueSubscription subscriberSTM (normaliseTag tag) sub postList leaseTime -postFetch :: PostService d -> Txt.Text -> Handler Txt.Text +postFetch :: PostService d -> Text -> Handler Text postFetch serv postID = do postSet <- liftIO . readTVarIO . ownPosts $ serv if HSet.member postID postSet @@ -255,7 +232,7 @@ postFetch serv postID = do else throwError $ err404 { errBody = "No post found with this ID" } -postMultiFetch :: PostService d -> Txt.Text -> Handler Txt.Text +postMultiFetch :: PostService d -> Text -> Handler Text postMultiFetch serv postIDs = do let idList = Txt.lines postIDs postSet <- liftIO . readTVarIO . ownPosts $ serv @@ -267,7 +244,7 @@ postMultiFetch serv postIDs = do ) "" idList -postInbox :: PostService d -> Txt.Text -> Handler NoContent +postInbox :: PostService d -> Text -> Handler NoContent postInbox serv post = do -- extract contained hashtags let @@ -277,13 +254,13 @@ postInbox serv post = do -- add ID to own posts liftIO . atomically $ modifyTVar' (ownPosts serv) (HSet.insert postId) -- enqueue a relay job for each tag - liftIO $ forM_ (containedTags :: [Txt.Text]) (\tag -> + liftIO $ forM_ (containedTags :: [Text]) (\tag -> atomically $ writeTQueue (relayInQueue serv) (tag, postId, post) ) pure NoContent -tagDelivery :: PostService d -> Txt.Text -> Txt.Text -> Handler Txt.Text +tagDelivery :: PostService d -> Text -> Text -> Handler Text tagDelivery serv hashtag posts = do let postIDs = Txt.lines posts subscriptions <- liftIO . readTVarIO . ownSubscriptions $ serv @@ -294,7 +271,7 @@ tagDelivery serv hashtag posts = do pure () pure $ "Received a postID for tag " <> hashtag -tagSubscribe :: DHT d => PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Integer +tagSubscribe :: DHT d => PostService d -> Text -> Maybe Text -> Handler Integer tagSubscribe serv hashtag origin = do responsible <- liftIO $ isResponsibleFor (baseDHT serv) (hashtagToId hashtag) if not responsible @@ -313,7 +290,7 @@ tagSubscribe serv hashtag origin = do pure $ round leaseTime -tagUnsubscribe :: DHT d => PostService d -> Txt.Text -> Maybe Txt.Text -> Handler Txt.Text +tagUnsubscribe :: DHT d => PostService d -> Text -> Maybe Text -> Handler Text tagUnsubscribe serv hashtag origin = do responsible <- liftIO $ isResponsibleFor (baseDHT serv) (hashtagToId hashtag) if not responsible @@ -333,8 +310,15 @@ tagUnsubscribe serv hashtag origin = do clientAPI :: Proxy PostServiceAPI clientAPI = Proxy - -relayInboxClient :<|> subscriptionDeliveryClient :<|> postFetchClient :<|> postMultiFetchClient :<|> postInboxClient :<|> tagDeliveryClient :<|> tagSubscribeClient :<|> tagUnsubscribeClient = client clientAPI +relayInboxClient :: Text -> Text -> ClientM NoContent +relayInboxClient :<|> subscriptionDeliveryClient + :<|> postFetchClient + :<|> postMultiFetchClient + :<|> postInboxClient + :<|> tagDeliveryClient + :<|> tagSubscribeClient + :<|> tagUnsubscribeClient + = client clientAPI -- | Deliver the subscriber list of all hashtags in the interval [fromTag, toTag] @@ -542,7 +526,7 @@ lookupTagSubscriptions tag = rMapLookup (hashtagToId tag) -- normalise the unicode representation of a string to NFC -normaliseTag :: Txt.Text -> Txt.Text +normaliseTag :: Text -> Text normaliseTag = Txt.fromStrict . normalize NFC . Txt.toStrict diff --git a/src/Hash2Pub/PostService/API.hs b/src/Hash2Pub/PostService/API.hs new file mode 100644 index 0000000..1484631 --- /dev/null +++ b/src/Hash2Pub/PostService/API.hs @@ -0,0 +1,37 @@ +{-# LANGUAGE DataKinds #-} +{-# LANGUAGE FlexibleInstances #-} +{-# LANGUAGE InstanceSigs #-} +{-# LANGUAGE MultiParamTypeClasses #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE TypeOperators #-} +module Hash2Pub.PostService.API where + +import Data.Text.Lazy (Text) + +import Servant + +type PostServiceAPI = "relay" :> "inbox" :> Capture "hashtag" Text :> ReqBody '[PlainText] Text :> PutCreated '[PlainText] NoContent + -- delivery endpoint at responsible relay for delivering posts of $tag for distribution + :<|> "relay" :> "subscribers" :> Capture "senderID" Integer :> ReqBody '[PlainText] Text :> PostNoContent '[PlainText] Text + -- endpoint for delivering the subscriptions and outstanding queue + :<|> "post" :> Capture "postid" Text :> Get '[PlainText] Text + -- fetch endpoint for posts, full post ID is http://$domain/post/$postid + :<|> "posts" :> ReqBody '[PlainText] Text :> Post '[PlainText] Text + -- endpoint for fetching multiple posts at once + :<|> "posts" :> "inbox" :> ReqBody '[PlainText] Text :> PutCreated '[PlainText] NoContent + -- delivery endpoint of newly published posts of the relay's instance + :<|> "tags" :> Capture "hashtag" Text :> ReqBody '[PlainText] Text :> PostCreated '[PlainText] Text + -- delivery endpoint for posts of $tag at subscribing instance + :<|> "tags" :> Capture "hashtag" Text :> "subscribe" :> Header "Origin" Text :> Get '[PlainText] Integer + -- endpoint for subscribing the instance specified in + -- the Origin header to $hashtag. + -- Returns subscription lease time in seconds. + :<|> "tags" :> Capture "hashtag" Text :> "unsubscribe" :> Header "Origin" Text :> Get '[PlainText] Text + -- endpoint for unsubscribing the instance specified in + -- the Origin header to $hashtag + +-- | needed for guiding type inference +exposedPostServiceAPI :: Proxy PostServiceAPI +exposedPostServiceAPI = Proxy + diff --git a/src/Hash2Pub/ProtocolTypes.hs b/src/Hash2Pub/ProtocolTypes.hs index 86825a7..a5af10c 100644 --- a/src/Hash2Pub/ProtocolTypes.hs +++ b/src/Hash2Pub/ProtocolTypes.hs @@ -1,7 +1,5 @@ module Hash2Pub.ProtocolTypes where -import qualified Data.Map as Map -import Data.Maybe (mapMaybe) import qualified Data.Set as Set import Data.Time.Clock.POSIX (POSIXTime) From 6b166ac4ca7861aa339289cd5772b3dfca452c0d Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 7 Sep 2020 10:32:58 +0200 Subject: [PATCH 10/43] fixup! Merge branch 'measurement_logging' into mainline --- Hash2Pub.cabal | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index 2953d97..8970aaa 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -58,7 +58,7 @@ library exposed-modules: Hash2Pub.FediChord, Hash2Pub.FediChordTypes, Hash2Pub.DHTProtocol, Hash2Pub.ASN1Coding, Hash2Pub.ProtocolTypes, Hash2Pub.PostService, Hash2Pub.RingMap -- Modules included in this library but not exported. - other-modules: Hash2Pub.Utils + other-modules: Hash2Pub.Utils, Hash2Pub.PostService.API -- LANGUAGE extensions used by modules in this package. other-extensions: GeneralizedNewtypeDeriving, DataKinds, OverloadedStrings From c823e6357a911693b5254095e44b57b2ddd0e94f Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 7 Sep 2020 13:00:15 +0200 Subject: [PATCH 11/43] accumulate all statistic/ measurement events to a measurement summary - RingMap can now be mapped over --- src/Hash2Pub/PostService.hs | 106 ++++++++++++++++++++++++++++++++++-- src/Hash2Pub/RingMap.hs | 13 ++++- 2 files changed, 114 insertions(+), 5 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 89c14d2..baa2b70 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -19,7 +19,7 @@ import qualified Data.ByteString.Lazy.UTF8 as BSUL import qualified Data.ByteString.UTF8 as BSU import qualified Data.HashMap.Strict as HMap import qualified Data.HashSet as HSet -import Data.Maybe (isJust) +import Data.Maybe (isJust, fromJust) import Data.String (fromString) import Data.Text.Lazy (Text) import qualified Data.Text.Lazy as Txt @@ -56,6 +56,7 @@ data PostService d = PostService , postFetchQueue :: TQueue PostID , migrationsInProgress :: TVar (HMap.HashMap NodeID (MVar ())) , httpMan :: HTTP.Manager + , statsQueue :: TQueue StatsEvent } deriving (Typeable) @@ -84,9 +85,10 @@ instance DHT d => Service PostService d where postFetchQueue' <- newTQueueIO migrationsInProgress' <- newTVarIO HMap.empty httpMan' <- HTTP.newManager HTTP.defaultManagerSettings + statsQueue' <- newTQueueIO let - thisService = PostService { - serviceConf = conf + thisService = PostService + { serviceConf = conf , baseDHT = dht , serviceThread = threadVar , subscribers = subscriberVar @@ -96,7 +98,8 @@ instance DHT d => Service PostService d where , postFetchQueue = postFetchQueue' , migrationsInProgress = migrationsInProgress' , httpMan = httpMan' - } + , statsQueue = statsQueue' + } port' = fromIntegral (confServicePort conf) warpSettings = Warp.setPort port' . Warp.setHost (fromString . confServiceHost $ conf) $ Warp.defaultSettings -- Run 'concurrently_' from another thread to be able to return the @@ -599,3 +602,98 @@ fetchTagPosts serv = forever $ do -- TODO error handling, retry pure () + +-- ======= statistics/measurement and logging ======= + +data StatsEventType = PostPublishEvent + -- ^ initial publishing of a post by an instance + | RelayReceiveEvent + -- ^ receiving of posts because of being the responsible relay, for estimation of \tau_t + -- TODO: record for which hashtag + | RelayDeliveryEvent + -- ^ delivering (or at least attempt) a post to a subscriber + | IncomingPostFetchEvent + -- ^ another instance fetches a post from this instance + deriving (Enum, Show, Eq) + +-- | Represents measurement event of a 'StatsEventType' with a count relevant for a certain key +data StatsEvent = StatsEvent StatsEventType Int NodeID + +-- TODO: make delay configurable +statsAccuDelay = 300000 + +-- | periodically flush the stats queue and accumulate all events inside +accumulateStatsThread :: TQueue StatsEvent -> IO () +accumulateStatsThread statsQ = getPOSIXTime >>= flushLoop + where + flushLoop previousRun = do + now <- getPOSIXTime + -- TODO: instead of letting the events accumulate in the queue and allocate linear memory, immediately fold the result + -- but how to achieve the periodicity when blocking on a queue? + -- idea: let another thread periodically exchange the RelayStats, modify it atomically (Konzept "unterm Arsch wegziehen") + threadDelay statsAccuDelay + latestEvents <- atomically $ flushTQueue statsQ + -- accumulate the events + -- and now what? write a log to file, probably as a forkIO + -- persistently store in a TVar so it can be retrieved later by the DHT + flushLoop now + + +accumulateStats :: POSIXTime -> [StatsEvent] -> RelayStats +accumulateStats timeInterval events = + -- first sum all event numbers, then divide through number of seconds passed to + -- get rate per second + RelayStats + { relayReceiveRates = mapRMap (/ intervalSeconds) $ relayReceiveRates summedStats + , relayDeliveryRates = mapRMap (/ intervalSeconds) $ relayDeliveryRates summedStats + , postPublishRate = postPublishRate summedStats / intervalSeconds + , postFetchRate = postFetchRate summedStats / intervalSeconds + } + where + intervalSeconds = fromIntegral (fromEnum timeInterval) / 10^12 + summedStats = foldl (\stats event -> case event of + StatsEvent PostPublishEvent num _ -> + stats {postPublishRate = fromIntegral num + postPublishRate stats} + StatsEvent RelayReceiveEvent num key -> + stats {relayReceiveRates = sumIfEntryExists key (fromIntegral num) (relayReceiveRates stats)} + StatsEvent RelayDeliveryEvent num key -> + stats {relayDeliveryRates = sumIfEntryExists key (fromIntegral num) (relayDeliveryRates stats)} + StatsEvent IncomingPostFetchEvent num _ -> + stats {postFetchRate = fromIntegral num + postFetchRate stats} + ) + emptyStats + events + sumIfEntryExists = addRMapEntryWith (\newVal oldVal -> + let toInsert = fromJust $ extractRingEntry newVal + in + case oldVal of + KeyEntry n -> KeyEntry (n + toInsert) + ProxyEntry pointer (Just (KeyEntry n)) -> ProxyEntry pointer (Just (KeyEntry $ n + toInsert)) + ProxyEntry pointer Nothing -> ProxyEntry pointer (Just newVal) + _ -> error "RingMap nested too deeply" + ) + +-- idea: first just sum with foldl, and then map the time division over all values + + + +emptyStats :: RelayStats +emptyStats = RelayStats + { relayReceiveRates = emptyRMap + , relayDeliveryRates = emptyRMap + , postFetchRate = 0 + , postPublishRate = 0 + } + +-- | measured rates of relay performance +-- TODO: maybe include other metrics in here as well, like number of subscribers? +data RelayStats = RelayStats + { relayReceiveRates :: RingMap NodeID Double + -- ^ rate of incoming posts in the responsibility of this relay + , relayDeliveryRates :: RingMap NodeID Double + -- ^ rate of relayed outgoing posts + , postFetchRate :: Double -- no need to differentiate between tags + -- ^ number of post-fetches delivered + , postPublishRate :: Double + -- ^ rate of initially publishing posts through this instance + } diff --git a/src/Hash2Pub/RingMap.hs b/src/Hash2Pub/RingMap.hs index ae1ec15..8416278 100644 --- a/src/Hash2Pub/RingMap.hs +++ b/src/Hash2Pub/RingMap.hs @@ -133,7 +133,7 @@ rMapLookupPred :: (Bounded k, Ord k, Num k) rMapLookupPred = lookupWrapper Map.lookupLT Map.lookupLE Backwards addRMapEntryWith :: (Bounded k, Ord k) - => (RingEntry k a -> RingEntry k a -> RingEntry k a) + => (RingEntry k a -> RingEntry k a -> RingEntry k a) -- ^ f new_value mold_value -> k -- ^ key -> a -- ^ value -> RingMap k a @@ -247,3 +247,14 @@ takeRMapSuccessorsFromTo :: (Bounded k, Ord k, Num k) -> RingMap k a -> [a] takeRMapSuccessorsFromTo fromVal toVal rmap = takeEntriesUntil_ rmap rMapLookupSucc toVal fromVal Nothing [] + + +-- | map a function over all payload values of a 'RingMap' +mapRMap :: (Bounded k, Ord k, Num k) + => (a -> b) -> RingMap k a -> RingMap k b +mapRMap f = RingMap . Map.map traversingF . getRingMap + where + --traversingF :: RingEntry k a -> RingEntry k b + traversingF (KeyEntry a) = KeyEntry (f a) + traversingF (ProxyEntry pointer (Just entry)) = ProxyEntry pointer (Just $ traversingF entry) + traversingF (ProxyEntry pointer Nothing) = ProxyEntry pointer Nothing From 5c338b9cd7d5ea5a7b7061bd28eac16a7b05e356 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 7 Sep 2020 16:27:19 +0200 Subject: [PATCH 12/43] split up stats summing and evaluating, launch threads --- src/Hash2Pub/PostService.hs | 165 +++++++++++++++++++++--------------- 1 file changed, 99 insertions(+), 66 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index baa2b70..938ca0e 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -19,7 +19,7 @@ import qualified Data.ByteString.Lazy.UTF8 as BSUL import qualified Data.ByteString.UTF8 as BSU import qualified Data.HashMap.Strict as HMap import qualified Data.HashSet as HSet -import Data.Maybe (isJust, fromJust) +import Data.Maybe (fromJust, isJust) import Data.String (fromString) import Data.Text.Lazy (Text) import qualified Data.Text.Lazy as Txt @@ -54,9 +54,11 @@ data PostService d = PostService , relayInQueue :: TQueue (Hashtag, PostID, PostContent) -- ^ Queue for processing incoming posts of own instance asynchronously , postFetchQueue :: TQueue PostID + -- ^ queue of posts to be fetched , migrationsInProgress :: TVar (HMap.HashMap NodeID (MVar ())) , httpMan :: HTTP.Manager , statsQueue :: TQueue StatsEvent + , loadStats :: TVar RelayStats } deriving (Typeable) @@ -86,6 +88,7 @@ instance DHT d => Service PostService d where migrationsInProgress' <- newTVarIO HMap.empty httpMan' <- HTTP.newManager HTTP.defaultManagerSettings statsQueue' <- newTQueueIO + loadStats' <- newTVarIO emptyStats let thisService = PostService { serviceConf = conf @@ -99,6 +102,7 @@ instance DHT d => Service PostService d where , migrationsInProgress = migrationsInProgress' , httpMan = httpMan' , statsQueue = statsQueue' + , loadStats = loadStats' } port' = fromIntegral (confServicePort conf) warpSettings = Warp.setPort port' . Warp.setHost (fromString . confServiceHost $ conf) $ Warp.defaultSettings @@ -109,7 +113,11 @@ instance DHT d => Service PostService d where concurrently_ -- web server (Warp.runSettings warpSettings $ postServiceApplication thisService) - (processIncomingPosts thisService) + $ concurrently + -- post queue processing + (processIncomingPosts thisService) + -- statistics/ measurements + (launchStatsThreads thisService) -- update thread ID after fork atomically $ writeTVar threadVar servThreadID pure thisService @@ -606,63 +614,64 @@ fetchTagPosts serv = forever $ do -- ======= statistics/measurement and logging ======= data StatsEventType = PostPublishEvent - -- ^ initial publishing of a post by an instance - | RelayReceiveEvent - -- ^ receiving of posts because of being the responsible relay, for estimation of \tau_t - -- TODO: record for which hashtag - | RelayDeliveryEvent - -- ^ delivering (or at least attempt) a post to a subscriber - | IncomingPostFetchEvent - -- ^ another instance fetches a post from this instance - deriving (Enum, Show, Eq) + | RelayReceiveEvent + | RelayDeliveryEvent + | IncomingPostFetchEvent + deriving (Enum, Show, Eq) -- | Represents measurement event of a 'StatsEventType' with a count relevant for a certain key data StatsEvent = StatsEvent StatsEventType Int NodeID --- TODO: make delay configurable -statsAccuDelay = 300000 --- | periodically flush the stats queue and accumulate all events inside -accumulateStatsThread :: TQueue StatsEvent -> IO () -accumulateStatsThread statsQ = getPOSIXTime >>= flushLoop - where - flushLoop previousRun = do - now <- getPOSIXTime - -- TODO: instead of letting the events accumulate in the queue and allocate linear memory, immediately fold the result - -- but how to achieve the periodicity when blocking on a queue? - -- idea: let another thread periodically exchange the RelayStats, modify it atomically (Konzept "unterm Arsch wegziehen") - threadDelay statsAccuDelay - latestEvents <- atomically $ flushTQueue statsQ - -- accumulate the events - -- and now what? write a log to file, probably as a forkIO - -- persistently store in a TVar so it can be retrieved later by the DHT - flushLoop now - - -accumulateStats :: POSIXTime -> [StatsEvent] -> RelayStats -accumulateStats timeInterval events = - -- first sum all event numbers, then divide through number of seconds passed to - -- get rate per second - RelayStats - { relayReceiveRates = mapRMap (/ intervalSeconds) $ relayReceiveRates summedStats - , relayDeliveryRates = mapRMap (/ intervalSeconds) $ relayDeliveryRates summedStats - , postPublishRate = postPublishRate summedStats / intervalSeconds - , postFetchRate = postFetchRate summedStats / intervalSeconds +-- | measured rates of relay performance +-- TODO: maybe include other metrics in here as well, like number of subscribers? +data RelayStats = RelayStats + { relayReceiveRates :: RingMap NodeID Double + -- ^ rate of incoming posts in the responsibility of this relay + , relayDeliveryRates :: RingMap NodeID Double + -- ^ rate of relayed outgoing posts + , postFetchRate :: Double -- no need to differentiate between tags + -- ^ number of post-fetches delivered + , postPublishRate :: Double + -- ^ rate of initially publishing posts through this instance } + + +-- TODO: make delay configurable +statsEvalDelay = 300000 + + +launchStatsThreads :: PostService d -> IO () +launchStatsThreads serv = do + -- create shared accumulator + sharedAccum <- newTVarIO emptyStats + concurrently_ + (accumulateStatsThread sharedAccum $ statsQueue serv) + (evaluateStatsThread serv sharedAccum) + + +-- | Read stats events from queue and add them to a shared accumulator. +-- Instead of letting the events accumulate in the queue and allocate linear memory, immediately fold the result. +accumulateStatsThread :: TVar RelayStats -> TQueue StatsEvent -> IO () +accumulateStatsThread statsAccumulator statsQ = forever $ do + -- blocks until stats event arrives + event <- atomically $ readTQueue statsQ + -- add the event number to current accumulator + atomically $ modifyTVar' statsAccumulator $ statsAdder event + + +-- | add incoming stats events to accumulator value +statsAdder :: StatsEvent -> RelayStats -> RelayStats +statsAdder event stats = case event of + StatsEvent PostPublishEvent num _ -> + stats {postPublishRate = fromIntegral num + postPublishRate stats} + StatsEvent RelayReceiveEvent num key -> + stats {relayReceiveRates = sumIfEntryExists key (fromIntegral num) (relayReceiveRates stats)} + StatsEvent RelayDeliveryEvent num key -> + stats {relayDeliveryRates = sumIfEntryExists key (fromIntegral num) (relayDeliveryRates stats)} + StatsEvent IncomingPostFetchEvent num _ -> + stats {postFetchRate = fromIntegral num + postFetchRate stats} where - intervalSeconds = fromIntegral (fromEnum timeInterval) / 10^12 - summedStats = foldl (\stats event -> case event of - StatsEvent PostPublishEvent num _ -> - stats {postPublishRate = fromIntegral num + postPublishRate stats} - StatsEvent RelayReceiveEvent num key -> - stats {relayReceiveRates = sumIfEntryExists key (fromIntegral num) (relayReceiveRates stats)} - StatsEvent RelayDeliveryEvent num key -> - stats {relayDeliveryRates = sumIfEntryExists key (fromIntegral num) (relayDeliveryRates stats)} - StatsEvent IncomingPostFetchEvent num _ -> - stats {postFetchRate = fromIntegral num + postFetchRate stats} - ) - emptyStats - events sumIfEntryExists = addRMapEntryWith (\newVal oldVal -> let toInsert = fromJust $ extractRingEntry newVal in @@ -673,8 +682,45 @@ accumulateStats timeInterval events = _ -> error "RingMap nested too deeply" ) --- idea: first just sum with foldl, and then map the time division over all values +-- Periodically exchange the accumulated statistics with empty ones, evaluate them +-- and make them the current statistics of the service. +evaluateStatsThread :: PostService d -> TVar RelayStats -> IO () +evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop + where + loop previousTs = do + threadDelay statsEvalDelay + -- get and reset the stats accumulator + summedStats <- atomically $ do + stats <- readTVar statsAcc + writeTVar statsAcc emptyStats + pure stats + -- as the transaction might retry several times, current time needs to + -- be read afterwards + now <- getPOSIXTime + -- evaluate stats rate and replace server stats + atomically . writeTVar (loadStats serv) . evaluateStats (now - previousTs) $ summedStats + -- idea: let another thread periodically exchange the RelayStats, modify it atomically (Konzept "unterm Arsch wegziehen") + -- and now what? write a log to file, probably as a forkIO + -- persistently store in a TVar so it can be retrieved later by the DHT + loop now + + +-- | Evaluate the accumulated statistic events: Currently mostly calculates the event +-- rates by dividing through the collection time frame +evaluateStats :: POSIXTime -> RelayStats -> RelayStats +evaluateStats timeInterval summedStats = + -- first sum all event numbers, then divide through number of seconds passed to + -- get rate per second + RelayStats + { relayReceiveRates = mapRMap (/ intervalSeconds) $ relayReceiveRates summedStats + , relayDeliveryRates = mapRMap (/ intervalSeconds) $ relayDeliveryRates summedStats + , postPublishRate = postPublishRate summedStats / intervalSeconds + , postFetchRate = postFetchRate summedStats / intervalSeconds + } + where + -- TODO: take speedup into account + intervalSeconds = fromIntegral (fromEnum timeInterval) / 10^12 emptyStats :: RelayStats @@ -684,16 +730,3 @@ emptyStats = RelayStats , postFetchRate = 0 , postPublishRate = 0 } - --- | measured rates of relay performance --- TODO: maybe include other metrics in here as well, like number of subscribers? -data RelayStats = RelayStats - { relayReceiveRates :: RingMap NodeID Double - -- ^ rate of incoming posts in the responsibility of this relay - , relayDeliveryRates :: RingMap NodeID Double - -- ^ rate of relayed outgoing posts - , postFetchRate :: Double -- no need to differentiate between tags - -- ^ number of post-fetches delivered - , postPublishRate :: Double - -- ^ rate of initially publishing posts through this instance - } From c536994afe762016f70b91ea5bd3f59cd23d6b63 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 7 Sep 2020 16:35:59 +0200 Subject: [PATCH 13/43] re-format Servant client pattern matching --- src/Hash2Pub/PostService.hs | 18 +++++++++--------- 1 file changed, 9 insertions(+), 9 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 938ca0e..c1ea936 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -321,15 +321,15 @@ tagUnsubscribe serv hashtag origin = do clientAPI :: Proxy PostServiceAPI clientAPI = Proxy -relayInboxClient :: Text -> Text -> ClientM NoContent -relayInboxClient :<|> subscriptionDeliveryClient - :<|> postFetchClient - :<|> postMultiFetchClient - :<|> postInboxClient - :<|> tagDeliveryClient - :<|> tagSubscribeClient - :<|> tagUnsubscribeClient - = client clientAPI +relayInboxClient + :<|> subscriptionDeliveryClient + :<|> postFetchClient + :<|> postMultiFetchClient + :<|> postInboxClient + :<|> tagDeliveryClient + :<|> tagSubscribeClient + :<|> tagUnsubscribeClient + = client clientAPI -- | Deliver the subscriber list of all hashtags in the interval [fromTag, toTag] From df479982fa15b463a2c0f0daab99cad4755eb32c Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Tue, 8 Sep 2020 08:46:36 +0200 Subject: [PATCH 14/43] make RingMap instance of Functor and Foldable --- src/Hash2Pub/PostService.hs | 4 ++-- src/Hash2Pub/RingMap.hs | 34 +++++++++++++++++++++++----------- 2 files changed, 25 insertions(+), 13 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index c1ea936..b111455 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -713,8 +713,8 @@ evaluateStats timeInterval summedStats = -- first sum all event numbers, then divide through number of seconds passed to -- get rate per second RelayStats - { relayReceiveRates = mapRMap (/ intervalSeconds) $ relayReceiveRates summedStats - , relayDeliveryRates = mapRMap (/ intervalSeconds) $ relayDeliveryRates summedStats + { relayReceiveRates = (/ intervalSeconds) <$> relayReceiveRates summedStats + , relayDeliveryRates = (/ intervalSeconds) <$> relayDeliveryRates summedStats , postPublishRate = postPublishRate summedStats / intervalSeconds , postFetchRate = postFetchRate summedStats / intervalSeconds } diff --git a/src/Hash2Pub/RingMap.hs b/src/Hash2Pub/RingMap.hs index 8416278..a2fe3ae 100644 --- a/src/Hash2Pub/RingMap.hs +++ b/src/Hash2Pub/RingMap.hs @@ -25,6 +25,29 @@ instance (Bounded k, Ord k, Eq a) => Eq (RingMap k a) where instance (Bounded k, Ord k, Show k, Show a) => Show (RingMap k a) where show rmap = shows ("RingMap " :: String) (show $ getRingMap rmap) + +instance (Bounded k, Ord k) => Functor (RingMap k) where + -- | map a function over all payload values of a 'RingMap' + fmap f = RingMap . Map.map traversingF . getRingMap + where + traversingF (KeyEntry a) = KeyEntry (f a) + traversingF (ProxyEntry pointer (Just entry)) = ProxyEntry pointer (Just $ traversingF entry) + traversingF (ProxyEntry pointer Nothing) = ProxyEntry pointer Nothing + + +instance (Bounded k, Ord k) => Foldable (RingMap k) where + foldr f initVal = Map.foldr traversingFR initVal . getRingMap + where + traversingFR (KeyEntry a) acc = f a acc + traversingFR (ProxyEntry _ Nothing) acc = acc + traversingFR (ProxyEntry _ (Just entry)) acc = traversingFR entry acc + foldl f initVal = Map.foldl traversingFL initVal . getRingMap + where + traversingFL acc (KeyEntry a) = f acc a + traversingFL acc (ProxyEntry _ Nothing) = acc + traversingFL acc (ProxyEntry _ (Just entry)) = traversingFL acc entry + + -- | entry of a 'RingMap' that holds a value and can also -- wrap around the lookup direction at the edges of the name space. data RingEntry k a = KeyEntry a @@ -247,14 +270,3 @@ takeRMapSuccessorsFromTo :: (Bounded k, Ord k, Num k) -> RingMap k a -> [a] takeRMapSuccessorsFromTo fromVal toVal rmap = takeEntriesUntil_ rmap rMapLookupSucc toVal fromVal Nothing [] - - --- | map a function over all payload values of a 'RingMap' -mapRMap :: (Bounded k, Ord k, Num k) - => (a -> b) -> RingMap k a -> RingMap k b -mapRMap f = RingMap . Map.map traversingF . getRingMap - where - --traversingF :: RingEntry k a -> RingEntry k b - traversingF (KeyEntry a) = KeyEntry (f a) - traversingF (ProxyEntry pointer (Just entry)) = ProxyEntry pointer (Just $ traversingF entry) - traversingF (ProxyEntry pointer Nothing) = ProxyEntry pointer Nothing From 2b39648a77dcc16c5689020d26dcd6abc8218a89 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 11:39:48 +0200 Subject: [PATCH 15/43] actually implement simple relaying of posts was still missing for #41 --- src/Hash2Pub/DHTProtocol.hs | 1 - src/Hash2Pub/PostService.hs | 28 ++++++++++++++++++++++++++++ 2 files changed, 28 insertions(+), 1 deletion(-) diff --git a/src/Hash2Pub/DHTProtocol.hs b/src/Hash2Pub/DHTProtocol.hs index 8258ca3..3639c08 100644 --- a/src/Hash2Pub/DHTProtocol.hs +++ b/src/Hash2Pub/DHTProtocol.hs @@ -587,7 +587,6 @@ sendQueryIdMessages targetID ns lParam targets = do nodeConf <- nodeConfig <$> readTVarIO (parentRealNode ns) let srcAddr = confIP nodeConf - -- ToDo: make attempts and timeout configurable queryThreads <- mapM (\resultNode -> async $ bracket (mkSendSocket srcAddr (getDomain resultNode) (getDhtPort resultNode)) close ( sendRequestTo (confRequestTimeout nodeConf) (confRequestRetries nodeConf) (lookupMessage targetID ns Nothing) )) targets diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index b111455..099855d 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -17,6 +17,7 @@ import Control.Monad.IO.Class (liftIO) import Data.Bifunctor import qualified Data.ByteString.Lazy.UTF8 as BSUL import qualified Data.ByteString.UTF8 as BSU +import Data.Either (rights) import qualified Data.HashMap.Strict as HMap import qualified Data.HashSet as HSet import Data.Maybe (fromJust, isJust) @@ -611,6 +612,33 @@ fetchTagPosts serv = forever $ do pure () +-- TODO: paralellelisation +-- TODO: make sure it doesn't busy-wait +relayWorker :: PostService d -> IO () +relayWorker serv = forever $ do + subscriptionMap <- readTVarIO $ subscribers serv + -- for each tag, try to deliver some posts to subscriber + forM_ subscriptionMap (\(subscriberMapSTM, _, tag) -> do + subscriberMap <- readTVarIO subscriberMapSTM + forM_ (HMap.toList subscriberMap) (\((subHost, subPort), (postChan, _)) -> do + postsToDeliver <- readUpTo 500 postChan + response <- runClientM (tagDeliveryClient tag (Txt.unlines postsToDeliver)) (mkClientEnv (httpMan serv) (BaseUrl Http subHost (fromIntegral subPort) "")) + -- so far just dropping failed attempts, TODO: retry mechanism + -- TODO: stats + pure () + ) + ) + where + readUpTo :: Int -> TChan a -> IO [a] + readUpTo 0 _ = pure [] + readUpTo n chan = do + readFromChan <- atomically (tryReadTChan chan) + case readFromChan of + Nothing -> pure [] + Just val -> do + moreReads <- readUpTo (pred n) chan + pure (val:moreReads) + -- ======= statistics/measurement and logging ======= data StatsEventType = PostPublishEvent From 0ffe9effc0dee2c8ab1591dd44565f3e6f8a17b7 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 14:23:36 +0200 Subject: [PATCH 16/43] refactor relay processing to STM-retry instead of busy-wait --- Hash2Pub.cabal | 2 +- src/Hash2Pub/PostService.hs | 48 +++++++++++++++++++++++++------------ 2 files changed, 34 insertions(+), 16 deletions(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index 2953d97..5e8d25d 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -46,7 +46,7 @@ category: Network extra-source-files: CHANGELOG.md common deps - build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays + build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist ghc-options: -Wall -Wpartial-fields -O2 diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 099855d..c556d7f 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -17,6 +17,7 @@ import Control.Monad.IO.Class (liftIO) import Data.Bifunctor import qualified Data.ByteString.Lazy.UTF8 as BSUL import qualified Data.ByteString.UTF8 as BSU +import qualified Data.DList as D import Data.Either (rights) import qualified Data.HashMap.Strict as HMap import qualified Data.HashSet as HSet @@ -39,6 +40,7 @@ import Servant.Client import Hash2Pub.FediChordTypes import Hash2Pub.PostService.API import Hash2Pub.RingMap +import Hash2Pub.Utils data PostService d = PostService @@ -612,27 +614,43 @@ fetchTagPosts serv = forever $ do pure () +-- TODO: make configurable +numParallelDeliveries = 10 + -- TODO: paralellelisation --- TODO: make sure it doesn't busy-wait relayWorker :: PostService d -> IO () relayWorker serv = forever $ do - subscriptionMap <- readTVarIO $ subscribers serv - -- for each tag, try to deliver some posts to subscriber - forM_ subscriptionMap (\(subscriberMapSTM, _, tag) -> do - subscriberMap <- readTVarIO subscriberMapSTM - forM_ (HMap.toList subscriberMap) (\((subHost, subPort), (postChan, _)) -> do - postsToDeliver <- readUpTo 500 postChan - response <- runClientM (tagDeliveryClient tag (Txt.unlines postsToDeliver)) (mkClientEnv (httpMan serv) (BaseUrl Http subHost (fromIntegral subPort) "")) - -- so far just dropping failed attempts, TODO: retry mechanism - -- TODO: stats - pure () - ) - ) + -- atomically (to be able to retry) fold a list of due delivery actions + jobsToProcess <- atomically $ do + subscriptionMap <- readTVar $ subscribers serv + jobList <- D.toList <$> foldM (\jobAcc (subscriberMapSTM, _, tag) -> do + subscriberMap <- readTVar subscriberMapSTM + foldM (\jobAcc' ((subHost, subPort), (postChan, _)) -> do + postsToDeliver <- readUpTo 500 postChan + -- append fetch job to job list + pure $ if not (null postsToDeliver) + then jobAcc' `D.snoc` runClientM (tagDeliveryClient tag (Txt.unlines postsToDeliver)) (mkClientEnv (httpMan serv) (BaseUrl Http subHost (fromIntegral subPort) "")) + else jobAcc' + ) jobAcc $ HMap.toList subscriberMap + ) D.empty subscriptionMap + -- if no relay jobs, then retry + if null jobList + then retry + else pure jobList + + -- when processing the list, send several deliveries in parallel + forM_ (chunksOf numParallelDeliveries jobsToProcess) $ \jobset -> do + runningJobs <- mapM async jobset + -- so far just dropping failed attempts, TODO: retry mechanism + successfulResults <- rights <$> mapM waitCatch runningJobs + -- TODO: stats + pure () + where - readUpTo :: Int -> TChan a -> IO [a] + readUpTo :: Int -> TChan a -> STM [a] readUpTo 0 _ = pure [] readUpTo n chan = do - readFromChan <- atomically (tryReadTChan chan) + readFromChan <- tryReadTChan chan case readFromChan of Nothing -> pure [] Just val -> do From 72eca0f4fe13c4636163329f0c9a9c03c4924cac Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 17:12:56 +0200 Subject: [PATCH 17/43] log metrics to file contributes to #60 --- app/Main.hs | 13 +++++++------ src/Hash2Pub/FediChordTypes.hs | 2 ++ src/Hash2Pub/PostService.hs | 17 ++++++++++++++--- 3 files changed, 23 insertions(+), 9 deletions(-) diff --git a/app/Main.hs b/app/Main.hs index c08cd3c..80c0520 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -54,8 +54,8 @@ readConfig = do bootstrapHost : bootstrapPortString : _ -> [(bootstrapHost, read bootstrapPortString)] _ -> [] - fConf = FediChordConf { - confDomain = confDomainString + fConf = FediChordConf + { confDomain = confDomainString , confIP = toHostAddress6 . read $ ipString , confDhtPort = read portString , confBootstrapNodes = confBootstrapNodes' @@ -67,11 +67,12 @@ readConfig = do , confResponsePurgeAge = 60 / fromIntegral speedup , confRequestTimeout = 5 * 10^6 `div` speedup , confRequestRetries = 3 - } - sConf = ServiceConf { - confSubscriptionExpiryTime = fromIntegral $ 2*3600 `div` speedup + } + sConf = ServiceConf + { confSubscriptionExpiryTime = fromIntegral $ 2*3600 `div` speedup , confServicePort = read servicePortString , confServiceHost = confDomainString - } + , confLogfilePath = "../simulationData/logs/" <> confDomainString <> ".log" + } pure (fConf, sConf) diff --git a/src/Hash2Pub/FediChordTypes.hs b/src/Hash2Pub/FediChordTypes.hs index 3b563e6..725031e 100644 --- a/src/Hash2Pub/FediChordTypes.hs +++ b/src/Hash2Pub/FediChordTypes.hs @@ -457,6 +457,8 @@ data ServiceConf = ServiceConf -- ^ listening port for service , confServiceHost :: String -- ^ hostname of service + , confLogfilePath :: String + -- ^ where to store the (measurement) log file } class DHT d where diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index c556d7f..2abf3b8 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -25,6 +25,7 @@ import Data.Maybe (fromJust, isJust) import Data.String (fromString) import Data.Text.Lazy (Text) import qualified Data.Text.Lazy as Txt +import qualified Data.Text.Lazy.IO as TxtI import Data.Text.Normalize (NormalizationMode (NFC), normalize) import Data.Time.Clock.POSIX import Data.Typeable (Typeable) @@ -109,6 +110,8 @@ instance DHT d => Service PostService d where } port' = fromIntegral (confServicePort conf) warpSettings = Warp.setPort port' . Warp.setHost (fromString . confServiceHost $ conf) $ Warp.defaultSettings + -- log a start message, this also truncates existing files + TxtI.writeFile (confLogfilePath conf) "# Starting mock relay implementation\n" -- Run 'concurrently_' from another thread to be able to return the -- 'PostService'. -- Terminating that parent thread will make all child threads terminate as well. @@ -745,10 +748,18 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop -- be read afterwards now <- getPOSIXTime -- evaluate stats rate and replace server stats - atomically . writeTVar (loadStats serv) . evaluateStats (now - previousTs) $ summedStats - -- idea: let another thread periodically exchange the RelayStats, modify it atomically (Konzept "unterm Arsch wegziehen") - -- and now what? write a log to file, probably as a forkIO -- persistently store in a TVar so it can be retrieved later by the DHT + atomically . writeTVar (loadStats serv) . evaluateStats (now - previousTs) $ summedStats + -- and now what? write a log to file + -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate + -- later: current (reported) load, target load + TxtI.appendFile (confLogfilePath . serviceConf $ serv) $ + Txt.intercalate ";" (Txt.pack <$> ( + [ show . sum . relayReceiveRates + , show . sum . relayDeliveryRates + , show . postPublishRate + , show . postFetchRate + ] <*> pure summedStats)) <> "\n" loop now From 12fcd137541c8813ecc7e2b87733799616b3337e Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 18:01:51 +0200 Subject: [PATCH 18/43] annotate the PostService server/ request-handler functions --- src/Hash2Pub/PostService.hs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 2abf3b8..3d1df68 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -167,6 +167,7 @@ postServer service = relayInbox service :<|> tagUnsubscribe service +-- | delivery endpoint: receive posts of a handled tag and enqueue them for relaying relayInbox :: DHT d => PostService d -> Hashtag -> Text -> Handler NoContent relayInbox serv tag posts = do let @@ -195,6 +196,7 @@ newtype UnhandledTagException = UnhandledTagException String instance Exception UnhandledTagException +-- | delivery endpoint: receives a list of subscribers of tags and their outstanding queues for migration subscriptionDelivery :: DHT d => PostService d -> Integer -> Text -> Handler Text subscriptionDelivery serv senderID subList = do let @@ -240,6 +242,7 @@ subscriptionDelivery serv senderID subList = do enqueueSubscription subscriberSTM (normaliseTag tag) sub postList leaseTime +-- | endpoint for fetching a post by its ID postFetch :: PostService d -> Text -> Handler Text postFetch serv postID = do postSet <- liftIO . readTVarIO . ownPosts $ serv @@ -249,6 +252,7 @@ postFetch serv postID = do else throwError $ err404 { errBody = "No post found with this ID" } +-- | endpoint for fetching multiple posts of this instance by their IDs postMultiFetch :: PostService d -> Text -> Handler Text postMultiFetch serv postIDs = do let idList = Txt.lines postIDs @@ -261,6 +265,7 @@ postMultiFetch serv postIDs = do ) "" idList +-- | delivery endpoint: inbox for initially publishing a post at an instance postInbox :: PostService d -> Text -> Handler NoContent postInbox serv post = do -- extract contained hashtags @@ -277,6 +282,7 @@ postInbox serv post = do pure NoContent +-- | delivery endpoint: receive postIDs of a certain subscribed hashtag tagDelivery :: PostService d -> Text -> Text -> Handler Text tagDelivery serv hashtag posts = do let postIDs = Txt.lines posts @@ -288,6 +294,8 @@ tagDelivery serv hashtag posts = do pure () pure $ "Received a postID for tag " <> hashtag + +-- | receive subscription requests to a handled hashtag tagSubscribe :: DHT d => PostService d -> Text -> Maybe Text -> Handler Integer tagSubscribe serv hashtag origin = do responsible <- liftIO $ isResponsibleFor (baseDHT serv) (hashtagToId hashtag) @@ -307,6 +315,7 @@ tagSubscribe serv hashtag origin = do pure $ round leaseTime +-- | receive and handle unsubscription requests regarding a handled tag tagUnsubscribe :: DHT d => PostService d -> Text -> Maybe Text -> Handler Text tagUnsubscribe serv hashtag origin = do responsible <- liftIO $ isResponsibleFor (baseDHT serv) (hashtagToId hashtag) From e3a8912360f4dad0a5f808ddd2e5966d91d24057 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 18:50:45 +0200 Subject: [PATCH 19/43] process incoming posts in parallel --- src/Hash2Pub/PostService.hs | 105 ++++++++++++++++++++++-------------- 1 file changed, 64 insertions(+), 41 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 3d1df68..02278bf 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -12,7 +12,8 @@ import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM import Control.Exception (Exception (..), try) -import Control.Monad (foldM, forM_, forever, void, when) +import Control.Monad (foldM, forM, forM_, forever, void, + when) import Control.Monad.IO.Class (liftIO) import Data.Bifunctor import qualified Data.ByteString.Lazy.UTF8 as BSUL @@ -560,6 +561,28 @@ normaliseTag = Txt.fromStrict . normalize NFC . Txt.toStrict hashtagToId :: Hashtag -> NodeID hashtagToId = genKeyID . Txt.unpack + +readUpToTChan :: Int -> TChan a -> STM [a] +readUpToTChan 0 _ = pure [] +readUpToTChan n chan = do + readFromChan <- tryReadTChan chan + case readFromChan of + Nothing -> pure [] + Just val -> do + moreReads <- readUpToTChan (pred n) chan + pure (val:moreReads) + + +readUpToTQueue :: Int -> TQueue a -> STM [a] +readUpToTQueue 0 _ = pure [] +readUpToTQueue n q = do + readFromQueue <- tryReadTQueue q + case readFromQueue of + Nothing -> pure [] + Just val -> do + moreReads <- readUpToTQueue (pred n) q + pure (val:moreReads) + -- | define how to convert all showable types to PlainText -- No idea what I'm doing with these overlappable instances though ¯\_(ツ)_/¯ -- TODO: figure out how this overlapping stuff actually works https://downloads.haskell.org/~ghc/latest/docs/html/users_guide/glasgow_exts.html#instance-overlap @@ -571,36 +594,50 @@ instance {-# OVERLAPPABLE #-} Read a => MimeUnrender PlainText a where -- ====== worker threads ====== +-- TODO: make configurable +numParallelDeliveries = 10 + + -- | process the pending relay inbox of incoming posts from the internal queue: -- Look up responsible relay node for given hashtag and forward post to it processIncomingPosts :: DHT d => PostService d -> IO () processIncomingPosts serv = forever $ do -- blocks until available - -- TODO: process multiple in parallel - (tag, pID, pContent) <- atomically . readTQueue $ relayInQueue serv - let pIdUri = "http://" <> (Txt.pack . confServiceHost . serviceConf $ serv) <> ":" <> (fromString . show . confServicePort . serviceConf $ serv) <> "/post/" <> pID - lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag) - case lookupRes of - -- no vserver active => wait and retry - Nothing -> threadDelay $ 10 * 10^6 - Just (responsibleHost, responsiblePort) -> do - resp <- runClientM (relayInboxClient tag $ pIdUri <> "," <> pContent) (mkClientEnv (httpMan serv) (BaseUrl Http responsibleHost (fromIntegral responsiblePort) "")) - case resp of - Left err -> do - putStrLn $ "Error: " <> show err - -- 410 error indicates outdated responsibility mapping - -- Simplification: just invalidate the mapping entry on all errors, force a re-lookup and re-queue the post - -- TODO: keep track of maximum retries - _ <- forceLookupKey (baseDHT serv) (Txt.unpack tag) - atomically . writeTQueue (relayInQueue serv) $ (tag, pID, pContent) - Right _ -> do - -- TODO: stats - -- idea for the experiment: each post publication makes the initial posting instance subscribe to all contained tags - now <- getPOSIXTime - subscriptionStatus <- HMap.lookup (hashtagToId tag) <$> readTVarIO (ownSubscriptions serv) - -- if not yet subscribed or subscription expires within 2 minutes, (re)subscribe to tag - when (maybe False (\subLease -> now - subLease < 120) subscriptionStatus) $ - void $ clientSubscribeTo serv tag + deliveriesToProcess <- atomically $ do + readResult <- readUpToTQueue numParallelDeliveries $ relayInQueue serv + if null readResult + then retry + else pure readResult + runningJobs <- forM deliveriesToProcess $ \(tag, pID, pContent) -> async $ do + let pIdUri = "http://" <> (Txt.pack . confServiceHost . serviceConf $ serv) <> ":" <> (fromString . show . confServicePort . serviceConf $ serv) <> "/post/" <> pID + lookupRes <- lookupKey (baseDHT serv) (Txt.unpack tag) + case lookupRes of + -- no vserver active => wait and retry + Nothing -> threadDelay (10 * 10^6) >> pure (Left "no vserver active") + Just (responsibleHost, responsiblePort) -> do + resp <- runClientM (relayInboxClient tag $ pIdUri <> "," <> pContent) (mkClientEnv (httpMan serv) (BaseUrl Http responsibleHost (fromIntegral responsiblePort) "")) + case resp of + Left err -> do + -- 410 error indicates outdated responsibility mapping + -- Simplification: just invalidate the mapping entry on all errors, force a re-lookup and re-queue the post + -- TODO: keep track of maximum retries + _ <- forceLookupKey (baseDHT serv) (Txt.unpack tag) + atomically . writeTQueue (relayInQueue serv) $ (tag, pID, pContent) + pure . Left $ "Error: " <> show err + Right _ -> do + -- idea for the experiment: each post publication makes the initial posting instance subscribe to all contained tags + now <- getPOSIXTime + subscriptionStatus <- HMap.lookup (hashtagToId tag) <$> readTVarIO (ownSubscriptions serv) + -- if not yet subscribed or subscription expires within 2 minutes, (re)subscribe to tag + when (maybe False (\subLease -> now - subLease < 120) subscriptionStatus) $ + void $ clientSubscribeTo serv tag + + -- for evaluation, return the tag of the successfully forwarded post + pure $ Right tag + + -- collect async results + results <- mapM waitCatch runningJobs + -- TODO: statistics -- | process the pending fetch jobs of delivered post IDs: Delivered posts are tried to be fetched from their URI-ID @@ -626,10 +663,6 @@ fetchTagPosts serv = forever $ do pure () --- TODO: make configurable -numParallelDeliveries = 10 - --- TODO: paralellelisation relayWorker :: PostService d -> IO () relayWorker serv = forever $ do -- atomically (to be able to retry) fold a list of due delivery actions @@ -638,7 +671,7 @@ relayWorker serv = forever $ do jobList <- D.toList <$> foldM (\jobAcc (subscriberMapSTM, _, tag) -> do subscriberMap <- readTVar subscriberMapSTM foldM (\jobAcc' ((subHost, subPort), (postChan, _)) -> do - postsToDeliver <- readUpTo 500 postChan + postsToDeliver <- readUpToTChan 500 postChan -- append fetch job to job list pure $ if not (null postsToDeliver) then jobAcc' `D.snoc` runClientM (tagDeliveryClient tag (Txt.unlines postsToDeliver)) (mkClientEnv (httpMan serv) (BaseUrl Http subHost (fromIntegral subPort) "")) @@ -658,16 +691,6 @@ relayWorker serv = forever $ do -- TODO: stats pure () - where - readUpTo :: Int -> TChan a -> STM [a] - readUpTo 0 _ = pure [] - readUpTo n chan = do - readFromChan <- tryReadTChan chan - case readFromChan of - Nothing -> pure [] - Just val -> do - moreReads <- readUpTo (pred n) chan - pure (val:moreReads) -- ======= statistics/measurement and logging ======= From 85d10f677314bea16cd20479f6cd5f4095fa8ea2 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 18:50:55 +0200 Subject: [PATCH 20/43] report published posts to statistics --- src/Hash2Pub/PostService.hs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 02278bf..c6bac4a 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -637,7 +637,10 @@ processIncomingPosts serv = forever $ do -- collect async results results <- mapM waitCatch runningJobs - -- TODO: statistics + -- report the count of published posts for statistics + atomically . writeTQueue (statsQueue serv) $ StatsEvent PostPublishEvent (length . rights $ results) 0 -- hashtag published to doesn't matter + pure () + -- | process the pending fetch jobs of delivered post IDs: Delivered posts are tried to be fetched from their URI-ID From 620e998876e449c4694eb1b7b73cfea06d7c0dfd Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 19:25:48 +0200 Subject: [PATCH 21/43] report incoming relay posts to statistics --- src/Hash2Pub/PostService.hs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index c6bac4a..f9b2fc4 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -185,8 +185,10 @@ relayInbox serv tag posts = do -- if noone subscribed to the tag, nothing needs to be done (pure ()) -- otherwise enqueue posts into broadcast queue of the tag - (\queue -> + (\queue -> do liftIO $ forM_ postIDs (atomically . writeTChan queue) + -- report the received post for statistic purposes + liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent RelayReceiveEvent (length postIDs) (hashtagToId tag) ) broadcastChan pure NoContent From f8d30d0cc40b4b13cbcf1c740a33bf37441bd50c Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 9 Sep 2020 19:55:34 +0200 Subject: [PATCH 22/43] report post fetches to statistics --- src/Hash2Pub/PostService.hs | 23 ++++++++++++++++++----- 1 file changed, 18 insertions(+), 5 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index f9b2fc4..608551f 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -251,7 +251,9 @@ postFetch serv postID = do postSet <- liftIO . readTVarIO . ownPosts $ serv if HSet.member postID postSet -- decision: always return the same placeholder post - then pure placeholderPost + then do + liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent IncomingPostFetchEvent 1 0 -- tag fetched for is irrelevant + pure placeholderPost else throwError $ err404 { errBody = "No post found with this ID" } @@ -261,11 +263,14 @@ postMultiFetch serv postIDs = do let idList = Txt.lines postIDs postSet <- liftIO . readTVarIO . ownPosts $ serv -- look up existence of all given post IDs, fail if even one is missing - foldM (\response postID -> + response <- foldM (\response postID -> if HSet.member postID postSet then pure $ placeholderPost <> "\n" <> response else throwError $ err404 { errBody = "No post found with this ID" } ) "" idList + -- this shouldn't be reached in case of error + liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent IncomingPostFetchEvent (length idList) 0 -- tag fetched for is irrelevant + pure response -- | delivery endpoint: inbox for initially publishing a post at an instance @@ -677,9 +682,18 @@ relayWorker serv = forever $ do subscriberMap <- readTVar subscriberMapSTM foldM (\jobAcc' ((subHost, subPort), (postChan, _)) -> do postsToDeliver <- readUpToTChan 500 postChan - -- append fetch job to job list + let postDeliveryAction = runClientM (tagDeliveryClient tag (Txt.unlines postsToDeliver)) (mkClientEnv (httpMan serv) (BaseUrl Http subHost (fromIntegral subPort) "")) + -- append relay push job to job list pure $ if not (null postsToDeliver) - then jobAcc' `D.snoc` runClientM (tagDeliveryClient tag (Txt.unlines postsToDeliver)) (mkClientEnv (httpMan serv) (BaseUrl Http subHost (fromIntegral subPort) "")) + then jobAcc' `D.snoc` (do + deliveryResult <- postDeliveryAction + either + (const $ pure ()) + -- on successful push, record that event for statistics + (const . atomically . writeTQueue (statsQueue serv) $ StatsEvent RelayDeliveryEvent (length postsToDeliver) (hashtagToId tag)) + deliveryResult + pure deliveryResult + ) else jobAcc' ) jobAcc $ HMap.toList subscriberMap ) D.empty subscriptionMap @@ -693,7 +707,6 @@ relayWorker serv = forever $ do runningJobs <- mapM async jobset -- so far just dropping failed attempts, TODO: retry mechanism successfulResults <- rights <$> mapM waitCatch runningJobs - -- TODO: stats pure () From 3c76544afbd017a4704e62c67cf1a21def6324ad Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 10 Sep 2020 12:00:17 +0200 Subject: [PATCH 23/43] launch background worker threads --- src/Hash2Pub/PostService.hs | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 608551f..b943ea6 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -121,8 +121,8 @@ instance DHT d => Service PostService d where -- web server (Warp.runSettings warpSettings $ postServiceApplication thisService) $ concurrently - -- post queue processing - (processIncomingPosts thisService) + -- background processing workers + (launchWorkerThreads thisService) -- statistics/ measurements (launchStatsThreads thisService) -- update thread ID after fork @@ -604,6 +604,12 @@ instance {-# OVERLAPPABLE #-} Read a => MimeUnrender PlainText a where -- TODO: make configurable numParallelDeliveries = 10 +launchWorkerThreads :: DHT d => PostService d -> IO () +launchWorkerThreads serv = concurrently_ + (processIncomingPosts serv) + $ concurrently_ + (fetchTagPosts serv) + (relayWorker serv) -- | process the pending relay inbox of incoming posts from the internal queue: -- Look up responsible relay node for given hashtag and forward post to it From 3ac89d301c9755cab1ccedc548fd1e62a8c48d7c Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 10 Sep 2020 13:09:28 +0200 Subject: [PATCH 24/43] bugfix: subscribe as default if not subscribed yet, when posting to a tag --- src/Hash2Pub/PostService.hs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index b943ea6..d84b58b 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -611,6 +611,7 @@ launchWorkerThreads serv = concurrently_ (fetchTagPosts serv) (relayWorker serv) + -- | process the pending relay inbox of incoming posts from the internal queue: -- Look up responsible relay node for given hashtag and forward post to it processIncomingPosts :: DHT d => PostService d -> IO () @@ -642,7 +643,7 @@ processIncomingPosts serv = forever $ do now <- getPOSIXTime subscriptionStatus <- HMap.lookup (hashtagToId tag) <$> readTVarIO (ownSubscriptions serv) -- if not yet subscribed or subscription expires within 2 minutes, (re)subscribe to tag - when (maybe False (\subLease -> now - subLease < 120) subscriptionStatus) $ + when (maybe True (\subLease -> now - subLease < 120) subscriptionStatus) $ void $ clientSubscribeTo serv tag -- for evaluation, return the tag of the successfully forwarded post From 8f917130c495d932ae7132a88710a83964bb6a10 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 10 Sep 2020 13:14:48 +0200 Subject: [PATCH 25/43] tag normalisation includes lower case conversion --- Hash2Pub.cabal | 2 +- src/Hash2Pub/PostService.hs | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index 5e8d25d..f7a1676 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -47,7 +47,7 @@ extra-source-files: CHANGELOG.md common deps build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist - ghc-options: -Wall -Wpartial-fields -O2 + ghc-options: -Wall -Wpartial-fields diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index d84b58b..7a082d0 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -559,9 +559,9 @@ lookupTagSubscriptions :: Hashtag -> RingMap NodeID a -> Maybe a lookupTagSubscriptions tag = rMapLookup (hashtagToId tag) --- normalise the unicode representation of a string to NFC +-- normalise the unicode representation of a string to NFC and convert to lower case normaliseTag :: Text -> Text -normaliseTag = Txt.fromStrict . normalize NFC . Txt.toStrict +normaliseTag = Txt.toLower . Txt.fromStrict . normalize NFC . Txt.toStrict -- | convert a hashtag to its representation on the DHT From 34ecdd66e1b92af9b9dcd262f9c1c994f614d8bb Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 10 Sep 2020 21:23:21 +0200 Subject: [PATCH 26/43] make stats measurement delay configurable, take speedup into account --- app/Main.hs | 2 ++ src/Hash2Pub/FediChordTypes.hs | 4 ++++ src/Hash2Pub/PostService.hs | 9 +++------ 3 files changed, 9 insertions(+), 6 deletions(-) diff --git a/app/Main.hs b/app/Main.hs index 80c0520..a620fe8 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -73,6 +73,8 @@ readConfig = do , confServicePort = read servicePortString , confServiceHost = confDomainString , confLogfilePath = "../simulationData/logs/" <> confDomainString <> ".log" + , confSpeedupFactor = speedup + , confStatsEvalDelay = 35 * 10^6 `div` speedup } pure (fConf, sConf) diff --git a/src/Hash2Pub/FediChordTypes.hs b/src/Hash2Pub/FediChordTypes.hs index 725031e..4ce20a7 100644 --- a/src/Hash2Pub/FediChordTypes.hs +++ b/src/Hash2Pub/FediChordTypes.hs @@ -459,6 +459,10 @@ data ServiceConf = ServiceConf -- ^ hostname of service , confLogfilePath :: String -- ^ where to store the (measurement) log file + , confStatsEvalDelay :: Int + -- ^ delay between statistic rate measurement samplings, in microseconds + , confSpeedupFactor :: Int + -- While the speedup factor needs to be already included in all } class DHT d where diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 7a082d0..d3e7daf 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -743,9 +743,6 @@ data RelayStats = RelayStats } --- TODO: make delay configurable -statsEvalDelay = 300000 - launchStatsThreads :: PostService d -> IO () launchStatsThreads serv = do @@ -795,7 +792,7 @@ evaluateStatsThread :: PostService d -> TVar RelayStats -> IO () evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop where loop previousTs = do - threadDelay statsEvalDelay + threadDelay $ confStatsEvalDelay (serviceConf serv) -- get and reset the stats accumulator summedStats <- atomically $ do stats <- readTVar statsAcc @@ -806,7 +803,8 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop now <- getPOSIXTime -- evaluate stats rate and replace server stats -- persistently store in a TVar so it can be retrieved later by the DHT - atomically . writeTVar (loadStats serv) . evaluateStats (now - previousTs) $ summedStats + let timePassed = (now - previousTs) * fromIntegral (confSpeedupFactor $ serviceConf serv) + atomically . writeTVar (loadStats serv) . evaluateStats timePassed $ summedStats -- and now what? write a log to file -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate -- later: current (reported) load, target load @@ -833,7 +831,6 @@ evaluateStats timeInterval summedStats = , postFetchRate = postFetchRate summedStats / intervalSeconds } where - -- TODO: take speedup into account intervalSeconds = fromIntegral (fromEnum timeInterval) / 10^12 From 0f9727c05a44875270582a8d57cdbdb6c1078792 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 10 Sep 2020 22:40:56 +0200 Subject: [PATCH 27/43] log the post rates instead of the absolute sums --- src/Hash2Pub/PostService.hs | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index d3e7daf..2170945 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -44,6 +44,7 @@ import Hash2Pub.PostService.API import Hash2Pub.RingMap import Hash2Pub.Utils +import Debug.Trace data PostService d = PostService { serviceConf :: ServiceConf @@ -727,6 +728,7 @@ data StatsEventType = PostPublishEvent -- | Represents measurement event of a 'StatsEventType' with a count relevant for a certain key data StatsEvent = StatsEvent StatsEventType Int NodeID + deriving (Show, Eq) -- | measured rates of relay performance @@ -741,6 +743,7 @@ data RelayStats = RelayStats , postPublishRate :: Double -- ^ rate of initially publishing posts through this instance } + deriving (Show, Eq) @@ -804,7 +807,8 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop -- evaluate stats rate and replace server stats -- persistently store in a TVar so it can be retrieved later by the DHT let timePassed = (now - previousTs) * fromIntegral (confSpeedupFactor $ serviceConf serv) - atomically . writeTVar (loadStats serv) . evaluateStats timePassed $ summedStats + let rateStats = evaluateStats timePassed summedStats + atomically $ writeTVar (loadStats serv) rateStats -- and now what? write a log to file -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate -- later: current (reported) load, target load @@ -814,7 +818,7 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop , show . sum . relayDeliveryRates , show . postPublishRate , show . postFetchRate - ] <*> pure summedStats)) <> "\n" + ] <*> pure rateStats)) <> "\n" loop now @@ -831,7 +835,7 @@ evaluateStats timeInterval summedStats = , postFetchRate = postFetchRate summedStats / intervalSeconds } where - intervalSeconds = fromIntegral (fromEnum timeInterval) / 10^12 + intervalSeconds = realToFrac timeInterval emptyStats :: RelayStats From e12d8ef70af043b17f764409767c69850a4dac72 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 10 Sep 2020 23:54:51 +0200 Subject: [PATCH 28/43] properly format stats log numbers: no e-notation --- Hash2Pub.cabal | 2 +- src/Hash2Pub/PostService.hs | 17 ++++++++++------- 2 files changed, 11 insertions(+), 8 deletions(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index f7a1676..92ec096 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -46,7 +46,7 @@ category: Network extra-source-files: CHANGELOG.md common deps - build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist + build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist, formatting ghc-options: -Wall -Wpartial-fields diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 2170945..662b0a1 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -35,6 +35,7 @@ import qualified Network.HTTP.Types as HTTPT import System.Random import Text.Read (readEither) +import Formatting (float, format, (%), fixed) import qualified Network.Wai.Handler.Warp as Warp import Servant import Servant.Client @@ -113,7 +114,10 @@ instance DHT d => Service PostService d where port' = fromIntegral (confServicePort conf) warpSettings = Warp.setPort port' . Warp.setHost (fromString . confServiceHost $ conf) $ Warp.defaultSettings -- log a start message, this also truncates existing files - TxtI.writeFile (confLogfilePath conf) "# Starting mock relay implementation\n" + TxtI.writeFile (confLogfilePath conf) $ Txt.unlines + [ "# Starting mock relay implementation\n" + , "#relay receive rate ;relay delivery rate ;instance publish rate ;instance fetch rate" + ] -- Run 'concurrently_' from another thread to be able to return the -- 'PostService'. -- Terminating that parent thread will make all child threads terminate as well. @@ -813,12 +817,11 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate -- later: current (reported) load, target load TxtI.appendFile (confLogfilePath . serviceConf $ serv) $ - Txt.intercalate ";" (Txt.pack <$> ( - [ show . sum . relayReceiveRates - , show . sum . relayDeliveryRates - , show . postPublishRate - , show . postFetchRate - ] <*> pure rateStats)) <> "\n" + format ((fixed 20) % ";" % (fixed 20) % ";" % (fixed 20) % ";" % (fixed 20) % "\n") + (sum . relayReceiveRates $ rateStats) + (sum . relayDeliveryRates $ rateStats) + (postPublishRate rateStats) + (postFetchRate rateStats) loop now From da579a0756986c5fb5b6aa1ef4848ed6fda52ae3 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Fri, 11 Sep 2020 00:36:00 +0200 Subject: [PATCH 29/43] decrease logging verbosity --- src/Hash2Pub/FediChord.hs | 4 ---- src/Hash2Pub/PostService.hs | 16 ++++++++-------- 2 files changed, 8 insertions(+), 12 deletions(-) diff --git a/src/Hash2Pub/FediChord.hs b/src/Hash2Pub/FediChord.hs index 33044fa..9f14a1e 100644 --- a/src/Hash2Pub/FediChord.hs +++ b/src/Hash2Pub/FediChord.hs @@ -345,7 +345,6 @@ nodeCacheWriter nsSTM = -- | Periodically iterate through cache, clean up expired entries and verify unverified ones nodeCacheVerifyThread :: LocalNodeStateSTM s -> IO () nodeCacheVerifyThread nsSTM = forever $ do - putStrLn "cache verify run: begin" -- get cache (ns, cache, maxEntryAge) <- atomically $ do ns <- readTVar nsSTM @@ -398,7 +397,6 @@ nodeCacheVerifyThread nsSTM = forever $ do forkIO $ sendQueryIdMessages targetID latestNs (Just (1 + jEntriesPerSlice latestNs)) (nodesToQuery targetID) >> pure () -- ask for 1 entry more than j because of querying the middle ) - putStrLn "cache verify run: end" threadDelay $ fromEnum (maxEntryAge / 20) `div` 10^6 -- convert from pico to milliseconds @@ -465,7 +463,6 @@ stabiliseThread :: Service s (RealNodeSTM s) => LocalNodeStateSTM s -> IO () stabiliseThread nsSTM = forever $ do oldNs <- readTVarIO nsSTM - putStrLn "stabilise run: begin" -- iterate through the same snapshot, collect potential new neighbours -- and nodes to be deleted, and modify these changes only at the end of @@ -544,7 +541,6 @@ stabiliseThread nsSTM = forever $ do ) newPredecessor - putStrLn "stabilise run: end" stabiliseDelay <- confStabiliseInterval . nodeConfig <$> readTVarIO (parentRealNode newNs) threadDelay stabiliseDelay where diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 662b0a1..5e6ddcb 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -673,13 +673,13 @@ fetchTagPosts serv = forever $ do resp <- try $ HTTP.httpLbs fetchReq (httpMan serv) :: IO (Either HTTP.HttpException (HTTP.Response BSUL.ByteString)) case resp of Right response -> - if HTTPT.statusCode (HTTP.responseStatus response) == 200 - then - -- success, TODO: statistics - putStrLn "post fetch success" - else - -- TODO error handling, retry - pure () + -- TODO error handling, retry + --if HTTPT.statusCode (HTTP.responseStatus response) == 200 + -- then + -- -- success, TODO: statistics + -- putStrLn "post fetch success" + -- else + pure () Left _ -> -- TODO error handling, retry pure () @@ -817,7 +817,7 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate -- later: current (reported) load, target load TxtI.appendFile (confLogfilePath . serviceConf $ serv) $ - format ((fixed 20) % ";" % (fixed 20) % ";" % (fixed 20) % ";" % (fixed 20) % "\n") + format (fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % "\n") (sum . relayReceiveRates $ rateStats) (sum . relayDeliveryRates $ rateStats) (postPublishRate rateStats) From 1fc264a226419355683657e1f1ca692062c0dcea Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Fri, 11 Sep 2020 14:04:35 +0200 Subject: [PATCH 30/43] manage logging via file handle reason: `appendFile` combined with lazy evaluation lead to exhaustion of open file descriptors, as each file is opened again for each write and due to lazy evaluation is kept open multiple times. --- src/Hash2Pub/PostService.hs | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 5e6ddcb..bb94e86 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -32,10 +32,11 @@ import Data.Time.Clock.POSIX import Data.Typeable (Typeable) import qualified Network.HTTP.Client as HTTP import qualified Network.HTTP.Types as HTTPT +import System.IO import System.Random import Text.Read (readEither) -import Formatting (float, format, (%), fixed) +import Formatting (fixed, float, format, (%)) import qualified Network.Wai.Handler.Warp as Warp import Servant import Servant.Client @@ -66,6 +67,7 @@ data PostService d = PostService , httpMan :: HTTP.Manager , statsQueue :: TQueue StatsEvent , loadStats :: TVar RelayStats + , logFileHandle :: Handle } deriving (Typeable) @@ -96,6 +98,7 @@ instance DHT d => Service PostService d where httpMan' <- HTTP.newManager HTTP.defaultManagerSettings statsQueue' <- newTQueueIO loadStats' <- newTVarIO emptyStats + loggingFile <- openFile (confLogfilePath conf) WriteMode let thisService = PostService { serviceConf = conf @@ -110,12 +113,13 @@ instance DHT d => Service PostService d where , httpMan = httpMan' , statsQueue = statsQueue' , loadStats = loadStats' + , logFileHandle = loggingFile } port' = fromIntegral (confServicePort conf) warpSettings = Warp.setPort port' . Warp.setHost (fromString . confServiceHost $ conf) $ Warp.defaultSettings -- log a start message, this also truncates existing files - TxtI.writeFile (confLogfilePath conf) $ Txt.unlines - [ "# Starting mock relay implementation\n" + TxtI.hPutStrLn loggingFile $ Txt.unlines + [ "# Starting mock relay implementation" , "#relay receive rate ;relay delivery rate ;instance publish rate ;instance fetch rate" ] -- Run 'concurrently_' from another thread to be able to return the @@ -816,8 +820,8 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop -- and now what? write a log to file -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate -- later: current (reported) load, target load - TxtI.appendFile (confLogfilePath . serviceConf $ serv) $ - format (fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % "\n") + TxtI.hPutStrLn (logFileHandle serv) $ + format (fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20) (sum . relayReceiveRates $ rateStats) (sum . relayDeliveryRates $ rateStats) (postPublishRate rateStats) From 3c28cde9421a296acf687e183e811f557e92ecef Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Sat, 12 Sep 2020 15:45:03 +0200 Subject: [PATCH 31/43] catch and print all Socket bind exceptions --- src/Hash2Pub/DHTProtocol.hs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/src/Hash2Pub/DHTProtocol.hs b/src/Hash2Pub/DHTProtocol.hs index 3639c08..c86c0f1 100644 --- a/src/Hash2Pub/DHTProtocol.hs +++ b/src/Hash2Pub/DHTProtocol.hs @@ -865,7 +865,7 @@ mkServerSocket ip port = do sockAddr <- addrAddress <$> resolve (Just $ show . fromHostAddress6 $ ip) (Just port) sock <- socket AF_INET6 Datagram defaultProtocol setSocketOption sock IPv6Only 1 - bind sock sockAddr + bind sock sockAddr `catch` (\e -> putStrLn $ "Caught exception while bind " <> show sock <> " " <> show sockAddr <> ": " <> show (e :: SomeException)) pure sock -- | create a UDP datagram socket, connected to a destination. @@ -881,6 +881,6 @@ mkSendSocket srcIp dest destPort = do setSocketOption sendSock IPv6Only 1 -- bind to the configured local IP to make sure that outgoing packets are sent from -- this source address - bind sendSock srcAddr + bind sendSock srcAddr `catch` (\e -> putStrLn $ "Caught exception while mkSendSocket bind " <> show sendSock <> " " <> show srcAddr <> ": " <> show (e :: SomeException)) connect sendSock destAddr pure sendSock From a0e7142a7d8f8eff19469d71d4d07b8c8fc29021 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 14 Sep 2020 14:57:25 +0200 Subject: [PATCH 32/43] report number of subscriptions --- Hash2Pub.cabal | 2 +- app/Experiment.hs | 2 +- app/Main.hs | 2 +- src/Hash2Pub/PostService.hs | 21 ++++++++++++++++----- 4 files changed, 19 insertions(+), 8 deletions(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index 92ec096..7be7ecf 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -47,7 +47,7 @@ extra-source-files: CHANGELOG.md common deps build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist, formatting - ghc-options: -Wall -Wpartial-fields + ghc-options: -Wall -Wpartial-fields -O2 diff --git a/app/Experiment.hs b/app/Experiment.hs index ffa8869..a999dea 100644 --- a/app/Experiment.hs +++ b/app/Experiment.hs @@ -33,7 +33,7 @@ parseSchedule = fmap (parseEntry . Txt.split (== ';')) . Txt.lines where parseEntry [delayT, contactT, tag] = (read $ Txt.unpack delayT, tag, read $ Txt.unpack contactT) - parseEntry _ = error "invalid schedule input format" + parseEntry entry = error $ "invalid schedule input format: " <> show entry executeSchedule :: Int -- ^ speedup factor -> [(Int, Hashtag, (String, Int))] -- ^ [(delay in microseconds, hashtag, (hostname, port))] diff --git a/app/Main.hs b/app/Main.hs index a620fe8..c10e0c8 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -74,7 +74,7 @@ readConfig = do , confServiceHost = confDomainString , confLogfilePath = "../simulationData/logs/" <> confDomainString <> ".log" , confSpeedupFactor = speedup - , confStatsEvalDelay = 35 * 10^6 `div` speedup + , confStatsEvalDelay = 120 * 10^6 `div` speedup } pure (fConf, sConf) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index bb94e86..f2a8a18 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -36,7 +36,7 @@ import System.IO import System.Random import Text.Read (readEither) -import Formatting (fixed, float, format, (%)) +import Formatting (fixed, format, int, (%)) import qualified Network.Wai.Handler.Warp as Warp import Servant import Servant.Client @@ -67,6 +67,7 @@ data PostService d = PostService , httpMan :: HTTP.Manager , statsQueue :: TQueue StatsEvent , loadStats :: TVar RelayStats + -- ^ current load stats, replaced periodically , logFileHandle :: Handle } deriving (Typeable) @@ -120,7 +121,7 @@ instance DHT d => Service PostService d where -- log a start message, this also truncates existing files TxtI.hPutStrLn loggingFile $ Txt.unlines [ "# Starting mock relay implementation" - , "#relay receive rate ;relay delivery rate ;instance publish rate ;instance fetch rate" + , "#relay receive rate ;relay delivery rate ;instance publish rate ;instance fetch rate ;total subscriptions" ] -- Run 'concurrently_' from another thread to be able to return the -- 'PostService'. @@ -681,7 +682,6 @@ fetchTagPosts serv = forever $ do --if HTTPT.statusCode (HTTP.responseStatus response) == 200 -- then -- -- success, TODO: statistics - -- putStrLn "post fetch success" -- else pure () Left _ -> @@ -723,6 +723,7 @@ relayWorker serv = forever $ do runningJobs <- mapM async jobset -- so far just dropping failed attempts, TODO: retry mechanism successfulResults <- rights <$> mapM waitCatch runningJobs + putStrLn $ "successfully relayed " <> show (length successfulResults) pure () @@ -818,16 +819,26 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop let rateStats = evaluateStats timePassed summedStats atomically $ writeTVar (loadStats serv) rateStats -- and now what? write a log to file - -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate + -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate; subscriberSum -- later: current (reported) load, target load + subscriberSum <- sumSubscribers TxtI.hPutStrLn (logFileHandle serv) $ - format (fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20) + format (fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % int ) (sum . relayReceiveRates $ rateStats) (sum . relayDeliveryRates $ rateStats) (postPublishRate rateStats) (postFetchRate rateStats) + subscriberSum loop now + sumSubscribers = do + tagMap <- readTVarIO $ subscribers serv + foldM (\subscriberSum (subscriberMapSTM, _, _) -> do + subscriberMap <- readTVarIO subscriberMapSTM + pure $ subscriberSum + HMap.size subscriberMap + ) + 0 tagMap + -- | Evaluate the accumulated statistic events: Currently mostly calculates the event -- rates by dividing through the collection time frame From c036dea7f9f91a03f3d068ac9cb3a827961c5db4 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 14 Sep 2020 15:49:44 +0200 Subject: [PATCH 33/43] periodically purge expired subscriptions --- app/Main.hs | 2 +- src/Hash2Pub/PostService.hs | 26 ++++++++++++++++++++++---- 2 files changed, 23 insertions(+), 5 deletions(-) diff --git a/app/Main.hs b/app/Main.hs index c10e0c8..d7be0a5 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -69,7 +69,7 @@ readConfig = do , confRequestRetries = 3 } sConf = ServiceConf - { confSubscriptionExpiryTime = fromIntegral $ 2*3600 `div` speedup + { confSubscriptionExpiryTime = fromIntegral 12*3600 / fromIntegral speedup , confServicePort = read servicePortString , confServiceHost = confDomainString , confLogfilePath = "../simulationData/logs/" <> confDomainString <> ".log" diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index f2a8a18..69f1b13 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -618,8 +618,26 @@ launchWorkerThreads :: DHT d => PostService d -> IO () launchWorkerThreads serv = concurrently_ (processIncomingPosts serv) $ concurrently_ - (fetchTagPosts serv) - (relayWorker serv) + (purgeSubscriptionsThread serv) + $ concurrently_ + (fetchTagPosts serv) + (relayWorker serv) + + +-- | periodically remove expired subscription entries from relay subscribers +purgeSubscriptionsThread :: PostService d -> IO () +purgeSubscriptionsThread serv = forever $ do + -- read config + now <- getPOSIXTime + let + purgeInterval = confSubscriptionExpiryTime (serviceConf serv) / 10 + -- no need to atomically lock this, as newly incoming subscriptions do not + -- need to be purged + tagMap <- readTVarIO $ subscribers serv + forM_ tagMap $ \(subscriberMapSTM, _, _) -> + -- but each subscriberMap needs to be modified atomically + atomically . modifyTVar' subscriberMapSTM $ HMap.filter (\(_, ts) -> ts > now) + threadDelay $ fromEnum purgeInterval `div` 10^6 -- | process the pending relay inbox of incoming posts from the internal queue: @@ -652,8 +670,8 @@ processIncomingPosts serv = forever $ do -- idea for the experiment: each post publication makes the initial posting instance subscribe to all contained tags now <- getPOSIXTime subscriptionStatus <- HMap.lookup (hashtagToId tag) <$> readTVarIO (ownSubscriptions serv) - -- if not yet subscribed or subscription expires within 2 minutes, (re)subscribe to tag - when (maybe True (\subLease -> now - subLease < 120) subscriptionStatus) $ + -- if not yet subscribed or subscription expires within 5 minutes, (re)subscribe to tag + when (maybe True (\subLease -> now - subLease < 300) subscriptionStatus) $ void $ clientSubscribeTo serv tag -- for evaluation, return the tag of the successfully forwarded post From bb17b136d6505090d3f10a1cc231f404d1abec20 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 16 Sep 2020 01:54:40 +0200 Subject: [PATCH 34/43] increase stabilise interval --- app/Main.hs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/app/Main.hs b/app/Main.hs index d7be0a5..d02507e 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -59,7 +59,7 @@ readConfig = do , confIP = toHostAddress6 . read $ ipString , confDhtPort = read portString , confBootstrapNodes = confBootstrapNodes' - , confStabiliseInterval = 60 * 10^6 + , confStabiliseInterval = 80 * 10^6 , confBootstrapSamplingInterval = 180 * 10^6 `div` speedup , confMaxLookupCacheAge = 300 / fromIntegral speedup , confJoinAttemptsInterval = 60 * 10^6 `div` speedup @@ -69,7 +69,7 @@ readConfig = do , confRequestRetries = 3 } sConf = ServiceConf - { confSubscriptionExpiryTime = fromIntegral 12*3600 / fromIntegral speedup + { confSubscriptionExpiryTime = 12*3600 / fromIntegral speedup , confServicePort = read servicePortString , confServiceHost = confDomainString , confLogfilePath = "../simulationData/logs/" <> confDomainString <> ".log" From a2f268d374982c3987b154bae66bfffb1e383e3b Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 16 Sep 2020 01:54:50 +0200 Subject: [PATCH 35/43] improve logging: line buffering, time stamps contributes to #60 --- src/Hash2Pub/PostService.hs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 69f1b13..bd83506 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -100,6 +100,7 @@ instance DHT d => Service PostService d where statsQueue' <- newTQueueIO loadStats' <- newTVarIO emptyStats loggingFile <- openFile (confLogfilePath conf) WriteMode + hSetBuffering loggingFile LineBuffering let thisService = PostService { serviceConf = conf @@ -121,7 +122,7 @@ instance DHT d => Service PostService d where -- log a start message, this also truncates existing files TxtI.hPutStrLn loggingFile $ Txt.unlines [ "# Starting mock relay implementation" - , "#relay receive rate ;relay delivery rate ;instance publish rate ;instance fetch rate ;total subscriptions" + , "#time stamp ; relay receive rate ;relay delivery rate ;instance publish rate ;instance fetch rate ;total subscriptions" ] -- Run 'concurrently_' from another thread to be able to return the -- 'PostService'. @@ -841,7 +842,8 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop -- later: current (reported) load, target load subscriberSum <- sumSubscribers TxtI.hPutStrLn (logFileHandle serv) $ - format (fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % int ) + format (fixed 9 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % fixed 20 % ";" % int ) + (realToFrac now :: Double) (sum . relayReceiveRates $ rateStats) (sum . relayDeliveryRates $ rateStats) (postPublishRate rateStats) From f5de7601bbef988d82e756d42e3f2197f2646325 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 16 Sep 2020 13:49:26 +0200 Subject: [PATCH 36/43] do not store published posts for reducing memory consumption --- src/Hash2Pub/PostService.hs | 36 +++++++++++++++--------------------- 1 file changed, 15 insertions(+), 21 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index bd83506..75bdd33 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -57,7 +57,7 @@ data PostService d = PostService -- ^ for each tag store the subscribers + their queue , ownSubscriptions :: TVar (HMap.HashMap NodeID POSIXTime) -- ^ tags subscribed by the own node have an assigned lease time - , ownPosts :: TVar (HSet.HashSet Text) + --, ownPosts :: TVar (HSet.HashSet Text) -- ^ just store the existence of posts for saving memory, , relayInQueue :: TQueue (Hashtag, PostID, PostContent) -- ^ Queue for processing incoming posts of own instance asynchronously @@ -92,7 +92,7 @@ instance DHT d => Service PostService d where threadVar <- newTVarIO =<< myThreadId -- own thread ID as placeholder subscriberVar <- newTVarIO emptyRMap ownSubsVar <- newTVarIO HMap.empty - ownPostVar <- newTVarIO HSet.empty + --ownPostVar <- newTVarIO HSet.empty relayInQueue' <- newTQueueIO postFetchQueue' <- newTQueueIO migrationsInProgress' <- newTVarIO HMap.empty @@ -108,7 +108,7 @@ instance DHT d => Service PostService d where , serviceThread = threadVar , subscribers = subscriberVar , ownSubscriptions = ownSubsVar - , ownPosts = ownPostVar + --, ownPosts = ownPostVar , relayInQueue = relayInQueue' , postFetchQueue = postFetchQueue' , migrationsInProgress = migrationsInProgress' @@ -258,28 +258,23 @@ subscriptionDelivery serv senderID subList = do -- | endpoint for fetching a post by its ID postFetch :: PostService d -> Text -> Handler Text -postFetch serv postID = do - postSet <- liftIO . readTVarIO . ownPosts $ serv - if HSet.member postID postSet - -- decision: always return the same placeholder post - then do - liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent IncomingPostFetchEvent 1 0 -- tag fetched for is irrelevant - pure placeholderPost - else throwError $ err404 { errBody = "No post found with this ID" } +postFetch serv _ = do + -- decision: for saving memory do not store published posts, just + -- pretend there is a post for each requested ID + liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent IncomingPostFetchEvent 1 0 -- tag fetched for is irrelevant + pure placeholderPost -- | endpoint for fetching multiple posts of this instance by their IDs postMultiFetch :: PostService d -> Text -> Handler Text postMultiFetch serv postIDs = do - let idList = Txt.lines postIDs - postSet <- liftIO . readTVarIO . ownPosts $ serv - -- look up existence of all given post IDs, fail if even one is missing - response <- foldM (\response postID -> - if HSet.member postID postSet - then pure $ placeholderPost <> "\n" <> response - else throwError $ err404 { errBody = "No post found with this ID" } + let + idList = Txt.lines postIDs + -- decision: for saving memory do not store published posts, just + -- pretend there is a post for each requested ID + response = foldl (\response' _ -> + placeholderPost <> "\n" <> response' ) "" idList - -- this shouldn't be reached in case of error liftIO . atomically . writeTQueue (statsQueue serv) $ StatsEvent IncomingPostFetchEvent (length idList) 0 -- tag fetched for is irrelevant pure response @@ -292,8 +287,7 @@ postInbox serv post = do containedTags = fmap (normaliseTag . Txt.tail) . filter ((==) '#' . Txt.head) . Txt.words $ post -- generate post ID postId <- liftIO $ Txt.pack . show <$> (randomRIO (0, 2^(128::Integer)-1) :: IO Integer) - -- add ID to own posts - liftIO . atomically $ modifyTVar' (ownPosts serv) (HSet.insert postId) + -- decision: for saving memory do not store published post IDs, just deliver a post for any requested ID -- enqueue a relay job for each tag liftIO $ forM_ (containedTags :: [Text]) (\tag -> atomically $ writeTQueue (relayInQueue serv) (tag, postId, post) From 556b69d887dcbf99287d819a7625f59a6163476c Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Wed, 16 Sep 2020 16:07:39 +0200 Subject: [PATCH 37/43] increase subscription lease to 1 (simulated) day for achieving higher subscriber numbers --- app/Main.hs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/Main.hs b/app/Main.hs index d02507e..eac223d 100644 --- a/app/Main.hs +++ b/app/Main.hs @@ -69,7 +69,7 @@ readConfig = do , confRequestRetries = 3 } sConf = ServiceConf - { confSubscriptionExpiryTime = 12*3600 / fromIntegral speedup + { confSubscriptionExpiryTime = 24*3600 / fromIntegral speedup , confServicePort = read servicePortString , confServiceHost = confDomainString , confLogfilePath = "../simulationData/logs/" <> confDomainString <> ".log" From eee40ce4fb7bd74fd0ebd8b9455a5cab22bcd5e5 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Thu, 17 Sep 2020 02:03:45 +0200 Subject: [PATCH 38/43] add log messages for failed relays as well --- src/Hash2Pub/PostService.hs | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/src/Hash2Pub/PostService.hs b/src/Hash2Pub/PostService.hs index 75bdd33..ffeef17 100644 --- a/src/Hash2Pub/PostService.hs +++ b/src/Hash2Pub/PostService.hs @@ -12,14 +12,14 @@ import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM import Control.Exception (Exception (..), try) -import Control.Monad (foldM, forM, forM_, forever, void, - when) +import Control.Monad (foldM, forM, forM_, forever, unless, + void, when) import Control.Monad.IO.Class (liftIO) import Data.Bifunctor import qualified Data.ByteString.Lazy.UTF8 as BSUL import qualified Data.ByteString.UTF8 as BSU import qualified Data.DList as D -import Data.Either (rights) +import Data.Either (lefts, rights) import qualified Data.HashMap.Strict as HMap import qualified Data.HashSet as HSet import Data.Maybe (fromJust, isJust) @@ -57,8 +57,6 @@ data PostService d = PostService -- ^ for each tag store the subscribers + their queue , ownSubscriptions :: TVar (HMap.HashMap NodeID POSIXTime) -- ^ tags subscribed by the own node have an assigned lease time - --, ownPosts :: TVar (HSet.HashSet Text) - -- ^ just store the existence of posts for saving memory, , relayInQueue :: TQueue (Hashtag, PostID, PostContent) -- ^ Queue for processing incoming posts of own instance asynchronously , postFetchQueue :: TQueue PostID @@ -325,6 +323,7 @@ tagSubscribe serv hashtag origin = do let leaseTime = now + confSubscriptionExpiryTime (serviceConf serv) -- setup subscription entry _ <- liftIO . atomically $ setupSubscriberChannel (subscribers serv) hashtag (BSU.toString $ HTTP.host req, HTTP.port req) leaseTime + --liftIO . putStrLn $ "just got a subscription to " <> Txt.unpack hashtag pure $ round leaseTime @@ -427,10 +426,12 @@ clientSubscribeTo serv tag = do Left (FailureResponse _ fresp) |(HTTPT.statusCode . responseStatusCode $ fresp) == 410 && allowRetry -> do -- responsibility gone, force new lookup newRes <- forceLookupKey (baseDHT serv) (Txt.unpack tag) + --putStrLn $ "failed subscribing to " <> Txt.unpack tag <> " on " <> foundHost doSubscribe newRes False Left err -> pure . Left . show $ err Right lease -> do atomically . modifyTVar' (ownSubscriptions serv) $ HMap.insert (hashtagToId tag) (fromInteger lease) + --putStrLn $ "just subscribed to " <> Txt.unpack tag <> " on " <> foundHost pure . Right $ lease ) lookupResponse @@ -735,7 +736,11 @@ relayWorker serv = forever $ do forM_ (chunksOf numParallelDeliveries jobsToProcess) $ \jobset -> do runningJobs <- mapM async jobset -- so far just dropping failed attempts, TODO: retry mechanism - successfulResults <- rights <$> mapM waitCatch runningJobs + results <- mapM waitCatch runningJobs + let + successfulResults = rights results + unsuccessfulResults = lefts results + unless (null unsuccessfulResults) $ putStrLn ("ERR: " <> show (length unsuccessfulResults) <> " failed deliveries!") putStrLn $ "successfully relayed " <> show (length successfulResults) pure () @@ -829,7 +834,7 @@ evaluateStatsThread serv statsAcc = getPOSIXTime >>= loop -- evaluate stats rate and replace server stats -- persistently store in a TVar so it can be retrieved later by the DHT let timePassed = (now - previousTs) * fromIntegral (confSpeedupFactor $ serviceConf serv) - let rateStats = evaluateStats timePassed summedStats + rateStats = evaluateStats timePassed summedStats atomically $ writeTVar (loadStats serv) rateStats -- and now what? write a log to file -- format: total relayReceiveRates;total relayDeliveryRates;postFetchRate;postPublishRate; subscriberSum From d8b21860166d4b6e894c32f92a2df94e2fb289f8 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 21 Sep 2020 22:14:33 +0200 Subject: [PATCH 39/43] make inclusion of HIE overlay conditional as well --- default.nix | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/default.nix b/default.nix index cea4aa3..126975a 100644 --- a/default.nix +++ b/default.nix @@ -14,13 +14,13 @@ let name = "nixpkgs-pinned"; url = https://github.com/NixOS/nixpkgs/; ref = "refs/heads/release-20.03"; - rev = "de3780b937d2984f9b5e20d191f23be4f857b3aa"; + rev = "faf5bdea5d9f0f9de26deaa7e864cdcd3b15b4e8"; }) { # Pass no config for purity config = {}; - overlays = [ + overlays = if withHIE then [ (import all-hies {}).overlay - ]; + ] else []; }; hp = pkgs.haskell.packages."${compiler}"; src = pkgs.nix-gitignore.gitignoreSource [] ./.; From d7355aa04d1cd307a83e92ddf4689b4dda5f0627 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Tue, 22 Sep 2020 19:47:39 +0200 Subject: [PATCH 40/43] increase HTTP timeout for initial post publication to 60 seconds After a while, experiments made some publication events time-out. Increasing the timeout just in case, although it i likely to be a mere symptom but the core fault. --- app/Experiment.hs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/app/Experiment.hs b/app/Experiment.hs index a999dea..f2fa586 100644 --- a/app/Experiment.hs +++ b/app/Experiment.hs @@ -40,7 +40,7 @@ executeSchedule :: Int -- ^ speedup factor -> IO () executeSchedule speedup events = do -- initialise HTTP manager - httpMan <- HTTP.newManager HTTP.defaultManagerSettings + httpMan <- HTTP.newManager $ HTTP.defaultManagerSettings { HTTP.managerResponseTimeout = HTTP.responseTimeoutMicro 60000000 } forM_ events $ \(delay, tag, (pubHost, pubPort)) -> do _ <- forkIO $ clientPublishPost httpMan pubHost pubPort ("foobar #" <> tag) From 9d8df6d3d8b82bd78565088bf01593250d7bdcc2 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Fri, 2 Oct 2020 02:36:02 +0200 Subject: [PATCH 41/43] make the multithread-runtime use all cores by default --- Hash2Pub.cabal | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index b343df3..1d3ac7f 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -91,7 +91,7 @@ executable Hash2Pub -- Base language which the package is written in. default-language: Haskell2010 - ghc-options: -threaded + ghc-options: -threaded -rtsopts -with-rtsopts=-N executable Experiment -- experiment runner From ea14ff9b09032a68814ccfa59c24d8f104975ae9 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Fri, 1 Jan 2021 14:30:33 +0100 Subject: [PATCH 42/43] update ghc to 8.6.4, nixpkgs base to 20.09 - relaxes some version constraints as dirty update quickfix - removes hie integration as that project is abandoned, todo: switch to haskell-languageserver instead --- Hash2Pub.cabal | 2 +- default.nix | 19 +++++-------------- shell.nix | 2 +- 3 files changed, 7 insertions(+), 16 deletions(-) diff --git a/Hash2Pub.cabal b/Hash2Pub.cabal index 1d3ac7f..376d675 100644 --- a/Hash2Pub.cabal +++ b/Hash2Pub.cabal @@ -46,7 +46,7 @@ category: Network extra-source-files: CHANGELOG.md common deps - build-depends: base ^>=4.12.0.0, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=2.8.0.1, time ^>=1.8.0.2, cmdargs ^>= 0.10, cryptonite ^>= 0.25, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist, formatting + build-depends: base >=4, containers ^>=0.6.0.1, bytestring, utf8-string ^>=1.0.1.1, network ^>=3.1, time, cmdargs ^>= 0.10, cryptonite, memory, async, stm, asn1-encoding, asn1-types, asn1-parse, publicsuffix, network-byte-order, safe, iproute, mtl, random, servant, servant-server, servant-client, warp, text, unordered-containers, hashable, unicode-transforms, http-client, http-types, unbounded-delays, dlist, formatting ghc-options: -Wall -Wpartial-fields -O2 diff --git a/default.nix b/default.nix index 126975a..a3f7640 100644 --- a/default.nix +++ b/default.nix @@ -1,26 +1,18 @@ { - compiler ? "ghc865", - withHIE ? false + compiler ? "ghc884" }: let - # pin all-hies for getting the language server - all-hies = fetchTarball { - url = "https://github.com/infinisil/all-hies/tarball/b8fb659620b99b4a393922abaa03a1695e2ca64d"; - sha256 = "sha256:0br6wsqpfk1lzz90f7zw439w1ir2p54268qilw9l2pk6yz7ganfx"; - }; pkgs = import ( builtins.fetchGit { name = "nixpkgs-pinned"; url = https://github.com/NixOS/nixpkgs/; - ref = "refs/heads/release-20.03"; - rev = "faf5bdea5d9f0f9de26deaa7e864cdcd3b15b4e8"; + ref = "refs/heads/release-20.09"; + rev = "e065200fc90175a8f6e50e76ef10a48786126e1c"; }) { # Pass no config for purity config = {}; - overlays = if withHIE then [ - (import all-hies {}).overlay - ] else []; + overlays = []; }; hp = pkgs.haskell.packages."${compiler}"; src = pkgs.nix-gitignore.gitignoreSource [] ./.; @@ -38,7 +30,6 @@ in hlint stylish-haskell pkgs.python3Packages.asn1ate - ] - ++ (if withHIE then [ hie ] else []); + ]; }; } diff --git a/shell.nix b/shell.nix index dafd212..82fb296 100644 --- a/shell.nix +++ b/shell.nix @@ -1 +1 @@ -(import ./default.nix {withHIE = true;}).shell +(import ./default.nix {}).shell From b46f66e2c0dff72cda7723f2344a0fc68f27c411 Mon Sep 17 00:00:00 2001 From: Trolli Schmittlauch Date: Mon, 16 Aug 2021 20:12:53 +0200 Subject: [PATCH 43/43] update Readme with latest branch name and pointer to SocialHub --- Readme.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/Readme.md b/Readme.md index daf9e38..e3cff3d 100644 --- a/Readme.md +++ b/Readme.md @@ -1,7 +1,7 @@ # Hash2Pub ***This is heavily WIP and does not provide any useful functionality yet***. -I aim for always having the master branch at a state where it builds and tests pass. +I aim for always having the `mainline` branch in a state where it builds and tests pass. A fully-decentralised relay for global hashtag federation in [ActivityPub](https://activitypub.rocks) based on a distributed hash table. It allows querying and subscribing to all posts of a certain hashtag and is implemented in Haskell. @@ -10,6 +10,8 @@ This is the practical implementation of the concept presented in the paper [Dece The ASN.1 module schema used for DHT messages can be found in `FediChord.asn1`. +For further questions and discussins, please refer to the **Hash2Pub topic in [SocialHub](https://socialhub.activitypub.rocks/c/software/hash2pub/48)**. + ## Building The project and its developent environment are built with [Nix](https://nixos.org/nix/).