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