Dispose ClientChangesImpl ChangeSets to minimize memory footprint
[simantics/platform.git] / bundles / org.simantics.db.procore / src / fi / vtt / simantics / procore / internal / State.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.util.concurrent.TimeUnit;
15 import java.util.concurrent.locks.Condition;
16 import java.util.concurrent.locks.Lock;
17 import java.util.concurrent.locks.ReentrantLock;
18
19 import org.simantics.db.ChangeSet;
20 import org.simantics.db.Disposable;
21 import org.simantics.db.Operation;
22 import org.simantics.db.VirtualGraph;
23 import org.simantics.db.common.utils.Logger;
24 import org.simantics.db.exception.DatabaseException;
25 import org.simantics.db.exception.InternalException;
26 import org.simantics.db.exception.RuntimeDatabaseException;
27 import org.simantics.db.impl.graph.WriteGraphImpl;
28 import org.simantics.db.impl.query.QueryProcessor;
29 import org.simantics.db.request.WriteOnly;
30 import org.simantics.db.request.WriteTraits;
31 import org.simantics.db.service.LifecycleSupport.LifecycleState;
32 import org.simantics.db.service.TransactionPolicySupport;
33
34 import fi.vtt.simantics.procore.internal.SessionImplSocket.WriteOnlySupport;
35
36 class State {
37     static final int          Closed           = 1;
38     static final int          Closing          = 2;
39     static final int          WriteTransaction = 4;
40     static final int          ReadTransaction  = 8;
41     static final int          All              = 15;
42     private volatile int state = 0; // Must be volatile so we don't have to synchronize getState.
43     private SessionImplSocket session          = null;
44     private GraphSession      graphSession     = null;
45     private QueryProcessor    queryProvider    = null;
46     private Lock              lock             = new ReentrantLock();
47     private Condition condition = lock.newCondition(); // Tried this with tree conditions but it wasn't worth the effort.
48     private volatile int readCount = 0; // Must be volatile so we don't have to synchronize getReadCount.
49     private volatile int writeCount = 0; // Must be volatile so we don't have to synchronize getWriteCount.
50     private Thread            writeOwner       = null;
51     private volatile int asyncCount = 1; // Must be volatile so we don't have to synchronize getAsyncCount.
52     private TransactionToken  transactionToken = null;
53     void setCombine(boolean a) {
54         if (null != transactionToken)
55             transactionToken.setCombine(a);
56     }
57     void setGraphSession(SessionImplSocket session, GraphSession graphSession, QueryProcessor queryProvider, ClusterTable clusterTable) {
58         this.session = session;
59         this.graphSession = graphSession;
60         this.queryProvider = queryProvider;
61         resetTransactionPolicy();
62     }
63     void resetTransactionPolicy() {
64         TransactionPolicySupport tps = session.getService(TransactionPolicySupport.class);
65         this.transactionToken = new TransactionToken(tps, session, graphSession, this.transactionToken);
66     }
67     int getAsyncCount() {
68         return asyncCount;
69     }
70     int getReadCount() {
71         return readCount;
72     }
73     int getWriteCount() {
74         return writeCount;
75     }
76     boolean isWriteTransaction() {
77         return 0 != (state & WriteTransaction);
78     }
79     boolean isAlive() {
80         return !isClosed() && !isClosing();
81     }
82     boolean isClosed() {
83         return 0 != (state & Closed);
84     }
85     boolean isClosing() {
86         return 0 != (state & Closing);
87     }
88     boolean isWriting() {
89         return 0 != (state & WriteTransaction);
90     }
91     void close() {
92         boolean acquired = false;
93         try {
94             acquired = lock.tryLock(1, TimeUnit.MINUTES);
95         } catch (InterruptedException e1) {
96         }
97         try {
98             if (null != transactionToken)
99                 transactionToken.close();
100             graphSession.stop();
101         } catch (DatabaseException e) {
102         } finally {
103             state = Closed;
104             session.lifecycleSupport.fireListeners(LifecycleState.CLOSED);
105             if (acquired) {
106                 condition.signalAll();
107                 lock.unlock();
108             }
109         }
110     }
111     void incAsync()
112     {
113         lock.lock();
114         try {
115             boolean closed = isClosed();
116             boolean closing = isClosing();
117             if (closed || closing)
118                 throw new RuntimeDatabaseException(session + ": closed=" + closed + ", closing=" + closing);
119             ++asyncCount;
120         } finally {
121             lock.unlock();
122         }
123     }
124     void decAsync()
125     {
126         lock.lock();
127         try {
128             if (asyncCount < 1)
129                 throw new RuntimeDatabaseException(session + ": asyncCount=" + asyncCount);
130             if (0 == --asyncCount) {
131                 condition.signal();
132             }
133         } finally {
134             lock.unlock();
135         }
136     }
137     enum WaitStatus { IsClosed, IsClosing, CanBeClosed, Timeout, Deadlock }
138     WaitStatus waitClosing(long timeout) {
139         lock.lock();
140         try {
141             if (isClosed())
142                 return WaitStatus.IsClosed;
143             else if (isClosing())
144                 return WaitStatus.IsClosing;
145             state |= Closing;
146             try {
147                 if (timeout < 0) { // Infinite timeout
148                     for (int i=0; i<100; i++)
149                     while (asyncCount > 0  || readCount > 0 ||
150                            (null != writeOwner && Thread.currentThread() != writeOwner)) {
151                         condition.signalAll();
152                         condition.awaitNanos(10000000); // nanos
153                     }
154                     while (asyncCount > 0  || readCount > 0 ||
155                             (null != writeOwner && Thread.currentThread() != writeOwner)) {
156                          condition.signalAll();
157                          condition.await();
158                      }
159                 } else if (timeout > 0) {
160                     boolean interrupted = false;
161                     while (!interrupted &&
162                            (asyncCount > 0  || readCount > 0 ||
163                             (null != writeOwner && Thread.currentThread() != writeOwner)))
164                         interrupted = !condition.await(timeout, TimeUnit.MILLISECONDS);
165                 }
166                 transactionToken.closing();
167             } catch (Exception e) {
168                 e.printStackTrace();
169             }
170             state &= All ^ Closing;
171             if (readCount == 0 && writeCount == 0 && asyncCount == 0)
172                 return WaitStatus.CanBeClosed;
173             else if (null != writeOwner && Thread.currentThread() == writeOwner)
174                 return WaitStatus.Deadlock;
175             return WaitStatus.Timeout;
176         } finally {
177             lock.unlock();
178         }
179     }
180     void startReadTransaction(int thread) throws DatabaseException {
181         lock.lock();
182         try {
183             assert(readCount == 0);
184             transactionToken.startReadTransaction(thread);
185             state |= ReadTransaction;
186             session.fireStartReadTransaction();
187             ++readCount;
188         } catch (Throwable e) {
189             throw new DatabaseException("Failed to start read transaction.", e);
190         } finally {
191             lock.unlock();
192         }
193     }
194     void stopReadTransaction() throws DatabaseException {
195         lock.lock();
196         try {
197             assert (!queryProvider.hasScheduledUpdates());
198             assert (readCount == 1);
199             session.writeSupport.gc();
200             transactionToken.stopReadTransaction();
201             readCount = 0;
202             state &= All ^ ReadTransaction;
203             condition.signal();
204         } finally {
205             lock.unlock();
206         }
207     }
208
209     void startWriteTransaction(int thread)
210     throws DatabaseException {
211         lock.lock();
212         try {
213 //            System.out.println("stawt");
214             boolean closed = isClosed();
215             boolean closing = isClosing();
216             int ac = asyncCount;
217             int wc = writeCount;
218             if (closed || (closing && ac < 1) || wc < 0)
219                 throw new DatabaseException(session + ": closed=" + closed + ", closing=" + closing + ", asyncCount=" + asyncCount + ", writeCount=" + writeCount);
220             if (writeCount == 0) {
221                 transactionToken.startWriteTransaction(thread);
222                 state |= WriteTransaction;
223                 writeOwner = Thread.currentThread();
224                 session.fireStartWriteTransaction();
225             }
226             ++writeCount;
227             assert(writeCount == 1);
228         } catch (Throwable e) {
229             throw new DatabaseException(e);
230         } finally {
231             lock.unlock();
232         }
233     }
234
235     void writeTransactionEnded() {
236         session.defaultClusterSet = null;
237         session.fireFinishWriteTransaction();
238     }
239
240     void stopWriteTransaction(ClusterStream clusterStream) {
241         if(!isAlive()) return;
242         lock.lock();
243         try {
244 //            System.out.println("stowt");
245             if (writeCount < 1)
246                 throw new IllegalStateException(session + ": writeCount=" + writeCount);
247             else if (1 == writeCount) {
248                 boolean empty = session.clusterStream.reallyFlush();
249                 if (!empty) {
250                     String msg = "We have modified graph (on server) without accept/cancel acknowledgment.\n"
251                             + "This is probably a serious error. Automatic cancel done as default recovery.";
252                     Logger.defaultLogInfo(msg);
253                     ClientChangesImpl cs = new ClientChangesImpl(session);
254                     SynchronizeContext context = new SynchronizeContext(session, cs, 1);
255                     transactionToken.cancelBegin(session.writeSupport, context, clusterStream);
256                 }
257                 transactionToken.stopWriteTransaction();
258                 state &= All ^ WriteTransaction;
259                 writeOwner = null;
260                 condition.signal();
261                 writeTransactionEnded();
262             }
263             --writeCount;
264         } catch (Throwable e) {
265             e.printStackTrace();
266             this.close(); // Everything is lost anyway.
267             throw new IllegalStateException(e);
268         } finally {
269             lock.unlock();
270         }
271     }
272
273     void cancelWriteTransaction(WriteGraphImpl graph) {
274         lock.lock();
275         try {
276             if (writeCount < 1)
277                 throw new IllegalStateException(session + ": writeCount=" + writeCount);
278             VirtualGraph vg = graph.getProvider();
279             if (null == vg) { // Skip if virtual graph.
280                 boolean empty = session.clusterStream.reallyFlush();
281                 if (!empty) { // Something to cancel.
282                     // Send all local changes to server so it can calculate correct reverse change set.
283                     ClientChangesImpl cs = new ClientChangesImpl(session);
284                     SynchronizeContext context = new SynchronizeContext(session, cs, 1);
285                     // This will send cancel and fetch reverse change set from server.
286                     transactionToken.cancelBegin(session.writeSupport, context, session.clusterStream);
287                     try {
288                         final boolean undo = false;
289                         if (!context.isOk(undo)) // this is a blocking operation
290                             throw new InternalException("Cancel failed. This should never happen. Contact application support.");
291 //                        System.out.println("session cs: " + session.clientChanges);
292 //                        System.out.println("reverse cs: " + cs);
293                         queryProvider.performDirtyUpdates(graph);
294                         queryProvider.performScheduledUpdates(graph);
295                     } catch (DatabaseException e) {
296                         Logger.defaultLogError(e);
297                     }
298                     // This will send and accept the reverse change set.
299                     transactionToken.cancelEnd(session.writeSupport, null, session.clusterStream);
300                 }
301             }
302             session.writeSupport.clearMetadata();
303             if (--writeCount == 0) {
304                 transactionToken.stopWriteTransaction();
305                 state &= All ^ WriteTransaction;
306                 writeOwner = null;
307                 condition.signal();
308                 writeTransactionEnded();
309             }
310         } catch (Throwable e) {
311             e.printStackTrace();
312             this.close(); // Everything is lost anyway.
313             throw new IllegalStateException(e);
314         } finally {
315             lock.unlock();
316         }
317     }
318
319     void commitWriteTransaction(WriteGraphImpl graph, ClusterStream clusterStream, ChangeSet cs, WriteTraits request, Operation op) {
320         assert(request != null);
321         graph.commitAccessorChanges(request);
322         boolean writeOnly = request instanceof WriteOnly;
323         lock.lock();
324         try {
325             VirtualGraph vg = graph.getProvider();
326             if (writeCount == 1) {
327
328                 if (vg != null && clusterStream.isDirty())
329                     new Exception("Internal error: virtual transaction committed changes into core (" + request + ")").printStackTrace();
330
331                 // This is needed even when the write targets a virtual graph -
332                 // deny can always remove a persistent statement. 
333                 clusterStream.reallyFlush();
334
335                 Disposable.safeDispose(session.clientChanges);
336                 session.clientChanges = new ClientChangesImpl(session);
337
338
339 //                start = System.nanoTime();
340                 queryProvider.performScheduledUpdates(graph);
341 //                duration = System.nanoTime() - start;
342 //                System.out.println("performScheduledUpdates " + 1e-9*duration + "s. ");
343
344                 // Process updates as long as pending primitives exist
345                 while (session.dirtyPrimitives) {
346                     session.dirtyPrimitives = false;
347                     queryProvider.performDirtyUpdates(graph);
348                     queryProvider.performScheduledUpdates(graph);
349                 }
350
351                 if (!writeOnly) // TODO: fix me! FIX ME! Please fix me! Please!
352                     session.fireReactionsToCommit(graph, cs); // Does not throw exception!
353
354                 writeTransactionEnded();
355                 session.writeSupport.gc();
356             } else // Only the last writer can commit.
357                 throw new IllegalStateException(session + ": writeCount=" + writeCount);
358 //            long start = System.nanoTime();
359 //            System.err.println("commit");
360
361 //            start = System.nanoTime();
362             if (null == vg)
363                 transactionToken.commitWriteTransaction(graph.writeSupport, request, clusterStream, op);
364             else {
365                 // There should not be meta data because this is transient graph.
366                 // This is an insurance that this is the case for now.
367                 int n = session.writeSupport.clearMetadata();
368                 if (n > 0)
369                     if (SessionImplSocket.DEBUG)
370                         System.out.println("DEBUG: Transient graph has accumulated metadata. Size=" + n);
371             }
372             transactionToken.stopWriteTransaction();
373 //            long duration = System.nanoTime() - start;
374 //            System.out.println("transactionToken.commitWriteTransaction " + 1e-9*duration + "s. ");
375
376
377 //            long duration = System.nanoTime() - start;
378 //            System.err.println("commit2 " + 1e-9*duration);
379             state &= All ^ WriteTransaction;
380             writeCount = 0;
381             writeOwner = null;
382             condition.signal();
383         } catch (Throwable e) {
384             Logger.defaultLogError(e);
385             this.close(); // Everything is lost anyway.
386             throw new IllegalStateException(e);
387         } finally {
388             lock.unlock();
389         }
390     }
391
392     void commitAndContinue(WriteGraphImpl graph, ClusterStream clusterStream, WriteTraits request) {
393         VirtualGraph vg = graph.getProvider();
394         if (null != vg)
395             return;
396         lock.lock();
397         try {
398             clusterStream.reallyFlush();
399             transactionToken.commitWriteTransaction(graph.writeSupport, request, clusterStream, null);
400             // See below for explanation, the same reasoning should apply here too.
401             if (graph.writeSupport instanceof WriteOnlySupport)
402                 graph.writeSupport.flushCluster();
403         } catch (Throwable e) {
404             Logger.defaultLogError(e);
405             this.close(); // Everything is lost anyway.
406             throw new IllegalStateException(e);
407         } finally {
408             lock.unlock();
409         }
410     }
411     void commitAndContinue2(WriteGraphImpl graph, ClusterStream clusterStream, WriteTraits request) {
412         VirtualGraph vg = graph.getProvider();
413         if (null != vg)
414             return;
415         lock.lock();
416         try {
417             transactionToken.commitWriteTransaction(graph.writeSupport, request, clusterStream, null);
418             transactionToken.setCombine(true);
419             // This was added because test Issue3199Test2 failed because
420             // write only cluster does not need cluster when allocating resource index
421             // and the first claim after commitAndContinue2 caused cluster to be loaded
422             // from server and of course it's resource index was off by one.
423             if (graph.writeSupport instanceof WriteOnlySupport)
424                 graph.writeSupport.flushCluster();
425         } catch (Throwable e) {
426             Logger.defaultLogError(e);
427             this.close(); // Everything is lost anyway.
428             throw new IllegalStateException(e);
429         } finally {
430             lock.unlock();
431         }
432     }
433     Operation getLastOperation() {
434         return transactionToken.getLastOperation();
435     }
436     long getHeadRevisionId() {
437         return transactionToken.getHeadRevisionId();
438     }
439 }