Recommendations for effective query blocking

Question

I would like to store tuple objects in a collection of matching java, and then have an efficient, blocking query method that returns the first element matching the pattern. If such an element is not available, it is blocked until such an element appears.

For example, if I have a class:

public class Pair { public final String first; public final String Second; public Pair( String first, String second ) { this.first = first; this.second = second; } } 

And a collection like:

 public class FunkyCollection { public void add( Pair p ) { /* ... */ } public Pair get( Pair p ) { /* ... */ } } 

I would like to request it as:

 myFunkyCollection.get( new Pair( null, "foo" ) ); 

which returns the first available pair with the second field equal to "foo", or blocks until this element is added. Another example request:

 myFunkyCollection.get( new Pair( null, null ) ); 

must return the first available pair regardless of its values.

Does a solution already exist? If this is not the case, what do you suggest implementing the get( Pair p ) method?

Clarification: The get( Pair p) method should also remove the item. The choice of name was not very smart. The best name would be take( ... ) .

Answer 1

Here is some source code. This is basically the same as cb160, but the source code can help resolve any issues you may have. In particular, methods on FunkyCollection must be synchronized.

As the Meriton noted, the get method performs an O (n) check for each blocked receipt every time a new object is added. It also performs O (n) operation to delete objects. This could be improved using a data structure similar to a linked list, where you can save the iterator to the last checked item. I did not provide source code for this optimization, but it should not be too difficult to implement if you need additional performance.

 import java.util.*; public class BlockingQueries { public class Pair { public final String first; public final String second; public Pair(String first, String second) { this.first = first; this.second = second; } } public class FunkyCollection { final ArrayList<Pair> pairs = new ArrayList<Pair>(); public synchronized void add( Pair p ) { pairs.add(p); notifyAll(); } public synchronized Pair get( Pair p ) throws InterruptedException { while (true) { for (Iterator<Pair> i = pairs.iterator(); i.hasNext(); ) { Pair pair = i.next(); boolean firstOk = p.first == null || p.first.equals(pair.first); boolean secondOk = p.second == null || p.second.equals(pair.second); if (firstOk && secondOk) { i.remove(); return pair; } } wait(); } } } class Producer implements Runnable { private FunkyCollection funkyCollection; public Producer(FunkyCollection funkyCollection) { this.funkyCollection = funkyCollection; } public void run() { try { for (int i = 0; i < 10; ++i) { System.out.println("Adding item " + i); funkyCollection.add(new Pair("foo" + i, "bar" + i)); Thread.sleep(1000); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } public void go() throws InterruptedException { FunkyCollection funkyCollection = new FunkyCollection(); new Thread(new Producer(funkyCollection)).start(); System.out.println("Fetching bar5."); funkyCollection.get(new Pair(null, "bar5")); System.out.println("Fetching foo2."); funkyCollection.get(new Pair("foo2", null)); System.out.println("Fetching foo8, bar8"); funkyCollection.get(new Pair("foo8", "bar8")); System.out.println("Finished."); } public static void main(String[] args) throws InterruptedException { new BlockingQueries().go(); } } 

Output:

 Fetching bar5. Adding item 0 Adding item 1 Adding item 2 Adding item 3 Adding item 4 Adding item 5 Fetching foo2. Fetching foo8, bar8 Adding item 6 Adding item 7 Adding item 8 Finished. Adding item 9 

Please note that I put everything in one source file to make it easier to run.

Answer 2

I do not know the existing container that will provide this behavior. One of the problems you encounter is when no existing record matches the query. In this case, you have to wait for new entries to appear, and these new entries should appear in the tail of the sequence. Given that you are blocking, you do not want to check all records prior to the last addition, because you have already checked them and determined that they do not match. Therefore, you need a way to record your current position and the ability to search forward from there when a new record arrives.

This expectation is the work of Condition . As suggested in cb160's answer , you should highlight the Condition instance inside your collection and lock it with Condition#await() . You should also set the concurrent message overload to your get() method to allow timeouts:

 public Pair get(Pair p) throws InterruptedException; public Pair get(Pair p, long time, TimeUnit unit) throws InterruptedException; 

Each time add() called, call Condition#signalAll() to unlock threads waiting for unsatisfied get() requests, which allows them to scan for the latest additions.

You did not specify how and when items are removed from this container. If the container is only growing, it simplifies how the threads can scan its contents without worrying about competing with other threads mutating the container. Each thread can confidently start its request regarding the minimum number of records available for verification. However, if you allow the removal of objects, you still have many problems.

Answer 3

In your FunkyCollection add method, you can call notifyAll in the collection itself each time you add an item.

In the get method, if the base container (Any suitable conatiner is fine) does not contain the required value, wait on the FunkyCollection. When the wait is notified, check if the base container contains the desired result. If so, return the value, otherwise wait again.

Answer 4

You seem to be looking for an implementation of Tuple Spaces. The Wikipedia article about them contains several implementations for Java, maybe you can use one of them. Otherwise, you may find an open source implementation to follow or related research documents.