2006-08-09 Gonzalo Paniagua Javier <gonzalo@ximian.com>

	* mcs/class/System/System.Diagnostics/Process.cs: add support for 2.0
	asynchronous reads on stdout and stderr.

	* mono/mono/metadata/threadpool.c: treat pipes from process
	asynchronous reads as sockets when reading from them, so we get
	select/poll or epoll to wait for data.


svn path=/trunk/mcs/; revision=63565

1 files changed, 284 insertions, 19 deletions

diff --git a/mcs/class/System/System.Diagnostics/Process.cs b/mcs/class/System/System.Diagnostics/Process.cs
index 79b04a98280..e94aeaee1cd 100644
--- a/mcs/class/System/System.Diagnostics/Process.cs
+++ b/mcs/class/System/System.Diagnostics/Process.cs
@@ -8,7 +8,7 @@
 //
 // (C) 2002 Ximian, Inc.
 // (C) 2003 Andreas Nahr
-// (c) 2004 Novell, Inc. (http://www.novell.com)
+// (c) 2004,2005,2006 Novell, Inc. (http://www.novell.com)
 //
 
 //
@@ -33,6 +33,7 @@
 //
 
 using System.IO;
+using System.Text;
 using System.ComponentModel;
 using System.ComponentModel.Design;
 using System.Runtime.CompilerServices;
@@ -71,6 +72,8 @@ namespace System.Diagnostics {
 		bool already_waiting;
 		ISynchronizeInvoke synchronizingObject;
 		EventHandler exited_event;
+		IntPtr stdout_rd;
+		IntPtr stderr_rd;
 		
 		/* Private constructor called from other methods */
 		private Process(IntPtr handle, int id) {
@@ -556,6 +559,12 @@ namespace System.Diagnostics {
 				if (error_stream == null) {
 					throw new InvalidOperationException("Standard error has not been redirected");
 				}
+#if NET_2_0
+				if ((async_mode & AsyncModes.AsyncError) != 0)
+					throw new InvalidOperationException ("Cannot mix asynchronous and synchonous reads.");
+
+				async_mode |= AsyncModes.SyncError;
+#endif
 
 				return(error_stream);
 			}
@@ -584,6 +593,12 @@ namespace System.Diagnostics {
 				if (output_stream == null) {
 					throw new InvalidOperationException("Standard output has not been redirected");
 				}
+#if NET_2_0
+				if ((async_mode & AsyncModes.AsyncOutput) != 0)
+					throw new InvalidOperationException ("Cannot mix asynchronous and synchonous reads.");
+
+				async_mode |= AsyncModes.SyncOutput;
+#endif
 
 				return(output_stream);
 			}
@@ -873,8 +888,8 @@ namespace System.Diagnostics {
 				throw new FileNotFoundException  ("Executable not found: " + startInfo.FileName);
 			ProcInfo proc_info=new ProcInfo();
 			IntPtr stdin_rd, stdin_wr;
-			IntPtr stdout_rd, stdout_wr;
-			IntPtr stderr_rd, stderr_wr;
+			IntPtr stdout_wr;
+			IntPtr stderr_wr;
 			bool ret;
 			MonoIOError error;
 
@@ -904,8 +919,11 @@ namespace System.Diagnostics {
 			}
 
 			if (startInfo.RedirectStandardOutput == true) {
-				ret = MonoIO.CreatePipe (out stdout_rd,
+				IntPtr out_rd;
+				ret = MonoIO.CreatePipe (out out_rd,
 						         out stdout_wr);
+
+				process.stdout_rd = out_rd;
 				if (ret == false) {
 					if (startInfo.RedirectStandardInput == true) {
 						MonoIO.Close (stdin_rd, out error);
@@ -915,27 +933,30 @@ namespace System.Diagnostics {
 					throw new IOException ("Error creating standard output pipe");
 				}
 			} else {
-				stdout_rd = (IntPtr)0;
+				process.stdout_rd = (IntPtr)0;
 				stdout_wr = MonoIO.ConsoleOutput;
 			}
 
 			if (startInfo.RedirectStandardError == true) {
-				ret = MonoIO.CreatePipe (out stderr_rd,
+				IntPtr err_rd;
+				ret = MonoIO.CreatePipe (out err_rd,
 						         out stderr_wr);
+
+				process.stderr_rd = err_rd;
 				if (ret == false) {
 					if (startInfo.RedirectStandardInput == true) {
 						MonoIO.Close (stdin_rd, out error);
 						MonoIO.Close (stdin_wr, out error);
 					}
 					if (startInfo.RedirectStandardOutput == true) {
-						MonoIO.Close (stdout_rd, out error);
+						MonoIO.Close (process.stdout_rd, out error);
 						MonoIO.Close (stdout_wr, out error);
 					}
 					
 					throw new IOException ("Error creating standard error pipe");
 				}
 			} else {
-				stderr_rd = (IntPtr)0;
+				process.stderr_rd = (IntPtr)0;
 				stderr_wr = MonoIO.ConsoleError;
 			}
 			
@@ -949,12 +970,12 @@ namespace System.Diagnostics {
 				}
 
 				if (startInfo.RedirectStandardOutput == true) {
-					MonoIO.Close (stdout_rd, out error);
+					MonoIO.Close (process.stdout_rd, out error);
 					MonoIO.Close (stdout_wr, out error);
 				}
 
 				if (startInfo.RedirectStandardError == true) {
-					MonoIO.Close (stderr_rd, out error);
+					MonoIO.Close (process.stderr_rd, out error);
 					MonoIO.Close (stderr_wr, out error);
 				}
 
@@ -976,12 +997,12 @@ namespace System.Diagnostics {
 
 			if (startInfo.RedirectStandardOutput == true) {
 				MonoIO.Close (stdout_wr, out error);
-				process.output_stream = new StreamReader (new FileStream (stdout_rd, FileAccess.Read, true));
+				process.output_stream = new StreamReader (new FileStream (process.stdout_rd, FileAccess.Read, true));
 			}
 
 			if (startInfo.RedirectStandardError == true) {
 				MonoIO.Close (stderr_wr, out error);
-				process.error_stream = new StreamReader (new FileStream (stderr_rd, FileAccess.Read, true));
+				process.error_stream = new StreamReader (new FileStream (process.stderr_rd, FileAccess.Read, true));
 			}
 
 			process.StartExitCallbackIfNeeded ();
@@ -1043,18 +1064,43 @@ namespace System.Diagnostics {
 		 * exit.  ms can be <0 to mean wait forever.
 		 */
 		[MethodImplAttribute(MethodImplOptions.InternalCall)]
-		private extern bool WaitForExit_internal(IntPtr handle,
-							 int ms);
+		private extern bool WaitForExit_internal(IntPtr handle, int ms);
 
-		public void WaitForExit() {
-			WaitForExit_internal(process_handle, -1);
+		public void WaitForExit ()
+		{
+			WaitForExit (-1);
 		}
 
 		public bool WaitForExit(int milliseconds) {
-			if (milliseconds == int.MaxValue)
-				milliseconds = -1;
+			int ms = milliseconds;
+			if (ms == int.MaxValue)
+				ms = -1;
 
-			return WaitForExit_internal (process_handle, milliseconds);
+#if NET_2_0
+			DateTime start = DateTime.UtcNow;
+			if (async_output != null && !async_output.IsCompleted) {
+				if (false == async_output.WaitHandle.WaitOne (ms, false))
+					return false; // Timed out
+
+				if (ms >= 0) {
+					ms -= (int) (DateTime.UtcNow - start).TotalMilliseconds;
+					if (ms <= 0)
+						return false;
+				}
+			}
+
+			if (async_error != null && !async_error.IsCompleted) {
+				if (false == async_error.WaitHandle.WaitOne (ms, false))
+					return false; // Timed out
+
+				if (ms >= 0) {
+					ms -= (int) (DateTime.UtcNow - start).TotalMilliseconds;
+					if (ms <= 0)
+						return false;
+				}
+			}
+#endif
+			return WaitForExit_internal (process_handle, ms);
 		}
 
 		[MonoTODO]
@@ -1067,6 +1113,225 @@ namespace System.Diagnostics {
 			return(false);
 		}
 
+
+#if NET_2_0
+		public event DataReceivedEventHandler OutputDataReceived;
+		public event DataReceivedEventHandler ErrorDataReceived;
+
+		void OnOutputDataReceived (string str)
+		{
+			if (OutputDataReceived != null)
+				OutputDataReceived (this, new DataReceivedEventArgs (str));
+		}
+
+		void OnErrorDataReceived (string str)
+		{
+			if (ErrorDataReceived != null)
+				ErrorDataReceived (this, new DataReceivedEventArgs (str));
+		}
+
+		[Flags]
+		enum AsyncModes {
+			NoneYet = 0,
+			SyncOutput = 1,
+			SyncError = 1 << 1,
+			AsyncOutput = 1 << 2,
+			AsyncError = 1 << 3
+		}
+
+		[StructLayout (LayoutKind.Sequential)]
+		sealed class ProcessAsyncReader
+		{
+			/*
+			   The following fields match those of SocketAsyncResult.
+			   This is so that changes needed in the runtime to handle
+			   asynchronous reads are trivial
+			*/
+			/* DON'T shuffle fields around. DON'T remove fields */
+			public object Sock;
+			public IntPtr handle;
+			public object state;
+			public AsyncCallback callback;
+			public ManualResetEvent wait_handle;
+
+			public Exception delayedException;
+
+			public object EndPoint;
+			byte [] buffer = new byte [4196];
+			public int Offset;
+			public int Size;
+			public int SockFlags;
+
+			public object acc_socket;
+			public int total;
+			public bool completed_sync;
+			bool completed;
+			bool err_out; // true -> stdout, false -> stderr
+			internal int error;
+			public int operation = 6; // MAGIC NUMBER: Maximum in SocketOperation + 1
+			public object ares;
+
+
+			// These fields are not in SocketAsyncResult
+			Process process;
+			Stream stream;
+			StringBuilder sb = new StringBuilder ();
+			public AsyncReadHandler ReadHandler;
+
+			public ProcessAsyncReader (Process process, IntPtr handle, bool err_out)
+			{
+				this.process = process;
+				this.handle = handle;
+				stream = new FileStream (handle, FileAccess.Read, false);
+				this.ReadHandler = new AsyncReadHandler (AddInput);
+				this.err_out = err_out;
+			}
+
+			public void AddInput ()
+			{
+				lock (this) {
+					int nread = stream.Read (buffer, 0, buffer.Length);
+					if (nread == 0) {
+						completed = true;
+						if (wait_handle != null)
+							wait_handle.Set ();
+						Flush (true);
+						return;
+					}
+
+					try {
+						sb.Append (Encoding.Default.GetString (buffer, 0, nread));
+					} catch {
+						// Just in case the encoding fails...
+						for (int i = 0; i < nread; i++) {
+							sb.Append ((char) buffer [i]);
+						}
+					}
+
+					Flush (false);
+					ReadHandler.BeginInvoke (null, this);
+				}
+			}
+
+			void Flush (bool last)
+			{
+				if (sb.Length == 0 ||
+				    (err_out && process.output_canceled) ||
+				    (!err_out && process.error_canceled))
+					return;
+
+				string total = sb.ToString ();
+				sb.Length = 0;
+				string [] strs = total.Split ('\n');
+				int len = strs.Length;
+				if (len == 0)
+					return;
+
+				for (int i = 0; i < len - 1; i++) {
+					if (err_out)
+						process.OnOutputDataReceived (strs [i]);
+					else
+						process.OnErrorDataReceived (strs [i]);
+				}
+
+				string end = strs [len - 1];
+				if (last || end == "") {
+					if (err_out)
+						process.OnOutputDataReceived (end);
+					else
+						process.OnErrorDataReceived (end);
+				} else {
+					sb.Append (end);
+				}
+			}
+
+			public bool IsCompleted {
+				get { return completed; }
+			}
+
+			public WaitHandle WaitHandle {
+				get {
+					lock (this) {
+						if (wait_handle == null)
+							wait_handle = new ManualResetEvent (completed);
+						return wait_handle;
+					}
+				}
+			}
+		}
+
+		AsyncModes async_mode;
+		bool output_canceled;
+		bool error_canceled;
+		ProcessAsyncReader async_output;
+		ProcessAsyncReader async_error;
+		delegate void AsyncReadHandler ();
+
+		[ComVisibleAttribute(false)] 
+		public void BeginOutputReadLine ()
+		{
+			if (process_handle == IntPtr.Zero || output_stream == null || StartInfo.RedirectStandardOutput == false)
+				throw new InvalidOperationException ("Standard output has not been redirected or process has not been started.");
+
+			if ((async_mode & AsyncModes.SyncOutput) != 0)
+				throw new InvalidOperationException ("Cannot mix asynchronous and synchonous reads.");
+
+			async_mode |= AsyncModes.AsyncOutput;
+			output_canceled = false;
+			if (async_output == null) {
+				async_output = new ProcessAsyncReader (this, stdout_rd, true);
+				async_output.ReadHandler.BeginInvoke (null, async_output);
+			}
+		}
+
+		[ComVisibleAttribute(false)] 
+		public void CancelOutputRead ()
+		{
+			if (process_handle == IntPtr.Zero || output_stream == null || StartInfo.RedirectStandardOutput == false)
+				throw new InvalidOperationException ("Standard output has not been redirected or process has not been started.");
+
+			if ((async_mode & AsyncModes.SyncOutput) != 0)
+				throw new InvalidOperationException ("OutputStream is not enabled for asynchronous read operations.");
+
+			if (async_output == null)
+				throw new InvalidOperationException ("No async operation in progress.");
+
+			output_canceled = true;
+		}
+
+		[ComVisibleAttribute(false)] 
+		public void BeginErrorReadLine ()
+		{
+			if (process_handle == IntPtr.Zero || error_stream == null || StartInfo.RedirectStandardError == false)
+				throw new InvalidOperationException ("Standard error has not been redirected or process has not been started.");
+
+			if ((async_mode & AsyncModes.SyncError) != 0)
+				throw new InvalidOperationException ("Cannot mix asynchronous and synchonous reads.");
+
+			async_mode |= AsyncModes.AsyncError;
+			error_canceled = false;
+			if (async_error == null) {
+				async_error = new ProcessAsyncReader (this, stderr_rd, false);
+				async_error.ReadHandler.BeginInvoke (null, async_error);
+			}
+		}
+
+		[ComVisibleAttribute(false)] 
+		public void CancelErrorRead ()
+		{
+			if (process_handle == IntPtr.Zero || output_stream == null || StartInfo.RedirectStandardOutput == false)
+				throw new InvalidOperationException ("Standard output has not been redirected or process has not been started.");
+
+			if ((async_mode & AsyncModes.SyncOutput) != 0)
+				throw new InvalidOperationException ("OutputStream is not enabled for asynchronous read operations.");
+
+			if (async_error == null)
+				throw new InvalidOperationException ("No async operation in progress.");
+
+			error_canceled = true;
+		}
+#endif
+
 		[Category ("Behavior")]
 		[MonitoringDescription ("Raised when this process exits.")]
 		public event EventHandler Exited {