Event loop for pipeline processing #2209

lebdron · 2019-04-02T16:42:04Z

Description of the Change

Create a synchronize subject in consensus to queue the events from the network without locking the operation of consensus

Patch rxcpp to remove data race and allow using current_thread scheduler with synchronize
Add subscriptions for subjects in irohad to manage their lifetime
Add round parameter to Yac to close rounds that are less than or equal to current round
Add publish().ref_count() to YacGateImpl observable to eliminate redundant calls to flat_map if multiple subscribers are present
Fix data race in getSpdlogLogLevel
Fix nondeterminism in consensus_sunny_day test 🎉
Allow usage of blocking_observable in TestSubscriber

Benefits

Consensus can process messages from the network while the pipeline is running

Possible Drawbacks

A separate thread is allocated to the pipeline

Signed-off-by: Andrei Lebedev <[email protected]>

luckychess · 2019-04-03T13:37:20Z

cmake/Modules/Findrxcpp.cmake

@@ -1,6 +1,7 @@
 add_library(rxcpp INTERFACE IMPORTED)

-find_path(rxcpp_INCLUDE_DIR rxcpp/rx.hpp)
+# TODO 2019-04-02 lebdron: IR-346 Update Rx commit
+#find_path(rxcpp_INCLUDE_DIR rxcpp/rx.hpp)


I guess we don't need this line anymore.

luckychess · 2019-04-03T13:37:37Z

cmake/Modules/Findrxcpp.cmake

 set_target_description(rxcpp "Library for reactive programming" ${URL} ${VERSION})


-if (NOT rxcpp_FOUND)
+#if (NOT rxcpp_FOUND)


Maybe just remove?

luckychess · 2019-04-03T13:37:47Z

cmake/Modules/Findrxcpp.cmake

      TEST_COMMAND "" # remove test step
      )
  externalproject_get_property(reactive_extensions_rxcpp source_dir)
  set(rxcpp_INCLUDE_DIR ${source_dir}/Rx/v2/src)
  file(MAKE_DIRECTORY ${rxcpp_INCLUDE_DIR})

  add_dependencies(rxcpp reactive_extensions_rxcpp)
-endif ()
+#endif ()


And here as well.

luckychess · 2019-04-03T14:00:19Z

CI confuses me a bit - looks like all tests are passed but it shows failure status.

MBoldyrev

I will finish the review later.

MBoldyrev · 2019-04-03T14:15:36Z

irohad/consensus/yac/impl/yac.cpp

@@ -150,13 +172,17 @@ namespace iroha {

        network_->sendState(current_leader, {vote});
        cluster_order_.switchToNext();
-        if (cluster_order_.hasNext()) {
+        auto has_next = cluster_order_.hasNext();
+        lock.unlock();


should not we synchronize the other accesses to cluster_order_ for example with shared locks?
https://github.com/hyperledger/iroha/blob/6b97c3e467ca628d3a1e7f44a22b7d3420498813/irohad/consensus/yac/impl/yac.cpp#L190
https://github.com/hyperledger/iroha/blob/6b97c3e467ca628d3a1e7f44a22b7d3420498813/irohad/consensus/yac/impl/yac.cpp#L203
https://github.com/hyperledger/iroha/blob/6b97c3e467ca628d3a1e7f44a22b7d3420498813/irohad/consensus/yac/impl/yac.cpp#L286

and round_ btw

They can be synchronized with shared lock as optimization, but it will introduce more complexity to code. I am not sure that it is required at the moment.
The accesses you mentioned are covered by lock in onState now.

MBoldyrev · 2019-04-03T14:35:30Z

irohad/consensus/yac/impl/yac.cpp

          timer_->invokeAfterDelay([this, vote] { this->votingStep(vote); });
        }
      }

-      void Yac::closeRound() {
-        timer_->deny();
+      void Yac::closeRound(Round round) {


as for me, the closeRound function should better take no round argument, because we do not maintain more than one round at a time. Better would be to call it only if the proposal_round matches the condition here:
https://github.com/hyperledger/iroha/blob/6b97c3e467ca628d3a1e7f44a22b7d3420498813/irohad/consensus/yac/impl/yac.cpp#L247

if (proposal_round >= round_) { this->closeRound(); }

MBoldyrev · 2019-04-03T14:46:43Z

irohad/main/impl/on_demand_ordering_init.hpp

@@ -98,7 +100,8 @@ namespace iroha {
      /**
       * Initializes on-demand ordering gate and ordering sevice components
       *
-       * @param max_number_of_transactions maximum number of transaction in a proposal
+       * @param max_number_of_transactions maximum number of transaction in a


Suggested change

* @param max_number_of_transactions maximum number of transaction in a

* @param max_number_of_transactions maximum number of transactions in a

MBoldyrev · 2019-04-03T14:50:37Z

libs/logger/logger_spdlog.cpp

+        return spdlog::level::critical;
+      default:
+        BOOST_ASSERT_MSG(false, "Unknown log level!");
+        return getSpdlogLogLevel(logger::kDefaultLogLevel);


looks like a place for stack overflow! if something changes and logger::kDefaultLogLevel will hit default too.. better something like

Suggested change

return getSpdlogLogLevel(logger::kDefaultLogLevel);

return getSpdlogLogLevel(spdlog::level::info);

MBoldyrev · 2019-04-04T09:28:53Z

also it would be nice to give the patch patch/synchronize-lock-recursion-thread-safety.patch a more meaningful and descriptive name. I would suggest either renaming it with some rxcpp prefix or moving to a directory like patches/external/rxcpp (it other patches to rxcpp are probable).

Signed-off-by: Andrei Lebedev <[email protected]>

lebdron added 3 commits April 2, 2019 16:45

Add rxcpp patch and bump version

cfadddd

Signed-off-by: Andrei Lebedev <[email protected]>

Add support for running pipeline in a separate thread

e1b9571

Signed-off-by: Andrei Lebedev <[email protected]>

Fix yac gate flat map being called multiple times

6b97c3e

Signed-off-by: Andrei Lebedev <[email protected]>

lebdron requested review from luckychess and MBoldyrev April 2, 2019 16:42

luckychess approved these changes Apr 3, 2019

View reviewed changes

MBoldyrev reviewed Apr 3, 2019

View reviewed changes

MBoldyrev approved these changes Apr 4, 2019

View reviewed changes

lebdron added 6 commits April 4, 2019 21:01

Merge branch 'develop' into refactor/yac-threads

135a08b

Signed-off-by: Andrei Lebedev <[email protected]>

Merge branch 'develop' into refactor/yac-threads

4697562

Fix review

45a7e3f

Signed-off-by: Andrei Lebedev <[email protected]>

Reduce scope of no iostream definition

2566d8c

Signed-off-by: Andrei Lebedev <[email protected]>

Reduce the scope of yac linkage

7b5699f

Signed-off-by: Andrei Lebedev <[email protected]>

Fix deadlock in yac; fix data race in timer

f6c2de6

Signed-off-by: Andrei Lebedev <[email protected]>

lebdron requested a review from MBoldyrev April 6, 2019 16:56

Fix compilation

04e334b

Signed-off-by: Andrei Lebedev <[email protected]>

MBoldyrev approved these changes Apr 6, 2019

View reviewed changes

lebdron added 6 commits April 6, 2019 20:52

Fix review

be2f239

Signed-off-by: Andrei Lebedev <[email protected]>

Update rxcpp patch

75f6719

Signed-off-by: Andrei Lebedev <[email protected]>

Merge branch 'develop' into refactor/yac-threads

0fddbcb

Signed-off-by: Andrei Lebedev <[email protected]>

Merge branch 'develop' into refactor/yac-threads

a3c1ae3

Replace rxcpp patch with commit

2886b1e

Signed-off-by: Andrei Lebedev <[email protected]>

Add update command to rxcpp module

0cb8795

Signed-off-by: Andrei Lebedev <[email protected]>

lebdron merged commit 681dc16 into hyperledger-iroha:develop Apr 12, 2019

lebdron deleted the refactor/yac-threads branch April 12, 2019 09:21

lebdron mentioned this pull request May 6, 2019

Hyperledger Iroha v1.0 hyperledger-iroha/iroha#61

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Event loop for pipeline processing #2209

Event loop for pipeline processing #2209

lebdron commented Apr 2, 2019

luckychess Apr 3, 2019

luckychess Apr 3, 2019

luckychess Apr 3, 2019

luckychess commented Apr 3, 2019

MBoldyrev left a comment

MBoldyrev Apr 3, 2019

MBoldyrev Apr 3, 2019

lebdron Apr 4, 2019

MBoldyrev Apr 3, 2019

MBoldyrev Apr 3, 2019

MBoldyrev Apr 3, 2019

MBoldyrev commented Apr 4, 2019

	* @param max_number_of_transactions maximum number of transaction in a
	* @param max_number_of_transactions maximum number of transactions in a

	return getSpdlogLogLevel(logger::kDefaultLogLevel);
	return getSpdlogLogLevel(spdlog::level::info);

Event loop for pipeline processing #2209

Event loop for pipeline processing #2209

Conversation

lebdron commented Apr 2, 2019

Description of the Change

Benefits

Possible Drawbacks

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

luckychess commented Apr 3, 2019

MBoldyrev 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

MBoldyrev commented Apr 4, 2019