Merge pull request #36 from Symphony-DAS/1.5.0

Fixed failed tasks from not clearing out of task list

Author: cafarm

src/symphony-core/Symphony.Core/Controller.cs

@@ -682,6 +682,8 @@ private void Process(ConcurrentQueue<Epoch> epochQueue, EpochPersistor persistor
         {
             try
             {
+                CompletedEpochTasks.Clear();
+
                 ProcessLoop(epochQueue, persistor);
             }
             finally
@@ -761,7 +763,7 @@ private void ProcessLoop(ConcurrentQueue<Epoch> epochQueue, EpochPersistor persi
                     // Throw if any previous completed epoch tasks faulted
                     if (CompletedEpochTasks.Any(t => t.IsFaulted))
                     {
-                        throw new AggregateException(CompletedEpochTasks.Select(t => t.Exception));
+                        throw new AggregateException(CompletedEpochTasks.Where(t => t.IsFaulted).Select(t => t.Exception));
                     }
 
                     Epoch currentEpoch;

src/symphony-core/Symphony.Core/DAQController.cs

@@ -271,6 +271,9 @@ protected void Process(bool waitForTrigger)
             // pushing the results back through the input pipeline
             try
             {
+                OutputTasks.Clear();
+                InputTasks.Clear();
+
                 WillBeginProcessLoop();
                 ProcessLoop(waitForTrigger);
             }
@@ -401,13 +404,9 @@ private void PushIncomingData(IEnumerable<KeyValuePair<IDAQInputStream, IInputDa
             // Throw if any previous tasks faulted
             if (InputTasks.Any(t => t.IsFaulted))
             {
-                var lookup = InputTasks.ToLookup(t => t.IsFaulted);
-                var faultedTasks = lookup[true];
-                InputTasks = lookup[false].ToList();
-                throw new AggregateException(faultedTasks.Select(t => t.Exception));
+                throw new AggregateException(InputTasks.Where(t => t.IsFaulted).Select(t => t.Exception));
             }
 
-
             var newTask = InputTasks.Any()
                               ? Task.Factory.ContinueWhenAll(InputTasks.ToArray(),
                                                              (tasks) =>
@@ -425,8 +424,6 @@ private void PushIncomingData(IEnumerable<KeyValuePair<IDAQInputStream, IInputDa
                                           }
                                       });
 
-
-
             InputTasks = InputTasks.Where(t => !t.IsCompleted).ToList();
             InputTasks.Add(newTask);
         }
@@ -626,9 +623,8 @@ protected virtual void CommonStop()
             }
             catch (Exception ex)
             {
-                log.WarnFormat("An output task failed while stopping: {0}", ex);
+                log.ErrorFormat("An output task failed while stopping: {0}", ex);
             }
-            OutputTasks.Clear();
 
             SetStreamsBackground();
         }