ProcessHandler will terminate execution prematurely if a FrameConsumer is too slow

[marklassau]

Describe the issue
Under certain conditions, my FrameConsumer is not able to finish processing frames within the 10 second timeout that is hard-coded in ProcessHandler.EXECUTOR_TIMEOUT_MILLIS

To Reproduce
Steps to reproduce the behavior:

1. Use an existing audio file as input using UrlInput
2. Create a FrameConsumer with a slow implementation of consume(Frame)
   I was able to reproduce with Thread.sleep(100), but this will depend on environment /

Note that not all frames will be passed to your FrameConsumer.

Sample file
I was able to reproduce with various small audio files in various formats.

Expected behavior
There should be a way to allow for FrameConsumer implementations that require longer processing times.

Logs

16:42:17.651 WARN  [process.ProcessHandler] Executor hasn't stopped in 10000 millis, won't wait longer
16:42:17.651 WARN  [process.Executor] Interrupting ALIVE thread: Runnable-0

Additional context
I agree that the 10 second timeout is probably a fair amount for the majority of use cases, it's just that in some cases application developers will need more time to complete processing.
I suggest the simplest short term solution is simply to make the timeout configurable so developers can adjust depending on circumstances.
I plan to open a PR with a suggested fix sometime soon :)

In the long term it might be nice to have some smarter way to detect when processing is complete - eg if we could somehow know that not all frames have been passed to the FrameConsumer then we would know that processing is on-going.

[marklassau]

PR: #356

[kokorin]

@marklassau this timeout is applied only when ffmpeg process completed but helper threads (used by FrameConsumer) have not.
There is no timeout on each individual frame.

Note that not all frames will be passed to your FrameConsumer.

Could you clarify? ffmpeg sends frames one by one (through TCP using NUT format).

[kokorin]

Could you describe what you are going to achieve in more details? Why handling audio samples takes so long?

[marklassau]

Could you describe what you are going to achieve in more details? Why handling audio samples takes so long?

We read an audio file - and then output a video file that includes that audio.
This works when outputting video at 720p, but when outputting at 1080p it fails intermittently due to the timeout.
In particular it fails on AWS EC2 instances where we are running virtual hardware.

The processing time with a 1080p output in that particular environment is just slow enough to need more than the 10 seconds to finish off the last buffer load of audio frames.

It is probably best if I post some simplified example code that will trigger the issue and you can see for yourself.

[kokorin]

Do you visualize audio wave? Do you use another ffmpeg/Jaffree instance to produce video frames and encode them as video file?
If so I can recommend to have synchronized FIFO queue. FrameConsumer should just push audio sample to the queue (probably it should sleep if queue's length exceeds 2 or 3). FrameProducer should in that case read the queue and produce video frames to be encoded by ffmpeg. In that case first ffmpeg will complete without timeout issue, while the second can finish video frame processing.

[marklassau]

Do you use another ffmpeg/Jaffree instance to produce video frames and encode them as video file?

Exactly

If so I can recommend to have synchronized FIFO queue.

We already use a java.util.concurrent.LinkedBlockingQueue in the real code.

[marklassau]

Here is a small sample code to replicate the problem:

import com.github.kokorin.jaffree.ffmpeg.*;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;

public class FrameConsumerTest {
    private static final Logger LOGGER = LoggerFactory.getLogger(FrameConsumerTest.class);

    @Test
    public void testTimeout() {
        Path inputFile = Paths.get("test-input", "test_audio.m4a");

        FrameConsumer frameConsumer = new FrameConsumer() {
            @Override
            public void consumeStreams(List<Stream> streams) {
                System.out.println("#streams: " + streams.size());
            }

            int frameCount = 0;

            @Override
            public void consume(Frame frame) {
                if (frame == null) {
                    System.out.println("========================================");
                    System.out.println("EOF: null frame detected");
                    System.out.println("========================================");
                    return;
                }
                LOGGER.info("consume() started frame " + frameCount);
                try {
                    Thread.sleep(100);
                } catch (InterruptedException e) {
                    throw new IllegalStateException("InterruptedException", e);
                }
                LOGGER.info("consume() finished frame " + frameCount);

                frameCount++;
            }
        };

        FFmpegResultFuture fFmpegResultFuture = FFmpeg.atPath(Paths.get("/Users/mark/bin/"))
                .addInput(UrlInput.fromPath(inputFile))
                .setOverwriteOutput(true)
                .addOutput(FrameOutput.withConsumer(frameConsumer))
                .executeAsync();

        while (!fFmpegResultFuture.isDone()) {
            LOGGER.info("fFmpegResultFuture.isDone() = " + fFmpegResultFuture.isDone());
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                throw new IllegalStateException("InterruptedException", e);
            }
        }
        LOGGER.info("fFmpegResultFuture.isCancelled() = " + fFmpegResultFuture.isCancelled());
        LOGGER.info("fFmpegResultFuture.isDone() = " + fFmpegResultFuture.isDone());
    }
}

You would have to add a Path to an existing audio file ... and put in your own ffmpeg path.

Note that we use the executeAsync() method - this seems to be part of the problem (it worked when I tested with execute() ).

[marklassau]

When I run the above code I get logs like this:

22:19:02.827 [Runnable-0] INFO  t.FrameConsumerTest - consume() finished frame 895
22:19:02.827 [Runnable-0] INFO  t.FrameConsumerTest - consume() started frame 896
22:19:02.921 [FFmpeg-async-runner] WARN  c.g.k.j.p.ProcessHandler - Executor hasn't stopped in 10000 millis, won't wait longer
22:19:02.921 [FFmpeg-async-runner] WARN  c.g.k.j.p.Executor - Interrupting ALIVE thread: Runnable-0

Note that I should see 915 frames for test audio file I am using and we should see the null frame that signifies EOF.
But this never gets logged.

[kokorin]

It looks like ffmpeg batches several audio frames and sends them to FrameConsumer with one package. So when FrameConsumer receives final batch ffmpeg exits and there is only 10 seconds for FrameConsumer to handle all frames or to pass them somewhere else.
Could you provide the same log with execute method being used (not executeAsync)?
Could you try to run your code with extra -r flag for input audio? BaseInput.setReadAtFrameRate method. I believe it takes longer (in your use-case) to visualize audio stream rather than to play the same audio stream.

[marklassau]

Last night I thought that execute() was working ... but I must have set it up wrong.
I just tried it again and I got the same problem as executeAsync():

17:41:34.639 [Runnable-0] INFO  t.FrameConsumerTest - consume() started frame 897
17:41:34.744 [Runnable-0] INFO  t.FrameConsumerTest - consume() finished frame 897
17:41:34.744 [Runnable-0] INFO  t.FrameConsumerTest - consume() started frame 898
17:41:34.754 [main] WARN  c.g.k.j.p.ProcessHandler - Executor hasn't stopped in 10000 millis, won't wait longer
17:41:34.754 [main] WARN  c.g.k.j.p.Executor - Interrupting ALIVE thread: Runnable-0

I tried the test code using setReadAtFrameRate(true) and now it works as expected:

17:49:14.831 [StdErr] INFO  c.g.k.j.p.BaseStdReader - [info] size=    7335kB time=00:00:19.52 bitrate=3076.8kbits/s speed=0.978x    
17:49:14.832 [StdErr] INFO  c.g.k.j.p.BaseStdReader - [info] video:0kB audio:7324kB subtitle:0kB other streams:0kB global headers:0kB muxing overhead: 0.155895%
17:49:14.833 [main] INFO  c.g.k.j.p.ProcessHandler - Process has finished with status: 0
17:49:14.931 [Runnable-0] INFO  t.FrameConsumerTest - consume() finished frame 829
17:49:14.931 [Runnable-0] INFO  t.FrameConsumerTest - consume() started frame 830
...
17:49:23.651 [Runnable-0] INFO  t.FrameConsumerTest - consume() started frame 915
17:49:23.754 [Runnable-0] INFO  t.FrameConsumerTest - consume() finished frame 915
========================================
EOF: null frame detected
========================================

But you can see in the log times that it only just scrapes in before the 10 second timeout would activate - if I change the 100ms sleep to 200ms then it will timeout again.

[kokorin]

I see your point. And to be honest I'm not sure now that we need helper thread timeout at all. Also I don't remember why this functionality was added. May be at that time Jaffree had issues with threads. And AFAIK this is the first issue regarding executor timeout.

[marklassau]

And AFAIK this is the first issue regarding executor timeout.

That makes sense - 10 seconds should usually be more than enough for most use cases.

We use Jaffree successfully for a bunch of different things (its a great library - thanks for your hard work and sharing it!) and it was only just now that we ran into the timeout for one particular use case.

I think 99% of users will never know or need to know about the timeout.