[WIP] Attestation pool take 1: interop

chain/src/main/java/org/ethereum/beacon/chain/pool/InMemoryAttestationPool.java

@@ -1,12 +1,15 @@
 package org.ethereum.beacon.chain.pool;
 
+import java.util.List;
 import org.ethereum.beacon.chain.pool.checker.SanityChecker;
 import org.ethereum.beacon.chain.pool.checker.SignatureEncodingChecker;
 import org.ethereum.beacon.chain.pool.checker.TimeFrameFilter;
 import org.ethereum.beacon.chain.pool.churn.OffChainAggregates;
 import org.ethereum.beacon.chain.pool.reactor.ChurnProcessor;
+import org.ethereum.beacon.chain.pool.reactor.DoubleWorkProcessor;
 import org.ethereum.beacon.chain.pool.reactor.IdentificationProcessor;
 import org.ethereum.beacon.chain.pool.reactor.SanityProcessor;
+import org.ethereum.beacon.chain.pool.reactor.SignatureEncodingProcessor;
 import org.ethereum.beacon.chain.pool.reactor.TimeProcessor;
 import org.ethereum.beacon.chain.pool.reactor.VerificationProcessor;
 import org.ethereum.beacon.chain.pool.registry.ProcessedAttestations;
@@ -15,13 +18,11 @@
 import org.ethereum.beacon.core.BeaconBlock;
 import org.ethereum.beacon.core.state.Checkpoint;
 import org.ethereum.beacon.core.types.SlotNumber;
-import org.ethereum.beacon.schedulers.Scheduler;
 import org.ethereum.beacon.schedulers.Schedulers;
-import org.ethereum.beacon.stream.Fluxes;
-import org.ethereum.beacon.stream.Fluxes.FluxSplit;
 import org.reactivestreams.Publisher;
 import reactor.core.publisher.DirectProcessor;
 import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Scheduler;
 
 /**
  * An implementation of attestation pool based on <a href="https://projectreactor.io/">Reactor</a>
@@ -36,16 +37,17 @@ public class InMemoryAttestationPool implements AttestationPool {
   private final Publisher<BeaconBlock> chainHeads;
   private final Schedulers schedulers;
 
-  private final TimeProcessor timeProcessor;
-  private final SanityProcessor sanityChecker;
+  private final TimeFrameFilter timeFrameFilter;
+  private final SanityChecker sanityChecker;
   private final SignatureEncodingChecker encodingChecker;
   private final ProcessedAttestations processedFilter;
-  private final IdentificationProcessor identifier;
-  private final VerificationProcessor verifier;
-  private final ChurnProcessor churn;
+  private final UnknownAttestationPool unknownPool;
+  private final BatchVerifier verifier;
 
   private final DirectProcessor<ReceivedAttestation> invalidAttestations = DirectProcessor.create();
   private final DirectProcessor<ReceivedAttestation> validAttestations = DirectProcessor.create();
+  private final DirectProcessor<ReceivedAttestation> unknownAttestations = DirectProcessor.create();
+  private final DirectProcessor<OffChainAggregates> offChainAggregates = DirectProcessor.create();
 
   public InMemoryAttestationPool(
       Publisher<ReceivedAttestation> source,
@@ -58,82 +60,85 @@ public InMemoryAttestationPool(
       SanityChecker sanityChecker,
       SignatureEncodingChecker encodingChecker,
       ProcessedAttestations processedFilter,
-      UnknownAttestationPool unknownAttestationPool,
+      UnknownAttestationPool unknownPool,
       BatchVerifier batchVerifier) {
     this.source = source;
     this.newSlots = newSlots;
     this.finalizedCheckpoints = finalizedCheckpoints;
     this.importedBlocks = importedBlocks;
     this.chainHeads = chainHeads;
     this.schedulers = schedulers;
-    this.timeProcessor = new TimeProcessor(timeFrameFilter);
-    this.sanityChecker = new SanityProcessor(sanityChecker);
+    this.timeFrameFilter = timeFrameFilter;
+    this.sanityChecker = sanityChecker;
     this.encodingChecker = encodingChecker;
     this.processedFilter = processedFilter;
-    this.identifier = new IdentificationProcessor(unknownAttestationPool);
-    this.verifier = new VerificationProcessor(batchVerifier);
-    this.churn = new ChurnProcessor();
+    this.unknownPool = unknownPool;
+    this.verifier = batchVerifier;
   }
 
   @Override
   public void start() {
-    Scheduler executor =
-        schedulers.newParallelDaemon("attestation-pool-%d", AttestationPool.MAX_THREADS);
-
-    Flux<?> sourceFx = Flux.from(source).publishOn(executor.toReactor());
-    Flux<?> newSlotsFx = Flux.from(newSlots).publishOn(executor.toReactor());
-    Flux<?> importedBlocksFx = Flux.from(importedBlocks).publishOn(executor.toReactor());
-    Flux<?> finalizedCheckpointsFx =
-        Flux.from(finalizedCheckpoints).publishOn(executor.toReactor());
-    Flux<?> chainHeadsFx = Flux.from(chainHeads).publishOn(executor.toReactor());
-
-    // start from time frame processor
-    Flux.merge(sourceFx, newSlotsFx, finalizedCheckpointsFx).subscribe(timeProcessor);
-    FluxSplit<CheckedAttestation> timeFrameOut =
-        Fluxes.split(timeProcessor, CheckedAttestation::isPassed);
-
-    // subscribe sanity checker
-    Flux.merge(
-            timeFrameOut.getSatisfied().map(CheckedAttestation::getAttestation),
-            finalizedCheckpointsFx)
-        .subscribe(sanityChecker);
-    FluxSplit<CheckedAttestation> sanityOut =
-        Fluxes.split(sanityChecker, CheckedAttestation::isPassed);
-
-    // filter already processed attestations
-    Flux<ReceivedAttestation> newAttestations =
-        sanityOut
-            .getSatisfied()
-            .map(CheckedAttestation::getAttestation)
-            .filter(processedFilter::add);
+    Scheduler parallelExecutor =
+        schedulers
+            .newParallelDaemon("attestation-pool-%d", AttestationPool.MAX_THREADS)
+            .toReactor();
+
+    // create sources
+    Flux<ReceivedAttestation> sourceFx = Flux.from(source);
+    Flux<SlotNumber> newSlotsFx = Flux.from(newSlots);
+    Flux<Checkpoint> finalizedCheckpointsFx = Flux.from(finalizedCheckpoints);
+    Flux<BeaconBlock> importedBlocksFx = Flux.from(importedBlocks);
+    Flux<BeaconBlock> chainHeadsFx = Flux.from(chainHeads);
+
+    // check time frames
+    TimeProcessor timeProcessor =
+        new TimeProcessor(
+            timeFrameFilter, schedulers, sourceFx, finalizedCheckpointsFx, newSlotsFx);
+
+    // run sanity check
+    SanityProcessor sanityProcessor =
+        new SanityProcessor(sanityChecker, schedulers, timeProcessor, finalizedCheckpointsFx);
+
+    // discard already processed attestations
+    DoubleWorkProcessor doubleWorkProcessor =
+        new DoubleWorkProcessor(processedFilter, schedulers, sanityProcessor.getValid());
 
     // check signature encoding
-    FluxSplit<ReceivedAttestation> encodingCheckOut =
-        Fluxes.split(newAttestations, encodingChecker::check);
+    SignatureEncodingProcessor encodingProcessor =
+        new SignatureEncodingProcessor(
+            encodingChecker, doubleWorkProcessor.publishOn(parallelExecutor));
 
     // identify attestation target
-    Flux.merge(encodingCheckOut.getSatisfied(), newSlotsFx, importedBlocksFx).subscribe(identifier);
+    IdentificationProcessor identificationProcessor =
+        new IdentificationProcessor(
+            unknownPool, schedulers, encodingProcessor.getValid(), newSlotsFx, importedBlocksFx);
 
     // verify attestations
-    identifier.bufferTimeout(VERIFIER_BUFFER_SIZE, VERIFIER_INTERVAL).subscribe(verifier);
-    FluxSplit<CheckedAttestation> verifierOut =
-        Fluxes.split(verifier, CheckedAttestation::isPassed);
+    Flux<List<ReceivedAttestation>> verificationThrottle =
+        identificationProcessor
+            .getIdentified()
+            .publishOn(parallelExecutor)
+            .bufferTimeout(VERIFIER_BUFFER_SIZE, VERIFIER_INTERVAL);
+    VerificationProcessor verificationProcessor =
+        new VerificationProcessor(verifier, verificationThrottle);
 
     // feed churn
-    Flux.merge(
-        verifierOut.getSatisfied().map(CheckedAttestation::getAttestation),
-        newSlotsFx,
-        chainHeadsFx);
+    ChurnProcessor churnProcessor = new ChurnProcessor(schedulers, chainHeadsFx, newSlotsFx);
 
+    Scheduler outScheduler = schedulers.events().toReactor();
+    // expose valid attestations
+    verificationProcessor.getValid().publishOn(outScheduler).subscribe(validAttestations);
+    // expose not yet identified
+    identificationProcessor.getUnknown().publishOn(outScheduler).subscribe(unknownAttestations);
+    // expose aggregates
+    churnProcessor.publishOn(outScheduler).subscribe(offChainAggregates);
     // expose invalid attestations
     Flux.merge(
-            sanityOut.getUnsatisfied().map(CheckedAttestation::getAttestation),
-            encodingCheckOut.getUnsatisfied(),
-            verifierOut.getUnsatisfied().map(CheckedAttestation::getAttestation))
+            sanityProcessor.getInvalid(),
+            encodingProcessor.getInvalid(),
+            verificationProcessor.getInvalid())
+        .publishOn(outScheduler)
         .subscribe(invalidAttestations);
-
-    // expose valid attestations
-    verifierOut.getSatisfied().map(CheckedAttestation::getAttestation).subscribe(validAttestations);
   }
 
   @Override
@@ -148,11 +153,11 @@ public Publisher<ReceivedAttestation> getInvalid() {
 
   @Override
   public Publisher<ReceivedAttestation> getUnknownAttestations() {
-    return identifier.getUnknownAttestations();
+    return unknownAttestations;
   }
 
   @Override
   public Publisher<OffChainAggregates> getAggregates() {
-    return churn;
+    return offChainAggregates;
   }
 }

chain/src/main/java/org/ethereum/beacon/chain/pool/reactor/ChurnProcessor.java

@@ -1,13 +1,27 @@
 package org.ethereum.beacon.chain.pool.reactor;
 
+import org.ethereum.beacon.chain.pool.ReceivedAttestation;
 import org.ethereum.beacon.chain.pool.churn.OffChainAggregates;
-import org.ethereum.beacon.stream.AbstractDelegateProcessor;
+import org.ethereum.beacon.core.BeaconBlock;
+import org.ethereum.beacon.core.types.SlotNumber;
+import org.ethereum.beacon.schedulers.Schedulers;
+import org.ethereum.beacon.stream.OutsourcePublisher;
+import reactor.core.CoreSubscriber;
+import reactor.core.publisher.Flux;
 
-public class ChurnProcessor
-    extends AbstractDelegateProcessor<Object, OffChainAggregates> {
+public class ChurnProcessor extends Flux<OffChainAggregates> {
 
-  @Override
-  protected void hookOnNext(Object value) {
+  private final OutsourcePublisher<OffChainAggregates> out = new OutsourcePublisher<>();
+
+  public ChurnProcessor(
+      Schedulers schedulers, Flux<BeaconBlock> chainHeads, Flux<SlotNumber> newSlots) {}
+
+  protected void hookOnNext(ReceivedAttestation value) {
     // TODO implement
   }
+
+  @Override
+  public void subscribe(CoreSubscriber<? super OffChainAggregates> actual) {
+    out.subscribe(actual);
+  }
 }

chain/src/main/java/org/ethereum/beacon/chain/pool/reactor/DoubleWorkProcessor.java

@@ -0,0 +1,51 @@
+package org.ethereum.beacon.chain.pool.reactor;
+
+import org.ethereum.beacon.chain.pool.ReceivedAttestation;
+import org.ethereum.beacon.chain.pool.registry.ProcessedAttestations;
+import org.ethereum.beacon.schedulers.Schedulers;
+import org.ethereum.beacon.stream.OutsourcePublisher;
+import reactor.core.CoreSubscriber;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Scheduler;
+
+/**
+ * Passes attestations through {@link ProcessedAttestations} filter.
+ *
+ * <p>Input:
+ *
+ * <ul>
+ *   <li>attestations
+ * </ul>
+ *
+ * <p>Output:
+ *
+ * <ul>
+ *   <li>Not yet processed attestations
+ * </ul>
+ */
+public class DoubleWorkProcessor extends Flux<ReceivedAttestation> {
+
+  private final ProcessedAttestations processedAttestations;
+  private final OutsourcePublisher<ReceivedAttestation> out = new OutsourcePublisher<>();
+
+  public DoubleWorkProcessor(
+      ProcessedAttestations processedAttestations,
+      Schedulers schedulers,
+      Flux<ReceivedAttestation> source) {
+    this.processedAttestations = processedAttestations;
+
+    Scheduler scheduler = schedulers.newSingleThreadDaemon("pool-double-work").toReactor();
+    source.publishOn(scheduler).subscribe(this::hookOnNext);
+  }
+
+  private void hookOnNext(ReceivedAttestation attestation) {
+    if (!processedAttestations.add(attestation)) {
+      out.publishOut(attestation);
+    }
+  }
+
+  @Override
+  public void subscribe(CoreSubscriber<? super ReceivedAttestation> actual) {
+    out.subscribe(actual);
+  }
+}

chain/src/main/java/org/ethereum/beacon/chain/pool/reactor/IdentificationProcessor.java

@@ -4,12 +4,13 @@
 import org.ethereum.beacon.chain.pool.registry.UnknownAttestationPool;
 import org.ethereum.beacon.core.BeaconBlock;
 import org.ethereum.beacon.core.types.SlotNumber;
-import org.ethereum.beacon.stream.AbstractDelegateProcessor;
-import reactor.core.publisher.DirectProcessor;
-import reactor.core.publisher.FluxSink;
+import org.ethereum.beacon.schedulers.Schedulers;
+import org.ethereum.beacon.stream.OutsourcePublisher;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Scheduler;
 
 /**
- * Processor throttling attestations through {@link UnknownAttestationPool}.
+ * Passes attestations through {@link UnknownAttestationPool}.
  *
  * <p>Input:
  *
@@ -24,46 +25,49 @@
  * <ul>
  *   <li>instantly identified attestations
  *   <li>attestations identified upon a new block come
+ *   <li>attestations with not yet imported block
  * </ul>
  */
-public class IdentificationProcessor extends AbstractDelegateProcessor<Object, ReceivedAttestation> {
+public class IdentificationProcessor {
 
   private final UnknownAttestationPool pool;
-  private final DirectProcessor<ReceivedAttestation> unknownAttestations = DirectProcessor.create();
-  private final FluxSink<ReceivedAttestation> unknownOut = unknownAttestations.sink();
+  private final OutsourcePublisher<ReceivedAttestation> identified = new OutsourcePublisher<>();
+  private final OutsourcePublisher<ReceivedAttestation> unknown = new OutsourcePublisher<>();
 
-  public IdentificationProcessor(UnknownAttestationPool pool) {
+  public IdentificationProcessor(
+      UnknownAttestationPool pool,
+      Schedulers schedulers,
+      Flux<ReceivedAttestation> source,
+      Flux<SlotNumber> newSlots,
+      Flux<BeaconBlock> newImportedBlocks) {
     this.pool = pool;
-  }
 
-  @Override
-  protected void hookOnNext(Object value) {
-    if (value.getClass().equals(BeaconBlock.class)) {
-      // forward attestations identified with a new block
-      pool.feedNewImportedBlock((BeaconBlock) value).forEach(this::publishOut);
-    } else if (value.getClass().equals(SlotNumber.class)) {
-      pool.feedNewSlot((SlotNumber) value);
-    } else if (value.getClass().equals(ReceivedAttestation.class)) {
-      if (pool.isInitialized()) {
-        return;
-      }
+    Scheduler scheduler =
+        schedulers.newSingleThreadDaemon("pool-attestation-identifier").toReactor();
+    newSlots.publishOn(scheduler).subscribe(this.pool::feedNewSlot);
+    newImportedBlocks.publishOn(scheduler).subscribe(this::hookOnNext);
+    source.publishOn(scheduler).subscribe(this::hookOnNext);
+  }
 
-      ReceivedAttestation attestation = (ReceivedAttestation) value;
+  private void hookOnNext(BeaconBlock block) {
+    pool.feedNewImportedBlock(block).forEach(identified::publishOut);
+  }
 
-      if (!pool.add(attestation)) {
-        // forward attestations not added to the pool
-        publishOut(attestation);
+  private void hookOnNext(ReceivedAttestation attestation) {
+    if (pool.isInitialized()) {
+      if (pool.add(attestation)) {
+        unknown.publishOut(attestation);
       } else {
-        // expose not yet identified attestations
-        unknownOut.next(attestation);
+        identified.publishOut(attestation);
       }
-    } else {
-      throw new IllegalArgumentException(
-          "Unsupported input type: " + value.getClass().getSimpleName());
     }
   }
 
-  public DirectProcessor<ReceivedAttestation> getUnknownAttestations() {
-    return unknownAttestations;
+  public OutsourcePublisher<ReceivedAttestation> getIdentified() {
+    return identified;
+  }
+
+  public OutsourcePublisher<ReceivedAttestation> getUnknown() {
+    return unknown;
   }
 }