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