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