/*
|
* 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 2006-2008 Sun Microsystems, Inc.
|
* Portions Copyright 2014-2016 ForgeRock AS.
|
* Portions Copyright 2026 3A Systems, LLC.
|
*/
|
package org.opends.server.plugins;
|
|
|
|
import static org.opends.server.util.CollectionUtils.*;
|
|
import java.io.IOException;
|
import java.util.List;
|
import java.util.Locale;
|
import java.util.Map;
|
import java.util.Set;
|
import java.util.concurrent.ConcurrentHashMap;
|
import java.util.concurrent.TimeUnit;
|
import java.util.concurrent.TimeoutException;
|
import java.util.concurrent.atomic.AtomicInteger;
|
import java.util.function.Predicate;
|
|
import org.forgerock.i18n.LocalizableMessage;
|
import org.forgerock.opendj.config.server.ConfigException;
|
import org.forgerock.opendj.io.ASN1;
|
import org.forgerock.opendj.io.ASN1Reader;
|
import org.forgerock.opendj.io.ASN1Writer;
|
import org.forgerock.opendj.ldap.ByteString;
|
import org.forgerock.opendj.ldap.ResultCode;
|
import org.forgerock.opendj.server.config.server.PluginCfg;
|
import org.opends.server.api.plugin.DirectoryServerPlugin;
|
import org.opends.server.api.plugin.PluginResult;
|
import org.opends.server.api.plugin.PluginType;
|
import org.opends.server.controls.ControlDecoder;
|
import org.opends.server.types.Control;
|
import org.opends.server.types.DirectoryException;
|
import org.opends.server.types.OperationType;
|
import org.opends.server.types.operation.*;
|
|
/**
|
* This class defines a very simple plugin that causes request processing to end
|
* immediately and send a specific result code to the client. It will be
|
* triggered by a control contained in the client request, and may be invoked
|
* during either pre-parse or pre-operation processing. Short circuits can
|
* also be registered for operations regardless of controls.
|
*/
|
public class ShortCircuitPlugin
|
extends DirectoryServerPlugin<PluginCfg>
|
{
|
/**
|
* The OID for the short circuit request control, which is used to flag
|
* operations that should cause the operation processing to end immediately.
|
*/
|
public static final String OID_SHORT_CIRCUIT_REQUEST =
|
"1.3.6.1.4.1.26027.1.999.3";
|
|
/**
|
* The control used by this plugin.
|
*/
|
public static class ShortCircuitRequestControl extends Control
|
{
|
/**
|
* ControlDecoder implementation to decode this control from a ByteString.
|
*/
|
private static final class Decoder
|
implements ControlDecoder<ShortCircuitRequestControl>
|
{
|
/** {@inheritDoc} */
|
@Override
|
public ShortCircuitRequestControl decode(boolean isCritical, ByteString value)
|
throws DirectoryException
|
{
|
ASN1Reader reader = ASN1.getReader(value);
|
|
try
|
{
|
reader.readStartSequence();
|
int resultCode = (int)reader.readInteger();
|
String section = reader.readOctetStringAsString();
|
reader.readEndSequence();
|
|
return new ShortCircuitRequestControl(isCritical,
|
resultCode, section);
|
}
|
catch (Exception e)
|
{
|
// TODO: Need a better message
|
throw new DirectoryException(ResultCode.PROTOCOL_ERROR, null, e);
|
}
|
}
|
|
@Override
|
public String getOID()
|
{
|
return OID_SHORT_CIRCUIT_REQUEST;
|
}
|
|
}
|
|
/**
|
* The Control Decoder that can be used to decode this control.
|
*/
|
public static final ControlDecoder<ShortCircuitRequestControl> DECODER =
|
new Decoder();
|
|
|
private int resultCode;
|
private String section;
|
|
/**
|
* Constructs a new control of this class.
|
*
|
* @param isCritical
|
* Indicates whether support for this control should be considered
|
* a critical part of the server processing.
|
* @param resultCode
|
* The result code to return to the client.
|
* @param section
|
* The section to use to determine when to short circuit.
|
*/
|
public ShortCircuitRequestControl(boolean isCritical, int resultCode,
|
String section)
|
{
|
super(OID_SHORT_CIRCUIT_REQUEST, isCritical);
|
this.resultCode = resultCode;
|
this.section = section;
|
}
|
|
/**
|
* Writes this control's value to an ASN.1 writer. The value (if any)
|
* must be written as an ASN1OctetString.
|
*
|
* @param writer The ASN.1 writer to use.
|
* @throws IOException If a problem occurs while writing to the stream.
|
*/
|
@Override
|
protected void writeValue(ASN1Writer writer) throws IOException {
|
writer.writeStartSequence(ASN1.UNIVERSAL_OCTET_STRING_TYPE);
|
writer.writeStartSequence();
|
writer.writeInteger(resultCode);
|
writer.writeOctetString(section);
|
writer.writeEndSequence();
|
writer.writeEndSequence();
|
}
|
|
/**
|
* Retrieves the resultCode.
|
*
|
* @return The resultCode.
|
*/
|
public int getResultCode()
|
{
|
return resultCode;
|
}
|
|
/**
|
* Retrieves the section.
|
*
|
* @return The section.
|
*/
|
public String getSection()
|
{
|
return section;
|
}
|
}
|
|
|
|
/**
|
* Creates a new instance of this Directory Server plugin. Every
|
* plugin must implement a default constructor (it is the only one
|
* that will be used to create plugins defined in the
|
* configuration), and every plugin constructor must call
|
* <CODE>super()</CODE> as its first element.
|
*/
|
public ShortCircuitPlugin()
|
{
|
super();
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public void initializePlugin(Set<PluginType> pluginTypes,
|
PluginCfg configuration)
|
throws ConfigException
|
{
|
// This plugin may only be used as a pre-parse or pre-operation plugin.
|
for (PluginType t : pluginTypes)
|
{
|
switch (t)
|
{
|
case PRE_PARSE_ABANDON:
|
case PRE_PARSE_ADD:
|
case PRE_PARSE_BIND:
|
case PRE_PARSE_COMPARE:
|
case PRE_PARSE_DELETE:
|
case PRE_PARSE_EXTENDED:
|
case PRE_PARSE_MODIFY:
|
case PRE_PARSE_MODIFY_DN:
|
case PRE_PARSE_SEARCH:
|
case PRE_PARSE_UNBIND:
|
case PRE_OPERATION_ADD:
|
case PRE_OPERATION_BIND:
|
case PRE_OPERATION_COMPARE:
|
case PRE_OPERATION_DELETE:
|
case PRE_OPERATION_EXTENDED:
|
case PRE_OPERATION_MODIFY:
|
case PRE_OPERATION_MODIFY_DN:
|
case PRE_OPERATION_SEARCH:
|
// This is fine.
|
break;
|
default:
|
throw new ConfigException(LocalizableMessage.raw("Invalid plugin type " + t +
|
" for the short circuit plugin."));
|
}
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public void finalizePlugin()
|
{
|
/*
|
* A park which outlives the test which took it holds a replay thread of this server,
|
* and every replayed operation queued behind it, for as long as this plugin is
|
* loaded: the map is static and nothing but the test itself removes an entry from it.
|
*/
|
for (ParkedReplay park : parks.values())
|
{
|
park.deregister();
|
}
|
parks.clear();
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseAbandonOperation abandonOperation)
|
{
|
int resultCode = shortCircuitInternal(abandonOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse doPreParse(PreParseAddOperation addOperation)
|
{
|
int resultCode = shortCircuitInternal(addOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse doPreParse(PreParseBindOperation bindOperation)
|
{
|
int resultCode = shortCircuitInternal(bindOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseCompareOperation compareOperation)
|
{
|
int resultCode = shortCircuitInternal(compareOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseDeleteOperation deleteOperation)
|
{
|
int resultCode = shortCircuitInternal(deleteOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseExtendedOperation extendedOperation)
|
{
|
int resultCode = shortCircuitInternal(extendedOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseModifyOperation modifyOperation)
|
{
|
int resultCode = shortCircuitInternal(modifyOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseModifyDNOperation modifyDNOperation)
|
{
|
int resultCode = shortCircuitInternal(modifyDNOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseSearchOperation searchOperation)
|
{
|
int resultCode = shortCircuitInternal(searchOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreParse
|
doPreParse(PreParseUnbindOperation unbindOperation)
|
{
|
int resultCode = shortCircuitInternal(unbindOperation, "PreParse");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreParse.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-parse"));
|
}
|
else
|
{
|
return PluginResult.PreParse.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationAddOperation addOperation)
|
{
|
int resultCode = shortCircuitInternal(addOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationBindOperation bindOperation)
|
{
|
int resultCode = shortCircuitInternal(bindOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationCompareOperation compareOperation)
|
{
|
int resultCode = shortCircuitInternal(compareOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationDeleteOperation deleteOperation)
|
{
|
int resultCode = shortCircuitInternal(deleteOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationExtendedOperation extendedOperation)
|
{
|
int resultCode = shortCircuitInternal(extendedOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationModifyOperation modifyOperation)
|
{
|
int resultCode = shortCircuitInternal(modifyOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationModifyDNOperation modifyDNOperation)
|
{
|
int resultCode = shortCircuitInternal(modifyDNOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/** {@inheritDoc} */
|
@Override
|
public PluginResult.PreOperation
|
doPreOperation(PreOperationSearchOperation searchOperation)
|
{
|
int resultCode = shortCircuitInternal(searchOperation, "PreOperation");
|
if (resultCode >= 0)
|
{
|
return PluginResult.PreOperation.stopProcessing(
|
ResultCode.valueOf(resultCode),
|
LocalizableMessage.raw("Short-circuit in pre-operation"));
|
}
|
else
|
{
|
return PluginResult.PreOperation.continueOperationProcessing();
|
}
|
}
|
|
|
|
/**
|
* Looks for a short-circuit request control in the operation and if one is
|
* found with the correct section then generate the appropriate result.
|
*
|
* @param operation The operation to be processed.
|
* @param section The section to match in the control value.
|
*
|
* @return The result code that should be immediately sent to the client, or
|
* -1 if operation processing should continue as normal.
|
*/
|
private int shortCircuitInternal(PluginOperation operation, String section)
|
{
|
try
|
{
|
ShortCircuitRequestControl control =
|
operation.getRequestControl(ShortCircuitRequestControl.DECODER);
|
if (control != null && section.equalsIgnoreCase(control.getSection()))
|
{
|
return control.resultCode;
|
}
|
}
|
catch (Exception e)
|
{
|
System.err.println("***** ERROR: Could not decode short circuit " +
|
"control value: " + e);
|
e.printStackTrace();
|
return -1;
|
}
|
|
// Check for registered short circuits.
|
final String key = keyFor(operation.getOperationType(), section);
|
Integer resultCode = shortCircuits.get(key);
|
if (resultCode != null)
|
{
|
final int reached = shortCircuitCounts.computeIfAbsent(key, k -> new AtomicInteger()).incrementAndGet();
|
final Integer maxTimes = shortCircuitLimits.get(key);
|
if (maxTimes == null || reached <= maxTimes)
|
{
|
return resultCode;
|
}
|
// The short circuit was applied as many times as it was asked for: from now on the
|
// operations are let through, which is how a transient failure is simulated.
|
}
|
|
/*
|
* A parked replay is held here, which is inside the run() of the operation and before
|
* anything of the backend was taken: the thread which is replaying a change sits on
|
* this monitor while it still owns that change, which is what lets a test act on the
|
* thread rather than race it. It is consulted last, so that a park never takes an
|
* operation away from a control or from a registered short circuit.
|
*/
|
if (operation.isSynchronizationOperation())
|
{
|
final ParkedReplay park = parks.get(key);
|
if (park != null && park.parks(operation))
|
{
|
final int parkResultCode = park.hold();
|
if (parkResultCode >= 0)
|
{
|
return parkResultCode;
|
}
|
}
|
}
|
|
// If we've gotten here, then we shouldn't short-circuit the operation
|
// processing.
|
return -1;
|
}
|
|
|
|
/**
|
* Creates a short circuit request control with the specified result code and
|
* section.
|
*
|
* @param resultCode The result code to return to the client.
|
* @param section The section to use to determine when to short circuit.
|
*
|
* @return The appropriate short circuit request control.
|
*/
|
public static Control createShortCircuitControl(int resultCode,
|
String section)
|
{
|
return new ShortCircuitRequestControl(false, resultCode, section);
|
}
|
|
|
|
/**
|
* Retrieves a list containing a short circuit control with the specified
|
* result code and section.
|
*
|
* @param resultCode The result code to return to the client.
|
* @param section The section to use to determine when to short circuit.
|
*
|
* @return A list containing the appropriate short circuit request control.
|
*/
|
public static List<Control> createShortCircuitControlList(int resultCode, String section)
|
{
|
return newArrayList(createShortCircuitControl(resultCode, section));
|
}
|
|
/** Registered short circuits for operations regardless of controls. */
|
private static Map<String, Integer> shortCircuits = new ConcurrentHashMap<>();
|
|
/** How many times a registered short circuit was reached. */
|
private static final Map<String, AtomicInteger> shortCircuitCounts = new ConcurrentHashMap<>();
|
|
/** How many times a registered short circuit must be applied, when it is limited. */
|
private static final Map<String, Integer> shortCircuitLimits = new ConcurrentHashMap<>();
|
|
/**
|
* Returns how many times the short circuit registered for the given operation type and
|
* plugin point was reached. A short circuit registered for a limited number of times is
|
* counted as reached by the operations it let through once that number was used up.
|
*
|
* @param operation The type of operation the short circuit applies to.
|
* @param section The plugin point the short circuit applies to.
|
* @return the number of operations which reached the short circuit
|
*/
|
public static int getShortCircuitCount(OperationType operation, String section)
|
{
|
final AtomicInteger count = shortCircuitCounts.get(keyFor(operation, section));
|
return count != null ? count.get() : 0;
|
}
|
|
|
/**
|
* Register a short circuit for the given operation type and plugin point.
|
* @param operation The type of operation the short circuit applies to.
|
* @param section The plugin point the short circuit applies to.
|
* @param resultCode The result code to be returned for the short circuit.
|
*/
|
public static void registerShortCircuit(OperationType operation, String section, int resultCode)
|
{
|
final String key = keyFor(operation, section);
|
// This registration applies to every operation, and it counts from zero: a limit or
|
// a count left behind by a previous registration is not part of it.
|
shortCircuitCounts.remove(key);
|
shortCircuitLimits.remove(key);
|
shortCircuits.put(key, resultCode);
|
}
|
|
/**
|
* Register a short circuit which only applies to the given number of operations, the
|
* ones which follow being let through: this is how a transient failure is simulated.
|
*
|
* @param operation The type of operation the short circuit applies to.
|
* @param section The plugin point the short circuit applies to.
|
* @param resultCode The result code to be returned for the short circuit.
|
* @param maxTimes How many operations must be short circuited.
|
*/
|
public static void registerShortCircuit(OperationType operation, String section, int resultCode, int maxTimes)
|
{
|
final String key = keyFor(operation, section);
|
shortCircuitCounts.remove(key);
|
shortCircuitLimits.put(key, maxTimes);
|
shortCircuits.put(key, resultCode);
|
}
|
|
/**
|
* Deregister a short circuit for the given operation type and plugin point.
|
* @param operation The type of operation the short circuit applies to.
|
* @param section The plugin point the short circuit applies to.
|
*/
|
public static void deregisterShortCircuit(OperationType operation, String section)
|
{
|
final String key = keyFor(operation, section);
|
shortCircuits.remove(key);
|
shortCircuitLimits.remove(key);
|
// The count belongs to the registration which is being removed: a test which counts
|
// the operations it short circuits must not inherit the count of the previous one.
|
shortCircuitCounts.remove(key);
|
}
|
|
/** Registered parks for the replayed operations, keyed like the short circuits. */
|
private static final Map<String, ParkedReplay> parks = new ConcurrentHashMap<>();
|
|
/**
|
* Holds the replayed operations of one type where they are, one at a time, until the
|
* test lets each of them go.
|
* <p>
|
* The hold is taken at a plugin point which runs inside {@code op.run()}, so the thread
|
* which is replaying a change is stopped while it still owns that change: a test can
|
* then do something to that thread - stop it, disable its domain - and know the change
|
* is in flight rather than hope it is. Nothing of the backend has been taken at that
|
* point, so a parked operation blocks the replay and nothing else.
|
*/
|
public static final class ParkedReplay
|
{
|
/**
|
* The value which lets the operation run rather than short circuit it.
|
* <p>
|
* {@code ResultCode.UNDEFINED} is registered on {@code -1} as well, so
|
* {@code release(ResultCode.UNDEFINED.intValue())} lets the operation run instead of
|
* making it report that code - the same hole {@code registerShortCircuit(-1)} has.
|
* No caller has a use for it, and a park releases with a real result code or with
|
* none at all.
|
*/
|
private static final int LET_THROUGH = -1;
|
|
/**
|
* How long an operation is held before this park gives up on the test which took it.
|
* <p>
|
* It is far longer than any release a test waits for - the fixture itself waits a
|
* minute for a park - and it exists for the test which never releases at all: a park
|
* leaked by a method killed on a timeout would otherwise hold a replay thread of this
|
* server, and every replayed operation queued behind it, for the life of the JVM.
|
*/
|
private static final long MAX_HOLD_IN_MS = TimeUnit.MINUTES.toMillis(5);
|
|
private final String key;
|
/** Which of the replayed operations of that type this park is for. */
|
private final Predicate<PluginOperation> parked;
|
private final Object lock = new Object();
|
/** Whether an operation is parked right now. */
|
private boolean occupied;
|
/**
|
* The thread of the operation which parked last. It is never cleared, so that a test
|
* which waited for a park is handed the thread of that park even when the operation
|
* has left the park since - a park which is let go of by {@link #deregister()}, or by
|
* the thread it holds being interrupted, would otherwise hand out no thread at all
|
* and have an assertion on which thread replays the change pass without asserting it.
|
*/
|
private Thread lastParkedThread;
|
/** How many operations were parked, which is what tells one park from the next. */
|
private int parkedOperations;
|
/** How many of them the test has waited for already. */
|
private int awaitedOperations;
|
private boolean released;
|
private int releasedResultCode;
|
private boolean deregistered;
|
|
private ParkedReplay(String key, Predicate<PluginOperation> parked)
|
{
|
this.key = key;
|
this.parked = parked;
|
}
|
|
/** Returns whether the provided operation is one this park is for. */
|
private boolean parks(PluginOperation operation)
|
{
|
return parked.test(operation);
|
}
|
|
/**
|
* Parks the calling operation until the test releases it. Runs on the thread which is
|
* replaying the change.
|
*
|
* @return the result code the operation must be short circuited with, or a negative
|
* value to let it run
|
*/
|
private int hold()
|
{
|
final long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(MAX_HOLD_IN_MS);
|
synchronized (lock)
|
{
|
// One operation at a time, so that a release belongs to the operation the test
|
// waited for rather than to whichever of them the scheduler let in first.
|
while (occupied && !deregistered)
|
{
|
if (!waitOnLock(deadline))
|
{
|
return LET_THROUGH;
|
}
|
}
|
if (deregistered)
|
{
|
return LET_THROUGH;
|
}
|
occupied = true;
|
lastParkedThread = Thread.currentThread();
|
parkedOperations++;
|
released = false;
|
lock.notifyAll();
|
try
|
{
|
while (!released && !deregistered)
|
{
|
if (!waitOnLock(deadline))
|
{
|
return LET_THROUGH;
|
}
|
}
|
return released ? releasedResultCode : LET_THROUGH;
|
}
|
finally
|
{
|
occupied = false;
|
lock.notifyAll();
|
}
|
}
|
}
|
|
/**
|
* Waits on the monitor until the provided deadline, reporting whether waiting can go
|
* on. A deadline which has passed gives up on this park altogether rather than only
|
* on the operation which reached it: the operations behind it would each pay the
|
* whole wait again otherwise.
|
*/
|
private boolean waitOnLock(long deadlineInNanos)
|
{
|
final long leftInNanos = deadlineInNanos - System.nanoTime();
|
if (leftInNanos <= 0)
|
{
|
giveUpOnTheTest();
|
return false;
|
}
|
try
|
{
|
// Rounded up, so that a budget shorter than a millisecond is still waited out
|
// rather than truncated to a wait with no timeout at all.
|
lock.wait(TimeUnit.NANOSECONDS.toMillis(leftInNanos + 999999L));
|
return true;
|
}
|
catch (InterruptedException e)
|
{
|
// Whatever wants this thread to stop wins over the park: let the operation run
|
// rather than hold a thread which is being taken down.
|
Thread.currentThread().interrupt();
|
return false;
|
}
|
}
|
|
/** Stops parking anything and says so, after a test held an operation for too long. */
|
private void giveUpOnTheTest()
|
{
|
System.err.println("***** ERROR: a replayed operation was parked on " + key
|
+ " for " + MAX_HOLD_IN_MS + " ms and was never released: the test which took"
|
+ " this park left it behind. Letting the operation run and parking no more.");
|
deregister();
|
}
|
|
/**
|
* Waits for a replayed operation which was not waited for yet to be parked, and
|
* reports which thread is replaying it. The operations are parked one at a time, so
|
* that thread is the one which was parked when this returns; the thread of the last
|
* park is reported when several of them were let go of without being waited for.
|
*
|
* @param timeout how long to wait for it
|
* @param unit the unit of the timeout
|
* @return the thread which is replaying the parked operation
|
* @throws InterruptedException if this thread is interrupted while waiting
|
* @throws TimeoutException if no operation was parked in time
|
* @throws IllegalStateException if this park is gone, so that nothing can be parked
|
* on it any more
|
*/
|
public Thread awaitParked(long timeout, TimeUnit unit)
|
throws InterruptedException, TimeoutException
|
{
|
final long deadline = System.nanoTime() + unit.toNanos(timeout);
|
synchronized (lock)
|
{
|
while (parkedOperations <= awaitedOperations)
|
{
|
if (deregistered)
|
{
|
// Waiting out the budget here would report a timeout naming the operations
|
// which never parked, rather than the park which cannot park them any more.
|
throw new IllegalStateException("the park on " + key + " is gone - it was"
|
+ " deregistered, or displaced by another park of the same operations -"
|
+ " so no replayed operation will be parked on it again");
|
}
|
final long leftInNanos = deadline - System.nanoTime();
|
if (leftInNanos <= 0)
|
{
|
throw new TimeoutException("no replayed operation was parked on " + key
|
+ " within " + timeout + " " + unit);
|
}
|
// Rounded up, so that a budget shorter than a millisecond is still waited out
|
// rather than truncated to a wait with no timeout at all.
|
lock.wait(TimeUnit.NANOSECONDS.toMillis(leftInNanos + 999999L));
|
}
|
awaitedOperations = parkedOperations;
|
return lastParkedThread;
|
}
|
}
|
|
/**
|
* Lets the parked operation run. Valid once {@link #awaitParked} has reported that
|
* operation: see there for what a release which arrives before it costs.
|
*/
|
public void release()
|
{
|
release(LET_THROUGH);
|
}
|
|
/**
|
* Lets the parked operation go, short circuiting it with the provided result code.
|
* <p>
|
* Valid once {@link #awaitParked} has reported the operation being released. A
|
* release which arrives before an operation is parked is wiped by the park it was
|
* meant for - a park starts out unreleased - and that operation then waits for a
|
* release which has already been spent.
|
*
|
* @param resultCode the result code the operation must report
|
*/
|
public void release(int resultCode)
|
{
|
synchronized (lock)
|
{
|
if (!occupied)
|
{
|
throw new IllegalStateException("nothing is parked on " + key + " to release:"
|
+ " a release is spent by the park it arrives before, and the operation"
|
+ " which parks next then waits for one which has already been given");
|
}
|
released = true;
|
releasedResultCode = resultCode;
|
lock.notifyAll();
|
}
|
}
|
|
/**
|
* Stops parking the replayed operations and lets go of the one which is parked, if
|
* any. A test must call this however it ends, or it leaves a replay thread of this
|
* server parked for good.
|
*/
|
public void deregister()
|
{
|
parks.remove(key, this);
|
synchronized (lock)
|
{
|
deregistered = true;
|
lock.notifyAll();
|
}
|
}
|
}
|
|
/**
|
* Parks the replayed operations of the given type at the given plugin point, until the
|
* test releases each of them.
|
*
|
* @param operation the type of operation to park
|
* @param section the plugin point to park them at, which can only be {@code PreParse}
|
* @param parked which of them to park - the change a test acts on rather than whatever
|
* of that type reaches this point first, which is somebody else's change as
|
* soon as more than one of them is in flight
|
* @return the park, which the test must {@link ParkedReplay#deregister()} when it is
|
* done with it
|
* @throws IllegalArgumentException if asked for any plugin point but {@code PreParse}
|
*/
|
public static ParkedReplay parkReplayedOperations(
|
OperationType operation, String section, Predicate<PluginOperation> parked)
|
{
|
if (!"PreParse".equalsIgnoreCase(section))
|
{
|
/*
|
* The pre-operation plugins are not invoked for synchronization operations at all,
|
* so a park anywhere else is never reached: the test which took it would wait out
|
* its whole budget for an operation which cannot park, and be told that none did
|
* rather than that none could.
|
*/
|
throw new IllegalArgumentException("replayed operations can only be parked at"
|
+ " PreParse, which is the only plugin point they reach, not at " + section);
|
}
|
final String key = keyFor(operation, section);
|
final ParkedReplay park = new ParkedReplay(key, parked);
|
final ParkedReplay previous = parks.put(key, park);
|
if (previous != null)
|
{
|
// A park a test left behind holds a replay thread of this server for good once the
|
// map stops pointing at it: let go of it rather than lose the last reference to it.
|
previous.deregister();
|
}
|
return park;
|
}
|
|
/** Returns the key a short circuit or a park of the given operations is kept under. */
|
private static String keyFor(OperationType operation, String section)
|
{
|
return operation + "/" + section.toLowerCase(Locale.ROOT);
|
}
|
}
|
