Is Data.Pool not thread-safe or have I done something silly?
I wrote some code with a Data.Pool that has 1 stripe, 1 max resource, and then basically did `Async.replicateConcurrently_ . withResource $ \res -> f res`. I expect withResource to block in each of those threads until the first thread spawned is done with that Resource and releases. To be clear, I get output like: ```shell /tmp $ stack exec --resolver lts-17.5 --package resource-pool --package stm --package hslogger --package time --package concurrent-output -- ghc -O2 -threaded -fwarn-unused-imports testpool.hs && ./testpool acquire 0 - .81428 acquire 1 - .81431 acquire 2 - .81438 acquire 3 - .81440 acquire 4 - .81448 ThreadId 8: processing 1 - .82460 ThreadId 17: processing 4 - .82461 ThreadId 11: processing 2 - .82464 ThreadId 14: processing 3 - .82464 ThreadId 5: processing 0 - .82465 anything else? release 4 - .14427 release 3 - .14430 release 2 - .14431 release 1 - .14431 release 0 - .14432 anything else? anything else? ``` I expect output like: ```shell /tmp $ stack exec --resolver lts-17.5 --package resource-pool --package stm --package hslogger --package time --package concurrent-output -- ghc -O2 -threaded -fwarn-unused-imports testpool.hs && ./testpool [1 of 1] Compiling Main ( testpool.hs, testpool.o ) Linking testpool ... acquire 0 ThreadId 5: processing 0 - .01129 release 0 acquire 1 -- I thought this would have blocked until 0 was released ThreadId 8: processing 1 - .01120 release 1 acquire 2 ThreadId 11: processing 2 - .01123 release 2 acquire 3 ThreadId 14: processing 3 - .01129 release 3 acquire 4 ThreadId 17: processing 4 - .01129 release 4 anything else? anything else? anything else? ``` Here is the code: ``` #!/usr/bin/env stack -- stack script --resolver lts-17.5 --package resource-pool --package stm --package hslogger --package time --package concurrent-output import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM import Data.Pool import Data.Time import System.Console.Concurrent import System.Clock main :: IO () main = do counter <- newTVarIO 0 let acquire = do k <- atomically $ do k <- readTVar counter writeTVar counter (k + 1) return k now <- getTime Monotonic outputConcurrent ("acquire " ++ show k ++ " - "++ show now ++ "\n") return k release k = do now <- getTime Monotonic outputConcurrent ("acquire " ++ show k ++ " - "++ show now ++ "\n") withConcurrentOutput $ do -- create a pool that only allows 1 resource pool <- createPool acquire release 1 500 1 replicateConcurrently_ 5 $ do useResourceFor (seconds 10) pool -- Why do you need these to see the release messages? putStrLn "anything else?" >> threadDelay (seconds 5) putStrLn "anything else?" >> threadDelay (seconds 5) putStrLn "anything else?" >> threadDelay (seconds 5) useResourceFor waitSeconds pool = withResource pool $ \i -> do threadDelay waitSeconds tid <- myThreadId now <- getTime Monotonic outputConcurrent $ show tid <> ": " <> " processing " <> show i <> " - " <> show now <> "\n" seconds = (* 1000000) ``` I'm hoping someone could help explain what I did wrong or confirm it's a bug. Thanks, Cody
Hi Cody, I think part of your confusion may come from the naming of your functions. Your 'acquire' function is passed as first argument to createPool, and thus is a 'create' function rather than an acquire function. As your pool only ever has a single resource, I would expect that it would be called only once, because once a resource has been created it can be reused by all threads. With stackage lts-16.15 I get exactly the behaviour I would expect on my machine: acquire 0 - TimeSpec {sec = 221871, nsec = 95964800} ThreadId 6: processing 0 - TimeSpec {sec = 221881, nsec = 97250800} ThreadId 8: processing 0 - TimeSpec {sec = 221891, nsec = 98211800} ThreadId 10: processing 0 - TimeSpec {sec = 221901, nsec = 99347300} ThreadId 12: processing 0 - TimeSpec {sec = 221911, nsec = 100904500} ThreadId 14: processing 0 - TimeSpec {sec = 221921, nsec = 102292000} anything else? acquire 0 - TimeSpec {sec = 221921, nsec = 442620100} anything else? anything else? First a resource is created, then every 10 seconds a thread completed, and finally the resource is freed (your logging in release also prints acquire, but the second instance is from release). I can't run GHC 8.10 yet so not sure what happens there. Regards, Jeroen Bransen Op 13-3-2021 om 02:09 schreef Cody Gman:
I wrote some code with a Data.Pool that has 1 stripe, 1 max resource, and then basically did `Async.replicateConcurrently_ . withResource $ \res -> f res`.
I expect withResource to block in each of those threads until the first thread spawned is done with that Resource and releases.
To be clear, I get output like:
```shell /tmp $ stack exec --resolver lts-17.5 --package resource-pool --package stm --package hslogger --package time --package concurrent-output -- ghc -O2 -threaded -fwarn-unused-imports testpool.hs && ./testpool acquire 0 - .81428 acquire 1 - .81431 acquire 2 - .81438 acquire 3 - .81440 acquire 4 - .81448 ThreadId 8: processing 1 - .82460 ThreadId 17: processing 4 - .82461 ThreadId 11: processing 2 - .82464 ThreadId 14: processing 3 - .82464 ThreadId 5: processing 0 - .82465 anything else? release 4 - .14427 release 3 - .14430 release 2 - .14431 release 1 - .14431 release 0 - .14432 anything else? anything else?
```
I expect output like:
```shell /tmp $ stack exec --resolver lts-17.5 --package resource-pool --package stm --package hslogger --package time --package concurrent-output -- ghc -O2 -threaded -fwarn-unused-imports testpool.hs && ./testpool [1 of 1] Compiling Main ( testpool.hs, testpool.o ) Linking testpool ... acquire 0 ThreadId 5: processing 0 - .01129 release 0 acquire 1 -- I thought this would have blocked until 0 was released ThreadId 8: processing 1 - .01120 release 1 acquire 2 ThreadId 11: processing 2 - .01123 release 2 acquire 3 ThreadId 14: processing 3 - .01129 release 3 acquire 4 ThreadId 17: processing 4 - .01129 release 4 anything else? anything else? anything else? ```
Here is the code:
``` #!/usr/bin/env stack -- stack script --resolver lts-17.5 --package resource-pool --package stm --package hslogger --package time --package concurrent-output
import Control.Concurrent import Control.Concurrent.Async import Control.Concurrent.STM import Data.Pool import Data.Time import System.Console.Concurrent import System.Clock
main :: IO () main = do counter <- newTVarIO 0 let acquire = do k <- atomically $ do k <- readTVar counter writeTVar counter (k + 1) return k now <- getTime Monotonic outputConcurrent ("acquire " ++ show k ++ " - "++ show now ++ "\n") return k release k = do now <- getTime Monotonic outputConcurrent ("acquire " ++ show k ++ " - "++ show now ++ "\n")
withConcurrentOutput $ do -- create a pool that only allows 1 resource pool <- createPool acquire release 1 500 1 replicateConcurrently_ 5 $ do useResourceFor (seconds 10) pool
-- Why do you need these to see the release messages? putStrLn "anything else?" >> threadDelay (seconds 5) putStrLn "anything else?" >> threadDelay (seconds 5) putStrLn "anything else?" >> threadDelay (seconds 5)
useResourceFor waitSeconds pool = withResource pool $ \i -> do threadDelay waitSeconds tid <- myThreadId now <- getTime Monotonic outputConcurrent $ show tid <> ": " <> " processing " <> show i <> " - " <> show now <> "\n"
seconds = (* 1000000) ```
I'm hoping someone could help explain what I did wrong or confirm it's a bug.
Thanks,
Cody _______________________________________________ Haskell-Cafe mailing list To (un)subscribe, modify options or view archives go to: http://mail.haskell.org/cgi-bin/mailman/listinfo/haskell-cafe Only members subscribed via the mailman list are allowed to post.
-- Jeroen Bransen Back-end Developer at Chordify -- https://chordify.net
participants (2)
-
Cody Gman -
Jeroen Bransen