bundles/org.simantics.db.indexing/src/org/simantics/db/indexing/IndexedRelationsImpl.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.indexing;
  13
  14 import java.io.File;
  15 import java.io.IOException;
  16 import java.nio.file.Path;
  17 import java.util.Collection;
  18 import java.util.Collections;
  19 import java.util.List;
  20 import java.util.Map;
  21 import java.util.WeakHashMap;
  22 import java.util.concurrent.locks.Lock;
  23 import java.util.concurrent.locks.ReentrantReadWriteLock;
  24
  25 import org.apache.lucene.queryparser.classic.ParseException;
  26 import org.eclipse.core.runtime.IProgressMonitor;
  27 import org.eclipse.core.runtime.SubMonitor;
  28 import org.simantics.db.ReadGraph;
  29 import org.simantics.db.RequestProcessor;
  30 import org.simantics.db.Resource;
  31 import org.simantics.db.common.request.UniqueRead;
  32 import org.simantics.db.exception.DatabaseException;
  33 import org.simantics.db.exception.InvalidResourceReferenceException;
  34 import org.simantics.db.indexing.IndexedRelationsSearcherBase.State;
  35 import org.simantics.db.layer0.adapter.GenericRelation;
  36 import org.simantics.db.layer0.genericrelation.IndexException;
  37 import org.simantics.db.layer0.genericrelation.IndexedRelations;
  38 import org.simantics.db.service.QueryControl;
  39 import org.simantics.db.service.SerialisationSupport;
  40 import org.simantics.utils.datastructures.Pair;
  41 import org.slf4j.Logger;
  42 import org.slf4j.LoggerFactory;
  43
  44 /**
  45  * @author Tuukka Lehtonen
  46  * @author Antti Villberg
  47  */
  48 public class IndexedRelationsImpl implements IndexedRelations {
  49
  50     private static final Logger LOGGER = LoggerFactory.getLogger(IndexedRelationsImpl.class);
  51
  52     Map<Object, RWLock> indexLocks = new WeakHashMap<Object, RWLock>();
  53
  54     static class LockHandle {
  55         public final Object id;
  56         public final Lock   lock;
  57
  58         public LockHandle(Object id, Lock lock) {
  59             this.id = id;
  60             this.lock = lock;
  61         }
  62
  63         public void unlock() {
  64             if (IndexPolicy.TRACE_INDEX_LOCKING)
  65                 System.out.println("Unlocking index " + id);
  66             lock.unlock();
  67         }
  68     }
  69
  70     static class RWLock {
  71         public final Object                 id;
  72         public final ReentrantReadWriteLock lock;
  73
  74         public RWLock(Object id) {
  75             this.id = id;
  76             this.lock = new ReentrantReadWriteLock(true);
  77         }
  78
  79         LockHandle lock(RequestProcessor processor, boolean write) {
  80             Lock l = write ? lock.writeLock() : lock.readLock();
  81             if(processor instanceof ReadGraph) {
  82                 ReadGraph graph = (ReadGraph)processor;
  83                 while(!l.tryLock()) {
  84                         QueryControl qc = processor.getService(QueryControl.class);
  85                         boolean executed = qc.resume(graph);
  86                         if(!executed) {
  87                                                 try {
  88                                                         Thread.sleep(1);
  89                                                 } catch (InterruptedException e) {
  90                                                 }
  91                         }
  92                 }
  93             } else {
  94                 l.lock();
  95             }
  96             if (IndexPolicy.TRACE_INDEX_LOCKING)
  97                 System.out.println("Locked index " + id);
  98             return new LockHandle(id, l);
  99         }
 100
 101         LockHandle tryLock(RequestProcessor processor, boolean write) {
 102             Lock l = write ? lock.writeLock() : lock.readLock();
 103             if(l.tryLock()) return new LockHandle(id, l);
 104             else return null;
 105         }
 106
 107     }
 108
 109     private LockHandle lock(RequestProcessor processor, Object lockIdentifier, boolean write) {
 110         RWLock rwlock = null;
 111         synchronized (indexLocks) {
 112             rwlock = indexLocks.get(lockIdentifier);
 113             if (rwlock == null) {
 114                 rwlock = new RWLock(lockIdentifier);
 115                 indexLocks.put(lockIdentifier, rwlock);
 116             }
 117         }
 118         return rwlock.lock(processor, write);
 119     }
 120
 121     private LockHandle tryLock(RequestProcessor processor, Object lockIdentifier, boolean write) {
 122         RWLock rwlock = null;
 123         synchronized (indexLocks) {
 124             rwlock = indexLocks.get(lockIdentifier);
 125             if (rwlock == null) {
 126                 rwlock = new RWLock(lockIdentifier);
 127                 indexLocks.put(lockIdentifier, rwlock);
 128             }
 129         }
 130         return rwlock.tryLock(processor, write);
 131     }
 132
 133     private IndexedRelationsSearcherBase makeSearcher(final RequestProcessor processor, final Resource relation, final Resource input) {
 134         try {
 135                         return processor.syncRequest(new UniqueRead<IndexedRelationsSearcherBase>() {
 136
 137                                 @Override
 138                                 public IndexedRelationsSearcherBase perform(ReadGraph graph) throws DatabaseException {
 139                                         if(graph.isImmutable(input)) {
 140                                                 return MemoryIndexing.getInstance(processor.getSession()).getImmutable(processor, relation, input);
 141                                         } else {
 142                                                 return MemoryIndexing.getInstance(processor.getSession()).get(processor, relation, input);
 143                                         }
 144                                 }
 145
 146                         });
 147                 } catch (DatabaseException e) {
 148                         throw new IllegalStateException(e);
 149                 }
 150     }
 151
 152     private LockHandle waitLoaded(SubMonitor progress, final IndexedRelationsSearcherBase searcher, RequestProcessor processor, LockHandle lock, final Object lockId, final Resource input) throws IndexException {
 153
 154         // Initial state: we are locked, no news about the index
 155         // Final state: we are locked and the index has been loaded, the current lock is returned
 156
 157         // First just check if the index is loaded
 158         if (searcher.isIndexAvailable()) {
 159                 // We have an index - try to start access
 160                 searcher.startAccess(progress.newChild(50), processor.getSession(), false);
 161                 // At this point we have three options:
 162                 // 1. we have access
 163                 if(searcher.hasAccess(false)) return lock;
 164                 // 2. something is wrong and the index cannot be cleaned
 165                 if(searcher.checkState(State.PROBLEM)) throw new IndexException("Searcher is in problematic state", searcher.getException());
 166                 // 3. something was wrong, but the index has been successfully cleaned
 167         }
 168
 169                 if(!searcher.checkState(State.NONE))
 170                         throw new IndexException("Illegal searcher state, contact application support.");
 171
 172         // We loop until the index is loaded
 173         while(true) {
 174
 175                 // With ReadGraph we can proceed to initialize
 176                 if(processor instanceof ReadGraph) {
 177
 178                 // (re)create the index.
 179                 try {
 180                     SerialisationSupport ss = processor.getService(SerialisationSupport.class);
 181                     searcher.initializeIndex(progress.newChild(40), (ReadGraph)processor, new Object[] { ss.getRandomAccessId(input) }, true);
 182                                         searcher.setReady();
 183                     searcher.startAccess(progress.newChild(10), processor.getSession(), false);
 184                         if(searcher.hasAccess(false)) return lock;
 185                 } catch (IOException e) {
 186                                         searcher.setProblem(e);
 187                     throw new IndexException(e);
 188                 } catch (DatabaseException e) {
 189                                         searcher.setProblem(e);
 190                     throw new IndexException(e);
 191                 }
 192
 193                 }
 194                 // With session we schedule the job
 195                 else {
 196
 197                         // Release lock
 198                         lock.unlock();
 199
 200 //                      final Semaphore s = new Semaphore(0);
 201
 202                         // Schedule job
 203
 204                         boolean success = false;
 205                         int tries = 0;
 206                         while(!success && (++tries)<10) {
 207
 208                                 try {
 209
 210                                         success = processor.sync(new UniqueRead<Boolean>() {
 211
 212                                                 @Override
 213                                                 public Boolean perform(ReadGraph graph) throws DatabaseException {
 214
 215                                                         // Obtain lock
 216                                                         LockHandle lock = tryLock(graph, lockId, true);
 217                                                         if(lock == null) return false;
 218
 219                                                         try {
 220
 221                                                                 boolean loaded = false;
 222                                                                 if (searcher.isIndexAvailable()) {
 223                                                                         searcher.startAccess(null, graph.getSession(), false);
 224                                                                         // At this point we have three options:
 225                                                                         // 1. we have access
 226                                                                         if(searcher.hasAccess(false)) loaded = true;
 227                                                                         // 2. something is wrong and the index cannot be cleaned
 228                                                                         if(searcher.checkState(State.PROBLEM)) throw new DatabaseException("Searcher is in problematic state", searcher.getException());
 229                                                                         // 3. something was wrong, but the index has been successfully cleaned
 230                                                                 }
 231
 232                                                                 if(!loaded) {
 233
 234                                                                 if(!searcher.checkState(State.NONE))
 235                                                                         throw new DatabaseException("Illegal searcher state, contact application support.");
 236
 237                                                                         try {
 238                                                                             SerialisationSupport ss = graph.getService(SerialisationSupport.class);
 239                                                                                 searcher.initializeIndex(null, graph, new Object[] { ss.getRandomAccessId(input) }, true);
 240                                                                                 searcher.setReady();
 241                                                                         } catch (IOException e) {
 242                                                                                 searcher.setProblem(e);
 243                                                                                 throw new DatabaseException(e);
 244                                                                         }
 245
 246                                                                 }
 247
 248                                                         } finally {
 249
 250                                                                 lock.unlock();
 251 //                                                              s.release();
 252
 253                                                         }
 254
 255                                                         return true;
 256
 257                                                 }
 258
 259                                         });
 260
 261                                 } catch (DatabaseException e) {
 262                                         throw new IndexException(e);
 263                                 }
 264
 265                         }
 266
 267                         if(!success)
 268                                 throw new IndexException("Did not manage to load index. Contact application support.");
 269
 270                 // Try again
 271
 272                 // Obtain lock
 273                                 lock= lock(processor, lockId, true);
 274
 275                         if (searcher.isIndexAvailable()) {
 276                                 searcher.startAccess(progress.newChild(50), processor.getSession(), false);
 277                                 if(searcher.hasAccess(false)) return lock;
 278                                 throw new IndexException("Illegal searcher state, contact application support.");
 279                         }
 280
 281                 }
 282
 283         }
 284
 285
 286     }
 287
 288     @Override
 289     public List<Resource> queryResources(IProgressMonitor monitor, String search, RequestProcessor processor,
 290             Resource relation, final Resource input, int maxResultCount) {
 291         if (processor == null)
 292             throw new IllegalArgumentException("null processor");
 293         if (relation == null)
 294             throw new IllegalArgumentException("null relation");
 295         if (input == null)
 296             throw new IllegalArgumentException("null input");
 297         if (search == null)
 298             throw new IllegalArgumentException("null search criterion");
 299
 300         SubMonitor progress = SubMonitor.convert(monitor, 100);
 301
 302         // Look for existing index.
 303         // Indexes always exist in secondary storage, i.e. disk.
 304         // Indexes can be cached in memory when necessary performance-wise.
 305
 306         final IndexedRelationsSearcherBase searcher = makeSearcher(processor, relation, input);
 307
 308         final Object lockId = Pair.make(relation, input);
 309
 310         LockHandle lock = lock(processor, lockId, false);
 311
 312         // Ensure that index is loaded & ready
 313         lock = waitLoaded(progress, searcher, processor, lock, lockId, input);
 314
 315         // Perform query
 316         try {
 317             return searcher.doSearchResources(progress.newChild(50), processor, search, maxResultCount);
 318         } catch (ParseException e) {
 319             // FIXME: should throw an exception, not just ignore.
 320             e.printStackTrace();
 321             return Collections.emptyList();
 322             //throw new IndexException(e);
 323         } catch (IOException e) {
 324             throw new IndexException(e);
 325         } catch (DatabaseException e) {
 326             throw new IndexException(e);
 327         } finally {
 328                 lock.unlock();
 329         }
 330     }
 331
 332     @Override
 333     public List<Map<String, Object>> query(IProgressMonitor monitor, String search, RequestProcessor processor,
 334             Resource relation, final Resource input, int maxResultCount) {
 335         if (processor == null)
 336             throw new IllegalArgumentException("null processor");
 337         if (relation == null)
 338             throw new IllegalArgumentException("null relation");
 339         if (input == null)
 340             throw new IllegalArgumentException("null input");
 341         if (search == null)
 342             throw new IllegalArgumentException("null search criterion");
 343
 344         SubMonitor progress = SubMonitor.convert(monitor, 100);
 345
 346         // Look for existing index.
 347         // Indexes always exist in secondary storage, i.e. disk.
 348         // Indexes can be cached in memory when necessary performance-wise.
 349
 350         final IndexedRelationsSearcherBase searcher = makeSearcher(processor, relation, input);
 351
 352         final Object lockId = Pair.make(relation, input);
 353
 354         LockHandle lock = lock(processor, lockId, false);
 355
 356         // Ensure that index is loaded & ready
 357         lock = waitLoaded(progress, searcher, processor, lock, lockId, input);
 358
 359         // Perform query
 360         try {
 361             return searcher.doSearch(progress.newChild(50), processor, search, maxResultCount);
 362         } catch (ParseException e) {
 363             // FIXME: should throw an exception, not just ignore.
 364             e.printStackTrace();
 365             return Collections.emptyList();
 366             //throw new IndexException(e);
 367         } catch (IOException e) {
 368             throw new IndexException(e);
 369         } catch (DatabaseException e) {
 370             throw new IndexException(e);
 371         } finally {
 372                 lock.unlock();
 373         }
 374     }
 375
 376     @Override
 377     public void insert(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation,
 378             Resource relationResource, Resource input, Collection<Object[]> documents) throws IndexException {
 379
 380 //        System.out.println("Inserting to index: " + input + " " + documents);
 381
 382         if (relation == null)
 383             throw new IllegalArgumentException("null relation");
 384         if (input == null)
 385             throw new IllegalArgumentException("null input");
 386         if (documents == null)
 387             throw new IllegalArgumentException("null documents");
 388
 389         if (documents.isEmpty())
 390             return;
 391
 392         final SubMonitor progress = SubMonitor.convert(monitor, 100);
 393
 394         final IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);
 395
 396         LockHandle handle = lock(processor, Pair.make(relationResource, input), true);
 397
 398         try {
 399
 400                 DatabaseIndexing.markIndexChanged(processor.getSession(), searcher.getIndexPath());
 401                 if(!searcher.startAccess(null, processor.getSession(), true)) {
 402                 // Could not write index for some reason. Ignore and let the next index query reinitialize the index.
 403                         return;
 404                 }
 405
 406             searcher.insertIndex(progress.newChild(40), relation, 1, documents);
 407
 408         } catch (InvalidResourceReferenceException e) {
 409             throw new IndexException(e);
 410         } catch (IOException e) {
 411             throw new IndexException(e);
 412         } catch (DatabaseException e) {
 413             throw new IndexException(e);
 414         } finally {
 415                 handle.unlock();
 416         }
 417     }
 418
 419     @Override
 420     public void remove(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation,
 421             Resource relationResource, Resource input, String key, Collection<Object> keyValues) throws IndexException {
 422
 423         if (relation == null)
 424             throw new IllegalArgumentException("null relation");
 425         if (input == null)
 426             throw new IllegalArgumentException("null input");
 427         if (key == null)
 428             throw new IllegalArgumentException("null key");
 429
 430         SubMonitor progress = SubMonitor.convert(monitor, 100);
 431
 432         IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);
 433
 434         LockHandle handle = lock(processor, Pair.make(relationResource, input), true);
 435         try {
 436
 437             DatabaseIndexing.markIndexChanged(processor.getSession(), searcher.getIndexPath());
 438                 if(!searcher.startAccess(null, processor.getSession(), true)) {
 439                 // Could not write index for some reason. Ignore and let the next index query reinitialize the index.
 440                         return;
 441                 }
 442
 443             searcher.removeIndex(progress.newChild(40), relation, processor, key, keyValues);
 444
 445         } catch (DatabaseException e) {
 446             throw new IndexException(e);
 447         } catch (IOException e) {
 448             throw new IndexException(e);
 449         } finally {
 450                 handle.unlock();
 451         }
 452     }
 453
 454     @Override
 455     public void removeAll(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation,
 456             Resource relationResource, Resource input) {
 457
 458         if (relation == null)
 459             throw new IllegalArgumentException("null relation");
 460         if (input == null)
 461             throw new IllegalArgumentException("null input");
 462
 463         IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);
 464
 465         LockHandle handle = lock(processor, Pair.make(relationResource, input), true);
 466
 467         try {
 468
 469                 Throwable t = searcher.bestEffortClear(monitor, processor.getSession());
 470                 if(t != null) searcher.setProblem(t);
 471                 else searcher.setNone();
 472
 473                 } finally {
 474                         handle.unlock();
 475                 }
 476
 477     }
 478
 479     @Override
 480     public boolean replace(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation,
 481             Resource relationResource, Resource input, String key, Collection<Object> keyValues, Collection<Object[]> documents) throws IndexException {
 482
 483         if (relation == null)
 484             throw new IllegalArgumentException("null relation");
 485         if (input == null)
 486             throw new IllegalArgumentException("null input");
 487         if (key == null)
 488             throw new IllegalArgumentException("null key");
 489
 490         SubMonitor progress = SubMonitor.convert(monitor, 100);
 491
 492         IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);
 493
 494         LockHandle handle = lock(processor, Pair.make(relationResource, input), true);
 495
 496         boolean didChange = false;
 497
 498         try {
 499
 500                 DatabaseIndexing.markIndexChanged(processor.getSession(), searcher.getIndexPath());
 501                 if(!searcher.startAccess(null, processor.getSession(), true)) {
 502                 // Could not write index for some reason. Ignore and let the next index query reinitialize the index.
 503                         return true;
 504                 }
 505             didChange |= searcher.replaceIndex(progress.newChild(40), key, keyValues, relation, 1, documents);
 506
 507         } catch (InvalidResourceReferenceException e) {
 508             throw new IndexException(e);
 509         } catch (IOException e) {
 510             throw new IndexException(e);
 511         } catch (DatabaseException e) {
 512             throw new IndexException(e);
 513         } catch (Throwable t) {
 514             throw new IndexException(t);
 515         } finally {
 516                 handle.unlock();
 517         }
 518
 519         return didChange;
 520
 521     }
 522
 523     @Override
 524     public void reset(IProgressMonitor monitor, RequestProcessor processor, Resource relationResource, Resource input) throws IndexException {
 525
 526         IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);
 527
 528         LockHandle handle = lock(processor, Pair.make(relationResource, input), true);
 529         Path path = DatabaseIndexing.getIndexLocation(processor.getSession(), relationResource, input);
 530         try {
 531             searcher.changeState(monitor, processor.getSession(), State.NONE);
 532             if (!searcher.checkState(State.NONE))
 533                 throw new IndexException("Could not close index for input " + input + " before removing it");
 534
 535             DatabaseIndexing.deleteIndex(path);
 536         } catch (IOException e) {
 537             LOGGER.error("Could not delete {}", path.toAbsolutePath(), e);
 538             throw new IndexException(e);
 539         } finally {
 540             handle.unlock();
 541         }
 542
 543     }
 544
 545 }