General event listening interface for DB and purge events.
[simantics/platform.git] / bundles / org.simantics.db.procore / src / fi / vtt / simantics / procore / internal / SessionImplSocket.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 fi.vtt.simantics.procore.internal;
13
14 import java.io.File;
15 import java.io.IOException;
16 import java.io.InputStream;
17 import java.nio.charset.Charset;
18 import java.nio.file.FileVisitResult;
19 import java.nio.file.Files;
20 import java.nio.file.Path;
21 import java.nio.file.SimpleFileVisitor;
22 import java.nio.file.attribute.BasicFileAttributes;
23 import java.util.ArrayList;
24 import java.util.Collection;
25 import java.util.HashSet;
26 import java.util.TreeMap;
27 import java.util.concurrent.CopyOnWriteArrayList;
28 import java.util.concurrent.Semaphore;
29 import java.util.concurrent.atomic.AtomicInteger;
30 import java.util.function.Consumer;
31
32 import org.eclipse.core.runtime.Platform;
33 import org.simantics.databoard.Bindings;
34 import org.simantics.db.AsyncReadGraph;
35 import org.simantics.db.ChangeSet;
36 import org.simantics.db.DevelopmentKeys;
37 import org.simantics.db.Disposable;
38 import org.simantics.db.ExternalValueSupport;
39 import org.simantics.db.Metadata;
40 import org.simantics.db.MonitorContext;
41 import org.simantics.db.MonitorHandler;
42 import org.simantics.db.ReadGraph;
43 import org.simantics.db.Resource;
44 import org.simantics.db.ResourceSerializer;
45 import org.simantics.db.Session;
46 import org.simantics.db.SessionManager;
47 import org.simantics.db.SessionVariables;
48 import org.simantics.db.VirtualGraph;
49 import org.simantics.db.WriteGraph;
50 import org.simantics.db.authentication.UserAuthenticationAgent;
51 import org.simantics.db.authentication.UserAuthenticator;
52 import org.simantics.db.common.Indexing;
53 import org.simantics.db.common.TransactionPolicyRelease;
54 import org.simantics.db.common.procedure.adapter.AsyncMultiProcedureAdapter;
55 import org.simantics.db.common.procedure.adapter.ProcedureAdapter;
56 import org.simantics.db.common.procedure.adapter.SyncMultiProcedureAdapter;
57 import org.simantics.db.common.procedure.wrapper.NoneToAsyncListener;
58 import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiListener;
59 import org.simantics.db.common.procedure.wrapper.NoneToAsyncMultiProcedure;
60 import org.simantics.db.common.procedure.wrapper.NoneToAsyncProcedure;
61 import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiListener;
62 import org.simantics.db.common.procedure.wrapper.NoneToSyncMultiProcedure;
63 import org.simantics.db.common.procedure.wrapper.SyncToAsyncListener;
64 import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiListener;
65 import org.simantics.db.common.procedure.wrapper.SyncToAsyncMultiProcedure;
66 import org.simantics.db.common.procedure.wrapper.SyncToAsyncProcedure;
67 import org.simantics.db.common.service.ServiceActivityMonitorImpl;
68 import org.simantics.db.common.service.ServiceActivityUpdaterForWriteTransactions;
69 import org.simantics.db.common.utils.Logger;
70 import org.simantics.db.event.ChangeEvent;
71 import org.simantics.db.event.ChangeListener;
72 import org.simantics.db.event.SessionEventListener;
73 import org.simantics.db.exception.CancelTransactionException;
74 import org.simantics.db.exception.ClusterSetExistException;
75 import org.simantics.db.exception.DatabaseException;
76 import org.simantics.db.exception.ImmutableException;
77 import org.simantics.db.exception.InvalidResourceReferenceException;
78 import org.simantics.db.exception.ResourceNotFoundException;
79 import org.simantics.db.exception.RuntimeDatabaseException;
80 import org.simantics.db.exception.ServiceException;
81 import org.simantics.db.exception.ServiceNotFoundException;
82 import org.simantics.db.impl.BlockingAsyncProcedure;
83 import org.simantics.db.impl.ClusterBase;
84 import org.simantics.db.impl.ClusterI;
85 import org.simantics.db.impl.ClusterTraitsBase;
86 import org.simantics.db.impl.ClusterTranslator;
87 import org.simantics.db.impl.ResourceImpl;
88 import org.simantics.db.impl.TransientGraph;
89 import org.simantics.db.impl.VirtualGraphImpl;
90 import org.simantics.db.impl.graph.DelayedWriteGraph;
91 import org.simantics.db.impl.graph.ReadGraphImpl;
92 import org.simantics.db.impl.graph.WriteGraphImpl;
93 import org.simantics.db.impl.graph.WriteSupport;
94 import org.simantics.db.impl.internal.RandomAccessValueSupport;
95 import org.simantics.db.impl.procedure.ResultCallWrappedQueryProcedure4;
96 import org.simantics.db.impl.procedure.ResultCallWrappedSyncQueryProcedure;
97 import org.simantics.db.impl.query.QueryCache;
98 import org.simantics.db.impl.query.QueryCacheBase;
99 import org.simantics.db.impl.query.QueryProcessor;
100 import org.simantics.db.impl.query.QueryProcessor.SessionRead;
101 import org.simantics.db.impl.query.QueryProcessor.SessionTask;
102 import org.simantics.db.impl.service.QueryDebug;
103 import org.simantics.db.impl.support.VirtualGraphServerSupport;
104 import org.simantics.db.impl.support.WriteRequestScheduleSupport;
105 import org.simantics.db.procedure.AsyncListener;
106 import org.simantics.db.procedure.AsyncMultiListener;
107 import org.simantics.db.procedure.AsyncMultiProcedure;
108 import org.simantics.db.procedure.AsyncProcedure;
109 import org.simantics.db.procedure.Listener;
110 import org.simantics.db.procedure.ListenerBase;
111 import org.simantics.db.procedure.MultiListener;
112 import org.simantics.db.procedure.MultiProcedure;
113 import org.simantics.db.procedure.Procedure;
114 import org.simantics.db.procedure.SyncListener;
115 import org.simantics.db.procedure.SyncMultiListener;
116 import org.simantics.db.procedure.SyncMultiProcedure;
117 import org.simantics.db.procedure.SyncProcedure;
118 import org.simantics.db.procore.cluster.ClusterImpl;
119 import org.simantics.db.procore.cluster.ClusterTraits;
120 import org.simantics.db.procore.protocol.Constants;
121 import org.simantics.db.procore.protocol.DebugPolicy;
122 import org.simantics.db.request.AsyncMultiRead;
123 import org.simantics.db.request.AsyncRead;
124 import org.simantics.db.request.DelayedWrite;
125 import org.simantics.db.request.DelayedWriteResult;
126 import org.simantics.db.request.ExternalRead;
127 import org.simantics.db.request.MultiRead;
128 import org.simantics.db.request.Read;
129 import org.simantics.db.request.ReadInterface;
130 import org.simantics.db.request.Write;
131 import org.simantics.db.request.WriteInterface;
132 import org.simantics.db.request.WriteOnly;
133 import org.simantics.db.request.WriteOnlyResult;
134 import org.simantics.db.request.WriteResult;
135 import org.simantics.db.request.WriteTraits;
136 import org.simantics.db.service.ByteReader;
137 import org.simantics.db.service.ClusterBuilder;
138 import org.simantics.db.service.ClusterBuilderFactory;
139 import org.simantics.db.service.ClusterControl;
140 import org.simantics.db.service.ClusterSetsSupport;
141 import org.simantics.db.service.ClusterUID;
142 import org.simantics.db.service.ClusteringSupport;
143 import org.simantics.db.service.CollectionSupport;
144 import org.simantics.db.service.DebugSupport;
145 import org.simantics.db.service.DirectQuerySupport;
146 import org.simantics.db.service.EventSupport;
147 import org.simantics.db.service.GraphChangeListenerSupport;
148 import org.simantics.db.service.InitSupport;
149 import org.simantics.db.service.LifecycleSupport;
150 import org.simantics.db.service.ManagementSupport;
151 import org.simantics.db.service.QueryControl;
152 import org.simantics.db.service.SerialisationSupport;
153 import org.simantics.db.service.ServerInformation;
154 import org.simantics.db.service.ServiceActivityMonitor;
155 import org.simantics.db.service.SessionEventSupport;
156 import org.simantics.db.service.SessionMonitorSupport;
157 import org.simantics.db.service.SessionUserSupport;
158 import org.simantics.db.service.StatementSupport;
159 import org.simantics.db.service.TransactionPolicySupport;
160 import org.simantics.db.service.TransactionSupport;
161 import org.simantics.db.service.TransferableGraphSupport;
162 import org.simantics.db.service.UndoRedoSupport;
163 import org.simantics.db.service.VirtualGraphSupport;
164 import org.simantics.db.service.XSupport;
165 import org.simantics.layer0.Layer0;
166 import org.simantics.utils.DataContainer;
167 import org.simantics.utils.Development;
168 import org.simantics.utils.threads.logger.ITask;
169 import org.simantics.utils.threads.logger.ThreadLogger;
170 import org.slf4j.LoggerFactory;
171
172 import gnu.trove.procedure.TLongProcedure;
173 import gnu.trove.set.hash.TLongHashSet;
174
175
176 public abstract class SessionImplSocket implements Session, WriteRequestScheduleSupport {
177
178     private static org.slf4j.Logger LOGGER = LoggerFactory.getLogger(SessionImplSocket.class);
179
180     protected static final boolean DEBUG        = false;
181
182     private static final boolean DIAGNOSTICS       = false;
183
184     private TransactionPolicySupport transactionPolicy;
185
186     final private ClusterControl clusterControl;
187
188     final protected BuiltinSupportImpl builtinSupport;
189     final protected VirtualGraphServerSupportImpl virtualGraphServerSupport;
190     final protected ClusterSetsSupport clusterSetsSupport;
191     final protected LifecycleSupportImpl lifecycleSupport;
192
193     protected QuerySupportImpl querySupport;
194     protected ResourceSupportImpl resourceSupport;
195     protected WriteSupport writeSupport;
196     public ClusterTranslator clusterTranslator;
197
198     boolean dirtyPrimitives = false;
199
200     public static final int SERVICE_MODE_CREATE = 2;
201     public static final int SERVICE_MODE_ALLOW = 1;
202     public int serviceMode = 0;
203     public boolean createdImmutableClusters = false;
204     public TLongHashSet createdClusters = new TLongHashSet();
205
206     private Layer0 L0;
207
208     /**
209      * The service locator maintained by the workbench. These services are
210      * initialized during workbench during the <code>init</code> method.
211      */
212     final protected ServiceLocatorImpl                           serviceLocator     = new ServiceLocatorImpl();
213
214     final public ResourceSerializerImpl                       resourceSerializer = new ResourceSerializerImpl();
215
216     final CopyOnWriteArrayList<ChangeListener>         changeListeners2   = new CopyOnWriteArrayList<ChangeListener>();
217
218     final CopyOnWriteArrayList<ChangeListener>         metadataListeners   = new CopyOnWriteArrayList<ChangeListener>();
219
220     final CopyOnWriteArrayList<SessionEventListener>   eventListeners     = new CopyOnWriteArrayList<SessionEventListener>();
221
222     final HashSet<Thread>                              sessionThreads     = new HashSet<Thread>();
223
224     final BijectionMap<MonitorContext, MonitorHandler> monitorContexts    = new BijectionMap<MonitorContext, MonitorHandler>();
225
226     final protected State                                        state              = new State();
227
228     protected GraphSession                                       graphSession       = null;
229
230     protected SessionManager                                     sessionManagerImpl = null;
231
232     protected UserAuthenticationAgent                            authAgent          = null;
233
234     protected UserAuthenticator                                  authenticator      = null;
235
236     protected Resource                                           user               = null;
237
238     protected ClusterStream                                      clusterStream      = null;
239
240     protected SessionRequestManager                         requestManager = null;
241
242     public ClusterTable                                       clusterTable       = null;
243
244     public QueryProcessor                                     queryProvider2 = null;
245
246     ClientChangesImpl                                  clientChanges      = null;
247
248     MonitorHandler[]                                   monitorHandlers    = new MonitorHandler[0];
249
250 //    protected long                                               newClusterId       = Constants.NullClusterId;
251
252     protected int                                                flushCounter       = 0;
253
254     protected boolean                                            writeOnly          = false;
255
256     WriteState<?>                                                writeState = null;
257     WriteStateBase<?>                                            delayedWriteState = null;
258     protected Resource defaultClusterSet = null; // If not null then used for newResource().
259
260     public SessionImplSocket(SessionManager sessionManagerImpl, UserAuthenticationAgent authAgent) {
261
262 //        if (authAgent == null)
263 //            throw new IllegalArgumentException("null authentication agent");
264
265         File t = StaticSessionProperties.virtualGraphStoragePath;
266         if (null == t)
267             t = new File(".");
268         this.clusterTable = new ClusterTable(this, t);
269         this.builtinSupport = new BuiltinSupportImpl(this);
270         this.sessionManagerImpl = sessionManagerImpl;
271         this.user = null;
272 //        this.authAgent = authAgent;
273
274         serviceLocator.registerService(Session.class, this);
275         serviceLocator.registerService(InitSupport.class, new InitSupportImpl(this));
276         serviceLocator.registerService(ManagementSupport.class, new ManagementSupportImpl(this));
277         serviceLocator.registerService(QueryControl.class, new QueryControlImpl(this));
278         serviceLocator.registerService(SessionUserSupport.class, new SessionUserSupportImpl(this));
279         serviceLocator.registerService(GraphChangeListenerSupport.class, new GraphChangeListenerSupportImpl(this));
280         serviceLocator.registerService(SessionEventSupport.class, new SessionEventSupportImpl(this));
281         serviceLocator.registerService(SerialisationSupport.class, new SerialisationSupportImpl(this));
282         serviceLocator.registerService(UndoRedoSupport.class, new UndoRedoSupportImpl(this));
283         serviceLocator.registerService(ClusteringSupport.class, new ClusteringSupportImpl(this));
284         serviceLocator.registerService(TransactionSupport.class, new TransactionSupportImpl(this));
285         serviceLocator.registerService(SessionMonitorSupport.class, new SessionMonitorSupportImpl(this));
286         serviceLocator.registerService(TransferableGraphSupport.class, new TransferableGraphSupportImpl(this));
287         serviceLocator.registerService(QueryDebug.class, new QueryDebugImpl(this));
288         serviceLocator.registerService(CollectionSupport.class, new CollectionSupportImpl(this));
289         serviceLocator.registerService(StatementSupport.class, new StatementSupportImpl(this));
290         serviceLocator.registerService(DirectQuerySupport.class, new DirectQuerySupportImpl(this));
291         serviceLocator.registerService(XSupport.class, new XSupportImpl(this));
292         serviceLocator.registerService(DebugSupport.class, new DebugSupportImpl());
293         serviceLocator.registerService(ExternalValueSupport.class, new ExternalValueSupportImpl(this));
294         serviceLocator.registerService(RandomAccessValueSupport.class, new RandomAccessValueSupportImpl());
295         serviceLocator.registerService(ServiceActivityMonitor.class, new ServiceActivityMonitorImpl());
296         serviceLocator.registerService(EventSupport.class, new EventSupportImpl());
297         ServiceActivityUpdaterForWriteTransactions.register(this);
298
299         this.virtualGraphServerSupport = new VirtualGraphServerSupportImpl(this, t);
300         serviceLocator.registerService(VirtualGraphSupport.class, virtualGraphServerSupport);
301         serviceLocator.registerService(VirtualGraphServerSupport.class, virtualGraphServerSupport);
302         this.lifecycleSupport = new LifecycleSupportImpl(this);
303         serviceLocator.registerService(LifecycleSupport.class, lifecycleSupport);
304         this.transactionPolicy = new TransactionPolicyRelease();
305         serviceLocator.registerService(TransactionPolicySupport.class, transactionPolicy);
306         this.clusterControl = new ClusterControlImpl(this);
307         serviceLocator.registerService(ClusterControl.class, clusterControl);
308         this.clusterSetsSupport = new ClusterSetsSupportImpl2(); // Using same path as virtual graphs.
309         this.clusterSetsSupport.setReadDirectory(t.toPath());
310         this.clusterSetsSupport.updateWriteDirectory(t.toPath());
311         serviceLocator.registerService(ClusterSetsSupport.class, clusterSetsSupport);
312     }
313
314     /*
315      *
316      * Session interface
317      */
318
319
320     @Override
321     public Resource getRootLibrary() {
322         return queryProvider2.getRootLibraryResource();
323     }
324
325     void refresh(int thread, ClusterUID[] clusterUID, long csid) throws DatabaseException {
326         if (!graphSession.dbSession.refreshEnabled())
327             return;
328         try {
329             getClusterTable().refresh(csid, this, clusterUID);
330         } catch (Throwable t) {
331             Logger.defaultLogError("Refesh failed.", t);
332         }
333     }
334
335     static final class TaskHelper {
336         private final String name;
337         private Object result;
338         final Semaphore sema = new Semaphore(0);
339         private Throwable throwable = null;
340         final Consumer<DatabaseException> callback = e -> {
341             synchronized (TaskHelper.this) {
342                 throwable = e;
343             }
344         };
345         final Procedure<Object> proc = new Procedure<Object>() {
346             @Override
347             public void execute(Object result) {
348                 callback.accept(null);
349             }
350             @Override
351             public void exception(Throwable t) {
352                 if (t instanceof DatabaseException)
353                     callback.accept((DatabaseException)t);
354                 else
355                     callback.accept(new DatabaseException("" + name + "operation failed.", t));
356             }
357         };
358         final WriteTraits writeTraits = new WriteTraits() {};
359         TaskHelper(String name) {
360             this.name = name;
361         }
362         @SuppressWarnings("unchecked")
363         <T> T getResult() {
364             return (T)result;
365         }
366         void setResult(Object result) {
367             this.result = result;
368         }
369 //        Throwable throwableGet() {
370 //            return throwable;
371 //        }
372         synchronized void throwableSet(Throwable t) {
373             throwable = t;
374         }
375 //        void throwableSet(String t) {
376 //            throwable = new InternalException("" + name + " operation failed. " + t);
377 //        }
378         synchronized void throwableCheck()
379         throws DatabaseException {
380             if (null != throwable)
381                 if (throwable instanceof DatabaseException)
382                     throw (DatabaseException)throwable;
383                 else
384                     throw new DatabaseException("Undo operation failed.", throwable);
385         }
386         void throw_(String message)
387         throws DatabaseException {
388             throw new DatabaseException("" + name + " operation failed. " + message);
389         }
390     }
391
392
393
394     private ListenerBase getListenerBase(Object procedure) {
395         if (procedure instanceof ListenerBase)
396             return (ListenerBase) procedure;
397         else
398             return null;
399     }
400
401     public <T> void scheduleRequest(final Write request, final Consumer<DatabaseException> callback, final Semaphore notify) {
402         scheduleRequest(request, callback, notify, null);
403     }
404
405     /* (non-Javadoc)
406      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.Write, org.simantics.utils.datastructures.Callback, java.util.concurrent.Semaphore, java.lang.Boolean)
407      */
408     @Override
409     public <T> void scheduleRequest(final Write request, final Consumer<DatabaseException> callback, final Semaphore notify, Boolean combine) {
410
411         assert (request != null);
412
413         if(Development.DEVELOPMENT) {
414             try {
415             if(Development.<Boolean>getProperty(DevelopmentKeys.SESSION_LOG_WRITES, Bindings.BOOLEAN))
416                 System.err.println("schedule write '" + request + "'");
417             } catch (Throwable t) {
418                 Logger.defaultLogError(t);
419             }
420         }
421
422         requestManager.scheduleWrite(new SessionTask(null) {
423
424             @Override
425             public void run0(int thread) {
426
427                 if(Development.DEVELOPMENT) {
428                     try {
429                     if(Development.<Boolean>getProperty(DevelopmentKeys.SESSION_LOG_WRITES, Bindings.BOOLEAN))
430                         System.err.println("perform write '" + request + "'");
431                     } catch (Throwable t) {
432                         Logger.defaultLogError(t);
433                     }
434                 }
435
436                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
437
438                 fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
439
440                 try {
441
442                     flushCounter = 0;
443                     Disposable.safeDispose(clientChanges);
444                     clientChanges = new ClientChangesImpl(SessionImplSocket.this);
445
446                     VirtualGraph vg = getProvider(request.getProvider());
447
448                     WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
449                     writeState = new WriteState<Object>(writer, request, notify, new Procedure<Object>() {
450
451                         @Override
452                         public void execute(Object result) {
453                             if(callback != null) callback.accept(null);
454                         }
455
456                         @Override
457                         public void exception(Throwable t) {
458                             if(callback != null) callback.accept((DatabaseException)t);
459                         }
460
461                     });
462
463                     assert (null != writer);
464 //                    writer.state.barrier.inc();
465
466                     try {
467                         request.perform(writer);
468                         assert (null != writer);
469                     } catch (Throwable t) {
470                         if (!(t instanceof CancelTransactionException))
471                             Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", t);
472                         writeState.except(t);
473                     } finally {
474 //                        writer.state.barrier.dec();
475 //                        writer.waitAsync(request);
476                     }
477
478
479                     assert(!queryProvider2.cache.dirty);
480
481                 } catch (Throwable e) {
482
483                     // Log it first, just to be safe that the error is always logged.
484                     if (!(e instanceof CancelTransactionException))
485                     Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", e);
486
487 //                    writeState.getGraph().state.barrier.dec();
488 //                    writeState.getGraph().waitAsync(request);
489
490                     writeState.except(e);
491
492 //                    try {
493 //                        // Callback is client code, we have to be prepared for it to throw unexpected exceptions.
494 //                        // All we can do here is to log those, can't really pass them anywhere.
495 //                        if (callback != null) {
496 //                            if(e instanceof DatabaseException) callback.run((DatabaseException)e);
497 //                            else callback.run(new DatabaseException(e));
498 //                        }
499 //                    } catch (Throwable e2) {
500 //                        Logger.defaultLogError("Write request callback caused an unexpected error, see exception.", e2);
501 //                    }
502 //                    boolean empty = clusterStream.reallyFlush();
503 //                    int callerThread = -1;
504 //                    ClientChangesImpl cs = new ClientChangesImpl(SessionImplSocket.this);
505 //                    SynchronizeContext context = new SynchronizeContext(callerThread, SessionImplSocket.this, cs, 1);
506 //                    try {
507 //                        // Send all local changes to server so it can calculate correct reverse change set.
508 //                        // This will call clusterStream.accept().
509 //                        state.cancelCommit(context, clusterStream);
510 //                        if (!empty) {
511 //                            if (!context.isOk()) // this is a blocking operation
512 //                                throw new InternalException("Cancel failed. This should never happen.");
513 //                            getQueryProvider2().performDirtyUpdates(writeState.getGraph());
514 //                        }
515 //                        state.cancelCommit2(context, clusterStream);
516 //                    } catch (DatabaseException e3) {
517 //                        Logger.defaultLogError("Write request cancel caused an unexpected error.", e3);
518 //                    } finally {
519 //                        clusterStream.setOff(false);
520 //                        clientChanges = new ClientChangesImpl(SessionImplSocket.this);
521 //                    }
522
523                 } finally {
524
525                     fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
526
527                 }
528
529                 task.finish();
530
531                 if(Development.DEVELOPMENT) {
532                     try {
533                     if(Development.<Boolean>getProperty(DevelopmentKeys.SESSION_LOG_WRITES, Bindings.BOOLEAN))
534                         System.err.println("finish write '" + request + "'");
535                     } catch (Throwable t) {
536                         Logger.defaultLogError(t);
537                     }
538                 }
539
540             }
541
542         }, combine);
543
544     }
545
546     public <T> void scheduleRequest(final WriteResult<T> request, final Procedure<T> procedure, final Semaphore notify) {
547         scheduleRequest(request, procedure, notify, null);
548     }
549
550     /* (non-Javadoc)
551      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.WriteResult, org.simantics.db.procedure.Procedure, java.util.concurrent.Semaphore, java.lang.Boolean)
552      */
553     @Override
554     public <T> void scheduleRequest(final WriteResult<T> request, final Procedure<T> procedure, final Semaphore notify, Boolean combine) {
555
556         assert (request != null);
557
558         requestManager.scheduleWrite(new SessionTask(null) {
559
560             @Override
561             public void run0(int thread) {
562
563                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
564
565                 fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
566
567                 flushCounter = 0;
568                 Disposable.safeDispose(clientChanges);
569                 clientChanges = new ClientChangesImpl(SessionImplSocket.this);
570
571                 VirtualGraph vg = getProvider(request.getProvider());
572                 WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
573
574                 try {
575                     WriteState<T> writeStateT = new WriteState<T>(writer, request, notify, procedure);
576                     writeState = writeStateT;
577
578                     assert (null != writer);
579 //                    writer.state.barrier.inc();
580                     writeStateT.setResult(request.perform(writer));
581                     assert (null != writer);
582
583 //                    writer.state.barrier.dec();
584 //                    writer.waitAsync(null);
585
586                 } catch (Throwable e) {
587
588 //                    writer.state.barrier.dec();
589 //                    writer.waitAsync(null);
590
591                     writeState.except(e);
592
593 //                  state.stopWriteTransaction(clusterStream);
594 //
595 //              } catch (Throwable e) {
596 //                  // Log it first, just to be safe that the error is always logged.
597 //                  Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", e);
598 //
599 //                  try {
600 //                      // Callback is client code, we have to be prepared for it to throw unexpected exceptions.
601 //                      // All we can do here is to log those, can't really pass them anywhere.
602 //                      if (procedure != null) {
603 //                          if(e instanceof DatabaseException) procedure.exception((DatabaseException)e);
604 //                          else procedure.exception(new DatabaseException(e));
605 //                      }
606 //                  } catch (Throwable e2) {
607 //                      Logger.defaultLogError("Write request callback caused an unexpected error, see exception.", e2);
608 //                  }
609 //
610 //                  clientChanges = new ClientChangesImpl(SessionImplSocket.this);
611 //
612 //                  state.stopWriteTransaction(clusterStream);
613
614                 } finally {
615                     fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
616                 }
617
618 //              if(notify != null) notify.release();
619
620                 task.finish();
621
622             }
623
624         }, combine);
625
626     }
627
628     public <T> void scheduleRequest(final DelayedWrite request, final Consumer<DatabaseException> callback, final Semaphore notify) {
629         scheduleRequest(request, callback, notify, null);
630     }
631
632     /* (non-Javadoc)
633      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.DelayedWrite, org.simantics.utils.datastructures.Callback, java.util.concurrent.Semaphore, java.lang.Boolean)
634      */
635     @Override
636     public <T> void scheduleRequest(final DelayedWrite request, final Consumer<DatabaseException> callback, final Semaphore notify, Boolean combine) {
637
638         final ITask total = ThreadLogger.getInstance().begin("ScheduleDelayedWrite");
639
640         assert (request != null);
641
642         requestManager.scheduleWrite(new SessionTask(null) {
643
644             @Override
645             public void run0(int thread) {
646                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
647
648                 Procedure<Object> stateProcedure = new Procedure<Object>() {
649                     @Override
650                     public void execute(Object result) {
651                         if (callback != null)
652                             callback.accept(null);
653                     }
654                     @Override
655                     public void exception(Throwable t) {
656                         if (callback != null) {
657                             if (t instanceof DatabaseException) callback.accept((DatabaseException) t);
658                             else callback.accept(new DatabaseException(t));
659                         } else
660                             Logger.defaultLogError("Unhandled exception", t);
661                     }
662                 };
663
664                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
665                 delayedWriteState = new WriteStateBase<Object>(request, notify, stateProcedure);
666                 DelayedWriteGraph dwg = null;
667 //                newGraph.state.barrier.inc();
668
669                 try {
670                     dwg = new DelayedWriteGraph(newGraph);
671                     request.perform(dwg);
672                 } catch (Throwable e) {
673                     delayedWriteState.except(e);
674                     total.finish();
675                     dwg.close();
676                     return;
677                 } finally {
678 //                    newGraph.state.barrier.dec();
679 //                    newGraph.waitAsync(request);
680                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
681                 }
682
683                 delayedWriteState = null;
684
685                 ITask task2 = ThreadLogger.getInstance().begin("DelayedWriteCommit");
686                 fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
687
688                 flushCounter = 0;
689                 Disposable.safeDispose(clientChanges);
690                 clientChanges = new ClientChangesImpl(SessionImplSocket.this);
691
692                 acquireWriteOnly();
693
694                 VirtualGraph vg = getProvider(request.getProvider());
695                 WriteSupport writeSupport = vg != null ? new VirtualWriteOnlySupport() : new WriteOnlySupport();
696
697                 WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
698
699                 writeState = new WriteState<Object>(writer, request, notify, stateProcedure);
700
701                 assert (null != writer);
702 //                writer.state.barrier.inc();
703
704                 try {
705                     // Cannot cancel
706                     dwg.commit(writer, request);
707
708                         if(defaultClusterSet != null) {
709                                 XSupport xs = getService(XSupport.class);
710                                 defaultClusterSet = xs.convertDelayedResourceToResource(defaultClusterSet);
711                                 ClusteringSupport cs = getService(ClusteringSupport.class);
712                                 clusterSetsSupport.put(defaultClusterSet.getResourceId(), cs.getCluster(defaultClusterSet));
713                         }
714
715                     // This makes clusters available from server
716                     clusterStream.reallyFlush();
717
718                     releaseWriteOnly(writer);
719                     handleUpdatesAndMetadata(writer);
720
721                 } catch (ServiceException e) {
722 //                    writer.state.barrier.dec();
723 //                    writer.waitAsync(null);
724
725                     // These shall be requested from server
726                     clusterTable.removeWriteOnlyClusters();
727                     // This makes clusters available from server
728                     clusterStream.reallyFlush();
729
730                     releaseWriteOnly(writer);
731                     writeState.except(e);
732                 } finally {
733                     // Debugging & Profiling
734                     fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
735                     task2.finish();
736                     total.finish();
737                 }
738             }
739
740         }, combine);
741
742     }
743
744     public <T> void scheduleRequest(final DelayedWriteResult<T> request, final Procedure<T> procedure, final Semaphore notify) {
745         scheduleRequest(request, procedure, notify, null);
746     }
747
748     /* (non-Javadoc)
749      * @see fi.vtt.simantics.procore.internal.WriteRequestScheduler#scheduleRequest(org.simantics.db.request.DelayedWriteResult, org.simantics.db.procedure.Procedure, java.util.concurrent.Semaphore, java.lang.Boolean)
750      */
751     @Override
752     public <T> void scheduleRequest(final DelayedWriteResult<T> request, final Procedure<T> procedure, final Semaphore notify, Boolean combine) {
753         throw new Error("Not implemented");
754     }
755
756     protected ClusterImpl getNewResourceCluster() throws DatabaseException {
757         ClusterImpl cluster = clusterTable.getNewResourceCluster(clusterTranslator, graphSession, writeOnly);
758         if((serviceMode & SERVICE_MODE_CREATE) > 0) {
759             createdClusters.add(cluster.clusterId);
760         }
761         return cluster;
762     }
763
764     class WriteOnlySupport implements WriteSupport {
765
766         ClusterStream stream;
767         ClusterImpl currentCluster;
768
769         public WriteOnlySupport() {
770             this.stream = clusterStream;
771         }
772
773         @Override
774         public void claim(VirtualGraph provider, Resource subject, Resource predicate, Resource object) throws ServiceException {
775             claim(provider, ((ResourceImpl)subject).id, ((ResourceImpl)predicate).id, ((ResourceImpl)object).id );
776         }
777
778         @Override
779         public void claim(VirtualGraph provider, int s, int p, int o) throws ServiceException {
780
781             ClusterImpl cluster = clusterTable.getClusterByResourceKey(s);
782
783             if (cluster.getImmutable() && (serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0)
784                 if(s != queryProvider2.getRootLibrary())
785                     throw new ImmutableException("Trying to modify immutable resource key=" + s);
786
787             try {
788                 maintainCluster(cluster, cluster.addRelation(s, p, o, clusterTranslator));
789             } catch (DatabaseException e) {
790                 Logger.defaultLogError(e);
791                 //throw e;
792             }
793
794             clientChanges.invalidate(s);
795
796             if (cluster.isWriteOnly())
797                 return;
798             queryProvider2.updateStatements(s, p);
799
800         }
801
802         @Override
803         public void claimValue(VirtualGraph provider, Resource resource, byte[] value) throws DatabaseException {
804             claimValue(provider, ((ResourceImpl)resource).id, value, value.length);
805         }
806
807         @Override
808         public void claimValue(VirtualGraph provider, int rid, byte[] value, int length) {
809
810             ClusterImpl cluster = clusterTable.getClusterByResourceKey(rid);
811             try {
812                 maintainCluster(cluster, cluster.setValue(rid, value, length, clusterTranslator));
813             } catch (DatabaseException e) {
814                 Logger.defaultLogError(e);
815             }
816
817             clientChanges.invalidate(rid);
818
819             if (cluster.isWriteOnly())
820                 return;
821             queryProvider2.updateValue(rid);
822
823         }
824
825         @Override
826         public void claimValue(VirtualGraph provider, Resource resource, ByteReader reader, int amount) throws DatabaseException {
827
828             if(amount < 65536) {
829                 claimValue(provider,resource, reader.readBytes(null, amount));
830                 return;
831             }
832
833             byte[] bytes = new byte[65536];
834
835             int rid = ((ResourceImpl)resource).id;
836             ClusterImpl cluster = clusterTable.getClusterByResourceKey(rid);
837             try {
838                 int left = amount;
839                 while(left > 0) {
840                     int block = Math.min(left, 65536);
841                     reader.readBytes(bytes, block);
842                     maintainCluster(cluster, cluster.modiValueEx(rid, amount-left, block, bytes, 0, clusterTranslator));
843                     left -= block;
844                 }
845             } catch (DatabaseException e) {
846                 Logger.defaultLogError(e);
847             }
848
849             clientChanges.invalidate(rid);
850
851             if (cluster.isWriteOnly())
852                 return;
853             queryProvider2.updateValue(rid);
854
855         }
856
857         private void maintainCluster(ClusterImpl before, ClusterI after_) {
858             if(after_ != null && after_ != before) {
859                 ClusterImpl after = (ClusterImpl)after_;
860                 if(currentCluster == before) {
861                     currentCluster = after;
862                 }
863                 clusterTable.replaceCluster(after);
864             }
865         }
866
867         public int createResourceKey(int foreignCounter) throws DatabaseException {
868             if(currentCluster == null) {
869                 currentCluster = getNewResourceCluster();
870             }
871             if(currentCluster.getNumberOfResources(clusterTranslator) == ClusterTable.CLUSTER_FILL_SIZE) {
872                 ClusterWriteOnly newCluster = (ClusterWriteOnly)getNewResourceCluster();
873                 newCluster.foreignLookup = new byte[foreignCounter];
874                 currentCluster = newCluster;
875                 if (DEBUG)
876                     System.err.println("foreignLookup " + currentCluster + " " + foreignCounter);
877             }
878             return currentCluster.createResource(clusterTranslator);
879         }
880
881         @Override
882         public Resource createResource(VirtualGraph provider) throws DatabaseException {
883             if(currentCluster == null) {
884                 if (null != defaultClusterSet) {
885                         ResourceImpl result = getNewResource(defaultClusterSet);
886                     currentCluster = clusterTable.getClusterByResourceKey(result.id);
887                     return result;
888                 } else {
889                     currentCluster = getNewResourceCluster();
890                 }
891             }
892             if(currentCluster.getNumberOfResources(clusterTranslator) >= ClusterTable.CLUSTER_FILL_SIZE) {
893                 if (null != defaultClusterSet) {
894                         ResourceImpl result = getNewResource(defaultClusterSet);
895                     currentCluster = clusterTable.getClusterByResourceKey(result.id);
896                     return result;
897                 } else {
898                     currentCluster = getNewResourceCluster();
899                 }
900             }
901             return new ResourceImpl(resourceSupport, currentCluster.createResource(clusterTranslator));
902         }
903
904         @Override
905         public Resource createResource(VirtualGraph provider, long clusterId)
906         throws DatabaseException {
907             return getNewResource(clusterId);
908         }
909
910         @Override
911         public Resource createResource(VirtualGraph provider, Resource clusterSet)
912         throws DatabaseException {
913             return getNewResource(clusterSet);
914         }
915
916         @Override
917         public void createClusterSet(VirtualGraph provider, Resource clusterSet)
918                 throws DatabaseException {
919             getNewClusterSet(clusterSet);
920         }
921
922         @Override
923         public boolean hasClusterSet(VirtualGraph dummy, Resource clusterSet)
924         throws ServiceException {
925             return containsClusterSet(clusterSet);
926         }
927
928         public void selectCluster(long cluster) {
929                 currentCluster = clusterTable.getClusterByClusterId(cluster);
930                 long setResourceId = clusterSetsSupport.getSet(cluster);
931                 clusterSetsSupport.put(setResourceId, cluster);
932         }
933
934         @Override
935         public Resource setDefaultClusterSet(Resource clusterSet)
936         throws ServiceException {
937                 Resource result = setDefaultClusterSet4NewResource(clusterSet);
938                 if(clusterSet != null) {
939                         long id = clusterSetsSupport.get(clusterSet.getResourceId());
940                         currentCluster = clusterTable.getClusterByClusterId(id);
941                         return result;
942                 } else {
943                         currentCluster = null;
944                         return null;
945                 }
946         }
947
948         @Override
949         public void denyValue(VirtualGraph provider, Resource resource) throws ServiceException {
950
951                  provider = getProvider(provider);
952              if (null == provider) {
953                  int key = ((ResourceImpl)resource).id;
954
955
956
957                  ClusterI cluster = clusterTable.getClusterProxyByResourceKey(key);
958 //                 if (cluster.getImmutable() && (serviceMode & SessionImplSocket.SERVICE_MODE_ALLOW) == 0)
959 //                      if(key != queryProvider2.getRootLibrary())
960 //                              throw new ImmutableException("Trying to modify immutable resource key=" + key);
961
962 //                 try {
963 //                     cluster.removeValue(key, clusterTranslator);
964 //                 } catch (DatabaseException e) {
965 //                     Logger.defaultLogError(e);
966 //                     return;
967 //                 }
968
969                  clusterTable.writeOnlyInvalidate(cluster);
970
971                  try {
972                          int sResourceKey = ClusterTraits.getResourceIndexFromResourceKey(key);
973                          clusterTranslator.addStatementIndex(cluster, sResourceKey, cluster.getClusterUID(), ClusterChange.DELETE_OPERATION);
974                          clusterTranslator.removeValue(cluster);
975                  } catch (DatabaseException e) {
976                          Logger.defaultLogError(e);
977                  }
978
979                  queryProvider2.invalidateResource(key);
980                  clientChanges.invalidate(key);
981
982              } else {
983                  ((VirtualGraphImpl)provider).denyValue(((ResourceImpl) resource).id);
984                  queryProvider2.updateValue(querySupport.getId(resource));
985                  clientChanges.claimValue(resource);
986              }
987
988
989         }
990
991         @Override
992         public void flush(boolean intermediate) {
993             throw new UnsupportedOperationException();
994         }
995
996         @Override
997         public void flushCluster() {
998             clusterTable.flushCluster(graphSession);
999             if(defaultClusterSet != null) {
1000                 clusterSetsSupport.put(defaultClusterSet.getResourceId(), Constants.NewClusterId);
1001             }
1002             currentCluster = null;
1003         }
1004
1005         @Override
1006         public void flushCluster(Resource r) {
1007             throw new UnsupportedOperationException("flushCluster resource " + r);
1008         }
1009
1010         @Override
1011         public void gc() {
1012         }
1013
1014         @Override
1015         public boolean removeStatement(VirtualGraph provider, Resource subject, Resource predicate,
1016                 Resource object) {
1017
1018                 int s = ((ResourceImpl)subject).id;
1019                 int p = ((ResourceImpl)predicate).id;
1020                 int o = ((ResourceImpl)object).id;
1021
1022                 provider = getProvider(provider);
1023                 if (null == provider) {
1024
1025                         ClusterI cluster = clusterTable.getClusterProxyByResourceKey(s);
1026                         clusterTable.writeOnlyInvalidate(cluster);
1027
1028                         try {
1029
1030                                 int sResourceKey = ClusterTraits.getResourceIndexFromResourceKey(s);
1031
1032                                 ClusterI pc = clusterTable.getClusterProxyByResourceKey(p);
1033                                 ClusterI oc = clusterTable.getClusterProxyByResourceKey(o);
1034
1035                                 clusterTranslator.addStatementIndex(cluster, sResourceKey, cluster.getClusterUID(), ClusterChange.REMOVE_OPERATION);
1036                                 clusterTranslator.addStatementIndex(cluster, p, pc.getClusterUID(), ClusterStream.NULL_OPERATION);
1037                                 clusterTranslator.addStatementIndex(cluster, o, oc.getClusterUID(), ClusterStream.NULL_OPERATION);
1038                                 clusterTranslator.removeStatement(cluster);
1039
1040                                 queryProvider2.invalidateResource(s);
1041                                 clientChanges.invalidate(s);
1042
1043                         } catch (DatabaseException e) {
1044
1045                                 Logger.defaultLogError(e);
1046
1047                         }
1048
1049                         return true;
1050
1051                 } else {
1052                         
1053                         ((VirtualGraphImpl)provider).deny(s, p, o);
1054                         queryProvider2.invalidateResource(s);
1055                         clientChanges.invalidate(s);
1056
1057                         return true;
1058
1059                 }
1060
1061         }
1062
1063         @Override
1064         public void setValue(VirtualGraph provider, Resource resource, byte[] value) {
1065             throw new UnsupportedOperationException();
1066         }
1067
1068         @Override
1069         public boolean writeOnly() {
1070             return true;
1071         }
1072
1073         @Override
1074         public void performWriteRequest(WriteGraph graph, Write request) throws DatabaseException {
1075             writeSupport.performWriteRequest(graph, request);
1076         }
1077
1078         @Override
1079         public <T> T performWriteRequest(WriteGraph graph, WriteResult<T> request) throws DatabaseException {
1080             throw new UnsupportedOperationException();
1081         }
1082
1083         @Override
1084         public void performWriteRequest(WriteGraph graph, WriteOnly request) throws DatabaseException {
1085             throw new UnsupportedOperationException();
1086         }
1087
1088         @Override
1089         public <T> void addMetadata(Metadata data) throws ServiceException {
1090             writeSupport.addMetadata(data);
1091         }
1092
1093         @Override
1094         public <T extends Metadata> T getMetadata(Class<T> clazz) throws ServiceException {
1095             return writeSupport.getMetadata(clazz);
1096         }
1097
1098         @Override
1099         public TreeMap<String, byte[]> getMetadata() {
1100             return writeSupport.getMetadata();
1101         }
1102
1103         @Override
1104         public void commitDone(WriteTraits writeTraits, long csid) {
1105             writeSupport.commitDone(writeTraits, csid);
1106         }
1107
1108         @Override
1109         public void clearUndoList(WriteTraits writeTraits) {
1110             writeSupport.clearUndoList(writeTraits);
1111         }
1112         @Override
1113         public int clearMetadata() {
1114             return writeSupport.clearMetadata();
1115         }
1116
1117                 @Override
1118                 public void startUndo() {
1119                         writeSupport.startUndo();
1120                 }
1121     }
1122
1123     class VirtualWriteOnlySupport implements WriteSupport {
1124
1125 //        @Override
1126 //        public void addStatement(VirtualGraph provider, Resource subject, Resource predicate,
1127 //                Resource object) {
1128 //            throw new UnsupportedOperationException();
1129 //        }
1130
1131         @Override
1132         public void claim(VirtualGraph provider, Resource subject, Resource predicate, Resource object) {
1133
1134             TransientGraph impl = (TransientGraph)provider;
1135             impl.claim(querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object));
1136             getQueryProvider2().updateStatements(querySupport.getId(subject), querySupport.getId(predicate));
1137             clientChanges.claim(subject, predicate, object);
1138
1139         }
1140
1141         @Override
1142         public void claim(VirtualGraph provider, int subject, int predicate, int object) {
1143
1144             TransientGraph impl = (TransientGraph)provider;
1145             impl.claim(subject, predicate, object);
1146             getQueryProvider2().updateStatements(subject, predicate);
1147             clientChanges.claim(subject, predicate, object);
1148
1149         }
1150
1151         @Override
1152         public void claimValue(VirtualGraph provider, Resource resource, byte[] value) throws DatabaseException {
1153             claimValue(provider, ((ResourceImpl)resource).id, value, value.length);
1154         }
1155
1156         @Override
1157         public void claimValue(VirtualGraph provider, int resource, byte[] value, int length) {
1158             ((VirtualGraphImpl)provider).claimValue(resource, value, length);
1159             getQueryProvider2().updateValue(resource);
1160             clientChanges.claimValue(resource);
1161         }
1162
1163         @Override
1164         public void claimValue(VirtualGraph provider, Resource resource, ByteReader reader, int amount) throws DatabaseException {
1165             byte[] value = reader.readBytes(null, amount);
1166             claimValue(provider, resource, value);
1167         }
1168
1169         @Override
1170         public Resource createResource(VirtualGraph provider) {
1171             TransientGraph impl = (TransientGraph)provider;
1172             return impl.getResource(impl.newResource(false));
1173         }
1174
1175         @Override
1176         public Resource createResource(VirtualGraph provider, long clusterId) {
1177             throw new UnsupportedOperationException();
1178         }
1179
1180         @Override
1181         public Resource createResource(VirtualGraph provider, Resource clusterSet)
1182         throws DatabaseException {
1183             throw new UnsupportedOperationException();
1184         }
1185
1186         @Override
1187         public void createClusterSet(VirtualGraph provider, Resource clusterSet)
1188         throws DatabaseException {
1189             throw new UnsupportedOperationException();
1190         }
1191
1192         @Override
1193         public boolean hasClusterSet(VirtualGraph provider, Resource clusterSet)
1194         throws ServiceException {
1195             throw new UnsupportedOperationException();
1196         }
1197
1198         @Override
1199         public Resource setDefaultClusterSet(Resource clusterSet)
1200         throws ServiceException {
1201                 return null;
1202         }
1203
1204         @Override
1205         public void denyValue(VirtualGraph provider, Resource resource) {
1206             ((VirtualGraphImpl)provider).denyValue(((ResourceImpl) resource).id);
1207             getQueryProvider2().updateValue(querySupport.getId(resource));
1208             // NOTE: this only keeps track of value changes by-resource.
1209             clientChanges.claimValue(resource);
1210         }
1211
1212         @Override
1213         public void flush(boolean intermediate) {
1214             throw new UnsupportedOperationException();
1215         }
1216
1217         @Override
1218         public void flushCluster() {
1219             throw new UnsupportedOperationException();
1220         }
1221
1222         @Override
1223         public void flushCluster(Resource r) {
1224             throw new UnsupportedOperationException("Resource " + r);
1225         }
1226
1227         @Override
1228         public void gc() {
1229         }
1230
1231         @Override
1232         public boolean removeStatement(VirtualGraph provider, Resource subject, Resource predicate, Resource object) {
1233             TransientGraph impl = (TransientGraph) provider;
1234             impl.deny(querySupport.getId(subject), querySupport.getId(predicate), querySupport.getId(object));
1235             getQueryProvider2().updateStatements(querySupport.getId(subject), querySupport.getId(predicate));
1236             clientChanges.deny(subject, predicate, object);
1237             return true;
1238         }
1239
1240         @Override
1241         public void setValue(VirtualGraph provider, Resource resource, byte[] value) {
1242             throw new UnsupportedOperationException();
1243         }
1244
1245         @Override
1246         public boolean writeOnly() {
1247             return true;
1248         }
1249
1250         @Override
1251         public void performWriteRequest(WriteGraph graph, Write request) throws DatabaseException {
1252             throw new UnsupportedOperationException();
1253         }
1254
1255         @Override
1256         public <T> T performWriteRequest(WriteGraph graph, WriteResult<T> request) throws DatabaseException {
1257             throw new UnsupportedOperationException();
1258         }
1259
1260         @Override
1261         public void performWriteRequest(WriteGraph graph, WriteOnly request) throws DatabaseException {
1262             throw new UnsupportedOperationException();
1263         }
1264
1265         @Override
1266         public <T> void addMetadata(Metadata data) throws ServiceException {
1267             throw new UnsupportedOperationException();
1268         }
1269
1270         @Override
1271         public <T extends Metadata> T getMetadata(Class<T> clazz) throws ServiceException {
1272             throw new UnsupportedOperationException();
1273         }
1274
1275         @Override
1276         public TreeMap<String, byte[]> getMetadata() {
1277             throw new UnsupportedOperationException();
1278         }
1279
1280         @Override
1281         public void commitDone(WriteTraits writeTraits, long csid) {
1282         }
1283
1284         @Override
1285         public void clearUndoList(WriteTraits writeTraits) {
1286         }
1287
1288         @Override
1289         public int clearMetadata() {
1290             return 0;
1291         }
1292
1293                 @Override
1294                 public void startUndo() {
1295                 }
1296     }
1297
1298     private <T> void performWriteOnly(WriteOnlyResult<T> request, Semaphore notify, Procedure<T> callback) {
1299
1300         try {
1301
1302             //int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1303
1304             fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1305
1306             flushCounter = 0;
1307             Disposable.safeDispose(clientChanges);
1308             clientChanges = new ClientChangesImpl(SessionImplSocket.this);
1309
1310             acquireWriteOnly();
1311
1312             WriteSupport writeSupport = request.getProvider() != null ? new VirtualWriteOnlySupport() : new WriteOnlySupport();
1313
1314             WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, request.getProvider());
1315
1316             WriteState<T> writeStateT = new WriteState<T>(writer, request, notify, callback);
1317             writeState = writeStateT;
1318
1319             assert (null != writer);
1320 //            writer.state.barrier.inc();
1321             long start = System.nanoTime();
1322             T result = request.perform(writer);
1323             long duration = System.nanoTime() - start;
1324             if (DEBUG) {
1325                 System.err.println("################");
1326                 System.err.println("WriteOnly duration " + 1e-9*duration);
1327             }
1328             writeStateT.setResult(result);
1329
1330             // This makes clusters available from server
1331             clusterStream.reallyFlush();
1332
1333             // This will trigger query updates
1334             releaseWriteOnly(writer);
1335             assert (null != writer);
1336
1337             handleUpdatesAndMetadata(writer);
1338
1339         } catch (CancelTransactionException e) {
1340
1341             releaseWriteOnly(writeState.getGraph());
1342
1343             clusterTable.removeWriteOnlyClusters();
1344             state.stopWriteTransaction(clusterStream);
1345
1346         } catch (Throwable e) {
1347
1348             e.printStackTrace();
1349
1350             releaseWriteOnly(writeState.getGraph());
1351
1352             clusterTable.removeWriteOnlyClusters();
1353
1354             if (callback != null)
1355                 callback.exception(new DatabaseException(e));
1356
1357             state.stopWriteTransaction(clusterStream);
1358             Logger.defaultLogError("Write transaction caused an unexpected error, see exception.", e);
1359
1360         } finally  {
1361
1362             fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1363
1364         }
1365
1366     }
1367
1368     public <T> void scheduleRequest(final WriteOnly request, final Consumer<DatabaseException> callback, final Semaphore notify) {
1369         scheduleRequest(request, callback, notify, null);
1370     }
1371
1372     @Override
1373     public <T> void scheduleRequest(final WriteOnly request, final Consumer<DatabaseException> callback, final Semaphore notify, Boolean combine) {
1374
1375         assertAlive();
1376
1377         assert (request != null);
1378
1379         requestManager.scheduleWrite(new SessionTask(null) {
1380
1381             @Override
1382             public void run0(int thread) {
1383
1384                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
1385
1386                     try {
1387
1388                         fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1389
1390                         flushCounter = 0;
1391                         Disposable.safeDispose(clientChanges);
1392                         clientChanges = new ClientChangesImpl(SessionImplSocket.this);
1393
1394                         acquireWriteOnly();
1395
1396                         VirtualGraph vg = getProvider(request.getProvider());
1397                         WriteSupport writeSupport = vg != null ? new VirtualWriteOnlySupport() : new WriteOnlySupport();
1398
1399                         WriteGraphImpl writer = WriteGraphImpl.create(getQueryProvider2(), writeSupport, vg);
1400
1401                         writeState = new WriteState<Object>(writer, request, notify, new Procedure<Object>() {
1402
1403                             @Override
1404                             public void execute(Object result) {
1405                                 if(callback != null) callback.accept(null);
1406                             }
1407
1408                             @Override
1409                             public void exception(Throwable t) {
1410                                 if(callback != null) callback.accept((DatabaseException)t);
1411                             }
1412
1413                         });
1414
1415                         assert (null != writer);
1416 //                        writer.state.barrier.inc();
1417
1418                         try {
1419
1420                             request.perform(writer);
1421
1422                         } catch (Throwable e) {
1423
1424 //                            writer.state.barrier.dec();
1425 //                            writer.waitAsync(null);
1426
1427                             releaseWriteOnly(writer);
1428
1429                             clusterTable.removeWriteOnlyClusters();
1430
1431                             if(!(e instanceof CancelTransactionException)) {
1432                                 if (callback != null)
1433                                     callback.accept(new DatabaseException(e));
1434                             }
1435
1436                             writeState.except(e);
1437
1438                             return;
1439
1440                         }
1441
1442                         // This makes clusters available from server
1443                         boolean empty = clusterStream.reallyFlush();
1444                         // This was needed to make WO requests call metadata listeners.
1445                         // NOTE: the calling event does not contain clientChanges information.
1446                         if (!empty && clientChanges.isEmpty())
1447                             clientChanges.setNotEmpty(true);
1448                         releaseWriteOnly(writer);
1449                         assert (null != writer);
1450                     } finally  {
1451
1452                         fireSessionVariableChange(SessionVariables.QUEUED_WRITES);
1453
1454                     }
1455
1456
1457                 task.finish();
1458
1459             }
1460
1461         }, combine);
1462
1463     }
1464
1465     public <T> void scheduleRequest(final WriteOnlyResult<T> request, final Procedure<T> callback, final Semaphore notify) {
1466         scheduleRequest(request, callback, notify, null);
1467     }
1468
1469     /* (non-Javadoc)
1470      * @see org.simantics.db.impl.support.WriteRequestScheduleSupport#scheduleRequest(org.simantics.db.request.WriteOnlyResult, org.simantics.db.procedure.Procedure, java.util.concurrent.Semaphore, java.lang.Boolean)
1471      */
1472     @Override
1473     public <T> void scheduleRequest(final WriteOnlyResult<T> request, final Procedure<T> callback, final Semaphore notify, Boolean combine) {
1474
1475         assert (request != null);
1476
1477         requestManager.scheduleWrite(new SessionTask(null) {
1478
1479             @Override
1480             public void run0(int thread) {
1481
1482                 ITask task = ThreadLogger.getInstance().begin("WriteRequest " + request);
1483
1484                 performWriteOnly(request, notify, callback);
1485
1486                 task.finish();
1487
1488             }
1489
1490         }, combine);
1491
1492     }
1493
1494     public <T> void scheduleRequest(final Read<T> request, final AsyncProcedure<T> procedure, final Semaphore notify, final DataContainer<Throwable> throwable, final DataContainer<T> result) {
1495
1496         assert (request != null);
1497         assert (procedure != null);
1498
1499         //int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1500
1501         requestManager.scheduleRead(new SessionRead(throwable, notify) {
1502
1503             @Override
1504             public void run0(int thread) {
1505
1506                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1507
1508                 ListenerBase listener = getListenerBase(procedure);
1509
1510                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1511
1512                 // This is never synced but increase to prevent it from visiting 0
1513                 newGraph.asyncBarrier.inc();
1514
1515                 try {
1516
1517                     if (listener != null) {
1518
1519                         try {
1520                                 
1521                                 AsyncProcedure ap = new AsyncProcedure<T>() {
1522
1523                                         @Override
1524                                         public void exception(AsyncReadGraph graph, Throwable t) {
1525                                                 procedure.exception(graph, t);
1526                                                 if(throwable != null) {
1527                                                         throwable.set(t);
1528                                                 } else {
1529                                                         //                                    ErrorLogger.defaultLogError("Unhandled exception", t);
1530                                                 }
1531                                         }
1532
1533                                         @Override
1534                                         public void execute(AsyncReadGraph graph, T t) {
1535                                                 if(result != null) result.set(t);
1536                                                 procedure.execute(graph, t);
1537                                         }
1538
1539                                 };
1540                                 
1541                                 QueryCache.runnerReadEntry(newGraph, request, null, listener, ap, true);
1542                                 
1543                         } catch (Throwable t) {
1544                             // This is handled by the AsyncProcedure
1545                                 //Logger.defaultLogError("Internal error", t);
1546                         }
1547
1548                     } else {
1549
1550                         try {
1551
1552 //                            newGraph.state.barrier.inc();
1553
1554                             T t = request.perform(newGraph);
1555
1556                             try {
1557
1558                                 if(result != null) result.set(t);
1559                                 procedure.execute(newGraph, t);
1560
1561                             } catch (Throwable th) {
1562
1563                                 if(throwable != null) {
1564                                     throwable.set(th);
1565                                 } else {
1566                                     Logger.defaultLogError("Unhandled exception", th);
1567                                 }
1568
1569                             }
1570
1571                         } catch (Throwable t) {
1572
1573                             if (DEBUG)
1574                                 t.printStackTrace();
1575
1576                             if(throwable != null) {
1577                                 throwable.set(t);
1578                             } else {
1579                                 Logger.defaultLogError("Unhandled exception", t);
1580                             }
1581
1582                             try {
1583
1584                                 procedure.exception(newGraph, t);
1585
1586                             } catch (Throwable t2) {
1587
1588                                 if(throwable != null) {
1589                                     throwable.set(t2);
1590                                 } else {
1591                                     Logger.defaultLogError("Unhandled exception", t2);
1592                                 }
1593
1594                             }
1595
1596                         }
1597
1598 //                        newGraph.state.barrier.dec();
1599 //                        newGraph.waitAsync(request);
1600
1601                     }
1602
1603                 } finally {
1604
1605                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1606
1607                 }
1608
1609             }
1610
1611         });
1612
1613     }
1614
1615     public <T> void scheduleRequest(final AsyncRead<T> request, final AsyncProcedure<T> procedure, final ListenerBase listener, final Semaphore notify) {
1616
1617         assert (request != null);
1618         assert (procedure != null);
1619
1620         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1621
1622         requestManager.scheduleRead(new SessionRead(null, notify) {
1623
1624             @Override
1625             public void run0(int thread) {
1626
1627                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1628
1629                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1630
1631                 try {
1632
1633                     if (listener != null) {
1634
1635                         try {
1636                                 QueryCacheBase.resultAsyncReadEntry(newGraph, request, null, listener, procedure);
1637                                 //QueryCache.runnerAsyncReadEntry(newGraph, request, null, listener, procedure, true);
1638                                                         //newGraph.processor.query(newGraph, request, null, procedure, listener);
1639                                                 } catch (DatabaseException e) {
1640                                                         Logger.defaultLogError(e);
1641                                                 }
1642
1643                     } else {
1644
1645 //                        final ResultCallWrappedSingleQueryProcedure4<T> wrapper = new ResultCallWrappedSingleQueryProcedure4<T>(
1646 //                                procedure, "request");
1647
1648                         BlockingAsyncProcedure<T> wrap = new BlockingAsyncProcedure<T>(newGraph.asyncBarrier, newGraph, procedure, request);
1649
1650                         try {
1651
1652                             request.perform(newGraph, wrap);
1653                             wrap.get();
1654
1655                         } catch (DatabaseException e) {
1656
1657                                                         Logger.defaultLogError(e);
1658
1659                         }
1660
1661                     }
1662
1663                 } finally {
1664
1665                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1666
1667                 }
1668
1669             }
1670
1671         });
1672
1673     }
1674
1675     public <T> void scheduleRequest(final MultiRead<T> request, final SyncMultiProcedure<T> procedure, final Semaphore notify) {
1676
1677         assert (request != null);
1678         assert (procedure != null);
1679
1680         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1681
1682         int sync = notify != null ? thread : -1;
1683
1684         requestManager.scheduleRead(new SessionRead(null, notify) {
1685
1686             @Override
1687             public void run0(int thread) {
1688
1689                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1690
1691                 ListenerBase listener = getListenerBase(procedure);
1692
1693                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1694
1695                 try {
1696
1697                     if (listener != null) {
1698
1699                         newGraph.processor.query(newGraph, request, null, procedure, listener);
1700
1701 //                        newGraph.waitAsync(request);
1702
1703                     } else {
1704
1705                         final ResultCallWrappedSyncQueryProcedure<T> wrapper = new ResultCallWrappedSyncQueryProcedure<T>(procedure);
1706
1707                         try {
1708
1709                             request.perform(newGraph, wrapper);
1710
1711                         } catch (Throwable t) {
1712
1713                             t.printStackTrace();
1714
1715                         }
1716
1717                     }
1718
1719                 } finally {
1720
1721                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1722
1723                 }
1724
1725             }
1726
1727         });
1728
1729     }
1730     
1731     public <T> void scheduleRequest(final AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure, final Semaphore notify) {
1732
1733         assert (request != null);
1734         assert (procedure != null);
1735
1736         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1737
1738         int sync = notify != null ? thread : -1;
1739
1740         requestManager.scheduleRead(new SessionRead(null, notify) {
1741
1742             @Override
1743             public void run0(int thread) {
1744
1745                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1746
1747                 ListenerBase listener = getListenerBase(procedure);
1748
1749                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1750
1751                 try {
1752
1753                     if (listener != null) {
1754
1755                         newGraph.processor.query(newGraph, request, null, procedure, listener);
1756
1757 //                        newGraph.waitAsync(request);
1758
1759                     } else {
1760
1761                         final ResultCallWrappedQueryProcedure4<T> wrapper = new ResultCallWrappedQueryProcedure4<T>(procedure);
1762
1763                         try {
1764
1765                             request.perform(newGraph, wrapper);
1766
1767                         } catch (Throwable t) {
1768
1769                             t.printStackTrace();
1770
1771                         }
1772
1773                     }
1774
1775                 } finally {
1776
1777                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1778
1779                 }
1780
1781             }
1782
1783         });
1784
1785     }
1786
1787     public <T> void scheduleRequest(final ExternalRead<T> request, final Procedure<T> procedure, final Semaphore notify, final DataContainer<Throwable> throwable, final DataContainer<T> result) {
1788
1789         assert (request != null);
1790         assert (procedure != null);
1791
1792         int thread = request.hashCode() & queryProvider2.THREAD_MASK;
1793
1794         requestManager.scheduleRead(new SessionRead(throwable, notify) {
1795
1796             @Override
1797             public void run0(int thread) {
1798
1799                 fireSessionVariableChange(SessionVariables.QUEUED_READS);
1800
1801                 ListenerBase listener = getListenerBase(procedure);
1802
1803                 final ReadGraphImpl newGraph = ReadGraphImpl.create(getQueryProvider2());
1804
1805                 try {
1806
1807                     if (listener != null) {
1808
1809                         try {
1810                             QueryCacheBase.resultExternalReadEntry(newGraph, request, null, listener, procedure);
1811                         } catch (DatabaseException e) {
1812                             Logger.defaultLogError(e);
1813                         }
1814
1815                     } else {
1816
1817 //                        newGraph.state.barrier.inc();
1818
1819                         request.register(newGraph, new Listener<T>() {
1820
1821                             @Override
1822                             public void exception(Throwable t) {
1823                                 if(throwable != null) throwable.set(t);
1824                                 procedure.exception(t);
1825 //                                newGraph.state.barrier.dec();
1826                             }
1827
1828                             @Override
1829                             public void execute(T t) {
1830                                 if(result != null) result.set(t);
1831                                 procedure.execute(t);
1832 //                                newGraph.state.barrier.dec();
1833                             }
1834
1835                             @Override
1836                             public boolean isDisposed() {
1837                                 return true;
1838                             }
1839
1840                         });
1841
1842 //                        newGraph.waitAsync(request);
1843
1844                     }
1845
1846                 } finally {
1847
1848                     fireSessionVariableChange(SessionVariables.QUEUED_READS);
1849
1850                 }
1851
1852             }
1853
1854         });
1855
1856     }
1857
1858
1859     @Override
1860     public <T> void asyncRequest(final Read<T> request, final AsyncProcedure<T> procedure) {
1861
1862         scheduleRequest(request, procedure, null, null, null);
1863
1864     }
1865
1866     @Override
1867     public <T> T syncRequest(Read<T> request, AsyncProcedure<T> procedure) throws DatabaseException {
1868         assertNotSession();
1869         Semaphore notify = new Semaphore(0);
1870         DataContainer<Throwable> container = new DataContainer<Throwable>();
1871         DataContainer<T> result = new DataContainer<T>();
1872         scheduleRequest(request, procedure, notify, container, result);
1873         acquire(notify, request);
1874         Throwable throwable = container.get();
1875         if(throwable != null) {
1876             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1877             else throw new DatabaseException("Unexpected exception", throwable);
1878         }
1879         return result.get();
1880     }
1881
1882     @Override
1883     public <T> T syncRequest(AsyncRead<T> request, final AsyncProcedure<T> procedure) throws DatabaseException {
1884         assertNotSession();
1885         Semaphore notify = new Semaphore(0);
1886         final DataContainer<Throwable> exceptionContainer = new DataContainer<Throwable>();
1887         final DataContainer<T> resultContainer = new DataContainer<T>();
1888         scheduleRequest(request, new AsyncProcedure<T>() {
1889             @Override
1890             public void exception(AsyncReadGraph graph, Throwable throwable) {
1891                 exceptionContainer.set(throwable);
1892                 procedure.exception(graph, throwable);
1893             }
1894             @Override
1895             public void execute(AsyncReadGraph graph, T result) {
1896                 resultContainer.set(result);
1897                 procedure.execute(graph, result);
1898             }
1899         }, getListenerBase(procedure), notify);
1900         acquire(notify, request, procedure);
1901         Throwable throwable = exceptionContainer.get();
1902         if (throwable != null) {
1903             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1904             else throw new DatabaseException("Unexpected exception", throwable);
1905         }
1906         return resultContainer.get();
1907     }
1908
1909
1910     @Override
1911     public <T> Collection<T> syncRequest(AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) throws DatabaseException {
1912         assertNotSession();
1913         Semaphore notify = new Semaphore(0);
1914         final DataContainer<Throwable> exceptionContainer = new DataContainer<Throwable>();
1915         scheduleRequest(request, new AsyncMultiProcedure<T>() {
1916             @Override
1917             public void exception(AsyncReadGraph graph, Throwable throwable) {
1918                 exceptionContainer.set(throwable);
1919                 procedure.exception(graph, throwable);
1920             }
1921             @Override
1922             public void execute(AsyncReadGraph graph, T result) {
1923                 procedure.execute(graph, result);
1924             }
1925             @Override
1926             public void finished(AsyncReadGraph graph) {
1927                 procedure.finished(graph);
1928             }
1929         }, notify);
1930         acquire(notify, request, procedure);
1931         Throwable throwable = exceptionContainer.get();
1932         if (throwable != null) {
1933             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1934             else throw new DatabaseException("Unexpected exception", throwable);
1935         }
1936         // TODO: implement return value
1937         System.err.println("TODO: implement return value for syncRequest(AsyncMultiRead, AsyncMultiProcedure)");
1938         return null;
1939     }
1940
1941     @Override
1942     public <T> T syncRequest(final ExternalRead<T> request) throws DatabaseException {
1943         return syncRequest(request, new ProcedureAdapter<T>());
1944
1945
1946 //        assert(request != null);
1947 //
1948 //        final DataContainer<T> result = new DataContainer<T>();
1949 //        final DataContainer<Throwable> exception = new DataContainer<Throwable>();
1950 //
1951 //        syncRequest(request, new Procedure<T>() {
1952 //
1953 //            @Override
1954 //            public void execute(T t) {
1955 //                result.set(t);
1956 //            }
1957 //
1958 //            @Override
1959 //            public void exception(Throwable t) {
1960 //                exception.set(t);
1961 //            }
1962 //
1963 //        });
1964 //
1965 //        Throwable t = exception.get();
1966 //        if(t != null) {
1967 //            if(t instanceof DatabaseException) throw (DatabaseException)t;
1968 //            else throw new DatabaseException("Unexpected exception in ReadGraph.syncRequest(PrimitiveRead)", t);
1969 //        }
1970 //
1971 //        return result.get();
1972
1973     }
1974
1975     @Override
1976     public <T> T syncRequest(ExternalRead<T> request, final Listener<T> procedure) throws DatabaseException {
1977         return syncRequest(request, (Procedure<T>)procedure);
1978     }
1979
1980
1981 //    @Override
1982 //    public <T> T syncRequest(Read<T> request, AsyncProcedure<T> procedure) throws DatabaseException {
1983 //        assertNotSession();
1984 //        Semaphore notify = new Semaphore(0);
1985 //        DataContainer<Throwable> container = new DataContainer<Throwable>();
1986 //        DataContainer<T> result = new DataContainer<T>();
1987 //        scheduleRequest(request, procedure, notify, container, result);
1988 //        acquire(notify, request);
1989 //        Throwable throwable = container.get();
1990 //        if(throwable != null) {
1991 //            if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
1992 //            else throw new DatabaseException("Unexpected exception", throwable);
1993 //        }
1994 //        return result.get();
1995 //    }
1996
1997
1998     @Override
1999     public <T> T syncRequest(ExternalRead<T> request, final Procedure<T> procedure) throws DatabaseException {
2000         assertNotSession();
2001         Semaphore notify = new Semaphore(0);
2002         final DataContainer<Throwable> container = new DataContainer<Throwable>();
2003         final DataContainer<T> result = new DataContainer<T>();
2004         scheduleRequest(request, procedure, notify, container, result);
2005         acquire(notify, request);
2006         Throwable throwable = container.get();
2007         if (throwable != null) {
2008             if(throwable instanceof DatabaseException) throw (DatabaseException)throwable;
2009             else throw new DatabaseException("Unexpected exception", throwable);
2010         }
2011         return result.get();
2012     }
2013
2014     @Override
2015     public void syncRequest(Write request) throws DatabaseException  {
2016         assertNotSession();
2017         assertAlive();
2018         Semaphore notify = new Semaphore(0);
2019         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2020         scheduleRequest(request, e -> exception.set(e), notify);
2021         acquire(notify, request);
2022         if(exception.get() != null) throw exception.get();
2023     }
2024
2025     @Override
2026     public <T> T syncRequest(WriteResult<T> request) throws DatabaseException  {
2027         assertNotSession();
2028         Semaphore notify = new Semaphore(0);
2029         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2030         final DataContainer<T> result = new DataContainer<T>();
2031         scheduleRequest(request, new Procedure<T>() {
2032
2033             @Override
2034             public void exception(Throwable t) {
2035               exception.set(t);
2036             }
2037
2038             @Override
2039             public void execute(T t) {
2040               result.set(t);
2041             }
2042
2043         }, notify);
2044         acquire(notify, request);
2045         if(exception.get() != null) {
2046             Throwable t = exception.get();
2047             if(t instanceof DatabaseException) throw (DatabaseException)t;
2048             else throw new DatabaseException(t);
2049         }
2050         return result.get();
2051     }
2052
2053     @Override
2054     public <T> T syncRequest(WriteOnlyResult<T> request) throws DatabaseException  {
2055         assertNotSession();
2056         Semaphore notify = new Semaphore(0);
2057         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2058         final DataContainer<T> result = new DataContainer<T>();
2059         scheduleRequest(request, new Procedure<T>() {
2060
2061             @Override
2062             public void exception(Throwable t) {
2063               exception.set(t);
2064             }
2065
2066             @Override
2067             public void execute(T t) {
2068               result.set(t);
2069             }
2070
2071         }, notify);
2072         acquire(notify, request);
2073         if(exception.get() != null) {
2074             Throwable t = exception.get();
2075             if(t instanceof DatabaseException) throw (DatabaseException)t;
2076             else throw new DatabaseException(t);
2077         }
2078         return result.get();
2079     }
2080
2081     @Override
2082     public <T> T syncRequest(DelayedWriteResult<T> request) throws DatabaseException  {
2083         assertNotSession();
2084         Semaphore notify = new Semaphore(0);
2085         final DataContainer<Throwable> exception = new DataContainer<Throwable>();
2086         final DataContainer<T> result = new DataContainer<T>();
2087         scheduleRequest(request, new Procedure<T>() {
2088
2089             @Override
2090             public void exception(Throwable t) {
2091               exception.set(t);
2092             }
2093
2094             @Override
2095             public void execute(T t) {
2096               result.set(t);
2097             }
2098
2099         }, notify);
2100         acquire(notify, request);
2101         if(exception.get() != null) {
2102             Throwable t = exception.get();
2103             if(t instanceof DatabaseException) throw (DatabaseException)t;
2104             else throw new DatabaseException(t);
2105         }
2106         return result.get();
2107     }
2108
2109     @Override
2110     public void syncRequest(DelayedWrite request) throws DatabaseException  {
2111         assertNotSession();
2112         Semaphore notify = new Semaphore(0);
2113         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2114         scheduleRequest(request, e -> exception.set(e), notify);
2115         acquire(notify, request);
2116         if(exception.get() != null) throw exception.get();
2117     }
2118
2119     @Override
2120     public void syncRequest(WriteOnly request) throws DatabaseException  {
2121         assertNotSession();
2122         assertAlive();
2123         Semaphore notify = new Semaphore(0);
2124         final DataContainer<DatabaseException> exception = new DataContainer<DatabaseException>();
2125         scheduleRequest(request, e -> exception.set(e), notify);
2126         acquire(notify, request);
2127         if(exception.get() != null) throw exception.get();
2128     }
2129
2130     /*
2131      *
2132      * GraphRequestProcessor interface
2133      */
2134
2135     @Override
2136     public <T> void asyncRequest(final AsyncRead<T> request, final AsyncProcedure<T> procedure) {
2137
2138         scheduleRequest(request, procedure, null, null);
2139
2140     }
2141
2142     @Override
2143     public <T> void asyncRequest(final AsyncMultiRead<T> request, final AsyncMultiProcedure<T> procedure) {
2144
2145         scheduleRequest(request, procedure, null);
2146
2147     }
2148
2149     @Override
2150     public <T> void asyncRequest(final ExternalRead<T> request, final Procedure<T> procedure) {
2151
2152         scheduleRequest(request, procedure, null, null, null);
2153
2154     }
2155
2156     @Override
2157     public void asyncRequest(final Write request, final Consumer<DatabaseException> callback) {
2158
2159         scheduleRequest(request, callback, null);
2160
2161     }
2162
2163     @Override
2164     public <T> void asyncRequest(final WriteResult<T> request, final Procedure<T> procedure) {
2165
2166         scheduleRequest(request, procedure, null);
2167
2168     }
2169
2170     @Override
2171     public <T> void asyncRequest(final WriteOnlyResult<T> request, final Procedure<T> procedure) {
2172
2173         scheduleRequest(request, procedure, null);
2174
2175     }
2176
2177     @Override
2178     public <T> void asyncRequest(final DelayedWriteResult<T> request, final Procedure<T> procedure) {
2179
2180         scheduleRequest(request, procedure, null);
2181
2182     }
2183
2184     @Override
2185     public void asyncRequest(final DelayedWrite request, final Consumer<DatabaseException> callback) {
2186
2187         scheduleRequest(request, callback, null);
2188
2189     }
2190
2191     @Override
2192     public void asyncRequest(final Write r) {
2193         asyncRequest(r, null);
2194     }
2195
2196     @Override
2197     public void asyncRequest(final DelayedWrite r) {
2198         asyncRequest(r, null);
2199     }
2200
2201     @Override
2202     public void asyncRequest(final WriteOnly request, final Consumer<DatabaseException> callback) {
2203
2204         scheduleRequest(request, callback, null);
2205
2206     }
2207
2208     @Override
2209     public void asyncRequest(final WriteOnly request) {
2210
2211         asyncRequest(request, null);
2212
2213     }
2214
2215     @Override
2216     public <T> void async(ReadInterface<T> r, AsyncProcedure<T> procedure) {
2217         r.request(this, procedure);
2218     }
2219
2220     @Override
2221     public <T> void async(ReadInterface<T> r, Procedure<T> procedure) {
2222         r.request(this, procedure);
2223     }
2224
2225     @Override
2226     public <T> void async(ReadInterface<T> r, SyncProcedure<T> procedure) {
2227         r.request(this, procedure);
2228     }
2229
2230     @Override
2231     public <T> void async(ReadInterface<T> r, AsyncListener<T> procedure) {
2232         r.request(this, procedure);
2233     }
2234
2235     @Override
2236     public <T> void async(ReadInterface<T> r, Listener<T> procedure) {
2237         r.request(this, procedure);
2238     }
2239
2240     @Override
2241     public <T> void async(ReadInterface<T> r, SyncListener<T> procedure) {
2242         r.request(this, procedure);
2243     }
2244
2245     @Override
2246     public <T> T sync(ReadInterface<T> r) throws DatabaseException {
2247         return r.request(this);
2248     }
2249
2250     @Override
2251     public <T> T sync(WriteInterface<T> r) throws DatabaseException {
2252         return r.request(this);
2253     }
2254
2255     @Override
2256     public <T> void async(WriteInterface<T> r, Procedure<T> procedure) {
2257         r.request(this, procedure);
2258     }
2259
2260     @Override
2261     public <T> void async(WriteInterface<T> r) {
2262         r.request(this, new ProcedureAdapter<T>());
2263     }
2264
2265     //@Override
2266     public void incAsync() {
2267         state.incAsync();
2268     }
2269
2270     //@Override
2271     public void decAsync() {
2272         state.decAsync();
2273     }
2274
2275     public long getCluster(ResourceImpl resource) {
2276         ClusterI cluster = clusterTable.getClusterByResourceKey(resource.id);
2277         return cluster.getClusterId();
2278     }
2279
2280     public long getCluster(int id) {
2281         if (clusterTable == null)
2282             System.out.println("SessionImplSocket.getCluster() clusterTable == null !!!!! how come");
2283         return clusterTable.getClusterIdByResourceKeyNoThrow(id);
2284     }
2285
2286     public ResourceImpl getResource(int id) {
2287         return new ResourceImpl(resourceSupport, id);
2288     }
2289
2290     public ResourceImpl getResource(int resourceIndex, long clusterId) {
2291         assert (!ClusterTraitsBase.isIllegalResourceIndex(resourceIndex));
2292         ClusterI proxy = clusterTable.getClusterByClusterId(clusterId);
2293         int key = proxy.getClusterKey();
2294         int resourceKey = ClusterTraitsBase.createResourceKeyNoThrow(key, resourceIndex);
2295         return new ResourceImpl(resourceSupport, resourceKey);
2296     }
2297
2298     public ResourceImpl getResource2(int id) {
2299         assert (id != 0);
2300         return new ResourceImpl(resourceSupport, id);
2301     }
2302
2303     final public int getId(ResourceImpl impl) {
2304         return impl.id;
2305     }
2306
2307     public static final Charset UTF8 = Charset.forName("utf-8");
2308
2309
2310
2311
2312     static boolean areVirtualStatementsLoaded(final VirtualGraphServerSupportImpl support, int subject) {
2313         for(TransientGraph g : support.providers) {
2314             if(g.isPending(subject)) return false;
2315         }
2316         return true;
2317     }
2318
2319     static boolean areVirtualStatementsLoaded(final VirtualGraphServerSupportImpl support, int subject, int predicate) {
2320         for(TransientGraph g : support.providers) {
2321             if(g.isPending(subject, predicate)) return false;
2322         }
2323         return true;
2324     }
2325
2326     static void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, final Consumer<ReadGraphImpl> runnable) {
2327
2328         Consumer<ReadGraphImpl> composite = new Consumer<ReadGraphImpl>() {
2329
2330             AtomicInteger ready = new AtomicInteger(support.providers.size() + 1);
2331
2332             @Override
2333             public void accept(ReadGraphImpl graph) {
2334                 if(ready.decrementAndGet() == 0) {
2335                     runnable.accept(graph);
2336                 }
2337             }
2338
2339         };
2340
2341         for(TransientGraph g : support.providers) {
2342             if(g.isPending(subject)) {
2343                 try {
2344                     g.load(graph, subject, composite);
2345                 } catch (DatabaseException e) {
2346                     e.printStackTrace();
2347                 }
2348             } else {
2349                 composite.accept(graph);
2350             }
2351         }
2352
2353         composite.accept(graph);
2354
2355     }
2356
2357     static  void loadVirtualStatements(final VirtualGraphServerSupportImpl support, ReadGraphImpl graph, int subject, int predicate, final Consumer<ReadGraphImpl> runnable) {
2358
2359         Consumer<ReadGraphImpl> composite = new Consumer<ReadGraphImpl>() {
2360
2361             AtomicInteger ready = new AtomicInteger(support.providers.size() + 1);
2362
2363             @Override
2364             public void accept(ReadGraphImpl graph) {
2365                 if(ready.decrementAndGet() == 0) {
2366                     runnable.accept(graph);
2367                 }
2368             }
2369
2370         };
2371
2372         for(TransientGraph g : support.providers) {
2373             if(g.isPending(subject, predicate)) {
2374                 try {
2375                     g.load(graph, subject, predicate, composite);
2376                 } catch (DatabaseException e) {
2377                     e.printStackTrace();
2378                 }
2379             } else {
2380                 composite.accept(graph);
2381             }
2382         }
2383
2384         composite.accept(graph);
2385
2386     }
2387
2388 //    void dumpHeap() {
2389 //
2390 //        try {
2391 //            ManagementFactory.newPlatformMXBeanProxy(ManagementFactory.getPlatformMBeanServer(),
2392 //                    "com.sun.management:type=HotSpotDiagnostic", HotSpotDiagnosticMXBean.class).dumpHeap(
2393 //                    "d:/heap" + flushCounter + ".txt", true);
2394 //        } catch (IOException e) {
2395 //            e.printStackTrace();
2396 //        }
2397 //
2398 //    }
2399
2400     void fireReactionsToSynchronize(ChangeSet cs) {
2401
2402         // Do not fire empty events
2403         if (cs.isEmpty())
2404             return;
2405
2406         ReadGraphImpl g = ReadGraphImpl.create(getQueryProvider2());
2407 //        g.state.barrier.inc();
2408
2409         try {
2410
2411             if (!cs.isEmpty()) {
2412                 ChangeEvent e2 = new ChangeEvent(g.getSession(), g, null, cs);
2413                 for (ChangeListener l : changeListeners2) {
2414                     try {
2415                         l.graphChanged(e2);
2416                     } catch (Exception ex) {
2417                         ex.printStackTrace();
2418                     }
2419                 }
2420             }
2421
2422         } finally {
2423
2424 //            g.state.barrier.dec();
2425 //            g.waitAsync(null);
2426
2427         }
2428
2429     }
2430
2431     void fireReactionsToCommit(ReadGraphImpl graph, ChangeSet cs2) {
2432         try {
2433
2434             // Do not fire empty events
2435             if (cs2.isEmpty())
2436                 return;
2437
2438 //            graph.restart();
2439 //            graph.state.barrier.inc();
2440
2441             try {
2442
2443                 if (!cs2.isEmpty()) {
2444
2445                     ChangeEvent e2 = new ChangeEvent(graph.getSession(), graph, null, cs2);
2446                     for (ChangeListener l : changeListeners2) {
2447                         try {
2448                             l.graphChanged(e2);
2449 //                            System.out.println("changelistener " + l);
2450                         } catch (Exception ex) {
2451                             ex.printStackTrace();
2452                         }
2453                     }
2454
2455                 }
2456
2457             } finally {
2458
2459 //                graph.state.barrier.dec();
2460 //                graph.waitAsync(null);
2461
2462             }
2463         } catch (Throwable t) {
2464             t.printStackTrace();
2465
2466         }
2467     }
2468     void fireMetadataListeners(WriteGraphImpl graph, ChangeSet cs2) {
2469
2470         try {
2471
2472             // Do not fire empty events
2473             if (cs2.isEmpty())
2474                 return;
2475
2476             // Do not fire on virtual requests
2477             if(graph.getProvider() != null)
2478                 return;
2479
2480             WriteGraphImpl reactionGraph = WriteGraphImpl.create(graph.processor, writeSupport, null);
2481
2482             try {
2483
2484                 ChangeEvent e2 = new ChangeEvent(graph.getSession(), reactionGraph, graph, cs2);
2485                 for (ChangeListener l : metadataListeners) {
2486                     try {
2487                         l.graphChanged(e2);
2488                     } catch (Throwable ex) {
2489                         LOGGER.error("Could not invoke listener {} with event {}", l, e2, ex);
2490                     }
2491                 }
2492
2493             } finally {
2494
2495             }
2496
2497         } catch (Throwable t) {
2498             LOGGER.error("Could not fire metadata listeners {} {}", graph, cs2, t);
2499         }
2500
2501     }
2502
2503
2504     /*
2505      * (non-Javadoc)
2506      *
2507      * @see org.simantics.db.ServiceLocator#getService(java.lang.Class)
2508      */
2509     @Override
2510     public <T> T getService(Class<T> api) {
2511         T t = peekService(api);
2512         if (t == null) {
2513             if (state.isClosed())
2514                 throw new ServiceNotFoundException(this, api, "Session has been shut down");
2515             throw new ServiceNotFoundException(this, api);
2516         }
2517         return t;
2518     }
2519
2520     /**
2521      * @return
2522      */
2523     protected abstract ServerInformation getCachedServerInformation();
2524
2525         private Class<?> serviceKey1 = null;
2526         private Class<?> serviceKey2 = null;
2527         private Object service1 = null;
2528         private Object service2 = null;
2529
2530     /*
2531      * (non-Javadoc)
2532      *
2533      * @see
2534      * fi.vtt.simantics.db.connection.ServiceLocator#peekService(java.lang.Class)
2535      */
2536     @SuppressWarnings("unchecked")
2537     @Override
2538     public synchronized <T> T peekService(Class<T> api) {
2539
2540                 if(serviceKey1 == api) {
2541                         return (T)service1;
2542                 } else if (serviceKey2 == api) {
2543                         // Promote this key
2544                         Object result = service2;
2545                         service2 = service1;
2546                         serviceKey2 = serviceKey1;
2547                         service1 = result;
2548                         serviceKey1 = api;
2549                         return (T)result;
2550                 }
2551
2552         if (Layer0.class == api)
2553                 return (T) L0;
2554         if (ServerInformation.class == api)
2555             return (T) getCachedServerInformation();
2556         else if (WriteGraphImpl.class == api)
2557             return (T) writeState.getGraph();
2558         else if (ClusterBuilder.class == api)
2559             return (T)new ClusterBuilderImpl(this, (WriteOnlySupport)writeState.getGraph().writeSupport);
2560         else if (ClusterBuilderFactory.class == api)
2561             return (T)new ClusterBuilderFactoryImpl(this);
2562
2563                 service2 = service1;
2564                 serviceKey2 = serviceKey1;
2565
2566         service1 = serviceLocator.peekService(api);
2567         serviceKey1 = api;
2568
2569         return (T)service1;
2570
2571     }
2572
2573     /*
2574      * (non-Javadoc)
2575      *
2576      * @see
2577      * fi.vtt.simantics.db.connection.ServiceLocator#hasService(java.lang.Class)
2578      */
2579     @Override
2580     public boolean hasService(Class<?> api) {
2581         return serviceLocator.hasService(api);
2582     }
2583
2584     /**
2585      * @param api the api that must be implemented by the specified service
2586      * @param service the service implementation
2587      */
2588     @Override
2589     public <T> void registerService(Class<T> api, T service) {
2590         if(Layer0.class == api) {
2591                 L0 = (Layer0)service;
2592                 return;
2593         }
2594         serviceLocator.registerService(api, service);
2595         if (TransactionPolicySupport.class == api) {
2596             transactionPolicy = (TransactionPolicySupport)service;
2597             state.resetTransactionPolicy();
2598         }
2599         if (api == serviceKey1)
2600                 service1 = service;
2601         else if (api == serviceKey2)
2602                 service2 = service;
2603     }
2604
2605     // ----------------
2606     // SessionMonitor
2607     // ----------------
2608
2609     void fireSessionVariableChange(String variable) {
2610         for (MonitorHandler h : monitorHandlers) {
2611             MonitorContext ctx = monitorContexts.getLeft(h);
2612             assert ctx != null;
2613             // SafeRunner functionality repeated here to avoid dependency.
2614             try {
2615                 h.valuesChanged(ctx);
2616             } catch (Exception e) {
2617                 Logger.defaultLogError("monitor handler notification produced the following exception", e);
2618             } catch (LinkageError e) {
2619                 Logger.defaultLogError("monitor handler notification produced a linkage error", e);
2620             }
2621         }
2622     }
2623
2624
2625     class ResourceSerializerImpl implements ResourceSerializer {
2626
2627         public long createRandomAccessId(int id) throws DatabaseException {
2628             if(id < 0)
2629                 return id;
2630             int index = ClusterTraitsBase.getResourceIndexFromResourceKey(id);
2631             long cluster = getCluster(id);
2632             if (0 == cluster)
2633                 return 0; // Better to return 0 then invalid id.
2634             long result = ClusterTraitsBase.createResourceId(cluster, index);
2635             return result;
2636         }
2637
2638         @Override
2639         public long getRandomAccessId(Resource resource) throws DatabaseException {
2640
2641             ResourceImpl resourceImpl = (ResourceImpl) resource;
2642             return createRandomAccessId(resourceImpl.id);
2643
2644         }
2645
2646         @Override
2647         public int getTransientId(Resource resource) throws DatabaseException {
2648
2649             ResourceImpl resourceImpl = (ResourceImpl) resource;
2650             return resourceImpl.id;
2651
2652         }
2653
2654         @Override
2655         public String createRandomAccessId(Resource resource)
2656         throws InvalidResourceReferenceException {
2657
2658             if(resource == null) throw new IllegalArgumentException();
2659
2660             ResourceImpl resourceImpl = (ResourceImpl) resource;
2661
2662             if (resourceImpl.id < 0) return String.valueOf(resourceImpl.id) + "_0";
2663
2664             int r;