0
\$\begingroup\$

I have concurrency tasks in my module: user code is able to instantiate my object and specify Executor service, then user can submit requests to this object which should be run on specified executor. I'm calling this object requests queue. Also, this object holds internal state, but it's not relevant for this question. I have these abstractions:

interface Request {
  void process(SomeState state);
}

class RequestsQueue {

  public RequestsQueue(Executor exec) {
    this.exec = exec;
  }

  void accept(Request req) {
    //
  }
}

The problem is that user is able to submit multiple requests simultaneously from different threads and all requests should be processed sequentially.

I'm using the combination of AtomicBoolean and ConcurrentLinkedQueue to achieve it:

class RequestsQueue implements Runnable {
    // queue of requests
    private final Queue<Request> queue = new ConcurrentLinkedQueue<>();

    // atomic running state flag
    private final AtomicBoolean running = new AtomicBoolean();

    // user specified executor
    private final Executor exec;

    // some internal state
    private final SomeState state;

    public RequestsQueue(SomeState state, Executor exec) {
        this.state = state;
        this.exec = exec;
    }

    @Override
    public void run() {
        // run this task until internal state will be marked as done by request
        while (!this.state.done()) {
            Request next = this.queue.poll();
            // if no next item, try to exit the loop
            boolean empty = next == null;
            if (empty) {
                // try to recover: user may submit next request between
                //   checking for empty and running.set(false) call,
                //   this is why I'm checking next item and running state twice

                // mark this loop as finished
                this.running.set(false);
                // check if any next item available
                next = this.queue.peek();
                empty = next == null;
                // if next item available and this loop is still not running
                // continue running this loop and process next item
                if (!empty && this.running.compareAndSet(false, true)) {
                    if (this.state.done()) {
                        // done - exit loop, perform cleanup
                        break;
                    }
                    // remove peeked item
                    this.queue.remove(next);
                } else {
                    // exit if empty or acquired by next loop, don't cleanup,
                    // wait for next request
                    return;
                }
            }
            next.process(this.state);
        }
        // some cleanup
        // ...

        // mark as not running
        this.running.set(false);
    }

    public void accept(final Request req) {
        if (this.state.done()) {
            return;
        }
        // add request to the queue
        this.queue.add(req);
        // start this task with queue loop if it is not running already
        if (this.running.compareAndSet(false, true)) {
            this.exec.execute(this);
        }
    }
}

Why such implementation: I tried to avoid synchronization and locks here, since expected use case is when the user will submit many requests before first will complete, and the queue will be populated without exiting from first task. Also, the amount of requests may be huge enough, so synchronizing each request may hurt performance a lot. So usually all requests will be processed within one loop without starting multiple tasks. But it's also possible that user will have some delay and tasks will be submitted to executor when previous exited the loop.

I wrote some unit tests to verify different scenarios and it seems everything works fine. But I'm not sure if I missed something. Is it enough to use double check the concurrent queue with conjunction of atomic boolean to be sure that none of the requests was lost? And can I be sure that all the requests will be processed without hanging in queue with stopped while-loop?

\$\endgroup\$
  • \$\begingroup\$ Shouldn't a SingleThreadExecutor fulfill your needs? \$\endgroup\$ – Xtreme Biker May 19 at 9:20
  • \$\begingroup\$ @XtremeBiker no, I can't specify executor, since it's configured from client side by protocol, the user (client programmer) will use shared fixed size thread pool for multiple different RequestsQueues. \$\endgroup\$ – g4s8 May 19 at 9:41
  • \$\begingroup\$ I don't understand it, so you only can process one request at the same time but you decide to do your own implementation because your client's implementation? As far as I see what you want to do here is FIFO processing, so then what's the problem? \$\endgroup\$ – Xtreme Biker May 19 at 10:22
  • \$\begingroup\$ Keep in mind that a Java ExecutorService provides the submit feature, that's not "execute this now", which is what you're doing in your code, but "enqueue this to be executed as soon as you can and notify me with the result". You just need to implement Callable instead of Runnable to use it. \$\endgroup\$ – Xtreme Biker May 19 at 10:26
  • \$\begingroup\$ @XtremeBiker actual implementation is a bit complexer than in the example, I removed all non relevant details from this implementation. You can check origin class here: github.com/g4s8/rio/blob/master/src/main/java/wtf/g4s8/rio/file/… \$\endgroup\$ – g4s8 May 19 at 17:44

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.