Difference between revisions of "DSDP/DD/DSF Concurrency"

Revision as of 15:16, 22 March 2007

DSF Concurrency Model

Version 1.0
Pawel Piech
© 2006, Wind River Systems.  Release under EPL version 1.0.

Introduction

Providing a solution to concurrency problems is the primary design goal of DSF. To that end DSF imposes a rather draconian restriction on services that use it: 1) All service interface methods must be called using a single designated dispatch thread, unless explicitly stated otherwise, 2) The dispatch thread should never be used to make a blocking call (a call that waits on I/O or a call that makes a long-running computation). What the first restriction effectively means, is that the dispatch thread becomes a global "lock" that all DSF services in a given session share with each other, and which controls access to most of services' shared data. It's important to note that multi-threading is still allowed

within individual service implementation. but when crossing the service interface boundaries, only the dispatch thread can be used. The second restriction just ensures that the performance of the whole system is not killed by one service that needs to read a huge file over the network. Another way of looking at it is that the service implementations practice co-operative multi-threading using the single dispatch thread.

There are a couple of obvious side effects that result from this rule:

1. When executing within the dispatch thread, the state of the services is guaranteed not to change. This means that thread-defensive programming techniques, such as making duplicates of lists before iterating over them, are not necessary. Also it's possible to implement much more complicated logic which polls the state of many objects, without the worry about dead-locks.
2. Whenever a blocking operation needs to be performed, it must be done using an asynchronous method. By the time the operation is completed, and the caller regains the dispatch thread, this caller may need to retest the relevant state of the system, because it could change completely while the asynchronous operation was executing.

The Mechanics

java.util.concurrent.ExecutorService

DSF builds on the vast array of tools added in Java 5.0's java.util.concurrent package (see Java 5 concurrency package API for details), where the most important is the ExecutorService interface. ExecutorService is a formal interface for submitting Runnable objects that will be executed according to executor's rules, which could be to execute the Runnable immediately, within a thread pool, using a display thread, etc. For DSF, the main rule for executors is that they have to use a single thread to execute the runnable and that the runnables be executed in the order that they were submitted. To give the DSF clients and services a method for checking whether they are being called on the dispatch thread, we extended the ExecutorService interface as such:

public interface DsfExecutor extends ScheduledExecutorService
{
    /**
     * Checks if the thread that this method is called in is the same as the
     * executor's dispatch thread.
     * @return true if in DSF executor's dispatch thread
     */
    public boolean isInExecutorThread();
}

java.lang.concurrent.Future vs org.eclipse.dd.dsf.concurrent.Done

The Done object encapsulates the return value of an asynchronous call in DSF. It is actually merely a Runnable with an attached org.eclipse.core.runtime.IStatus object , but it can be extended by the services or clients to hold whatever additional data is needed. Typical pattern in how the Done object is used, is as follows:

Service:
    public class Service {
        void asyncMethod(Done done) {
            new Job() {
                public void run() {
                    // perform calculation                    
                    ... 
                    done.setStatus(new Status(IStatus.ERROR, ...));
                    fExecutor.execute(done);
                }
            }.schedule();
        }
    }

Client:
    ...
    Service service = new Service();
    final String clientData = "xyz";
    ...
    service.asynMethod(new Done() {
        public void run() {
            if (getStatus().isOK()) {
                // Handle return data
                ...
            } else {
	        // Handle error
                ...
            }
        }
    }

The service performs the asynchronous operation a background thread, but it can still submit the Done runnable with the executor. In other words, the Done and other runnables can be submitted from any thread, but will always execute in the single dispatch thread. Also if the implementation of the asyncMethod() is non-blocking, it does not need to start a job, it could just perform the operation in the dispatch thread. On the client side, care has to be taken to save appropriate state before the asynchronous method is called, because by the time the Done is executed, the client state may change.

The java.lang.concurrent package doesn't already have a Done, because the generic concurrent package is geared more towards large thread pools, where clients submit tasks to be run in a style similar to Eclipse's Jobs, rather than using the single dispatch thread model of DSF. To this end, the concurrent package does have an equivalent object, Future. Future has methods that allows the client to call the get()

method, and block while waiting for a result, and for this reason it cannot be used from the dispatch thread. But it can be used, in a limited way, by clients which are running on background thread that still need to retrieve data from synchronous DSF methods. In this case the code might look like the following:

Service:
    public class Service {
        int syncMethod() {
	    // perform calculation
            ...
            return result;
        }
    }

Client:
    ...
    DsfExecutor executor = new DsfExecutor();
    final Service service = new Service(executor);
    Future<Integer> future = executor.submit(new Callable<Integer>() {
        Integer call() {
            return service.syncMethod();
        }
    });
    int result = future.get();

The biggest drawback to using Future with DSF services, is that it does not work with asynchronous methods. This is because the Callable.call() implementation has to return a value within a single dispatch cycle. To get around this, DSF has an additional object called DsfQuery, which works like a Future combined with a Callable, but allows the implementation to make multiple dispatches before setting the return value to the client. The DsfQuery object works as follows:

1. Client creates the query object with its own implementation of DsfQuery.execute().
   
2. Client calls the DsfQuery.get() method on non-dispatch thread, and blocks.
3. The query is queued with the executor, and eventually the DsfQuery.execute() method is called on the dispatch thread.
4. The query DsfQuery.execute() calls synchronous and asynchronous methods that are needed to do its job.
5. The query code calls DsfQuery.done() method with the result.
6. The DsfQuery.get() method un-blocks and returns the result to the client.
   

Slow Data Provider Example

See example for application of the concurrency model.

Cancellability

Table Viewer

Unlike coalescing, which can be implemented entirely within the service, cancellability requires that the client be modified as well to take advantage of this capability. For the table viewer content provider, this means that additional features have to be added. In CancellingSlowDataProviderContentProvider.java

ILazyContentProvider.updateElement() was changes as follows:

    public void updateElement(final int index) {
        assert fTableViewer != null;
        if (fDataProvider == null) return;
        
        // Calculate the visible index range.
        final int topIdx = fTableViewer.getTable().getTopIndex();
        final int botIdx = topIdx + getVisibleItemCount(topIdx);
        
        fCancelCallsPending.incrementAndGet();
        fDataProvider.getExecutor().execute(
            new Runnable() { public void run() {
                // Must check again, in case disposed while redispatching.
                if (fDataProvider == null || fTableViewer.getTable().isDisposed()) return;
                if (index >= topIdx && index <= botIdx) {
                    queryItemData(index);
                }
                cancelStaleRequests(topIdx, botIdx);
            }});
    }

Now the client keeps track of the requests it made to the service in fItemDataDones, and above, cancelStaleRequests() iterates through all the outstanding requests and cancels the ones that are no longer in the visible range.

Data Provider Service

The data provider implementation (CancellableInputCoalescingSlowDataProvider.java), builds on top of the coalescing data provider. To make the canceling feature useful, the data provider service has to limit the size of the request queue. This is because in this example which simulates communication with a target and once requests are filed into the request queue, they cannot be canceled, just like a client can't cancel request once it sends them over a socket. So instead, if a flood of getItem() calls comes in, the service has to hold most of them in the coalescing buffer in case the client decides to cancel them. Therefore the fileBufferedRequests() method includes a simple check before servicing the buffer, and if the request queue is full, the buffer servicing call is delayed.

     
        if (fQueue.size() >= REQUEST_QUEUE_SIZE_LIMIT) {
            if (fGetItemIndexesBuffer.isEmpty()) {
                fExecutor.schedule(
                    new Runnable() { public void run() {
                        fileBufferedRequests();
                    }},
                    REQUEST_BUFFER_FULL_RETRY_DELAY, 
                    TimeUnit.MILLISECONDS);
            }
            return;
        } 

Beyond this change, the only other significant change is that before the requests are queued, they are checked for cancellation.

Final Notes

The example given here is fairly simplistic, and chances are that the same example could be implemented using semaphores and free threading with perhaps fewer lines of code. But what we have found is that as the problem gets bigger, the amount of features in the data provider increases, the state of the communication protocol gets more complicated, and the number of modules needed in the service layer increases, using free threading and semaphores does not safely scale. Using a dispatch thread for synchronization certainly doesn't make the inherent problems of the system less complicated, but it does help eliminate the race conditions and deadlocks from the overall system.

Coalescing and Cancellability are both optimizations. Neither of these optimizations affected the original interface of the service, and one of them only needed a service-side modification. But as with all optimizations, it is often better to first make sure that the whole system is working correctly and then add optimizations where they can make the biggest difference in user experience.

The above examples of optimizations can take many forms, and as mentioned with coalescing, caching data that is retrieved from the data provider is the most common form of data coalescing. For cancellation, many services in DSF build on top of other services, which means that even a low-level service can cause a higher level service to retrieve data, while another event might cause it to cancel those requests. The perfect example of this is a Variables service, which is responsible for calculating the value of expressions shown in the Variables view. The Variables service reacts to the Run Control service, which issues a suspended event and then requests a set of variables to be evaluated by the debugger back end. But as soon as a resumed event is issued by Run Control, the Variables service needs to cancel the pending evaluation requests.