Fix concurrency bug in limitedMapConcurrently (#53)

Also simplifies implementation using lifted `bracket_` and async
routines, instead of lifted control operations like `liftBaseOp_`

Adds a test.

examples/FileUploader.hs

@@ -57,7 +57,7 @@ main :: IO ()
 main = do
   let bucket = "my-bucket"
 
-  -- Parse command line argument, namely --filename.
+  -- Parse command line argument
   filepath <- execParser cmdParser
   let object = pack $ takeBaseName filepath
 

minio-hs.cabal

@@ -106,6 +106,7 @@ test-suite minio-hs-live-server-test
                      , Network.Minio.S3API
                      , Network.Minio.Sign.V4
                      , Network.Minio.Utils
+                     , Network.Minio.Utils.Test
                      , Network.Minio.API.Test
                      , Network.Minio.XmlGenerator
                      , Network.Minio.XmlGenerator.Test
@@ -215,6 +216,7 @@ test-suite minio-hs-test
                      , Network.Minio.S3API
                      , Network.Minio.Sign.V4
                      , Network.Minio.Utils
+                     , Network.Minio.Utils.Test
                      , Network.Minio.API.Test
                      , Network.Minio.XmlGenerator
                      , Network.Minio.XmlGenerator.Test

src/Network/Minio/Utils.hs

@@ -17,10 +17,9 @@
 module Network.Minio.Utils where
 
 import qualified Control.Concurrent.Async.Lifted as A
-import qualified Control.Concurrent.QSem as Q
+import qualified Control.Concurrent.QSem.Lifted as Q
 import qualified Control.Exception.Lifted as ExL
 import qualified Control.Monad.Catch as MC
-import           Control.Monad.Trans.Control (liftBaseOp_, StM)
 import qualified Control.Monad.Trans.Resource as R
 
 import qualified Data.ByteString as B
@@ -162,25 +161,18 @@ http req mgr = do
     tryHttpEx = ExL.try
     contentTypeMay resp = lookupHeader Hdr.hContentType $ NC.responseHeaders resp
 
--- like mapConcurrently but with a limited number of concurrent
--- threads.
-limitedMapConcurrently :: forall t a (m :: * -> *) b.
-                          (MonadIO m, R.MonadBaseControl IO m,
-                           StM m a ~ StM m b)
-                       => Int -> (t -> m a) -> [t] -> m [b]
+-- Similar to mapConcurrently but limits the number of threads that
+-- can run using a quantity semaphore.
+limitedMapConcurrently :: (MonadIO m, R.MonadBaseControl IO m)
+                       => Int -> (t -> m a) -> [t] -> m [a]
 limitedMapConcurrently count act args = do
   qSem <- liftIO $ Q.newQSem count
-  threads <- workOn qSem args
+  threads <- mapM (A.async . wThread qSem) args
   mapM A.wait threads
   where
-    workOn _ [] = return []
-    workOn qs (a:as) = liftBaseOp_
-      (bracket_ (Q.waitQSem qs) (Q.signalQSem qs)) $
-      do
-        thread <- A.async $ act a
-        others <- workOn qs as
-        return (thread : others)
-
+    -- grab 1 unit from semaphore, run action and release it
+    wThread qs arg =
+      ExL.bracket_ (Q.waitQSem qs) (Q.signalQSem qs) $ act arg
 
 -- helper function to 'drop' empty optional parameter.
 mkQuery :: Text -> Maybe Text -> Maybe (Text, Text)

test/Network/Minio/Utils/Test.hs

@@ -0,0 +1,47 @@
+--
+-- Minio Haskell SDK, (C) 2017 Minio, Inc.
+--
+-- Licensed under the Apache License, Version 2.0 (the "License");
+-- you may not use this file except in compliance with the License.
+-- You may obtain a copy of the License at
+--
+--     http://www.apache.org/licenses/LICENSE-2.0
+--
+-- Unless required by applicable law or agreed to in writing, software
+-- distributed under the License is distributed on an "AS IS" BASIS,
+-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+-- See the License for the specific language governing permissions and
+-- limitations under the License.
+--
+
+module Network.Minio.Utils.Test
+  (
+    limitedMapConcurrentlyTests
+  ) where
+
+import Test.Tasty
+import Test.Tasty.HUnit
+
+import Lib.Prelude
+
+import Network.Minio.Utils
+
+limitedMapConcurrentlyTests :: TestTree
+limitedMapConcurrentlyTests = testGroup "limitedMapConcurrently Tests"
+  [ testCase "Test with various thread counts" testLMC
+  ]
+
+testLMC :: Assertion
+testLMC = do
+  let maxNum = 50
+  -- test with thread count of 1 to 2*maxNum
+  forM_ [1..(2*maxNum)] $ \threads -> do
+    res <- limitedMapConcurrently threads compute [1..maxNum]
+    sum res @?= overallResultCheck maxNum
+  where
+    -- simple function to run in each thread
+    compute :: Int -> IO Int
+    compute n = return $ sum [1..n]
+
+    -- function to check overall result
+    overallResultCheck n = sum $ map (\t -> (t * (t+1)) `div` 2) [1..n]

test/Spec.hs

@@ -23,6 +23,7 @@ import           Lib.Prelude
 
 import           Network.Minio.API.Test
 import           Network.Minio.PutObject
+import           Network.Minio.Utils.Test
 import           Network.Minio.XmlGenerator.Test
 import           Network.Minio.XmlParser.Test
 
@@ -113,4 +114,5 @@ qcProps = testGroup "(checked by QuickCheck)"
 unitTests :: TestTree
 unitTests = testGroup "Unit tests" [xmlGeneratorTests, xmlParserTests,
                                     bucketNameValidityTests,
-                                    objectNameValidityTests]
+                                    objectNameValidityTests,
+                                    limitedMapConcurrentlyTests]