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