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