001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.camel.util.concurrent;
018
019import java.util.concurrent.Callable;
020import java.util.concurrent.CompletionService;
021import java.util.concurrent.DelayQueue;
022import java.util.concurrent.Delayed;
023import java.util.concurrent.Executor;
024import java.util.concurrent.Future;
025import java.util.concurrent.FutureTask;
026import java.util.concurrent.TimeUnit;
027import java.util.concurrent.atomic.AtomicInteger;
028
029/**
030 * A {@link java.util.concurrent.CompletionService} that orders the completed tasks in the same order as they where
031 * submitted.
032 */
033public class SubmitOrderedCompletionService<V> implements CompletionService<V> {
034
035    private final Executor executor;
036
037    // the idea to order the completed task in the same order as they where submitted is to leverage
038    // the delay queue. With the delay queue we can control the order by the getDelay and compareTo methods
039    // where we can order the tasks in the same order as they where submitted.
040    private final DelayQueue<SubmitOrderFutureTask> completionQueue = new DelayQueue<>();
041
042    // id is the unique id that determines the order in which tasks was submitted (incrementing)
043    private final AtomicInteger id = new AtomicInteger();
044    // index is the index of the next id that should expire and thus be ready to take from the delayed queue
045    private final AtomicInteger index = new AtomicInteger();
046
047    private class SubmitOrderFutureTask extends FutureTask<V> implements Delayed {
048
049        // the id this task was assigned
050        private final long id;
051
052        SubmitOrderFutureTask(long id, Callable<V> voidCallable) {
053            super(voidCallable);
054            this.id = id;
055        }
056
057        SubmitOrderFutureTask(long id, Runnable runnable, V result) {
058            super(runnable, result);
059            this.id = id;
060        }
061
062        @Override
063        public long getDelay(TimeUnit unit) {
064            // if the answer is 0 then this task is ready to be taken
065            long answer = id - index.get();
066            if (answer <= 0) {
067                return answer;
068            }
069            // okay this task is not ready yet, and we don't really know when it would be
070            // so we have to return a delay value of one time unit
071            if (TimeUnit.NANOSECONDS == unit) {
072                // okay this is too fast so use a little more delay to avoid CPU burning cycles
073                // To avoid align with java 11 impl of
074                // "java.util.concurrent.locks.AbstractQueuedSynchronizer.SPIN_FOR_TIMEOUT_THRESHOLD", otherwise
075                // no sleep with very high CPU usage
076                answer = 1001L;
077            } else {
078                answer = unit.convert(1, unit);
079            }
080            return answer;
081        }
082
083        @Override
084        @SuppressWarnings("unchecked")
085        public int compareTo(Delayed o) {
086            SubmitOrderFutureTask other = (SubmitOrderFutureTask) o;
087            return (int) (this.id - other.id);
088        }
089
090        @Override
091        protected void done() {
092            // when we are done add to the completion queue
093            completionQueue.add(this);
094        }
095
096        @Override
097        public String toString() {
098            // output using zero-based index
099            return "SubmitOrderedFutureTask[" + (id - 1) + "]";
100        }
101    }
102
103    public SubmitOrderedCompletionService(Executor executor) {
104        this.executor = executor;
105    }
106
107    @Override
108    public Future<V> submit(Callable<V> task) {
109        if (task == null) {
110            throw new IllegalArgumentException("Task must be provided");
111        }
112        SubmitOrderFutureTask f = new SubmitOrderFutureTask(id.incrementAndGet(), task);
113        executor.execute(f);
114        return f;
115    }
116
117    @Override
118    public Future<V> submit(Runnable task, Object result) {
119        if (task == null) {
120            throw new IllegalArgumentException("Task must be provided");
121        }
122        SubmitOrderFutureTask f = new SubmitOrderFutureTask(id.incrementAndGet(), task, null);
123        executor.execute(f);
124        return f;
125    }
126
127    @Override
128    public Future<V> take() throws InterruptedException {
129        index.incrementAndGet();
130        return completionQueue.take();
131    }
132
133    @Override
134    public Future<V> poll() {
135        index.incrementAndGet();
136        Future<V> answer = completionQueue.poll();
137        if (answer == null) {
138            // decrease counter if we didnt get any data
139            index.decrementAndGet();
140        }
141        return answer;
142    }
143
144    @Override
145    public Future<V> poll(long timeout, TimeUnit unit) throws InterruptedException {
146        index.incrementAndGet();
147        Future<V> answer = completionQueue.poll(timeout, unit);
148        if (answer == null) {
149            // decrease counter if we didnt get any data
150            index.decrementAndGet();
151        }
152        return answer;
153    }
154
155    /**
156     * Marks the current task as timeout, which allows you to poll the next tasks which may already have been completed.
157     */
158    public void timeoutTask() {
159        index.incrementAndGet();
160    }
161
162}