TemporalSet.cs

using Microsoft.VisualBasic.Devices;
using MyCaffe.basecode;
using MyCaffe.basecode.descriptors;
using MyCaffe.db.image;
using SimpleGraphing;
using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
 
namespace MyCaffe.db.temporal
{
    [Serializable]
    public class TemporalSet : IDisposable
    {
        Exception m_loadException = null;
        int m_nHistoricSteps;
        int m_nFutureSteps;
        int m_nTotalSteps;
        DateTime? m_dtStart;
        DateTime? m_dtEnd;
        ManualResetEvent m_evtCancel = new ManualResetEvent(false);
        AutoResetEvent m_evtDone = new AutoResetEvent(false);
        List<ItemSet> m_rgItems = new List<ItemSet>();
        int m_nLoadLimit = 0;
        double m_dfReplacementPct;
        int m_nRefreshUpdateMs;
        DB_LOAD_METHOD m_loadMethod;
        DatabaseTemporal m_db;
        CryptoRandom m_random = null;
        SourceDescriptor m_src;
        Log m_log;
        Thread m_loadThread = null;
        int m_nChunks = 1024;
        int m_nItemIdx = 0;
        bool m_bNormalizeData = false;
        double m_dfLoadPct = 0;
        object m_objSync = new object();
 
        public TemporalSet(Log log, DatabaseTemporal db, SourceDescriptor src, DB_LOAD_METHOD loadMethod, int nLoadLimit, double dfReplacementPct, int nRefreshUpdateMs, CryptoRandom random, int nHistoricSteps, int nFutureSteps, int nChunks)
        {
            m_log = log;
            m_random = random;
            m_db = db;
            m_src = src;
            m_loadMethod = loadMethod;
            m_nLoadLimit = nLoadLimit;
            m_nHistoricSteps = nHistoricSteps;
            m_nFutureSteps = nFutureSteps;
            m_nTotalSteps = m_nHistoricSteps + m_nFutureSteps;
            m_nChunks = nChunks;
            m_dfReplacementPct = dfReplacementPct;
            m_nRefreshUpdateMs = nRefreshUpdateMs;
 
            if (m_nLoadLimit < 0)
                m_nLoadLimit = 0;
 
            if (m_nLoadLimit > 0 && m_nLoadLimit < 1000)
                m_nLoadLimit = 1000;
 
            if (m_dfReplacementPct > 0.9)
                m_dfReplacementPct = 0.9;
 
            if (m_nRefreshUpdateMs < 250)
                m_nRefreshUpdateMs = 250;
        }
 
        public void Dispose()
        {
            m_evtCancel.Set();
        }
 
        public void CleanUp()
        {
            if (m_loadThread != null)
            {
                m_evtCancel.Set();
                m_loadThread = null;
            }
 
            m_rgItems.Clear();
        }
 
        public void Reset()
        {
            m_nItemIdx = 0;
 
            foreach (ItemSet item in m_rgItems)
            {
                item.Reset();
            }
        }
 
        public bool Initialize(bool bNormalizedData, EventWaitHandle evtCancel)
        {
            m_bNormalizeData = bNormalizedData;
 
            if (m_loadThread == null)
            {
                m_loadThread = new Thread(new ThreadStart(loadThread));
                m_loadThread.Start();
            }
 
            if (m_loadMethod == DB_LOAD_METHOD.LOAD_ALL)
                return WaitForLoadingToComplete(evtCancel);
 
            return true;
        }
 
        private void removeItem(List<ItemSet> rgItems)
        {
            lock (m_objSync)
            {
                if (m_rgItems == null)
                    return;
 
                if (m_rgItems.Count == 0)
                    return;
 
                ItemSet item = rgItems[0];
                rgItems.RemoveAt(0);
                item.CleanUp();
            }
        }
 
        private void loadThread()
        {
            DatabaseLoader loader = new DatabaseLoader();
            List<ValueItem> rgItems = m_db.GetAllValueItems(m_src.ID);
            double dfMemMin = 1000.0;
 
            try
            {
                m_dfLoadPct = 0;
 
                TemporalDescriptor td = loader.LoadTemporalFromDb(m_src.ID);
                OrderedValueStreamDescriptorSet rgStrm = td.OrderedValueStreamDescriptors;
                
                m_dtStart = td.StartDate;
                m_dtEnd = td.EndDate;
 
                if (m_dtEnd.HasValue && m_dtStart.HasValue && m_dtEnd <= m_dtStart)
                    throw new Exception("The end date must be greater than the start date.");
 
                // Load the data.
                DateTime? dt = m_dtStart;
                bool bEOD = false;
 
                Stopwatch sw = new Stopwatch();
                sw.Start();
                m_dfLoadPct = 0;
                int nIdx = 0;
 
                ComputerInfo info = new ComputerInfo();
                Stopwatch swReplacement = new Stopwatch();
 
                swReplacement.Start();
 
                while (!m_evtCancel.WaitOne(0))
                {
                    bool bEOD1;
 
                    // Load one chunk for each item.
                    while (m_rgItems.Count < rgItems.Count && (m_nLoadLimit <= 0 || m_rgItems.Count < m_nLoadLimit))
                    {
                        ItemSet item = new ItemSet(m_random, m_db, rgItems[nIdx], rgStrm);
                        item.Load(out bEOD1);
 
                        lock (m_objSync)
                        {
                            m_rgItems.Add(item);
                        }
 
                        m_dfLoadPct = (double)m_rgItems.Count / rgItems.Count;                       
                        if (bEOD1)
                            bEOD = true;
 
                        if (sw.Elapsed.TotalMilliseconds > 1000)
                        {
                            sw.Restart();
                            m_log.Progress = m_dfLoadPct;
                            m_log.WriteLine("Loading '" + m_src.Name + "' data for item " + item.Item.Name + " (" + m_dfLoadPct.ToString("P") + ")...", true);
                        }
 
                        if (m_evtCancel.WaitOne(0))
                            break;
 
                        nIdx++;
 
                        if (nIdx >= rgItems.Count)
                            nIdx = 0;
 
                        if (nIdx % m_nChunks == 0)
                            break;
                    }
 
                    if (m_nLoadLimit > 0)
                    {
                        int nLoadLimit = m_nLoadLimit;
                        
                        if ((int)swReplacement.Elapsed.TotalMilliseconds > m_nRefreshUpdateMs)
                            nLoadLimit = (int)(m_nLoadLimit * (1 - m_dfReplacementPct));
 
                        if (nLoadLimit > 1000)
                        {
                            while (m_rgItems.Count > nLoadLimit)
                            {
                                removeItem(m_rgItems);
                            }
 
                            GC.Collect(2, GCCollectionMode.Forced);
                        }
                    }
 
                    ulong nMem = info.AvailablePhysicalMemory;
                    double dfMb = (double)nMem / (1024.0 * 1024.0);
                    int nRetryCount = 0;
 
                    while (dfMb < dfMemMin && nRetryCount < 10 && m_rgItems.Count > 1000)
                    {
                        m_log.WriteLine("Waiting for memory to free up...", true);
 
                        for (int i=0; i<1000 && m_rgItems.Count > 1000; i++)
                        {
                            removeItem(m_rgItems);
                        }
 
                        GC.Collect(2, GCCollectionMode.Forced);
                        nMem = info.AvailablePhysicalMemory;
                        dfMb = (double)nMem / (1024.0 * 1024.0);
 
                        nRetryCount++;
                        Thread.Sleep(250);
                    }
 
                    if (dfMb < dfMemMin)
                        return;
 
                    if (m_rgItems.Count == rgItems.Count && m_nLoadLimit <= 0)
                        break;
                }
            }
            catch (Exception excpt)
            {
                m_loadException = excpt;
                return;
            }
            finally
            {
                m_evtDone.Set();
            }
        }
 
        public bool WaitForLoadingToComplete(EventWaitHandle evtCancel, int nWaitMs = int.MaxValue)
        {
            WaitHandle[] rgWait = new WaitHandle[] { m_evtCancel, evtCancel, m_evtDone };
            int nWait = WaitHandle.WaitAny(rgWait, nWaitMs);
 
            if (nWait <= 1)
                return false;
 
            return true;
        }
 
        public double LoadPercent
        {
            get { return m_dfLoadPct; }
        }
 
        public DB_LOAD_METHOD LoadMethod
        {
            get { return m_loadMethod; }
        }
 
        public SimpleTemporalDatumCollection GetData(int nQueryIdx, ref int? nItemIdx, ref int? nValueIdx, DB_LABEL_SELECTION_METHOD itemSelectionMethod, DB_ITEM_SELECTION_METHOD valueSelectionMethod, int nValueStepOffset = 1, bool bEnableDebug = false, string strDebugPath = null)
        {
            SimpleTemporalDatumCollection data = null;
 
            lock (m_objSync)
            {
                if (itemSelectionMethod == DB_LABEL_SELECTION_METHOD.RANDOM)
                {
                    m_nItemIdx = m_random.Next(m_rgItems.Count);
                }
                else
                {
                    if (m_nItemIdx >= m_rgItems.Count)
                        m_nItemIdx = 0;
                }
 
                if (nItemIdx.HasValue)
                    m_nItemIdx = nItemIdx.Value;
                nItemIdx = m_nItemIdx;
 
                if (m_nItemIdx >= m_rgItems.Count)
                    return null;
 
                data = m_rgItems[m_nItemIdx].GetData(nQueryIdx, ref nValueIdx, valueSelectionMethod, m_nHistoricSteps, m_nFutureSteps, nValueStepOffset, bEnableDebug, strDebugPath);
 
                int nRetryCount = 0;
                while (data == null && nRetryCount < 40)
                {
                    if (itemSelectionMethod == DB_LABEL_SELECTION_METHOD.RANDOM)
                    {
                        m_nItemIdx = m_random.Next(m_rgItems.Count);
                    }
                    else
                    {
                        m_nItemIdx++;
 
                        if (m_nItemIdx >= m_rgItems.Count)
                            m_nItemIdx = 0;
                    }
 
                    nItemIdx = m_nItemIdx;
                    data = m_rgItems[m_nItemIdx].GetData(nQueryIdx, ref nValueIdx, valueSelectionMethod, m_nHistoricSteps, m_nFutureSteps, nValueStepOffset, bEnableDebug, strDebugPath);
                    nRetryCount++;
                }
            }
 
            return data;
        }
 
        public int GetCount()
        {
            lock (m_objSync)
            {
                return m_rgItems.Sum(p => p.GetCount(m_nHistoricSteps + m_nFutureSteps));
            }
        }
 
        public int Count
        {
            get 
            {
                lock (m_objSync)
                {
                    return m_rgItems.Count;
                }
            }
        }
    }
}