280 lines
10 KiB
C#
280 lines
10 KiB
C#
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;
|
|
|
|
namespace BCG {
|
|
using HighPerformanceTimer = mbts.spark.HighPerformanceTimer;
|
|
|
|
public class ViewStreamDatabase : IWorker, IDisposable {
|
|
internal delegate void TagFireDelegate(string tag);
|
|
public delegate void DoWorkDelegate();
|
|
public event DoWorkDelegate onDoWorkStart,
|
|
onDoWorkEnd;
|
|
internal 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;
|
|
private HighPerformanceTimer _timer = new HighPerformanceTimer();
|
|
private readonly double myLogTimeThreshold = 10000;
|
|
private readonly double _dowork_threshold = 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 ICollection<string> getCachedTables() {
|
|
List<string> result = new List<string>();
|
|
|
|
|
|
foreach (string s in _table_accumulator_map.Keys)
|
|
if (((ITableAccumulator)_table_accumulator_map[s]).isActive())
|
|
result.Add(s);
|
|
|
|
return result;
|
|
}
|
|
|
|
public ISubscriberCore subscriber { get { return mySubscriberCore; } }
|
|
|
|
|
|
public ITableAccumulator getITableAccumulator(string tableName) {
|
|
// If theres a ITableAccumulator for the tableName its returned (to share subscriptions), otherwise
|
|
// a new one is obtained from the Subscriber factory
|
|
|
|
_table_accumulator_map[tableName] = _table_accumulator_map[tableName] == null ?
|
|
AccumulatorFactory.getAccumulator(tableName, mySubscriberCore) : _table_accumulator_map[tableName];
|
|
|
|
return _table_accumulator_map[tableName] as ITableAccumulator;
|
|
}
|
|
|
|
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.start();
|
|
|
|
if (onTagFireStart != null)
|
|
onTagFireStart(tag);
|
|
table_accumulator.update(tag_updates.Select(t => new MemoryStream(t)).ToArray());
|
|
if (onTagFireEnd != null)
|
|
onTagFireEnd(tag);
|
|
_timer.stop();
|
|
if (_timer.lastIntervalTime > myLogTimeThreshold) {
|
|
bts.Log.Gen.error($"ViewStreamDatabase.doWork tag: {tag} took {_timer.lastIntervalTime} msecs");
|
|
}
|
|
}
|
|
}
|
|
|
|
// 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.start();
|
|
w.doWork();
|
|
_timer.stop();
|
|
if (_timer.lastIntervalTime > _dowork_threshold)
|
|
bts.Log.Gen.warn($"VSD.dowork on worker {w} took {_timer.lastIntervalTime}");
|
|
}
|
|
_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
|
|
}
|
|
|
|
/*
|
|
/// <summary>
|
|
/// Returns a subscriber based on the name of the tag. Hardcoded right now, will change to config
|
|
/// later
|
|
/// </summary>
|
|
public class SubscriberFactory
|
|
{
|
|
public static ITableAccumulator getITableAccumulator (string tag, SubscriberCore subCore)
|
|
{
|
|
string tableName = tag.Substring (3);
|
|
if (!DataStreamGatewaySettings.instance.isSerializableUnderNewScheme (tableName))
|
|
return new TableSubscriber (tag, subCore);
|
|
else
|
|
{
|
|
string subName = DataStreamGatewaySettings.instance.getSubscriber (tableName);
|
|
if (subName.Equals ("BCG.NonIncrementalTableSubscriber"))
|
|
return new NonIncrementalTableSubscriber (tag, subCore);
|
|
else
|
|
return new TableSubscriber (tag, subCore);
|
|
}
|
|
}
|
|
}*/
|
|
}
|