AKI-695: Use batch database operations in production code

modAionImpl/src/org/aion/zero/impl/db/AionRepositoryImpl.java

@@ -1213,7 +1213,7 @@ public List<byte[]> getReferencedStorageNodes(byte[] value, int limit, AionAddre
             byte[] subKey = h256(("details-storage/" + contract.toString()).getBytes());
 
             ByteArrayKeyValueStore db =
-                    new XorDataSource(selectDatabase(DatabaseType.STORAGE), subKey);
+                    new XorDataSource(selectDatabase(DatabaseType.STORAGE), subKey, false);
 
             Trie trie = new SecureTrie(db);
             Map<ByteArrayWrapper, byte[]> refs = trie.getReferencedTrieNodes(value, limit);
@@ -1273,7 +1273,8 @@ public TrieNodeResult importTrieNode(byte[] key, byte[] value, DatabaseType dbTy
             }
         }
 
-        db.put(key, value);
+        db.putToBatch(key, value);
+        db.commit();
         return TrieNodeResult.IMPORTED;
     }
 

modAionImpl/src/org/aion/zero/impl/db/AvmContractDetails.java

@@ -356,10 +356,11 @@ public byte[] getEncoded() {
     public void syncStorage() {
         byte[] graph = getObjectGraph();
         if (!Arrays.equals(graph, EMPTY_BYTE_ARRAY)) {
-            objectGraphSource.put(objectGraphHash, graph);
+            objectGraphSource.putToBatch(objectGraphHash, graph);
         }
-        objectGraphSource.put(computeAvmStorageHash(), RLP.encodeList(RLP.encodeElement(storageTrie.getRootHash()), RLP.encodeElement(objectGraphHash)));
-
+        objectGraphSource.putToBatch(computeAvmStorageHash(), RLP.encodeList(RLP.encodeElement(storageTrie.getRootHash()), RLP.encodeElement(objectGraphHash)));
+        // commit both updates together
+        objectGraphSource.commit();
         storageTrie.sync();
     }
 

modAionImpl/src/org/aion/zero/impl/db/DetailsDataStore.java

@@ -48,7 +48,8 @@ public DetailsDataStore(
      * @return the external storage data source associated with the given contract address
      */
     private ByteArrayKeyValueStore createStorageSource(AionAddress address) {
-        return new XorDataSource(storageDSPrune, h256(("details-storage/" + address.toString()).getBytes()));
+        // NOTE: The consensus-correct Trie use for contracts requires not pushing deletions via the XorDataSource.
+        return new XorDataSource(storageDSPrune, h256(("details-storage/" + address.toString()).getBytes()), false);
     }
 
     /**
@@ -59,7 +60,7 @@ private ByteArrayKeyValueStore createStorageSource(AionAddress address) {
      * @return the object graph data source associated with the given contract address
      */
     private ByteArrayKeyValueStore createGraphSource(AionAddress address) {
-        return new XorDataSource(graphSrc, h256(("details-graph/" + address.toString()).getBytes()));
+        return new XorDataSource(graphSrc, h256(("details-graph/" + address.toString()).getBytes()), true);
     }
 
     /**
@@ -113,15 +114,11 @@ public StoredContractDetails newContractDetails(AionAddress address, InternalVmT
     public synchronized void update(AionAddress key, StoredContractDetails contractDetails) {
         // Put into cache.
         byte[] rawDetails = contractDetails.getEncoded();
-        detailsSrc.put(key.toByteArray(), rawDetails);
-
+        detailsSrc.putToBatch(key.toByteArray(), rawDetails);
+        detailsSrc.commit(); // TODO AKI-309: flush in bulk by the repository
         contractDetails.syncStorage();
     }
 
-    public synchronized void remove(byte[] key) {
-        detailsSrc.delete(key);
-    }
-
     public JournalPruneDataSource getStorageDSPrune() {
         return storageDSPrune;
     }

modAionImpl/src/org/aion/zero/impl/db/PendingBlockStore.java

@@ -276,7 +276,7 @@ public int addBlockRange(List<Block> blocks, Logger log) {
             int stored = addBlockRange(first, blockRange);
 
             // save data to disk
-            indexSource.commitBatch();
+            indexSource.commit();
             levelSource.commit();
             queueSource.commit();
 
@@ -543,7 +543,7 @@ public void dropPendingQueues(long level, Collection<ByteArrayWrapper> queues, M
             }
 
             // push changed to disk
-            indexSource.commitBatch();
+            indexSource.commit();
             queueSource.commit();
             levelSource.commit();
 

modAionImpl/src/org/aion/zero/impl/trie/Cache.java

@@ -84,15 +84,6 @@ public synchronized Value get(byte[] key) {
         return null;
     }
 
-    public synchronized void delete(byte[] key) {
-        ByteArrayWrapper wrappedKey = wrap(key);
-        this.nodes.remove(wrappedKey);
-
-        if (dataSource != null) {
-            this.dataSource.delete(key);
-        }
-    }
-
     public synchronized void commit(boolean flushCache) {
         // Don't try to commit if it isn't dirty
         if ((dataSource == null) || !this.isDirty) {

modAionImpl/src/org/aion/zero/impl/trie/TrieImpl.java

@@ -1005,6 +1005,7 @@ private void appendHashes(byte[] bytes, ArrayList<byte[]> hashes) {
     public long saveFullStateToDatabase(byte[] stateRoot, ByteArrayKeyValueDatabase db) {
         ExtractToDatabase traceAction = new ExtractToDatabase(db);
         traceTrie(stateRoot, traceAction);
+        db.commit();
         return traceAction.count;
     }
 
@@ -1024,6 +1025,7 @@ private void traceDiffTrie(byte[] stateRoot, ScanAction action, ByteArrayKeyValu
     public long saveDiffStateToDatabase(byte[] stateRoot, ByteArrayKeyValueDatabase db) {
         ExtractToDatabase traceAction = new ExtractToDatabase(db);
         traceDiffTrie(stateRoot, traceAction, db);
+        db.commit();
         return traceAction.count;
     }
 }

modAionImpl/src/org/aion/zero/impl/trie/scan/ExtractToDatabase.java

@@ -17,7 +17,7 @@ public ExtractToDatabase(ByteArrayKeyValueDatabase _db) {
 
     @Override
     public void doOnNode(byte[] hash, Value node) {
-        db.put(hash, dummy_value);
+        db.putToBatch(hash, dummy_value);
         count++;
     }
 }

modDbImpl/src/org/aion/db/generic/LockedDatabase.java

@@ -67,21 +67,6 @@ public void close() {
         }
     }
 
-    @Override
-    public boolean commit() {
-        // acquire write lock
-        lock.writeLock().lock();
-
-        try {
-            return database.commit();
-        } catch (Exception e) {
-            throw e;
-        } finally {
-            // releasing write lock
-            lock.writeLock().unlock();
-        }
-    }
-
     @Override
     public void compact() {
         // acquire write lock
@@ -316,12 +301,12 @@ public void deleteInBatch(byte[] key) {
     }
 
     @Override
-    public void commitBatch() {
+    public void commit() {
         // acquire write lock
         lock.writeLock().lock();
 
         try {
-            database.commitBatch();
+            database.commit();
         } catch (Exception e) {
             if (e instanceof RuntimeException) {
                 throw e;

modDbImpl/src/org/aion/db/generic/TimedDatabase.java

@@ -52,16 +52,6 @@ public void close() {
         LOG.debug(database.toString() + " close() in " + (t2 - t1) + " ns.");
     }
 
-    @Override
-    public boolean commit() {
-        long t1 = System.nanoTime();
-        boolean cmt = database.commit();
-        long t2 = System.nanoTime();
-
-        LOG.debug(database.toString() + " commit() in " + (t2 - t1) + " ns.");
-        return cmt;
-    }
-
     @Override
     public void compact() {
         long t1 = System.nanoTime();
@@ -251,12 +241,12 @@ public void deleteInBatch(byte[] key) {
     }
 
     @Override
-    public void commitBatch() {
+    public void commit() {
         long t1 = System.nanoTime();
-        database.commitBatch();
+        database.commit();
         long t2 = System.nanoTime();
 
-        LOG.debug(database.toString() + " commitBatch() in " + (t2 - t1) + " ns.");
+        LOG.debug(database.toString() + " commit() in " + (t2 - t1) + " ns.");
     }
 
     @Override

modDbImpl/src/org/aion/db/impl/AbstractDB.java

@@ -57,14 +57,6 @@ protected String propertiesInfo() {
                 + ">"; //
     }
 
-    @Override
-    public boolean commit() {
-        // not implemented since we always commit the changes to the database for this
-        // implementation
-        throw new UnsupportedOperationException(
-                "Only automatic commits are supported by " + this.toString());
-    }
-
     @Override
     public void compact() {
         LOG.warn("Compact not supported by " + this.toString() + ".");
@@ -145,9 +137,6 @@ public boolean isLocked() {
         return false;
     }
 
-    /** Functionality for directly interacting with the heap cache. */
-    public abstract boolean commitCache(Map<ByteArrayWrapper, byte[]> cache);
-
     @Override
     public Optional<byte[]> get(byte[] key) {
         check(key);

modDbImpl/src/org/aion/db/impl/Database.java

@@ -44,18 +44,6 @@ public interface Database {
      */
     void close();
 
-    /**
-     * Makes all changes made since the previous commit/rollback permanent and releases any database
-     * locks currently held by this Connection object. This method should be used only when
-     * auto-commit mode has been disabled.
-     *
-     * @return {@code true} if the changes were successfully committed to storage, {@code false} if
-     *     the changes could not be committed to storage
-     * @throws RuntimeException if the data store is closed
-     * @implNote Returns {@code true} with no other effect when auto-commit is already enabled.
-     */
-    boolean commit();
-
     /** Reduce the size of the database when possible. */
     void compact();
 

modDbImpl/src/org/aion/db/impl/KeyValueStore.java

@@ -119,8 +119,11 @@ public interface KeyValueStore<KeyT, ValueT> extends AutoCloseable {
     /**
      * Pushes updates made using {@link #putToBatch(Object, Object)} and {@link
      * #deleteInBatch(Object)} to the underlying data source.
+     * All changes since the last commit will be made permanent.
+     *
+     * @throws RuntimeException if the underlying data store is closed
      */
-    void commitBatch();
+    void commit();
 
     /**
      * Puts or updates the data store with the given <i>key-value</i> pairs, as follows:

modDbImpl/src/org/aion/db/impl/h2/H2MVMap.java

@@ -246,7 +246,9 @@ public void deleteInBatchInternal(byte[] key) {
     }
 
     @Override
-    public void commitBatch() {
+    public void commit() {
+        check();
+
         // nothing to do since batch operations are not supported
     }
 
@@ -296,32 +298,6 @@ public void deleteBatchInternal(Collection<byte[]> keys) {
         }
     }
 
-    // AbstractDB functionality
-    // ----------------------------------------------------------------------------------------
-
-    public boolean commitCache(Map<ByteArrayWrapper, byte[]> cache) {
-        boolean success = false;
-
-        try {
-            check();
-
-            // doesn't actually have functionality for batch operations
-            for (Entry<ByteArrayWrapper, byte[]> e : cache.entrySet()) {
-                if (e.getValue() == null) {
-                    map.remove(e.getKey().toBytes());
-                } else {
-                    map.put(e.getKey().toBytes(), e.getValue());
-                }
-            }
-
-            success = true;
-        } catch (Exception e) {
-            LOG.error("Unable to commit heap cache to " + this.toString() + ".", e);
-        }
-
-        return success;
-    }
-
     /**
      * Compact the database file, that is, compact blocks that have a low fill rate, and move chunks
      * next to each other. This will typically shrink the database file. Changes are flushed to the

modDbImpl/src/org/aion/db/impl/leveldb/LevelDB.java

@@ -368,7 +368,9 @@ public void deleteInBatchInternal(byte[] key) {
     }
 
     @Override
-    public void commitBatch() {
+    public void commit() {
+        check();
+
         if (batch != null) {
             try {
                 db.write(batch);
@@ -426,36 +428,4 @@ public void deleteBatchInternal(Collection<byte[]> keys) {
             LOG.error("Unable to close WriteBatch object in " + this.toString() + ".", e);
         }
     }
-
-    // AbstractDB functionality
-    // ----------------------------------------------------------------------------------------
-
-    public boolean commitCache(Map<ByteArrayWrapper, byte[]> cache) {
-        boolean success = false;
-
-        check();
-
-        // try-with-resources will automatically close the batch object
-        try (WriteBatch batch = db.createWriteBatch()) {
-            // add put and delete operations to batch
-            for (Map.Entry<ByteArrayWrapper, byte[]> e : cache.entrySet()) {
-                if (e.getValue() == null) {
-                    batch.delete(e.getKey().toBytes());
-                } else {
-                    batch.put(e.getKey().toBytes(), e.getValue());
-                }
-            }
-
-            // bulk atomic update
-            db.write(batch);
-
-            success = true;
-        } catch (DBException e) {
-            LOG.error("Unable to commit heap cache to " + this.toString() + ".", e);
-        } catch (IOException e) {
-            LOG.error("Unable to close WriteBatch object in " + this.toString() + ".", e);
-        }
-
-        return success;
-    }
 }

modDbImpl/src/org/aion/db/impl/mockdb/MockDB.java

@@ -120,7 +120,9 @@ public void deleteInBatchInternal(byte[] key) {
     }
 
     @Override
-    public void commitBatch() {
+    public void commit() {
+        check();
+
         // nothing to do since batch operations are not supported
     }
 
@@ -155,28 +157,4 @@ public void deleteBatchInternal(Collection<byte[]> keys) {
     public void drop() {
         kv.clear();
     }
-
-    public boolean commitCache(Map<ByteArrayWrapper, byte[]> cache) {
-        boolean success = false;
-
-        try {
-            check();
-
-            // simply do a put, because setting a kv pair to null is same as delete
-            cache.forEach(
-                    (key, value) -> {
-                        if (value == null) {
-                            kv.remove(key);
-                        } else {
-                            kv.put(key, value);
-                        }
-                    });
-
-            success = true;
-        } catch (Exception e) {
-            LOG.error("Unable to commit heap cache to " + this.toString() + ".", e);
-        }
-
-        return success;
-    }
 }