import org.simantics.db.VirtualGraph;
import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter;
import org.simantics.db.common.utils.Logger;
-import org.simantics.db.debug.ListenerReport;
import org.simantics.db.exception.DatabaseException;
import org.simantics.db.exception.ManyObjectsForFunctionalRelationException;
import org.simantics.db.exception.NoInverseException;
import org.simantics.db.impl.graph.BarrierTracing;
import org.simantics.db.impl.graph.ReadGraphImpl;
import org.simantics.db.impl.graph.ReadGraphSupport;
-import org.simantics.db.impl.graph.WriteGraphImpl;
import org.simantics.db.impl.procedure.IntProcedureAdapter;
import org.simantics.db.impl.procedure.InternalProcedure;
import org.simantics.db.impl.procedure.TripleIntProcedureAdapter;
AtomicInteger sleepers = new AtomicInteger(0);
- private boolean updating = false;
+ boolean updating = false;
- private boolean firingListeners = false;
-
final public QueryCache cache;
final public QuerySupport querySupport;
final public Session session;
final public ResourceSupport resourceSupport;
-
- private THashSet<ListenerEntry> scheduledListeners = new THashSet<ListenerEntry>();
+
+ final public QueryListening listening = new QueryListening(this);
QueryThread[] executors;
}
final public void releaseWrite(ReadGraphImpl graph) {
- performDirtyUpdates(graph);
+ propagateChangesInQueryCache(graph);
modificationCounter++;
}
boolean isBound(ExternalReadEntry<?> entry) {
if(entry.hasParents()) return true;
- else if(hasListener(entry)) return true;
+ else if(listening.hasListener(entry)) return true;
else return false;
}
- synchronized public ListenerEntry registerDependencies(ReadGraphImpl graph, CacheEntry child, CacheEntry parent, ListenerBase listener, Object procedure, boolean inferred) {
-
- if (parent != null && !inferred) {
- try {
- if(!child.isImmutable(graph))
- child.addParent(parent);
- } catch (DatabaseException e) {
- Logger.defaultLogError(e);
- }
- if (Development.DEVELOPMENT) {
- if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_DEPENDENCIES, Bindings.BOOLEAN)) {
- System.out.println(child + " -> " + parent);
- }
- }
- }
-
- if (listener != null) {
- return registerListener(child, listener, procedure);
- } else {
- return null;
- }
-
- }
-
-
static class Dummy implements InternalProcedure<Object>, IntProcedure {
@Override
}
- public ListenerEntry registerListener(final CacheEntry entry, final ListenerBase base, final Object procedure) {
-
- assert (entry != null);
-
- if (base.isDisposed())
- return null;
-
- return addListener(entry, base, procedure);
-
- }
-
- private void primeListenerEntry(final ListenerEntry entry, final Object result) {
- entry.setLastKnown(result);
- }
-
- private ListenerEntry addListener(CacheEntry entry, ListenerBase base, Object procedure) {
-
- assert (entry != null);
- assert (procedure != null);
-
- ArrayList<ListenerEntry> list = cache.listeners.get(entry);
- if (list == null) {
- list = new ArrayList<ListenerEntry>(1);
- cache.listeners.put(entry, list);
- }
-
- ListenerEntry result = new ListenerEntry(entry, base, procedure);
- int currentIndex = list.indexOf(result);
- // There was already a listener
- if(currentIndex > -1) {
- ListenerEntry current = list.get(currentIndex);
- if(!current.base.isDisposed()) return null;
- list.set(currentIndex, result);
- } else {
- list.add(result);
- }
-
- if (Development.DEVELOPMENT) {
- if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
- new Exception().printStackTrace();
- System.err.println("addListener -> " + list.size() + " " + entry + " " + base + " " + procedure);
- }
- }
-
- return result;
-
- }
-
- private void scheduleListener(ListenerEntry entry) {
- assert (entry != null);
- if (Development.DEVELOPMENT) {
- if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
- System.err.println("Scheduled " + entry.procedure);
- }
- }
- scheduledListeners.add(entry);
- }
-
- private void removeListener(ListenerEntry entry) {
- assert (entry != null);
- ArrayList<ListenerEntry> list = cache.listeners.get(entry.entry);
- if(list == null) return;
- boolean success = list.remove(entry);
- assert (success);
- if (list.isEmpty())
- cache.listeners.remove(entry.entry);
- }
-
- private boolean hasListener(CacheEntry entry) {
- if(cache.listeners.get(entry) != null) return true;
- return false;
- }
-
- boolean hasListenerAfterDisposing(CacheEntry entry) {
- if(cache.listeners.get(entry) != null) {
- ArrayList<ListenerEntry> entries = cache.listeners.get(entry);
- ArrayList<ListenerEntry> list = null;
- for (ListenerEntry e : entries) {
- if (e.base.isDisposed()) {
- if(list == null) list = new ArrayList<ListenerEntry>();
- list.add(e);
- }
- }
- if(list != null) {
- for (ListenerEntry e : list) {
- entries.remove(e);
- }
- }
- if (entries.isEmpty()) {
- cache.listeners.remove(entry);
- return false;
- }
- return true;
- }
- return false;
- }
-
- List<ListenerEntry> getListenerEntries(CacheEntry entry) {
- hasListenerAfterDisposing(entry);
- if(cache.listeners.get(entry) != null)
- return cache.listeners.get(entry);
- else
- return Collections.emptyList();
- }
-
- void processListenerReport(CacheEntry<?> entry, Map<CacheEntry, Set<ListenerBase>> workarea) {
-
- if(!workarea.containsKey(entry)) {
-
- HashSet<ListenerBase> ls = new HashSet<ListenerBase>();
- for(ListenerEntry e : getListenerEntries(entry))
- ls.add(e.base);
-
- workarea.put(entry, ls);
-
- for(CacheEntry parent : entry.getParents(this)) {
- processListenerReport(parent, workarea);
- ls.addAll(workarea.get(parent));
- }
-
- }
-
- }
-
- public synchronized ListenerReport getListenerReport() throws IOException {
-
- class ListenerReportImpl implements ListenerReport {
-
- Map<CacheEntry, Set<ListenerBase>> workarea = new HashMap<CacheEntry, Set<ListenerBase>>();
-
- @Override
- public void print(PrintStream b) {
- Map<ListenerBase, Integer> hist = new HashMap<ListenerBase, Integer>();
- for(Map.Entry<CacheEntry, Set<ListenerBase>> e : workarea.entrySet()) {
- for(ListenerBase l : e.getValue()) {
- Integer i = hist.get(l);
- hist.put(l, i != null ? i-1 : -1);
- }
- }
-
- for(Pair<ListenerBase, Integer> p : CollectionUtils.valueSortedEntries(hist)) {
- b.print("" + -p.second + " " + p.first + "\n");
- }
-
- b.flush();
- }
-
- }
-
- ListenerReportImpl result = new ListenerReportImpl();
-
- Collection<CacheEntryBase> all = allCaches(new CacheCollectionResult()).toCollection();
- for(CacheEntryBase entry : all) {
- hasListenerAfterDisposing(entry);
- }
- for(CacheEntryBase entry : all) {
- processListenerReport(entry, result.workarea);
- }
-
- return result;
-
- }
-
- public synchronized String reportListeners(File file) throws IOException {
-
- if (!isAlive())
- return "Disposed!";
-
- PrintStream b = new PrintStream(new BufferedOutputStream(new FileOutputStream(file)));
- ListenerReport report = getListenerReport();
- report.print(b);
-
- return "Done reporting listeners.";
-
- }
void processParentReport(CacheEntry entry, Map<CacheEntry, Set<CacheEntry>> workarea) {
int listeners = 0;
for(CacheEntry entry : workarea.keySet()) {
- boolean listener = hasListenerAfterDisposing(entry);
+ boolean listener = listening.hasListenerAfterDisposing(entry);
boolean hasParents = entry.getParents(this).iterator().hasNext();
if(listener) {
// Bound
b.print("// Entry parent listing\n");
for(CacheEntry entry : workarea.keySet()) {
int status = flagMap.get(entry);
- boolean hasListener = hasListenerAfterDisposing(entry);
+ boolean hasListener = listening.hasListenerAfterDisposing(entry);
b.print("Q " + entry.toString());
if(hasListener) {
b.print(" (L" + status + ")");
}
- class UpdateEntry {
-
- public CacheEntry caller;
-
- public CacheEntry entry;
-
- public int indent;
-
- public UpdateEntry(CacheEntry caller, CacheEntry entry, int indent) {
- this.caller = caller;
- this.entry = entry;
- this.indent = indent;
- }
-
- };
-
boolean removeQuery(CacheEntry entry) {
// This entry has been removed before. No need to do anything here.
Query query = entry.getQuery();
int type = query.type();
- boolean hasListener = hasListener(entry);
+ boolean hasListener = listening.hasListener(entry);
if (Development.DEVELOPMENT) {
if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_UPDATE, Bindings.BOOLEAN)) {
- if(hasListener(entry)) {
+ if(listening.hasListener(entry)) {
System.err.println(" (L)");
} else {
System.err.println("");
// System.err.println(" => FOO " + type);
if (hasListener) {
- ArrayList<ListenerEntry> entries = cache.listeners.get(entry);
+ ArrayList<ListenerEntry> entries = listening.listeners.get(entry);
if(entries != null) {
for (ListenerEntry le : entries) {
- scheduleListener(le);
+ listening.scheduleListener(le);
}
}
}
// If invalid, update parents
if (type == RequestFlags.INVALIDATE) {
- updateParents(e.indent, entry, todo);
+ listening.updateParents(e.indent, entry, todo);
}
return hasListener;
}
- private void updateParents(int indent, CacheEntry entry, LinkedList<UpdateEntry> todo) {
-
- Iterable<CacheEntry> oldParents = entry.getParents(this);
- for (CacheEntry parent : oldParents) {
-// System.err.println("updateParents " + entry + " => " + parent);
- if(!parent.isDiscarded())
- todo.push(new UpdateEntry(entry, parent, indent + 2));
- }
-
- }
-
- private boolean pruneListener(ListenerEntry entry) {
- if (entry.base.isDisposed()) {
- removeListener(entry);
- return true;
- } else {
- return false;
- }
- }
-
/**
* @param av1 an array (guaranteed)
* @param av2 any object
- final private Object compareTo(ReadGraphImpl graph, final CacheEntry entry, final Object oldValue) {
+ final Object compareTo(ReadGraphImpl graph, final CacheEntry entry, final Object oldValue) {
try {
}
- public boolean hasScheduledUpdates() {
- return !scheduledListeners.isEmpty();
- }
-
- public void performScheduledUpdates(WriteGraphImpl graph) {
-
- assert (!updating);
- assert (!cache.collecting);
- assert (!firingListeners);
-
- firingListeners = true;
-
- try {
-
- // Performing may cause further events to be scheduled.
- while (!scheduledListeners.isEmpty()) {
-
-// graph.restart();
-// graph.state.barrier.inc();
-
- // Clone current events to make new entries possible during
- // firing.
- THashSet<ListenerEntry> entries = scheduledListeners;
- scheduledListeners = new THashSet<ListenerEntry>();
-
- ArrayList<ListenerEntry> schedule = new ArrayList<ListenerEntry>();
-
- for (ListenerEntry listenerEntry : entries) {
-
- if (pruneListener(listenerEntry)) {
- if (Development.DEVELOPMENT) {
- if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
- new Exception().printStackTrace();
- System.err.println("Pruned " + listenerEntry.procedure);
- }
- }
- continue;
- }
-
- final CacheEntry entry = listenerEntry.entry;
- assert (entry != null);
-
- Object newValue = compareTo(graph, entry, listenerEntry.getLastKnown());
-
- if (newValue != ListenerEntry.NOT_CHANGED) {
- if (Development.DEVELOPMENT) {
- if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
- new Exception().printStackTrace();
- System.err.println("Add to schedule " + listenerEntry.procedure + " with " + newValue);
- }
- }
- schedule.add(listenerEntry);
- listenerEntry.setLastKnown(entry.getResult());
- }
-
- }
-
- for(ListenerEntry listenerEntry : schedule) {
- final CacheEntry entry = listenerEntry.entry;
- if (Development.DEVELOPMENT) {
- if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
- System.err.println("Firing " + listenerEntry.procedure);
- }
- }
- try {
- if (Development.DEVELOPMENT) {
- if(Development.<Boolean>getProperty(DevelopmentKeys.QUERYPROCESSOR_LISTENERS, Bindings.BOOLEAN)) {
- System.err.println("Firing " + listenerEntry.procedure + " for " + listenerEntry.entry);
- }
- }
- entry.performFromCache(graph, listenerEntry.procedure);
- } catch (Throwable t) {
- t.printStackTrace();
- }
- }
-
-// graph.state.barrier.dec();
-// graph.waitAsync(null);
-// graph.state.barrier.assertReady();
-
- }
-
- } finally {
- firingListeners = false;
- }
-
- }
/**
*
Object newValue = compareTo(graph, immediate, ListenerEntry.NO_VALUE);
if (newValue != ListenerEntry.NOT_CHANGED)
- updateParents(0, immediate, todo);
+ listening.updateParents(0, immediate, todo);
} else {
Object newValue = compareTo(graph, immediate, oldValue);
if (newValue != ListenerEntry.NOT_CHANGED) {
- updateParents(0, immediate, todo);
+ listening.updateParents(0, immediate, todo);
} else {
// If not changed, keep the old value
immediate.setResult(oldValue);
}
- public void performDirtyUpdates(final ReadGraphImpl graph) {
+ public void propagateChangesInQueryCache(final ReadGraphImpl graph) {
+
+ // Make sure that listening has performed its work
+ listening.sync();
cache.dirty = false;
lastInvalidate = 0;
Code Review / simantics / platform.git / blobdiff