Merge #1900 #1907

1900: Label mux threads r=karknu a=karknu



1907: ImmutableDB: allow reading concurrently with appending r=mrBliss a=mrBliss

Read the potentially stale value of the internal state provided by the
`StrictMVar` instead of blocking on the `TMVar` in case an appender has locked
it. Reads can safely happen concurrently with appends, so this will allow for
some more concurrency.

I don't expect this to make any impact for a "client" node that is syncing, as
it will typically not read from its ImmutableDB, only append to it, which is
necessarily sequential. However, a "server" node might have to read
headers/blocks from its ImmutableDB to serve to other nodes. Serving multiple
requests won't require taking + putting a `TMVar` anymore, it's just reading a
`TVar` now.

I expect to see the most improvement for a node serving others while it is
also still syncing itself. Blocks being appended to the ImmutableDB will
happen concurrently with all read requests.

Co-authored-by: Karl Knutsson <[email protected]>
Co-authored-by: Thomas Winant <[email protected]>

Author: 3 people

network-mux/src/Network/Mux.hs

@@ -177,11 +177,11 @@ muxStart tracer (MuxApplication ptcls) bearer = do
 
     muxerJob tq cnt =
       JobPool.Job (mux cnt MuxState { egressQueue   = tq,  Egress.bearer })
-                  MuxerException
+                  MuxerException "muxer"
 
     demuxerJob tbl =
       JobPool.Job (demux DemuxState { dispatchTable = tbl, Ingress.bearer })
-                  DemuxerException
+                  DemuxerException "demuxer"
 
     miniProtocolInitiatorJob = miniProtocolJob selectInitiator ModeInitiator
     miniProtocolResponderJob = miniProtocolJob selectResponder ModeResponder
@@ -216,6 +216,7 @@ muxStart tracer (MuxApplication ptcls) bearer = do
       where
         job run = JobPool.Job (jobAction run)
                               (MiniProtocolException pnum pix pmode)
+                              ((show pix) ++ "." ++ (show pmode))
 
         jobAction run = do
           chan    <- mkChannel

network-mux/src/Network/Mux/JobPool.hs

@@ -28,7 +28,7 @@ data JobPool m a = JobPool {
        completionQueue :: !(TQueue m a)
      }
 
-data Job m a = Job (m a) (SomeException -> a)
+data Job m a = Job (m a) (SomeException -> a) String
 
 withJobPool :: forall m a b.
                (MonadAsync m, MonadThrow m)
@@ -52,12 +52,13 @@ forkJob :: forall m a.
         => JobPool m a
         -> Job     m a
         -> m ()
-forkJob JobPool{jobsVar, completionQueue} (Job action handler) =
+forkJob JobPool{jobsVar, completionQueue} (Job action handler label) =
     mask $ \restore -> do
       jobAsync <- async $ do
+        tid <- myThreadId
+        labelThread tid label
         res <- handleJust notAsyncExceptions (return . handler) $
                  restore action
-        tid <- myThreadId
         atomically $ do
           writeTQueue completionQueue res
           modifyTVar' jobsVar (Map.delete tid)

ouroboros-consensus/src/Ouroboros/Consensus/Storage/ImmutableDB/Impl/State.hs

@@ -155,7 +155,9 @@ getOpenState :: (HasCallStack, IOLike m)
              => ImmutableDBEnv m hash
              -> m (SomePair (HasFS m) (OpenState m hash))
 getOpenState ImmutableDBEnv {..} = do
-    internalState <- readMVar varInternalState
+    -- We use 'readMVarSTM' to read a potentially stale internal state if
+    -- somebody's appending to the ImmutableDB at the same time.
+    internalState <- atomically $ readMVarSTM varInternalState
     case internalState of
        DbClosed         -> throwUserError  ClosedDBError
        DbOpen openState -> return (SomePair hasFS openState)
@@ -244,15 +246,19 @@ withOpenState ImmutableDBEnv { hasFS = hasFS :: HasFS m h, .. } action = do
   where
     HasFS{..} = hasFS
 
+    -- We use 'readMVarSTM' to read a potentially stale internal state if
+    -- somebody's appending to the ImmutableDB at the same time. Reads can
+    -- safely happen concurrently with appends, so this is fine and allows for
+    -- some extra concurrency.
     open :: m (OpenState m hash h)
-    open = readMVar varInternalState >>= \case
+    open = atomically (readMVarSTM varInternalState) >>= \case
       DbOpen ost -> return ost
       DbClosed   -> throwUserError ClosedDBError
 
     -- close doesn't take the state that @open@ returned, because the state
     -- may have been updated by someone else since we got it (remember we're
-    -- using 'readMVar' here, 'takeMVar'). So we need to get the most recent
-    -- state anyway.
+    -- using 'readMVarSTM' here, not 'takeMVar'). So we need to get the most
+    -- recent state anyway.
     close :: ExitCase (Either ImmutableDBError r)
           -> m ()
     close ec = case ec of