Considerations for an Asynchronous Implementation

Thread Safety

These entry points to the node must be re-entrant:

• Ping
• Store
• FindNode
• FindValue

The first issue is with add contacts and the bucket manipulation that occurs. Collections should not be modified or otherwise manipulated while they are searched. We’ve seen the use of lock statements in previous code to ensure that collections are not modified asynchronously.

There are potentially more optimized approaches, such as locking only the specific KBucket being manipulated and only locking the BucketList when it itself is being modified; however, I will leave those for another time.

It is also assumed that the storage implementation can be re-entrant. In the virtual storage, this is handled by ConcurrentDictionary instances, for example:

protected ConcurrentDictionary<BigInteger, StoreValue> store;

Parallel queries

From the spec: “The initiator then sends parallel, asynchronous find_node RPCs to the a [sic] nodes it has chosen, a is a system-wide concurrency parameter, such as 3.”

In the lookup algorithm, Kademlia uses parallel, asynchronous queries to reduce timeout delays from failed nodes. Waiting for at least some of the nodes to respond in each batch of three closest nodes gives the system a chance to get even closer nodes to those first set of close nodes with the hope of acquiring k closer contacts without having to explore farther contacts.

It’s not particularly clear why all the k closer contacts aren’t queried in parallel to start with. Maybe the idea is that you want to try to get closer contacts from already close contacts. Certainly all the contacts could be queried and from the ones that respond first, we can select k closer ones. On the other hand, querying all the contacts simultaneously probably results in unnecessary network traffic as many of the FindNode RPC calls will be ignored.

The BaseRouter abstract class

For unit testing, it’s useful to keep the nonparallel implementation, but ideally, both parallel and nonparallel calls to the Router should be made in the same way. An abstract BaseRouter class allows for this.

Code Listing 88: The BaseRouter Class

The ParallelRouter

The ParallelRouter queues contacts to query, in addition to some other information each thread needs to know about when executing the RPC call.

Code Listing 89: ContactQueueItem

The ParallelRouter also initializes an internal thread pool.

Code Listing 90: InitializeThreadPool

Work is queued and a semaphore is released for a thread to pick up the work.

Code Listing 91: QueueWork

Each thread dequeues an item for work in the RpcCaller method.

Code Listing 92: RpcCaller

The salient point with the previous code is that when a value is found, it takes a snapshot of the current closer contacts and stores all the information about a closer contact in fields belonging to the ParallelLookup class.

The ParallelRouter must terminate its search after a certain amount of time, which handles unresponsive contacts. Whenever a response is received and new contacts are added to the list of contacts that can be queried, a timer is reset. The Lookup call exits when a value is found (for FindValue), or k closer contacts have been found, or the time period expires.

Code Listing 93: SetQueryTime

The Lookup inner loop is the where the work is done, as with the nonparallel version, but notice instead how work is queued and we wait for responses—particularly the check for whether we’ve waited long enough in the haveWork assignment.

Code Listing 94: haveWork

We can now take the Dht tests for the nonparallel version and create parallel versions of those tests, passing in the ParallelRouter instead:

Dht dht = new Dht(ID.RandomID, vp, () => new VirtualStorage(), new ParallelRouter());

The result is that the parallel router unit tests also pass.

Figure 9

A potential problem occurs when there are threads still waiting for a response, and that response possibly occurs at some point after the Lookup method exits. We deal with this in several ways:

1. Remove all pending queued work with DequeueRemainingWork(); 
2. Copy the closer contacts collection when executing the FindValue RPC call: foundContacts = item.CloserContacts.ToList();
3. Set a “stop work” flag that ignores any further FindValue RPC successful return values for the current lookup.
4. Use a local (per-lookup call) findResult instance.
5. Use a common locker object to prevent both closer/farther contacts from being updated and update the values in the findResult instance.
6. Use local closer/farther/contacted lists per lookup, such that any remaining thread will update only the collections associated with its current work item, which is why ContactQueueItem includes the closer/farther collections as well as the findResult instance in which to store any found value.

This ensures that even if there are threads still performing work on a previous lookup, they do not affect the results of the current lookup.

While the use of a single locker object for blocking updates to collections and updating the find value is slightly inefficient, it avoids using nested locks; otherwise the thread, when if finds a value, would technically have to lock both the closerContacts collection and the findResult instance. Nested locks should be avoided. Also note that the Lookup method itself is not intended to be re-entrant.