tests/org.simantics.db.tests/src/org/simantics/db/tests/client/TransactionTest1.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 org.simantics.db.tests.client;
  13
  14 import java.util.concurrent.Semaphore;
  15 import java.util.concurrent.atomic.AtomicBoolean;
  16 import java.util.concurrent.atomic.AtomicInteger;
  17
  18 import org.junit.Test;
  19 import org.simantics.db.AsyncReadGraph;
  20 import org.simantics.db.ReadGraph;
  21 import org.simantics.db.Session;
  22 import org.simantics.db.WriteGraph;
  23 import org.simantics.db.common.TransactionPolicyKeep;
  24 import org.simantics.db.common.TransactionPolicyRelease;
  25 import org.simantics.db.common.request.ReadRequest;
  26 import org.simantics.db.common.request.WriteRequest;
  27 import org.simantics.db.exception.DatabaseException;
  28 import org.simantics.db.procedure.AsyncProcedure;
  29 import org.simantics.db.request.Read;
  30 import org.simantics.db.service.LifecycleSupport;
  31 import org.simantics.db.service.TransactionPolicySupport;
  32 import org.simantics.db.testing.base.ExistingDatabaseTest;
  33 import org.simantics.db.testing.common.Tests;
  34 import org.simantics.db.testing.impl.Configuration;
  35
  36 public class TransactionTest1 extends ExistingDatabaseTest {
  37         @Test
  38     public void testTransaction1() throws DatabaseException {
  39         empty1();
  40         empty2();
  41 //        empty3();
  42     }
  43         void empty1() throws DatabaseException {
  44             TransactionClient1 client = new TransactionClient1(this);
  45         try {
  46             client.setPolicyRelease();
  47             client.emptySyncReadLoop();
  48             client.setPolicyKeep();
  49             client.emptySyncReadLoop();
  50         } finally {
  51             if (null != client)
  52                 client.close();
  53         }
  54         }
  55         void empty2() throws DatabaseException {
  56         TransactionClient1 client = new TransactionClient1(this);
  57         TransactionClient2 client2 = new TransactionClient2(this);
  58         try {
  59             client2.tryToKeepReadTransaction();
  60             client.setPolicyRelease();
  61             client.emptySyncReadLoop();
  62             client.setPolicyKeep();
  63             client.emptySyncReadLoop();
  64         } finally {
  65             client.close();
  66             client2.close();
  67         }
  68     }
  69         public void empty3() throws DatabaseException {
  70         TransactionClient1 client = new TransactionClient1(this);
  71         TransactionClient2 client2 = new TransactionClient2(this);
  72         try {
  73             client2.tryToKeepWriteTransaction();
  74             client.setPolicyRelease();
  75             client.emptySyncReadLoop();
  76             client.setPolicyKeep();
  77             client.emptySyncReadLoop();
  78         } finally {
  79             client.close();
  80             client2.close();
  81         }
  82     }
  83 }
  84 class TranasctionSession {
  85     protected static final boolean DEBUG = Configuration.get().debug;
  86     protected Session session;
  87     TranasctionSession() throws DatabaseException {
  88         session = Tests.getTestHandler().getSession();
  89     }
  90     public void close() throws DatabaseException {
  91         try {
  92             if (null != session) {
  93                 LifecycleSupport ls = session.getService(LifecycleSupport.class);
  94                 ls.close();
  95             }
  96         } finally {
  97             session = null;
  98         }
  99     }
 100 }
 101 class TransctionCommon extends TranasctionSession {
 102     protected final ExistingDatabaseTest testCommon;
 103     protected final int REQUEST_COUNT = 50;
 104     private AtomicInteger counter = new AtomicInteger(0);
 105     TransctionCommon(ExistingDatabaseTest testCommon)
 106     throws DatabaseException {
 107         this.testCommon = testCommon;
 108     }
 109     public void emptySyncReadLoop()
 110     throws DatabaseException {
 111         counter.set(0);
 112         long start = System.nanoTime();
 113         for(int i = 0; i < REQUEST_COUNT; ++i) {
 114             session.syncRequest(new ReadRequest() {
 115                 @Override
 116                 public void run(ReadGraph graph) throws DatabaseException {
 117                     int value = counter.incrementAndGet();
 118                     if (DEBUG)
 119                         System.out.println("DEBUG: Sync read count=" + value);
 120                 }
 121             });
 122         }
 123         long end = System.nanoTime();
 124         double time = (double)(end-start) * (double)(1e-9);
 125         if (REQUEST_COUNT != counter.get())
 126             throw new DatabaseException("Transaction count does not match. Transaction counter was " + counter);
 127         double speed = counter.get() / time;
 128         String t = "Transaction speed was " + speed + " empty synchronous read transactions per second.";
 129         if (DEBUG)
 130                 System.out.println(t);
 131     }
 132     void setPolicyKeep() {
 133         if (DEBUG)
 134             System.out.println("Transaction policy is keep.");
 135         session.registerService(TransactionPolicySupport.class, new TransactionPolicyKeep());
 136     }
 137     void setPolicyRelease() {
 138         if (DEBUG)
 139             System.out.println("Transaction policy is release.");
 140         session.registerService(TransactionPolicySupport.class, new TransactionPolicyRelease());
 141     }
 142 }
 143
 144 class TransactionClient1 extends TransctionCommon {
 145   TransactionClient1(ExistingDatabaseTest testCommon)
 146   throws DatabaseException {
 147       super(testCommon);
 148   }
 149 }
 150
 151 class TransactionClient2 extends TransctionCommon
 152 implements TransactionPolicySupport {
 153     Thread tryToKeepWrite;
 154     int in;
 155     int out;
 156     AtomicBoolean breakTransaction = new AtomicBoolean(false);
 157     Semaphore relinquish = new Semaphore(0);
 158     boolean goOn;
 159     TransactionClient2(ExistingDatabaseTest testCommon)
 160     throws DatabaseException {
 161         super(testCommon);
 162         init();
 163     }
 164     private void init() {
 165         synchronized (this) {
 166             in = 0;
 167             out = 0;
 168             breakTransaction.set(false);
 169             goOn = true;
 170         }
 171     }
 172     public void tryToKeepReadTransaction()
 173     throws DatabaseException {
 174         init();
 175         setPolicyKeep();
 176         session.registerService(TransactionPolicySupport.class, this);
 177         startReadTransaction();
 178     }
 179     public void tryToKeepWriteTransaction()
 180     throws DatabaseException {
 181         init();
 182         setPolicyKeep();
 183         session.registerService(TransactionPolicySupport.class, this);
 184 //        startWriteTransaction();
 185         this.tryToKeepWrite = new Thread(new TryToKeepWrite(), "TryToKeepWrite");
 186         this.tryToKeepWrite.start();
 187     }
 188     @Override
 189     public boolean holdOnToTransactionAfterRead() {
 190         return true;
 191     }
 192     @Override
 193     public boolean holdOnToTransactionAfterCancel() {
 194         return true;
 195     }
 196     @Override
 197     public boolean holdOnToTransactionAfterCommit() {
 198         return true;
 199     }
 200     @Override
 201     public void onRelinquish() {
 202         freeTransactionSync();
 203     }
 204
 205     @Override
 206     public void onRelinquishDone() {
 207         relinquish.release();
 208     }
 209
 210     @Override
 211     public void onRelinquishError() {
 212         relinquish.release();
 213         init();
 214         try {
 215             super.close();
 216         } catch (DatabaseException e) {
 217             System.out.println("OnRelinquishError: " + e.getMessage() + ".");
 218         }
 219         System.exit(-1);
 220     }
 221
 222     @Override
 223     public void close()
 224     throws DatabaseException {
 225         goOn = false;
 226         freeTransactionSync();
 227         // This HACK gives time for implementation to remove
 228         // the asynchronous call from bookkeeping. At this point
 229         // one asynchronous call might still be active.
 230         try {
 231             session.syncRequest(new WriteRequest() {
 232                 @Override
 233                 public void perform(WriteGraph graph) throws DatabaseException {
 234                 }
 235             });
 236         } catch (DatabaseException e) {
 237             System.out.println("Empty sync write failed: " + e.getMessage());
 238         }
 239         super.close();
 240     }
 241     void freeTransactionSync() {
 242         breakTransaction.set(true);
 243         synchronized (this) {
 244             while (in > out) {
 245                 this.notifyAll();
 246                 try {
 247                     TransactionClient2.this.wait();
 248                 } catch (InterruptedException e) {
 249                     System.out.println("Wait was interruped: " + e);
 250                 }
 251             }
 252         }
 253     }
 254     private void startReadTransaction()
 255     throws DatabaseException {
 256         session.asyncRequest(new Read<Integer>() {
 257             @Override
 258             public Integer perform(ReadGraph graph) throws DatabaseException {
 259                 synchronized (TransactionClient2.this) {
 260                     TransactionClient2.this.transactionIn();
 261                     if (DEBUG)
 262                         System.out.println("DEBUG: Read request begin ok, count= " + in);
 263                     while (TransactionClient2.this.transactionKeep()) {
 264                         try {
 265                             TransactionClient2.this.wait();
 266                         } catch (InterruptedException e) {
 267                             if (DEBUG)
 268                                 System.out.println("DEBUG: read request wait interrupted: " + e.getMessage());
 269                         }
 270                     }
 271                 }
 272                 return in;
 273             }
 274         }, new AsyncProcedure<Integer>() {
 275             @Override
 276             public void execute(AsyncReadGraph graph, Integer result) {
 277                 TransactionClient2.this.transactionOut();
 278                 if (DEBUG)
 279                     System.out.println("DEBUG: Read request end ok, count= " + result);
 280             }
 281             @Override
 282             public void exception(AsyncReadGraph graph, Throwable throwable) {
 283                 TransactionClient2.this.transactionOut();
 284                 if (DEBUG)
 285                     System.out.println("DEBUG: read request end failed: " + throwable.getMessage());
 286             }
 287         });
 288         synchronized (this) {
 289             while (in <= out) {
 290                 try {
 291                     TransactionClient2.this.wait();
 292                 } catch (InterruptedException e) {
 293                     if (DEBUG)
 294                         System.out.println("DEBUG: read request sleep was interruped: " + e);
 295                 }
 296             }
 297         }
 298     }
 299     synchronized boolean transactionKeep() {
 300         return !breakTransaction.get();
 301     }
 302     synchronized int transactionIn() {
 303         ++in;
 304         this.notify();
 305         return in;
 306     }
 307     synchronized int transactionOut() {
 308         ++out;
 309         this.notify();
 310         return out;
 311     }
 312     class TryToKeepWrite implements Runnable {
 313         TryToKeepWrite() {
 314         }
 315         @Override
 316         public void run() {
 317             while (goOn) {
 318                 try {
 319                     session.syncRequest(new WriteRequest() {
 320                         @Override
 321                         public void perform(WriteGraph graph) throws DatabaseException {
 322                             synchronized (TransactionClient2.this) {
 323                                 TransactionClient2.this.transactionIn();
 324                                 if (DEBUG)
 325                                     System.out.println("DEBUG: Write request begin ok, count= " + in);
 326                                 while (TransactionClient2.this.transactionKeep()) {
 327                                     try {
 328                                         TransactionClient2.this.wait();
 329                                     } catch (InterruptedException e) {
 330                                         if (DEBUG)
 331                                             System.out.println("DEBUG: Write request wait interrupted: " + e.getMessage());
 332                                     }
 333                                 }
 334                             }
 335                         }
 336                     });
 337                 } catch (DatabaseException e) {
 338                     System.out.println("Empty sync write failed: " + e.getMessage());
 339                 } finally {
 340                     TransactionClient2.this.transactionOut();
 341                 }
 342                 while (goOn)
 343                     try {
 344                         relinquish.acquire();
 345                         break;
 346                     } catch (InterruptedException e) {
 347                         if (DEBUG)
 348                             System.out.println("DEBUG: Relinquish wait interrupted: " + e.getMessage());
 349                     }
 350                 if (DEBUG)
 351                     System.out.println("DEBUG: Write request end, count= " + out);
 352             }
 353         }
 354     }
 355 }