Add multithreading for saving tracker data (Fix #163)

This adds multithreading when saving data. This becomes an issues when
large amounts of data is being written out, where the methods in
SaveData do alot of string processing. To ensure data is correctly
written out, this also wait for the threads to complete when session
is ending.

Author: ahmed-shariff

Assets/UXF/Scripts/Etc/ResultsDict.cs

@@ -13,6 +13,7 @@ public class ResultsDictionary
     {
         private Dictionary<string, object> baseDict;
         private bool allowAdHocAdding;
+        private Object lockObject = new Object();
 
         /// <summary>
         /// Dictionary of results for a trial.
@@ -25,7 +26,10 @@ public ResultsDictionary(IEnumerable<string> initialKeys, bool allowAdHocAdding)
             this.allowAdHocAdding = allowAdHocAdding;
             foreach (var key in initialKeys)
             {
-                baseDict.Add(key, string.Empty);
+                lock (lockObject)
+                {
+                    baseDict.Add(key, string.Empty);
+                }
             }
         }
 
@@ -38,13 +42,16 @@ public object this[string key]
         {
             get { return baseDict[key]; }
             set {
-                if (allowAdHocAdding || baseDict.ContainsKey(key))
-                {
-                    baseDict[key] = value;
-                }
-                else
+                lock (lockObject)
                 {
-                    throw new KeyNotFoundException(string.Format("Custom header \"{0}\" does not exist!", key));
+                    if (allowAdHocAdding || baseDict.ContainsKey(key))
+                    {
+                        baseDict[key] = value;
+                    }
+                    else
+                    {
+                        throw new KeyNotFoundException(string.Format("Custom header \"{0}\" does not exist!", key));
+                    }
                 }
             }
         }
@@ -73,4 +80,4 @@ public bool ContainsKey(string key)
 
     }
 
-}
+}

Assets/UXF/Scripts/Etc/Trial.cs

@@ -5,7 +5,9 @@
 using System.Threading;
 using UnityEngine;
 using System.Collections.Specialized;
-
+using System.Threading.Tasks;
+using System.Collections.Concurrent;
+using System.IO;
 
 namespace UXF
 {
@@ -62,6 +64,11 @@ public bool saveData
         /// </summary>
         public ResultsDictionary result;
 
+        // Used by the worker task
+        private static BlockingQueue<System.Action> blockingQueue = new BlockingQueue<System.Action>();
+        private static Task workerTask;
+        private static bool quitting = false;
+
         /// <summary>
         /// Manually create a trial. When doing this you need to add this trial to a block with block.trials.Add(trial)
         /// </summary>
@@ -277,7 +284,12 @@ private void SaveData()
                     tracker.StopRecording();
                     if (tracker.Data.CountRows() > 0)
                     {
-                        SaveDataTable(tracker.Data, tracker.DataName, dataType: UXFDataType.Trackers);
+                        UXFDataTable table = tracker.Data;
+                        string name = tracker.DataName;
+                        ManageInWorker(() =>
+                        {
+                            SaveDataTable(table, name, dataType: UXFDataType.Trackers);
+                        });
                     }
                 }
                 catch (NullReferenceException)
@@ -292,9 +304,68 @@ private void SaveData()
                 result[s] = settings.GetObject(s, string.Empty);
             }
         }
-    }
 
-    
+        /// <summary>
+        /// Adds a new command to a queue which is executed in a separate worker thread when it is available.
+        /// Warning: The Unity Engine API is not thread safe, so do not attempt to put any Unity commands here.
+        /// </summary>
+        /// <param name="action"></param>
+        public static void ManageInWorker(System.Action action)
+        {
+            if (workerTask == null)
+            {
+                workerTask = Task.Run(Worker);
+                quitting = false;
+            }
+
+            blockingQueue.Enqueue(action);
+        }
+
+        /// <summary>
+        /// The worker thread used when <see cref="ManageInWorker"/> is called.
+        /// </summary>
+        private static void Worker()
+        {
+            // performs FileIO tasks in seperate thread
+            foreach (var action in blockingQueue)
+            {
+                try
+                {
+                    action.Invoke();
+                }
+                catch (ThreadAbortException)
+                {
+                    break;
+                }
+                catch (IOException e)
+                {
+                    Utilities.UXFDebugLogError(string.Format("Error, file may be in use! Exception: {0}", e));
+                }
+                catch (System.Exception e)
+                {
+                    // stops thread aborting upon an exception
+                    Debug.LogException(e);
+                }
+
+                if (quitting && blockingQueue.NumItems() == 0)
+                {
+                    break;
+                }
+            }
+        }
+
+        /// <summary>
+        /// Wait for all tasks scheduled through <see cref="ManageInWorker"/>
+        /// </summary>
+        public static void WaitForTasks()
+        {
+            Utilities.UXFDebugLog("Waiting for tasks to finish");
+            quitting = true;
+            blockingQueue.Enqueue(() => {}); // ensures bq breaks from foreach loop
+            workerTask?.Wait();
+            Utilities.UXFDebugLog("Tasks finished");
+        }
+    }
 
     /// <summary>
     /// Status of a trial
@@ -307,4 +378,4 @@ public enum TrialStatus
     }
 
 
-}
+}

Assets/UXF/Scripts/Session.cs

@@ -695,6 +695,8 @@ public void End()
 
         void SaveResults()
         {
+            Trial.WaitForTasks();
+
             // generate list of all headers possible
             // hashset keeps unique set of keys
             HashSet<string> resultsHeaders = new HashSet<string>();