bundles/org.simantics.db.impl/src/org/simantics/db/impl/query/AsyncReadEntry.java

   1 /*******************************************************************************
   2  * Copyright (c) 2007, 2018 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package org.simantics.db.impl.query;
  13
  14 import org.simantics.db.AsyncReadGraph;
  15 import org.simantics.db.exception.DatabaseException;
  16 import org.simantics.db.impl.BlockingAsyncProcedure;
  17 import org.simantics.db.impl.DebugPolicy;
  18 import org.simantics.db.impl.graph.AsyncBarrierImpl;
  19 import org.simantics.db.impl.graph.BarrierTracing;
  20 import org.simantics.db.impl.graph.ReadGraphImpl;
  21 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
  22 import org.simantics.db.procedure.AsyncProcedure;
  23 import org.simantics.db.request.AsyncRead;
  24 import org.slf4j.Logger;
  25 import org.slf4j.LoggerFactory;
  26
  27 final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> implements AsyncProcedure<T> {
  28
  29     private static final Logger LOGGER = LoggerFactory.getLogger(AsyncReadEntry.class);
  30
  31     protected AsyncRead<T> request;
  32
  33     AsyncReadEntry(AsyncRead<T> request) {
  34         this.request = request;
  35         if (DebugPolicy.QUERY_STATE)
  36             System.out.println("[QUERY STATE]: created " + this);
  37     }
  38
  39     @Override
  40     int makeHash() {
  41         return request.hashCode();
  42     }
  43
  44     @Override
  45     public Object getOriginalRequest() {
  46         return request;
  47     }
  48
  49     @Override
  50     public void discard() {
  51         super.discard();
  52         setResult(null);
  53     }
  54
  55     public void except(AsyncReadGraph graph, Throwable t) {
  56
  57         assert (isPending());
  58
  59         synchronized (this) {
  60             except(t);
  61         }
  62
  63     }
  64
  65     @Override
  66     final public Query getQuery() {
  67
  68         return new Query() {
  69
  70             @Override
  71             public void recompute(ReadGraphImpl graph) {
  72
  73                 try {
  74
  75                     BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(graph.asyncBarrier, graph, new AsyncProcedure<T>() {
  76
  77                         @Override
  78                         public void execute(AsyncReadGraph graph, T result) {
  79                             setResult(result);
  80                             setReady();
  81                         }
  82
  83                         @Override
  84                         public void exception(AsyncReadGraph graph, Throwable t) {
  85                             except(t);
  86                         }
  87
  88                     }, request);
  89
  90                     request.perform(graph, proc);
  91
  92                     proc.get();
  93
  94                 } catch (Throwable t) {
  95                     except(t);
  96                 }
  97
  98             }
  99
 100             @Override
 101             public void removeEntry(QueryProcessor qp) {
 102                 qp.cache.remove(AsyncReadEntry.this);
 103             }
 104
 105             @Override
 106             public int type() {
 107                 return request.getFlags();
 108             }
 109
 110             @Override
 111             public String toString() {
 112                 if (request == null)
 113                     return "DISCARDED";
 114                 else if (isExcepted())
 115                     return request.toString() + " " + getResult();
 116                 else
 117                     return request.toString() + " " + statusOrException;
 118             }
 119
 120         };
 121
 122     }
 123
 124     @Override
 125     public Object performFromCache(ReadGraphImpl graph, AsyncProcedure<T> proc) {
 126
 127         if (isExcepted()) {
 128
 129             try {
 130                 proc.exception(graph, (Throwable) getResult());
 131             } catch (Throwable t) {
 132                 LOGGER.error("performFromCache proc.exception failed", t);
 133             }
 134
 135         } else {
 136
 137             try {
 138                 T result = (T) getResult();
 139                 proc.execute(graph, result);
 140             } catch (Throwable t) {
 141                 LOGGER.error("performFromCache proc.execute failed", t);
 142             }
 143
 144         }
 145
 146         return getResult();
 147
 148     }
 149
 150     public static <T> T computeForEach(ReadGraphImpl graph, AsyncRead<T> request, AsyncReadEntry<T> entry,
 151             AsyncProcedure<T> procedure_, boolean needsToBlock) throws DatabaseException {
 152
 153         AsyncProcedure<T> procedure = entry != null ? entry : procedure_;
 154
 155         ReadGraphImpl queryGraph = graph.withParent(entry);
 156         queryGraph.asyncBarrier.inc();
 157
 158         BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(queryGraph.asyncBarrier, graph, null, request);
 159
 160         class AsyncTask extends SessionTask {
 161
 162             int counter = 0;
 163             T result;
 164             DatabaseException exception;
 165
 166             public AsyncTask(ReadGraphImpl graph) {
 167                 super(graph);
 168             }
 169
 170             @Override
 171             public void run0(int thread) {
 172                 if(needsToBlock) proc.waitBarrier();
 173                 if(proc.isDone()) {
 174                     ReadGraphImpl executeGraph = graph.withParent(graph.parent);
 175                     executeGraph.asyncBarrier.inc();
 176                     try {
 177                         result = (T)proc.get();
 178                         if(procedure != null) {
 179                             procedure.execute(executeGraph, result);
 180                         }
 181                     } catch (DatabaseException e) {
 182                         if(procedure != null) procedure.exception(executeGraph, e);
 183                         exception = e;
 184                     } catch (Throwable t) {
 185                         DatabaseException dbe = new DatabaseException(t);
 186                         if(procedure != null) procedure.exception(executeGraph, dbe);
 187                         exception = dbe;
 188                     } finally {
 189                         if (entry != null) {
 190                             // This does not throw
 191                             entry.performFromCache(executeGraph, procedure_);
 192                         }
 193                         executeGraph.asyncBarrier.dec();
 194                         executeGraph.asyncBarrier.waitBarrier(procedure, executeGraph);
 195                     }
 196                 } else {
 197                     if(counter++ > 10000) {
 198                         if(BarrierTracing.BOOKKEEPING) {
 199                             AsyncBarrierImpl.printReverse(queryGraph.asyncBarrier, 2);
 200                             AsyncBarrierImpl caller = queryGraph.asyncBarrier.caller;
 201                             while(caller != null) {
 202                                 System.err.println("called by " + AsyncBarrierImpl.report(caller));
 203                                 caller = caller.caller;
 204                             }
 205                             for(AsyncBarrierImpl ab : BarrierTracing.debuggerMap.keySet()) {
 206                                 AsyncBarrierImpl.printReverse(ab, 2);
 207                             }
 208                         }
 209                         throw new IllegalStateException("Eternal loop in queries.");
 210                     }
 211                     graph.processor.schedule(new AsyncTask(graph));
 212                 }
 213             }
 214
 215         }
 216
 217         try {
 218             request.perform(queryGraph, proc);
 219         } finally {
 220             queryGraph.asyncBarrier.dec();
 221         }
 222
 223         AsyncTask task = new AsyncTask(graph);
 224
 225         if(needsToBlock) task.run(0);
 226         else if (proc.isDone()) task.run(0);
 227         else  {
 228             graph.processor.schedule(task);
 229             return null;
 230         }
 231
 232         if(task.exception != null) throw task.exception;
 233         else return task.result;
 234
 235     }
 236
 237     @Override
 238     public String toString() {
 239         if (isDiscarded())
 240             return "DISCARDED " + request.toString();
 241         else if (isExcepted())
 242             return request.toString() + " " + getResult();
 243         else
 244             return request.toString() + " " + statusOrException;
 245     }
 246
 247     @Override
 248     public void execute(AsyncReadGraph graph, T result) {
 249         setResult(result);
 250         setReady();
 251     }
 252
 253     @Override
 254     public void exception(AsyncReadGraph graph, Throwable throwable) {
 255         except(throwable);
 256     }
 257
 258 }