Oracles

src/neo/Ledger/Blockchain.cs

@@ -62,7 +62,7 @@ public class RelayResult { public IInventory Inventory; public VerifyResult Resu
         private uint stored_header_count = 0;
         private readonly Dictionary<UInt256, Block> block_cache = new Dictionary<UInt256, Block>();
         private readonly Dictionary<uint, LinkedList<Block>> block_cache_unverified = new Dictionary<uint, LinkedList<Block>>();
-        internal readonly RelayCache RelayCache = new RelayCache(100);
+        internal readonly RelayCache RelayCache = new RelayCache(200);
         private SnapshotView currentSnapshot;
 
         public IStore Store { get; }

src/neo/Ledger/MemoryPool.cs

@@ -27,7 +27,6 @@ public class MemoryPool : IReadOnlyCollection<Transaction>
 
         private readonly NeoSystem _system;
 
-        //
         /// <summary>
         /// Guarantees consistency of the pool data structures.
         ///
@@ -174,7 +173,7 @@ public IEnumerable<Transaction> GetVerifiedTransactions()
             _txRwLock.EnterReadLock();
             try
             {
-                return _unsortedTransactions.Select(p => p.Value.Tx).ToArray();
+                return GetTransactions(_unsortedTransactions.Values);
             }
             finally
             {
@@ -202,14 +201,36 @@ public IEnumerable<Transaction> GetSortedVerifiedTransactions()
             _txRwLock.EnterReadLock();
             try
             {
-                return _sortedTransactions.Reverse().Select(p => p.Tx).ToArray();
+                return GetTransactions(_sortedTransactions);
             }
             finally
             {
                 _txRwLock.ExitReadLock();
             }
         }
 
+        private IEnumerable<Transaction> GetTransactions(IEnumerable<PoolItem> set)
+        {
+            var state = new PoolItem.DelayState();
+            var ret = set
+                .Reverse()
+                .Where(u => u.IsReady(state))
+                .Select(p => p.Tx)
+                .ToArray();
+
+            // Fetch transactions
+
+            foreach (var tx in ret)
+            {
+                yield return tx;
+            }
+            foreach (var delayed in state.Delayed)
+            {
+                if (state.Allowed.Contains(delayed.Hash))
+                    yield return delayed;
+            }
+        }
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         private PoolItem GetLowestFeeTransaction(SortedSet<PoolItem> verifiedTxSorted,
             SortedSet<PoolItem> unverifiedTxSorted, out SortedSet<PoolItem> sortedPool)
@@ -239,7 +260,7 @@ private PoolItem GetLowestFeeTransaction(out Dictionary<UInt256, PoolItem> unsor
             }
             finally
             {
-                unsortedTxPool = Object.ReferenceEquals(sortedPool, _unverifiedSortedTransactions)
+                unsortedTxPool = ReferenceEquals(sortedPool, _unverifiedSortedTransactions)
                    ? _unverifiedTransactions : _unsortedTransactions;
             }
         }
@@ -447,6 +468,7 @@ internal void InvalidateAllTransactions()
                 {
                     if (_unsortedTransactions.TryAdd(item.Tx.Hash, item))
                     {
+                        item.CheckOracleResponse(snapshot);
                         verifiedSortedTxPool.Add(item);
 
                         if (item.LastBroadcastTimestamp < rebroadcastCutOffTime)

src/neo/Ledger/PoolItem.cs

@@ -1,5 +1,8 @@
 using Neo.Network.P2P.Payloads;
+using Neo.Persistence;
+using Neo.SmartContract.Native;
 using System;
+using System.Collections.Generic;
 
 namespace Neo.Ledger
 {
@@ -26,11 +29,69 @@ internal class PoolItem : IComparable<PoolItem>
         /// </summary>
         public DateTime LastBroadcastTimestamp;
 
+        /// <summary>
+        /// check if oracle response was already found in storage
+        /// </summary>
+        private bool _responseExists;
+
         internal PoolItem(Transaction tx)
         {
             Tx = tx;
             Timestamp = TimeProvider.Current.UtcNow;
             LastBroadcastTimestamp = Timestamp;
+            _responseExists = false;
+        }
+
+        internal class DelayState
+        {
+            public HashSet<UInt256> Allowed = new HashSet<UInt256>();
+            public HashSet<Transaction> Delayed = new HashSet<Transaction>();
+        }
+
+        public void CheckOracleResponse(StoreView snapshot)
+        {
+            if (!_responseExists &&
+                Tx.IsOracleRequest() &&
+                NativeContract.Oracle.ContainsResponse(snapshot, Tx.Hash))
+            {
+                _responseExists = true;
+            }
+        }
+
+        public bool IsReady(DelayState state)
+        {
+            if (Tx.IsOracleRequest())
+            {
+                if (state.Allowed.Remove(Tx.Hash))
+                {
+                    // The response was already fetched, we can put request and response in the same block
+
+                    return true;
+                }
+                else
+                {
+                    if (_responseExists)
+                    {
+                        // The response it's waiting to be consumed (block+n)
+
+                        return true;
+                    }
+                    else
+                    {
+                        // If the response it's in the pool it's located after the request
+                        // We save the request in order to put after the response
+
+                        state.Delayed.Add(Tx);
+                        return false;
+                    }
+                }
+            }
+            else if (Tx.IsOracleResponse(out var oracleRequestTx))
+            {
+                state.Allowed.Add(oracleRequestTx);
+            }
+
+            return true;
         }
 
         public int CompareTo(Transaction otherTx)

src/neo/Ledger/SortedBlockingCollection.cs

@@ -0,0 +1,84 @@
+using System.Collections.Concurrent;
+using System.Collections.Generic;
+using System.Threading;
+
+namespace Neo.Ledger
+{
+    public class SortedBlockingCollection<TKey, TValue>
+    {
+        /// <summary>
+        /// _oracleTasks will consume from this pool
+        /// </summary>
+        private readonly BlockingCollection<TValue> _asyncPool = new BlockingCollection<TValue>();
+
+        /// <summary>
+        /// Queue
+        /// </summary>
+        private readonly SortedConcurrentDictionary<TKey, TValue> _queue;
+
+        /// <summary>
+        /// Constructor
+        /// </summary>
+        /// <param name="comparer">Comparer</param>
+        /// <param name="capacity">Capacity</param>
+        public SortedBlockingCollection(IComparer<KeyValuePair<TKey, TValue>> comparer, int capacity)
+        {
+            _queue = new SortedConcurrentDictionary<TKey, TValue>(comparer, capacity);
+        }
+
+        /// <summary>
+        /// Add entry
+        /// </summary>
+        /// <param name="key">Key</param>
+        /// <param name="value">Value</param>
+        public void Add(TKey key, TValue value)
+        {
+            if (_queue.TryAdd(key, value) && _asyncPool.Count <= 0)
+            {
+                Pop();
+            }
+        }
+
+        /// <summary>
+        /// Clear
+        /// </summary>
+        public void Clear()
+        {
+            _queue.Clear();
+
+            while (_asyncPool.Count > 0)
+            {
+                _asyncPool.TryTake(out _);
+            }
+        }
+
+        /// <summary>
+        /// Get consuming enumerable
+        /// </summary>
+        /// <param name="token">Token</param>
+        public IEnumerable<TValue> GetConsumingEnumerable(CancellationToken token)
+        {
+            foreach (var entry in _asyncPool.GetConsumingEnumerable(token))
+            {
+                // Prepare other item in _asyncPool
+
+                Pop();
+
+                // Iterate items
+
+                yield return entry;
+            }
+        }
+
+        /// <summary>
+        /// Move one item from the sorted queue to _asyncPool, this will ensure that the threads process the entries according to the priority
+        /// </summary>
+        private void Pop()
+        {
+            if (_queue.TryPop(out var entry))
+            {
+                _asyncPool.Add(entry);
+            }
+        }
+    }
+}