feat: Implement consumer handler in the Simulator #360

georgi-l95 · 2024-11-22T16:10:17Z

Description:
This pull request aims to make use of the consumer mode in the simulator, by adding needed implementations.
We add:

gRPC client and observer for consuming blocks, which utilizes SubscribeStreamRequest and SubscribeStreamResponse, which are part of the subscribeBlockStream rpc.
Makes the neccessery changes to initilize managers and clients, depending on mode.
Adds neccessery unit and integration tests for sufficient code coverage, by making use of as less mocks as possible.

Related issue(s):

Fixes #121

Notes for reviewer:

Checklist

Documented (Code comments, README, etc.)
Tested (unit, integration, etc.)

Signed-off-by: georgi-l95 <[email protected]>

codecov · 2024-11-27T21:50:38Z

Codecov Report

Attention: Patch coverage is 97.87234% with 2 lines in your changes missing coverage. Please review.

Project coverage is 97.37%. Comparing base (ad18213) to head (b6e5369).
Report is 5 commits behind head on main.

Files with missing lines	Patch %	Lines
...edera/block/simulator/BlockStreamSimulatorApp.java	90.00%	0 Missing and 1 partial ⚠️
.../simulator/generator/BlockAsFileLargeDataSets.java	0.00%	1 Missing ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##               main     #360      +/-   ##
============================================
+ Coverage     94.62%   97.37%   +2.74%     
- Complexity      316      360      +44     
============================================
  Files            69       72       +3     
  Lines          1228     1333     +105     
  Branches         84       89       +5     
============================================
+ Hits           1162     1298     +136     
+ Misses           55       23      -32     
- Partials         11       12       +1

Files with missing lines	Coverage Δ
...ulator/generator/BlockAsDirBlockStreamManager.java	`100.00% <100.00%> (ø)`
...lator/generator/BlockAsFileBlockStreamManager.java	`90.90% <100.00%> (-9.10%)`	⬇️
...dera/block/simulator/grpc/GrpcInjectionModule.java	`0.00% <ø> (ø)`
...ulator/grpc/impl/ConsumerStreamGrpcClientImpl.java	`100.00% <100.00%> (ø)`
...ck/simulator/grpc/impl/ConsumerStreamObserver.java	`100.00% <100.00%> (ø)`
...mulator/grpc/impl/PublishStreamGrpcClientImpl.java	`90.19% <ø> (ø)`
...ock/simulator/grpc/impl/PublishStreamObserver.java	`100.00% <100.00%> (ø)`
.../block/simulator/metrics/SimulatorMetricTypes.java	`100.00% <100.00%> (ø)`
...dera/block/simulator/mode/CombinedModeHandler.java	`100.00% <100.00%> (ø)`
...dera/block/simulator/mode/ConsumerModeHandler.java	`100.00% <100.00%> (ø)`
... and 3 more

... and 16 files with indirect coverage changes

mattp-swirldslabs

I left a few suggestions

mattp-swirldslabs · 2024-11-27T22:46:25Z

simulator/src/main/java/com/hedera/block/simulator/BlockStreamSimulatorApp.java

 public class BlockStreamSimulatorApp {
-
+    /** Logger for this class */


You don't need to javadoc private methods or member variables

mattp-swirldslabs · 2024-11-27T22:51:26Z

simulator/src/main/java/com/hedera/block/simulator/BlockStreamSimulatorApp.java

        this.metricsService = requireNonNull(metricsService);
        this.publishStreamGrpcClient = requireNonNull(publishStreamGrpcClient);
+        this.consumerStreamGrpcClient = requireNonNull(consumerStreamGrpcClient);
+        this.isRunning = new AtomicBoolean(false);


It looks like you moved the assignment from line 46 here. Why move it?

mattp-swirldslabs · 2024-11-27T22:54:24Z

simulator/src/main/java/com/hedera/block/simulator/BlockStreamSimulatorApp.java

            default -> throw new IllegalArgumentException("Unknown SimulatorMode: " + simulatorMode);
-        }


IntelliJ is signaling the default case isn't possible given it's an enum. Bad data from the config should break before this class when BlockStreamConfig is injected.

mattp-swirldslabs · 2024-11-27T23:00:32Z

simulator/src/main/java/com/hedera/block/simulator/mode/ConsumerModeHandler.java


+    // Configuration
    private final BlockStreamConfig blockStreamConfig;


BlockStreamConfig is never read. Should it be removed?

mattp-swirldslabs · 2024-11-27T23:07:19Z

simulator/src/main/java/com/hedera/block/simulator/generator/BlockStreamManager.java

+    /**
+     * Initialize the block stream manager and load blocks into memory.
+     */
+    void init();


Maybe change this to a default interface method like: default void init() {} so you don't have to override with a no-op comment in BlockAsFileLargeDataSets

mattp-swirldslabs · 2024-11-27T23:08:05Z

simulator/src/main/java/com/hedera/block/simulator/generator/BlockAsFileLargeDataSets.java

+    @Override
+    public void init() {
+        // Do nothing, because we don't have real initializing and loading blocks into memory for this implementation.
+    }


You should be able to remove this init() method. See my comment in BlockStreamManager

mattp-swirldslabs · 2024-11-27T23:26:47Z

simulator/src/main/java/com/hedera/block/simulator/grpc/impl/ConsumerStreamGrpcClientImpl.java

+    public void completeStreaming() throws InterruptedException {
+        consumerStreamObserver.onCompleted();
+        // todo(352) Find a suitable solution for removing the sleep
+        Thread.sleep(100);


I'm not 100% sure but I think Netty will continue to send BlockItems to onNext() as long as the channel is open. Right now, I think onCompleted() will only release the starting thread waiting at line 97. You might need to close the channel in this method and then release the starting thread.

mattp-swirldslabs · 2024-11-27T23:48:23Z

simulator/src/main/java/com/hedera/block/simulator/grpc/impl/ConsumerStreamObserver.java

+    private void processBlockItems(List<BlockItem> blockItems) {
+        blockItems.stream()
+                .filter(BlockItem::hasBlockProof)
+                .forEach(__ -> metricsService.get(LiveBlocksConsumed).increment());


If you start the consumer up and start streaming blockitems via the producer to the BN, it looks like nothing is happening on the consumer side. Consider changing this to something like:

blockItems.stream() .filter(BlockItem::hasBlockProof) .forEach(blockItem -> { metricsService.get(LiveBlocksConsumed).increment(); LOGGER.log(INFO, "Received block number: " + blockItem.getBlockProof().getBlock()); });

That should print the last block received without overwhelming the terminal. This gives the consumer parity with the producer which prints the response hashes right now.

georgi-l95 added 2 commits November 21, 2024 16:56

feat: add consumer logic for the simulator

fa0e4c8

Signed-off-by: georgi-l95 <[email protected]>

chore: refactor and cleanup

c84095a

Signed-off-by: georgi-l95 <[email protected]>

georgi-l95 added the Simulator Issue related to Block Stream Simulator label Nov 22, 2024

georgi-l95 added this to the 0.3.0 milestone Nov 22, 2024

georgi-l95 self-assigned this Nov 22, 2024

georgi-l95 linked an issue Nov 22, 2024 that may be closed by this pull request

Implement consumer logic in the Simulator #121

Open

georgi-l95 added 8 commits November 22, 2024 18:14

chore: fix unit tests

c73d097

Signed-off-by: georgi-l95 <[email protected]>

chore: switch to publisher by default

98ad021

Signed-off-by: georgi-l95 <[email protected]>

test: add unit tests

a8b4589

Signed-off-by: georgi-l95 <[email protected]>

test: add unit tests

583aedd

Signed-off-by: georgi-l95 <[email protected]>

test: add more tests

4d37a65

Signed-off-by: georgi-l95 <[email protected]>

add temporary code

d6d5deb

Signed-off-by: georgi-l95 <[email protected]>

fix unit tests

87b6d89

Signed-off-by: georgi-l95 <[email protected]>

fix loading block manager in consumer mode

b6e5369

Signed-off-by: georgi-l95 <[email protected]>

georgi-l95 marked this pull request as ready for review November 27, 2024 21:49

georgi-l95 requested a review from a team as a code owner November 27, 2024 21:49

georgi-l95 changed the title ~~feat: WIP Implement consumer handler in the Simulator~~ feat: Implement consumer handler in the Simulator Nov 27, 2024

mattp-swirldslabs approved these changes Nov 27, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: Implement consumer handler in the Simulator #360

feat: Implement consumer handler in the Simulator #360

georgi-l95 commented Nov 22, 2024 •

edited

Loading

codecov bot commented Nov 27, 2024

mattp-swirldslabs left a comment

mattp-swirldslabs Nov 27, 2024

mattp-swirldslabs Nov 27, 2024

mattp-swirldslabs Nov 27, 2024

mattp-swirldslabs Nov 27, 2024

mattp-swirldslabs Nov 27, 2024

mattp-swirldslabs Nov 27, 2024

mattp-swirldslabs Nov 27, 2024

mattp-swirldslabs Nov 27, 2024

		public class BlockStreamSimulatorApp {

		/** Logger for this class */

		default -> throw new IllegalArgumentException("Unknown SimulatorMode: " + simulatorMode);
		}


		// Configuration
		private final BlockStreamConfig blockStreamConfig;

feat: Implement consumer handler in the Simulator #360

Are you sure you want to change the base?

feat: Implement consumer handler in the Simulator #360

Conversation

georgi-l95 commented Nov 22, 2024 • edited Loading

codecov bot commented Nov 27, 2024

Codecov Report

mattp-swirldslabs left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

georgi-l95 commented Nov 22, 2024 •

edited

Loading