When downloading the world state, persist children before parents (#3202)

* Fast sync should traverse the world state depth first

1. The pending requests queue in the world state downloader is now a different data structure. We now use a list of priority queues.
2. The node requests now know about the parent request that was responsible for spawning them.
3. When a node has data available and is asked to persist before its children are persisted, the node will not do anything. Instead, it will wait for all its children to persist first and will persist together with the last child.

Storing children before parents gives us the following benefits:
* We don't need to store pending requests on disk any more.
* When restarting download from a new pivot, we do not need to walk and check the whole tree any more.

And the following drawbacks:
* We now have pending nodes in memory for which we already downloaded data, but we do not store them in the database yet.

Overall expectations on performance:
We still need to download every single state node at least once, so there is no improvement there. We will save a significant amount of time in case we change pivots. And we save lots of read/writes on filesystem because tasks are not needed to be written to disk any more.

We want to avoid having too many pending unsaved nodes in memory, not to run out of it. If we were always handling only one request to our peers at the same time, we would not need to be worried, and we would just use a simple depth first search. Because we batch our requests, we might produce too many pending unprocessed nodes in memory if we are not careful about the order of processing requests. That is where the priority on node request comes from. We want to always process nodes lower in the tree before nodes higher in the tree, and preferably we want to first process children from the same parent so that we can save the current unsaved parent as soon as possible.

At the moment, I still left in the code several artefacts that I use for debugging the behaviour. I am planning to get rid of most of these counters, feel free to point them out in the review. There is for instance a weird counter in the NodeDataRequest class that I am using to monitor the total amount of unsaved nodes. In case pending unsaved node count rises too high, there is a warning printed into the logs. At the moment of writing, I would expect the counter to stay below 10 000 generally and not rise above 20 000 nodes. If you saw the number rise to for instance 100 000 that would signify a bug.

Similarly, because of the order of processing of the nodes, we do not need to store huge number of requests on the disk any more and the whole list fits comfortably into the memory. Without batching, we would not have more than a thousand requests around waiting. Because of the batching, we can see the number of requests occasionally rises all the way up to 300 000, but usually should be under 200 000.

Note that at any time there should not be more pending unsaved blocks than pending requests. Such a situation would be a bug to be reported.

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

* Addressing review comments

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

* Fixed failing test

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

* Improving test coverage

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

* Addressing review comments

Signed-off-by: Jiri Peinlich <jiri.peinlich@gmail.com>

services/tasks/src/main/java/org/hyperledger/besu/services/tasks/InMemoryTasksPriorityQueues.java

@@ -0,0 +1,171 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services.tasks;
+
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.HashSet;
+import java.util.List;
+import java.util.PriorityQueue;
+import java.util.Queue;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import org.jetbrains.annotations.NotNull;
+
+public class InMemoryTasksPriorityQueues<T extends TasksPriorityProvider>
+    implements TaskCollection<T> {
+  private final List<PriorityQueue<T>> internalQueues = new ArrayList<>(16);
+  private final Set<InMemoryTask<T>> unfinishedOutstandingTasks = new HashSet<>();
+  private final AtomicBoolean closed = new AtomicBoolean(false);
+
+  public InMemoryTasksPriorityQueues() {
+    clearInternalQueues();
+  }
+
+  private void clearInternalQueues() {
+    internalQueues.clear();
+    for (int i = 0; i < 16; i++) {
+      internalQueues.add(newEmptyQueue());
+    }
+  }
+
+  @NotNull
+  private PriorityQueue<T> newEmptyQueue() {
+    return new PriorityQueue<>(Comparator.comparingLong(TasksPriorityProvider::getPriority));
+  }
+
+  @Override
+  public synchronized void add(final T taskData) {
+    assertNotClosed();
+    var dequeue = findQueue(taskData.getDepth());
+    dequeue.add(taskData);
+  }
+
+  private PriorityQueue<T> findQueue(final int priority) {
+    while (priority + 1 > internalQueues.size()) {
+      internalQueues.add(newEmptyQueue());
+    }
+    return internalQueues.get(priority);
+  }
+
+  @Override
+  public synchronized Task<T> remove() {
+    assertNotClosed();
+    final Queue<T> lastNonEmptyQueue = findLastNonEmptyQueue();
+    if (lastNonEmptyQueue.isEmpty()) {
+      return null;
+    }
+    T data = lastNonEmptyQueue.remove();
+    InMemoryTask<T> task = new InMemoryTask<>(this, data);
+    unfinishedOutstandingTasks.add(task);
+    return task;
+  }
+
+  private Queue<T> findLastNonEmptyQueue() {
+    for (int i = internalQueues.size() - 1; i > 0; i--) {
+      final Queue<T> queue = internalQueues.get(i);
+      if (!queue.isEmpty()) {
+        return queue;
+      }
+    }
+    return internalQueues.get(0);
+  }
+
+  @Override
+  public synchronized long size() {
+    return internalQueues.stream().mapToInt(Queue::size).sum();
+  }
+
+  @Override
+  public synchronized boolean isEmpty() {
+    return findLastNonEmptyQueue().isEmpty();
+  }
+
+  @Override
+  public synchronized void clear() {
+    assertNotClosed();
+
+    unfinishedOutstandingTasks.clear();
+    clearInternalQueues();
+  }
+
+  @Override
+  public synchronized boolean allTasksCompleted() {
+    return isEmpty() && unfinishedOutstandingTasks.isEmpty();
+  }
+
+  @Override
+  public synchronized void close() {
+    if (closed.compareAndSet(false, true)) {
+      internalQueues.clear();
+      unfinishedOutstandingTasks.clear();
+    }
+  }
+
+  private void assertNotClosed() {
+    if (closed.get()) {
+      throw new IllegalStateException("Attempt to access closed " + getClass().getSimpleName());
+    }
+  }
+
+  private synchronized void handleFailedTask(final InMemoryTask<T> task) {
+    if (markTaskCompleted(task)) {
+      add(task.getData());
+    }
+  }
+
+  private synchronized boolean markTaskCompleted(final InMemoryTask<T> task) {
+    return unfinishedOutstandingTasks.remove(task);
+  }
+
+  public synchronized boolean contains(final T request) {
+    final PriorityQueue<T> queue = findQueue(request.getDepth());
+    return queue.contains(request)
+        || unfinishedOutstandingTasks.stream()
+            .map(InMemoryTask::getData)
+            .anyMatch(data -> data.equals(request));
+  }
+
+  private static class InMemoryTask<T extends TasksPriorityProvider> implements Task<T> {
+    private final T data;
+    private final InMemoryTasksPriorityQueues<T> queue;
+    private final AtomicBoolean completed = new AtomicBoolean(false);
+
+    public InMemoryTask(final InMemoryTasksPriorityQueues<T> queue, final T data) {
+      this.queue = queue;
+      this.data = data;
+    }
+
+    @Override
+    public T getData() {
+      return data;
+    }
+
+    @Override
+    public void markCompleted() {
+      if (completed.compareAndSet(false, true)) {
+        queue.markTaskCompleted(this);
+      }
+    }
+
+    @Override
+    public void markFailed() {
+      if (completed.compareAndSet(false, true)) {
+        queue.handleFailedTask(this);
+      }
+    }
+  }
+}

services/tasks/src/main/java/org/hyperledger/besu/services/tasks/TasksPriorityProvider.java

@@ -0,0 +1,24 @@
+/*
+ *
+ *  * Copyright Hyperledger Besu Contributors.
+ *  *
+ *  * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ *  * the License. You may obtain a copy of the License at
+ *  *
+ *  * http://www.apache.org/licenses/LICENSE-2.0
+ *  *
+ *  * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ *  * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ *  * specific language governing permissions and limitations under the License.
+ *  *
+ *  * SPDX-License-Identifier: Apache-2.0
+ *
+ */
+
+package org.hyperledger.besu.services.tasks;
+
+public interface TasksPriorityProvider {
+  long getPriority();
+
+  int getDepth();
+}

services/tasks/src/test/java/org/hyperledger/besu/services/tasks/InMemoryTasksPriorityQueuesTest.java

@@ -0,0 +1,175 @@
+/*
+ * Copyright ConsenSys AG.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+ * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ */
+package org.hyperledger.besu.services.tasks;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+
+import org.junit.Test;
+
+public class InMemoryTasksPriorityQueuesTest {
+
+  @Test
+  public void shouldBePossibleToAddElementsAndRetrieveThemInPriorityOrder() {
+    InMemoryTasksPriorityQueues<Item> queue = new InMemoryTasksPriorityQueues<>();
+
+    queue.add(item(1, 1));
+    queue.add(item(2, 30));
+    queue.add(item(2, 10));
+    queue.add(item(5, 1));
+    queue.add(item(99, Integer.MAX_VALUE));
+    queue.add(item(1, 20));
+
+    List<Item> items = new ArrayList<>();
+    while (!queue.isEmpty()) {
+      items.add(queue.remove().getData());
+    }
+
+    assertThat(items)
+        .containsExactly(
+            item(99, Integer.MAX_VALUE),
+            item(5, 1),
+            item(2, 10),
+            item(2, 30),
+            item(1, 1),
+            item(1, 20));
+  }
+
+  @Test
+  public void shouldNotInsertItemsToClosedQueue() {
+    InMemoryTasksPriorityQueues<Item> queue = new InMemoryTasksPriorityQueues<>();
+
+    queue.add(item(1, 1));
+
+    queue.close();
+
+    final Item item = item(2, 2);
+    assertThatThrownBy(() -> queue.add(item)).isInstanceOf(IllegalStateException.class);
+  }
+
+  @Test
+  public void shouldContainTaskUntilFinished() {
+    InMemoryTasksPriorityQueues<Item> queue = new InMemoryTasksPriorityQueues<>();
+    queue.add(item(1, 1));
+
+    final Item item = item(2, 3);
+    queue.add(item);
+
+    assertThat(queue.contains(item)).isTrue();
+
+    final Task<Item> removed = queue.remove();
+    assertThat(removed.getData()).isEqualTo(item);
+
+    assertThat(queue.contains(item)).isTrue();
+
+    removed.markCompleted();
+
+    assertThat(queue.contains(item)).isFalse();
+  }
+
+  @Test
+  public void shouldPutFailedItemBackIntoQueue() {
+    InMemoryTasksPriorityQueues<Item> queue = new InMemoryTasksPriorityQueues<>();
+    queue.add(item(1, 1));
+
+    final Item item = item(2, 3);
+    queue.add(item);
+
+    assertThat(queue.contains(item)).isTrue();
+
+    Task<Item> removed = queue.remove();
+    assertThat(removed.getData()).isEqualTo(item);
+
+    assertThat(queue.contains(item)).isTrue();
+
+    removed.markFailed();
+
+    assertThat(queue.contains(item)).isTrue();
+
+    removed = queue.remove();
+    assertThat(removed.getData()).isEqualTo(item);
+  }
+
+  @Test
+  public void shouldNotPutFailedItemBackIntoIfItWasCompletedAlreadyQueue() {
+    InMemoryTasksPriorityQueues<Item> queue = new InMemoryTasksPriorityQueues<>();
+    queue.add(item(1, 1));
+
+    final Item item = item(2, 3);
+    queue.add(item);
+
+    assertThat(queue.contains(item)).isTrue();
+
+    Task<Item> removed = queue.remove();
+    assertThat(removed.getData()).isEqualTo(item);
+
+    assertThat(queue.contains(item)).isTrue();
+
+    removed.markCompleted();
+    assertThat(queue.contains(item)).isFalse();
+
+    removed.markFailed();
+    assertThat(queue.contains(item)).isFalse();
+
+    removed = queue.remove();
+    assertThat(removed.getData()).isNotEqualTo(item);
+  }
+
+  private Item item(final int depth, final int priority) {
+    return new Item(depth, priority);
+  }
+
+  static class Item implements TasksPriorityProvider {
+    private final int depth;
+    private final long priority;
+
+    public Item(final int depth, final long priority) {
+      this.depth = depth;
+      this.priority = priority;
+    }
+
+    @Override
+    public long getPriority() {
+      return priority;
+    }
+
+    @Override
+    public int getDepth() {
+      return depth;
+    }
+
+    @Override
+    public boolean equals(final Object o) {
+      if (this == o) return true;
+      if (o == null || getClass() != o.getClass()) return false;
+      final Item item = (Item) o;
+      return depth == item.depth && priority == item.priority;
+    }
+
+    @Override
+    public int hashCode() {
+      return Objects.hash(depth, priority);
+    }
+
+    @Override
+    public String toString() {
+      return "Item{" + "depth=" + depth + ", priority=" + priority + '}';
+    }
+  }
+}