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