--- /dev/null
+/*******************************************************************************\r
+ * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
+ * in Industry THTH ry.\r
+ * All rights reserved. This program and the accompanying materials\r
+ * are made available under the terms of the Eclipse Public License v1.0\r
+ * which accompanies this distribution, and is available at\r
+ * http://www.eclipse.org/legal/epl-v10.html\r
+ *\r
+ * Contributors:\r
+ * VTT Technical Research Centre of Finland - initial API and implementation\r
+ *******************************************************************************/\r
+package org.simantics.db.indexing;\r
+\r
+import java.io.File;\r
+import java.io.IOException;\r
+import java.util.Collection;\r
+import java.util.Collections;\r
+import java.util.List;\r
+import java.util.Map;\r
+import java.util.WeakHashMap;\r
+import java.util.concurrent.locks.Lock;\r
+import java.util.concurrent.locks.ReentrantReadWriteLock;\r
+\r
+import org.apache.lucene.queryparser.classic.ParseException;\r
+import org.eclipse.core.runtime.IProgressMonitor;\r
+import org.eclipse.core.runtime.SubMonitor;\r
+import org.simantics.db.ReadGraph;\r
+import org.simantics.db.RequestProcessor;\r
+import org.simantics.db.Resource;\r
+import org.simantics.db.common.request.UniqueRead;\r
+import org.simantics.db.exception.DatabaseException;\r
+import org.simantics.db.exception.InvalidResourceReferenceException;\r
+import org.simantics.db.indexing.IndexedRelationsSearcherBase.State;\r
+import org.simantics.db.layer0.adapter.GenericRelation;\r
+import org.simantics.db.layer0.genericrelation.IndexException;\r
+import org.simantics.db.layer0.genericrelation.IndexedRelations;\r
+import org.simantics.db.service.QueryControl;\r
+import org.simantics.db.service.SerialisationSupport;\r
+import org.simantics.utils.datastructures.Pair;\r
+\r
+/**\r
+ * @author Tuukka Lehtonen\r
+ * @author Antti Villberg\r
+ */\r
+public class IndexedRelationsImpl implements IndexedRelations {\r
+\r
+ Map<Object, RWLock> indexLocks = new WeakHashMap<Object, RWLock>();\r
+\r
+ static class LockHandle {\r
+ public final Object id;\r
+ public final Lock lock;\r
+\r
+ public LockHandle(Object id, Lock lock) {\r
+ this.id = id;\r
+ this.lock = lock;\r
+ }\r
+\r
+ public void unlock() {\r
+ if (IndexPolicy.TRACE_INDEX_LOCKING)\r
+ System.out.println("Unlocking index " + id);\r
+ lock.unlock();\r
+ }\r
+ }\r
+\r
+ static class RWLock {\r
+ public final Object id;\r
+ public final ReentrantReadWriteLock lock;\r
+\r
+ public RWLock(Object id) {\r
+ this.id = id;\r
+ this.lock = new ReentrantReadWriteLock(true);\r
+ }\r
+\r
+ LockHandle lock(RequestProcessor processor, boolean write) {\r
+ Lock l = write ? lock.writeLock() : lock.readLock();\r
+ if(processor instanceof ReadGraph) {\r
+ ReadGraph graph = (ReadGraph)processor;\r
+ while(!l.tryLock()) {\r
+ QueryControl qc = processor.getService(QueryControl.class);\r
+ boolean executed = qc.resume(graph);\r
+ if(!executed) {\r
+ try {\r
+ Thread.sleep(1);\r
+ } catch (InterruptedException e) {\r
+ }\r
+ }\r
+ }\r
+ } else {\r
+ l.lock();\r
+ }\r
+ if (IndexPolicy.TRACE_INDEX_LOCKING)\r
+ System.out.println("Locked index " + id);\r
+ return new LockHandle(id, l);\r
+ }\r
+ \r
+ LockHandle tryLock(RequestProcessor processor, boolean write) {\r
+ Lock l = write ? lock.writeLock() : lock.readLock();\r
+ if(l.tryLock()) return new LockHandle(id, l);\r
+ else return null;\r
+ }\r
+ \r
+ }\r
+\r
+ private LockHandle lock(RequestProcessor processor, Object lockIdentifier, boolean write) {\r
+ RWLock rwlock = null;\r
+ synchronized (indexLocks) {\r
+ rwlock = indexLocks.get(lockIdentifier);\r
+ if (rwlock == null) {\r
+ rwlock = new RWLock(lockIdentifier);\r
+ indexLocks.put(lockIdentifier, rwlock);\r
+ }\r
+ }\r
+ return rwlock.lock(processor, write);\r
+ }\r
+\r
+ private LockHandle tryLock(RequestProcessor processor, Object lockIdentifier, boolean write) {\r
+ RWLock rwlock = null;\r
+ synchronized (indexLocks) {\r
+ rwlock = indexLocks.get(lockIdentifier);\r
+ if (rwlock == null) {\r
+ rwlock = new RWLock(lockIdentifier);\r
+ indexLocks.put(lockIdentifier, rwlock);\r
+ }\r
+ }\r
+ return rwlock.tryLock(processor, write);\r
+ }\r
+\r
+ private IndexedRelationsSearcherBase makeSearcher(final RequestProcessor processor, final Resource relation, final Resource input) {\r
+ try {\r
+ return processor.syncRequest(new UniqueRead<IndexedRelationsSearcherBase>() {\r
+\r
+ @Override\r
+ public IndexedRelationsSearcherBase perform(ReadGraph graph) throws DatabaseException {\r
+ if(graph.isImmutable(input)) {\r
+ return MemoryIndexing.getInstance(processor.getSession()).getImmutable(processor, relation, input);\r
+ } else {\r
+ return MemoryIndexing.getInstance(processor.getSession()).get(processor, relation, input);\r
+ }\r
+ }\r
+ \r
+ });\r
+ } catch (DatabaseException e) {\r
+ throw new IllegalStateException(e);\r
+ }\r
+ }\r
+ \r
+ private LockHandle waitLoaded(SubMonitor progress, final IndexedRelationsSearcherBase searcher, RequestProcessor processor, LockHandle lock, final Object lockId, final Resource input) throws IndexException {\r
+ \r
+ // Initial state: we are locked, no news about the index\r
+ // Final state: we are locked and the index has been loaded, the current lock is returned\r
+\r
+ // First just check if the index is loaded\r
+ if (searcher.isIndexAvailable()) {\r
+ // We have an index - try to start access\r
+ searcher.startAccess(progress.newChild(50), processor.getSession(), false);\r
+ // At this point we have three options:\r
+ // 1. we have access\r
+ if(searcher.hasAccess(false)) return lock;\r
+ // 2. something is wrong and the index cannot be cleaned\r
+ if(searcher.checkState(State.PROBLEM)) throw new IndexException("Searcher is in problematic state", searcher.getException());\r
+ // 3. something was wrong, but the index has been successfully cleaned\r
+ }\r
+\r
+ if(!searcher.checkState(State.NONE)) \r
+ throw new IndexException("Illegal searcher state, contact application support.");\r
+ \r
+ // We loop until the index is loaded\r
+ while(true) {\r
+ \r
+ // With ReadGraph we can proceed to initialize\r
+ if(processor instanceof ReadGraph) {\r
+\r
+ // (re)create the index.\r
+ try {\r
+ SerialisationSupport ss = processor.getService(SerialisationSupport.class);\r
+ searcher.initializeIndex(progress.newChild(40), (ReadGraph)processor, new Object[] { ss.getRandomAccessId(input) }, true);\r
+ searcher.setReady();\r
+ searcher.startAccess(progress.newChild(10), processor.getSession(), false);\r
+ if(searcher.hasAccess(false)) return lock;\r
+ } catch (IOException e) {\r
+ searcher.setProblem(e);\r
+ throw new IndexException(e);\r
+ } catch (DatabaseException e) {\r
+ searcher.setProblem(e);\r
+ throw new IndexException(e);\r
+ }\r
+ \r
+ }\r
+ // With session we schedule the job\r
+ else {\r
+\r
+ // Release lock\r
+ lock.unlock();\r
+ \r
+// final Semaphore s = new Semaphore(0);\r
+ \r
+ // Schedule job\r
+ \r
+ boolean success = false;\r
+ int tries = 0;\r
+ while(!success && (++tries)<10) {\r
+\r
+ try {\r
+\r
+ success = processor.sync(new UniqueRead<Boolean>() {\r
+\r
+ @Override\r
+ public Boolean perform(ReadGraph graph) throws DatabaseException {\r
+\r
+ // Obtain lock\r
+ LockHandle lock = tryLock(graph, lockId, true);\r
+ if(lock == null) return false;\r
+\r
+ try {\r
+\r
+ boolean loaded = false;\r
+ if (searcher.isIndexAvailable()) {\r
+ searcher.startAccess(null, graph.getSession(), false);\r
+ // At this point we have three options:\r
+ // 1. we have access\r
+ if(searcher.hasAccess(false)) loaded = true;\r
+ // 2. something is wrong and the index cannot be cleaned\r
+ if(searcher.checkState(State.PROBLEM)) throw new DatabaseException("Searcher is in problematic state", searcher.getException());\r
+ // 3. something was wrong, but the index has been successfully cleaned\r
+ }\r
+\r
+ if(!loaded) {\r
+ \r
+ if(!searcher.checkState(State.NONE)) \r
+ throw new DatabaseException("Illegal searcher state, contact application support."); \r
+\r
+ try {\r
+ SerialisationSupport ss = graph.getService(SerialisationSupport.class);\r
+ searcher.initializeIndex(null, graph, new Object[] { ss.getRandomAccessId(input) }, true);\r
+ searcher.setReady();\r
+ } catch (IOException e) {\r
+ searcher.setProblem(e);\r
+ throw new DatabaseException(e);\r
+ } \r
+\r
+ }\r
+\r
+ } finally {\r
+ \r
+ lock.unlock();\r
+// s.release();\r
+ \r
+ }\r
+\r
+ return true;\r
+\r
+ }\r
+\r
+ });\r
+\r
+ } catch (DatabaseException e) {\r
+ throw new IndexException(e);\r
+ }\r
+\r
+ }\r
+ \r
+ if(!success)\r
+ throw new IndexException("Did not manage to load index. Contact application support.");\r
+ \r
+ // Try again\r
+\r
+ // Obtain lock\r
+ lock= lock(processor, lockId, true); \r
+ \r
+ if (searcher.isIndexAvailable()) {\r
+ searcher.startAccess(progress.newChild(50), processor.getSession(), false);\r
+ if(searcher.hasAccess(false)) return lock;\r
+ throw new IndexException("Illegal searcher state, contact application support.");\r
+ }\r
+ \r
+ }\r
+ \r
+ }\r
+\r
+ \r
+ }\r
+ \r
+ @Override\r
+ public List<Resource> queryResources(IProgressMonitor monitor, String search, RequestProcessor processor,\r
+ Resource relation, final Resource input, int maxResultCount) {\r
+ if (processor == null)\r
+ throw new IllegalArgumentException("null processor");\r
+ if (relation == null)\r
+ throw new IllegalArgumentException("null relation");\r
+ if (input == null)\r
+ throw new IllegalArgumentException("null input");\r
+ if (search == null)\r
+ throw new IllegalArgumentException("null search criterion");\r
+\r
+ SubMonitor progress = SubMonitor.convert(monitor, 100);\r
+\r
+ // Look for existing index.\r
+ // Indexes always exist in secondary storage, i.e. disk.\r
+ // Indexes can be cached in memory when necessary performance-wise.\r
+\r
+ final IndexedRelationsSearcherBase searcher = makeSearcher(processor, relation, input);\r
+\r
+ final Object lockId = Pair.make(relation, input);\r
+\r
+ LockHandle lock = lock(processor, lockId, false);\r
+\r
+ // Ensure that index is loaded & ready\r
+ lock = waitLoaded(progress, searcher, processor, lock, lockId, input);\r
+\r
+ // Perform query\r
+ try {\r
+ return searcher.doSearchResources(progress.newChild(50), processor, search, maxResultCount);\r
+ } catch (ParseException e) {\r
+ // FIXME: should throw an exception, not just ignore.\r
+ e.printStackTrace();\r
+ return Collections.emptyList();\r
+ //throw new IndexException(e);\r
+ } catch (IOException e) {\r
+ throw new IndexException(e);\r
+ } catch (DatabaseException e) {\r
+ throw new IndexException(e);\r
+ } finally {\r
+ lock.unlock();\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public List<Map<String, Object>> query(IProgressMonitor monitor, String search, RequestProcessor processor,\r
+ Resource relation, final Resource input, int maxResultCount) {\r
+ if (processor == null)\r
+ throw new IllegalArgumentException("null processor");\r
+ if (relation == null)\r
+ throw new IllegalArgumentException("null relation");\r
+ if (input == null)\r
+ throw new IllegalArgumentException("null input");\r
+ if (search == null)\r
+ throw new IllegalArgumentException("null search criterion");\r
+\r
+ SubMonitor progress = SubMonitor.convert(monitor, 100);\r
+\r
+ // Look for existing index.\r
+ // Indexes always exist in secondary storage, i.e. disk.\r
+ // Indexes can be cached in memory when necessary performance-wise.\r
+\r
+ final IndexedRelationsSearcherBase searcher = makeSearcher(processor, relation, input);\r
+\r
+ final Object lockId = Pair.make(relation, input);\r
+\r
+ LockHandle lock = lock(processor, lockId, false);\r
+\r
+ // Ensure that index is loaded & ready\r
+ lock = waitLoaded(progress, searcher, processor, lock, lockId, input);\r
+\r
+ // Perform query\r
+ try {\r
+ return searcher.doSearch(progress.newChild(50), processor, search, maxResultCount);\r
+ } catch (ParseException e) {\r
+ // FIXME: should throw an exception, not just ignore.\r
+ e.printStackTrace();\r
+ return Collections.emptyList();\r
+ //throw new IndexException(e);\r
+ } catch (IOException e) {\r
+ throw new IndexException(e);\r
+ } catch (DatabaseException e) {\r
+ throw new IndexException(e);\r
+ } finally {\r
+ lock.unlock();\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public void insert(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation, \r
+ Resource relationResource, Resource input, Collection<Object[]> documents) {\r
+\r
+// System.out.println("Inserting to index: " + input + " " + documents);\r
+\r
+ if (relation == null)\r
+ throw new IllegalArgumentException("null relation");\r
+ if (input == null)\r
+ throw new IllegalArgumentException("null input");\r
+ if (documents == null)\r
+ throw new IllegalArgumentException("null documents");\r
+\r
+ if (documents.isEmpty())\r
+ return;\r
+\r
+ final SubMonitor progress = SubMonitor.convert(monitor, 100);\r
+\r
+ final IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);\r
+\r
+ LockHandle handle = lock(processor, Pair.make(relationResource, input), true);\r
+ \r
+ try {\r
+ \r
+ if(!searcher.startAccess(null, processor.getSession(), true)) {\r
+ // Could not write index for some reason. Ignore and let the next index query reinitialize the index.\r
+ return;\r
+ }\r
+ \r
+ searcher.insertIndex(progress.newChild(40), relation, 1, documents);\r
+ DatabaseIndexing.markIndexChanged(searcher.getIndexPath());\r
+ \r
+ } catch (InvalidResourceReferenceException e) {\r
+ throw new IndexException(e);\r
+ } catch (IOException e) {\r
+ throw new IndexException(e);\r
+ } catch (DatabaseException e) {\r
+ throw new IndexException(e);\r
+ } finally {\r
+ handle.unlock();\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public void remove(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation,\r
+ Resource relationResource, Resource input, String key, Collection<Object> keyValues) {\r
+\r
+ if (relation == null)\r
+ throw new IllegalArgumentException("null relation");\r
+ if (input == null)\r
+ throw new IllegalArgumentException("null input");\r
+ if (key == null)\r
+ throw new IllegalArgumentException("null key");\r
+\r
+ SubMonitor progress = SubMonitor.convert(monitor, 100);\r
+\r
+ IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);\r
+\r
+ LockHandle handle = lock(processor, Pair.make(relationResource, input), true);\r
+ try {\r
+ \r
+ if(!searcher.startAccess(null, processor.getSession(), true)) {\r
+ // Could not write index for some reason. Ignore and let the next index query reinitialize the index.\r
+ return;\r
+ }\r
+ \r
+ searcher.removeIndex(progress.newChild(40), relation, processor, key, keyValues);\r
+ DatabaseIndexing.markIndexChanged(searcher.getIndexPath());\r
+ \r
+ } catch (DatabaseException e) {\r
+ throw new IndexException(e);\r
+ } catch (IOException e) {\r
+ throw new IndexException(e);\r
+ } finally {\r
+ handle.unlock();\r
+ }\r
+ }\r
+\r
+ @Override\r
+ public void removeAll(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation,\r
+ Resource relationResource, Resource input) {\r
+\r
+ if (relation == null)\r
+ throw new IllegalArgumentException("null relation");\r
+ if (input == null)\r
+ throw new IllegalArgumentException("null input");\r
+\r
+ IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);\r
+ \r
+ LockHandle handle = lock(processor, Pair.make(relationResource, input), true);\r
+\r
+ try {\r
+ \r
+ Throwable t = searcher.bestEffortClear(monitor, processor.getSession());\r
+ if(t != null) searcher.setProblem(t);\r
+ else searcher.setNone();\r
+ \r
+ } finally {\r
+ handle.unlock();\r
+ }\r
+ \r
+ }\r
+ \r
+ @Override\r
+ public boolean replace(IProgressMonitor monitor, RequestProcessor processor, GenericRelation relation,\r
+ Resource relationResource, Resource input, String key, Collection<Object> keyValues, Collection<Object[]> documents) {\r
+\r
+ if (relation == null)\r
+ throw new IllegalArgumentException("null relation");\r
+ if (input == null)\r
+ throw new IllegalArgumentException("null input");\r
+ if (key == null)\r
+ throw new IllegalArgumentException("null key");\r
+\r
+ SubMonitor progress = SubMonitor.convert(monitor, 100);\r
+\r
+ IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);\r
+\r
+ LockHandle handle = lock(processor, Pair.make(relationResource, input), true);\r
+\r
+ boolean didChange = false;\r
+\r
+ try {\r
+ \r
+ if(!searcher.startAccess(null, processor.getSession(), true)) {\r
+ // Could not write index for some reason. Ignore and let the next index query reinitialize the index.\r
+ return true;\r
+ }\r
+ didChange = searcher.replaceIndex(progress.newChild(40), key, keyValues, relation, 1, documents);\r
+ if(didChange)\r
+ DatabaseIndexing.markIndexChanged(searcher.getIndexPath());\r
+ \r
+ } catch (InvalidResourceReferenceException e) {\r
+ throw new IndexException(e);\r
+ } catch (IOException e) {\r
+ throw new IndexException(e);\r
+ } catch (DatabaseException e) {\r
+ throw new IndexException(e);\r
+ } catch (Throwable t) {\r
+ throw new IndexException(t);\r
+ } finally {\r
+ handle.unlock();\r
+ }\r
+ \r
+ return didChange;\r
+\r
+ }\r
+ \r
+ @Override\r
+ public void reset(IProgressMonitor monitor, RequestProcessor processor, Resource relationResource, Resource input) throws IndexException {\r
+\r
+ IndexedRelationsSearcherBase searcher = makeSearcher(processor, relationResource, input);\r
+\r
+ LockHandle handle = lock(processor, Pair.make(relationResource, input), true);\r
+\r
+ try {\r
+ searcher.changeState(monitor, processor.getSession(), State.NONE);\r
+ if (!searcher.checkState(State.NONE))\r
+ throw new IndexException("Could not close index for input " + input + " before removing it");\r
+\r
+ File path = DatabaseIndexing.getIndexLocation(processor.getSession(), relationResource, input);\r
+ DatabaseIndexing.deleteIndex(path);\r
+\r
+ } catch (IOException e) {\r
+ throw new IndexException(e);\r
+ } finally {\r
+ handle.unlock();\r
+ }\r
+\r
+ }\r
+\r
+}\r
Code Review / simantics / platform.git / blobdiff