spark/src/DataStreamSuite/DataStreamCore/ViewStreamDatabase.cs

using System;
using System.Collections;
using System.IO;
using BCG.Utilities;
using BCG.Utilities.General;
//using BCG.Utilities.BlueConfiguration;
using System.Collections.Generic;
using System.Threading;
using mbts;
using System.Linq;
using System.Diagnostics;

namespace BCG {
   public class ViewStreamDatabase : IWorker, IDisposable {
      internal delegate void TagFireDelegate(string tag);
      public delegate void DoWorkDelegate();
      private event DoWorkDelegate onDoWorkStart,
                                    onDoWorkEnd;
      private event TagFireDelegate onTagFireStart,
                                    onTagFireEnd;
      //currently onTagUpdate is usd by the DataStreamPerfRecorder to keep track of how often the updates for various streams come in
      //and how big they are. 
      //public delegate void OnTagUpdateDelegate(PubMetaData metaData);
      //public event OnTagUpdateDelegate onTagUpdate;
      private Dictionary<string, ITableAccumulator> _table_accumulator_map = new Dictionary<string, ITableAccumulator>();
      private static ViewStreamDatabase mySelf;
      private ISubscriberCore mySubscriberCore;
      private List<IWorker> _workers = new List<IWorker>();
      private bool myIsDisposed = false;
      private bool _is_in_do_work = false;
      private static object myTimeInterval;
      readonly Stopwatch _timer = new();
      readonly TimeSpan myLogTimeThreshold = TimeSpan.FromSeconds(10);
      readonly TimeSpan _dowork_threshold = TimeSpan.FromMilliseconds(500);
      private Thread myThreadOwner = null;  // ViewStreamDatabase takes a thread affinity when created.  If called from a different thread it will assert.

      private static readonly object myInstanceLockObject = new object();
      private readonly object myStateLock = new object();

      public ViewStreamDatabase()
         : this(new SubscriberCore()) {
      }

      public ViewStreamDatabase(ISubscriberCore subscriberCore) {
         mySubscriberCore = subscriberCore;
         // if (DataStreamLoadFileConfig.instance.profileStreams)
         //    new DataStreamPerfRecorder(this);
      }

      private static int myGetTimeInterval() {
         if (myTimeInterval == null) {
            // TODO
            string timeInterval = String.Empty; //ConfigurationRequester.request("viewStreamDatabaseTimeOut");
            if (timeInterval == null || timeInterval == "")
               myTimeInterval = 0;
            else
               myTimeInterval = Int32.Parse(timeInterval);
         }
         return (int)myTimeInterval;
      }

      public static int instance_hashcode = -1;
      public static ViewStreamDatabase instance {
         get {
            if (mySelf == null || mySelf.myIsDisposed)
               lock (myInstanceLockObject) {
                  if (mySelf == null || mySelf.myIsDisposed) {
                     mySelf = new ViewStreamDatabase();
                     instance_hashcode = mySelf.GetHashCode();
                  }
               }
            return mySelf;
         }
         set {
            lock (myInstanceLockObject) {
               if (mySelf != null)
                  mySelf.mySubscriberCore.Dispose();
               mySelf = value;
               instance_hashcode = mySelf.GetHashCode();
            }
         }
      }

      public ISubscriberCore subscriber { get { return mySubscriberCore; } }


      public ITableAccumulator getITableAccumulator(string table_name) {
         // If theres a ITableAccumulator for the tableName its returned (to share subscriptions), otherwise 
         // a new one is obtained from the Subscriber factory
         ITableAccumulator rv = null;
         var table_tag_name = DSTagManager.get_stream_tag(table_name);
         if (!_table_accumulator_map.TryGetValue(table_tag_name, out rv)) {
            rv = AccumulatorFactory.getAccumulator(table_tag_name, mySubscriberCore);
            _table_accumulator_map.Add(table_tag_name, rv);
         }
         return rv;
      }

      public void doWork() {
         // TODO
         verifyAccess();

         if (_is_suspended)
            return;
         if (_is_worker_call_in_progress)
            return;
         else {
            _is_worker_call_in_progress = true; // to allow reentrancy
            try {
               if (onDoWorkStart != null)
                  onDoWorkStart();

               var updates = mySubscriberCore.get_updates(TimeSpan.FromMilliseconds(10));
               if (updates != null) {
                  var consolidated_update_map = _consolidate_updates(updates);

                  foreach (var kv in consolidated_update_map) {
                     string tag = kv.Key;
                     List<byte[]> tag_updates = kv.Value;
                     ITableAccumulator table_accumulator;
                     if (!_table_accumulator_map.TryGetValue(tag, out table_accumulator)) {
                        bts.Log.Gen.warn("Received callback for tag: {tag}");
                        continue;
                     }
                     _timer.Restart();

                     if (onTagFireStart != null)
                        onTagFireStart(tag);
                     table_accumulator.handle_updates(tag_updates);
                     if (onTagFireEnd != null)
                        onTagFireEnd(tag);
                     _timer.Stop();
                     if (_timer.Elapsed > myLogTimeThreshold) {
                        bts.Log.Gen.error($"ViewStreamDatabase.doWork tag: {tag} took {_timer.Elapsed}");
                     }
                  }
               }

               // to allow for registering of workers on the worker callback

               _is_worker_call_in_progress = true;
               _pending_worker_ops.Clear();
               foreach (IWorker w in _workers) {
                  _timer.Restart();
                  w.doWork();
                  _timer.Stop();
                  if (_timer.Elapsed > _dowork_threshold) {
                     bts.Log.Gen.warn($"VSD.dowork on worker {w} took {_timer.Elapsed}");
                  }
               }
               _is_worker_call_in_progress = false;

               foreach (PendingAddRemoveWorker op in _pending_worker_ops)
                  if (op.isAdd)
                     addWorker(op.w);
                  else
                     removeWorker(op.w);
            } finally {
               _pending_worker_ops.Clear();
               _is_in_do_work = false;
               _is_worker_call_in_progress = false;
               if (onDoWorkEnd != null)
                  onDoWorkEnd();
            }
         }
      }

      private bool _is_suspended = false;

      public void suspend() { _is_suspended = true; }

      public void resume() { _is_suspended = false; }

      public bool checkAccess() {
         if (this.myThreadOwner == null) this.myThreadOwner = Thread.CurrentThread;
         return this.myThreadOwner == Thread.CurrentThread;
      }

      public void verifyAccess() {
         if (!checkAccess()) {
            string errorMessage = string.Format("Invalid cross thread access on ViewStreamDatabase.  Owning thread: {0}.  Accessing thread: {1}", this.myThreadOwner.ManagedThreadId, Thread.CurrentThread.ManagedThreadId);

            //#if DEBUG
            //            BlueEventLog2.instance().assert(false, errorMessage);
            //#else
            //            BlueEventLog2.instance().logEvent(errorMessage, BlueEventLog2Level.Fatal);
            //#endif

         }
      }

      private IDictionary<string, List<byte[]>> _consolidate_updates(IEnumerable<bts.PubSubServerMessage> updates) {
         Dictionary<string, List<byte[]>> rv = new Dictionary<string, List<byte[]>>();
         foreach (var msg in updates) {
            List<byte[]> tag_updates = null;
            if (!rv.TryGetValue(msg.tag, out tag_updates)) {
               tag_updates = new List<byte[]>();
               rv.Add(msg.tag, tag_updates);
            }
            tag_updates.Add(msg.data);
         }
         return rv;
      }

      public bool isConsistent { get { return true; } }

      public void addWorker(IWorker w) {
         verifyAccess();
         if (_is_worker_call_in_progress)
            _pending_worker_ops.Add(new PendingAddRemoveWorker(true, w));
         else
            _workers.Add(w);
      }


      public void removeWorker(IWorker w) {
         verifyAccess();
         if (_is_worker_call_in_progress)
            _pending_worker_ops.Add(new PendingAddRemoveWorker(false, w));
         else
            _workers.Remove(w);
      }

      #region IDisposable Members

      public void Dispose() {
         if (!myIsDisposed) {
            myIsDisposed = true;
            mySubscriberCore.Dispose();
         }
      }

      private List<PendingAddRemoveWorker> _pending_worker_ops = new List<PendingAddRemoveWorker>();
      private bool _is_worker_call_in_progress = false;
      class PendingAddRemoveWorker {
         public PendingAddRemoveWorker(bool isAdd, IWorker w) {
            this.isAdd = isAdd;
            this.w = w;
         }

         public bool isAdd;
         public IWorker w;
      }

      #endregion
   }
}