第二十四章：并發(fā)和多核編程

線程的不確定性?

GHC 的運行時組件并不按特定順序執(zhí)行多個線程。所以，上面的例子中，文件 xyzzy 的創(chuàng)建時間在初始線程檢查其是否存在之前或之后都有可能。如果刪除 xyzzy 并且再執(zhí)行一次，我們可能得到完全相反的結(jié)果。

隱藏延遲?

假設我們要將一個大文件壓縮并寫入磁盤，但是希望快速處理用戶輸入以使他們感覺程序是立即響應的。如果使用 forkIO 來開啟一個單獨的線程去寫文件，這樣就可以同時做這兩件事。

-- file: ch24/Compressor.hs
import Control.Concurrent (forkIO)
import Control.Exception (handle)
import Control.Monad (forever)
import qualified Data.ByteString.Lazy as L
import System.Console.Readline (readline)

-- http://hackage.haskell.org/ 上的 zlib 包提供了壓縮功能
import Codec.Compression.GZip (compress)

main = do
    maybeLine <- readline "Enter a file to compress> "
    case maybeLine of
      Nothing -> return ()      -- 用戶輸入了 EOF
      Just "" -> return ()      -- 不輸入名字按 “想要退出” 處理
      Just name -> do
           handle
             (print :: (SomeException->IO ()))
             $ do
                 content <- L.readFile name
                 forkIO (compressFile name content)
                 return ()
           main
  where compressFile path = L.writeFile (path ++ ".gz") . compress

因為使用了惰性的 ByteString I/O ，主線程中做僅僅是打開文件。真正讀取文件內(nèi)容發(fā)生在子線程中。

當用戶輸入的文件名并不存在時將發(fā)生異常， handle (print :: (SomeException-> IO ())) 是一個低成本的打印錯誤信息的方式。

安全的修改 MVar?

forkManaged 中使用的 modifyMVar 函數(shù)很實用：它將 takeMVar 和 putMVar 安全的組合在一起。

ghci> :t modifyMVar
modifyMVar :: MVar a -> (a -> IO (a, b)) -> IO b

其從一個 MVar 中取出一個值，并傳入一個函數(shù)。這個函數(shù)生成一個新的值，且返回一個結(jié)果。如果函數(shù)拋出一個異常， modifyMVar 會將初始值重新放回 MVar ，否則其會寫入新值。它還會返回另一個返回值。

使用 modifyMVar 而非手動使用 takeMVar 和 putMVar 管理 MVar ， 可以避免兩類并發(fā)場景下的問題。

• 忘記將一個值放回 MVar 。有的線程會一直等待 MVar 中被放回一個值，如果一致沒有等到，就將導致死鎖。
• 沒有考慮可能出現(xiàn)的異常，擾亂了某端代碼的控制流。這可能導致一個本應執(zhí)行的 putMVar 沒有執(zhí)行，進而導致死鎖。

因為這些美妙的安全特性，盡可能的使用 modifyMVar 是明智的選擇。

安全資源管理：一個相對簡單的好主意。?

modifyMVar 遵循的模式適用很多場景。下面是這些模式：

1. 獲得一份資源。
2. 將資源傳入一個將處理它函數(shù)。
3. 始終釋放資源，即使函數(shù)拋出異常。如果發(fā)生異常，重新拋出異常，以便使其被程序捕獲。

除了安全性，這個方法還有其他好處：可以是代碼更簡短且容易理解。正如前面的 forkManaged ， Hakell 的簡潔語法和匿名函數(shù)使得這種風格的代碼看起來一點都不刺眼。

下面是 modifyMVar 的定義，從中可以了解這個模式的細節(jié)：

-- file: ch24/ModifyMVar.hs
import Control.Concurrent (MVar, putMVar, takeMVar)
import Control.Exception (block, catch, throw, unblock)
import Prelude hiding (catch) -- use Control.Exception's version

modifyMVar :: MVar a -> (a -> IO (a,b)) -> IO b
modifyMVar m io =
  block $ do
    a <- takeMVar m
    (b,r) <- unblock (io a) `catch` \e ->
             putMVar m a >> throw e
    putMVar m b
    return r

這種模式很容易用于你的特定需求，無論是處理網(wǎng)絡連接，數(shù)據(jù)庫句柄，或者被 C 庫函數(shù)管理的數(shù)據(jù)。

查看線程狀態(tài)?

我們編寫的 getStatus 函數(shù)用于獲取某個線程的當前狀態(tài)。若某線程已經(jīng)不被管理（或者未被管理），它返回 Nothing 。

-- file: ch24/NiceFork.hs
getStatus (Mgr mgr) tid =
  modifyMVar mgr $ \m ->
    case M.lookup tid m of
      Nothing -> return (m, Nothing)
      Just st -> tryTakeMVar st >>= \mst -> case mst of
                   Nothing -> return (m, Just Running)
                   Just sth -> return (M.delete tid m, Just sth)

若線程仍在運行，它返回 Just Running 。 否則，它指出將線程為何被終止，并停止管理這個線程。

若 tryTakeMVar 函數(shù)發(fā)現(xiàn) MVar 為空，它將立即返回 Nothing 而非阻塞等待。

ghci> :t tryTakeMVar
tryTakeMVar :: MVar a -> IO (Maybe a)

否則，它將從 MVar 取到一個值。

waitFor 函數(shù)的行為較簡單，其會阻塞等待給定線程終止，而非立即返回。

-- file: ch24/NiceFork.hs
waitFor (Mgr mgr) tid = do
  maybeDone <- modifyMVar mgr $ \m ->
    return $ case M.updateLookupWithKey (\_ _ -> Nothing) tid m of
      (Nothing, _) -> (m, Nothing)
      (done, m') -> (m', done)
  case maybeDone of
    Nothing -> return Nothing
    Just st -> Just `fmap` takeMVar st

首先讀取保存線程狀態(tài)的 MVar ，若其存在。 Map 類型的 updateLookupWithKey 函數(shù)很有用：它將查找某個值與更新或移除組合起來。

ghci> :m +Data.Map
ghci> :t updateLookupWithKey
updateLookupWithKey :: (Ord k) =>
                       (k -> a -> Maybe a) -> k -> Map k a -> (Maybe a, Map k a)

在此處，我們希望若保存線程狀態(tài)的 MVar 存在，則將其從 Map 中移除，這樣線線程管理器將不在管理這個線程。若從其中取到了值，則從中取出線程的退出狀態(tài)，并將其返回。

我們的最后一個實用函數(shù)簡單的等待所有當前被管理的線程完成，且忽略他們的退出狀態(tài)。

-- file: ch24/NiceFork.hs
waitAll (Mgr mgr) = modifyMVar mgr elems >>= mapM_ takeMVar
    where elems m = return (M.empty, M.elems m)

編寫更緊湊的代碼?

我們在上面定義的 waitFor 函數(shù)有點不完善，因為或多或少執(zhí)行了重復的模式分析：在 modifyMVar 內(nèi)部的回調(diào)函數(shù)，以及處理其返回值時。

當然，我們可以用一個函數(shù)消除這種重復。這是 Control.Monad 模塊中的 join 函數(shù)。

ghci> :m +Control.Monad
ghci> :t join
join :: (Monad m) => m (m a) -> m a

這是個有趣的主意：可以創(chuàng)建一個 monadic 函數(shù)或純代碼中的 action ，然后一直帶著它直到最終某處有個 monad 可以使用它。一旦我們了解這種寫法適用的場景，就可以更靈活的編寫代碼。

-- file: ch24/NiceFork.hs
waitFor2 (Mgr mgr) tid =
  join . modifyMVar mgr $ \m ->
    return $ case M.updateLookupWithKey (\_ _ -> Nothing) tid m of
      (Nothing, _) -> (m, return Nothing)
      (Just st, m') -> (m', Just `fmap` takeMVar st)

MVar 和 Chan 是非嚴格的?

正如大多數(shù) Haskell 容器類型， MVar 和 Char 都是非嚴格的：從不對其內(nèi)容求值。我們提到它，并非因為這是一個問題，而是因為這通常是一個盲點：人們傾向于假設這些類型是嚴格的，這大概是因為它們被用在 IO monad 中。

正如其他容器類型，誤認為 MVar 和 Chan 是嚴格的會導致空間和性能的泄漏?？紤]一下這個很可能發(fā)生的情況：

我們分離一個線程以在另一個核上執(zhí)行一些開銷較大的計算

-- file: ch24/Expensive.hs
import Control.Concurrent

notQuiteRight = do
  mv <- newEmptyMVar
  forkIO $ expensiveComputation_stricter mv
  someOtherActivity
  result <- takeMVar mv
  print result

它看上去做了一些事情并將結(jié)果存入 MVar 。

-- file: ch24/Expensive.hs
expensiveComputation mv = do
  let a = "this is "
      b = "not really "
      c = "all that expensive"
  putMVar mv (a ++ b ++ c)

當我們在父線程中從 MVar 獲取結(jié)果并嘗試用它做些事情時，我們的線程開始瘋狂的計算，因為我們從未強制指定在其他線程中的計算真正發(fā)生。

照舊，一旦我們知道了有個潛在問題，解決方案就很簡單：未分離的線程添加嚴格性，以確保計算確實發(fā)生。這個嚴格性最好加在一個位置，以避免我們忘記添加過它。

-- file: ch24/ModifyMVarStrict.hs
{-# LANGUAGE BangPatterns #-}

import Control.Concurrent (MVar, putMVar, takeMVar)
import Control.Exception (block, catch, throw, unblock)
import Prelude hiding (catch) -- 使用 Control.Exception's 中的 catch 而非 Prelude 中的。

modifyMVar_strict :: MVar a -> (a -> IO a) -> IO ()
modifyMVar_strict m io = block $ do
  a <- takeMVar m
  !b <- unblock (io a) `catch` \e ->
        putMVar m a >> throw e
  putMVar m b

上面代碼中的 ! 模式用起來很簡單，但是并不總是足以確保我們的數(shù)據(jù)已經(jīng)被求值。更完整的方法，請查看下面的段落“從求值中分離算法”。

Chan 是無邊界的?

因為 writeChan 總是立即成功，所以在使用 Chan 時有潛在風險。若對某個 Chan 的寫入多于其讀取， Chan 將用不檢查的方法增長：對未讀消息的讀取將遠遠落后于其增長。

死鎖?

死鎖的情況下，兩個或多個線程永遠卡在爭搶共享資源的訪問權上。制造多線程程序死鎖的一個經(jīng)典方法是不按順序加鎖。這種類型的 bug 很常見，它有個名字：鎖順序倒置。 Haskell 沒有提供鎖， 但 MVar 類型可能會有順序倒置問題。這有一個簡單例子：

-- file: ch24/LockHierarchy.hs
import Control.Concurrent

nestedModification outer inner = do
    modifyMVar_ outer $ \x -> do
        yield -- 強制當前線程讓出 CPU
        modifyMVar_ inner $ \y -> return (y + 1)
        return (x + 1)
    putStrLn "done"

main = do
    a <- newMVar 1
    b <- newMVar 2
    forkIO $ nestedModification a b
    forkIO $ nestedModification b a

在 ghci 中運行這段程序，它通常會（但不總是）不打印任何信息，表明兩個線程已經(jīng)卡住了。

容易看出 nestedModification 函數(shù)的問題。在第一個線程中，我們先取出 MVar a ，接著取出 b 。在第二個線程中，先取出 b 然后取出 a ，若第一個線程成功取出了 a 然后要取出 b ，這是兩個線程都會阻塞：每個線程都嘗試獲取一個 MVar ，而這個 MVar 已經(jīng)被另一個線程取空了，所以二者都不能完成整個流程。

無論何種語言，通常解決倒序問題的方法是申請資源時一直遵循一致的順序。因為這需要人工遵循編碼規(guī)范，在實踐中很容易遺忘。

更麻煩的是，這種倒序問題在實際代碼中很難被發(fā)現(xiàn)。獲取 MVar 的動作經(jīng)?？缭讲煌募械牟煌瘮?shù)，這使得通過觀察源碼檢查時更加棘手。更糟糕的是，這類問題通常是間歇性的，這使得它們難于重現(xiàn)，更不要說隔離和修復了。

饑餓?

并發(fā)軟件通?？赡軙е吗囸I問題，某個線程霸占了共享資源，阻止其他線程使用。很容易想象這是如何發(fā)生的：一個線程調(diào)用 modifyMVar 執(zhí)行一個 100 毫秒的代碼段，稍后另外一個線程對同一個 MVar 調(diào)用 modifyMVar 執(zhí)行一個 1 毫秒的代碼段。第二個線程在第一個線程完成前將無法執(zhí)行。

MVar 類型的非嚴格性質(zhì)使會導致或惡化饑餓的問題。若我們將一個求值開銷很大的 thunk 寫入一個 MVar ，在一個看上去開銷較小的線程中取出并求值，這個線程的執(zhí)行開銷馬上會變大。所以我們在 “MVar 和 Chan 是非嚴格的” 一章中特地給出了一些建議。

沒希望了嗎？?

幸運的是，我們已經(jīng)提及的并發(fā) API 并不是故事的全部。最近加入 Haskell 中的一個設施，軟件事務內(nèi)存，使用起來更加容易和安全。我們將在第 28 章，軟件事務內(nèi)存中介紹。

運行時選項?

運行程序時可以向 GHC 的運行時系統(tǒng)傳遞命令行參數(shù)。在將控制權交給我們的代碼前，運行時掃描程序的參數(shù)，看是否有命令行選項 +RTS 。其后跟隨的所有選項都被運行時解釋，直到特殊的選項 -RTS ，這些選項都是提供給運行時系統(tǒng)的，不為我們的程序。運行時會對我們的代碼隱藏所有這些選項。當我們使用 System.Environment 模塊的 getArgsfunction 來獲得我們的命令行參數(shù)是，我們不會在其中獲得運行時選項。

threaded 運行時接受參數(shù) -N [55] 。 其接受一個參數(shù)，指定了 GHC 的運行時系統(tǒng)將使用的核數(shù)。這個選項對輸入很挑剔： -N 和參數(shù)之間必須沒有空格。 -N4 可被接受， -N 4 則不被接受。

找出 Haskell 可以使用多少核?

GHC.Conc 模塊輸出一個變量， numCapabilities ，它會告訴我們運行時系統(tǒng)被 -NRTS 選項指定了多少核。

-- file: ch24/NumCapabilities.hs
import GHC.Conc (numCapabilities)
import System.Environment (getArgs)

main = do
    args <- getArgs
    putStrLn $ "command line arguments: " ++ show args
    putStrLn $ "number of cores: " ++ show numCapabilitie

若編譯上面的程序，我們可以看到運行時系統(tǒng)的選項對于程序來說是不可見的，但是它可以看其運行在多少核上。

$ ghc -c NumCapabilities.hs
$ ghc -threaded -o NumCapabilities NumCapabilities.o $ ./NumCapabilities +RTS -N4 -RTS foo
command line arguments: ["foo"]
number of cores: 4

選擇正確的運行時?

選擇正確的運行時需要花點心思。 threaded 運行時可以使用多核，但是也有相應的代價：線程間共享數(shù)據(jù)的成本比 non-threaded 運行時更大。

目前為止， GHC 的 6.8.3 版本使用的垃圾收集器是單線程的：它執(zhí)行時暫停其他所有線程，而且它是在單核上執(zhí)行。這限制了我們在使用多核的時候希望看到的性能改進[56]_。

很多真實世界中的并發(fā)程序中，一個單獨的線程多數(shù)時間實在等待一個網(wǎng)絡請求或響應。這些情況下，若以一個單獨的 Haskell 程序為數(shù)萬并發(fā)客戶端提供服務，使用低開銷的 non-threaded 運行時很可能是合適的。例如，與其用 4 個核跑 threaded 運行時的單個服務器程序，可能同時跑 4 個 non-threaded 運行時的相同服務器程序性能更好。

我們的目的并不是阻止你使用 threaded 運行時。相對于 non-threaded 運行時它并沒有特別大的開銷：相對于其他編程語言，線程依舊驚人的輕量。我們僅是希望說明 threaded 運行時并不是在所有場景都是最佳選擇。