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.ReadGraphImpl;
  20 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
  21 import org.simantics.db.procedure.AsyncProcedure;
  22 import org.simantics.db.request.AsyncRead;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25
  26 final public class AsyncReadEntry<T> extends CacheEntryBase<AsyncProcedure<T>> implements AsyncProcedure<T> {
  27
  28     private static final Logger LOGGER = LoggerFactory.getLogger(AsyncReadEntry.class);
  29
  30     protected AsyncRead<T> request;
  31
  32     AsyncReadEntry(AsyncRead<T> request) {
  33         this.request = request;
  34         if (DebugPolicy.QUERY_STATE)
  35             System.out.println("[QUERY STATE]: created " + this);
  36     }
  37
  38     @Override
  39     int makeHash() {
  40         return request.hashCode();
  41     }
  42
  43     @Override
  44     public Object getOriginalRequest() {
  45         return request;
  46     }
  47
  48     @Override
  49     public void discard() {
  50         super.discard();
  51         setResult(null);
  52     }
  53
  54     public void except(AsyncReadGraph graph, Throwable t) {
  55
  56         assert (isPending());
  57
  58         synchronized (this) {
  59             except(t);
  60         }
  61
  62     }
  63
  64     @Override
  65     final public Query getQuery() {
  66
  67         return new Query() {
  68
  69             @Override
  70             public void recompute(ReadGraphImpl graph) {
  71
  72                 try {
  73
  74                     BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(graph.asyncBarrier, graph, new AsyncProcedure<T>() {
  75
  76                         @Override
  77                         public void execute(AsyncReadGraph graph, T result) {
  78                             setResult(result);
  79                             setReady();
  80                         }
  81
  82                         @Override
  83                         public void exception(AsyncReadGraph graph, Throwable t) {
  84                             except(t);
  85                         }
  86
  87                     }, request);
  88
  89                     request.perform(graph, proc);
  90
  91                     proc.get();
  92
  93                 } catch (Throwable t) {
  94                     except(t);
  95                 }
  96
  97             }
  98
  99             @Override
 100             public void removeEntry(QueryProcessor qp) {
 101                 qp.cache.remove(AsyncReadEntry.this);
 102             }
 103
 104             @Override
 105             public int type() {
 106                 return request.getFlags();
 107             }
 108
 109             @Override
 110             public String toString() {
 111                 if (request == null)
 112                     return "DISCARDED";
 113                 else if (isExcepted())
 114                     return request.toString() + " " + getResult();
 115                 else
 116                     return request.toString() + " " + statusOrException;
 117             }
 118
 119         };
 120
 121     }
 122
 123     @Override
 124     public Object performFromCache(ReadGraphImpl graph, AsyncProcedure<T> proc) {
 125
 126         if (isExcepted()) {
 127
 128             try {
 129                 proc.exception(graph, (Throwable) getResult());
 130             } catch (Throwable t) {
 131                 LOGGER.error("performFromCache proc.exception failed", t);
 132             }
 133
 134         } else {
 135
 136             try {
 137                 T result = (T) getResult();
 138                 proc.execute(graph, result);
 139             } catch (Throwable t) {
 140                 LOGGER.error("performFromCache proc.execute failed", t);
 141             }
 142
 143         }
 144
 145         return getResult();
 146
 147     }
 148
 149     public static <T> T computeForEach(ReadGraphImpl graph, AsyncRead<T> request, AsyncReadEntry<T> entry,
 150             AsyncProcedure<T> procedure_, boolean needsToBlock) throws DatabaseException {
 151
 152         AsyncProcedure<T> procedure = entry != null ? entry : procedure_;
 153
 154         ReadGraphImpl queryGraph = graph.withParent(entry);
 155
 156         BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(queryGraph.asyncBarrier, graph, null, request);
 157
 158         class AsyncTask extends SessionTask {
 159
 160                 int counter = 0;
 161             T result;
 162             DatabaseException exception;
 163
 164             public AsyncTask(ReadGraphImpl graph) {
 165                 super(graph);
 166             }
 167
 168             @Override
 169             public void run(int thread) {
 170                 if(needsToBlock) proc.waitBarrier();
 171                 if(proc.isDone()) {
 172                     try {
 173                         result = (T)proc.get();
 174                         if(procedure != null) procedure.execute(graph, result);
 175                     } catch (DatabaseException e) {
 176                         if(procedure != null) procedure.exception(graph, e);
 177                         exception = e;
 178                     } catch (Throwable t) {
 179                         DatabaseException dbe = new DatabaseException(t);
 180                         if(procedure != null) procedure.exception(graph, dbe);
 181                         exception = dbe;
 182                     } finally {
 183                         if (entry != null)
 184                             entry.performFromCache(graph, procedure_);
 185                     }
 186                 } else {
 187                         if(counter++ > 10000) {
 188                                 AsyncBarrierImpl.printReverse(queryGraph.asyncBarrier, 2);
 189                                 AsyncBarrierImpl caller = queryGraph.asyncBarrier.caller;
 190                                 while(caller != null) {
 191                                         System.err.println("called by " + AsyncBarrierImpl.report(caller));
 192                                         caller = caller.caller;
 193                                 }
 194                                 for(AsyncBarrierImpl ab : AsyncBarrierImpl.debuggerMap.keySet()) {
 195                                 AsyncBarrierImpl.printReverse(ab, 2);
 196                                 }
 197                                 throw new IllegalStateException("Eternal loop in queries.");
 198                         }
 199                     graph.processor.schedule(this);
 200                 }
 201             }
 202
 203         }
 204
 205         request.perform(queryGraph, proc);
 206
 207         AsyncTask task = new AsyncTask(graph);
 208
 209         if(needsToBlock) task.run(0);
 210         else if (proc.isDone()) task.run(0);
 211         else  {
 212             graph.processor.schedule(task);
 213             return null;
 214         }
 215
 216         if(task.exception != null) throw task.exception;
 217         else return task.result;
 218
 219     }
 220
 221     @Override
 222     public String toString() {
 223         if (isDiscarded())
 224             return "DISCARDED " + request.toString();
 225         else if (isExcepted())
 226             return request.toString() + " " + getResult();
 227         else
 228             return request.toString() + " " + statusOrException;
 229     }
 230
 231     @Override
 232     public void execute(AsyncReadGraph graph, T result) {
 233         setResult(result);
 234         setReady();
 235     }
 236
 237     @Override
 238     public void exception(AsyncReadGraph graph, Throwable throwable) {
 239         except(throwable);
 240     }
 241
 242 }