* Only one save writes the state at a time: two of them would otherwise each * take their own snapshot, and the write of the older one landing last would * leave a state on disk that is both stale and marked as saved. *
* A save which finds another one writing gives up its turn instead of waiting * for it, and this method never blocks. Waiting would close a lock cycle: * when the write goes to the domain configuration entry it ends up in * {@code LDAPReplicationDomain.applyConfigurationChange()}, which takes the * very lock {@code disable()} holds while calling this method. *
* Giving up the turn loses nothing, because the state is marked as saved
* before the snapshot of the write in flight is taken. Whatever this save
* would have written is therefore either already in that snapshot, or has
* cleared the flag again after it was set - in which case the flag is still
* clear when that write completes, and the next save writes it.
*/
public void save()
{
if (state.isSaved())
{
// Nothing to write: stay out of the way of whoever is writing.
return;
}
if (!saveLock.tryLock())
{
return;
}
try
{
if (state.isSaved())
{
// The save which just completed carried what this one came to write.
return;
}
/*
* Mark the state as saved before the snapshot that goes to the backend
* is taken, so that an update landing while the write is in flight
* clears the flag again and gets written by the next save. Marking it
* afterwards would swallow such an update: it is not part of the write
* it raced with, yet the state would look saved. The persisted state
* would then stay stale until some later update happened to dirty it
* again - which, on a domain as quiet as cn=schema, may never happen.
*/
state.setSaved(true);
boolean written = false;
try
{
written = updateStateEntry();
}
finally
{
if (!written)
{
// The write reported a failure, or blew up on its way to the
// backend: the state is not on disk, so leave it to the next save.
state.setSaved(false);
}
}
}
finally
{
saveLock.unlock();
}
}
/**
* Load the ServerState from the backing entry in database to memory.
*/
public void loadState()
{
/*
* Whatever the state holds on the way in has, as far as this object knows,
* never been written: what follows only merges in what the backend holds.
* No shipped path comes in holding anything - the constructor is handed the
* state a ReplicationDomain has just created, and loadDataState() empties
* it first - so this guards a caller which does not exist yet rather than
* one which does.
*/
final boolean hadCSNs = !state.isEmpty();
// try to load the state from the base entry.
SearchResultEntry stateEntry = searchBaseEntry();
if (stateEntry == null)
{
/*
The base entry does not exist yet in the database or was deleted.
Try to read the ServerState from the configuration instead.
*/
stateEntry = searchConfigEntry();
}
if (stateEntry != null)
{
updateStateFromEntry(stateEntry);
}
/*
* In order to make sure that the replication never looses changes,
* the server needs to search all the entries that have been
* updated after the last write of the ServerState.
* Inconsistencies may append after a crash.
*/
checkAndUpdateServerState();
if (hadCSNs)
{
state.setSaved(false);
}
}
/**
* Run a search operation to find the base entry
* of the replication domain for which this ServerState was created.
*
* @return The base entry or null if no entry was found;
*/
private SearchResultEntry searchBaseEntry()
{
// Search the database entry that is used to periodically save the ServerState
final SearchRequest request = newSearchRequest(baseDN, SearchScope.BASE_OBJECT).addAttribute(REPLICATION_STATE);
final InternalSearchOperation search = getRootConnection().processSearch(request);
final ResultCode resultCode = search.getResultCode();
if (resultCode != ResultCode.SUCCESS
&& resultCode != ResultCode.NO_SUCH_OBJECT)
{
logger.error(ERR_ERROR_SEARCHING_RUV, search.getResultCode().getName(), search, search.getErrorMessage(), baseDN);
return null;
}
return getFirstResult(search);
}
/**
* Run a search operation to find the entry with the configuration
* of the replication domain for which this ServerState was created.
*
* @return The configuration Entry or null if no entry was found;
*/
private SearchResultEntry searchConfigEntry()
{
try
{
String filter = "(&(objectclass=ds-cfg-replication-domain)" + "(ds-cfg-base-dn=" + baseDN + "))";
final SearchRequest request = newSearchRequest("cn=config", SearchScope.SUBORDINATES, filter)
.setSizeLimit(1)
.addAttribute(REPLICATION_STATE);
final InternalSearchOperation op = getRootConnection().processSearch(request);
return getFirstResult(op);
}
catch (DirectoryException e)
{
// can not happen
return null;
}
}
private SearchResultEntry getFirstResult(InternalSearchOperation search)
{
if (search.getResultCode() == ResultCode.SUCCESS)
{
final LinkedList
* Kept separate, and overridable, so that a test can reach the point where
* the snapshot has been taken but the write has not gone through yet: see
* {@code PersistentServerStateTest}. Do not inline it.
*
* @param op The modify operation carrying the state to be written.
* @return A ResultCode indicating if the operation was successful.
*/
ResultCode runModify(ModifyOperationBasis op)
{
op.run();
return op.getResultCode();
}
/**
* Empty the ServerState in memory.
*
* The emptied state is marked as not saved, so the next save writes the empty
* state out - which is what {@link #clear()} is after. A caller that only
* means to drop the in-memory copy, and expects the backend to keep what it
* holds, has to keep saves away until it has loaded the state back.
*/
public void clearInMemory()
{
state.clear();
}
/**
* Empty the ServerState and write the emptied state to persistent storage.
*/
void clear()
{
clearInMemory();
save();
}
/**
* The ServerState is saved to the database periodically,
* therefore in case of crash it is possible that is does not contain
* the latest changes that have been processed and saved to the
* database.
* In order to make sure that we don't loose them, search all the entries
* that have been updated after this entry.
* This is done by using the HistoricalCsnOrderingMatchingRule
* and an ordering index for historical attribute
*/
private final void checkAndUpdateServerState()
{
// Retrieves the entries that have changed since the
// maxCsn stored in the serverState
synchronized (this)
{
CSN serverStateMaxCSN = state.getCSN(serverId);
if (serverStateMaxCSN == null)
{
return;
}
InternalSearchOperation op;
try
{
op = LDAPReplicationDomain.searchForChangedEntries(baseDN,
serverStateMaxCSN, null);
}
catch (Exception e)
{
return;
}
if (op.getResultCode() != ResultCode.SUCCESS)
{
// An error happened trying to search for the updates
// Log an error
logger.error(ERR_CANNOT_RECOVER_CHANGES, baseDN);
return;
}
CSN dbMaxCSN = serverStateMaxCSN;
for (SearchResultEntry resEntry : op.getSearchEntries())
{
for (ByteString attrValue : resEntry.getAllAttributes(HISTORICAL_ATTRIBUTE_NAME).iterator().next())
{
HistoricalAttributeValue histVal =
new HistoricalAttributeValue(attrValue.toString());
CSN csn = histVal.getCSN();
if (csn != null
&& csn.getServerId() == serverId
&& dbMaxCSN.isOlderThan(csn))
{
dbMaxCSN = csn;
}
}
}
if (dbMaxCSN.isNewerThan(serverStateMaxCSN))
{
// Update the serverState with the new maxCsn present in the database
update(dbMaxCSN);
logger.info(NOTE_SERVER_STATE_RECOVERY, baseDN, dbMaxCSN);
}
}
}
/**
* Get the largest CSN seen for a given LDAP server ID.
*
* @param serverId
* The serverId
* @return The largest CSN seen
*/
public CSN getMaxCSN(int serverId)
{
return state.getCSN(serverId);
}
/** {@inheritDoc} */
@Override
public String toString()
{
return getClass().getSimpleName()
+ " baseDN=" + baseDN
+ " serverId=" + serverId
+ " " + REPLICATION_STATE + "=" + state;
}
}