Infinite loop?

[eddiemundo]

Hi, I've been using streamly for a bit and it's a really cool library.

I have an application that batches IO actions and executes them asynchronously at various times, and if the IO actions fail then they are put into a retry queue to be retried at the appropriate times. In certain cases this runs into an infinite loop which also quickly eats up memory. I'm pretty sure isn't in my code, although I can't be sure. I've managed to write a smaller dummy example that reproduces the infinite loop and memory eating behaviour:

import Control.Concurrent (threadDelay)
import Control.Concurrent.Async (async)
import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
import Data.Functor (($>))
import Data.IORef (IORef, atomicModifyIORef', newIORef, readIORef)
import Data.Time.Clock (getCurrentTime)
import qualified Streamly as S
import qualified Streamly.Internal.Data.Fold as Fold
import qualified Streamly.Internal.Data.Stream.StreamD.Type as S
import Streamly.Internal.Data.Unfold.Types (Unfold (..))
import Streamly.Prelude ((|:))
import qualified Streamly.Prelude as S

signalAction :: IORef Int -> IORef Bool -> MVar Int -> Int -> IO Int
signalAction inFlightCountRef signalRef mvar d = do
  atomicModifyIORef' inFlightCountRef (\c -> (c + 1, ()))
  before <- getCurrentTime
  print $ ">>> signal " <> show before
  threadDelay d
  atomicModifyIORef' signalRef (\bool -> (True, ()))
  i <- takeMVar mvar
  after <- getCurrentTime
  print $ "<<< signal " <> show after
  atomicModifyIORef' inFlightCountRef (\c -> (c - 1, ()))
  pure i

regularAction :: IORef Int -> Int -> IO Int
regularAction inFlightCountRef d = do
  atomicModifyIORef' inFlightCountRef (\c -> (c + 1, ()))
  before <- getCurrentTime
  print $ ">>> regulr " <> show before
  threadDelay d
  after <- getCurrentTime
  print $ "<<< regulr " <> show after
  atomicModifyIORef' inFlightCountRef (\c -> (c - 1, ()))
  pure 2

stream :: S.IsStream s => IO (s IO Int)
stream = do
  inFlightCountRef <- newIORef 0
  signalRef <- newIORef False
  mvar <- newEmptyMVar
  pure $
    S.concatMapWith
      S.ahead
      id
      ( S.unfold
          ( Unfold
              ( \actions -> do
                  threadDelay $ 10 ^ 6
                  signal <- readIORef signalRef
                  if signal then async (threadDelay (10 ^ 6 + 10 ^ 4) >> putMVar mvar 1) >> atomicModifyIORef' signalRef (\bool -> (False, ())) else pure ()
                  let (currActions, remainingActions) = splitAt (if signal then 2 else 3) actions
                  print $ "action count: " <> show (length currActions)
                  if length currActions == 0
                    then do
                      inFlightCount <- readIORef inFlightCountRef
                      if inFlightCount > 0
                        then pure $ S.Skip actions
                        else pure $ S.Stop
                    else pure $ S.Yield (S.aheadly $ foldr (|:) S.nil currActions) remainingActions
              )
              pure
          )
          [ 
            regularAction inFlightCountRef $ 10 ^ 6 `div` 2,
            regularAction inFlightCountRef (10 ^ 6 + 10 ^ 4),
            regularAction inFlightCountRef (10 ^ 6 `div` 2 * 3),
            regularAction inFlightCountRef $ 10 ^ 6 `div` 2,
            signalAction inFlightCountRef signalRef mvar (10 ^ 6 + 10 ^ 4),
            regularAction inFlightCountRef (10 ^ 6 `div` 2 * 3),
            -- if you comment the below program works
            regularAction inFlightCountRef $ 10 ^ 6 `div` 2,
            regularAction inFlightCountRef (10 ^ 6 + 10 ^ 4),
            regularAction inFlightCountRef (10 ^ 6 `div` 2 * 3),
            regularAction inFlightCountRef $ 10 ^ 6 `div` 2,
            regularAction inFlightCountRef (10 ^ 6 + 10 ^ 4),
            regularAction inFlightCountRef (10 ^ 6 `div` 2 * 3)
          ]
      )

main :: IO ()
main = do
  putStrLn "Hello, Haskell!"
  s <- stream
  (S.fold Fold.toList s) >>= print

How the example works is that the unfold step outputs 3 actions inside an Ahead stream every second. This stream of streams is flattened via concatMapWith and ahead and so from the outside it looks like 3 actions are executed asynchronously every second. When a signalAction is executed, after a delay, it writes to a signal IORef and waits on an MVar. The unfold step will check if the signal IORef has been written and if so asynchronously (after a delay again) write some value to the MVar theoretically unblocking the signal action. Additionally if the signal IORef was written the unfold step will group 1 less action in the stream it outputs.

The delays in the action bodies affect whether the program infinite loops. If they are less than 1 second (10^6 microseconds) then I think everything works fine. However I've set it up so that for every 3 actions, the first action waits about 0.5 seconds, the second action a little more than 1 second, and the third action about 1.5 seconds.

If I shorten the action list to 6 items it works, but 7 items and more it doesn't work.
If I move actions 3-5 to the front it works.
Hopefully I haven't made a silly mistake.

[eddiemundo]

When S.async is used instead of S.ahead in S.concatMapWith then there is no infinite loop and the program finishes...

[harendra-kumar]

I have not looked at it in detail, but it could be due to the ordering of takeMVar and putMVar caused by concurrent execution. In general, you should try to design your application such as to avoid using MVar as long as possible.

Also, there is a significant difference in the implementation of ahead and async when flattening stream of streams. ahead is more likely to get deadlocked by out of order MVar take/put compared to async. Because async can execute many more actions in future.

Let's take a simplified example:

[x1 `ahead` x2 `ahead` x3] `ahead` [y1 `ahead` y2 `ahead` y3] `ahead` [z1 `ahead` z2 `ahead` z3]

The implementation of ahead is such that it can execute x1, x2, y1, z1 in advance, but the rest of y stream i.e.y2, y3 and the rest of z stream i.e. z2 and z3 won't be executed until the whole of x stream gets executed. Now, if let's say y1 or z1 took an MVar that is required for x1, x2 or x3 to finish then we are into a deadlock. Also if the release of MVar depends on y2, y3 or z2, z3 being executed then we can get into a deadlock.

async is different than ahead, in case of async y2, y3 and z2, z3 may get executed even if the whole x stream has not yet been executed.

Hopefully this gives you a clue.

[eddiemundo]

I didn't realize ahead worked like that when flattening streams but I don't think that's the cause of the infinite loop/deadlock. I think it's an infinite loop though because my system will run out of memory.

The scenario I have is say y2 fails. y2 will then create an MVar and create a retry action that fills this MVar with its result, then enqueue this retry action, then wait on the MVar. So y2 is now blocked waiting for the enqueued retry action to run.

It's the unfold step that will dequeue retry actions and run them asynchronously, before yielding a stream of unrelated actions. So as long as the unfold steps are being run y2 should eventually be unblocked.

OK, so I think the problem is at the end when I'm folding over the stream, it will fold up to the result of y2 then get stuck due to the enqueued retry, and never run the next unfold step.

This kinda shows that this is the problem

S.foldlM' (\_ _ -> print "hi") (pure ()) |$. stream

It will print "hi" up to the point of the y2 action, then never print again.

As a side note the parallel application operator doesn't get the next unfold step to run, but maybe that makes sense.

Anyway this is probably enough to go on to figure out a workaround so thanks.

[harendra-kumar]

@eddiemundo closing this, let us know if there is anything that needs to be done in the library.