How to atomically modify a value within a mutable array?

user_16533276 0
Viewed 180

I have this mutable array in an ST monad. And I have this loop function.

 runST $ do
    myarray <- newMArray (Sz 10) 0
    loopM_ 0 (<10) (+1) (\j ->
      loopM_ 0 (<10) (+1) (\i ->
        when (mytruthcheck j i)
             (modifyM_ myarray (pure . (+1)) ((funcofji j i) :: Int)
     )))

I want to use forkST_ to run the outer loop in parallel like this.

runST $ do
   myarray <- newMArray (Sz 10) 0
   loopM_ 0 (<10) (+1) (\j ->
     void (forkST_ (loopM_ 0 (<10) (+1) (\i ->
       when (mytruthcheck j i)
            (Data.Massiv.Array.Mutable.modifyM_
                 myarray (pure . (+1)) ((funcofji j i) :: Int)
    ))))

But I’m guessing this will cause thread collisions but I don’t really know although I do know it is possible that funcofji can output the same value for different values of j and therefore the loop can modify the same index of myarray for different j s. Is there a way to ensure this is done atomically or is that already the case?

Btw here’s the loopM_ function

loopM_ :: Monad m => Int -> (Int -> Bool) -> (Int -> Int) -> (Int -> m a) -> m ()
loopM_ !init' condition increment f = go init'
  where
    go !step
      | condition step = f step >> go (increment step)
      | otherwise = pure ()
massiv parallel-processing concurrency haskell
3 Answers

As a result of the discussion in the comments, I wrote a prototype of how this could be, maybe it will be useful (I didn't try to compile so there are maybe some type/syntax errors).

runST $ do
    let arrsz = 10 :: Int -- depends on codomain of funcofji
    let ncaps = 8 :: Int64 -- see also getNumCapabilities
    let outerLoopSize = 10^5  :: Int64
    let innerLoopSize = 10^12 :: Int64
    let chunksz = ceiling $ fromIntegral outerLoopSize / fromIntegral ncaps
    
    sync <- newEmptyMVar
    forM_ [0 .. ncaps - 1] $ \k -> forkST_ $ do
        localArr <- newMArray (Sz arrsz) 0
        forM_ [k * chunksz .. min outerLoopSize ((k + 1) * chunksz) - 1] $ \j -> do
            forM_ [0 .. innerLoopSize - 1] $ \i -> do
                when (mytruthcheck j i) $
                    modifyM_ localArr (pure . (+1)) $ funcofji j i
        putMVar sync localArr

    resultArr <- takeMVar sync
    replicateM_ (ncaps - 1) $ do
        localArr <- takeMVar sync
        forM_ [0 .. arrsz - 1] $ do \ix ->
            elm <- readM localArr ix
            modifyM_ resultArr (pure . (+elm)) ix

    ...
user_71751170

As it was mentioned in the comments, atomic modification is only useful for concurrency which doesn't look like what is needed here. What you need is parallelism., which is available in massiv for Ints: atomicAddIntArray

There is also a builtin way to do parallelism in massiv very efficiently, so there is definitely no need to reinvent the wheel:

  createArray_ Par (Sz 10) $ \scheduler myarray ->
    loopM_ 0 (<10) (+1) $ \j ->
      loopM_ 0 (<10) (+1) $ \i ->
        when (mytruthcheck j i) $ 
          scheduleWork_ scheduler $ 
            void $ atomicAddIntArray myarray ((funcofji j i) :: Int) 1

Also don't lie to yourself, ST (state thread) was not build for multi-threading, use IO instead. However, if you can guarantee, that despite multi-threaded setup, the outcome that is produced is still deterministic in the end, then it is ok to use unsafePerformIO.

Edit

I just noticed this comment:

The loop size of j is close to 100,000 and the loop size of i is close to 1 Billion.

Which makes me believe that it will be better to parallelize it in this way instead:

  createArray_ Par (Sz 10) $ \scheduler myarray ->
    iforSchedulerM_ scheduler (0 ..: (100000 :. 1000000000)) $ \_ (j :. i) ->
      when (mytruthcheck j i) $ 
        void $ atomicAddIntArray myarray ((funcofji j i) :: Int) 1

This will ensure that you only schedule a few jobs, instead of billions. Check out iforSchedulerM_ implementation in order to customize it parallelization if you have more insight into your per i and j workload.

user_18480290

based on @freestyle and @lehins answer i wrote this. This is closer to @freestyle's answer but utilizes unsafeAtomicAddIntArray as pointed out by @lehins and maintains the use of loopM_. For some reason @lehins' answer did not finish computing in a reasonable amount of time despite utilizing all cores fully. This was seemingly a result of using the massiv scheduler but i don't know for sure. Worth mentioning that swap was not used and memory was available at all times. With this solution i get a 3x speed up compared to the program in my question that wasn't using forkIO or forkST_. Also this also uses 50% more memory than my single threaded program. I might later add to this answer a program that generalizes commutative operation parallel nested loop processing by distributing threads from the outermost loop towards the innermost loop until all threads of the ncaps have been assigned while taking into account the size of each loop.

 unsafePerformIO $ do
        let arrz = 10 :: Int
        let ncaps = 8 :: Int -- see also getNumCapabilities
        let outerLoopSize = 10^5  :: Int
        let innerLoopSize = 10^12 :: Int
        let chunksz = outerLoopSize `div` (ncaps-1)
        myarray <- newMArray (Sz arrz) 0
        Control.Monad.Parallel.forM_ [0 .. (ncaps-1)] (\k -> (loopM_ (k * chunksz) (< (min outerLoopSize ((k+1) * chunksz) )) (+ 1) (\j -> (loopM_ 0 (< innerLoopSize) (+ 1) (\i -> when (mytruthcheck j i) (void (unsafeAtomicAddIntArray myarray (funcofji j i) 1)))))))
        unsafeFreeze Par myarray
user_165332760
Related