/* * CDDL HEADER START * * The contents of this file are subject to the terms of the * Common Development and Distribution License, Version 1.0 only * (the "License"). You may not use this file except in compliance * with the License. * * You can obtain a copy of the license at * trunk/opends/resource/legal-notices/OpenDS.LICENSE * or https://OpenDS.dev.java.net/OpenDS.LICENSE. * See the License for the specific language governing permissions * and limitations under the License. * * When distributing Covered Code, include this CDDL HEADER in each * file and include the License file at * trunk/opends/resource/legal-notices/OpenDS.LICENSE. If applicable, * add the following below this CDDL HEADER, with the fields enclosed * by brackets "[]" replaced with your own identifying information: * Portions Copyright [yyyy] [name of copyright owner] * * CDDL HEADER END * * * Copyright 2009-2010 Sun Microsystems, Inc. */ package org.opends.server.replication.server; import static org.opends.messages.ReplicationMessages.*; import static org.opends.server.loggers.ErrorLogger.logError; import static org.opends.server.loggers.debug.DebugLogger.debugEnabled; import static org.opends.server.loggers.debug.DebugLogger.getTracer; import static org.opends.server.util.StaticUtils.stackTraceToSingleLineString; import java.util.ArrayList; import java.util.concurrent.locks.ReentrantLock; import org.opends.messages.MessageBuilder; import org.opends.server.admin.std.server.MonitorProviderCfg; import org.opends.server.api.DirectoryThread; import org.opends.server.api.MonitorProvider; import org.opends.server.config.ConfigException; import org.opends.server.core.DirectoryServer; import org.opends.server.loggers.debug.DebugTracer; import org.opends.server.replication.common.ChangeNumber; import org.opends.server.replication.common.ServerState; import org.opends.server.replication.server.DraftCNDB.DraftCNDBCursor; import org.opends.server.types.Attribute; import org.opends.server.types.Attributes; import org.opends.server.types.InitializationException; import com.sleepycat.je.DatabaseException; import com.sleepycat.je.LockConflictException; /** * This class is used for managing the replicationServer database for each * server in the topology. * It is responsible for efficiently saving the updates that is received from * each master server into stable storage. * This class is also able to generate a ReplicationIterator that can be * used to read all changes from a given ChangeNUmber. * * This class publish some monitoring information below cn=monitor. * */ public class DraftCNDbHandler implements Runnable { /** * The tracer object for the debug logger. */ private static final DebugTracer TRACER = getTracer(); // A dedicated thread loops trim(). // trim() : deletes from the DB a number of changes that are older than a // certain date. // static int NO_KEY = 0; private DraftCNDB db; private int firstkey = NO_KEY; private int lastkey = NO_KEY; private DbMonitorProvider dbMonitor = new DbMonitorProvider(); private boolean shutdown = false; private boolean trimDone = false; private DirectoryThread thread = null; private ReplicationServer replicationServer; // The maximum number of retries in case of DatabaseDeadlock Exception. private static final int DEADLOCK_RETRIES = 10; /** * * The trim age in milliseconds. Changes record in the change DB that * are older than this age are removed. * */ private long trimage; /** * Creates a new dbHandler associated to a given LDAP server. * * @param replicationServer The ReplicationServer that creates this dbHandler. * @param dbenv the Database Env to use to create the ReplicationServer DB. * server for this domain. * @throws DatabaseException If a database problem happened */ public DraftCNDbHandler(ReplicationServer replicationServer, ReplicationDbEnv dbenv) throws DatabaseException { this.replicationServer = replicationServer; this.trimage = replicationServer.getTrimage(); // DB initialization db = new DraftCNDB(replicationServer, dbenv); firstkey = db.readFirstDraftCN(); lastkey = db.readLastDraftCN(); // Triming thread thread = new DirectoryThread(this, "Replication DraftCN db "); thread.start(); // Monitoring registration DirectoryServer.deregisterMonitorProvider( dbMonitor.getMonitorInstanceName()); DirectoryServer.registerMonitorProvider(dbMonitor); } /** * Add an update to the list of messages that must be saved to the db * managed by this db handler. * This method is blocking if the size of the list of message is larger * than its maximum. * @param key The key for this record in the db. * @param value The associated value. * @param serviceID The associated serviceID. * @param cn The associated replication change number. * */ public synchronized void add(int key, String value, String serviceID, ChangeNumber cn) { db.addEntry(key, value, serviceID, cn); if (debugEnabled()) TRACER.debugInfo( "In DraftCNDbhandler.add, added: " + " key=" + key + " value=" + value + " serviceID=" + serviceID + " cn=" + cn); } /** * Get the firstChange. * @return Returns the firstChange. */ public int getFirstKey() { return db.readFirstDraftCN(); } /** * Get the lastChange. * @return Returns the lastChange. */ public int getLastKey() { return db.readLastDraftCN(); } /** * Get the number of changes. * @return Returns the number of changes. */ public long count() { return db.count(); } /** * Get a read cursor on the database from a provided key. * The cursor MUST be released after use. * @param key The provided key. * @return the new cursor. */ public DraftCNDBCursor getReadCursor(int key) { try { return db.openReadCursor(key); } catch(Exception e) { return null; } } /** * Release a provided read cursor. * @param cursor The provided read cursor. */ public void releaseReadCursor(DraftCNDBCursor cursor) { try { cursor.close(); } catch(Exception e) { } } /** * Generate a new ReplicationIterator that allows to browse the db * managed by this dbHandler and starting at the position defined * by a given changeNumber. * * @param startDraftCN The position where the iterator must start. * * @return a new ReplicationIterator that allows to browse the db * managed by this dbHandler and starting at the position defined * by a given changeNumber. * * @throws DatabaseException if a database problem happened. * @throws Exception If there is no other change to push after change * with changeNumber number. */ public DraftCNDbIterator generateIterator(int startDraftCN) throws DatabaseException, Exception { DraftCNDbIterator it = new DraftCNDbIterator(db, startDraftCN); return it; } /** * Shutdown this dbHandler. */ public void shutdown() { if (shutdown == true) { return; } shutdown = true; synchronized (this) { this.notifyAll(); } synchronized (this) { while (trimDone == false) { try { this.wait(); } catch (Exception e) {} } } db.shutdown(); DirectoryServer.deregisterMonitorProvider( dbMonitor.getMonitorInstanceName()); } /** * Run method for this class. * Periodically Flushes the ReplicationServerDomain cache from memory to the * stable storage and trims the old updates. */ public void run() { while (shutdown == false) { try { trim(); synchronized (this) { try { this.wait(1000); } catch (InterruptedException e) { } } } catch (Exception end) { MessageBuilder mb = new MessageBuilder(); mb.append(ERR_EXCEPTION_CHANGELOG_TRIM_FLUSH.get()); mb.append(stackTraceToSingleLineString(end)); logError(mb.toMessage()); if (replicationServer != null) replicationServer.shutdown(); break; } } synchronized (this) { trimDone = true; this.notifyAll(); } } /** * Trim old changes from this database. * @throws DatabaseException In case of database problem. * @throws Exception In case of database problem. */ public void trim() throws DatabaseException, Exception { if (trimage == 0) return; clear(null); } /** * Clear the changes from this DB (from both memory cache and DB storage) * for the provided serviceID. * @param serviceIDToClear The serviceID for which we want to remove * all records from the DraftCNDb - null means all. * @throws DatabaseException When an exception occurs while removing the * changes from the DB. * @throws Exception When an exception occurs while accessing a resource * from the DB. * */ public void clear(String serviceIDToClear) throws DatabaseException, Exception { if (this.count()==0) return; int size = 0; int tries = 0; boolean finished = false; boolean done = false; ChangeNumber crossDomainEligibleCN = replicationServer.getEligibleCN(); // In case of deadlock detection by the Database, this thread can // by aborted by a DeadlockException. This is a transient error and // the transaction should be attempted again. // We will try DEADLOCK_RETRIES times before failing. while ((tries++ < DEADLOCK_RETRIES) && (!done)) { DraftCNDBCursor cursor = db.openDeleteCursor(); try { while ((size < 5000 ) && (!finished)) { // let's traverse the DraftCNDb if (!cursor.next()) { finished=true; } else { ChangeNumber cn = cursor.currentChangeNumber(); // From the draftCNDb change record, get the domain and changeNumber String serviceID = cursor.currentServiceID(); if ((serviceIDToClear!=null) && (serviceIDToClear.equalsIgnoreCase(serviceID))) { size++; cursor.delete(); continue; } ReplicationServerDomain domain = replicationServer.getReplicationServerDomain(serviceID, false); if (domain==null) { // the domain has been removed since the record was written in the // draftCNDb, thus it makes no sense to keep the record in the // draftCNDb. size++; cursor.delete(); } else { // let's get the eligible part of the domain ServerState startSS = domain.getStartState(); ServerState endSS= domain.getEligibleState(crossDomainEligibleCN, true); ChangeNumber fcn = startSS.getMaxChangeNumber(cn.getServerId()); ChangeNumber lcn = endSS.getMaxChangeNumber(cn.getServerId()); // if the draftCNDb change record, is out of the eligible part // of the domain, then it can be removed. if (cn.older(fcn)||cn.newer(lcn)) { size++; cursor.delete(); } } } } cursor.close(); done = true; } catch (LockConflictException e) { cursor.abort(); if (tries == DEADLOCK_RETRIES) { // could not handle the Deadlock after DEADLOCK_RETRIES tries. // shutdown the ReplicationServer. shutdown = true; throw (e); } } catch (DatabaseException e) { // mark shutdown for this db so that we don't try again to // stop it from cursor.close() or methods called by cursor.close() shutdown = true; cursor.abort(); throw (e); } } } /** * This internal class is used to implement the Monitoring capabilities * of the dbHandler. */ private class DbMonitorProvider extends MonitorProvider { /** * {@inheritDoc} */ @Override public ArrayList getMonitorData() { ArrayList attributes = new ArrayList(); attributes.add(Attributes.create("first-draft-changenumber", Integer.toString(db.readFirstDraftCN()))); attributes.add(Attributes.create("last-draft-changenumber", Integer.toString(db.readLastDraftCN()))); attributes.add(Attributes.create("count", Long.toString(count()))); return attributes; } /** * {@inheritDoc} */ @Override public String getMonitorInstanceName() { return "ReplicationServer DraftCN database "; } /** * {@inheritDoc} */ @Override public void initializeMonitorProvider(MonitorProviderCfg configuration) throws ConfigException,InitializationException { // Nothing to do for now } } /** * {@inheritDoc} */ @Override public String toString() { return("draftCNdb:" + " " + firstkey + " " + lastkey); } /** * Set the Purge delay for this db Handler. * @param delay The purge delay in Milliseconds. */ public void setPurgeDelay(long delay) { trimage = delay; } /** * Clear the changes from this DB (from both memory cache and DB storage). * @throws DatabaseException When an exception occurs while removing the * changes from the DB. * @throws Exception When an exception occurs while accessing a resource * from the DB. * */ public void clear() throws DatabaseException, Exception { db.clear(); firstkey = db.readFirstDraftCN(); lastkey = db.readLastDraftCN(); } private ReentrantLock lock = new ReentrantLock(); /** * Tests if the current thread has the lock on this object. * @return True if the current thread has the lock. */ public boolean hasLock() { return (lock.getHoldCount() > 0); } /** * Takes the lock on this object (blocking until lock can be acquired). * @throws java.lang.InterruptedException If interrupted. */ public void lock() throws InterruptedException { lock.lockInterruptibly(); } /** * Releases the lock on this object. */ public void release() { lock.unlock(); } /** * Get the value associated to a provided key. * @param key the provided key. * @return the associated value, null when none. */ public String getValue(int key) { String value = null; DraftCNDBCursor draftCNDBCursor = null; try { draftCNDBCursor = db.openReadCursor(key); value = draftCNDBCursor.currentValue(); } catch(Exception e) { if (debugEnabled()) TRACER.debugInfo("In DraftCNDbHandler.getValue, read: " + " key=" + key + " value returned is null" + " first=" + db.readFirstDraftCN() + " last=" + db.readLastDraftCN() + " count=" + db.count() + " exception" + e + " " + e.getMessage()); return null; } finally { if (draftCNDBCursor != null) draftCNDBCursor.close(); } return value; } /** * Get the CN associated to a provided key. * @param key the provided key. * @return the associated CN, null when none. */ public ChangeNumber getChangeNumber(int key) { ChangeNumber cn = null; DraftCNDBCursor draftCNDBCursor = null; try { draftCNDBCursor = db.openReadCursor(key); cn = draftCNDBCursor.currentChangeNumber(); } catch(Exception e) { if (debugEnabled()) TRACER.debugInfo("In DraftCNDbHandler.getChangeNumber, read: " + " key=" + key + " changeNumber returned is null" + " first=" + db.readFirstDraftCN() + " last=" + db.readLastDraftCN() + " count=" + db.count() + " exception" + e + " " + e.getMessage()); return null; } finally { if (draftCNDBCursor != null) draftCNDBCursor.close(); } return cn; } /** * Get the serviceID associated to a provided key. * @param key the provided key. * @return the serviceID, null when none. */ public String getServiceID(int key) { String sid = null; DraftCNDBCursor draftCNDBCursor = null; try { draftCNDBCursor = db.openReadCursor(key); sid = draftCNDBCursor.currentServiceID(); } catch(Exception e) { if (debugEnabled()) TRACER.debugInfo("In DraftCNDbHandler.getServiceID, read: " + " key=" + key + " serviceID returned is null" + " first=" + db.readFirstDraftCN() + " last=" + db.readLastDraftCN() + " count=" + db.count() + " exception" + e + " " + e.getMessage()); return null; } finally { if (draftCNDBCursor != null) draftCNDBCursor.close(); } return sid; } }