Simple concurrent_unordered_set

[jmjatlanta]

Potential fix for #1256

This code is for testing only. I would certainly imagine that optimizations can be found. Please do not do a full code review yet. Please take a look at the approach and comment here.

[abitmore]

Perhaps try the Intel TBB concurrent_unorderred_set which is Apache 2.0 licensed?

• source: https://github.com/01org/tbb
• docs: https://www.threadingbuildingblocks.org/documentation

Or perhaps the Microsoft implementation ?

[pmconrad]

Back in my university days I was in a class called "System programming concepts and techniques" or sth like that. One day, the professor was talking about mutexes, and gave us a pseudocode example for a simple producer/consumer queue. One of the students pointed out an error in the code. The professor admitted (with obvious frustration) that he was giving out slightly different versions of that example code every year, because every year someone would find a new bug in his logic.

What I'm trying to say with this is - although I haven't even checked your code, please use a commonly available solution instead of rolling your own. :-)

[jmjatlanta]

I agree (the roll-your-own part), but was surprised to find not too many baked versions of an unordered set. The Intel TBB was the only one I found. I haven't looked over the MS solution.

Does anyone have a problem with the Apache license? I'm no lawyer.

The begin() and end() stuff is the most troublesome. Things will possibly have to be locked until the end of the loop, unless we examine each loop individually, looking for optimizations.

Often the implementation is "make a lambda as small as possible, acquire lock with lock_guard, call std::for_each"

BTW: concurrency/parallel programming was a good subject at cppcon2017. Search "Is Parallel Programming Still Hard" on YouTube.

[jmjatlanta]

Update: I have begun to include Intel TBB as part of the build process (different branch in case we decide against it). The default (and highly recommended by Intel) is to build shared libraries due to scheduler issues.

There is an undocumented (on purpose) workaround to build and link statically. My question to all is should we use it?

Yet another update: I added TBB as a submodule, using static linking, and made a PR: #1315

[OpenLedgerDev]

Let's imagine simple code snippets executed from different threads.

get_first_object_name() {
   auto first = shared_concurrent_unordered_set.begin();
   return first->name();
}
clear() {
   shared_concurrent_unordered_set.clear();
}

so, i see the problem here

thread 1 : auto first = shared_concurrent_unordered_set.begin();  // get first object iterator under lock
thread 2 : shared_concurrent_unordered_set.clear();  // remove all objects under lock, all iterators will 
be invalid after that
thread 1 : return first->name();   // access by invalid iterator, segfault

[jmjatlanta]

so, i see the problem here

Yes, clear() is certainly a problem. But even erase() can be mean. Imagine thread A iterating, and the collection sees the next one. Just before it grabs it, thread B deletes it. In some implementations of iterators, having thread B delete any element prior to where A is at can cause problems.

I have a test program where I torture an unordered_set with 3 threads. Thread A adds an element, thread D deletes the first element if one exists (using unordered_set.begin()), and Thread F attempts a find(). All do their jobs at slightly different speeds. It only takes about 4 seconds before the program segfaults. Stacktrace normally points to find()

If you put locks in place, the program will continue to run without problems.

The "hard" part of fixing this is the iterators. Something intelligent must be done to either

• (A) "do the right thing" when an iterator wants to iterate, but something bad happened
• (B) have exclusive access to the entire collection during iteration
• (C) lock a portion of the collection so that iterators cannot traverse that area

All of those options require examining each iteration loop to determine its purpose. Then a decision can be made to either use A, B, or C.

C is an interesting option. You would think it would be the best option ( performant, lack of contention ). But depending on the implementation and use, it could actually be a bad choice ( increase in complexity, decrease in performance ).

[pmconrad]

https://github.com/bitshares/bitshares-core/wiki/Threading lists a number of tasks handled by the P2P thread. Can you document which collections are used by which tasks and in what way? Would be a good starting point for a high-level documentation of the P2P layer.

[jmjatlanta]

Can you document which collections are used by which tasks and in what way? Would be a good starting point for a high-level documentation of the P2P layer.

Here is what I found. I am posting here, as it will probably need to be reformatted to fit the goals of the wiki. Please let me know if you are looking for something different.

This has to do with _handshaking_connections, _active_connections, _closing_connections, and _terminating connections which are unordered sets in libraries/net/node.cpp.

_active_connections contains peer connections that are completely set up.
_handshaking_connections contains peer connections that are in the negotiation process.
_closing_connections peers that we are attempting to close
_terminating_connections peers that we have attempted to close, but seem to not be responding.

Methods that use the collections above, and how they use them:

node_impl destructor:
During destruction of the node object, all entries in _active_connections are stored as a potential_peer_record.

fetch_sync_items_loop:
Each peer in _active_connections is checked. If the peer is idle and we have something to send and we have not already
scheduled something, we schedule a request.

is_item_in_any_peers_inventory:
Checks all _active_connections to see if they have advertized a particular item as in their inventory.

fetch_items_loop:
This loops through _active_connections looking for previously requested items, so as to build a request to send to that peer. This method tries to be smart and removes items that were requested so long ago that the foreign node probably no longer has it in its cache.

advertise_inventory_loop:
For all peers in _active_connections that are in sync with us, advertise our inventory

terminate_inactive_connections_loop:
For peers in _active_connections or _handshaking_connections that are slow to respond, or not responding, terminate the connection. For peers in _closing_connections that are taking too long to close, forcibly close them. For peers in _terminating_connections (which contains peers that we have previously attempted to close nicely), terminate them.

fetch_updated_peer_lists_loop:
For peers in _active_connections, request the addresses of their _active_connections

schedule_peer_for_deletion:
Validates that the peer is no longer in any set (_handshaking_connections, _active_connections, closing_connections, terminating_connections).

on_address_request_message:
Responds to foreign node's request for a list of our _active_connections

get_number_of_connections:
Returns size of _handshaking_connections + _active_connections.

get_peer_by_node_id:
Find the peer matching the id by looking in _active_connections or _handshaking_connections.

is_already_connected_to_id:
Determine if this node id is already in _active_connections or _handshaking_connections.

display_current_connections:
Sends information about peers in _active_connections and _handshaking_connections to log file.

calculate_unsynced_block_count_from_all_peers:
For all peers in _active_connections, total all unfetched items.

on_blockchain_item_ids_inventory_message:
When we receive an item, flush it if we have already received it.

on_item_ids_inventory_message:
Checks each inventory item, then checks each peer in _active_connections to see if we advertised it or requested it.

on_connection_closed:
removes peer from _closing_connections, _handshaking_connections, _terminating_connections, and then updates the potential_eer_record with the last_seen_time, before deleting from _active_connections.

send_sync_block_to_node_delegate:
Sends blocks to _active_connections. Interestingly, keeps track of hardfork blocks, in an attempt to notify foreign peer to the need to upgrade (which it then promptly disconnects). If an invalid message is received, which indicates that the foreign peer did not accept the block, it disconnects from the peer.

process_backlog_of_sync_blocks:
For peers in _active_connections, looks to see if this block is the next block on the chain (or one of the forks). It does this by checking each peer to determine if we are in sync with that peer or not. This method also iterates through _active_connections again to mark the item as processed.

process_block_during_normal_operation:
For peers in _active_connections, if they have advertized this block, update the peer object to that block id. This method also contains hardfork code similar to send_sync_block_to_node_delegate. It also disconnects nodes that send invalid responses.

forward_firewall_check_to_next_available_peer:
For peers in _active_connections, propogates a firewall check (if they haven't already tested that peer).

on_get_current_connections_request_message:
Gathers statistics for each peer in _active_connections, and sends it to the peer that requested it.

start_synchronizing:
Begins syncronizing with each peer in _active_connections

new_peer_just_added:
checks _active_connections.size() to see if the connection_count_change event should be fired.

close:
destroys connections on all peers in _active_connections, _handshaking_connections, and _closing_connections

accept_loop:
Accept incoming connection, adds to _handshaking_connections.

connect_to_task:
If the connection failed, erase from _active_connections and _closing_connections

get_connection_to_endpoint:
Given an endpoint, find the related peer in _active_connections or _handshaking_connections.

move_peer_to_active_list:
Removes a peer from _handshaking_connections and places it in _active_connections.

move_peer_to_closing_list:
Removes a peer from _handshaking_connections, _active_connections, or _terminating_connections, and places it in _closing_connections.

move_peer_to_terminating_list:
Removes a peer from _handshaking_connections, _active_connections, or _closing_connections, and places it in _terminating_connections.

dump_node_status:
Places statistics of each node in _active_connections in the log file.

get_connected_peers:
Returns a vector of peer_status objects that contain info about each peer in _active_connections.

get_connection_count:
Return the size of _active_connections.

is_connected:
Return true if _active_connections is not empty.

set_advanced_node_parameters:
Disconnects from peers in _active_connections until the maximum_number_of_connections is not exceeded.

set_allowed_peers:
Disconnects any peers in _active_connections that are not in the _allowed_peers list.

[pmconrad]

Thanks.
I think it is possible that while a task is iterating through one of these collections it will yield to another task. That other task would successfully acquire the recursive lock (because it is running in the same thread). That other task could then access the same collection in a destructive way, IOW the locks don't work like this.
Can you confirm or deny that this is the case?

[jmjatlanta]

That other task would successfully acquire the recursive lock (because it is running in the same thread).

I do not believe that is possible. I believe that would violate memory/process boundary rules. I will investigate.

[jmjatlanta]

I think it is possible that while a task is iterating through one of these collections it will yield to another task.

How would it yield? Through std::this_thread::yield()? This yields to other threads. It simply allows this_thread to be rescheduled. It does not allow this_thread to jump to another section of code.

Or perhaps I am talking threads and you are talking tasks. Is there a difference?

[pmconrad]

fc::thread::yield pauses the current task and continues another ready task, if one is available. This could happen (maybe) if the current task is waiting for I/O for example .

[jmjatlanta]

fc::thread::yield pauses the current task and continues another ready task, if one is available. This could happen (maybe) if the current task is waiting for I/O for example .

Thank you for straightening me out. I am digging through boost::context to see how this works. From what I can tell, this would take an explicit call to yield on our (core or fc) part. Some things are still unclear. I am still digging.

[jmjatlanta]

Latest notes: I have played in my sandbox and verified that what @pmconrad has said is correct. std::recursive_mutex is unaware of fc::thread. As such, it can successfully get a lock when it is not supposed to. In short, this code is not thread safe.

fc::thread uses primitives that are part of boost::context to act like threads, but provide lightweight context switching. Therefore, an fc::thread can yield, and the underlying user thread can be used for something else, even a similar call that will attempt a lock on a recursive_mutex. This may succeed, as it could be the same user thread.

The boost primitives used by fc::thread have also been used to build out boost fibers. Boost fibers include much of the functionality of fc::thread, and include a recursive_mutex that is "fiber-aware".

There are now several ways to go about fixing this issue:

1. Use boost::fibers
   • Requires an overhaul (or elimination) of fc::thread
   • Afterward would be less code to maintain
   • More reliance on boost ( with associated pros and cons )
   • Would need all users to be on newer versions of Boost
2. Make a recursive mutex that can be used with fc::thread
   • More code to maintain and test
   • Probably the quickest fix
   • Further from "standard"
3. Wait for coroutines
   • Much of the benefits of fc::thread and boost::fibers are also in the coming std::coroutine pieces
   • That standard is not published yet
   • Compiler implementations will need time after standards publication
   • Could possibly use boost coroutines in the hopes of being closer to the standard, then possibly convert to standard later. Update: boost coroutines work differently, I do not think implementing boost coroutines would help.

None of these seem appealing. Cost and benefits need to be thought about. I will pause here to welcome other comments / suggestions / ideas.

[pmconrad]

AFAICS fc::mutex allows recursive access from current context.

[jmjatlanta]

AFAICS fc::mutex allows recursive access from current context.

Yes, it does. I have created a test that proves it. I may check it in to FC, but not sure if it is worth it. See https://gist.github.com/jmjatlanta/6624f524214b7c72184c36e43e346848

I will switch the concurrent_unordered_set to use fc::mutex and see if it works (update: it does).

[jmjatlanta]

Closing this PR, but will keep the branch. This does not seem to solve the problem it was designed for. Will take another look later.

[abitmore]

I thought this PR did fix something, if it's harmless, we can merge it.

[jmjatlanta]

It does make the internal sets thread-safe. It does not fix issue #1256. In short, it probably fixes an issue if these sets are accessed from multiple threads. I think we have yet to prove that is the case.

To merge it would require more testing. Being that such testing is difficult to do thoroughly, it was thought best to close this ticket, leave the branch, and revisit when we have a clearer picture of what is causing #1256 (or another).

[abitmore]

Merged this to dbg331b branch for testing in mainnet, and dbgtest333b for testnet. If no issue is found, I suggest we include it in next release.

[abitmore]

Need to remove ASSERT_TASK_NOT_PREEMPTED(); from code where added the lock, otherwise it will lead to crashes in debug build. See #1256 (comment).

[abitmore]

I've been running this in production for a few months, it seems to be working fine, so I will merge it.
Thanks to @jmjatlanta and the reviewers/testers.