timeseriesimpl.cs

Perst开源实时数据库

            return ((IEnumerable)iterator(0, Int64.MaxValue, IterationOrder.AscentOrder)).GetEnumerator();
        }

        public IEnumerator<T> GetEnumerator() 
#else
        public override IEnumerator GetEnumerator() 
#endif
        { 
            return iterator(0, Int64.MaxValue, IterationOrder.AscentOrder).GetEnumerator();
        }

#if USE_GENERICS
        public IEnumerator<T> GetEnumerator(DateTime from, DateTime till) 
#else
        public IEnumerator GetEnumerator(DateTime from, DateTime till) 
#endif
        {
            return iterator(from.Ticks, till.Ticks, IterationOrder.AscentOrder).GetEnumerator();
        }

#if USE_GENERICS
        public IEnumerator<T> GetEnumerator(DateTime from, DateTime till, IterationOrder order) 
#else
        public IEnumerator GetEnumerator(DateTime from, DateTime till, IterationOrder order) 
#endif
        {
            return iterator(from.Ticks, till.Ticks, order).GetEnumerator();
        }

#if USE_GENERICS
        public IEnumerator<T> GetEnumerator(IterationOrder order) 
#else
        public IEnumerator GetEnumerator(IterationOrder order) 
#endif
        {
            return iterator(0, Int64.MaxValue, order).GetEnumerator();
        }

#if USE_GENERICS
        public IEnumerable<T> Range(DateTime from, DateTime till) 
#else
        public IEnumerable Range(DateTime from, DateTime till) 
#endif
        {
            return iterator(from.Ticks, till.Ticks, IterationOrder.AscentOrder);
        }

#if USE_GENERICS
        public IEnumerable<T> Range(DateTime from, DateTime till, IterationOrder order) 
#else
        public IEnumerable Range(DateTime from, DateTime till, IterationOrder order) 
#endif
        {
            return iterator(from.Ticks, till.Ticks, order);
        }

#if USE_GENERICS
        public IEnumerable<T> Range(IterationOrder order) 
#else
        public IEnumerable Range(IterationOrder order) 
#endif
        {
            return iterator(0, Int64.MaxValue, order);
        }

#if USE_GENERICS
        public IEnumerable<T> Till(DateTime till) 
#else
        public IEnumerable Till(DateTime till) 
#endif
        { 
            return iterator(0, till.Ticks, IterationOrder.DescentOrder);
        }

#if USE_GENERICS
        public IEnumerable<T> From(DateTime from) 
#else
        public IEnumerable From(DateTime from) 
#endif
        { 
            return iterator(from.Ticks, Int64.MaxValue, IterationOrder.AscentOrder);
        }

#if USE_GENERICS
        public IEnumerable<T> Reverse() 
#else
        public IEnumerable Reverse() 
#endif
        { 
            return iterator(0, Int64.MaxValue, IterationOrder.DescentOrder);
        }


#if USE_GENERICS
        private IEnumerable<T> iterator(long from, long till, IterationOrder order) 
        { 
            IEnumerator<TimeSeriesBlock> enumerator = index.GetEnumerator(from - maxBlockTimeInterval, till, order);
            return order == IterationOrder.AscentOrder
                ? (IEnumerable<T>)new TimeSeriesEnumerator(enumerator, from, till)
                : (IEnumerable<T>)new TimeSeriesReverseEnumerator(enumerator, from, till);
        }
#else
        private IEnumerable iterator(long from, long till, IterationOrder order) 
        { 
            IEnumerator enumerator = index.GetEnumerator(from - maxBlockTimeInterval, till, order);
            return order == IterationOrder.AscentOrder
                ? (IEnumerable)new TimeSeriesEnumerator(enumerator, from, till)
                : (IEnumerable)new TimeSeriesReverseEnumerator(enumerator, from, till);
        }
#endif

        public DateTime FirstTime 
        {
            get 
            { 
                foreach (TimeSeriesBlock block in index) 
                {
                    return new DateTime(block.timestamp);
                } 
                throw new StorageError(StorageError.ErrorCode.KEY_NOT_FOUND);
            }
        }

        public DateTime LastTime 
        {
            get 
            {
                foreach (TimeSeriesBlock block in index.Range(null, null, IterationOrder.DescentOrder)) 
                {
                    return new DateTime(block[block.used-1].Time);
                } 
                throw new StorageError(StorageError.ErrorCode.KEY_NOT_FOUND);
            }
        }

#if USE_GENERICS
        public int Count 
#else
        public override int Count 
#endif
        {
            get 
            {
                return (int)CountTicks();
            }
        }

        public long CountTicks()
        {
            long n = 0;
            foreach (TimeSeriesBlock block in index) 
            {
                n += block.used;
            }
            return n;
        }
       
#if USE_GENERICS
        public T this[DateTime timestamp]     
#else
        public TimeSeriesTick this[DateTime timestamp]     
#endif
        {
            get 
            {
                long time = timestamp.Ticks;
                foreach (TimeSeriesBlock block in index.Range(time - maxBlockTimeInterval, time, IterationOrder.AscentOrder))
                {
                    int n = block.used;
                    int l = 0, r = n;
                    while (l < r)  
                    {
                        int i = (l+r) >> 1;
                        if (time > block[i].Time) 
                        { 
                            l = i+1;
                        } 
                        else 
                        { 
                            r = i;
                        }
                    }
                    Debug.Assert(l == r && (l == n || block[l].Time >= time)); 
                    if (l < n && block[l].Time == time) 
                    { 
                        return block[l];
                    }
                }
                throw new StorageError(StorageError.ErrorCode.KEY_NOT_FOUND);
            }
        }

        public bool Contains(DateTime timestamp) 
        {
            return this[timestamp] != null;
        }

        public int Remove(DateTime from, DateTime till) 
        {
            return remove(from.Ticks, till.Ticks);
        }

        public int RemoveFrom(DateTime from) 
        {
            return remove(from.Ticks, Int64.MaxValue);
        }

        public int RemoveTill(DateTime till) 
        {
            return remove(0, till.Ticks);
        }

        public int RemoveAll() 
        {
            return remove(0, Int64.MaxValue);
        }

        private int remove(long from, long till)
        {
            int nRemoved = 0;
#if USE_GENERICS
            IEnumerator<TimeSeriesBlock> blockIterator = index.GetEnumerator(from - maxBlockTimeInterval, till);
#else
            IEnumerator blockIterator = index.GetEnumerator(from - maxBlockTimeInterval, till);
#endif
            while (blockIterator.MoveNext()) 
            {
                TimeSeriesBlock block = (TimeSeriesBlock)blockIterator.Current;
                int n = block.used;
                int l = 0, r = n;
                while (l < r)  
                {
                    int i = (l+r) >> 1;
                    if (from > block[i].Time) 
                    { 
                        l = i+1;
                    } 
                    else 
                    { 
                        r = i;
                    }
                }
                Debug.Assert(l == r && (l == n || block[l].Time >= from)); 
                while (r < n && block[r].Time <= till) 
                {
                    r += 1;
                    nRemoved += 1;
                }
                if (l == 0 && r == n) 
                { 
                    index.Remove(block.timestamp, block);
                    block.Deallocate();
#if USE_GENERICS
                    blockIterator = index.GetEnumerator(from - maxBlockTimeInterval, till);
#else
                    blockIterator.Reset();
#endif
                } 
                else if (l != r) 
                { 
                    if (l == 0) 
                    { 
                        index.Remove(block.timestamp, block);
                        block.timestamp = block[r].Time;
                        index.Put(block.timestamp, block);
#if USE_GENERICS
                        blockIterator = index.GetEnumerator(from - maxBlockTimeInterval, till);
#else
                        blockIterator.Reset();
#endif
                    }
                    Array.Copy(block.Ticks, r, block.Ticks, l, n-r);
                    block.used = l + n - r;
                    block.Modify();
                }
            }
            return nRemoved;        
        }

#if USE_GENERICS
        private void addNewBlock(T t)
        {
            TimeSeriesBlock block = new TimeSeriesBlock(blockSize);
#else
        private void addNewBlock(TimeSeriesTick t)
        {
            TimeSeriesBlock block = (TimeSeriesBlock)blockConstructor.Invoke(ClassDescriptor.noArgs);
            if (block == null) 
            {
                throw new StorageError(StorageError.ErrorCode.CONSTRUCTOR_FAILURE);
            }
#endif
            block.timestamp = t.Time;
            block.used = 1;
            block[0] = t;
            index.Put(block.timestamp, block);
        }

#if USE_GENERICS
        private void insertInBlock(TimeSeriesBlock block, T tick)
#else
        private void insertInBlock(TimeSeriesBlock block, TimeSeriesTick tick)
#endif
        {
            long t = tick.Time;
            int i, n = block.used;
        
            int l = 0, r = n;
            while (l < r)  
            {
                i = (l+r) >> 1;
                if (t >= block[i].Time) 
                { 
                    l = i+1;
                } 
                else 
                { 
                    r = i;
                }
            }
            Debug.Assert(l == r && (l == n || block[l].Time >= t));
            if (r == 0) 
            { 
                if (block[n-1].Time - t > maxBlockTimeInterval || n == block.Ticks.Length) 
                { 
                    addNewBlock(tick);
                    return;
                }
                if (block.timestamp != t) 
                { 
                    index.Remove(new Key(block.timestamp), block);                
                    block.timestamp = t;
                    index.Put(new Key(block.timestamp), block);
                }
            } 
            else if (r == n) 
            {
                if (t - block[0].Time > maxBlockTimeInterval || n == block.Ticks.Length) 
                { 
                    addNewBlock(tick);
                    return;
                } 
            }
            if (n == block.Ticks.Length) 
            { 
                addNewBlock(block[n-1]);
                Array.Copy(block.Ticks, r, block.Ticks, r+1, n-r-1);
            } 
            else 
            { 
                if (n != r) 
                { 
                    Array.Copy(block.Ticks, r, block.Ticks, r+1, n-r);
                }
                block.used += 1;
            }
            block[r] = tick;
            block.Modify();
        }

#if USE_GENERICS
        internal TimeSeriesImpl(Storage storage, int blockSize, long maxBlockTimeInterval) 
        {
            this.blockSize = blockSize;
            this.maxBlockTimeInterval = maxBlockTimeInterval;
            index = storage.CreateIndex<long,TimeSeriesBlock>(false);
        }
#else
        private void lookupConstructor(Type cls)
        {
            blockConstructor = cls.GetConstructor(BindingFlags.Instance|BindingFlags.Instance|BindingFlags.Public|BindingFlags.NonPublic|BindingFlags.DeclaredOnly, null, ClassDescriptor.defaultConstructorProfile, null);
            if (blockConstructor == null) 
            { 
                throw new StorageError(StorageError.ErrorCode.DESCRIPTOR_FAILURE, cls);
            }
        }

        internal TimeSeriesImpl(Storage storage, Type blockClass, long maxBlockTimeInterval) 
        {
            lookupConstructor(blockClass);
            this.maxBlockTimeInterval = maxBlockTimeInterval;
            blockClassName = ClassDescriptor.getTypeName(blockClass);
            index = storage.CreateIndex(typeof(long), false);
        }

        public override void OnLoad() 
        {  
            lookupConstructor(ClassDescriptor.lookup(Storage, blockClassName));
        }
#endif
        internal TimeSeriesImpl() {}
   
        public override void Deallocate() 
        {
            foreach (TimeSeriesBlock block in index) 
            {
                block.Deallocate();
            }
            index.Deallocate();
            base.Deallocate();
        }

#if USE_GENERICS
        public IEnumerable<T> Select(string predicate) 
        { 
            Query<T> query = new QueryImpl<T>(Storage);
            return query.Select(this, predicate);
        }
#endif


#if USE_GENERICS
        private Index<long,TimeSeriesBlock> index;
        private long                        maxBlockTimeInterval;        
        private int                         blockSize;
#else
        private Index  index;
        private long   maxBlockTimeInterval;        
        private string blockClassName;
        [NonSerialized()]
        private ConstructorInfo blockConstructor;
#endif
    }
}