Implement swap method for indexed parallel skip list

Question

I implement a Java concurrentSkipListMap parallel skip list map, the differences are that I want the list to allow duplication, and I also want this list to be indexable (so finding the Nth element of the list takes O (lg (n) ) time, instead of O (n) time, as with the standard skip list). These changes are not a problem.

In addition, the keys of the skip list are changed. For example, if the list items are integers {0, 4, 7}, then the key of the middle item can be changed to any value in [0, 7] without asking to change the list structure; if the key is changed to (-inf, -1] or [8, + inf), then the item is deleted and re-added to maintain the order of the list. Instead of implementing this as a deletion, followed by an O (lg (n)) insert, I will implement this as a deletion, followed by a linear walk, followed by an O (1) insert (with an expected execution time of O (1) - 99% time node will be replaced by an adjacent node).

Inserting a completely new node is rare (after starting) and deleting a node (without immediately re-adding it) never occurs; almost all operations are elementAt (i) to retrieve an element in the i-th index or operations to replace nodes after changing the key.

The problem I am facing is how to implement the key modification class. Conceptually, I would like to do something like

public class Node implements Runnable { private int key; private Node prev, next; private BlockingQueue<Integer> queue; public void update(int i) { queue.offer(i); } public void run() { while(true) { int temp = queue.take(); temp += key; if(prev.getKey() > temp) { // remove node, update key to temp, perform backward linear traversal, and insert } else if(next.getKey() < temp) { // remove node, update key to temp, perform forward linear traveral, and insert } else { key = temp; // node doesn't change position } } } } 

(The insert -method called from run uses CAS to solve the problem of two nodes trying to insert at the same place at the same time (similar to how ConcurrentSkipListMap handles conflicting inserts) - conceptually this is the same as if the first node blocked the nodes adjacent to the insertion point, except that the overhead is reduced when there is no conflict.)

Thus, I can guarantee that the list is always in order (this is normal if the key update is delayed a bit, because I can be sure that the update will eventually happen, however if the list is unordered, then everything can go fragile). The problem is that implementing the list in this way will lead to the creation of a large number of threads, one on Node (with several thousand nodes in the list) - most of them will be blocked at any given time, but I'm interested in several thousand blocking threads still lead to too high costs.

Another option is to synchronize the update method and remove the Runnable interface from Node , so that instead of having two threads that depend on updates in Node , which then process these updates on their own separate thread, two threads will instead run with using the Node#update method. The problem is that this could potentially create a bottleneck; if eight different threads decide to update the same node at the same time, then the queue implementation will scale just fine, but a synchronized implementation will block seven out of eight threads (and then will block six threads, then five, etc.).

So my question is how to implement something like a queue implementation, except for a reduced number of threads, or how to implement something like a synchronized implementation, except without a potential bottleneck problem.

Answer 1

I think I can solve this with ThreadPoolExecutor , something like

 public class Node { private int key; private Node prev, next; private ConcurrentLinkedQueue<Integer> queue; private AtomicBoolean lock = new AtomicBoolean(false); private ThreadPoolExecutor executor; private UpdateNode updater = new UpdateNode(); public void update(int i) { queue.offer(i); if(lock.compareAndSet(false, true)) { executor.execute(updater); } } private class UpdateNode implements Runnable { public void run() { do { try { int temp = key; while(!queue.isEmpty()) { temp += queue.poll(); } if(prev.getKey() > temp) { // remove node, update key to temp, perform backward linear traversal, and insert } else if(next.getKey() < temp) { // remove node, update key to temp, perform forward linear traveral, and insert } else { key = temp; // node doesn't change position } } finally { lock.set(false); } } while (!queue.isEmpty() && lock.compareAndSet(false, true)); } } } 

Thus, I have the advantages of a UpdateNode approach without blocking thousands of threads - instead, I execute UpdateNode every time I need to update the node (if UpdateNode no longer running on this Node , therefore AtomicBoolean acting as a lock) and rely on ThreadPoolExecutor to make it inexpensive to create several thousand runnables.