Merge branch 'async'

RServeCLI2.Example/RserveExamples.cs

@@ -20,7 +20,7 @@ public class MainClass
         /// <param name="args">The arguments.</param>
         public static void Main( string[] args )
         {
-            using ( var s = new RConnection( new System.Net.IPAddress( new byte[] { 127 , 0 , 0 , 1 } ) ) )
+            using ( var s = RConnection.Connect( new System.Net.IPAddress( new byte[] { 127 , 0 , 0 , 1 } ) ) )
             {
                 // Generate some example data
                 var x = Enumerable.Range( 1 , 20 ).ToArray();

RServeCLI2.Test/Rservice.cs

@@ -73,7 +73,7 @@ public Rservice( bool showWindow = false , int maxInputBufferSizeInKb = 0 )
 
             // create a connection to the server
             // ReSharper disable RedundantArgumentDefaultValue
-            RConnection = new RConnection( port: Port );
+            RConnection = RConnection.Connect( port: Port );
             // ReSharper restore RedundantArgumentDefaultValue
 
         }

RServeCLI2/Async.cs

@@ -0,0 +1,101 @@
+//-----------------------------------------------------------------------
+// Original work Copyright (c) 2015, Atif Aziz. All rights reserved.
+// Portions Copyright (c) Microsoft. All rights reserved.
+//-----------------------------------------------------------------------
+
+namespace RserveCLI2
+{
+    using System;
+    using System.Collections.Generic;
+    using System.Linq;
+    using System.Runtime.CompilerServices;
+    using System.Runtime.ExceptionServices;
+    using System.Threading.Tasks;
+
+    static class Async
+    {
+        static readonly ConditionalWeakTable<Exception, AggregateException>
+            SyncRunErrors = new ConditionalWeakTable<Exception, AggregateException>();
+
+        public static T RunSynchronously<T>(Task<T> task)
+        {
+            RunSynchronously((Task) task);
+            return task.Result;
+        }
+
+        public static void RunSynchronously(Task task)
+        {
+            try
+            {
+                task.Wait();
+            }
+            catch (AggregateException e)
+            {
+                // When waiting on a task, all exceptions surface as
+                // AggregateException but this is mostly useless because
+                // synchronous code is generally written with catch blocks for
+                // specific exception types like IOException. However, these
+                // will never be caught given that they are housed and masked
+                // by AggregateException. To remedy the situation somewhat,
+                // a base exception is arbitrarily selected and throw instead.
+                // The exception is captured and thrown via
+                // ExceptionDispatchInfo to avoid resetting its stack trace.
+                //
+                // In the rare event that the original AggregateException may
+                // be useful, e.g. to log other errors in a fork-join
+                // scenario, the aggregate is saved so that it can be looked
+                // up later and as long as the base exception is still alive.
+
+                var someBaseException = e.DeepEnumerateInnerExceptions().FirstOrDefault();
+                if (someBaseException == null)
+                    throw;
+                SyncRunErrors.Add(someBaseException, e);
+                ExceptionDispatchInfo.Capture(someBaseException).Throw();
+                throw; // Should never end up here!
+            }
+        }
+
+        public static AggregateException GetSyncRunAggregateException(Exception exception)
+        {
+            if (exception == null) throw new ArgumentNullException(nameof(exception));
+            AggregateException aggregate;
+            return SyncRunErrors.TryGetValue(exception, out aggregate) ? aggregate : null;
+        }
+
+        /// <remarks>
+        /// This is similar to <see cref="AggregateException.Flatten"/> except
+        /// it flattens the inner exceptions lazily.
+        /// </remarks>
+
+        static IEnumerable<Exception> DeepEnumerateInnerExceptions(this AggregateException exception)
+        {
+            if (exception == null) throw new ArgumentNullException(nameof(exception));
+            return DeepEnumerateInnerExceptionsImpl(exception);
+        }
+
+        static IEnumerable<Exception> DeepEnumerateInnerExceptionsImpl(AggregateException exception)
+        {
+            // For sake of compatibility, adapted from the algorithm found in
+            // AggregateException.Flatten:
+            // https://github.com/dotnet/coreclr/blob/ef1e2ab328087c61a6878c1e84f4fc5d710aebce/src/mscorlib/src/System/AggregateException.cs#L390
+            // Portion Copyright (c) Microsoft. All rights reserved.
+            // Licensed under the MIT license.
+
+            var pending = new Queue<AggregateException>();
+            pending.Enqueue(exception);
+            while (pending.Count > 0)
+            {
+                foreach (var inner in pending.Dequeue().InnerExceptions)
+                {
+                    if (inner == null)
+                        continue;
+                    var nestedAggregate = inner as AggregateException;
+                    if (nestedAggregate != null)
+                        pending.Enqueue(nestedAggregate);
+                    else
+                        yield return inner;
+                }
+            }
+        }
+    }
+}

RServeCLI2/DelegateExtensions.cs

@@ -0,0 +1,45 @@
+#region Copyright (c) 2009 Atif Aziz. All rights reserved.
+//
+// Mannex - Extension methods for .NET
+// Copyright (c) 2009 Atif Aziz. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//    http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+#endregion
+
+namespace Mannex
+{
+    using System;
+
+    /// <summary>
+    /// Extension methods for <see cref="Delegate"/>.
+    /// </summary>
+
+    static partial class DelegateExtensions
+    {
+        /// <summary>
+        /// Sequentially invokes each delegate in the invocation list as
+        /// <see cref="EventHandler{TEventArgs}"/> and ignores exceptions
+        /// thrown during the invocation of any one handler (continuing
+        /// with the next handler in the list).
+        /// </summary>
+
+        public static void InvokeAsEventHandlerWhileIgnoringErrors<T>(this Delegate del, object sender, T args)
+        {
+            if (del == null) throw new ArgumentNullException("del");
+            // ReSharper disable once PossibleInvalidCastExceptionInForeachLoop
+            foreach (EventHandler<T> handler in del.GetInvocationList())
+                try { handler(sender, args); } catch { /* ignored */ }
+        }
+    }
+}

RServeCLI2/FaultedTask.cs

@@ -0,0 +1,67 @@
+//-----------------------------------------------------------------------
+// Original work Copyright (c) 2015, Atif Aziz. All rights reserved.
+// Portions Copyright (c) Microsoft. All rights reserved.
+//-----------------------------------------------------------------------
+
+namespace RserveCLI2
+{
+    using System;
+    using System.Threading.Tasks;
+    using Mannex;
+
+    // This infrastructure is essentially there to make sure no exception of a
+    // faulted task goes unobserved by this library code!
+
+    class UnhandledTaskExceptionEventArgs : EventArgs
+    {
+        public Exception Exception { get; }
+        public Task Task { get; }
+
+        public UnhandledTaskExceptionEventArgs(Task task)
+        {
+            if (task == null) throw new ArgumentNullException(nameof(task));
+            if (!task.IsFaulted) throw new ArgumentException(null, nameof(task));
+
+            Task = task;
+
+            // This effectively causes the task to not go "unobserved"!
+
+            Exception = task.Exception;
+        }
+    }
+
+    static class FaultedTask
+    {
+        /// <remarks>
+        /// The execution of this event's handler is not synchronized with
+        /// the context.
+        /// </remarks>
+
+        public static event EventHandler<UnhandledTaskExceptionEventArgs> UnhandledTaskException;
+
+        // If this class is ever made public, the following methods should
+        // remain internal to the library code only.
+
+        internal static void OnUnhandledException(UnhandledTaskExceptionEventArgs args)
+        {
+            if (args == null) throw new ArgumentNullException(nameof(args));
+            UnhandledTaskException?.InvokeAsEventHandlerWhileIgnoringErrors(null, args);
+        }
+
+        internal static void NotifyUnhandledException(this Task task) =>
+            OnUnhandledException(new UnhandledTaskExceptionEventArgs(task));
+
+        internal static void NotifyUnhandledExceptionIfFaulted(this Task task)
+        {
+            if (task == null) throw new ArgumentNullException(nameof(task));
+            if (task.IsFaulted)
+                task.NotifyUnhandledException();
+        }
+
+        internal static void IgnoreFault(this Task task)
+        {
+            task.ContinueWith(t => t.NotifyUnhandledExceptionIfFaulted(),
+                              TaskScheduler.Default);
+        }
+    }
+}