bundles/org.simantics.db.common/src/org/simantics/db/common/utils/RequestUtil.java

   1 package org.simantics.db.common.utils;
   2
   3 import java.util.concurrent.Semaphore;
   4 import java.util.concurrent.TimeUnit;
   5 import java.util.concurrent.atomic.AtomicInteger;
   6
   7 import org.simantics.db.ReadGraph;
   8 import org.simantics.db.Session;
   9
  10 import org.simantics.db.exception.DatabaseException;
  11 import org.simantics.db.procedure.Procedure;
  12 import org.simantics.db.request.Read;
  13
  14 /**
  15  * @author Tuukka Lehtonen
  16  */
  17 public final class RequestUtil {
  18
  19         private static final boolean TRACE_SCHEDULE = false;
  20         private static final boolean TRACE_WAITED_TIME = false;
  21         private static final boolean TRACE_TIMEOUT  = false;
  22         private static final boolean TRACE_RESULT_TIME = false;
  23
  24         static class ResultProcedure<R> extends Semaphore implements Procedure<R> {
  25
  26                 private static final long serialVersionUID = 8645926264625496924L;
  27
  28                 private static final Object NO_RESULT = new Object();
  29                 private Object resultOrException = NO_RESULT;
  30
  31                 public ResultProcedure() {
  32                         super(0);
  33                 }
  34
  35                 @Override
  36                 public void execute(R result) {
  37                         resultOrException = result;
  38                         release();
  39                 }
  40
  41                 @Override
  42                 public void exception(Throwable t) {
  43                         resultOrException = t;
  44                         release();
  45                 }
  46
  47                 @SuppressWarnings("unchecked")
  48                 public R waitForResultOrException(long timeout, R timeoutResult) throws DatabaseException, InterruptedException {
  49                         long nt0 = System.nanoTime();
  50                         if (timeout < 0) {
  51                                 acquire();
  52                         } else {
  53                                 tryAcquire(timeout, TimeUnit.MILLISECONDS);
  54                         }
  55                         Object r = resultOrException;
  56                         if (TRACE_WAITED_TIME)
  57                                 System.out.println("slept waiting for result for " + ((double) (System.nanoTime() - nt0) / 1e6) + " ms");
  58                         if (r == NO_RESULT) {
  59                                 if (TRACE_TIMEOUT)
  60                                         System.out.println("timed out (" + timeout + " ms) waiting for request result, returning default " + timeoutResult);
  61                                 return timeoutResult;
  62                         }
  63                         if (r instanceof Throwable) {
  64                                 if (r instanceof DatabaseException)
  65                                         throw (DatabaseException) r;
  66                                 throw new DatabaseException((Throwable) r);
  67                         }
  68                         if (TRACE_RESULT_TIME)
  69                                 System.out.println("got result in " + ((double) (System.nanoTime() - nt0) / 1e6) + " ms");
  70                         return (R) r;
  71                 }
  72
  73         }
  74
  75         static class TimeoutingRequest<R> extends Semaphore implements Read<R> {
  76
  77                 private static final long serialVersionUID = -5216095211800988013L;
  78
  79                 Read<R> wrappedRequest;
  80                 Procedure<R> wrappedProcedure;
  81                 /**
  82                  * 0 = not running, not timed out
  83                  * 1 = running
  84                  * 2 = timed out
  85                  */
  86                 AtomicInteger state = new AtomicInteger();
  87
  88                 public TimeoutingRequest(Read<R> request, Procedure<R> procedure) {
  89                         super(0);
  90                         this.wrappedRequest = request;
  91                         this.wrappedProcedure = procedure;
  92                 }
  93
  94                 @Override
  95                 public R perform(ReadGraph graph) throws DatabaseException {
  96                         if (state.compareAndSet(0, 1)) {
  97                                 release();
  98                                 return wrappedProcedure != null
  99                                                 ? graph.syncRequest(wrappedRequest, wrappedProcedure)
 100                                                 : graph.syncRequest(wrappedRequest);
 101                         } else {
 102                                 if (TRACE_TIMEOUT)
 103                                         System.out.println("[" + System.currentTimeMillis() + "] canceling execution of request " + wrappedRequest + ", startup timed out");
 104                                 release();
 105                                 return null;
 106                         }
 107                 }
 108
 109                 public boolean waitForRequestStart(long timeout) throws InterruptedException {
 110                         long nt0 = System.nanoTime();
 111                         tryAcquire(timeout, TimeUnit.MILLISECONDS);
 112                         if (TRACE_WAITED_TIME)
 113                                 System.out.println("[" + wrappedRequest + "] request start wait time " + ((double) (System.nanoTime() - nt0) / 1e6) + " ms");
 114                         if (state.compareAndSet(0, 2)) {
 115                                 if (TRACE_TIMEOUT)
 116                                         System.out.println("[" + wrappedRequest + "] starting of request timed out (>= " + timeout + " ms)");
 117                                 return false;
 118                         }
 119                         return true;
 120                 }
 121
 122         }
 123
 124         /**
 125          * @param processor
 126          *            database session handle
 127          * @param requestStartTimeout
 128          *            timeout in milliseconds for the request execution to start. If
 129          *            request execution is not started within this period of time,
 130          *            timeoutResult<code>timeoutResult</code>requestExecutionTimeout
 131          *            will be returned. Must be &ge; 0.
 132          * @param requestExecutionTimeout
 133          *            the maximum time to wait for the request execution to complete
 134          *            from the time the request execution was started. If negative,
 135          *            there is no timeout.
 136          * @param timeoutResult
 137          *            the result
 138          * @param read
 139          * @param procedure
 140          * @return
 141          * @throws DatabaseException
 142          * @throws InterruptedException
 143          */
 144         public static final <R> R trySyncRequest(
 145                         Session session,
 146                         long requestStartTimeout,
 147                         long requestExecutionTimeout,
 148                         R timeoutResult,
 149                         final Read<R> read,
 150                         final Procedure<R> procedure)
 151                                         throws DatabaseException, InterruptedException
 152         {
 153                 if (requestStartTimeout < 0)
 154                         throw new IllegalArgumentException("request start timeout must be >= 0, got " + requestStartTimeout);
 155                 if (TRACE_SCHEDULE)
 156                         System.out.println("[" + read + "] scheduling timed request with timeouts " + requestStartTimeout + "/" + requestExecutionTimeout + " ms");
 157                 ResultProcedure<R> proc = new ResultProcedure<>();
 158                 TimeoutingRequest<R> req = new TimeoutingRequest<>(read, procedure);
 159                 session.asyncRequest(req, proc);
 160                 if (!req.waitForRequestStart(requestStartTimeout))
 161                         return timeoutResult;
 162                 return proc.waitForResultOrException(requestExecutionTimeout, timeoutResult);
 163         }
 164
 165         public static final <R> R trySyncRequest(
 166                         Session session,
 167                         long requestStartTimeout,
 168                         long requestExecutionTimeout,
 169                         Read<R> read,
 170                         Procedure<R> procedure)
 171                                         throws DatabaseException, InterruptedException
 172         {
 173                 return trySyncRequest(session, requestStartTimeout, requestExecutionTimeout, null, read, procedure);
 174         }
 175
 176         public static final <R> R trySyncRequest(
 177                         Session session,
 178                         long requestStartTimeout,
 179                         long requestExecutionTimeout,
 180                         R timeoutResult,
 181                         final Read<R> read)
 182                                         throws DatabaseException, InterruptedException
 183         {
 184                 return trySyncRequest(session, requestStartTimeout, requestExecutionTimeout, timeoutResult, read, null);
 185         }
 186
 187         public static final <R> R trySyncRequest(
 188                         Session session,
 189                         long requestStartTimeout,
 190                         long requestExecutionTimeout,
 191                         Read<R> read)
 192                                         throws DatabaseException, InterruptedException
 193         {
 194                 return trySyncRequest(session, requestStartTimeout, requestExecutionTimeout, null, read, null);
 195         }
 196
 197 }