This repository has been archived by the owner on Aug 23, 2020. It is now read-only.
Feature: Add solidification stage to pipeline and only broadcast solid transactions #1646
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,8 +1,10 @@ | ||
| package com.iota.iri.network.pipeline; | ||
|
|
||
| import com.iota.iri.service.validation.TransactionSolidifier; | ||
| import com.iota.iri.service.validation.TransactionValidator; | ||
| import com.iota.iri.conf.NodeConfig; | ||
| import com.iota.iri.controllers.TipsViewModel; | ||
| import com.iota.iri.controllers.TransactionViewModel; | ||
| import com.iota.iri.crypto.batched.BatchedHasher; | ||
| import com.iota.iri.crypto.batched.BatchedHasherFactory; | ||
| import com.iota.iri.crypto.batched.HashRequest; | ||
|
|
@@ -19,7 +21,10 @@ | |
| import com.iota.iri.utils.Converter; | ||
|
|
||
| import java.nio.ByteBuffer; | ||
| import java.util.Iterator; | ||
| import java.util.LinkedHashSet; | ||
| import java.util.List; | ||
| import java.util.Set; | ||
| import java.util.concurrent.ArrayBlockingQueue; | ||
| import java.util.concurrent.BlockingQueue; | ||
| import java.util.concurrent.ExecutorService; | ||
|
|
@@ -62,12 +67,13 @@ public class TransactionProcessingPipelineImpl implements TransactionProcessingP | |
| private BroadcastStage broadcastStage; | ||
| private BatchedHasher batchedHasher; | ||
| private HashingStage hashingStage; | ||
| private TransactionSolidifier txSolidifier; | ||
|
|
||
| private BlockingQueue<ProcessingContext> preProcessStageQueue = new ArrayBlockingQueue<>(100); | ||
| private BlockingQueue<ProcessingContext> validationStageQueue = new ArrayBlockingQueue<>(100); | ||
| private BlockingQueue<ProcessingContext> receivedStageQueue = new ArrayBlockingQueue<>(100); | ||
| private BlockingQueue<ProcessingContext> replyStageQueue = new ArrayBlockingQueue<>(100); | ||
| private BlockingQueue<ProcessingContext> broadcastStageQueue = new ArrayBlockingQueue<>(100); | ||
|
|
||
| /** | ||
| * Creates a {@link TransactionProcessingPipeline}. | ||
|
|
@@ -82,9 +88,9 @@ public class TransactionProcessingPipelineImpl implements TransactionProcessingP | |
| * reply stage | ||
| */ | ||
| public TransactionProcessingPipelineImpl(NeighborRouter neighborRouter, NodeConfig config, | ||
| TransactionValidator txValidator, Tangle tangle, SnapshotProvider snapshotProvider, | ||
| TipsViewModel tipsViewModel, LatestMilestoneTracker latestMilestoneTracker, | ||
| TransactionRequester transactionRequester, TransactionSolidifier txSolidifier) { | ||
DyrellC marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| FIFOCache<Long, Hash> recentlySeenBytesCache = new FIFOCache<>(config.getCacheSizeBytes()); | ||
| this.preProcessStage = new PreProcessStage(recentlySeenBytesCache); | ||
| this.replyStage = new ReplyStage(neighborRouter, config, tangle, tipsViewModel, latestMilestoneTracker, | ||
|
|
@@ -94,6 +100,7 @@ public TransactionProcessingPipelineImpl(NeighborRouter neighborRouter, NodeConf | |
| this.receivedStage = new ReceivedStage(tangle, txValidator, snapshotProvider, transactionRequester); | ||
| this.batchedHasher = BatchedHasherFactory.create(BatchedHasherFactory.Type.BCTCURL81, 20); | ||
| this.hashingStage = new HashingStage(batchedHasher); | ||
| this.txSolidifier = txSolidifier; | ||
| } | ||
|
|
||
| @Override | ||
|
|
@@ -119,6 +126,7 @@ private void addStage(String name, BlockingQueue<ProcessingContext> queue, | |
| try { | ||
| while (!Thread.currentThread().isInterrupted()) { | ||
| ProcessingContext ctx = stage.process(queue.take()); | ||
|
|
||
| switch (ctx.getNextStage()) { | ||
| case REPLY: | ||
| replyStageQueue.put(ctx); | ||
|
|
@@ -177,6 +185,7 @@ public BlockingQueue<ProcessingContext> getValidationStageQueue() { | |
| public void process(Neighbor neighbor, ByteBuffer data) { | ||
| try { | ||
| preProcessStageQueue.put(new ProcessingContext(new PreProcessPayload(neighbor, data))); | ||
| refillBroadcastQueue(); | ||
| } catch (InterruptedException e) { | ||
| e.printStackTrace(); | ||
| } | ||
|
|
@@ -191,6 +200,23 @@ public void process(byte[] txTrits) { | |
| hashAndValidate(new ProcessingContext(payload)); | ||
| } | ||
|
|
||
| @Override | ||
| public void refillBroadcastQueue(){ | ||
| try{ | ||
| Iterator<TransactionViewModel> hashIterator = txSolidifier.getBroadcastQueue().iterator(); | ||
| Set<TransactionViewModel> toRemove = new LinkedHashSet<>(); | ||
| while(!Thread.currentThread().isInterrupted() && hashIterator.hasNext()){ | ||
| TransactionViewModel t = hashIterator.next(); | ||
| broadcastStageQueue.put(new ProcessingContext(new BroadcastPayload(null, t))); | ||
| toRemove.add(t); | ||
|
There was a problem hiding this comment. Question: Small peev: There was a problem hiding this comment. This refill is what adds solid transactions to the broadcast queue, so here, yes you would send it to all neighbours. I'll change the 1 letter variable though. There was a problem hiding this comment. Just making sure, now with the new stage, this problem is resolved? There was a problem hiding this comment. Due to the asynchronous nature of solidification, there is still no neighbour variable to pass through unless we create a mapping, which imo is not worth it to do. So anything that gets updated as solid will be broadcast to all neighbours. But now only transactions that are solid will end up in the broadcast queue. We no longer forward transactions that have just been received. There was a problem hiding this comment. I think we need to create this mapping There was a problem hiding this comment. @DyrellC I am marking this as a point to discuss before merging There was a problem hiding this comment. We will need to discuss this, because now that the solidify stage does not add the incoming transactions to the solidification queue anymore, we have no entry point to put the origin neighbour into the process. As is, the only place that places transactions into the solidification queue is from the milestone solidifier. This is something that could be introduced in the milestone stage, I mention that in the new issue #1701 |
||
| hashIterator.remove(); | ||
| } | ||
| txSolidifier.clearBroadcastQueue(toRemove); | ||
| } catch(InterruptedException e){ | ||
| log.info(e.getMessage()); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Sets up the given hashing stage {@link ProcessingContext} so that up on success, it will submit further to the | ||
| * validation stage. | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Currently not sure why this has changed...
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
In line 138 you gossip the tx to neighbor
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This was to bolster broadcast rate to neighbours to increase the rate that
transactionsToRequestare sent out. It was meant to increase solidification speed.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Seems a bit hacky...
I've been thinking about it and I think this should be the correct design:
You need to add a solidification stage and a solidification queue in
TransactionProcessingPipeline.The flow should be like this:
From the
ReceivedStage-> if solid (due to quickly solid) -> Broadcast-> else go to solidification stage -> Broadcast
Also remember to pass neighbor information along so we don't broadcast the solid tx to the neighbor that sent this to us..