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.exception.RuntimeDatabaseException;
  17 import org.simantics.db.impl.BlockingAsyncProcedure;
  18 import org.simantics.db.impl.DebugPolicy;
  19 import org.simantics.db.impl.graph.AsyncBarrierImpl;
  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, 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
 157         BlockingAsyncProcedure<T> proc = new BlockingAsyncProcedure<>(queryGraph, null, request);
 158
 159         class AsyncTask extends SessionTask {
 160
 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(queryGraph, procedure_);
 185                     }
 186                 } else {
 187                     graph.processor.schedule(this);
 188                 }
 189             }
 190
 191         }
 192
 193         request.perform(queryGraph, proc);
 194
 195         AsyncTask task = new AsyncTask(graph);
 196
 197         if(needsToBlock) task.run(0);
 198         else if (proc.isDone()) task.run(0);
 199         else  {
 200             graph.processor.schedule(task);
 201             return null;
 202         }
 203
 204         if(task.exception != null) throw task.exception;
 205         else return task.result;
 206
 207     }
 208
 209     @Override
 210     public String toString() {
 211         if (isDiscarded())
 212             return "DISCARDED " + request.toString();
 213         else if (isExcepted())
 214             return request.toString() + " " + getResult();
 215         else
 216             return request.toString() + " " + statusOrException;
 217     }
 218
 219     @Override
 220     public void execute(AsyncReadGraph graph, T result) {
 221         setResult(result);
 222         setReady();
 223     }
 224
 225     @Override
 226     public void exception(AsyncReadGraph graph, Throwable throwable) {
 227         except(throwable);
 228     }
 229
 230 }