This repository was 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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 hidden or 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.
Outdated
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); | ||
|
||
| 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 hidden or 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 hidden or 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..