bundles/org.simantics.db.indexing/src/org/simantics/db/indexing/MemoryIndexing.java

   1 package org.simantics.db.indexing;
   2
   3 import java.io.IOException;
   4 import java.nio.file.Path;
   5 import java.util.Collection;
   6 import java.util.List;
   7 import java.util.Map;
   8 import java.util.concurrent.ConcurrentHashMap;
   9 import java.util.concurrent.Semaphore;
  10 import java.util.stream.Stream;
  11
  12 import org.apache.lucene.analysis.Analyzer;
  13 import org.apache.lucene.index.IndexWriter;
  14 import org.apache.lucene.index.IndexWriterConfig;
  15 import org.apache.lucene.index.IndexWriterConfig.OpenMode;
  16 import org.apache.lucene.store.Directory;
  17 import org.apache.lucene.store.RAMDirectory;
  18 import org.apache.lucene.util.Version;
  19 import org.eclipse.core.runtime.IProgressMonitor;
  20 import org.eclipse.core.runtime.SubMonitor;
  21 import org.simantics.db.RequestProcessor;
  22 import org.simantics.db.Resource;
  23 import org.simantics.db.Session;
  24 import org.simantics.db.common.request.Adapt;
  25 import org.simantics.db.indexing.IndexedRelationsSearcherBase.State;
  26 import org.simantics.db.layer0.adapter.GenericRelation;
  27 import org.slf4j.LoggerFactory;
  28
  29 /**
  30  * @author Tuukka Lehtonen
  31  * @author Antti Villberg
  32  */
  33 public class MemoryIndexing {
  34
  35     private static final org.slf4j.Logger LOGGER = LoggerFactory.getLogger(MemoryIndexing.class);
  36
  37     final private Session session;
  38
  39     final Map<String,Map<String,List<Map<String, Object>>>> persistentCache = new ConcurrentHashMap<>();
  40     final Map<String,Map<String,List<Resource>>> persistentCacheResources = new ConcurrentHashMap<>();
  41
  42     final private ConcurrentHashMap<String,RAMDirectory> directories = new ConcurrentHashMap<>();
  43
  44     final private ConcurrentHashMap<String,IndexedRelationsSearcherBase> immutableSearchers = new ConcurrentHashMap<>();
  45     final private ConcurrentHashMap<String,IndexedRelationsSearcher> searchers = new ConcurrentHashMap<>();
  46
  47     public MemoryIndexing(Session session) {
  48         this.session = session;
  49     }
  50
  51     protected Path getIndexDirectory(Resource relation, Resource input) {
  52         return DatabaseIndexing.getIndexLocation(session, relation, input);
  53     }
  54
  55     public IndexedRelationsSearcher get(RequestProcessor processor, Resource relation, Resource input) {
  56         Path location = getIndexDirectory(relation, input);
  57         String key = location.toAbsolutePath().toString();
  58         return searchers.computeIfAbsent(key, t -> {
  59             try {
  60                 GenericRelation r = processor.sync(new Adapt<GenericRelation>(relation, GenericRelation.class));
  61                 return new IndexedRelationsSearcher(processor, relation, input, r);
  62             } catch (Exception e) {
  63                 LOGGER.error("Could not get searcher for relation {} and input {} in location {}", relation, input, location, e);
  64                 return null;
  65             }
  66         });
  67     }
  68
  69     public IndexedRelationsSearcherBase getImmutable(RequestProcessor processor, Resource relation, Resource input) {
  70         Path location = getIndexDirectory(relation, input);
  71         String key = location.toAbsolutePath().toString();
  72         return immutableSearchers.computeIfAbsent(key, t -> {
  73             try {
  74                 return new ImmutableIndexedRelationsSearcher(processor, relation, input);
  75             } catch (Exception e) {
  76                 LOGGER.error("Could not get searcher base for relation {} and input {} in location {}", relation, input, location, e);
  77                 return null;
  78             }
  79         });
  80     }
  81
  82     public static MemoryIndexing getInstance(Session session) {
  83         MemoryIndexing ret = session.peekService(MemoryIndexing.class);
  84         if(ret == null) {
  85             ret = new MemoryIndexing(session);
  86             session.registerService(MemoryIndexing.class, ret);
  87         }
  88         return ret;
  89     }
  90
  91     public Directory getDirectory(String path, Analyzer analyzer) throws IOException {
  92         try {
  93             return directories.computeIfAbsent(path, t -> {
  94                 try {
  95                     RAMDirectory directory = new RAMDirectory();
  96                     IndexWriterConfig config = new IndexWriterConfig(Version.LUCENE_4_9, analyzer);
  97                     new IndexWriter(directory, config.setOpenMode(OpenMode.CREATE)).close();
  98                     return directory;
  99                 } catch (IOException e) {
 100                     throw new RuntimeException(e);
 101                 }
 102             });
 103         } catch (RuntimeException e) {
 104             throw (IOException) e.getCause();
 105         }
 106     }
 107
 108     public void remove(String path) {
 109         directories.remove(path);
 110     }
 111
 112     public void flush(IProgressMonitor progress) throws Exception {
 113         long startTime = System.currentTimeMillis();
 114         SubMonitor monitor = SubMonitor.convert(progress);
 115         Collection<IndexedRelationsSearcher> searcherEntries = searchers.values();
 116         Collection<IndexedRelationsSearcherBase> immutableSearcherEntries = immutableSearchers.values();
 117         int count = searcherEntries.size() + immutableSearcherEntries.size();
 118         Semaphore sema = new Semaphore(0);
 119         Stream.concat(searcherEntries.stream(), immutableSearcherEntries.stream()).parallel().forEach(base -> {
 120             try {
 121                 if (base.isIndexAvailable()) {
 122                     if (base instanceof IndexedRelationsSearcher) {
 123                         IndexedRelationsMemorySearcher searcher = ((IndexedRelationsSearcher) base).cache;
 124                         try {
 125                             List<Object[]> os = searcher.allDocs(monitor, session);
 126                             ((IndexedRelationsSearcher) base).applyChanges(monitor, session, searcher.r, os);
 127                         } catch (Exception e) {
 128                             LOGGER.error("Could not flush", e);
 129                         }
 130                     }
 131                 }
 132                 monitor.worked(1);
 133                 base.changeState(monitor, session, State.READY);
 134             } finally {
 135                 sema.release();
 136             }
 137         });
 138         sema.acquire(count);
 139         long totalTime = System.currentTimeMillis() - startTime;
 140         LOGGER.info("index flush " + totalTime);
 141     }
 142
 143 }