tests/org.simantics.db.tests/src/org/simantics/db/tests/api/request/misc/ComplexReadWriteQueryTest.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.api.request.misc;
  13
  14 import java.util.ArrayList;
  15 import java.util.List;
  16 import java.util.UUID;
  17
  18 import org.junit.Test;
  19 import org.simantics.databoard.Bindings;
  20 import org.simantics.db.ReadGraph;
  21 import org.simantics.db.Resource;
  22 import org.simantics.db.Session;
  23 import org.simantics.db.WriteGraph;
  24 import org.simantics.db.exception.DatabaseException;
  25 import org.simantics.db.procedure.Listener;
  26 import org.simantics.db.request.Read;
  27 import org.simantics.db.testing.base.ExistingDatabaseTest;
  28 import org.simantics.db.testing.common.WriteQuery;
  29 import org.simantics.db.tests.common.Configuration;
  30 import org.simantics.layer0.Layer0;
  31
  32 /**
  33  * This test uses several threads to write the graph information and queries to react information changes.
  34  *
  35  * @author Marko Luukkainen <marko.luukkainen@vtt.fi>
  36  *
  37  */
  38 public class ComplexReadWriteQueryTest extends ExistingDatabaseTest {
  39
  40     public static int NUM_THREADS = Configuration.get().rwQueryThreds;;
  41     public static int MAX_CHANGES = Configuration.get().rwQueryCount;
  42
  43     List<NameUpdateThread> threads = new ArrayList<NameUpdateThread>();
  44     int changes; // not quite correct but works for me
  45     volatile int endCount = 0;
  46     Thread self = null;
  47     boolean ended = false;
  48
  49     private synchronized void incChanges() {
  50 //        System.out.println("changes = " + changes);
  51         ++changes;
  52     }
  53
  54         @Test
  55     public void test() throws Exception {
  56         final Resource rootLib = getSession().getRootLibrary();
  57         for (int i = 0; i < NUM_THREADS; i++) {
  58             getSession().asyncRequest(new WriteQuery(this) {
  59                 @Override
  60                 public void run(WriteGraph g) throws Throwable {
  61                     //System.out.println("creating test resource");
  62                     Layer0 b = Layer0.getInstance(g);
  63                     Resource newResource = g.newResource();
  64                     g.claim(newResource, b.InstanceOf, b.Type);
  65                     g.claim(rootLib, b.ConsistsOf, newResource);
  66                     String name = getRandomString();
  67                     g.claimLiteral(newResource, b.HasName, name, Bindings.STRING);
  68                     NameUpdateThread t = new NameUpdateThread(getSession(),newResource,name);
  69                     threads.add(t);
  70                     t.start();
  71                 }
  72             });
  73         }
  74         if (DEBUG)
  75             System.out.println("Created " + NUM_THREADS + " threads.");
  76         self = Thread.currentThread();
  77         while(!ended) {
  78             int lastChanges = changes;
  79             boolean test = true;
  80             try {
  81                 Thread.sleep(10000);
  82             } catch (InterruptedException e) {
  83                 if (DEBUG)
  84                     System.out.println("Interrupted.");
  85                 test = false;
  86             }
  87             if (test && lastChanges == changes) {
  88                 end();
  89                 throw new Exception("No changes in ten seconds, probably database has died. changes="+changes);
  90             }
  91             try {
  92                 checkException();
  93             } catch (Exception e) {
  94                 end();
  95                 throw e;
  96             }
  97             checkException();
  98         }
  99         if (DEBUG)
 100             System.out.println("ended, changes = " + changes);
 101         checkException();
 102
 103     }
 104
 105     public void end() {
 106         for (NameUpdateThread t : threads) {
 107             t.dispose();
 108         }
 109         try {
 110             Thread.sleep(1000);
 111         } catch (InterruptedException e) {
 112
 113         }
 114         ended = true;
 115     }
 116
 117     static volatile int threadCount=0;
 118     public class NameUpdateThread extends Thread {
 119         private boolean disposed = false;
 120         private Resource resource;
 121         private String currentName;
 122         private Session session;
 123         //private NameQuery query;
 124         private Listener<String> procedure;
 125         private NameRead nameRead;
 126         public NameUpdateThread(Session session, Resource resource, String name) {
 127             super("RWQueryTest-" + ++threadCount);
 128             this.currentName = name;
 129             this.resource = resource;
 130             this.session = session;
 131         }
 132
 133         @Override
 134         public void run() {
 135             procedure = new Listener<String>() {
 136                 @Override
 137                 public void execute(String result) {
 138                     NameUpdateThread.this.interrupt();
 139                 }
 140
 141                 @Override
 142                 public void exception(Throwable t) {
 143                     ComplexReadWriteQueryTest.fail();
 144                 }
 145
 146                 @Override
 147                 public boolean isDisposed() {
 148                     return disposed;
 149                 }
 150
 151             };
 152             while(!disposed) {
 153                 try {
 154                     nameRead = new NameRead(resource);
 155                     //System.out.println("creating query");
 156                     session.asyncRequest(new WriteQuery(ComplexReadWriteQueryTest.this) {
 157                         @Override
 158                         public void run(WriteGraph g) throws Throwable {
 159                             g.syncRequest(nameRead, procedure);
 160                         }
 161                     });
 162                     // sleep until interrupted (if query fails, we still are updating the value)
 163                     Thread.sleep(1000000000);
 164                 } catch (InterruptedException e) {
 165                 }
 166                 if(disposed)
 167                     return;
 168                 updateName();
 169             }
 170
 171
 172         }
 173
 174         public void updateName() {
 175             session.asyncRequest(new WriteQuery(ComplexReadWriteQueryTest.this) {
 176                 @Override
 177                 public void run(WriteGraph g) throws Throwable {
 178                     Layer0 b = Layer0.getInstance(g);
 179                     String graphName = g.getRelatedValue(resource, b.HasName);
 180                     if(!graphName.equals(currentName)) {
 181                         throw new Exception("Graph contains name " + graphName + " while name was set to " + currentName);
 182                     }
 183                     currentName = UUID.randomUUID().toString();
 184                     g.claimLiteral(resource, b.HasName, currentName, Bindings.STRING);
 185                     if (changes >= MAX_CHANGES) {
 186                         dispose();
 187                         return;
 188                     }
 189                     incChanges();
 190                     // System.out.println("changes="+changes);
 191                     //System.out.println("Name change " +procedure.toString());
 192                 }
 193             });
 194
 195         }
 196
 197         public void dispose() {
 198             if (!disposed) {
 199                 ++endCount;
 200                 if (endCount >= NUM_THREADS) {
 201                     ended = true;
 202                     if (null != self)
 203                         self.interrupt();
 204                 }
 205             }
 206             disposed = true;
 207         }
 208
 209     }
 210
 211     public class NameRead implements Read<String> {
 212         private Resource resource;
 213
 214         public NameRead(Resource resource) {
 215             this.resource = resource;
 216         }
 217
 218         @Override
 219         public String perform(ReadGraph graph) throws DatabaseException {
 220             Layer0 b = Layer0.getInstance(graph);
 221             String name = graph.getRelatedValue(resource, b.HasName, Bindings.STRING);
 222             return name;
 223         }
 224     }
 225
 226 }