thread instrumentation: contention reduction experiment

2 files changed, 33 insertions, 20 deletions

diff --git a/lifted-concurrent/lifted-concurrent.cabal b/lifted-concurrent/lifted-concurrent.cabal
index 26e9df68..78a027db 100644
--- a/lifted-concurrent/lifted-concurrent.cabal
+++ b/lifted-concurrent/lifted-concurrent.cabal
@@ -25,6 +25,7 @@ library
   other-extensions:    FlexibleContexts
   build-depends:
           base
+        , vector
         , containers
         , time
         , lifted-async
diff --git a/lifted-concurrent/src/Control/Concurrent/Lifted/Instrument.hs b/lifted-concurrent/src/Control/Concurrent/Lifted/Instrument.hs
index bd6ee4b8..b4a67583 100644
--- a/lifted-concurrent/src/Control/Concurrent/Lifted/Instrument.hs
+++ b/lifted-concurrent/src/Control/Concurrent/Lifted/Instrument.hs
@@ -27,7 +27,9 @@ import Data.Time()
 import Data.Time.Clock
 import DPut
 import DebugTag
-
+import qualified Data.Vector as V
+import Data.Vector (Vector)
+import Data.Char
 
 data PerThread = PerThread
  { lbl :: String
@@ -35,15 +37,25 @@ data PerThread = PerThread
  }
  deriving (Eq,Ord,Show)
 
+{-# NOINLINE globalMVarArray #-}
+globalMVarArray :: Vector (MVar (Map.Map ThreadId PerThread))
+globalMVarArray = unsafePerformIO (sequence (V.replicate 256 (newMVar Map.empty)))
+
 data GlobalState = GlobalState
- { threads :: !(Map.Map ThreadId PerThread)
- , reportException :: String -> IO ()
+ { reportException :: String -> IO ()
  }
 
+hashThreadId :: ThreadId -> Int
+hashThreadId tid = hash 0 (dropThreadIdAndSPace (show tid))
+    where
+        dropThreadIdAndSPace ('T':'h':'r':'e':'a':'d':'I':'d':' ':xs) = xs
+        dropThreadIdAndSPace xs = xs
+        hash n xs = read xs `mod` 256
+        
+
 globals :: MVar GlobalState
 globals = unsafePerformIO $ newMVar $ GlobalState
-    { threads = Map.empty
-    , reportException = dput XMisc
+    { reportException = dput XMisc
     }
 {-# NOINLINE globals #-}
 
@@ -92,24 +104,25 @@ instrumented rawFork action = do
     t <- mask_ $ rawFork $ \unmask -> do
             tid <- myThreadId
             let scrapIt = do takeMVar mvar
-                             modifyThreads $! Map.delete tid
+                             modifyThreads tid $! Map.delete tid
             io <- catch (unmask action >> return scrapIt) $ \e -> case fromException e of
                     Just ThreadKilled -> return scrapIt
                     Nothing           -> liftBase $ do
-                        g <- takeMVar globals
+                        g <- readMVar globals
+                        mp <- readMVar (globalMVarArray V.! hashThreadId tid)
                         let l = concat [ show e
                                        , " ("
-                                       , maybe "" lbl $ Map.lookup tid (threads g)
+                                       , maybe "" lbl $ Map.lookup tid mp
                                        , ")"
                                        ]
                         reportException g l
                         let l = concat [ show e
                                        , " ("
-                                       , maybe "" lbl $ Map.lookup tid (threads g)
+                                       , maybe "" lbl $ Map.lookup tid mp
                                        , ")"
                                        ]
                         foldr seq (return ()) l
-                        putMVar globals $! g { threads = Map.insert tid (PerThread l tm) $ threads g }
+                        modifyThreads tid $! Map.insert tid (PerThread l tm)
                         return $ return () -- Remove the thread only if it terminated normally or was killed.
             io -- scrap record on normal termination
     liftBase $ labelThread_ t (defaultLabel callStack) tm
@@ -122,25 +135,24 @@ labelThread_ tid s tm = do
     GHC.labelThread tid s
     let updateIt (Just pt) = Just $ pt { lbl = s }
         updateIt Nothing   = Just $ PerThread s tm
-    modifyThreads $! Map.alter updateIt tid
+    modifyThreads tid $! Map.alter updateIt tid
 
 labelThread :: ThreadId -> String -> IO ()
 labelThread tid s = do
     foldr seq (return ()) s
     GHC.labelThread tid s
-    modifyThreads $! Map.adjust (\pt -> pt { lbl = s }) tid
+    modifyThreads tid $! Map.adjust (\pt -> pt { lbl = s }) tid
 {-# INLINE labelThread #-}
 
 threadsInformation :: IO [(ThreadId,PerThread)]
 threadsInformation = do
-    m <- threads <$> readMVar globals
-    return $ Map.toList m
+    ms <- mapM readMVar (V.toList globalMVarArray)
+    return $ Prelude.concatMap Map.toList ms
 
 
-modifyThreads :: MonadBaseControl IO m =>
+modifyThreads :: MonadBaseControl IO m => ThreadId ->
                     (Map.Map ThreadId PerThread -> Map.Map ThreadId PerThread) -> m ()
-modifyThreads f = do
-    let f' st = st { threads = f (threads st) }
-    bracket (takeMVar globals)
-            (\g -> putMVar globals $! f' g)
-            (\g -> return ())
+modifyThreads tid f = do
+    let tidHash = hashThreadId tid
+    let mvar = globalMVarArray V.! tidHash
+    modifyMVarMasked_ mvar (return . f)