Unmask the forked thread in newPipeline

The `newPipeline` function, used as part of `connect`, forks a
listener thread. Before this commit, the thread is forked with
`forkFinally`, where the thread action is run in the same mask as the
parent thread. The thread is then killed by a `killThread` when
closing a connection.

This is typically not a problem if the mask is “masked” (or,
obviously, “unmasked”), because the listener is generally blocked on a
channel at some time or other, and therefore will accept the
asynchronous exception thrown by `killThread`, and terminate.

However, if the mask is “masked uninterruptible”, then the listener
definitely doesn't receive asynchronous exceptions, and the
`killThread` calls hangs, and never returns.

One should probably never call `connect` in a “masked uninterruptible”
action. However, it sounds better to protect the mongoDB library
against the user accidentally doing so than to add a big warning
saying that calling `connect` in “masked uninterruptible” will cause
the program to hang down the line.

Therefore, this commit uses `forkIOWithUnmask`, in order to run the
thread action always in an “unmasked” state. In which case we can be
sure that we can always kill the listener thread regardless of the
client code.

Database/MongoDB/Internal/Protocol.hs

@@ -48,10 +48,10 @@ import Data.Maybe (maybeToList)
 import GHC.Conc (ThreadStatus(..), threadStatus)
 import Control.Monad (forever)
 import Control.Monad.STM (atomically)
-import Control.Concurrent (ThreadId, killThread, forkFinally)
+import Control.Concurrent (ThreadId, killThread, forkIOWithUnmask)
 import Control.Concurrent.STM.TChan (TChan, newTChan, readTChan, writeTChan, isEmptyTChan)
 
-import Control.Exception.Lifted (onException, throwIO, try)
+import Control.Exception.Lifted (SomeException, mask_, onException, throwIO, try)
 
 import qualified Data.ByteString.Lazy as L
 
@@ -103,6 +103,12 @@ data ServerData = ServerData
                 , maxWriteBatchSize   :: Int
                 }
 
+-- | @'forkUnmaskedFinally' action and_then@ behaves the same as @'forkFinally' action and_then@, except that @action@ is run completely unmasked, whereas with 'forkFinally', @action@ is run with the same mask as the parent thread.
+forkUnmaskedFinally :: IO a -> (Either SomeException a -> IO ()) -> IO ThreadId
+forkUnmaskedFinally action and_then =
+  mask_ $ forkIOWithUnmask $ \unmask ->
+    try (unmask action) >>= and_then
+
 -- | Create new Pipeline over given handle. You should 'close' pipeline when finished, which will also close handle. If pipeline is not closed but eventually garbage collected, it will be closed along with handle.
 newPipeline :: ServerData -> Transport -> IO Pipeline
 newPipeline serverData stream = do
@@ -124,9 +130,9 @@ newPipeline serverData stream = do
 
     rec
         let pipe = Pipeline{..}
-        listenThread <- forkFinally (listen pipe) $ \_ -> do
-                                                       putMVar finished ()
-                                                       drainReplies
+        listenThread <- forkUnmaskedFinally (listen pipe) $ \_ -> do
+                                                              putMVar finished ()
+                                                              drainReplies
 
     _ <- mkWeakMVar vStream $ do
         killThread listenThread