From 342e33008ca4f1f3142b8660b027078cb8c766e5 Mon Sep 17 00:00:00 2001 From: elmato Date: Fri, 3 Feb 2017 19:25:52 +0000 Subject: [PATCH 1/9] Restore bloom filter usage: Check if the object changed/removed was previously subscribed (read) before sending updates back to ws clients --- libraries/app/database_api.cpp | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/libraries/app/database_api.cpp b/libraries/app/database_api.cpp index b811ebcd..f296005a 100644 --- a/libraries/app/database_api.cpp +++ b/libraries/app/database_api.cpp @@ -160,7 +160,7 @@ class database_api_impl : public std::enable_shared_from_this { if( !_subscribe_callback ) return false; - return true; + return _subscribe_filter.contains( i ); } @@ -1818,8 +1818,10 @@ void database_api_impl::on_objects_removed( const vector& objs ) vector updates; updates.reserve(objs.size()); - for( auto obj : objs ) - updates.emplace_back( obj->id ); + for( auto obj : objs ) { + if ( is_subscribed_to_item(obj->id) ) + updates.emplace_back( obj->id ); + } broadcast_updates( updates ); } @@ -1859,7 +1861,7 @@ void database_api_impl::on_objects_changed(const vector& ids) for(auto id : ids) { const object* obj = nullptr; - if( _subscribe_callback ) + if( is_subscribed_to_item(id) ) { obj = _db.find_object( id ); if( obj ) @@ -1874,7 +1876,7 @@ void database_api_impl::on_objects_changed(const vector& ids) if( _market_subscriptions.size() ) { - if( !_subscribe_callback ) + if( !is_subscribed_to_item(id) ) obj = _db.find_object( id ); if( obj ) { @@ -1895,7 +1897,7 @@ void database_api_impl::on_objects_changed(const vector& ids) /// if a connection hangs then this could get backed up and result in /// a failure to exit cleanly. fc::async([capture_this,this,updates,market_broadcast_queue](){ - if( _subscribe_callback ) _subscribe_callback( updates ); + if( _subscribe_callback && updates.size() ) _subscribe_callback( updates ); for( const auto& item : market_broadcast_queue ) { From 0bc0513e206ce221cec37ae4fb6ca5b4395b655a Mon Sep 17 00:00:00 2001 From: elmato Date: Mon, 6 Feb 2017 15:39:42 +0000 Subject: [PATCH 2/9] set_subscribe_callback: always initialize bloom filter --- libraries/app/database_api.cpp | 16 +++++++--------- 1 file changed, 7 insertions(+), 9 deletions(-) diff --git a/libraries/app/database_api.cpp b/libraries/app/database_api.cpp index f296005a..f92bf977 100644 --- a/libraries/app/database_api.cpp +++ b/libraries/app/database_api.cpp @@ -267,15 +267,13 @@ void database_api_impl::set_subscribe_callback( std::function cb ) From 823beb7fe5a5a888a373bfe439a1189cf0ec6023 Mon Sep 17 00:00:00 2001 From: elmato Date: Fri, 10 Feb 2017 08:18:23 +0000 Subject: [PATCH 3/9] remove call to notify_changed_objects in _push_transaction --- libraries/chain/db_block.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libraries/chain/db_block.cpp b/libraries/chain/db_block.cpp index 63306e66..f7ecd885 100644 --- a/libraries/chain/db_block.cpp +++ b/libraries/chain/db_block.cpp @@ -241,7 +241,7 @@ processed_transaction database::_push_transaction( const signed_transaction& trx auto processed_trx = _apply_transaction( trx ); _pending_tx.push_back(processed_trx); - notify_changed_objects(); + // notify_changed_objects(); // The transaction applied successfully. Merge its changes into the pending block session. temp_session.merge(); From 02b2672a394b17f70927db73eb0000a50b1818a7 Mon Sep 17 00:00:00 2001 From: elmato Date: Fri, 10 Feb 2017 08:25:39 +0000 Subject: [PATCH 4/9] split notifications (notify_changed_objects) in three signals: new_objects, changed_objects, removed_objects --- libraries/chain/db_block.cpp | 18 ++++++++++-------- .../chain/include/graphene/chain/database.hpp | 6 ++++++ 2 files changed, 16 insertions(+), 8 deletions(-) diff --git a/libraries/chain/db_block.cpp b/libraries/chain/db_block.cpp index f7ecd885..9d8cee4f 100644 --- a/libraries/chain/db_block.cpp +++ b/libraries/chain/db_block.cpp @@ -549,17 +549,19 @@ void database::notify_changed_objects() if( _undo_db.enabled() ) { const auto& head_undo = _undo_db.head(); + + vector new_ids; new_ids.reserve(head_undo.new_ids.size()); + for( const auto& item : head_undo.new_ids ) new_ids.push_back(item); + vector changed_ids; changed_ids.reserve(head_undo.old_values.size()); for( const auto& item : head_undo.old_values ) changed_ids.push_back(item.first); - for( const auto& item : head_undo.new_ids ) changed_ids.push_back(item); - vector removed; - removed.reserve( head_undo.removed.size() ); - for( const auto& item : head_undo.removed ) - { - changed_ids.push_back( item.first ); - removed.emplace_back( item.second.get() ); - } + + vector removed; removed.reserve( head_undo.removed.size() ); + for( const auto& item : head_undo.removed ) removed.emplace_back( item.second.get() ); + + new_objects(new_ids); changed_objects(changed_ids); + removed_objects(removed); } } FC_CAPTURE_AND_RETHROW() } diff --git a/libraries/chain/include/graphene/chain/database.hpp b/libraries/chain/include/graphene/chain/database.hpp index 1b721253..47fd63b9 100644 --- a/libraries/chain/include/graphene/chain/database.hpp +++ b/libraries/chain/include/graphene/chain/database.hpp @@ -189,6 +189,12 @@ namespace graphene { namespace chain { */ fc::signal on_pending_transaction; + /** + * Emitted After a block has been applied and committed. The callback + * should not yield and should execute quickly. + */ + fc::signal&)> new_objects; + /** * Emitted After a block has been applied and committed. The callback * should not yield and should execute quickly. From b2b895ac94c064f91c17be863255da4a84a90ccd Mon Sep 17 00:00:00 2001 From: elmato Date: Fri, 10 Feb 2017 08:29:48 +0000 Subject: [PATCH 5/9] allow to apply_block when exception in database signal handler --- libraries/chain/db_block.cpp | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/libraries/chain/db_block.cpp b/libraries/chain/db_block.cpp index 9d8cee4f..33403edf 100644 --- a/libraries/chain/db_block.cpp +++ b/libraries/chain/db_block.cpp @@ -563,7 +563,7 @@ void database::notify_changed_objects() changed_objects(changed_ids); removed_objects(removed); } -} FC_CAPTURE_AND_RETHROW() } +} FC_CAPTURE_AND_LOG( () ) } processed_transaction database::apply_transaction(const signed_transaction& trx, uint32_t skip) { From 29c636fcef74aeb8a56a0067b04b22e469eece4f Mon Sep 17 00:00:00 2001 From: elmato Date: Fri, 10 Feb 2017 08:33:44 +0000 Subject: [PATCH 6/9] add get_market() function to call_order_object --- libraries/chain/include/graphene/chain/market_object.hpp | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/libraries/chain/include/graphene/chain/market_object.hpp b/libraries/chain/include/graphene/chain/market_object.hpp index c41def13..b56f4e9c 100644 --- a/libraries/chain/include/graphene/chain/market_object.hpp +++ b/libraries/chain/include/graphene/chain/market_object.hpp @@ -120,6 +120,13 @@ class call_order_object : public abstract_object share_type collateral; ///< call_price.base.asset_id, access via get_collateral share_type debt; ///< call_price.quote.asset_id, access via get_collateral price call_price; ///< Debt / Collateral + + pair get_market()const + { + auto tmp = std::make_pair( call_price.base.asset_id, call_price.quote.asset_id ); + if( tmp.first > tmp.second ) std::swap( tmp.first, tmp.second ); + return tmp; + } }; /** From bfa600c5594cd05519fff72917f47133ebeeb516 Mon Sep 17 00:00:00 2001 From: elmato Date: Fri, 10 Feb 2017 08:36:07 +0000 Subject: [PATCH 7/9] handle new database_object signals, refactor --- libraries/app/database_api.cpp | 166 ++++++++++++++++++++------------- 1 file changed, 99 insertions(+), 67 deletions(-) diff --git a/libraries/app/database_api.cpp b/libraries/app/database_api.cpp index f92bf977..6b52053b 100644 --- a/libraries/app/database_api.cpp +++ b/libraries/app/database_api.cpp @@ -41,6 +41,8 @@ #define GET_REQUIRED_FEES_MAX_RECURSION 4 +typedef std::map< std::pair, std::vector > market_queue_type; + namespace graphene { namespace app { class database_api_impl; @@ -164,9 +166,26 @@ class database_api_impl : public std::enable_shared_from_this return _subscribe_filter.contains( i ); } - void broadcast_updates( const vector& updates ); + template + void enqueue_if_subscribed_to_market(const object* obj, market_queue_type& queue, bool full_object=true) + { + const T* order = dynamic_cast(obj); + FC_ASSERT( order != nullptr); + auto market = order->get_market(); + + auto sub = _market_subscriptions.find( market ); + if( sub != _market_subscriptions.end() ) { + queue[market].emplace_back( full_object ? obj->to_variant() : fc::variant(obj->id) ); + } + } + + void broadcast_updates( const vector& updates ); + void broadcast_market_updates( const market_queue_type& queue); + void check_for_market_objects(const vector& ids); + /** called every time a block is applied to report the objects that were changed */ + void on_objects_new(const vector& ids); void on_objects_changed(const vector& ids); void on_objects_removed(const vector& objs); void on_applied_block(); @@ -176,6 +195,7 @@ class database_api_impl : public std::enable_shared_from_this std::function _pending_trx_callback; std::function _block_applied_callback; + boost::signals2::scoped_connection _new_connection; boost::signals2::scoped_connection _change_connection; boost::signals2::scoped_connection _removed_connection; boost::signals2::scoped_connection _applied_block_connection; @@ -198,6 +218,9 @@ database_api::~database_api() {} database_api_impl::database_api_impl( graphene::chain::database& db ):_db(db) { wlog("creating database api ${x}", ("x",int64_t(this)) ); + _new_connection = _db.new_objects.connect([this](const vector& ids) { + on_objects_new(ids); + }); _change_connection = _db.changed_objects.connect([this](const vector& ids) { on_objects_changed(ids); }); @@ -628,7 +651,7 @@ std::map database_api_impl::get_full_accounts( const [&acnt] (const call_order_object& call) { acnt.call_orders.emplace_back(call); }); - + // get assets issued by user auto asset_range = _db.get_index_type().indices().get().equal_range(account->id); std::for_each(asset_range.first, asset_range.second, @@ -1800,10 +1823,27 @@ vector database_api_impl::get_blinded_balances( const fl void database_api_impl::broadcast_updates( const vector& updates ) { - if( updates.size() ) { + if( updates.size() && _subscribe_callback ) { auto capture_this = shared_from_this(); fc::async([capture_this,updates](){ - capture_this->_subscribe_callback( fc::variant(updates) ); + if(capture_this->_subscribe_callback) + capture_this->_subscribe_callback( fc::variant(updates) ); + }); + } +} + +void database_api_impl::broadcast_market_updates( const market_queue_type& queue) +{ + if( queue.size() ) + { + auto capture_this = shared_from_this(); + fc::async([capture_this, this, queue](){ + for( const auto& item : queue ) + { + auto sub = _market_subscriptions.find(item.first); + if( sub != _market_subscriptions.end() ) + sub->second( fc::variant(item.second ) ); + } }); } } @@ -1813,97 +1853,89 @@ void database_api_impl::on_objects_removed( const vector& objs ) /// we need to ensure the database_api is not deleted for the life of the async operation if( _subscribe_callback ) { - vector updates; - updates.reserve(objs.size()); + vector updates; - for( auto obj : objs ) { + for( auto obj : objs ) + { if ( is_subscribed_to_item(obj->id) ) - updates.emplace_back( obj->id ); + { + updates.push_back( fc::variant(obj->id) ); + } } + broadcast_updates( updates ); } if( _market_subscriptions.size() ) { - map< pair, vector > broadcast_queue; + market_queue_type broadcast_queue; + for( const auto& obj : objs ) { - const limit_order_object* order = dynamic_cast(obj); - if( order ) + if( obj->id.is() ) { - auto sub = _market_subscriptions.find( order->get_market() ); - if( sub != _market_subscriptions.end() ) - broadcast_queue[order->get_market()].emplace_back( order->id ); + enqueue_if_subscribed_to_market( obj, broadcast_queue, false ); + } + else if( obj->id.is() ) + { + enqueue_if_subscribed_to_market( obj, broadcast_queue, false ); } } - if( broadcast_queue.size() ) - { - auto capture_this = shared_from_this(); - fc::async([capture_this,this,broadcast_queue](){ - for( const auto& item : broadcast_queue ) - { - auto sub = _market_subscriptions.find(item.first); - if( sub != _market_subscriptions.end() ) - sub->second( fc::variant(item.second ) ); - } - }); - } + + broadcast_market_updates(broadcast_queue); } } +void database_api_impl::check_for_market_objects(const vector& ids) +{ + if( _market_subscriptions.size() ) + { + market_queue_type broadcast_queue; + + for(auto id : ids) + { + if( id.is() ) + { + enqueue_if_subscribed_to_market( _db.find_object(id), broadcast_queue ); + } + else if( id.is() ) + { + enqueue_if_subscribed_to_market( _db.find_object(id), broadcast_queue ); + } + } + + broadcast_market_updates(broadcast_queue); + } +} + +void database_api_impl::on_objects_new(const vector& ids) +{ + check_for_market_objects(ids); +} + void database_api_impl::on_objects_changed(const vector& ids) { - vector updates; - map< pair, vector > market_broadcast_queue; - - for(auto id : ids) + if( _subscribe_callback ) { - const object* obj = nullptr; - if( is_subscribed_to_item(id) ) + vector updates; + + for(auto id : ids) { - obj = _db.find_object( id ); - if( obj ) + const object* obj = nullptr; + if( is_subscribed_to_item(id) ) { - updates.emplace_back( obj->to_variant() ); - } - else - { - updates.emplace_back(id); // send just the id to indicate removal - } - } - - if( _market_subscriptions.size() ) - { - if( !is_subscribed_to_item(id) ) obj = _db.find_object( id ); - if( obj ) - { - const limit_order_object* order = dynamic_cast(obj); - if( order ) + if( obj ) { - auto sub = _market_subscriptions.find( order->get_market() ); - if( sub != _market_subscriptions.end() ) - market_broadcast_queue[order->get_market()].emplace_back( order->id ); + updates.emplace_back( obj->to_variant() ); } } } + + broadcast_updates(updates); } - auto capture_this = shared_from_this(); - - /// pushing the future back / popping the prior future if it is complete. - /// if a connection hangs then this could get backed up and result in - /// a failure to exit cleanly. - fc::async([capture_this,this,updates,market_broadcast_queue](){ - if( _subscribe_callback && updates.size() ) _subscribe_callback( updates ); - - for( const auto& item : market_broadcast_queue ) - { - auto sub = _market_subscriptions.find(item.first); - if( sub != _market_subscriptions.end() ) - sub->second( fc::variant(item.second ) ); - } - }); + check_for_market_objects(ids); } /** note: this method cannot yield because it is called in the middle of From 6f1d8b548be55210bc63fbb296c67039434b0ed9 Mon Sep 17 00:00:00 2001 From: elmato Date: Tue, 14 Feb 2017 21:34:08 +0000 Subject: [PATCH 8/9] Replace clear_filter parameter with notify_remove_create --- libraries/app/database_api.cpp | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) diff --git a/libraries/app/database_api.cpp b/libraries/app/database_api.cpp index 6b52053b..4df13551 100644 --- a/libraries/app/database_api.cpp +++ b/libraries/app/database_api.cpp @@ -58,7 +58,7 @@ class database_api_impl : public std::enable_shared_from_this fc::variants get_objects(const vector& ids)const; // Subscriptions - void set_subscribe_callback( std::function cb, bool clear_filter ); + void set_subscribe_callback( std::function cb, bool notify_remove_create ); void set_pending_transaction_callback( std::function cb ); void set_block_applied_callback( std::function cb ); void cancel_all_subscriptions(); @@ -190,7 +190,8 @@ class database_api_impl : public std::enable_shared_from_this void on_objects_removed(const vector& objs); void on_applied_block(); - mutable fc::bloom_filter _subscribe_filter; + bool _notify_remove_create = false; + mutable fc::bloom_filter _subscribe_filter; std::function _subscribe_callback; std::function _pending_trx_callback; std::function _block_applied_callback; @@ -281,15 +282,16 @@ fc::variants database_api_impl::get_objects(const vector& ids)co // // ////////////////////////////////////////////////////////////////////// -void database_api::set_subscribe_callback( std::function cb, bool clear_filter ) +void database_api::set_subscribe_callback( std::function cb, bool notify_remove_create ) { - my->set_subscribe_callback( cb, clear_filter ); + my->set_subscribe_callback( cb, notify_remove_create ); } -void database_api_impl::set_subscribe_callback( std::function cb, bool clear_filter ) +void database_api_impl::set_subscribe_callback( std::function cb, bool notify_remove_create ) { - edump((clear_filter)); + //edump((clear_filter)); _subscribe_callback = cb; + _notify_remove_create = notify_remove_create; static fc::bloom_parameters param; param.projected_element_count = 10000; @@ -589,7 +591,6 @@ std::map database_api_impl::get_full_accounts( const if( subscribe ) { - ilog( "subscribe to ${id}", ("id",account->name) ); subscribe_to_item( account->id ); } From 0ecdc90d4df2ebd80a6d4cfa4388f4b8a608c9c1 Mon Sep 17 00:00:00 2001 From: elmato Date: Tue, 14 Feb 2017 21:38:40 +0000 Subject: [PATCH 9/9] Add the array of ids (that are being removed) to the removed_objects signal --- libraries/chain/db_block.cpp | 9 +++++++-- libraries/chain/include/graphene/chain/database.hpp | 2 +- 2 files changed, 8 insertions(+), 3 deletions(-) diff --git a/libraries/chain/db_block.cpp b/libraries/chain/db_block.cpp index 33403edf..f55f0396 100644 --- a/libraries/chain/db_block.cpp +++ b/libraries/chain/db_block.cpp @@ -556,12 +556,17 @@ void database::notify_changed_objects() vector changed_ids; changed_ids.reserve(head_undo.old_values.size()); for( const auto& item : head_undo.old_values ) changed_ids.push_back(item.first); + vector removed_ids; removed_ids.reserve( head_undo.removed.size() ); vector removed; removed.reserve( head_undo.removed.size() ); - for( const auto& item : head_undo.removed ) removed.emplace_back( item.second.get() ); + for( const auto& item : head_undo.removed ) + { + removed_ids.emplace_back( item.first ); + removed.emplace_back( item.second.get() ); + } new_objects(new_ids); changed_objects(changed_ids); - removed_objects(removed); + removed_objects(removed_ids, removed); } } FC_CAPTURE_AND_LOG( () ) } diff --git a/libraries/chain/include/graphene/chain/database.hpp b/libraries/chain/include/graphene/chain/database.hpp index 47fd63b9..9e0f1af1 100644 --- a/libraries/chain/include/graphene/chain/database.hpp +++ b/libraries/chain/include/graphene/chain/database.hpp @@ -204,7 +204,7 @@ namespace graphene { namespace chain { /** this signal is emitted any time an object is removed and contains a * pointer to the last value of every object that was removed. */ - fc::signal&)> removed_objects; + fc::signal&, const vector&)> removed_objects; //////////////////// db_witness_schedule.cpp ////////////////////