4f6066d · typeable/postgresql-simple

5-----------------------------------------------------------------------------

6-- |

7-- Module : Database.PostgreSQL.Simple.Notification

9-- License : BSD3

10-- License : BSD3

10--

11--

11-- Maintainer : leon@melding-monads.com

12-- Maintainer : leon@melding-monads.com

35 , getBackendPID

36 , getBackendPID

36 ) where

37 ) where

37

38

38−import Control.Concurrent

39+import Control.Monad ( join, void )

39−import Control.Monad ( when )

40−import Control.Exception ( throwIO )

40+import Control.Exception ( throwIO, try )

41import qualified Data.ByteString as B

42+import qualified Data.ByteString.Char8 as B8

42import Database.PostgreSQL.Simple.Internal

43import Database.PostgreSQL.Simple.Internal

43import qualified Database.PostgreSQL.LibPQ as PQ

44import qualified Database.PostgreSQL.LibPQ as PQ

44import System.Posix.Types ( CPid )

45import System.Posix.Types ( CPid )

46+import GHC.IO.Exception ( ioe_location )

47+

48+#if defined(mingw32_HOST_OS)

49+import Control.Concurrent ( threadDelay )

50+#elif !MIN_VERSION_base(4,7,0)

51+import Control.Concurrent ( threadWaitRead )

52+#else

53+import GHC.Conc ( atomically )

54+import Control.Concurrent ( threadWaitReadSTM )

55+#endif

45

56

46data Notification = Notification

57data Notification = Notification

47 { notificationPid :: !CPid

58 { notificationPid :: !CPid

59-- 'getNotification' blocks until one arrives.

70-- 'getNotification' blocks until one arrives.

60

71

61getNotification :: Connection -> IO Notification

72getNotification :: Connection -> IO Notification

62−getNotification conn = loop False

73+getNotification conn = join $ withConnection conn fetch

63 where

74 where

64 funcName = "Database.PostgreSQL.Simple.Notification.getNotification"

75 funcName = "Database.PostgreSQL.Simple.Notification.getNotification"

65− loop doConsume = do

76+

66− res <- withConnection conn $ \c -> do

77+ fetch c = do

67− when doConsume (PQ.consumeInput c >> return ())

68− mmsg <- PQ.notifies c

78+ mmsg <- PQ.notifies c

69− case mmsg of

79+ case mmsg of

80+ Just msg -> return (return $! convertNotice msg)

70− Nothing -> do

81+ Nothing -> do

71− mfd <- PQ.socket c

82+ mfd <- PQ.socket c

72− case mfd of

83+ case mfd of

73− Nothing -> throwIO $ fdError funcName

84+ Nothing -> return (throwIO $! fdError funcName)

74− Just fd -> return (Left fd)

75− Just msg -> return (Right msg)

76− -- FIXME? what happens if the connection is closed/reset right here?

77− case res of

78#if defined(mingw32_HOST_OS)

85#if defined(mingw32_HOST_OS)

79− -- threadWaitRead doesn't work for sockets on Windows, so just poll

86+ -- threadWaitRead doesn't work for sockets on Windows, so just

80− -- for input every second (PQconsumeInput is non-blocking).

87+ -- poll for input every second (PQconsumeInput is non-blocking).

81− --

88+ --

82− -- We could call select(), but FFI calls can't be interrupted with

89+ -- We could call select(), but FFI calls can't be interrupted

83− -- async exceptions, whereas threadDelay can.

90+ -- with async exceptions, whereas threadDelay can.

91+ Just _fd -> do

84− Left _fd -> threadDelay 1000000 >> loop True

92+ return (threadDelay 1000000 >> loop)

93+#elif !MIN_VERSION_base(4,7,0)

94+ -- Technically there's a race condition that is usually benign.

95+ -- If the connection is closed or reset after we drop the

96+ -- lock, and then the fd index is reallocated to a new

97+ -- descriptor before we call threadWaitRead, then

98+ -- we could end up waiting on the wrong descriptor.

99+ --

100+ -- Now, if the descriptor becomes readable promptly, then

101+ -- it's no big deal as we'll wake up and notice the change

102+ -- on the next iteration of the loop. But if are very

103+ -- unlucky, then we could end up waiting a long time.

104+ Just fd -> do

105+ return $ do

106+ merr <- try (threadWaitRead fd)

107+ case merr of

108+ Left err -> throwIO $! setLoc err

109+ Right _ -> loop

85#else

110#else

86− Left fd -> threadWaitRead fd >> loop True

111+ -- This case fixes the race condition above. By registering

112+ -- our interest in the descriptor before we drop the lock,

113+ -- there is no opportunity for the descriptor index to be

114+ -- reallocated on us.

115+ --

116+ -- (That is, assuming there isn't concurrently executing

117+ -- code that manipulates the descriptor without holding

118+ -- the lock... but such a major bug is likely to exhibit

119+ -- itself in an at least somewhat more dramatic fashion.)

120+ Just fd -> do

121+ (waitRead, _) <- threadWaitReadSTM fd

122+ return $ do

123+ merr <- try (atomically waitRead)

124+ case merr of

125+ Left err -> throwIO $! setLoc err

126+ Right _ -> loop

87#endif

127#endif

88− Right msg -> return $! convertNotice msg

128+

129+ loop = join $ withConnection conn $ \c -> do

130+ void $ PQ.consumeInput c

131+ fetch c

132+

133+ setLoc :: IOError -> IOError

134+ setLoc err = err { ioe_location = B8.unpack funcName }

135+

89

136

90-- | Non-blocking variant of 'getNotification'. Returns a single notification,

137-- | Non-blocking variant of 'getNotification'. Returns a single notification,

91-- if available. If no notifications are available, returns 'Nothing'.

138-- if available. If no notifications are available, returns 'Nothing'.