Parallel transaction execution

src/Neo/Ledger/Blockchain.cs

@@ -23,6 +23,8 @@
 using System.Collections.Immutable;
 using System.Diagnostics;
 using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 
 namespace Neo.Ledger
 {
@@ -414,26 +416,47 @@ private void Persist(Block block)
                     all_application_executed.Add(application_executed);
                     transactionStates = engine.GetState<TransactionState[]>();
                 }
-                DataCache clonedSnapshot = snapshot.CreateSnapshot();
-                // Warning: Do not write into variable snapshot directly. Write into variable clonedSnapshot and commit instead.
+
+                var tasks = new List<Task<(TransactionState, DataCache, ApplicationExecuted)>>();
+
+                // Warning: Do not write into variable snapshot directly.
+                // Write into variable clonedSnapshot and commit instead.
                 foreach (TransactionState transactionState in transactionStates)
                 {
-                    Transaction tx = transactionState.Transaction;
-                    using ApplicationEngine engine = ApplicationEngine.Create(TriggerType.Application, tx, clonedSnapshot, block, system.Settings, tx.SystemFee);
-                    engine.LoadScript(tx.Script);
-                    transactionState.State = engine.Execute();
-                    if (transactionState.State == VMState.HALT)
+                    DataCache txSnapshot = snapshot.CreateSnapshot();
+                    var task = OnExecuteTransactionAsync(system, block, txSnapshot, transactionState);
+                    tasks.Add(task);
+                }
+
+                // var readSet = new HashSet<StorageKey>();
+                var writeSet = new HashSet<StorageKey>();
+
+                var randomUsed = false;
+                foreach (var task in tasks)
+                {
+                    var (transactionState, txSnapshot, executed) = task.Result;
+
+                    if (txSnapshot.GetReadSet().Overlaps(writeSet) || (randomUsed && txSnapshot.isRandomNumberCalled))
                     {
-                        clonedSnapshot.Commit();
+                        DataCache tmp = snapshot.CreateSnapshot();
+                        var task2 = OnExecuteTransactionAsync(system, block, tmp, transactionState);
+                        tasks.Add(task2);
+                        (transactionState, txSnapshot, executed) = task.Result;
                     }
-                    else
+
+                    if (transactionState.State == VMState.HALT)
                     {
-                        clonedSnapshot = snapshot.CreateSnapshot();
+                        txSnapshot.Commit();
+                        writeSet.UnionWith(txSnapshot.GetWriteSet());
+                        randomUsed = txSnapshot.isRandomNumberCalled;
                     }
-                    ApplicationExecuted application_executed = new(engine);
-                    Context.System.EventStream.Publish(application_executed);
-                    all_application_executed.Add(application_executed);
+
+                    // ApplicationExecuted application_executed = new(engine);
+                    Context.System.EventStream.Publish(executed);
+                    all_application_executed.Add(executed);
+
                 }
+
                 using (ApplicationEngine engine = ApplicationEngine.Create(TriggerType.PostPersist, null, snapshot, block, system.Settings, 0))
                 {
                     engine.LoadScript(postPersistScript);
@@ -491,6 +514,23 @@ private static ImmutableHashSet<UInt160> UpdateExtensibleWitnessWhiteList(Protoc
             }
             return builder.ToImmutable();
         }
+
+        private static async Task<(TransactionState, DataCache, ApplicationExecuted)> OnExecuteTransactionAsync(NeoSystem system, Block block, DataCache clonedSnapshot, TransactionState transactionState)
+        {
+            ApplicationExecuted application_executed = null;
+
+            await Task.Run(() =>
+            {
+                Transaction tx = transactionState.Transaction;
+                using ApplicationEngine engine = ApplicationEngine.Create(TriggerType.Application, tx, clonedSnapshot, block, system.Settings, tx.SystemFee);
+                engine.LoadScript(tx.Script);
+                transactionState.State = engine.Execute();
+                application_executed = new(engine);
+            });
+
+            return (transactionState, clonedSnapshot, application_executed);
+
+        }
     }
 
     internal class BlockchainMailbox : PriorityMailbox

src/Neo/Persistence/DataCache.cs

@@ -45,6 +45,10 @@ public class Trackable
         private readonly Dictionary<StorageKey, Trackable> dictionary = new();
         private readonly HashSet<StorageKey> changeSet = new();
 
+        private readonly HashSet<StorageKey> readSet = new();
+
+        public bool isRandomNumberCalled = false;
+
         /// <summary>
         /// Reads a specified entry from the cache. If the entry is not in the cache, it will be automatically loaded from the underlying storage.
         /// </summary>
@@ -67,7 +71,7 @@ public StorageItem this[StorageKey key]
                         trackable = new Trackable
                         {
                             Key = key,
-                            Item = GetInternal(key),
+                            Item = GetInternalWrapper(key),
                             State = TrackState.None
                         };
                         dictionary.Add(key, trackable);
@@ -179,7 +183,7 @@ public void Delete(StorageKey key)
                 }
                 else
                 {
-                    StorageItem item = TryGetInternal(key);
+                    StorageItem item = TryGetInternalWrapper(key);
                     if (item == null) return;
                     dictionary.Add(key, new Trackable
                     {
@@ -273,6 +277,11 @@ public IEnumerable<Trackable> GetChangeSet()
             }
         }
 
+        public HashSet<StorageKey> GetWriteSet()
+        {
+            return changeSet;
+        }
+
         /// <summary>
         /// Determines whether the cache contains the specified entry.
         /// </summary>
@@ -284,7 +293,7 @@ public bool Contains(StorageKey key)
             {
                 if (dictionary.TryGetValue(key, out Trackable trackable))
                     return trackable.State != TrackState.Deleted && trackable.State != TrackState.NotFound;
-                return ContainsInternal(key);
+                return ContainsInternalWrapper(key);
             }
         }
 
@@ -295,13 +304,25 @@ public bool Contains(StorageKey key)
         /// <returns><see langword="true"/> if the underlying storage contains an entry with the specified key; otherwise, <see langword="false"/>.</returns>
         protected abstract bool ContainsInternal(StorageKey key);
 
+        private bool ContainsInternalWrapper(StorageKey key)
+        {
+            RecordRead(key);
+            return ContainsInternal(key);
+        }
+
         /// <summary>
         /// Reads a specified entry from the underlying storage.
         /// </summary>
         /// <param name="key">The key of the entry.</param>
         /// <returns>The data of the entry. Or <see langword="null"/> if the entry doesn't exist.</returns>
         protected abstract StorageItem GetInternal(StorageKey key);
 
+        private StorageItem GetInternalWrapper(StorageKey key)
+        {
+            RecordRead(key);
+            return GetInternal(key);
+        }
+
         /// <summary>
         /// Reads a specified entry from the cache, and mark it as <see cref="TrackState.Changed"/>. If the entry is not in the cache, it will be automatically loaded from the underlying storage.
         /// </summary>
@@ -339,7 +360,7 @@ public StorageItem GetAndChange(StorageKey key, Func<StorageItem> factory = null
                     trackable = new Trackable
                     {
                         Key = key,
-                        Item = TryGetInternal(key)
+                        Item = TryGetInternalWrapper(key)
                     };
                     if (trackable.Item == null)
                     {
@@ -389,7 +410,7 @@ public StorageItem GetOrAdd(StorageKey key, Func<StorageItem> factory)
                     trackable = new Trackable
                     {
                         Key = key,
-                        Item = TryGetInternal(key)
+                        Item = TryGetInternalWrapper(key)
                     };
                     if (trackable.Item == null)
                     {
@@ -472,6 +493,16 @@ public StorageItem GetOrAdd(StorageKey key, Func<StorageItem> factory)
         /// <returns>An enumerator containing all the entries after seeking.</returns>
         protected abstract IEnumerable<(StorageKey Key, StorageItem Value)> SeekInternal(byte[] keyOrPrefix, SeekDirection direction);
 
+
+        private IEnumerable<(StorageKey Key, StorageItem Value)> SeekInternalWrapper(byte[] keyOrPrefix, SeekDirection direction)
+        {
+            foreach (var (key, value) in SeekInternal(keyOrPrefix, direction))
+            {
+                RecordRead(key);
+                yield return (key, value);
+            }
+        }
+
         /// <summary>
         /// Reads a specified entry from the cache. If the entry is not in the cache, it will be automatically loaded from the underlying storage.
         /// </summary>
@@ -487,7 +518,7 @@ public StorageItem TryGet(StorageKey key)
                         return null;
                     return trackable.Item;
                 }
-                StorageItem value = TryGetInternal(key);
+                StorageItem value = TryGetInternalWrapper(key);
                 if (value == null) return null;
                 dictionary.Add(key, new Trackable
                 {
@@ -506,11 +537,33 @@ public StorageItem TryGet(StorageKey key)
         /// <returns>The data of the entry. Or <see langword="null"/> if it doesn't exist.</returns>
         protected abstract StorageItem TryGetInternal(StorageKey key);
 
+        private StorageItem TryGetInternalWrapper(StorageKey key)
+        {
+            RecordRead(key);
+            return TryGetInternal(key);
+        }
+
         /// <summary>
         /// Updates an entry in the underlying storage.
         /// </summary>
         /// <param name="key">The key of the entry.</param>
         /// <param name="value">The data of the entry.</param>
         protected abstract void UpdateInternal(StorageKey key, StorageItem value);
+
+        private void RecordRead(StorageKey key)
+        {
+            lock (dictionary)
+            {
+                readSet.Add(key);
+            }
+        }
+
+        public HashSet<StorageKey> GetReadSet()
+        {
+            lock (dictionary)
+            {
+                return readSet;
+            }
+        }
     }
 }

src/Neo/SmartContract/ApplicationEngine.Runtime.cs

@@ -316,6 +316,11 @@ protected internal BigInteger GetRandom()
                 price = 1 << 4;
             }
             AddGas(price * ExecFeeFactor);
+            if (Snapshot != null)
+            {
+                Snapshot.isRandomNumberCalled = true;
+            }
+
             return new BigInteger(buffer, isUnsigned: true);
         }