change BlockBuffer's buffer to ConcurrentQueue

Discreet/DB/BlockBuffer.cs

@@ -22,9 +22,7 @@ namespace Discreet.DB
     /// </summary>
     public class BlockBuffer : IView
     {
-        private Channel<Block> _buffer;
-
-        public ChannelWriter<Block> Writer { get => _buffer.Writer; }
+        private ConcurrentQueue<Block> _buffer;
 
         private static BlockBuffer _instance;
 
@@ -244,7 +242,7 @@ public TXOutput[] GetMixins(uint[] idxs)
 
         public BlockBuffer()
         {
-            _buffer = Channel.CreateUnbounded<Block>();
+            _buffer = new ConcurrentQueue<Block>();
             _view = DataView.GetView();
         }
 
@@ -257,9 +255,24 @@ public long GetChainHeight()
             }
         }
 
-        public async Task ForceFlush()
+        public void WriteToBuffer(Block blk)
+        {
+            _buffer.Enqueue(blk);
+            UpdateBuffers(blk);
+        }
+
+        public void ForceFlush()
         {
-            await _buffer.Writer.WriteAsync(_signaler);
+            lock (buffer)
+            {
+                if (buffer.Count == 0)
+                {
+                    return;
+                }
+
+                Flush(buffer);
+                buffer.Clear();
+            }
         }
 
         /// <summary>
@@ -270,49 +283,23 @@ public async Task Start()
         {
             _pIndex = DataView.GetView().GetOutputIndex();
 
-            _ = Task.Factory.StartNew(async () =>
-            {
-                var timer = new PeriodicTimer(_flushInterval);
-                while (await timer.WaitForNextTickAsync())
-                {
-                    await _buffer.Writer.WriteAsync(_signaler);
-                }
-            });
-
-            await foreach(var block in _buffer.Reader.ReadAllAsync())
+            var timer = new PeriodicTimer(_flushInterval);
+            while (await timer.WaitForNextTickAsync())
             {
-                if (block == _signaler)
+                lock (buffer)
                 {
-                    lock (buffer)
+                    while (_buffer.TryDequeue(out var block))
                     {
-                        if (buffer.Count == 0)
-                        {
-                            continue;
-                        }
+                        buffer.Add(block);
                     }
 
-                    Flush(buffer);
-
-                    lock (buffer)
+                    if (buffer.Count == 0)
                     {
-                        buffer.Clear();
+                        continue;
                     }
-                }
-                else
-                {
-                    if (_flushEveryBlock)
-                    {
-                        Flush(new List<Block> { block });
-                    }
-                    else
-                    {
-                        lock (buffer)
-                        {
-                            buffer.Add(block);
-                        }
 
-                        UpdateBuffers(block);
-                    }
+                    Flush(buffer);
+                    buffer.Clear();
                 }
             }
         }
@@ -325,7 +312,7 @@ public async Task Start()
 
         public void ClearBlockCache() => _view.ClearBlockCache();
 
-        public void AddBlock(Block blk) => _buffer.Writer.TryWrite(blk);
+        public void AddBlock(Block blk) => _buffer.Enqueue(blk);
 
         public Dictionary<long, Block> GetBlockCache() => _view.GetBlockCache();
 

Discreet/DB/ValidationCache.cs

@@ -626,20 +626,20 @@ public async Task Flush(List<Block> goodBlocks = null, bool force = false)
                 {
                     foreach (var blk in goodBlocks)
                     {
-                        await BlockBuffer.Instance.Writer.WriteAsync(blk);
+                        BlockBuffer.Instance.WriteToBuffer(blk);
                     }
                 }
                 else
                 {
                     foreach (var blk in blocks)
                     {
-                        await BlockBuffer.Instance.Writer.WriteAsync(blk);
+                        BlockBuffer.Instance.WriteToBuffer(blk);
                     }
                 }
             }
             else
             {
-                await BlockBuffer.Instance.Writer.WriteAsync(block);
+                BlockBuffer.Instance.WriteToBuffer(block);
             }
 
             if (blocks != null)
@@ -664,7 +664,7 @@ public async Task Flush(List<Block> goodBlocks = null, bool force = false)
                 Daemon.TXPool.GetTXPool().UpdatePool(block.Transactions);
             }
 
-            if (force) await BlockBuffer.Instance.ForceFlush();
+            if (force) BlockBuffer.Instance.ForceFlush();
         }
 
         public static async Task Process(Block block)

Discreet/Entry.cs

@@ -28,6 +28,7 @@
 using Discreet.Common.Converters;
 using System.Text.Json.Serialization;
 using Discreet.Wallets;
+using System.Collections.Concurrent;
 
 namespace Discreet
 {

Discreet/Network/MessageCache.cs

@@ -50,7 +50,7 @@ public MessageCache()
             Versions = new ConcurrentDictionary<IPEndPoint, Core.Packets.Peerbloom.VersionPacket>();
             BadVersions = new ConcurrentDictionary<IPEndPoint, Core.Packets.Peerbloom.VersionPacket>();
             BlockCache = new ConcurrentDictionary<long, Block>();
-            OrphanBlocks = new ConcurrentDictionary<Cipher.SHA256, Block>();
+            OrphanBlocks = new ConcurrentDictionary<Cipher.SHA256, Block>(new Cipher.SHA256EqualityComparer());
             HeaderCache = new ConcurrentDictionary<long, BlockHeader>();
         }