/*
|
* The contents of this file are subject to the terms of the Common Development and
|
* Distribution License (the License). You may not use this file except in compliance with the
|
* License.
|
*
|
* You can obtain a copy of the License at legal/CDDLv1.0.txt. See the License for the
|
* specific language governing permission and limitations under the License.
|
*
|
* When distributing Covered Software, include this CDDL Header Notice in each file and include
|
* the License file at legal/CDDLv1.0.txt. If applicable, add the following below the CDDL
|
* Header, with the fields enclosed by brackets [] replaced by your own identifying
|
* information: "Portions Copyright [year] [name of copyright owner]".
|
*
|
* Copyright 2015-2016 ForgeRock AS.
|
* Portions Copyright 2026 3A Systems, LLC.
|
*/
|
package org.opends.server.backends.jeb;
|
|
import static com.sleepycat.je.EnvironmentConfig.*;
|
import static com.sleepycat.je.LockMode.READ_COMMITTED;
|
import static com.sleepycat.je.LockMode.RMW;
|
import static com.sleepycat.je.OperationStatus.*;
|
|
import static org.forgerock.util.Utils.*;
|
import static org.opends.messages.BackendMessages.*;
|
import static org.opends.messages.UtilityMessages.*;
|
import static org.opends.server.backends.pluggable.spi.StorageUtils.*;
|
import static org.opends.server.util.StaticUtils.*;
|
|
import java.io.File;
|
import java.io.FileFilter;
|
import java.io.IOException;
|
import java.nio.file.Files;
|
import java.nio.file.Path;
|
import java.util.ArrayList;
|
import java.util.Collections;
|
import java.util.HashMap;
|
import java.util.HashSet;
|
import java.util.List;
|
import java.util.ListIterator;
|
import java.util.Map;
|
import java.util.NoSuchElementException;
|
import java.util.Objects;
|
import java.util.Set;
|
import java.util.TreeMap;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.ConcurrentMap;
|
import java.util.concurrent.TimeUnit;
|
|
import org.forgerock.i18n.LocalizableMessage;
|
import org.forgerock.i18n.slf4j.LocalizedLogger;
|
import org.forgerock.opendj.config.server.ConfigChangeResult;
|
import org.forgerock.opendj.config.server.ConfigException;
|
import org.forgerock.opendj.ldap.ByteSequence;
|
import org.forgerock.opendj.ldap.ByteString;
|
import org.forgerock.opendj.ldap.ResultCode;
|
import org.forgerock.util.Reject;
|
import org.forgerock.opendj.config.server.ConfigurationChangeListener;
|
import org.forgerock.opendj.server.config.server.JEBackendCfg;
|
import org.opends.server.api.Backupable;
|
import org.opends.server.api.DiskSpaceMonitorHandler;
|
import org.opends.server.backends.pluggable.spi.EmptyCursor;
|
import org.opends.server.backends.pluggable.spi.AccessMode;
|
import org.opends.server.backends.pluggable.spi.Cursor;
|
import org.opends.server.backends.pluggable.spi.Importer;
|
import org.opends.server.backends.pluggable.spi.ReadOnlyStorageException;
|
import org.opends.server.backends.pluggable.spi.ReadOperation;
|
import org.opends.server.backends.pluggable.spi.SequentialCursor;
|
import org.opends.server.backends.pluggable.spi.Storage;
|
import org.opends.server.backends.pluggable.spi.StorageRuntimeException;
|
import org.opends.server.backends.pluggable.spi.StorageStatus;
|
import org.opends.server.backends.pluggable.spi.StorageUtils;
|
import org.opends.server.backends.pluggable.spi.TreeName;
|
import org.opends.server.backends.pluggable.spi.UpdateFunction;
|
import org.opends.server.backends.pluggable.spi.WriteOperation;
|
import org.opends.server.backends.pluggable.spi.WriteableTransaction;
|
import org.opends.server.core.DirectoryServer;
|
import org.opends.server.core.MemoryQuota;
|
import org.opends.server.core.ServerContext;
|
import org.opends.server.extensions.DiskSpaceMonitor;
|
import org.opends.server.types.BackupConfig;
|
import org.opends.server.types.BackupDirectory;
|
import org.opends.server.types.DirectoryException;
|
import org.opends.server.types.RestoreConfig;
|
import org.opends.server.util.BackupManager;
|
|
import com.sleepycat.je.CursorConfig;
|
import com.sleepycat.je.Database;
|
import com.sleepycat.je.DatabaseConfig;
|
import com.sleepycat.je.DatabaseEntry;
|
import com.sleepycat.je.DatabaseException;
|
import com.sleepycat.je.DatabaseNotFoundException;
|
import com.sleepycat.je.Durability;
|
import com.sleepycat.je.Environment;
|
import com.sleepycat.je.EnvironmentConfig;
|
import com.sleepycat.je.LockConflictException;
|
import com.sleepycat.je.OperationStatus;
|
import com.sleepycat.je.Transaction;
|
import com.sleepycat.je.TransactionConfig;
|
import com.sleepycat.je.config.ConfigParam;
|
import com.sleepycat.je.config.EnvironmentParams;
|
|
/** Berkeley DB Java Edition (JE for short) database implementation of the {@link Storage} engine. */
|
public final class JEStorage implements Storage, Backupable, ConfigurationChangeListener<JEBackendCfg>,
|
DiskSpaceMonitorHandler
|
{
|
/** JE implementation of the {@link Cursor} interface. */
|
private static final class CursorImpl implements Cursor<ByteString, ByteString>
|
{
|
private ByteString currentKey;
|
private ByteString currentValue;
|
private boolean isDefined;
|
private final com.sleepycat.je.Cursor cursor;
|
private final DatabaseEntry dbKey = new DatabaseEntry();
|
private final DatabaseEntry dbValue = new DatabaseEntry();
|
|
private CursorImpl(com.sleepycat.je.Cursor cursor)
|
{
|
this.cursor = cursor;
|
}
|
|
@Override
|
public void close()
|
{
|
closeSilently(cursor);
|
}
|
|
@Override
|
public boolean isDefined()
|
{
|
return isDefined;
|
}
|
|
@Override
|
public ByteString getKey()
|
{
|
if (currentKey == null)
|
{
|
throwIfNotSuccess();
|
currentKey = ByteString.wrap(dbKey.getData());
|
}
|
return currentKey;
|
}
|
|
@Override
|
public ByteString getValue()
|
{
|
if (currentValue == null)
|
{
|
throwIfNotSuccess();
|
currentValue = ByteString.wrap(dbValue.getData());
|
}
|
return currentValue;
|
}
|
|
@Override
|
public boolean next()
|
{
|
clearCurrentKeyAndValue();
|
try
|
{
|
isDefined = cursor.getNext(dbKey, dbValue, null) == SUCCESS;
|
return isDefined;
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public void delete() throws NoSuchElementException, UnsupportedOperationException
|
{
|
throwIfNotSuccess();
|
try
|
{
|
cursor.delete();
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public boolean positionToKey(final ByteSequence key)
|
{
|
clearCurrentKeyAndValue();
|
setData(dbKey, key);
|
try
|
{
|
isDefined = cursor.getSearchKey(dbKey, dbValue, null) == SUCCESS;
|
return isDefined;
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public boolean positionToKeyOrNext(final ByteSequence key)
|
{
|
clearCurrentKeyAndValue();
|
setData(dbKey, key);
|
try
|
{
|
isDefined = cursor.getSearchKeyRange(dbKey, dbValue, null) == SUCCESS;
|
return isDefined;
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public boolean positionToIndex(int index)
|
{
|
clearCurrentKeyAndValue();
|
try
|
{
|
isDefined = cursor.getFirst(dbKey, dbValue, null) == SUCCESS;
|
if (!isDefined)
|
{
|
return false;
|
}
|
else if (index == 0)
|
{
|
return true;
|
}
|
|
// equivalent to READ_UNCOMMITTED
|
long skipped = cursor.skipNext(index, dbKey, dbValue, null);
|
if (skipped == index)
|
{
|
isDefined = cursor.getCurrent(dbKey, dbValue, null) == SUCCESS;
|
}
|
else
|
{
|
isDefined = false;
|
}
|
return isDefined;
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public boolean positionToLastKey()
|
{
|
clearCurrentKeyAndValue();
|
try
|
{
|
isDefined = cursor.getLast(dbKey, dbValue, null) == SUCCESS;
|
return isDefined;
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
private void clearCurrentKeyAndValue()
|
{
|
currentKey = null;
|
currentValue = null;
|
}
|
|
private void throwIfNotSuccess()
|
{
|
if (!isDefined())
|
{
|
throw new NoSuchElementException();
|
}
|
}
|
}
|
|
/** JE implementation of the {@link Importer} interface. */
|
private final class ImporterImpl implements Importer
|
{
|
private final Map<TreeName, Database> trees = new HashMap<>();
|
|
private Database getOrOpenTree(TreeName treeName)
|
{
|
return getOrOpenTree0(trees, treeName);
|
}
|
|
@Override
|
public void put(final TreeName treeName, final ByteSequence key, final ByteSequence value)
|
{
|
try
|
{
|
getOrOpenTree(treeName).put(null, db(key), db(value));
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public ByteString read(final TreeName treeName, final ByteSequence key)
|
{
|
try
|
{
|
DatabaseEntry dbValue = new DatabaseEntry();
|
boolean isDefined = getOrOpenTree(treeName).get(null, db(key), dbValue, null) == SUCCESS;
|
return valueToBytes(dbValue, isDefined);
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public SequentialCursor<ByteString, ByteString> openCursor(TreeName treeName)
|
{
|
try
|
{
|
return new CursorImpl(getOrOpenTree(treeName).openCursor(null, new CursorConfig()));
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public void clearTree(TreeName treeName)
|
{
|
env.truncateDatabase(null, toDatabaseName(treeName), false);
|
}
|
|
@Override
|
public void close()
|
{
|
closeSilently(trees.values());
|
trees.clear();
|
JEStorage.this.close();
|
}
|
}
|
|
/** JE implementation of the {@link WriteableTransaction} interface. */
|
private final class WriteableTransactionImpl implements WriteableTransaction
|
{
|
private final Transaction txn;
|
|
private WriteableTransactionImpl(Transaction txn)
|
{
|
this.txn = txn;
|
}
|
|
/**
|
* This is currently needed for import-ldif:
|
* <ol>
|
* <li>Opening the EntryContainer calls {@link #openTree(TreeName, boolean)} for each index</li>
|
* <li>Then the underlying storage is closed</li>
|
* <li>Then {@link #startImport()} is called</li>
|
* <li>Then ID2Entry#put() is called</li>
|
* <li>Which in turn calls ID2Entry#encodeEntry()</li>
|
* <li>Which in turn finally calls PersistentCompressedSchema#store()</li>
|
* <li>Which uses a reference to the storage (that was closed before calling startImport()) and
|
* uses it as if it was open</li>
|
* </ol>
|
*/
|
private Database getOrOpenTree(TreeName treeName)
|
{
|
try
|
{
|
return getOrOpenTree0(trees, treeName);
|
}
|
catch (Exception e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public void put(final TreeName treeName, final ByteSequence key, final ByteSequence value)
|
{
|
try
|
{
|
final OperationStatus status = getOrOpenTree(treeName).put(txn, db(key), db(value));
|
if (status != SUCCESS)
|
{
|
throw new StorageRuntimeException(putErrorMsg(treeName, key, value, "did not succeed: " + status));
|
}
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(putErrorMsg(treeName, key, value, "threw an exception"), e);
|
}
|
}
|
|
private String putErrorMsg(TreeName treeName, ByteSequence key, ByteSequence value, String msg)
|
{
|
return "put(treeName=" + treeName + ", key=" + key + ", value=" + value + ") " + msg;
|
}
|
|
@Override
|
public boolean delete(final TreeName treeName, final ByteSequence key)
|
{
|
try
|
{
|
return getOrOpenTree(treeName).delete(txn, db(key)) == SUCCESS;
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(deleteErrorMsg(treeName, key, "threw an exception"), e);
|
}
|
}
|
|
private String deleteErrorMsg(TreeName treeName, ByteSequence key, String msg)
|
{
|
return "delete(treeName=" + treeName + ", key=" + key + ") " + msg;
|
}
|
|
@Override
|
public long getRecordCount(TreeName treeName)
|
{
|
try
|
{
|
return getOrOpenTree(treeName).count();
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public boolean treeExists(TreeName treeName)
|
{
|
// Deliberately not getOrOpenTree(): dbConfig() sets allowCreate, so asking the tree itself
|
// would create the very database whose absence is being tested.
|
try
|
{
|
return env.getDatabaseNames().contains(toDatabaseName(treeName));
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public Cursor<ByteString, ByteString> openCursor(final TreeName treeName)
|
{
|
try
|
{
|
return new CursorImpl(getOrOpenTree(treeName).openCursor(txn, CursorConfig.READ_COMMITTED));
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public ByteString read(final TreeName treeName, final ByteSequence key)
|
{
|
try
|
{
|
DatabaseEntry dbValue = new DatabaseEntry();
|
boolean isDefined = getOrOpenTree(treeName).get(txn, db(key), dbValue, READ_COMMITTED) == SUCCESS;
|
return valueToBytes(dbValue, isDefined);
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public boolean update(final TreeName treeName, final ByteSequence key, final UpdateFunction f)
|
{
|
try
|
{
|
final Database tree = getOrOpenTree(treeName);
|
final DatabaseEntry dbKey = db(key);
|
final DatabaseEntry dbValue = new DatabaseEntry();
|
for (;;)
|
{
|
final boolean isDefined = tree.get(txn, dbKey, dbValue, RMW) == SUCCESS;
|
final ByteSequence oldValue = valueToBytes(dbValue, isDefined);
|
final ByteSequence newValue = f.computeNewValue(oldValue);
|
if (Objects.equals(newValue, oldValue))
|
{
|
return false;
|
}
|
if (newValue == null)
|
{
|
return tree.delete(txn, dbKey) == SUCCESS;
|
}
|
setData(dbValue, newValue);
|
if (isDefined)
|
{
|
return tree.put(txn, dbKey, dbValue) == SUCCESS;
|
}
|
else if (tree.putNoOverwrite(txn, dbKey, dbValue) == SUCCESS)
|
{
|
return true;
|
}
|
// else retry due to phantom read: another thread inserted a record
|
}
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public void openTree(final TreeName treeName, boolean createOnDemand)
|
{
|
getOrOpenTree(treeName);
|
}
|
|
@Override
|
public void deleteTree(final TreeName treeName)
|
{
|
try
|
{
|
synchronized (trees)
|
{
|
closeSilently(trees.remove(treeName));
|
env.removeDatabase(txn, toDatabaseName(treeName));
|
}
|
}
|
catch (DatabaseNotFoundException e)
|
{
|
// This is fine: end result is what we wanted
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
}
|
|
/** JE read-only implementation of {@link WriteableTransaction} interface. */
|
private static final class ReadOnlyTransactionImpl implements WriteableTransaction
|
{
|
private final WriteableTransactionImpl delegate;
|
|
ReadOnlyTransactionImpl(WriteableTransactionImpl delegate)
|
{
|
this.delegate = delegate;
|
}
|
|
@Override
|
public ByteString read(TreeName treeName, ByteSequence key)
|
{
|
return delegate.read(treeName, key);
|
}
|
|
@Override
|
public Cursor<ByteString, ByteString> openCursor(TreeName treeName)
|
{
|
return delegate.openCursor(treeName);
|
}
|
|
@Override
|
public long getRecordCount(TreeName treeName)
|
{
|
return delegate.getRecordCount(treeName);
|
}
|
|
@Override
|
public boolean treeExists(TreeName treeName)
|
{
|
return delegate.treeExists(treeName);
|
}
|
|
@Override
|
public void openTree(TreeName treeName, boolean createOnDemand)
|
{
|
if (createOnDemand)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
delegate.openTree(treeName, false);
|
}
|
|
@Override
|
public void deleteTree(TreeName name)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
|
@Override
|
public void put(TreeName treeName, ByteSequence key, ByteSequence value)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
|
@Override
|
public boolean update(TreeName treeName, ByteSequence key, UpdateFunction f)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
|
@Override
|
public boolean delete(TreeName treeName, ByteSequence key)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
}
|
|
/** No operation storage transaction faking database files are present and empty. */
|
private static final class ReadOnlyEmptyTransactionImpl implements WriteableTransaction
|
{
|
@Override
|
public void openTree(TreeName name, boolean createOnDemand)
|
{
|
if (createOnDemand)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
}
|
|
@Override
|
public void deleteTree(TreeName name)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
|
@Override
|
public void put(TreeName treeName, ByteSequence key, ByteSequence value)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
|
@Override
|
public boolean update(TreeName treeName, ByteSequence key, UpdateFunction f)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
|
@Override
|
public boolean delete(TreeName treeName, ByteSequence key)
|
{
|
throw new ReadOnlyStorageException();
|
}
|
|
@Override
|
public ByteString read(TreeName treeName, ByteSequence key)
|
{
|
return null;
|
}
|
|
@Override
|
public Cursor<ByteString, ByteString> openCursor(TreeName treeName)
|
{
|
return new EmptyCursor<>();
|
}
|
|
@Override
|
public long getRecordCount(TreeName treeName)
|
{
|
return 0;
|
}
|
|
@Override
|
public boolean treeExists(TreeName treeName)
|
{
|
// No environment was ever opened, so nothing is stored.
|
return false;
|
}
|
}
|
|
private WriteableTransaction newWriteableTransaction(Transaction txn)
|
{
|
// If no database files have been created yet and we're opening READ-ONLY
|
// there is no db to use, since open was not called. Fake it.
|
if (env == null)
|
{
|
return new ReadOnlyEmptyTransactionImpl();
|
}
|
final WriteableTransactionImpl writeableStorage = new WriteableTransactionImpl(txn);
|
return accessMode.isWriteable() ? writeableStorage : new ReadOnlyTransactionImpl(writeableStorage);
|
}
|
|
private static final int IMPORT_DB_CACHE_SIZE = 32 * MB;
|
|
/**
|
* Number of attempts a {@link #write} makes before it propagates the conflict to the caller.
|
* <p>
|
* It is a budget of attempts and not of time, so it is only ever reached by the conflicts that report quickly.
|
* With the shipped configuration every conflict does: {@code je.lock.timeout} is 0, so a writer waiting for a
|
* lock never times out, and the only conflict JE raises is the deadlock, which it detects as soon as the wait
|
* would close a cycle. An operator who sets a lock timeout through {@code ds-cfg-je-property} turns a wait that
|
* long into a conflict as well, and a conflict slower to report than {@link #MAX_RETRY_WINDOW_NANOS} spends the
|
* whole window inside its first attempt, is granted the single replay that window's exemption guarantees, and
|
* gives up on the window after two attempts rather than after this many.
|
*/
|
static final int MAX_RETRIES = 10;
|
|
/**
|
* Wall-clock budget the replays of a {@link #write} may spend, in nanoseconds. It is checked between attempts,
|
* so an attempt already running is never interrupted, and never before one replay has been made: the loop
|
* returns after at most this window plus two attempts. It bounds the conflicts that are slow to report, which
|
* {@link #MAX_RETRIES} alone does not - an operation whose own work takes seconds would otherwise multiply that
|
* wait by the attempt count.
|
*/
|
static final long MAX_RETRY_WINDOW_NANOS = 10L * 1000L * 1000L * 1000L; //10 s
|
|
/**
|
* Upper bound of the random delay before the second attempt, in milliseconds; it doubles with every attempt.
|
* The ladder is {@code PDBStorage}'s, kept so that the two engines back off alike.
|
*/
|
private static final double BASE_SLEEP_ON_RETRY_MS = 50.0;
|
|
/** Upper bound the doubled delay is capped at, in milliseconds. */
|
private static final double MAX_SLEEP_ON_RETRY_MS = 1000.0;
|
|
private static final LocalizedLogger logger = LocalizedLogger.getLoggerForThisClass();
|
|
/** Use read committed isolation instead of the default which is repeatable read. */
|
private static final TransactionConfig TXN_READ_COMMITTED = new TransactionConfig().setReadCommitted(true);
|
|
private final ServerContext serverContext;
|
private final File backendDirectory;
|
/**
|
* The mode last written to the directory the storage runs on. A mode changed along with a move of
|
* db-directory is written to the directory moved to alone, so the configuration as last changed
|
* does not say what the running directory has.
|
*/
|
private String runningDirectoryPermissions;
|
private JEBackendCfg config;
|
private AccessMode accessMode;
|
|
/** It is NULL when opening the storage READ-ONLY and no files have been created yet. */
|
private Environment env;
|
private EnvironmentConfig envConfig;
|
private MemoryQuota memQuota;
|
/**
|
* The cache size of the configuration this storage opened with, in bytes - what the memory quota
|
* was asked for - and of it, what the quota granted, which is what {@link #close()} gives back.
|
* Both are zero while the storage is closed. Neither is read from {@link #config} again: a
|
* configuration change replaces that while the environment and the reservation stay as the open
|
* made them, so a release computed from it would give back a size that was never taken. For a
|
* cache sized by db-cache-percent this is the quota's count, a percent of its reservable pool, and
|
* not the cache JE runs: JE takes that percent of the maximum heap.
|
*/
|
private long configuredCacheSize;
|
private long reservedCacheSize;
|
private JEMonitor monitor;
|
private DiskSpaceMonitor diskMonitor;
|
private StorageStatus storageStatus = StorageStatus.working();
|
private final ConcurrentMap<TreeName, Database> trees = new ConcurrentHashMap<>();
|
/** Attempt bound of a {@link #write}, {@link #MAX_RETRIES} outside the tests. */
|
private final int maxRetries;
|
/** Wall-clock bound of a {@link #write}, {@link #MAX_RETRY_WINDOW_NANOS} outside the tests. */
|
private final long retryWindowNanos;
|
|
/**
|
* Creates a new JE storage with the provided configuration.
|
*
|
* @param cfg
|
* The configuration.
|
* @param serverContext
|
* This server instance context
|
* @throws ConfigException
|
* if memory cannot be reserved
|
*/
|
// Public as PDBStorage's is: a pluggable backend test which runs the same case over both storages builds them.
|
public JEStorage(final JEBackendCfg cfg, ServerContext serverContext) throws ConfigException
|
{
|
this(cfg, serverContext, MAX_RETRIES, MAX_RETRY_WINDOW_NANOS);
|
}
|
|
/**
|
* Creates a new JE storage whose replay bounds are the given ones rather than {@link #MAX_RETRIES} and
|
* {@link #MAX_RETRY_WINDOW_NANOS}.
|
* <p>
|
* Only a test builds one of these, and it does so to stop the two bounds racing each other: with the shipped
|
* values a run of replays spends a random share of the window on backoff alone, so a test of the attempt cap
|
* can be ended by the window on a loaded machine, and a test of the window has to spend seconds of build time
|
* to reach it.
|
*
|
* @param cfg
|
* The configuration.
|
* @param serverContext
|
* This server instance context
|
* @param maxRetries
|
* Number of attempts a write makes before it propagates the conflict to the caller.
|
* @param retryWindowNanos
|
* Wall-clock budget the replays of a write may spend, in nanoseconds.
|
* @throws ConfigException
|
* if memory cannot be reserved
|
*/
|
JEStorage(final JEBackendCfg cfg, ServerContext serverContext, int maxRetries, long retryWindowNanos)
|
throws ConfigException
|
{
|
this.serverContext = serverContext;
|
this.maxRetries = maxRetries;
|
this.retryWindowNanos = retryWindowNanos;
|
backendDirectory = getBackendDirectory(cfg);
|
runningDirectoryPermissions = cfg.getDBDirectoryPermissions();
|
config = cfg;
|
cfg.addJEChangeListener(this);
|
}
|
|
private Database getOrOpenTree0(Map<TreeName, Database> trees, TreeName treeName)
|
{
|
Database tree = trees.get(treeName);
|
if (tree == null)
|
{
|
synchronized (trees)
|
{
|
tree = trees.get(treeName);
|
if (tree == null)
|
{
|
tree = env.openDatabase(null, toDatabaseName(treeName), dbConfig());
|
trees.put(treeName, tree);
|
}
|
}
|
}
|
return tree;
|
}
|
|
private void buildConfiguration(AccessMode accessMode, boolean isImport) throws ConfigException
|
{
|
this.accessMode = accessMode;
|
|
if (isImport)
|
{
|
envConfig = new EnvironmentConfig();
|
envConfig
|
.setTransactional(false)
|
.setAllowCreate(true)
|
.setLockTimeout(0, TimeUnit.SECONDS)
|
.setTxnTimeout(0, TimeUnit.SECONDS)
|
.setCacheSize(IMPORT_DB_CACHE_SIZE)
|
.setDurability(Durability.COMMIT_NO_SYNC)
|
.setConfigParam(CLEANER_MIN_UTILIZATION, String.valueOf(config.getDBCleanerMinUtilization()))
|
.setConfigParam(LOG_FILE_MAX, String.valueOf(config.getDBLogFileMax()))
|
.setConfigParam("je.freeDisk",String.valueOf(50*1024*1024));
|
}
|
else
|
{
|
envConfig = ConfigurableEnvironment.parseConfigEntry(config);
|
}
|
|
diskMonitor = serverContext.getDiskSpaceMonitor();
|
memQuota = serverContext.getMemoryQuota();
|
configuredCacheSize = computeSize(config);
|
// A reservation the quota refuses - its budget spent by the other backends, which an open at
|
// startup is not checked against - is nothing to give back: the open goes ahead without it.
|
reservedCacheSize = memQuota.acquireMemory(configuredCacheSize) ? configuredCacheSize : 0;
|
}
|
|
private DatabaseConfig dbConfig()
|
{
|
boolean isImport = !envConfig.getTransactional();
|
return new DatabaseConfig()
|
.setKeyPrefixing(true)
|
.setAllowCreate(true)
|
.setTransactional(!isImport)
|
.setDeferredWrite(isImport);
|
}
|
|
@Override
|
public void close()
|
{
|
synchronized (trees)
|
{
|
closeSilently(trees.values());
|
trees.clear();
|
}
|
|
try
|
{
|
if (env != null)
|
{
|
// Not yet registered when a failed open got no further than the environment itself.
|
if (monitor != null)
|
{
|
DirectoryServer.deregisterMonitorProvider(monitor);
|
monitor = null;
|
}
|
env.close();
|
env = null;
|
}
|
}
|
catch (DatabaseException e)
|
{
|
throw new IllegalStateException(e);
|
}
|
finally
|
{
|
// Given back after the environment, and whether or not its close threw: reporting the
|
// memory free before the environment has actually freed it would let a racing enable of
|
// another backend be admitted while this one's cache is still resident.
|
if (memQuota != null)
|
{
|
// What the open reserved, not what the configuration says by now: a cache size changed
|
// while the storage was open is applied by the next open, which reserves it then.
|
memQuota.releaseMemory(reservedCacheSize);
|
reservedCacheSize = 0;
|
configuredCacheSize = 0;
|
// Released once: what an open takes, the next open takes again, and a close which follows
|
// a close - BackendImpl.importLDIF closes the storage of its root container however the
|
// import ended, on top of the close the import itself made - releases nothing more.
|
memQuota = null;
|
}
|
config.removeJEChangeListener(this);
|
envConfig = null;
|
if (diskMonitor != null)
|
{
|
diskMonitor.deregisterMonitoredDirectory(getDirectory(), this);
|
}
|
}
|
}
|
|
@Override
|
public void open(AccessMode accessMode) throws ConfigException, StorageRuntimeException
|
{
|
Reject.ifNull(accessMode, "accessMode must not be null");
|
if (isBackendIncomplete(accessMode))
|
{
|
envConfig = new EnvironmentConfig();
|
envConfig.setAllowCreate(false).setTransactional(false).setConfigParam("je.freeDisk",String.valueOf(50*1024*1024));
|
// Do not open files on disk
|
return;
|
}
|
rejectIfOpen();
|
buildConfiguration(accessMode, false);
|
openOrGiveBack();
|
}
|
|
/**
|
* Refuses to open an environment which is open, before anything is taken for the attempt: the
|
* refusal guards against a programming error, and what this storage holds is left as it is.
|
*/
|
private void rejectIfOpen()
|
{
|
if (env != null)
|
{
|
throw new IllegalStateException(
|
"Database is already open, either the backend is enabled or an import is currently running.");
|
}
|
}
|
|
/**
|
* Opens the environment, or gives back what the attempt took before it failed. Nothing else will:
|
* a root container does not close a storage whose {@code open()} threw, and a backend whose open
|
* failed is thrown away with the storage still registered as a listener of its configuration and
|
* the cache size it reserved still drawn from the memory quota - once per attempt to enable it.
|
*/
|
private void openOrGiveBack() throws ConfigException
|
{
|
boolean opened = false;
|
try
|
{
|
open0();
|
opened = true;
|
}
|
finally
|
{
|
if (!opened)
|
{
|
try
|
{
|
close();
|
}
|
catch (RuntimeException e)
|
{
|
// The failure being given up after is the one worth reporting, and this must not replace it.
|
logger.traceException(e);
|
}
|
}
|
}
|
}
|
|
private boolean isBackendIncomplete(AccessMode accessMode)
|
{
|
return !accessMode.isWriteable() && (!backendDirectory.exists() || backendDirectoryIncomplete());
|
}
|
|
// TODO: it belongs to disk-based Storage Interface.
|
private boolean backendDirectoryIncomplete()
|
{
|
try
|
{
|
return !getFilesToBackup().hasNext();
|
}
|
catch (DirectoryException ignored)
|
{
|
return true;
|
}
|
}
|
|
private void open0() throws ConfigException
|
{
|
setupStorageFiles(backendDirectory, config.getDBDirectoryPermissions(), config.dn());
|
runningDirectoryPermissions = config.getDBDirectoryPermissions();
|
try
|
{
|
env = new Environment(backendDirectory, envConfig);
|
monitor = new JEMonitor(config.getBackendId() + " JE Database", env);
|
DirectoryServer.registerMonitorProvider(monitor);
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
registerMonitoredDirectory(config);
|
}
|
|
@Override
|
public <T> T read(final ReadOperation<T> operation) throws Exception
|
{
|
try
|
{
|
return operation.run(newWriteableTransaction(null));
|
}
|
catch (final StorageRuntimeException e)
|
{
|
if (e.getCause() != null)
|
{
|
throw (Exception) e.getCause();
|
}
|
throw e;
|
}
|
}
|
|
@Override
|
public Importer startImport() throws ConfigException, StorageRuntimeException
|
{
|
rejectIfOpen();
|
buildConfiguration(AccessMode.READ_WRITE, true);
|
openOrGiveBack();
|
return new ImporterImpl();
|
}
|
|
private static String toDatabaseName(final TreeName treeName)
|
{
|
return treeName.toString();
|
}
|
|
/**
|
* {@inheritDoc}
|
* <p>
|
* A transaction JE ends with a {@link LockConflictException} is replayed, bounded twice: by {@link #MAX_RETRIES}
|
* attempts and by the {@link #MAX_RETRY_WINDOW_NANOS} wall-clock window, whichever is spent first - except that
|
* the window alone never ends the replays before one has been made. It is bounded for the reason
|
* {@code PDBStorage} bounds its loop: the configuration change paths of the pluggable backend hold an entry
|
* container's exclusive lock across this method, and every reader of that suffix then waits - untimed and
|
* uninterruptibly - until it returns, so a conflict that never clears would park every worker thread of that
|
* suffix rather than fail one operation.
|
* <p>
|
* JE raises the conflict from inside the operation - a record read or write, never the commit, which takes no
|
* lock - and the transaction wraps it in a {@link StorageRuntimeException}, which is unwrapped below before it is
|
* matched. The operation may catch it there; the transaction is then abort-only and {@code commit()} raises the
|
* conflict again, bare, so an attempt which swallowed its conflict commits nothing and is replayed all the same.
|
* With the shipped configuration the only conflict is the deadlock: {@code je.lock.timeout} is 0, so a writer
|
* waits for a lock rather than times out, and JE ends one transaction of a cycle - chosen at random - as soon as
|
* a wait would close it. A lock timeout set through {@code ds-cfg-je-property} makes a wait that long a conflict
|
* as well.
|
* <p>
|
* The replay backs off first, though JE would make it wait for the locks it lost anyway: JE locks a record by
|
* the LSN of its current version, and the abort of the victim, which hands the waiting survivor the lock it
|
* asked for, undoes the version that lock belongs to - the survivor then has to lock the version the undo put
|
* back, and a replay which comes back at once wins that race, holds the record when the survivor asks again,
|
* and the same deadlock forms with the roles drawn afresh. JE's own retry example sleeps before retrying for
|
* this reason; without the sleep the deadlock of two writers was seen to form three times in a row.
|
* <p>
|
* An interrupt reaches the loop only in that sleep, and the loop does not restore the flag the sleep cleared:
|
* JE invalidates the whole environment when a thread carrying the interrupt flag touches it - the transaction
|
* registry's latch is acquired interruptibly - and the caller's own failure road makes such a call
|
* ({@code EntryContainer.writeTrustState}). The interrupt is reported instead, with the conflict being replayed,
|
* as the suppressed exceptions of the {@link StorageRuntimeException} thrown.
|
* <p>
|
* Once the bound is spent the conflict is reported as a {@link StorageRuntimeException} naming the backend, the
|
* attempts spent, the time they took and which of the two bounds ran out. It carries the conflict as a
|
* suppressed exception rather than as its cause: a cause is unwrapped below and thrown in its place, and
|
* {@code EntryContainer.throwAllowedExceptionTypes} likewise passes a {@link StorageRuntimeException} through
|
* untouched only while it has no cause. Given a cause, both hand the caller the bare conflict instead, whose
|
* message is JE's account of its lock table - all an LDAP client used to be told after the single attempt.
|
*/
|
@Override
|
public void write(final WriteOperation operation) throws Exception
|
{
|
final long startedAt = System.nanoTime();
|
final long giveUpAt = startedAt + retryWindowNanos;
|
for (int attempt = 1;; attempt++)
|
{
|
final LockConflictException conflict;
|
final Transaction txn = beginTransaction();
|
try
|
{
|
try
|
{
|
operation.run(newWriteableTransaction(txn));
|
commit(txn);
|
return;
|
}
|
catch (final StorageRuntimeException e)
|
{
|
throw unwrap(e);
|
}
|
}
|
catch (final LockConflictException e)
|
{
|
conflict = e;
|
}
|
finally
|
{
|
abort(txn);
|
}
|
// System.nanoTime() - giveUpAt is the overflow safe form of the comparison, and attempt > 1 keeps the window
|
// from ending the loop before a single replay: a lock timeout set by the operator can outlast the window on
|
// its own, and it is the attempt after that one which is likeliest to succeed, the transaction that blocked
|
// it having just finished. One clock sample for both, so that the elapsed time reported is the one the give
|
// up was decided on
|
final long now = System.nanoTime();
|
final boolean capSpent = attempt >= maxRetries;
|
if (capSpent || (attempt > 1 && now - giveUpAt >= 0))
|
{
|
final long elapsedMs = TimeUnit.NANOSECONDS.toMillis(now - startedAt);
|
// which of the two bounds was spent, so that the config change paths - which report this as a bare stack
|
// trace with no message id - say whether raising the attempts or the window is what would have helped
|
final String boundSpent = capSpent ? "attempt cap" : "retry window";
|
final StorageRuntimeException spent = new StorageRuntimeException(
|
"je: backend '" + config.getBackendId() + "' did not apply the transaction after " + attempt
|
+ " attempts in " + elapsedMs + " ms, the " + boundSpent + " being spent; the last conflict was "
|
+ conflict);
|
spent.addSuppressed(conflict);
|
// warned once, at exhaustion only: a conflict is answered by a replay, and the trace below is what says
|
// so. stackTraceToSingleLineString, the form the config change paths report this exception with, walks
|
// the causes and never prints a suppressed exception, so this line is the only rendering that carries the
|
// stack of the conflict
|
logger.warn(LocalizableMessage.raw("je: giving up on the transaction of backend '%s' after %d attempts"
|
+ " in %d ms, the %s being spent: %s", config.getBackendId(), attempt, elapsedMs, boundSpent,
|
stackTraceToSingleLineString(conflict)));
|
throw spent;
|
}
|
if (logger.isTraceEnabled())
|
{
|
logger.trace("je: replaying the transaction after %s, attempt %d of %d", conflict, attempt, maxRetries);
|
}
|
try
|
{
|
// slept for after the finally has ended the rolled back transaction, so that nothing of it is held
|
// through the delay; random and growing with every attempt, so that two writers replaying the same
|
// deadlock do not come back in step
|
Thread.sleep(retryDelayMillis(attempt));
|
}
|
catch (final InterruptedException e)
|
{
|
// the flag stays cleared - see above - and the conflict being replayed is reported next to the interrupt,
|
// wrapped the way the exhausted loop wraps it and for the same reason: every caller strips a cause off
|
final StorageRuntimeException interrupted = new StorageRuntimeException(
|
"je: backend '" + config.getBackendId() + "' was interrupted while replaying the transaction after "
|
+ attempt + " attempts; the last conflict was " + conflict);
|
interrupted.addSuppressed(conflict);
|
interrupted.addSuppressed(e);
|
throw interrupted;
|
}
|
}
|
}
|
|
/** Returns the randomized delay before the given attempt is replayed, doubling with each attempt up to a cap. */
|
// package private like the PDBStorage copy it is taken from, so that a test can pin the growth and the cap
|
static long retryDelayMillis(int attempt)
|
{
|
final double bound = Math.min(MAX_SLEEP_ON_RETRY_MS, BASE_SLEEP_ON_RETRY_MS * (1 << Math.min(attempt - 1, 5)));
|
return (long) (Math.random() * bound);
|
}
|
|
private static Exception unwrap(final StorageRuntimeException e) throws Exception
|
{
|
if (e.getCause() != null)
|
{
|
throw (Exception) e.getCause();
|
}
|
throw e;
|
}
|
|
private Transaction beginTransaction()
|
{
|
if (envConfig.getTransactional())
|
{
|
final Transaction txn = env.beginTransaction(null, TXN_READ_COMMITTED);
|
logger.trace("beginTransaction txnid=%d", txn.getId());
|
return txn;
|
}
|
return null;
|
}
|
|
private void commit(final Transaction txn)
|
{
|
if (txn != null)
|
{
|
txn.commit();
|
logger.trace("commit txnid=%d", txn.getId());
|
}
|
}
|
|
private void abort(final Transaction txn)
|
{
|
if (txn != null)
|
{
|
txn.abort();
|
logger.trace("abort txnid=%d", txn.getId());
|
}
|
}
|
|
@Override
|
public boolean supportsBackupAndRestore()
|
{
|
return true;
|
}
|
|
@Override
|
public File getDirectory()
|
{
|
// The directory the storage runs on: a db-directory moved while it runs is used from the next open,
|
// which a new storage makes.
|
return backendDirectory;
|
}
|
|
private static File getBackendDirectory(JEBackendCfg cfg)
|
{
|
return getDBDirectory(cfg.getDBDirectory(), cfg.getBackendId());
|
}
|
|
@Override
|
public ListIterator<Path> getFilesToBackup() throws DirectoryException
|
{
|
return new JELogFilesIterator(getDirectory(), config.getBackendId());
|
}
|
|
/**
|
* Iterator on JE log files to backup.
|
* <p>
|
* The cleaner thread may delete some log files during the backup. The iterator is automatically
|
* renewed if at least one file has been deleted.
|
*/
|
static class JELogFilesIterator implements ListIterator<Path>
|
{
|
/** Root directory where all files are located. */
|
private final File rootDirectory;
|
private final String backendID;
|
|
/** Underlying iterator on files. */
|
private ListIterator<Path> iterator;
|
/** Files to backup. Used to renew the iterator if necessary. */
|
private List<Path> files;
|
|
private String lastFileName = "";
|
private long lastFileSize;
|
|
JELogFilesIterator(File rootDirectory, String backendID) throws DirectoryException
|
{
|
this.rootDirectory = rootDirectory;
|
this.backendID = backendID;
|
setFiles(BackupManager.getFiles(rootDirectory, new JELogFileFilter(), backendID));
|
}
|
|
private void setFiles(List<Path> files)
|
{
|
this.files = files;
|
Collections.sort(files);
|
if (!files.isEmpty())
|
{
|
Path lastFile = files.get(files.size() - 1);
|
lastFileName = lastFile.getFileName().toString();
|
lastFileSize = lastFile.toFile().length();
|
}
|
iterator = files.listIterator();
|
}
|
|
@Override
|
public boolean hasNext()
|
{
|
boolean hasNext = iterator.hasNext();
|
if (!hasNext && !files.isEmpty())
|
{
|
try
|
{
|
List<Path> allFiles = BackupManager.getFiles(rootDirectory, new JELogFileFilter(), backendID);
|
List<Path> compare = new ArrayList<>(files);
|
compare.removeAll(allFiles);
|
if (!compare.isEmpty())
|
{
|
// at least one file was deleted,
|
// the iterator must be renewed based on last file previously available
|
List<Path> newFiles =
|
BackupManager.getFiles(rootDirectory, new JELogFileFilter(lastFileName, lastFileSize), backendID);
|
logger.info(NOTE_JEB_BACKUP_CLEANER_ACTIVITY.get(newFiles.size()));
|
if (!newFiles.isEmpty())
|
{
|
setFiles(newFiles);
|
hasNext = iterator.hasNext();
|
}
|
}
|
}
|
catch (DirectoryException e)
|
{
|
logger.error(ERR_BACKEND_LIST_FILES_TO_BACKUP.get(backendID, stackTraceToSingleLineString(e)));
|
}
|
}
|
return hasNext;
|
}
|
|
@Override
|
public Path next()
|
{
|
if (hasNext())
|
{
|
return iterator.next();
|
}
|
throw new NoSuchElementException();
|
}
|
|
@Override
|
public boolean hasPrevious()
|
{
|
return iterator.hasPrevious();
|
}
|
|
@Override
|
public Path previous()
|
{
|
return iterator.previous();
|
}
|
|
@Override
|
public int nextIndex()
|
{
|
return iterator.nextIndex();
|
}
|
|
@Override
|
public int previousIndex()
|
{
|
return iterator.previousIndex();
|
}
|
|
@Override
|
public void remove()
|
{
|
throw new UnsupportedOperationException("remove() is not implemented");
|
}
|
|
@Override
|
public void set(Path e)
|
{
|
throw new UnsupportedOperationException("set() is not implemented");
|
}
|
|
@Override
|
public void add(Path e)
|
{
|
throw new UnsupportedOperationException("add() is not implemented");
|
}
|
}
|
|
/**
|
* This class implements a FilenameFilter to detect a JE log file, possibly with a constraint on
|
* the file name and file size.
|
*/
|
private static class JELogFileFilter implements FileFilter
|
{
|
private final String latestFilename;
|
private final long latestFileSize;
|
|
/**
|
* Creates the filter for log files that are newer than provided file name
|
* or equal to provided file name and of larger size.
|
* @param latestFilename the latest file name
|
* @param latestFileSize the latest file size
|
*/
|
JELogFileFilter(String latestFilename, long latestFileSize)
|
{
|
this.latestFilename = latestFilename;
|
this.latestFileSize = latestFileSize;
|
}
|
|
/** Creates the filter for any JE log file. */
|
JELogFileFilter()
|
{
|
this("", 0);
|
}
|
|
@Override
|
public boolean accept(File file)
|
{
|
String name = file.getName();
|
int cmp = name.compareTo(latestFilename);
|
return name.endsWith(".jdb")
|
&& (cmp > 0 || (cmp == 0 && file.length() > latestFileSize));
|
}
|
}
|
|
@Override
|
public Path beforeRestore() throws DirectoryException
|
{
|
return null;
|
}
|
|
@Override
|
public boolean isDirectRestore()
|
{
|
// restore is done in an intermediate directory
|
return false;
|
}
|
|
@Override
|
public void afterRestore(Path restoreDirectory, Path saveDirectory) throws DirectoryException
|
{
|
// intermediate directory content is moved to database directory
|
File targetDirectory = getDirectory();
|
recursiveDelete(targetDirectory);
|
try
|
{
|
Files.move(restoreDirectory, targetDirectory.toPath());
|
}
|
catch(IOException e)
|
{
|
LocalizableMessage msg = ERR_CANNOT_RENAME_RESTORE_DIRECTORY.get(restoreDirectory, targetDirectory.getPath());
|
throw new DirectoryException(DirectoryServer.getCoreConfigManager().getServerErrorResultCode(), msg);
|
}
|
}
|
|
@Override
|
public void createBackup(BackupConfig backupConfig) throws DirectoryException
|
{
|
new BackupManager(config.getBackendId()).createBackup(this, backupConfig);
|
}
|
|
@Override
|
public void removeBackup(BackupDirectory backupDirectory, String backupID) throws DirectoryException
|
{
|
new BackupManager(config.getBackendId()).removeBackup(backupDirectory, backupID);
|
}
|
|
@Override
|
public void restoreBackup(RestoreConfig restoreConfig) throws DirectoryException
|
{
|
new BackupManager(config.getBackendId()).restoreBackup(this, restoreConfig);
|
}
|
|
@Override
|
public Set<TreeName> listTrees()
|
{
|
if (env == null)
|
{
|
return Collections.<TreeName>emptySet();
|
}
|
try
|
{
|
List<String> treeNames = env.getDatabaseNames();
|
final Set<TreeName> results = new HashSet<>(treeNames.size());
|
for (String treeName : treeNames)
|
{
|
results.add(TreeName.valueOf(treeName));
|
}
|
return results;
|
}
|
catch (DatabaseException e)
|
{
|
throw new StorageRuntimeException(e);
|
}
|
}
|
|
@Override
|
public boolean isConfigurationChangeAcceptable(JEBackendCfg newCfg,
|
List<LocalizableMessage> unacceptableReasons)
|
{
|
// A size which does not grow past the one configured asks the quota for nothing, as before: every
|
// change of the backend entry comes here, the disable of an online import included, and after an
|
// open the quota refused this storage holds nothing to measure such a change against. A growth is
|
// measured against what this storage holds, which is what the next open adds to - not against
|
// config, which a change admitted but not yet applied has already moved to the new size.
|
final long newSize = computeSize(newCfg);
|
final MemoryQuota quota = serverContext.getMemoryQuota();
|
return (newSize <= Math.max(reservedCacheSize, computeSize(config))
|
|| quota.isMemoryAvailable(newSize - reservedCacheSize))
|
&& checkConfigurationDirectories(newCfg, unacceptableReasons)
|
&& checkEnvironmentConfiguration(newCfg, unacceptableReasons);
|
}
|
|
private long computeSize(JEBackendCfg cfg)
|
{
|
return cfg.getDBCacheSize() > 0
|
? cfg.getDBCacheSize()
|
: serverContext.getMemoryQuota().memPercentToBytes(cfg.getDBCachePercent());
|
}
|
|
/**
|
* Checks newly created backend has a valid configuration.
|
* @param cfg the new configuration
|
* @param unacceptableReasons the list of accumulated errors and their messages
|
* @param context the server context
|
* @return true if newly created backend has a valid configuration
|
*/
|
static boolean isConfigurationAcceptable(JEBackendCfg cfg, List<LocalizableMessage> unacceptableReasons,
|
ServerContext context)
|
{
|
if (context != null)
|
{
|
MemoryQuota memQuota = context.getMemoryQuota();
|
if (cfg.getDBCacheSize() > 0 && !memQuota.isMemoryAvailable(cfg.getDBCacheSize()))
|
{
|
unacceptableReasons.add(ERR_BACKEND_CONFIG_CACHE_SIZE_GREATER_THAN_JVM_HEAP.get(
|
cfg.getDBCacheSize(), memQuota.getAvailableMemory()));
|
return false;
|
}
|
else if (!memQuota.isMemoryAvailable(memQuota.memPercentToBytes(cfg.getDBCachePercent())))
|
{
|
unacceptableReasons.add(ERR_BACKEND_CONFIG_CACHE_PERCENT_GREATER_THAN_JVM_HEAP.get(
|
cfg.getDBCachePercent(), memQuota.memBytesToPercent(memQuota.getAvailableMemory())));
|
return false;
|
}
|
}
|
return checkConfigurationDirectories(cfg, unacceptableReasons)
|
&& checkEnvironmentConfiguration(cfg, unacceptableReasons);
|
}
|
|
/**
|
* Whether an environment can be configured from the given configuration. A durability which
|
* sets both flags, or a native property JE does not know, is refused here, before the change
|
* is written - rather than by the next open of the backend, which is where a configuration
|
* nothing checked used to fail.
|
*/
|
private static boolean checkEnvironmentConfiguration(JEBackendCfg cfg, List<LocalizableMessage> unacceptableReasons)
|
{
|
try
|
{
|
ConfigurableEnvironment.toEnvironmentConfig(cfg);
|
return true;
|
}
|
catch (ConfigException e)
|
{
|
unacceptableReasons.add(e.getMessageObject());
|
return false;
|
}
|
}
|
|
private static boolean checkConfigurationDirectories(JEBackendCfg cfg,
|
List<LocalizableMessage> unacceptableReasons)
|
{
|
final ConfigChangeResult ccr = new ConfigChangeResult();
|
File newBackendDirectory = getBackendDirectory(cfg);
|
|
checkDBDirExistsOrCanCreate(newBackendDirectory, ccr, true);
|
checkDBDirPermissions(cfg.getDBDirectoryPermissions(), cfg.dn(), ccr);
|
if (!ccr.getMessages().isEmpty())
|
{
|
unacceptableReasons.addAll(ccr.getMessages());
|
return false;
|
}
|
return true;
|
}
|
|
@Override
|
public ConfigChangeResult applyConfigurationChange(JEBackendCfg cfg)
|
{
|
final ConfigChangeResult ccr = new ConfigChangeResult();
|
|
try
|
{
|
File newBackendDirectory = getBackendDirectory(cfg);
|
// Against the directory the storage runs on rather than the configuration as last changed, so
|
// that a later change still asks for the restart a move is waiting for.
|
final boolean moved = !newBackendDirectory.equals(backendDirectory);
|
|
// Create the directory if it doesn't exist.
|
if (moved)
|
{
|
checkDBDirExistsOrCanCreate(newBackendDirectory, ccr, false);
|
if (!ccr.getMessages().isEmpty())
|
{
|
return ccr;
|
}
|
|
ccr.setAdminActionRequired(true);
|
ccr.addMessage(NOTE_CONFIG_DB_DIR_REQUIRES_RESTART.get(backendDirectory, newBackendDirectory));
|
}
|
|
if (!cfg.getDBDirectoryPermissions().equalsIgnoreCase(runningDirectoryPermissions)
|
|| moved)
|
{
|
checkDBDirPermissions(cfg.getDBDirectoryPermissions(), cfg.dn(), ccr);
|
// By its result code: the note of a moved directory is in the result already, and the rest of
|
// the change is still applied and reported alongside it.
|
if (ccr.getResultCode() != ResultCode.SUCCESS)
|
{
|
return ccr;
|
}
|
|
setDBDirPermissions(newBackendDirectory, cfg.getDBDirectoryPermissions(), cfg.dn(), ccr);
|
if (ccr.getResultCode() != ResultCode.SUCCESS)
|
{
|
return ccr;
|
}
|
if (!moved)
|
{
|
runningDirectoryPermissions = cfg.getDBDirectoryPermissions();
|
}
|
}
|
final long newCacheSize = computeSize(cfg);
|
if (env != null && newCacheSize != configuredCacheSize)
|
{
|
// The cache is sized when the environment opens and this storage never resizes it: the next
|
// open of the backend builds it to the new size and reserves that, and until then the
|
// reservation stays with the cache it was made for.
|
ccr.setAdminActionRequired(true);
|
ccr.addMessage(
|
NOTE_CONFIG_DB_CACHE_REQUIRES_RESTART.get(cfg.getBackendId(), configuredCacheSize, newCacheSize));
|
}
|
// An import runs the environment on a configuration of its own, which goes with it: the backend
|
// opens again on the configuration as changed once the import is over.
|
if (env != null && envConfig.getTransactional())
|
{
|
applyToEnvironment(cfg, ccr);
|
}
|
registerMonitoredDirectory(cfg);
|
config = cfg;
|
}
|
catch (Exception e)
|
{
|
addErrorMessage(ccr, LocalizableMessage.raw(stackTraceToSingleLineString(e)));
|
}
|
return ccr;
|
}
|
|
/**
|
* Applies to the running environment what JE takes while it runs, and asks for a restart for what
|
* it takes at the open alone. The environment is configured when it opens, from the configuration
|
* as it is then: a change of a property JE accepts as mutable is handed to the environment here,
|
* and a change of one it does not is reported - the change result reaches the error log, where a
|
* change reported as applied while the environment ran on unchanged until its next open did not.
|
* <p>
|
* The cache is the one mutable setting left where the open put it: it is sized with the memory
|
* reserved for it, and a change of its size asks for a restart above, at which the next open
|
* reserves the new size.
|
*/
|
private void applyToEnvironment(JEBackendCfg cfg, ConfigChangeResult ccr) throws ConfigException
|
{
|
final EnvironmentConfig next = ConfigurableEnvironment.toEnvironmentConfig(cfg);
|
final EnvironmentConfig running = env.getConfig();
|
for (ConfigParam param : new TreeMap<>(EnvironmentParams.SUPPORTED_PARAMS).values())
|
{
|
// Replication parameters are not set through an environment configuration; a multi-value
|
// parameter is not read as one value. Neither is set by this storage.
|
if (param.isForReplication() || param.isMultiValueParam())
|
{
|
continue;
|
}
|
final String runningValue = running.getConfigParam(param.getName());
|
final String nextValue = next.getConfigParam(param.getName());
|
if (Objects.equals(runningValue, nextValue)
|
|| (param.isMutable()
|
&& !switchesOffHeapCache(param.getName(), runningValue, nextValue)
|
&& (next.isConfigParamSet(param.getName()) || resetsToDefault(next, param.getName(), nextValue))))
|
{
|
continue;
|
}
|
ccr.setAdminActionRequired(true);
|
ccr.addMessage(NOTE_CONFIG_DB_PROPERTY_REQUIRES_RESTART.get(
|
ConfigurableEnvironment.configuredNameOf(param.getName()), cfg.getBackendId(), runningValue, nextValue));
|
}
|
next.setConfigParam(MAX_MEMORY, running.getConfigParam(MAX_MEMORY));
|
next.setConfigParam(MAX_MEMORY_PERCENT, running.getConfigParam(MAX_MEMORY_PERCENT));
|
// The off-heap cache is switched on or off by the next open alone, which the change asks for above.
|
final String offHeapRunning = running.getConfigParam(MAX_OFF_HEAP_MEMORY);
|
if (switchesOffHeapCache(MAX_OFF_HEAP_MEMORY, offHeapRunning, next.getConfigParam(MAX_OFF_HEAP_MEMORY)))
|
{
|
next.setConfigParam(MAX_OFF_HEAP_MEMORY, offHeapRunning);
|
}
|
// What JE takes while it runs, of the properties the configuration sets; the rest it ignores.
|
env.setMutableConfig(next);
|
}
|
|
/**
|
* Tells whether a change of the given parameter switches JE's off-heap cache on or off. JE takes a
|
* change of the off-heap cache size while it runs, but not one between zero and non-zero: it throws
|
* once it has already taken the new value as its own, and so throws again on every change which
|
* follows, until the next open - which does it.
|
*
|
* @param name the name of the parameter
|
* @param runningValue the value the environment runs with
|
* @param nextValue the value configured
|
* @return whether the change switches the off-heap cache on or off
|
*/
|
private static boolean switchesOffHeapCache(String name, String runningValue, String nextValue)
|
{
|
return MAX_OFF_HEAP_MEMORY.equals(name)
|
&& (Long.parseLong(runningValue) > 0) != (Long.parseLong(nextValue) > 0);
|
}
|
|
/**
|
* Sets a mutable parameter the configuration no longer sets - a je-property removed - to JE's
|
* default, since the environment keeps the value it runs with of every parameter it is not handed.
|
* A default JE does not take as a value, such as the 0 of je.cleaner.readSize, which JE reads as
|
* "computed at the open", leaves the parameter to the next open.
|
*
|
* @param next the environment configuration handed to the running environment
|
* @param name the name of the parameter
|
* @param defaultValue JE's default of the parameter, as the configuration reads it
|
* @return whether the default is handed to the environment along with the rest
|
*/
|
private static boolean resetsToDefault(EnvironmentConfig next, String name, String defaultValue)
|
{
|
if (defaultValue == null)
|
{
|
return false;
|
}
|
try
|
{
|
next.setConfigParam(name, defaultValue);
|
return true;
|
}
|
catch (IllegalArgumentException e)
|
{
|
return false;
|
}
|
}
|
|
private void registerMonitoredDirectory(JEBackendCfg cfg)
|
{
|
diskMonitor.registerMonitoredDirectory(
|
cfg.getBackendId() + " backend",
|
getDirectory(),
|
cfg.getDiskLowThreshold(),
|
cfg.getDiskFullThreshold(),
|
this);
|
}
|
|
@Override
|
public void removeStorageFiles() throws StorageRuntimeException
|
{
|
StorageUtils.removeStorageFiles(backendDirectory);
|
}
|
|
@Override
|
public StorageStatus getStorageStatus()
|
{
|
return storageStatus;
|
}
|
|
@Override
|
public void diskFullThresholdReached(File directory, long thresholdInBytes) {
|
storageStatus = statusWhenDiskSpaceFull(directory, thresholdInBytes, config.getBackendId());
|
}
|
|
@Override
|
public void diskLowThresholdReached(File directory, long thresholdInBytes) {
|
storageStatus = statusWhenDiskSpaceLow(directory, thresholdInBytes, config.getBackendId());
|
}
|
|
@Override
|
public void diskSpaceRestored(File directory, long lowThresholdInBytes, long fullThresholdInBytes) {
|
storageStatus = StorageStatus.working();
|
}
|
|
private static void setData(final DatabaseEntry dbEntry, final ByteSequence bs)
|
{
|
dbEntry.setData(bs != null ? bs.toByteArray() : null);
|
}
|
|
private static DatabaseEntry db(final ByteSequence bs)
|
{
|
return new DatabaseEntry(bs != null ? bs.toByteArray() : null);
|
}
|
|
private static ByteString valueToBytes(final DatabaseEntry dbValue, boolean isDefined)
|
{
|
if (isDefined)
|
{
|
return ByteString.wrap(dbValue.getData());
|
}
|
return null;
|
}
|
}
|
