Merge pull request #81 from hudl/ImplementTimeOuts

Implement timeout functionality and process monitoring

Author: Casey-Bateman

Hudl.FFmpeg.Core/Command/BaseTypes/ICommand.cs

@@ -12,7 +12,11 @@ ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>()
             where TProcessorType : class, ICommandProcessor, new()
             where TBuilderType : class, ICommandBuilder, new();
 
-        ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>(TProcessorType commandProcessor)
+        ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>(int? timeoutMilliseconds)
+            where TProcessorType : class, ICommandProcessor, new()
+            where TBuilderType : class, ICommandBuilder, new();
+
+        ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>(TProcessorType commandProcessor, int? timeoutMilliseconds)
             where TProcessorType : class, ICommandProcessor
             where TBuilderType : class, ICommandBuilder, new();
     }

Hudl.FFmpeg.Core/Command/BaseTypes/ICommandProcessor.cs

@@ -37,6 +37,9 @@ public interface ICommandProcessor
         /// </summary>
         bool Send(string command);
 
-       
+        /// <summary>
+        /// processes the given command string against the processor engine
+        /// </summary>
+        bool Send(string command, int? timeoutMilliseconds);
     }
 }

Hudl.FFmpeg.Core/Command/Models/FFcommandBase.cs

@@ -4,9 +4,9 @@
 
 namespace Hudl.FFmpeg.Command.Models
 {
-    public abstract class FFcommandBase : ICommand
+    public abstract class FFCommandBase : ICommand
     {
-        protected FFcommandBase()
+        protected FFCommandBase()
         {
             PreExecutionAction = EmptyOperation;
             PostExecutionAction = EmptyOperation;
@@ -27,6 +27,13 @@ protected FFcommandBase()
         public ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>()
             where TProcessorType : class, ICommandProcessor, new()
             where TBuilderType : class, ICommandBuilder, new()
+        {
+            return ExecuteWith<TProcessorType, TBuilderType>(null);
+        }
+
+        public ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>(int? timeoutMilliseconds)
+            where TProcessorType : class, ICommandProcessor, new()
+            where TBuilderType : class, ICommandBuilder, new()
         {
             var commandProcessor = new TProcessorType();
 
@@ -35,7 +42,7 @@ public ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>()
                 throw new FFmpegRenderingException(commandProcessor.Error);
             }
 
-            var returnType = ExecuteWith<TProcessorType, TBuilderType>(commandProcessor);
+            var returnType = ExecuteWith<TProcessorType, TBuilderType>(commandProcessor, timeoutMilliseconds);
 
             if (!commandProcessor.Close())
             {
@@ -45,7 +52,7 @@ public ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>()
             return returnType;
         }
 
-        public ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>(TProcessorType commandProcessor)
+        public ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>(TProcessorType commandProcessor, int? timeoutMilliseconds)
             where TProcessorType : class, ICommandProcessor
             where TBuilderType : class, ICommandBuilder, new()
         {
@@ -59,7 +66,7 @@ public ICommandProcessor ExecuteWith<TProcessorType, TBuilderType>(TProcessorTyp
 
             PreExecutionAction(Owner, this, true);
 
-            if (!commandProcessor.Send(commandBuilder.ToString()))
+            if (!commandProcessor.Send(commandBuilder.ToString(), timeoutMilliseconds))
             {
                 PostExecutionAction(Owner, this, false);
 

Hudl.FFmpeg.Core/Command/Models/FFcommandBuilderBase.cs

@@ -2,11 +2,11 @@
 
 namespace Hudl.FFmpeg.Command.Models
 {
-    public abstract class FFcommandBuilderBase 
+    public abstract class FFCommandBuilderBase 
     {
         protected readonly StringBuilder BuilderBase;
 
-        protected FFcommandBuilderBase()
+        protected FFCommandBuilderBase()
         {
             BuilderBase = new StringBuilder(100);            
         }

Hudl.FFmpeg.Core/Command/StreamReaders/BaseStandardStreamReader.cs

@@ -0,0 +1,98 @@
+using Hudl.FFmpeg.Logging;
+using System;
+using System.Diagnostics;
+using System.Text;
+using System.Threading;
+
+namespace Hudl.FFmpeg.Command.StreamReaders
+{
+    public abstract class BaseStandardStreamReader
+    {
+        private static readonly LogUtility Log = LogUtility.GetLogger(typeof(BaseStandardStreamReader));
+        private const int OutputBuilderLimit = 10000;
+
+        protected BaseStandardStreamReader(Process processToListenTo)
+        {
+            OutputBuilder = new StringBuilder();
+            ProcessToListenTo = processToListenTo;
+            _stopSignal = new ManualResetEvent(false);
+        }
+
+        protected Thread MonitoringThread;
+        protected StringBuilder OutputBuilder;
+        protected readonly Process ProcessToListenTo;
+        protected volatile bool _stopped;
+        protected ManualResetEvent _stopSignal;
+
+        public string LastLineReceived { get; set; }
+
+        public void Listen()
+        {
+            //workaround for a bug in the mono process when attempting to read async from console output events
+            // - link http://mono.1490590.n4.nabble.com/System-Diagnostic-Process-and-event-handlers-td3246096.html
+            if (ResourceManagement.IsMonoRuntime())
+            {
+                ListenAsThread();
+            }
+            else
+            {
+                ListenAsAsync();
+            }
+        }
+        protected abstract void ListenAsAsync();
+        protected abstract void ListenAsThread();
+
+        public void Stop()
+        {
+            _stopped = true;
+            _stopSignal.WaitOne(1000);
+            if (ResourceManagement.IsMonoRuntime())
+            {
+                try
+                {
+                    if (MonitoringThread.ThreadState != System.Threading.ThreadState.Stopped)
+                    {
+                        MonitoringThread.Abort();
+                    }
+                }
+                catch (Exception e)
+                {
+                    Log.Error("Unable to abort the monitoring thread", e);
+                }
+            }
+        }
+
+        private void HandleDataReceived(string data)
+        {
+            if (!string.IsNullOrWhiteSpace(data))
+            {
+                LastLineReceived = data;
+            }
+
+            //there is no specific reason behind this number other than, it seems like a pretty good number baseline. 
+            //ultimately ffmpeg can blow up the standard output/error stream with logs. we dont want to create
+            //an OOM exception. so we will trash and reset.
+            if (OutputBuilder.Length > OutputBuilderLimit)
+            {
+                var newBuilder = new StringBuilder(OutputBuilderLimit);
+                newBuilder.Append(OutputBuilder.ToString(), OutputBuilderLimit / 2, OutputBuilderLimit / 2);
+                OutputBuilder = newBuilder;
+            }
+
+            OutputBuilder.AppendLine(data);
+        }
+        protected void HandleDataReceivedAsAsync(object sender, DataReceivedEventArgs dataReceivedEventArgs)
+        {
+            HandleDataReceived(dataReceivedEventArgs.Data);
+        }
+        protected void HandleDataReceivedAsThread(string data)
+        {
+            HandleDataReceived(data);
+        }
+       
+        public override string ToString()
+        {
+            return OutputBuilder.ToString();
+        }
+    }
+}

Hudl.FFmpeg.Core/Command/StreamReaders/StandardErrorAsyncStreamReader.cs

@@ -0,0 +1,54 @@
+using System.Diagnostics;
+using System.Threading;
+
+namespace Hudl.FFmpeg.Command.StreamReaders
+{
+    public class StandardErrorAsyncStreamReader : BaseStandardStreamReader
+    {
+        private StandardErrorAsyncStreamReader(Process processToListenTo)
+            : base(processToListenTo)
+        {
+            ProcessToListenTo.StartInfo.RedirectStandardError = true;
+            if (!ResourceManagement.IsMonoRuntime())
+            {
+                ProcessToListenTo.ErrorDataReceived += HandleDataReceivedAsAsync;
+            }
+        }
+
+        public static StandardErrorAsyncStreamReader AttachReader(Process processToListenTo)
+        {
+            return new StandardErrorAsyncStreamReader(processToListenTo);
+        }
+
+        protected override void ListenAsAsync()
+        {
+            ProcessToListenTo.BeginErrorReadLine();
+        }
+
+        protected override void ListenAsThread()
+        {
+            MonitoringThread = new Thread(AsyncStdErrorMonitor);
+            MonitoringThread.Start();
+        }
+
+        private void AsyncStdErrorMonitor()
+        {
+            try
+            {
+                do
+                {
+                    string line = ProcessToListenTo.StandardError.ReadLine();
+                    if (line != null)
+                    {
+                        HandleDataReceivedAsThread(line);
+                    }
+                }
+                while (!_stopped && !ProcessToListenTo.HasExited);
+                _stopSignal.Set();
+            }
+            catch
+            { }
+        }
+    }
+
+}

Hudl.FFmpeg.Core/Command/StreamReaders/StandardOutputAsyncStreamReader.cs

@@ -0,0 +1,54 @@
+using System.Diagnostics;
+using System.Threading;
+
+namespace Hudl.FFmpeg.Command.StreamReaders
+{
+    public class StandardOutputAsyncStreamReader : BaseStandardStreamReader
+    {
+        private StandardOutputAsyncStreamReader(Process processToListenTo)
+            : base(processToListenTo)
+        {
+            ProcessToListenTo.StartInfo.RedirectStandardOutput = true;
+            if (!ResourceManagement.IsMonoRuntime())
+            {
+                ProcessToListenTo.ErrorDataReceived += HandleDataReceivedAsAsync;
+            }
+        }
+
+        public static StandardOutputAsyncStreamReader AttachReader(Process processToListenTo)
+        {
+            return new StandardOutputAsyncStreamReader(processToListenTo);
+        }
+
+        protected override void ListenAsAsync()
+        {
+            ProcessToListenTo.BeginErrorReadLine();
+        }
+
+        protected override void ListenAsThread()
+        {
+            MonitoringThread = new Thread(AsyncStdErrorMonitor);
+            MonitoringThread.Start();
+        }
+
+        private void AsyncStdErrorMonitor()
+        {
+            try
+            {
+                do
+                {
+                    string line = ProcessToListenTo.StandardError.ReadLine();
+                    if (line != null)
+                    {
+                        HandleDataReceivedAsThread(line);
+                    }
+                }
+                while (!_stopped && !ProcessToListenTo.HasExited);
+                _stopSignal.Set();
+            }
+            catch
+            { }
+        }
+    }
+
+}

Hudl.FFmpeg.Core/Exceptions/FfmpegTimeoutException.cs

@@ -0,0 +1,20 @@
+using System;
+
+namespace Hudl.FFmpeg.Exceptions
+{
+    /// <summary>
+    /// exception that is thrown when FFmpeg commands have passed the timeout period
+    /// </summary>
+    public class FFmpegTimeoutException : Exception
+    {
+        public FFmpegTimeoutException(string arguments)
+            : base("FFmpeg timed out during processing")
+        {
+            base.Data["Arguments"] = arguments;
+
+            Arguments = arguments;
+        }
+
+        public string Arguments { get; private set; }
+    }
+}

Hudl.FFmpeg.Core/Extensions/ProcessExtensions.cs

@@ -0,0 +1,71 @@
+using System;
+using System.Diagnostics;
+using System.Threading;
+
+namespace Hudl.FFmpeg.Extensions
+{
+    public static class ProcessExtensions
+    {
+        public static bool WaitForProcessStart(this Process process)
+        {
+            return process.WaitForProcessStart(null);
+        }
+        public static bool WaitForProcessStart(this Process process, int? timeoutMilliseconds)
+        {
+            var processTimeout = TimeSpan.FromMilliseconds(timeoutMilliseconds ?? 10000);
+            return process.WaitForProcessStart(processTimeout);
+        }
+        public static bool WaitForProcessStart(this Process process, TimeSpan processTimeout)
+        {
+            var processStopwatch = Stopwatch.StartNew();
+            var isProcessRunning = false;
+
+            while (processStopwatch.ElapsedMilliseconds <= processTimeout.TotalMilliseconds && !isProcessRunning)
+            {
+                //give a little bit of breathing room here....
+                Thread.Sleep(20.Milliseconds());
+
+                try
+                {
+                    isProcessRunning = (process != null && process.HasExited && process.Id != 0);
+                }
+                catch
+                {
+                    isProcessRunning = false;
+                }
+            }
+
+            return isProcessRunning;
+        }
+
+        public static bool WaitForProcessStop(this Process process)
+        {
+            return process.WaitForProcessStop(null);
+        }
+        public static bool WaitForProcessStop(this Process process, int? timeoutMilliseconds)
+        {
+            var processTimeout = TimeSpan.FromMilliseconds(timeoutMilliseconds ?? 0);
+            return process.WaitForProcessStop(processTimeout);
+        }
+        public static bool WaitForProcessStop(this Process process, TimeSpan processTimeout)
+        {
+            var processStopwatch = Stopwatch.StartNew();
+
+            while (!process.HasExited)
+            {
+                Thread.Sleep(1.Seconds());
+
+                if (processTimeout.TotalMilliseconds > 0 && processStopwatch.ElapsedMilliseconds > processTimeout.TotalMilliseconds)
+                {
+                    process.Kill();
+
+                    process.WaitForExit((int)5.Seconds().TotalMilliseconds);
+
+                    return false;
+                }
+            }
+
+            return true;
+        }
+    }
+}