[exo-jcr-commits] exo-jcr SVN: r3373 - in jcr/branches/1.14-CNK/exo.jcr.component.core: src/main/java/org/exoplatform/services/jcr/impl/core/query and 6 other directories.
do-not-reply at jboss.org
do-not-reply at jboss.org
Tue Nov 2 03:25:51 EDT 2010
Author: nzamosenchuk
Date: 2010-11-02 03:25:50 -0400 (Tue, 02 Nov 2010)
New Revision: 3373
Added:
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkIndexerCacheLoader.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkService.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkServiceImpl.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/JBossCacheChunkIndexChangesFilter.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/PersistentIndexChunk.java
Modified:
jcr/branches/1.14-CNK/exo.jcr.component.core/pom.xml
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/AbstractQueryHandler.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/QueryHandler.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChunkIndex.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/storage/jbosscache/AbstractWriteOnlyCacheLoader.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/util/io/PrivilegedCacheHelper.java
jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-indexer.xml
jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jcr-config.xml
Log:
EXOJCR-987 : POC commit. Nodes can join and leave the cluster. Some places contain hardcode and woraround, that are acceptable for now
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/pom.xml
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/pom.xml 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/pom.xml 2010-11-02 07:25:50 UTC (rev 3373)
@@ -30,7 +30,7 @@
<name>eXo JCR :: Component :: Core Service</name>
<description>eXo JCR Service core component</description>
<properties>
- <jcr.test.configuration.file>/conf/standalone/test-configuration.xml</jcr.test.configuration.file>
+ <jcr.test.configuration.file>/conf/standalone/cluster/test-configuration.xml</jcr.test.configuration.file>
<jbosscache.shareable>true</jbosscache.shareable>
</properties>
<dependencies>
@@ -434,6 +434,9 @@
<exclude>org/exoplatform/services/jcr/**/impl/**/SQLBenchmarkTest.java</exclude>
<exclude>org/exoplatform/services/jcr/**/impl/**/TestLockPerstistentDataManager.java</exclude>
<exclude>org/exoplatform/services/jcr/**/impl/**/TestCleanableFileStreamValueData.java</exclude>
+ <!-- After restore - default changes filter is used, so no ChunkService injected into serachManager, but
+ ChunkIndex is hardcoded and requires ChunkService -->
+ <exclude>org/exoplatform/services/jcr/**/TestSVNodeDataOptimization.java</exclude>
</excludes>
</configuration>
</plugin>
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/AbstractQueryHandler.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/AbstractQueryHandler.java 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/AbstractQueryHandler.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -18,6 +18,7 @@
import org.exoplatform.services.jcr.config.RepositoryConfigurationException;
import org.exoplatform.services.jcr.datamodel.NodeData;
+import org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService;
import org.exoplatform.services.jcr.impl.core.query.lucene.ChunkIndex;
import org.exoplatform.services.jcr.impl.core.query.lucene.DefaultIndexUpdateMonitor;
import org.exoplatform.services.jcr.impl.core.query.lucene.IndexInfos;
@@ -74,6 +75,12 @@
*/
protected IndexInfos indexInfos;
+ /**
+ * TODO: re-view this part of dependency delivery
+ * Temporary solution to deliver ChunkService to ChunkIndex (former MultiIndex)
+ */
+ protected ChunkService chunkService;
+
private IndexUpdateMonitor indexUpdateMonitor;
public boolean isInitialized()
@@ -82,6 +89,22 @@
}
/**
+ * @see org.exoplatform.services.jcr.impl.core.query.QueryHandler#getChunkService()
+ */
+ public ChunkService getChunkService()
+ {
+ return chunkService;
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.QueryHandler#setChunkService(org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService)
+ */
+ public void setChunkService(ChunkService chunkService)
+ {
+ this.chunkService = chunkService;
+ }
+
+ /**
* @see org.exoplatform.services.jcr.impl.core.query.QueryHandler#setIndexerIoModeHandler(org.exoplatform.services.jcr.impl.core.query.IndexerIoModeHandler)
*/
public void setIndexerIoModeHandler(IndexerIoModeHandler modeHandler) throws IOException
@@ -209,6 +232,7 @@
*
* @param idleTime the query handler idle time.
*/
+ @Deprecated
public void setIdleTime(String idleTime)
{
log.warn("Parameter 'idleTime' is not supported anymore. "
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/QueryHandler.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/QueryHandler.java 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/QueryHandler.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -21,6 +21,7 @@
import org.exoplatform.services.jcr.datamodel.NodeData;
import org.exoplatform.services.jcr.impl.core.SessionDataManager;
import org.exoplatform.services.jcr.impl.core.SessionImpl;
+import org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService;
import org.exoplatform.services.jcr.impl.core.query.lucene.ChunkIndex;
import org.exoplatform.services.jcr.impl.core.query.lucene.IndexInfos;
import org.exoplatform.services.jcr.impl.core.query.lucene.IndexUpdateMonitor;
@@ -161,7 +162,7 @@
* @param indexInfos
*/
void setIndexInfos(IndexInfos indexInfos);
-
+
/**
* Returns {@link IndexInfos} instance that was set into QueryHandler.
* @return
@@ -178,4 +179,16 @@
*/
void setIndexUpdateMonitor(IndexUpdateMonitor indexUpdateMonitor);
+ /**
+ * Temporary solution to deliver {@link ChunkService} to {@link ChunkIndex} (former MultiIndex)
+ * @param chunkService
+ */
+ void setChunkService(ChunkService chunkService);
+
+ /**
+ * Temporary solution to deliver {@link ChunkService} to {@link ChunkIndex} (former MultiIndex)
+ * @return
+ */
+ ChunkService getChunkService();
+
}
Added: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkIndexerCacheLoader.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkIndexerCacheLoader.java (rev 0)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkIndexerCacheLoader.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -0,0 +1,288 @@
+/*
+ * Copyright (C) 2003-2009 eXo Platform SAS.
+ *
+ * This program is free software; you can redistribute it and/or
+ * modify it under the terms of the GNU Affero General Public License
+ * as published by the Free Software Foundation; either version 3
+ * of the License, or (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program; if not, see<http://www.gnu.org/licenses/>.
+ */
+package org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk;
+
+import org.exoplatform.services.jcr.config.RepositoryConfigurationException;
+import org.exoplatform.services.jcr.impl.core.query.QueryHandler;
+import org.exoplatform.services.jcr.impl.core.query.SearchManager;
+import org.exoplatform.services.jcr.impl.core.query.jbosscache.ChangesFilterListsWrapper;
+import org.exoplatform.services.jcr.impl.storage.jbosscache.AbstractWriteOnlyCacheLoader;
+import org.exoplatform.services.log.ExoLogger;
+import org.exoplatform.services.log.Log;
+import org.jboss.cache.Fqn;
+import org.jboss.cache.Modification;
+
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
+import javax.jcr.RepositoryException;
+
+/**
+ * @author <a href="mailto:nikolazius at gmail.com">Nikolay Zamosenchuk</a>
+ * @version $Id: IndexerCacheLoader.java 34360 2009-07-22 23:58:59Z nzamosenchuk $
+ *
+ */
+public class ChunkIndexerCacheLoader extends AbstractWriteOnlyCacheLoader
+{
+ private static final Log log = ExoLogger.getLogger("exo.jcr.component.core.IndexerCacheLoader");
+
+ /**
+ * A map of all the indexers that has been registered
+ */
+ private final Map<Fqn<String>, Indexer> indexers = new HashMap<Fqn<String>, Indexer>();
+
+ /**
+ * @see org.jboss.cache.loader.AbstractCacheLoader#commit(java.lang.Object)
+ */
+ @Override
+ public void commit(Object tx) throws Exception
+ {
+ // do nothing. Everything is done on prepare phase.
+ }
+
+ /**
+ * This method will register a new Indexer according to the given parameters.
+ *
+ * @param searchManager
+ * @param parentSearchManager
+ * @param handler
+ * @param parentHandler
+ * @throws RepositoryConfigurationException
+ */
+ public void register(SearchManager searchManager, SearchManager parentSearchManager, QueryHandler handler,
+ QueryHandler parentHandler, ChunkService chunkService) throws RepositoryConfigurationException
+ {
+ indexers.put(Fqn.fromElements(searchManager.getWsId()), new Indexer(searchManager, parentSearchManager, handler,
+ parentHandler, chunkService));
+ if (log.isDebugEnabled())
+ {
+ log.debug("Register " + searchManager.getWsId() + " " + this + " in " + indexers);
+ }
+ }
+
+ /**
+ * @see org.jboss.cache.loader.CacheLoader#put(org.jboss.cache.Fqn, java.lang.Object, java.lang.Object)
+ */
+ public Object put(Fqn name, Object key, Object value) throws Exception
+ {
+ return null;
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.storage.jbosscache.AbstractWriteOnlyCacheLoader#put(org.jboss.cache.Fqn, java.util.Map)
+ */
+ public void put(Fqn fqn, Map<Object, Object> data) throws Exception
+ {
+ // ignoring call cacheRoot.addChild(PARAMETER_ROOT).setResident(true);
+ if (log.isDebugEnabled())
+ {
+ log.info("Received list wrapper, start indexing...");
+ }
+
+ // TODO Review it!!
+ if (fqn.getLastElementAsString().equals(ChunkServiceImpl.CHUNK_LIST))
+ {
+ // skipping all the rest of the code if service data written
+ log.info("skipping fqn " + fqn);
+ return;
+ }
+
+ Set<Integer> processingChunks = null;
+
+ ChunkService chunkService = null;
+
+ Indexer indexer = indexers.get(fqn.getParent());
+ if (indexer == null)
+ {
+ log.warn("No indexer could be found for the fqn " + fqn.getParent());
+ if (log.isDebugEnabled())
+ {
+ log.debug("The current content of the map of indexers is " + indexers);
+ }
+ }
+ else
+ {
+ try
+ {
+ // updating index
+ ChangesFilterListsWrapper wrapper =
+ new ChangesFilterListsWrapper(new HashSet<String>(), new HashSet<String>(), new HashSet<String>(),
+ new HashSet<String>());
+
+ // combine wrappers. Aggregating them into one wrapper, to pass through SearchManager and SearchIndex
+ chunkService = indexer.getChunkService();
+ processingChunks = chunkService.getAssignedChunks(true);
+ chunkService.acquireJob();
+
+ for (Integer chunkId : processingChunks)
+ {
+ ChangesFilterListsWrapper wrapperPart = (ChangesFilterListsWrapper)data.get(chunkId);
+ if (wrapperPart != null)
+ {
+ wrapper.getAddedNodes().addAll((wrapperPart).getAddedNodes());
+ wrapper.getParentAddedNodes().addAll((wrapperPart).getParentAddedNodes());
+ wrapper.getRemovedNodes().addAll((wrapperPart).getRemovedNodes());
+ wrapper.getParentRemovedNodes().addAll((wrapperPart).getParentRemovedNodes());
+ }
+ }
+ indexer.updateIndex(wrapper.getAddedNodes(), wrapper.getRemovedNodes(), wrapper.getParentAddedNodes(),
+ wrapper.getParentRemovedNodes());
+ }
+ finally
+ {
+ if (chunkService != null)
+ {
+ chunkService.releaseJob();
+ }
+ // removing processed chunk jobs
+ if (processingChunks != null)
+ {
+ for (Integer chunkId : processingChunks)
+ {
+ cache.remove(fqn, chunkId);
+ }
+ }
+ // if no attributes present in JBC node remove it.
+ if (cache.getNode(fqn).getKeys().isEmpty())
+ {
+ cache.removeNode(fqn);
+ }
+ }
+ }
+
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.storage.jbosscache.AbstractWriteOnlyCacheLoader#put(java.util.List)
+ */
+ @Override
+ public void put(List<Modification> modifications) throws Exception
+ {
+ // do nothing. Index is updated on prepare phase.
+ }
+
+ /**
+ * @see org.jboss.cache.loader.CacheLoader#remove(org.jboss.cache.Fqn)
+ */
+ public void remove(Fqn arg0) throws Exception
+ {
+ // do nothing
+ }
+
+ @Override
+ public void removeData(Fqn arg0) throws Exception
+ {
+ log.info("REMOVE DATA: FQN = " + arg0);
+ }
+
+ /**
+ * This class will update the indexes of the related workspace
+ */
+ private static class Indexer
+ {
+
+ private final SearchManager searchManager;
+
+ private final SearchManager parentSearchManager;
+
+ private final QueryHandler handler;
+
+ private final QueryHandler parentHandler;
+
+ private final ChunkService chunkService;
+
+ public Indexer(SearchManager searchManager, SearchManager parentSearchManager, QueryHandler handler,
+ QueryHandler parentHandler, ChunkService chunkService) throws RepositoryConfigurationException
+ {
+ this.searchManager = searchManager;
+ this.parentSearchManager = parentSearchManager;
+ this.handler = handler;
+ this.parentHandler = parentHandler;
+ this.chunkService = chunkService;
+ }
+
+ public ChunkService getChunkService()
+ {
+ return chunkService;
+ }
+
+ /**
+ * Flushes lists of added/removed nodes to SearchManagers, starting indexing.
+ *
+ * @param addedNodes
+ * @param removedNodes
+ * @param parentAddedNodes
+ * @param parentRemovedNodes
+ */
+ protected void updateIndex(Set<String> addedNodes, Set<String> removedNodes, Set<String> parentAddedNodes,
+ Set<String> parentRemovedNodes)
+ {
+ // pass lists to search manager
+ if (searchManager != null && (addedNodes.size() > 0 || removedNodes.size() > 0))
+ {
+ try
+ {
+ searchManager.updateIndex(removedNodes, addedNodes);
+ }
+ catch (RepositoryException e)
+ {
+ log.error("Error indexing changes " + e, e);
+ }
+ catch (IOException e)
+ {
+ log.error("Error indexing changes " + e, e);
+ try
+ {
+ handler.logErrorChanges(removedNodes, addedNodes);
+ }
+ catch (IOException ioe)
+ {
+ log.warn("Exception occure when errorLog writed. Error log is not complete. " + ioe, ioe);
+ }
+ }
+ }
+ // pass lists to parent search manager
+ if (parentSearchManager != null && (parentAddedNodes.size() > 0 || parentRemovedNodes.size() > 0))
+ {
+ try
+ {
+ parentSearchManager.updateIndex(parentRemovedNodes, parentAddedNodes);
+ }
+ catch (RepositoryException e)
+ {
+ log.error("Error indexing changes " + e, e);
+ }
+ catch (IOException e)
+ {
+ log.error("Error indexing changes " + e, e);
+ try
+ {
+ parentHandler.logErrorChanges(removedNodes, addedNodes);
+ }
+ catch (IOException ioe)
+ {
+ log.warn("Exception occure when errorLog writed. Error log is not complete. " + ioe, ioe);
+ }
+ }
+ }
+ }
+ }
+}
Property changes on: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkIndexerCacheLoader.java
___________________________________________________________________
Name: svn:mime-type
+ text/plain
Added: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkService.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkService.java (rev 0)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkService.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -0,0 +1,87 @@
+/*
+ * Copyright (C) 2010 eXo Platform SAS.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk;
+
+import org.exoplatform.services.jcr.impl.core.query.IndexerIoModeHandler;
+
+import java.util.Set;
+
+/**
+ * This service defines all the methods needed to be able to:
+ * <ul>
+ * <li>split the Lucene indexes into several chunks</li>
+ * <li>assign the chunks to a cluster node</li>
+ * <li>assign JCR Node to a chunk</li>
+ * </ul>
+ *
+ * @author Nicolas Filotto
+ * @author <a href="mailto:nikolazius at gmail.com">Nikolay Zamosenchuk</a>
+ * @version $Id: ChunkService.java 34360 2009-07-22 23:58:59Z nzamosenchuk $
+ *
+ */
+public interface ChunkService
+{
+
+ /**
+ * Returns the configured number of index chunks
+ * @return number of chunks
+ */
+ int getChunkCount();
+
+ /**
+ * Gives the id of the chunk corresponding to the given node id
+ * @param nodeId the node Id for which we want to find the corresponding chunk id
+ * @return the corresponding chunk index
+ */
+ int getChunkId(String nodeId);
+
+ /**
+ * TODO review this method
+ * Gives the full list of all the ids of chunks managed by the local machine
+ * @param blocking block current thread is reassigning is in progress
+ * @return the list of chunk ids managed by the local machine
+ */
+ Set<Integer> getAssignedChunks(boolean blocking);
+
+ /**
+ * Returns the instance of {@link IndexerIoModeHandler} managed by {@link ChunkService}. So
+ * {@link ChunkService} instance has an ability to control indexer I/O mode.
+ * @return instance of IndexerIoModeHandler
+ */
+ IndexerIoModeHandler getModeHandler();
+
+ /**
+ * Must be threadsafe! Notifies {@link ChunkService} that one more thread is working
+ * for indexing now. This is a job tracking feature. It is responsible for releasing
+ * and re-acquiring IndexChunks when cluster view changes. {@link ChunkService} must be
+ * aware of indexes that can be released and re-assigned to another JCR instance. After
+ * cluster view changed and no more jobs are currently than Chunks are marked as free.
+ */
+ void acquireJob();
+
+ /**
+ * Must be threadsafe! Notifies {@link ChunkService} that one thread is no more
+ * working for indexing now. This is a job tracking feature. It is responsible for releasing
+ * and re-acquiring IndexChunks when cluster view changes. {@link ChunkService} must be
+ * aware of indexes that can be released and re-assigned to another JCR instance. After
+ * cluster view changed and no more jobs are currently than Chunks are marked as free.
+ */
+ void releaseJob();
+
+}
\ No newline at end of file
Property changes on: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkService.java
___________________________________________________________________
Name: svn:mime-type
+ text/plain
Added: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkServiceImpl.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkServiceImpl.java (rev 0)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkServiceImpl.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -0,0 +1,406 @@
+/*
+ * Copyright (C) 2010 eXo Platform SAS.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk;
+
+import org.exoplatform.services.jcr.impl.core.query.IndexerIoMode;
+import org.exoplatform.services.jcr.impl.core.query.IndexerIoModeHandler;
+import org.exoplatform.services.log.ExoLogger;
+import org.exoplatform.services.log.Log;
+import org.jboss.cache.Cache;
+import org.jboss.cache.CacheSPI;
+import org.jboss.cache.Fqn;
+import org.jboss.cache.Node;
+import org.jboss.cache.notifications.annotation.CacheListener;
+import org.jboss.cache.notifications.annotation.NodeModified;
+import org.jboss.cache.notifications.annotation.ViewChanged;
+import org.jboss.cache.notifications.event.NodeModifiedEvent;
+import org.jboss.cache.notifications.event.NodeRemovedEvent;
+import org.jboss.cache.notifications.event.ViewChangedEvent;
+import org.jboss.cache.notifications.event.NodeModifiedEvent.ModificationType;
+import org.jgroups.Address;
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * ASSIGNING:
+ * 1. {@link ChunkServiceImpl#onClusterViewChanged(ViewChangedEvent)} : notification to start ASSIGNING process, block {@link ChunkServiceImpl#getAssignedChunks()}
+ * 2. wait until all local threads invokes {@link ChunkServiceImpl#releaseJob()}
+ * 3. cleanup corresponding attributes in cache and the whole JBC node if attributes are empty (Chunk<-->ClusterNode attributes).
+ * Set IOMode read only
+ * 4. if JBC node was removed by local machine - ({@link ChunkServiceImpl#onNodeRemoved(NodeRemovedEvent)}) assign chunks, unblock, set IOMode write
+ * 5. Else, if {@link ChunkServiceImpl#onNodeModified(NodeModifiedEvent)} then read map and unblock, set IOMode write
+ *
+ * @author <a href="mailto:nikolazius at gmail.com">Nikolay Zamosenchuk</a>
+ * @version $Id: ChunkServiceImpl.java 34360 2009-07-22 23:58:59Z nzamosenchuk $
+ *
+ */
+ at CacheListener
+public class ChunkServiceImpl implements ChunkService
+{
+ public static final String CHUNK_LIST = "$chunks";
+
+ private static final Log log = ExoLogger.getLogger("exo.jcr.component.core.ChunkService");
+
+ private final IndexerIoModeHandler modeHandler;
+
+ private int chunkNum;
+
+ private final AtomicInteger jobCount = new AtomicInteger(0);
+
+ private final Cache<Serializable, Object> cache;
+
+ private final Fqn rootFqn;
+
+ private ServiceState state;
+
+ private Object reassigninMonitor = new Object();
+
+ private Set<Integer> currentlyAssigned;
+
+ private List<Address> clusterView;
+
+ public ChunkServiceImpl(int chunkNum, Cache<Serializable, Object> cache, Fqn rootFqn)
+ {
+ super();
+ this.chunkNum = chunkNum;
+ this.cache = cache;
+ this.cache.addCacheListener(this);
+ this.modeHandler = new IndexerIoModeHandler(IndexerIoMode.READ_ONLY);
+ this.rootFqn = rootFqn;
+ this.currentlyAssigned = new HashSet<Integer>();
+ this.state = ServiceState.ASSIGNING;
+ clusterView = new ArrayList<Address>(cache.getMembers());
+ Collections.sort(clusterView);
+ // first initial start
+ Map<Serializable, Object> data = cache.getData(Fqn.fromElements(rootFqn, CHUNK_LIST));
+ if (data == null || data.isEmpty())
+ {
+ log.info(rootFqn + "\t creating ChunkSErviceImpl, doing init...");
+ doAssign();
+ }
+ else
+ {
+ log.info(rootFqn + "\t creating ChunkSErviceImpl, skipping init...");
+ if (data.size() == chunkNum)
+ {
+ log.info(rootFqn + "\t creating ChunkSErviceImpl, reading:" + data);
+ Address local = cache.getLocalAddress();
+ // if local address already present in data map - then it has been already assugned
+ for (Object address : data.values())
+ {
+ if (local.equals(address))
+ {
+ readAssigned(data);
+ doFinishAssign();
+ break;
+ }
+ }
+ }
+ }
+ }
+
+ private void doStartAssign(List<Address> members)
+ {
+ // mark index as ASSIGNING
+ // Waiting till all the local jobs finished
+ if (this.state != ServiceState.ASSIGNING)
+ {
+ this.state = ServiceState.ASSIGNING;
+ // cleaning currently assigned chunks
+ log.info(rootFqn.toString() + "\tdoStartAssign. Members = " + members.toString());
+
+ if (((CacheSPI)cache).getRPCManager().isCoordinator())
+ {
+ log.info(rootFqn.toString() + "\tonCLusterViewChanged cleaning suspected member");
+ // cleaning up suspected member
+ if (((CacheSPI)cache).getRPCManager().isCoordinator())
+ {
+ for (Address address : clusterView)
+ {
+ Map<Serializable, Object> data = cache.getData(Fqn.fromElements(rootFqn, CHUNK_LIST));
+ if (!members.contains(address))
+ {
+ // this member is suspected
+ // trying to remove information about chunks used by it
+ for (Integer i = 0; i < chunkNum; i++)
+ {
+ if (data.get(i) != null && data.get(i).equals(address))
+ {
+ cache.remove(Fqn.fromElements(rootFqn, CHUNK_LIST), i);
+ }
+ }
+ }
+ }
+ }
+ }
+
+ if (jobCount.get() == 0)
+ {
+ doReleaseChunks();
+ }
+ }
+ }
+
+ private void doAssign()
+ {
+ List<Address> members = new ArrayList<Address>(cache.getMembers());
+ Collections.sort(members);
+ log.info(rootFqn.toString() + "\tdoAssign. Members = " + members.toString());
+
+ Address localAddress = cache.getLocalAddress();
+
+ if (members.size() == 0)
+ {
+ throw new UnsupportedOperationException("List of memebers is empty. Not yet implemented usecase.");
+ }
+ {
+ log.info(rootFqn.toString() + "\tdoAssign. LocalMachine is coordinator, performing...");
+ // initializing the list of chunks in cache
+ Map<Integer, Address> chunkDistribution = new HashMap<Integer, Address>();
+ int clusterSize = members.size();
+ for (Integer i = 0; i < chunkNum; i++)
+ {
+ chunkDistribution.put(i, members.get(i % clusterSize));
+ if (members.get(i % clusterSize).equals(localAddress))
+ {
+ currentlyAssigned.add(i);
+ }
+ }
+ log.info(rootFqn.toString() + "\tdoAssign. Putting ChunkDistribution = " + chunkDistribution.toString());
+ this.cache.put(Fqn.fromElements(rootFqn, CHUNK_LIST), chunkDistribution);
+ log.info(rootFqn.toString() + "\tdoAssign. Put done ");
+ //this.cache.getNode(Fqn.fromElements(rootFqn, CHUNK_LIST)).setResident(true);
+ doFinishAssign();
+ }
+ }
+
+ private void readAssigned(Map<Serializable, Object> chunkDistribution)
+ {
+ {
+ // read in local variable current chunks
+ Address localAddress = cache.getLocalAddress();
+ // TODO
+ if (currentlyAssigned.isEmpty())
+ {
+ //Map<Serializable, Object> chunkDistribution = cache.getData(Fqn.fromElements(rootFqn, CHUNK_LIST));
+ for (Integer i = 0; i < chunkNum; i++)
+ {
+ if (chunkDistribution.get(i) != null && chunkDistribution.get(i).equals(localAddress))
+ {
+ currentlyAssigned.add(i);
+ }
+ }
+ }
+ }
+ }
+
+ private void doFinishAssign()
+ {
+ log.info(rootFqn.toString() + "\tdoFinishAssign: currentlyAssigned: " + currentlyAssigned);
+ this.state = ServiceState.NORMAL;
+ modeHandler.setMode(IndexerIoMode.READ_WRITE);
+ synchronized (reassigninMonitor)
+ {
+ log.info(rootFqn.toString() + "\tdoFinishAssign. ASSIGNING finished: Notifying");
+ reassigninMonitor.notifyAll();
+ }
+ }
+
+ /**
+ * Closes indexWriters on owning indexes and cleans up corresponding items from map: Map<Chunk, CluserNode>
+ * stored in cache.
+ */
+ private void doReleaseChunks()
+ {
+ // Notifies ChunkIndex to release indexes
+ modeHandler.setMode(IndexerIoMode.READ_ONLY);
+ // clean up Map<Chunk, CluserNode>
+ Iterator<Integer> it = currentlyAssigned.iterator();
+ log.info(rootFqn.toString() + "\tdoReleaseChunks. Removing from info cache...");
+ try
+ {
+ Thread.sleep(4000);
+ }
+ catch (InterruptedException e1)
+ {
+ // TODO Auto-generated catch block
+ e1.printStackTrace();
+ }
+ while (it.hasNext())
+ {
+ Integer chunkId = it.next();
+ Object remove = this.cache.remove(Fqn.fromElements(rootFqn, CHUNK_LIST), chunkId);
+ if (remove == null)
+ {
+ System.out.println("NULL RETURNED BY REMOVE " + rootFqn + " " + chunkId);
+ }
+ }
+ log.info(rootFqn.toString() + "\tdoReleaseChunks. Removed info ");
+ currentlyAssigned.clear();
+ // if this is the last one cluster machine releasing it's chunks - clean the node containing info about chunks.
+ Node listNode = this.cache.getNode(Fqn.fromElements(rootFqn, CHUNK_LIST));
+ if (listNode != null && listNode.getData().isEmpty())
+ {
+ try
+ {
+ log.info(rootFqn.toString() + "\tdoReleaseChunks. Removing node ");
+
+ this.cache.removeNode(Fqn.fromElements(rootFqn, CHUNK_LIST));
+ log.info(rootFqn.toString() + "\tdoReleaseChunks. Removed node ");
+ doAssign();
+ }
+ catch (Exception e)
+ {
+ System.out.println("EXCEPTION FOUND:");
+ e.printStackTrace();
+ }
+ }
+ else
+ {
+ if (listNode != null)
+ {
+ log.info("not removed!!! WS=" + rootFqn + " " + listNode.getData() + " localaddr="
+ + cache.getLocalAddress());
+ }
+ }
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService#getAssignedChunks()
+ */
+ public Set<Integer> getAssignedChunks(boolean blocking)
+ {
+ // Block until chunk transfer not finished
+ if (state != ServiceState.NORMAL && blocking)
+ {
+ synchronized (reassigninMonitor)
+ {
+ log.info(rootFqn.toString() + "\tgetAssignedChunks. Waitning ASSIGNING finished");
+ while (state != ServiceState.NORMAL)
+ {
+ try
+ {
+ reassigninMonitor.wait();
+ }
+ catch (InterruptedException e)
+ {
+ log.error("InterruptedException");
+ }
+ }
+ }
+ log.info(rootFqn.toString() + "\tgetAssignedChunks. ASSIGNING finished: Working");
+ }
+ return new HashSet<Integer>(currentlyAssigned);
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService#getChunkCount()
+ */
+ public int getChunkCount()
+ {
+ return chunkNum;
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService#getChunkId(java.lang.String)
+ */
+ public int getChunkId(String nodeId)
+ {
+ return Math.abs(nodeId.hashCode()) % chunkNum;
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService#acquireJob()
+ */
+ public void acquireJob()
+ {
+ jobCount.incrementAndGet();
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService#releaseJob()
+ */
+ public void releaseJob()
+ {
+ int currentJobs = jobCount.decrementAndGet();
+ if (this.state == ServiceState.ASSIGNING && currentJobs == 0)
+ {
+ // all local threads finished their jobs, so indexes can be released
+ doReleaseChunks();
+ }
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService#getModeHandler()
+ */
+ public IndexerIoModeHandler getModeHandler()
+ {
+ return modeHandler;
+ }
+
+ /**
+ * Listener, receiving the events, when cluster changes
+ * @param e
+ */
+ @ViewChanged
+ public void onClusterViewChanged(ViewChangedEvent e)
+ {
+ List<Address> members = new ArrayList<Address>(e.getNewView().getMembers());
+ Collections.sort(members);
+ if (!e.isPre())
+ {
+ log.info(rootFqn.toString() + "\tonCLusterViewChanged doing start assign");
+ // initiate the process of ASSIGNING
+ doStartAssign(members);
+ }
+ }
+
+ /**
+ * Listener, receiving the events, when node is modified. It handles only events when
+ *
+ * @param e
+ */
+ @NodeModified
+ public void onNodeModified(NodeModifiedEvent e)
+ {
+ // if this modification is PUT_MAP belongs to /WS1/CHUNK_LISTS and current state is ASSIGNING
+ if (!e.isPre() && this.state == ServiceState.ASSIGNING
+ && e.getFqn().equals(Fqn.fromElements(rootFqn, CHUNK_LIST))
+ && e.getModificationType() == ModificationType.PUT_MAP)
+ {
+ // re-assigning is about to finish. Map<Chunk, ClusterNode> already filled
+ readAssigned(e.getData());
+ doFinishAssign();
+ }
+ }
+
+ private enum ServiceState {
+ NORMAL, ASSIGNING;
+ }
+
+}
Property changes on: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/ChunkServiceImpl.java
___________________________________________________________________
Name: svn:mime-type
+ text/plain
Added: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/JBossCacheChunkIndexChangesFilter.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/JBossCacheChunkIndexChangesFilter.java (rev 0)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/JBossCacheChunkIndexChangesFilter.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -0,0 +1,252 @@
+/*
+ * Copyright (C) 2009 eXo Platform SAS.
+ *
+ * This is free software; you can redistribute it and/or modify it
+ * under the terms of the GNU Lesser General Public License as
+ * published by the Free Software Foundation; either version 2.1 of
+ * the License, or (at your option) any later version.
+ *
+ * This software is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
+ * Lesser General Public License for more details.
+ *
+ * You should have received a copy of the GNU Lesser General Public
+ * License along with this software; if not, write to the Free
+ * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
+ * 02110-1301 USA, or see the FSF site: http://www.fsf.org.
+ */
+package org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk;
+
+import org.exoplatform.container.configuration.ConfigurationManager;
+import org.exoplatform.services.jcr.config.QueryHandlerEntry;
+import org.exoplatform.services.jcr.config.RepositoryConfigurationException;
+import org.exoplatform.services.jcr.impl.core.query.IndexerChangesFilter;
+import org.exoplatform.services.jcr.impl.core.query.IndexerIoModeHandler;
+import org.exoplatform.services.jcr.impl.core.query.IndexingTree;
+import org.exoplatform.services.jcr.impl.core.query.QueryHandler;
+import org.exoplatform.services.jcr.impl.core.query.SearchManager;
+import org.exoplatform.services.jcr.impl.core.query.jbosscache.ChangesFilterListsWrapper;
+import org.exoplatform.services.jcr.impl.util.io.PrivilegedCacheHelper;
+import org.exoplatform.services.jcr.jbosscache.ExoJBossCacheFactory;
+import org.exoplatform.services.jcr.jbosscache.ExoJBossCacheFactory.CacheType;
+import org.exoplatform.services.jcr.util.IdGenerator;
+import org.exoplatform.services.log.ExoLogger;
+import org.exoplatform.services.log.Log;
+import org.jboss.cache.Cache;
+import org.jboss.cache.CacheException;
+import org.jboss.cache.CacheSPI;
+import org.jboss.cache.Fqn;
+import org.jboss.cache.config.CacheLoaderConfig;
+import org.jboss.cache.config.CacheLoaderConfig.IndividualCacheLoaderConfig;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Set;
+
+import javax.jcr.RepositoryException;
+
+/**
+ * @author <a href="mailto:Sergey.Kabashnyuk at exoplatform.org">Sergey Kabashnyuk</a>
+ * @version $Id: exo-jboss-codetemplates.xml 34360 2009-07-22 23:58:59Z ksm $
+ *
+ */
+public class JBossCacheChunkIndexChangesFilter extends IndexerChangesFilter
+{
+ /**
+ * Logger instance for this class
+ */
+ private final Log log = ExoLogger.getLogger("exo.jcr.component.core.JBossCacheIndexChangesFilter");
+
+ public static final String PARAM_JBOSSCACHE_CONFIGURATION = "jbosscache-configuration";
+
+ public static final String PARAM_JBOSSCACHE_PUSHSTATE = "jbosscache-sscl-push.state.enabled";
+
+ public static final String PARAM_JBOSSCACHE_PUSHSTATE_TIMEOUT = "jbosscache-sscl-push.state.timeout";
+
+ /**
+ * Indicate whether the JBoss Cache instance used can be shared with other caches
+ */
+ public static final String PARAM_JBOSSCACHE_SHAREABLE = "jbosscache-shareable";
+
+ public static final Boolean PARAM_JBOSSCACHE_SHAREABLE_DEFAULT = Boolean.FALSE;
+
+ private final Cache<Serializable, Object> cache;
+
+ private final Fqn<String> rootFqn;
+
+ public static final String LISTWRAPPER = "$lists".intern();
+
+ private final ChunkService chunkService;
+
+ /**
+ * @param searchManager
+ * @param config
+ * @param indexingTree
+ * @throws RepositoryConfigurationException
+ */
+ public JBossCacheChunkIndexChangesFilter(SearchManager searchManager, SearchManager parentSearchManager,
+ QueryHandlerEntry config, IndexingTree indexingTree, IndexingTree parentIndexingTree, QueryHandler handler,
+ QueryHandler parentHandler, ConfigurationManager cfm) throws IOException, RepositoryException,
+ RepositoryConfigurationException
+ {
+ super(searchManager, parentSearchManager, config, indexingTree, parentIndexingTree, handler, parentHandler, cfm);
+ // create cache using custom factory
+ ExoJBossCacheFactory<Serializable, Object> factory = new ExoJBossCacheFactory<Serializable, Object>(cfm);
+ Cache<Serializable, Object> initCache = factory.createCache(config);
+
+ // initialize IndexerCacheLoader
+ ChunkIndexerCacheLoader indexerCacheLoader = new ChunkIndexerCacheLoader();
+
+ // create CacheLoaderConfig
+ IndividualCacheLoaderConfig individualCacheLoaderConfig = new IndividualCacheLoaderConfig();
+ // set CacheLoader
+ individualCacheLoaderConfig.setCacheLoader(indexerCacheLoader);
+ // set parameters
+ individualCacheLoaderConfig.setFetchPersistentState(false);
+ individualCacheLoaderConfig.setAsync(false);
+ individualCacheLoaderConfig.setIgnoreModifications(false);
+ individualCacheLoaderConfig.setPurgeOnStartup(false);
+ // create CacheLoaderConfig
+ CacheLoaderConfig cacheLoaderConfig = new CacheLoaderConfig();
+ cacheLoaderConfig.setShared(false);
+ cacheLoaderConfig.setPassivation(false);
+ cacheLoaderConfig.addIndividualCacheLoaderConfig(individualCacheLoaderConfig);
+ // insert CacheLoaderConfig
+ initCache.getConfiguration().setCacheLoaderConfig(cacheLoaderConfig);
+ this.rootFqn = Fqn.fromElements(searchManager.getWsId());
+ this.cache =
+ ExoJBossCacheFactory.getUniqueInstance(CacheType.INDEX_CACHE, rootFqn, initCache, config.getParameterBoolean(
+ PARAM_JBOSSCACHE_SHAREABLE, PARAM_JBOSSCACHE_SHAREABLE_DEFAULT));
+
+ PrivilegedCacheHelper.create(cache);
+ PrivilegedCacheHelper.start(cache);
+
+ this.chunkService = new ChunkServiceImpl(4, cache, rootFqn);
+
+ indexerCacheLoader = (ChunkIndexerCacheLoader)((CacheSPI)cache).getCacheLoaderManager().getCacheLoader();
+ indexerCacheLoader.register(searchManager, parentSearchManager, handler, parentHandler, this.chunkService);
+ // TODO Dummy IOModeHandler
+ IndexerIoModeHandler modeHandler = chunkService.getModeHandler(); //new IndexerIoModeHandler(IndexerIoMode.READ_WRITE);
+ handler.setIndexerIoModeHandler(modeHandler);
+ parentHandler.setIndexerIoModeHandler(modeHandler);
+
+ // TODO: review this injection
+ handler.setChunkService(chunkService);
+ parentHandler.setChunkService(chunkService);
+
+ if (!parentHandler.isInitialized())
+ {
+ parentHandler.init();
+ }
+ if (!handler.isInitialized())
+ {
+ handler.init();
+ }
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.IndexerChangesFilter#doUpdateIndex(java.util.Set, java.util.Set, java.util.Set, java.util.Set)
+ */
+ @Override
+ protected void doUpdateIndex(Set<String> removedNodes, Set<String> addedNodes, Set<String> parentRemovedNodes,
+ Set<String> parentAddedNodes)
+ {
+ String id = IdGenerator.generate();
+ try
+ {
+ // this map contains ChangesListWrapper in it's values. Key is the index of Chunk,
+ // starting from 0.
+ Map<Serializable, Object> map = new HashMap<Serializable, Object>();
+
+ // removedNodes
+ for (String uuid : removedNodes)
+ {
+ Integer chunkId = chunkService.getChunkId(uuid);
+ ChangesFilterListsWrapper list = (ChangesFilterListsWrapper)map.get(chunkId);
+ if (list == null)
+ {
+ list =
+ new ChangesFilterListsWrapper(new HashSet<String>(), new HashSet<String>(), new HashSet<String>(),
+ new HashSet<String>());
+ map.put(chunkId, list);
+ }
+ list.getRemovedNodes().add(uuid);
+ }
+
+ // addedNodes
+ for (String uuid : addedNodes)
+ {
+ Integer chunkId = chunkService.getChunkId(uuid);
+ ChangesFilterListsWrapper list = (ChangesFilterListsWrapper)map.get(chunkId);
+ if (list == null)
+ {
+ list =
+ new ChangesFilterListsWrapper(new HashSet<String>(), new HashSet<String>(), new HashSet<String>(),
+ new HashSet<String>());
+ map.put(chunkId, list);
+ }
+ list.getAddedNodes().add(uuid);
+ }
+
+ // parentRemovedNodes
+ for (String uuid : parentRemovedNodes)
+ {
+ Integer chunkId = chunkService.getChunkId(uuid);
+ ChangesFilterListsWrapper list = (ChangesFilterListsWrapper)map.get(chunkId);
+ if (list == null)
+ {
+ list =
+ new ChangesFilterListsWrapper(new HashSet<String>(), new HashSet<String>(), new HashSet<String>(),
+ new HashSet<String>());
+ map.put(chunkId, list);
+ }
+ list.getParentRemovedNodes().add(uuid);
+ }
+
+ // parentAddedNodes
+ for (String uuid : parentAddedNodes)
+ {
+ Integer chunkId = chunkService.getChunkId(uuid);
+ ChangesFilterListsWrapper list = (ChangesFilterListsWrapper)map.get(chunkId);
+ if (list == null)
+ {
+ list =
+ new ChangesFilterListsWrapper(new HashSet<String>(), new HashSet<String>(), new HashSet<String>(),
+ new HashSet<String>());
+ map.put(chunkId, list);
+ }
+ list.getParentAddedNodes().add(uuid);
+ }
+
+ PrivilegedCacheHelper.put(cache, Fqn.fromRelativeElements(rootFqn, id + "MAP"), map);
+ }
+ catch (CacheException e)
+ {
+ log.error(e.getLocalizedMessage(), e);
+ logErrorChanges(handler, removedNodes, addedNodes);
+ logErrorChanges(parentHandler, parentRemovedNodes, parentAddedNodes);
+ }
+ }
+
+ /**
+ * Log errors
+ * @param logHandler
+ * @param removedNodes
+ * @param addedNodes
+ */
+ private void logErrorChanges(QueryHandler logHandler, Set<String> removedNodes, Set<String> addedNodes)
+ {
+ try
+ {
+ logHandler.logErrorChanges(addedNodes, removedNodes);
+ }
+ catch (IOException ioe)
+ {
+ log.warn("Exception occure when errorLog writed. Error log is not complete. " + ioe, ioe);
+ }
+ }
+}
Property changes on: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/jbosscache/chunk/JBossCacheChunkIndexChangesFilter.java
___________________________________________________________________
Name: svn:mime-type
+ text/plain
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChunkIndex.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChunkIndex.java 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/ChunkIndex.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -22,8 +22,11 @@
import org.exoplatform.services.jcr.dataflow.ItemDataConsumer;
import org.exoplatform.services.jcr.datamodel.ItemData;
import org.exoplatform.services.jcr.datamodel.NodeData;
-import org.exoplatform.services.jcr.impl.core.query.ChunkService;
+import org.exoplatform.services.jcr.impl.core.query.IndexerIoMode;
+import org.exoplatform.services.jcr.impl.core.query.IndexerIoModeHandler;
+import org.exoplatform.services.jcr.impl.core.query.IndexerIoModeListener;
import org.exoplatform.services.jcr.impl.core.query.IndexingTree;
+import org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.ChunkService;
import org.exoplatform.services.jcr.impl.core.query.lucene.directory.DirectoryManager;
import org.exoplatform.services.jcr.impl.util.SecurityHelper;
import org.slf4j.Logger;
@@ -47,7 +50,7 @@
/**
* TODO REWRITE JAVADOC!!!
*/
-public class ChunkIndex
+public class ChunkIndex implements IndexerIoModeListener
{
/**
@@ -116,6 +119,8 @@
private final ChunkService chunkService;
+ private final IndexerIoModeHandler modeHandler;
+
/**
* Creates a new MultiIndex.
*
@@ -127,7 +132,8 @@
* @throws IOException
* if an error occurs
*/
- ChunkIndex(SearchIndex handler, IndexingTree indexingTree) throws IOException
+ ChunkIndex(SearchIndex handler, IndexingTree indexingTree, IndexerIoModeHandler modeHandler,
+ ChunkService chunkService) throws IOException
{
this.directoryManager = handler.getDirectoryManager();
// this method is run in privileged mode internally
@@ -137,8 +143,26 @@
this.indexingTree = indexingTree;
this.nsMappings = handler.getNamespaceMappings();
// TODO remove this stub
- this.chunkService = new ChunkServiceImpl();
+ this.chunkService = chunkService;
+ if (chunkService == null)
+ {
+ System.out.println("\n\n\n");
+ System.out.println("null");
+ System.out.println("\n\n\n");
+ try
+ {
+ throw new Exception();
+ }
+ catch (Exception e)
+ {
+ e.printStackTrace();
+ }
+ }
+
+ this.modeHandler = modeHandler;
+ this.modeHandler.addIndexerIoModeListener(this);
+
// this method is run in privileged mode internally
IndexingQueueStore store = new IndexingQueueStore(indexDir);
@@ -147,7 +171,7 @@
// open persistent indexes
- Set<Integer> assignedChunks = chunkService.getAssignedChunks();
+ Set<Integer> assignedChunks = chunkService.getAssignedChunks(false);
for (int i = 0; i < chunkService.getChunkCount(); i++)
{
@@ -166,7 +190,7 @@
PersistentIndexChunk index =
new PersistentIndexChunk(i, handler.getTextAnalyzer(), handler.getSimilarity(), cache, indexingQueue,
directoryManager, !assignedChunks.contains(Integer.valueOf(i)));
-
+
index.setMaxFieldLength(handler.getMaxFieldLength());
index.setUseCompoundFile(handler.getUseCompoundFile());
index.setTermInfosIndexDivisor(handler.getTermInfosIndexDivisor());
@@ -519,6 +543,7 @@
*/
public synchronized void update(final Collection<String> remove, final Collection<Document> add) throws IOException
{
+ log.info("Indexing: added " + add.size() + " and removed " + remove.size());
SecurityHelper.doPriviledgedIOExceptionAction(new PrivilegedExceptionAction<Object>()
{
public Object run() throws Exception
@@ -563,10 +588,14 @@
}
}
}
- // TODO for owning indexes only
+ // commit owning indexes only
+ Set<Integer> chunkIds = chunkService.getAssignedChunks(false);
for (PersistentIndexChunk idx : indexes)
{
- idx.commit();
+ if (chunkIds.contains(Integer.valueOf(idx.getId())))
+ {
+ idx.commit();
+ }
}
}
finally
@@ -579,7 +608,7 @@
}
/**
- * Checks the indexing queue for finished text extrator jobs and updates the
+ * Checks the indexing queue for finished text extractor jobs and updates the
* index accordingly if there are any new ones.
*
* @param transactionPresent
@@ -745,4 +774,38 @@
}
});
}
+
+ private synchronized void doReleaseChunks()
+ {
+ for (PersistentIndexChunk idx : indexes)
+ {
+ idx.setReadOnly(true);
+ }
+ }
+
+ private synchronized void doAcquireChunks()
+ {
+ Set<Integer> chunkIds = chunkService.getAssignedChunks(false);
+ for (Integer id : chunkIds)
+ {
+ indexes.get(id).setReadOnly(false);
+ }
+ }
+
+ /**
+ * @see org.exoplatform.services.jcr.impl.core.query.IndexerIoModeListener#onChangeMode(org.exoplatform.services.jcr.impl.core.query.IndexerIoMode)
+ */
+ public void onChangeMode(IndexerIoMode mode)
+ {
+ switch (mode)
+ {
+ case READ_ONLY :
+ doReleaseChunks();
+ break;
+ case READ_WRITE :
+ doAcquireChunks();
+ break;
+ }
+
+ }
}
\ No newline at end of file
Added: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/PersistentIndexChunk.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/PersistentIndexChunk.java (rev 0)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/PersistentIndexChunk.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -0,0 +1,132 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.exoplatform.services.jcr.impl.core.query.lucene;
+
+import org.apache.lucene.analysis.Analyzer;
+import org.apache.lucene.document.Document;
+import org.apache.lucene.index.IndexWriter;
+import org.apache.lucene.index.Term;
+import org.apache.lucene.search.Similarity;
+import org.exoplatform.services.jcr.impl.core.query.lucene.directory.DirectoryManager;
+
+import java.io.IOException;
+
+/**
+ * TODO Write javadoc. This index extends Abstract with checks, if index is
+ * owned by this machine
+ * Implements a lucene index which is based on a
+ * {@link org.apache.jackrabbit.core.fs.FileSystem}.
+ */
+class PersistentIndexChunk extends AbstractIndex
+{
+
+ // TODO review checks. Some more methods should check if index is read only.
+
+ /** The Id of this persistent index */
+ private final int id;
+
+ private boolean readOnly;
+
+ /**
+ * Creates a new <code>PersistentIndex</code>.
+ *
+ * @param name the name of this index.
+ * @param analyzer the analyzer for text tokenizing.
+ * @param similarity the similarity implementation.
+ * @param cache the document number cache
+ * @param indexingQueue the indexing queue.
+ * @param directoryManager the directory manager.
+ * @param readOnly if current index is not owned by local machine
+ * @throws IOException if an error occurs while opening / creating the
+ * index.
+ */
+ PersistentIndexChunk(int id, Analyzer analyzer, Similarity similarity, DocNumberCache cache,
+ IndexingQueue indexingQueue, final DirectoryManager directoryManager, boolean readOnly) throws IOException
+ {
+ super(analyzer, similarity, directoryManager.getDirectory("chunk_" + id), cache, indexingQueue);
+ this.id = id;
+ this.readOnly = readOnly;
+ }
+
+ @Override
+ void addDocuments(Document[] docs) throws IOException
+ {
+ if (readOnly)
+ {
+ throw new IOException("Index is not owned by current instance of ChunkIndex and can't be modified!");
+ }
+ super.addDocuments(docs);
+ }
+
+ @Override
+ protected void commit() throws IOException
+ {
+ if (readOnly)
+ {
+ throw new IOException("Index is not owned by current instance of ChunkIndex and can't be modified!");
+ }
+ super.commit();
+ }
+
+ @Override
+ protected synchronized void commit(boolean optimize) throws IOException
+ {
+ if (readOnly)
+ {
+ throw new IOException("Index is not owned by current instance of ChunkIndex and can't be modified!");
+ }
+ super.commit(optimize);
+ }
+
+ @Override
+ protected synchronized IndexWriter getIndexWriter() throws IOException
+ {
+ if (readOnly)
+ {
+ throw new IOException("Index is not owned by current instance of ChunkIndex and can't be modified!");
+ }
+ return super.getIndexWriter();
+ }
+
+ @Override
+ int removeDocument(Term idTerm) throws IOException
+ {
+ if (readOnly)
+ {
+ throw new IOException("Index is not owned by current instance of ChunkIndex and can't be modified!");
+ }
+ return super.removeDocument(idTerm);
+ }
+
+ /**
+ * Returns the name of this index.
+ * @return the name of this index.
+ */
+ int getId()
+ {
+ return id;
+ }
+
+ public void setReadOnly(boolean readOnly)
+ {
+ this.readOnly = readOnly;
+ if (readOnly)
+ {
+ releaseWriterAndReaders();
+ }
+ }
+}
Property changes on: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/PersistentIndexChunk.java
___________________________________________________________________
Name: svn:mime-type
+ text/plain
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/core/query/lucene/SearchIndex.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -570,7 +570,7 @@
indexingConfig = createIndexingConfiguration(nsMappings);
analyzer.setIndexingConfig(indexingConfig);
- index = new ChunkIndex(this, context.getIndexingTree());
+ index = new ChunkIndex(this, context.getIndexingTree(), modeHandler, chunkService);
// if RW mode, create initial index and start check
if (modeHandler.getMode() == IndexerIoMode.READ_WRITE)
{
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/storage/jbosscache/AbstractWriteOnlyCacheLoader.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/storage/jbosscache/AbstractWriteOnlyCacheLoader.java 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/storage/jbosscache/AbstractWriteOnlyCacheLoader.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -74,6 +74,7 @@
/**
* {@inheritDoc}
*/
+ @Override
public void prepare(Object tx, List<Modification> modifications, boolean onePhase) throws Exception
{
throw new WriteOnlyCacheLoaderException("The method 'prepare(Object tx, List<Modification> modifications, boolean onePhase)' should not be called.");
@@ -84,7 +85,8 @@
*/
public Object remove(Fqn arg0, Object arg1) throws Exception
{
- throw new WriteOnlyCacheLoaderException("The method 'remove(Fqn arg0, Object arg1)' should not be called.");
+// throw new WriteOnlyCacheLoaderException("The method 'remove(Fqn arg0, Object arg1)' should not be called.");
+ return null;
}
/**
@@ -106,6 +108,7 @@
/**
* {@inheritDoc}
*/
+ @Override
public abstract void put(List<Modification> modifications) throws Exception ;
}
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/util/io/PrivilegedCacheHelper.java
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/util/io/PrivilegedCacheHelper.java 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/main/java/org/exoplatform/services/jcr/impl/util/io/PrivilegedCacheHelper.java 2010-11-02 07:25:50 UTC (rev 3373)
@@ -27,6 +27,7 @@
import java.security.PrivilegedAction;
import java.security.PrivilegedActionException;
import java.security.PrivilegedExceptionAction;
+import java.util.Map;
/**
* @author <a href="anatoliy.bazko at exoplatform.org">Anatoliy Bazko</a>
@@ -163,4 +164,42 @@
}
}
}
+
+ /**
+ * Put in cache in privileged mode.
+ *
+ * @param cache
+ */
+ public static void put(final Cache<Serializable, Object> cache, final Fqn fqn, final Map<Serializable, Object> data)
+ throws CacheException
+ {
+ PrivilegedExceptionAction<Object> action = new PrivilegedExceptionAction<Object>()
+ {
+ public Object run() throws Exception
+ {
+ cache.put(fqn, data);
+ return null;
+ }
+ };
+ try
+ {
+ AccessController.doPrivileged(action);
+ }
+ catch (PrivilegedActionException pae)
+ {
+ Throwable cause = pae.getCause();
+ if (cause instanceof CacheException)
+ {
+ throw (CacheException)cause;
+ }
+ else if (cause instanceof RuntimeException)
+ {
+ throw (RuntimeException)cause;
+ }
+ else
+ {
+ throw new RuntimeException(cause);
+ }
+ }
+ }
}
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-indexer.xml
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-indexer.xml 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jbosscache-indexer.xml 2010-11-02 07:25:50 UTC (rev 3373)
@@ -4,18 +4,18 @@
<locking useLockStriping="false" concurrencyLevel="500" lockParentForChildInsertRemove="false"
lockAcquisitionTimeout="20000" />
<!-- Configure the TransactionManager -->
- <transaction transactionManagerLookupClass="org.jboss.cache.transaction.JBossStandaloneJTAManagerLookup" />
+ <!-- transaction transactionManagerLookupClass="org.jboss.cache.transaction.JBossStandaloneJTAManagerLookup" /-->
<clustering mode="replication" clusterName="${jbosscache-cluster-name}">
- <stateRetrieval timeout="20000" fetchInMemoryState="false" />
- <sync />
+ <stateRetrieval timeout="20000" fetchInMemoryState="true" />
+ <sync replTimeout = "30000"/>
</clustering>
<!-- Eviction configuration -->
- <eviction wakeUpInterval="5000">
+ <!--eviction wakeUpInterval="5000">
<default algorithmClass="org.jboss.cache.eviction.FIFOAlgorithm" eventQueueSize="1000000">
<property name="maxNodes" value="10000" />
<property name="minTimeToLive" value="60000" />
</default>
- </eviction>
+ </eviction-->
</jbosscache>
Modified: jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jcr-config.xml
===================================================================
--- jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jcr-config.xml 2010-11-01 10:50:31 UTC (rev 3372)
+++ jcr/branches/1.14-CNK/exo.jcr.component.core/src/test/resources/conf/standalone/cluster/test-jcr-config.xml 2010-11-02 07:25:50 UTC (rev 3373)
@@ -59,7 +59,7 @@
<properties>
<property name="index-dir" value="target/temp/index/db1/ws" />
<property name="changesfilter-class"
- value="org.exoplatform.services.jcr.impl.core.query.jbosscache.JBossCacheIndexChangesFilter" />
+ value="org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.JBossCacheChunkIndexChangesFilter" />
<property name="jbosscache-configuration" value="test-jbosscache-indexer.xml" />
<property name="jgroups-configuration" value="udp-mux.xml" />
<property name="jgroups-multiplexer-stack" value="false" />
@@ -123,7 +123,7 @@
<properties>
<property name="index-dir" value="target/temp/index/db1/ws1" />
<property name="changesfilter-class"
- value="org.exoplatform.services.jcr.impl.core.query.jbosscache.JBossCacheIndexChangesFilter" />
+ value="org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.JBossCacheChunkIndexChangesFilter" />
<property name="jbosscache-configuration" value="test-jbosscache-indexer.xml" />
<property name="jgroups-configuration" value="udp-mux.xml" />
<property name="jgroups-multiplexer-stack" value="false" />
@@ -197,7 +197,7 @@
<properties>
<property name="index-dir" value="target/temp/index/db1/ws2" />
<property name="changesfilter-class"
- value="org.exoplatform.services.jcr.impl.core.query.jbosscache.JBossCacheIndexChangesFilter" />
+ value="org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.JBossCacheChunkIndexChangesFilter" />
<property name="jbosscache-configuration" value="test-jbosscache-indexer.xml" />
<property name="jgroups-configuration" value="udp-mux.xml" />
<property name="jgroups-multiplexer-stack" value="false" />
@@ -285,7 +285,7 @@
<properties>
<property name="index-dir" value="target/temp/index/db1/ws3" />
<property name="changesfilter-class"
- value="org.exoplatform.services.jcr.impl.core.query.jbosscache.JBossCacheIndexChangesFilter" />
+ value="org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.JBossCacheChunkIndexChangesFilter" />
<property name="jbosscache-configuration" value="test-jbosscache-indexer.xml" />
<property name="jgroups-configuration" value="udp-mux.xml" />
<property name="jgroups-multiplexer-stack" value="false" />
@@ -378,7 +378,7 @@
<properties>
<property name="index-dir" value="target/temp/index/db1tck/ws" />
<property name="changesfilter-class"
- value="org.exoplatform.services.jcr.impl.core.query.jbosscache.JBossCacheIndexChangesFilter" />
+ value="org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.JBossCacheChunkIndexChangesFilter" />
<property name="jbosscache-configuration" value="test-jbosscache-indexer.xml" />
<property name="jgroups-configuration" value="udp-mux.xml" />
<property name="jgroups-multiplexer-stack" value="false" />
@@ -459,7 +459,7 @@
<properties>
<property name="index-dir" value="target/temp/index/db1tck/ws1" />
<property name="changesfilter-class"
- value="org.exoplatform.services.jcr.impl.core.query.jbosscache.JBossCacheIndexChangesFilter" />
+ value="org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.JBossCacheChunkIndexChangesFilter" />
<property name="jbosscache-configuration" value="test-jbosscache-indexer.xml" />
<property name="jgroups-configuration" value="udp-mux.xml" />
<property name="jgroups-multiplexer-stack" value="false" />
@@ -538,7 +538,7 @@
<properties>
<property name="index-dir" value="target/temp/index/db1tck/ws2" />
<property name="changesfilter-class"
- value="org.exoplatform.services.jcr.impl.core.query.jbosscache.JBossCacheIndexChangesFilter" />
+ value="org.exoplatform.services.jcr.impl.core.query.jbosscache.chunk.JBossCacheChunkIndexChangesFilter" />
<property name="jbosscache-configuration" value="test-jbosscache-indexer.xml" />
<property name="jgroups-configuration" value="udp-mux.xml" />
<property name="jgroups-multiplexer-stack" value="false" />
More information about the exo-jcr-commits
mailing list