OPENDJ-1693 OPENDJ-1694 OPENDJ-1721 (CR-5748) addrate tool new version This...

Gaetan Boismal

15.26.2015 3070174573f3217558d17db2a57815963a2cb027

OPENDJ-1693 OPENDJ-1694 OPENDJ-1721 (CR-5748) addrate tool new version
This version includes
    *Bug fixes
        ** OPENDJ-1693: Solve NoSuchElementException by checking race conditions.
        ** OPENDJ-1694: Changes the age threshold feature implementation.
    *Improvement: Adding purge phase

 opendj-ldap-toolkit/src/main/java/com/forgerock/opendj/ldap/tools/AddRate.java

@@ -21,16 +21,29 @@
 * CDDL HEADER END
 *
 *
 *      Copyright 2014 ForgeRock AS
 *      Copyright 2014-2015 ForgeRock AS
 */

package com.forgerock.opendj.ldap.tools;

import static java.util.concurrent.TimeUnit.*;

import static org.forgerock.opendj.ldap.LdapException.*;
import static org.forgerock.opendj.ldap.ResultCode.*;
import static org.forgerock.opendj.ldap.requests.Requests.*;
import static org.forgerock.util.promise.Promises.*;

import static com.forgerock.opendj.cli.ArgumentConstants.*;
import static com.forgerock.opendj.cli.Utils.*;
import static com.forgerock.opendj.ldap.tools.ToolsMessages.*;

import java.io.IOException;
import java.io.PrintStream;
import java.util.Arrays;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.forgerock.i18n.LocalizableMessage;
@@ -40,9 +53,7 @@
import org.forgerock.opendj.ldap.LdapException;
import org.forgerock.opendj.ldap.ResultCode;
import org.forgerock.opendj.ldap.ResultHandler;
import org.forgerock.opendj.ldap.requests.AddRequest;
import org.forgerock.opendj.ldap.requests.DeleteRequest;
import org.forgerock.opendj.ldap.requests.Requests;
import org.forgerock.opendj.ldap.responses.Responses;
import org.forgerock.opendj.ldap.responses.Result;
import org.forgerock.opendj.ldif.EntryGenerator;
import org.forgerock.util.promise.Promise;
@@ -57,23 +68,25 @@
import com.forgerock.opendj.cli.MultiChoiceArgument;
import com.forgerock.opendj.cli.StringArgument;

import static org.forgerock.opendj.ldap.LdapException.*;

import static com.forgerock.opendj.cli.ArgumentConstants.*;
import static com.forgerock.opendj.cli.Utils.*;
import static com.forgerock.opendj.ldap.tools.ToolsMessages.*;

/**
 * A load generation tool that can be used to load a Directory Server with Add
 * and Delete requests using one or more LDAP connections.
 */
public class AddRate extends ConsoleApplication {

    private static final class AddPerformanceRunner extends PerformanceRunner {
    @SuppressWarnings("serial")
    private static final class AddRateExecutionEndedException extends LdapException {
        private AddRateExecutionEndedException() {
            super(Responses.newResult(OTHER));
        }
    }

    private final class AddPerformanceRunner extends PerformanceRunner {

        private final class AddStatsHandler extends UpdateStatsResultHandler<Result> {
            private final String entryDN;

            private AddStatsHandler(final long currentTime, String entryDN) {
            private AddStatsHandler(final long currentTime, final String entryDN) {
                super(currentTime);
                this.entryDN = entryDN;
            }
@@ -87,11 +100,11 @@
                    long newKey;
                    do {
                        newKey = randomSeq.get().nextInt();
                    } while (dnEntriesAdded.putIfAbsent(newKey, this.entryDN) != null);
                    } while (dnEntriesAdded.putIfAbsent(newKey, entryDN) != null);
                    break;
                case FIFO:
                    long uniqueTime = currentTime;
                    while (dnEntriesAdded.putIfAbsent(uniqueTime, this.entryDN) != null) {
                    while (dnEntriesAdded.putIfAbsent(uniqueTime, entryDN) != null) {
                        uniqueTime++;
                    }
                    break;
@@ -99,12 +112,13 @@
                    break;
                }

                nbAdd.getAndIncrement();
                recentAdds.getAndIncrement();
                totalAdds.getAndIncrement();
                entryCount.getAndIncrement();
            }
        }

        private final class DeleteStatsHandler extends UpdateStatsResultHandler<Result> {

            private DeleteStatsHandler(final long startTime) {
                super(startTime);
            }
@@ -112,7 +126,8 @@
            @Override
            public void handleResult(final Result result) {
                super.handleResult(result);
                nbDelete.getAndIncrement();
                recentDeletes.getAndIncrement();
                entryCount.getAndDecrement();
            }
        }

@@ -126,96 +141,139 @@
            @Override
            void resetStats() {
                super.resetStats();
                nbAdd.set(0);
                nbDelete.set(0);
                recentAdds.set(0);
                recentDeletes.set(0);
            }

            @Override
            String[] getAdditionalColumns() {
                final int nbAddStat = nbAdd.getAndSet(0);
                final int nbDelStat = nbDelete.getAndSet(0);
                final int total = nbAddStat + nbDelStat;
                final int adds = recentAdds.getAndSet(0);
                final int deleteStat = recentDeletes.getAndSet(0);
                final int total = adds + deleteStat;

                extraColumn[0] = String.format("%.2f", total > 0 ? ((double) nbAddStat / total) * 100 : 0.0);
                extraColumn[0] = String.format("%.2f", total > 0 ? ((double) adds / total) * 100 : 0.0);

                return extraColumn;
            }
        }

        private final class AddWorkerThread extends WorkerThread {

            AddWorkerThread(Connection connection, ConnectionFactory connectionFactory) {
        private final class AddDeleteWorkerThread extends WorkerThread {
            private AddDeleteWorkerThread(final Connection connection, final ConnectionFactory connectionFactory) {
                super(connection, connectionFactory);
            }

            @Override
            public Promise<?, LdapException> performOperation(Connection connection, DataSource[] dataSources,
                    long currentTime) {
                if (needsDelete(currentTime)) {
                    DeleteRequest dr = Requests.newDeleteRequest(getDNEntryToRemove());
                    ResultHandler<Result> deleteHandler = new DeleteStatsHandler(currentTime);

                    return connection.deleteAsync(dr).onSuccess(deleteHandler).onFailure(deleteHandler);
                } else {
                    return performAddOperation(connection, currentTime);
                }
            }

            private Promise<Result, LdapException> performAddOperation(Connection connection, long currentTime) {
            public Promise<?, LdapException> performOperation(
                    final Connection connection, final DataSource[] dataSources, final long currentTime) {
                startPurgeIfMaxNumberAddReached();
                startToggleDeleteIfAgeThresholdReached(currentTime);
                try {
                    Entry entry;
                    synchronized (generator) {
                        entry = generator.readEntry();
                    String entryToRemove = getEntryToRemove(currentTime);
                    if (entryToRemove != null) {
                        return doDelete(connection, currentTime, entryToRemove);
                    }

                    AddRequest ar = Requests.newAddRequest(entry);
                    ResultHandler<Result> addHandler = new AddStatsHandler(currentTime, entry.getName().toString());
                    return connection.addAsync(ar).onSuccess(addHandler).onFailure(addHandler);
                } catch (IOException e) {
                    // faking an error result by notifying the Handler
                    UpdateStatsResultHandler<Result> resHandler = new UpdateStatsResultHandler<Result>(currentTime);
                    resHandler.handleError(newLdapException(ResultCode.OTHER, e));
                    return null;
                    return doAdd(connection, currentTime);
                } catch (final AddRateExecutionEndedException a) {
                    return newSuccessfulPromise(OTHER);
                } catch (final IOException e) {
                    return newFailedPromise(newLdapException(OTHER, e));
                }
            }

            private boolean needsDelete(final long currentTime) {
                if (dnEntriesAdded.isEmpty() || delStrategy == DeleteStrategy.OFF) {
                    return false;
                }

                switch (delThreshold) {
                case SIZE_THRESHOLD:
                    return dnEntriesAdded.size() > sizeThreshold;
                case AGE_THRESHOLD:
                    long olderEntryTimestamp = dnEntriesAdded.firstKey();
                    return (olderEntryTimestamp + timeToWait) < currentTime;
                default:
                    return false;
            private void startToggleDeleteIfAgeThresholdReached(long currentTime) {
                if (!toggleDelete
                        && delThreshold == DeleteThreshold.AGE_THRESHOLD
                        && !dnEntriesAdded.isEmpty()
                        && dnEntriesAdded.firstKey() + timeToWait < currentTime) {
                    setSizeThreshold(entryCount.get());
                }
            }

            private String getDNEntryToRemove() {
                String removedEntry = null;

                while (removedEntry == null) {
                    long minKey = dnEntriesAdded.firstKey();
                    long maxKey = dnEntriesAdded.lastKey();
                    long randomIndex = Math.round(Math.random() * (maxKey - minKey) + minKey);
                    Long key = dnEntriesAdded.ceilingKey(randomIndex);

                    if (key != null) {
                        removedEntry = dnEntriesAdded.remove(key);
            private void startPurgeIfMaxNumberAddReached() {
                AtomicBoolean purgeLatch = new AtomicBoolean();
                if (!isPurgeBranchRunning.get()
                            && 0 < maxNbAddIterations && maxNbAddIterations < totalAdds.get()) {
                    if (purgeLatch.compareAndSet(false, true)) {
                        newPurgerThread().start();
                    }
                }

                return removedEntry;
            }

            // FIXME Followings @Checkstyle:ignore tags are related to the maven-checkstyle-plugin
            // issue related here: https://github.com/checkstyle/checkstyle/issues/5
            // @Checkstyle:ignore
            private String getEntryToRemove(final long currentTime) throws AddRateExecutionEndedException {
                if (isPurgeBranchRunning.get()) {
                    return purgeEntry();
                }

                if (toggleDelete && entryCount.get() > sizeThreshold) {
                    return removeFirstAddedEntry();
                }

                return null;
            }

            // @Checkstyle:ignore
            private String purgeEntry() throws AddRateExecutionEndedException {
                if (!dnEntriesAdded.isEmpty()) {
                    return removeFirstAddedEntry();
                }
                localStopRequested = true;
                throw new AddRateExecutionEndedException();
            }

            private String removeFirstAddedEntry() {
                final Map.Entry<Long, String> entry = dnEntriesAdded.pollFirstEntry();
                return entry != null ? entry.getValue() : null;
            }

            private Promise<Result, LdapException> doAdd(
                    final Connection connection, final long currentTime) throws IOException {
                Entry entry;
                synchronized (generator) {
                    entry = generator.readEntry();
                }

                final ResultHandler<Result> addHandler = new AddStatsHandler(currentTime, entry.getName().toString());
                return connection.addAsync(newAddRequest(entry))
                                 .onSuccess(addHandler)
                                 .onFailure(addHandler);
            }

            private Promise<?, LdapException> doDelete(
                    final Connection connection, final long currentTime, final String entryToRemove) {
                final ResultHandler<Result> deleteHandler = new DeleteStatsHandler(currentTime);
                return connection.deleteAsync(newDeleteRequest(entryToRemove))
                                 .onSuccess(deleteHandler)
                                 .onFailure(deleteHandler);
            }
        }

        private final ConcurrentSkipListMap<Long, String> dnEntriesAdded =
            new ConcurrentSkipListMap<Long, String>();
        private final class AddRateTimerThread extends TimerThread {
            private AddRateTimerThread(final long timeToWait) {
                super(timeToWait);
            }

            @Override
            void performStopOperations() {
                if (purgeEnabled && isPurgeBranchRunning.compareAndSet(false, true)) {
                    if (!isScriptFriendly()) {
                        println(LocalizableMessage.raw("Purge phase..."));
                    }
                    try {
                        joinAllWorkerThreads();
                    } catch (final InterruptedException e) {
                        throw new IllegalStateException();
                    }
                } else if (!purgeEnabled) {
                    stopRequested = true;
                }
            }
        }

        private final ConcurrentSkipListMap<Long, String> dnEntriesAdded = new ConcurrentSkipListMap<Long, String>();
        private final ThreadLocal<Random> randomSeq = new ThreadLocal<Random>() {
            @Override
            protected Random initialValue() {
@@ -226,18 +284,25 @@
        private EntryGenerator generator;
        private DeleteStrategy delStrategy;
        private DeleteThreshold delThreshold;
        private Integer sizeThreshold;
        private int sizeThreshold;
        private volatile boolean toggleDelete;
        private long timeToWait;
        private final AtomicInteger nbAdd = new AtomicInteger();
        private final AtomicInteger nbDelete = new AtomicInteger();
        private int maxNbAddIterations;
        private boolean purgeEnabled;
        private final AtomicInteger recentAdds = new AtomicInteger();
        private final AtomicInteger recentDeletes = new AtomicInteger();
        private final AtomicInteger totalAdds = new AtomicInteger();
        private final AtomicInteger entryCount = new AtomicInteger();
        private final AtomicBoolean isPurgeBranchRunning = new AtomicBoolean();

        private AddPerformanceRunner(final PerformanceRunnerOptions options) throws ArgumentException {
            super(options);
            maxIterationsArgument.setPropertyName("maxNumberOfAdd");
        }

        @Override
        WorkerThread newWorkerThread(Connection connection, ConnectionFactory connectionFactory) {
            return new AddWorkerThread(connection, connectionFactory);
        WorkerThread newWorkerThread(final Connection connection, final ConnectionFactory connectionFactory) {
            return new AddDeleteWorkerThread(connection, connectionFactory);
        }

        @Override
@@ -245,10 +310,23 @@
            return new AddRateStatsThread();
        }

        public void validate(MultiChoiceArgument<DeleteStrategy> delModeArg, IntegerArgument delSizeThresholdArg,
                                IntegerArgument delAgeThresholdArg) throws ArgumentException {
        @Override
        TimerThread newEndTimerThread(final long timeTowait) {
            return new AddRateTimerThread(timeTowait);
        }

        TimerThread newPurgerThread() {
            return newEndTimerThread(0);
        }

        public void validate(final MultiChoiceArgument<DeleteStrategy> delModeArg,
                final IntegerArgument delSizeThresholdArg, final IntegerArgument delAgeThresholdArg,
                final BooleanArgument noPurgeArgument) throws ArgumentException {
            super.validate();
            delStrategy = delModeArg.getTypedValue();
            maxNbAddIterations = maxIterationsArgument.getIntValue();
            purgeEnabled = !noPurgeArgument.isPresent();

            // Check for inconsistent use cases
            if (delSizeThresholdArg.isPresent() && delAgeThresholdArg.isPresent()) {
                throw new ArgumentException(ERR_ADDRATE_THRESHOLD_SIZE_AND_AGE.get());
@@ -263,12 +341,20 @@
                delThreshold =
                    delAgeThresholdArg.isPresent() ? DeleteThreshold.AGE_THRESHOLD : DeleteThreshold.SIZE_THRESHOLD;
                if (delThreshold == DeleteThreshold.SIZE_THRESHOLD) {
                    sizeThreshold = delSizeThresholdArg.getIntValue();
                    setSizeThreshold(delSizeThresholdArg.getIntValue());
                    if (0 < maxNbAddIterations && maxNbAddIterations < sizeThreshold) {
                        throw new ArgumentException(ERR_ADDRATE_SIZE_THRESHOLD_LOWER_THAN_ITERATIONS.get());
                    }
                } else {
                    timeToWait = delAgeThresholdArg.getIntValue() * 1000000000L;
                    timeToWait = NANOSECONDS.convert(delAgeThresholdArg.getIntValue(), SECONDS);
                }
            }
        }

        private void setSizeThreshold(int entriesSizeThreshold) {
            sizeThreshold = entriesSizeThreshold;
            toggleDelete = true;
        }
    }

    private enum DeleteStrategy {
@@ -280,6 +366,11 @@
    }

    private static final int EXIT_CODE_SUCCESS = 0;
    private static final int DEFAULT_SIZE_THRESHOLD = 10000;
    /** The minimum time to wait before starting add/delete phase (in seconds). */
    private static final int AGE_THRESHOLD_LOWERBOUND = 1;
    /** The minimum number of entries to add before starting add/delete phase. */
    private static final int SIZE_THRESHOLD_LOWERBOUND = 1;

    /**
     * The main method for AddRate tool.
@@ -293,14 +384,13 @@
    }

    private BooleanArgument verbose;

    private BooleanArgument scriptFriendly;

    private AddRate() {
        // Nothing to do
    }

    AddRate(PrintStream out, PrintStream err) {
    AddRate(final PrintStream out, final PrintStream err) {
        super(out, err);
    }

@@ -328,23 +418,24 @@
        final ArgumentParser argParser =
            new ArgumentParser(AddRate.class.getName(), toolDescription, false, true, 1, 1, "template-file-path");

        ConnectionFactoryProvider connectionFactoryProvider;
        ConnectionFactory connectionFactory;
        AddPerformanceRunner runner;
        final ConnectionFactoryProvider connectionFactoryProvider;
        final ConnectionFactory connectionFactory;
        final AddPerformanceRunner runner;

        /* Entries generation parameters */
        IntegerArgument randomSeedArg;
        StringArgument resourcePathArg;
        StringArgument constantsArg;
        final IntegerArgument randomSeedArg;
        final StringArgument resourcePathArg;
        final StringArgument constantsArg;

        /* addrate specifics arguments */
        MultiChoiceArgument<DeleteStrategy> deleteMode;
        IntegerArgument deleteSizeThreshold;
        IntegerArgument deleteAgeThreshold;
        final MultiChoiceArgument<DeleteStrategy> deleteMode;
        final IntegerArgument deleteSizeThreshold;
        final IntegerArgument deleteAgeThreshold;
        final BooleanArgument noPurgeArgument;

        try {
            Utils.setDefaultPerfToolProperties();
            PerformanceRunnerOptions options = new PerformanceRunnerOptions(argParser, this);
            final PerformanceRunnerOptions options = new PerformanceRunnerOptions(argParser, this);
            options.setSupportsGeneratorArgument(false);

            connectionFactoryProvider = new ConnectionFactoryProvider(argParser, this);
@@ -377,15 +468,21 @@
            argParser.addArgument(deleteMode);

            deleteSizeThreshold =
                new IntegerArgument("deletesizethreshold", 's', "deleteSizeThreshold", false, true,
                    INFO_DELETESIZETHRESHOLD_PLACEHOLDER.get(), INFO_ADDRATE_DESCRIPTION_DELETESIZETHRESHOLD.get());
            deleteSizeThreshold.setDefaultValue(String.valueOf(10000));
                new IntegerArgument("deletesizethreshold", 's', "deleteSizeThreshold", false, false, true,
                    INFO_DELETESIZETHRESHOLD_PLACEHOLDER.get(), DEFAULT_SIZE_THRESHOLD, "deleteSizeThreshold", true,
                    SIZE_THRESHOLD_LOWERBOUND, false, Integer.MAX_VALUE,
                    INFO_ADDRATE_DESCRIPTION_DELETESIZETHRESHOLD.get());
            argParser.addArgument(deleteSizeThreshold);

            deleteAgeThreshold =
                new IntegerArgument("deleteagethreshold", 'a', "deleteAgeThreshold", false, true,
                    INFO_DELETEAGETHRESHOLD_PLACEHOLDER.get(), INFO_ADDRATE_DESCRIPTION_DELETEAGETHRESHOLD.get());
                    INFO_DELETEAGETHRESHOLD_PLACEHOLDER.get(), true, AGE_THRESHOLD_LOWERBOUND, false,
                    Integer.MAX_VALUE, INFO_ADDRATE_DESCRIPTION_DELETEAGETHRESHOLD.get());
            deleteAgeThreshold.setPropertyName(deleteAgeThreshold.getLongIdentifier());
            argParser.addArgument(deleteAgeThreshold);

            noPurgeArgument = new BooleanArgument("nopurge", 'n', "noPurge", INFO_ADDRATE_DESCRIPTION_NOPURGE.get());
            argParser.addArgument(noPurgeArgument);
        } catch (final ArgumentException ae) {
            errPrintln(ERR_CANNOT_INITIALIZE_ARGS.get(ae.getMessage()));
            return ResultCode.CLIENT_SIDE_PARAM_ERROR.intValue();
@@ -400,32 +497,31 @@
            }

            connectionFactory = connectionFactoryProvider.getAuthenticatedConnectionFactory();
            runner.validate(deleteMode, deleteSizeThreshold, deleteAgeThreshold);
            runner.validate(deleteMode, deleteSizeThreshold, deleteAgeThreshold, noPurgeArgument);
        } catch (final ArgumentException ae) {
            final LocalizableMessage message = ERR_ERROR_PARSING_ARGS.get(ae.getMessage());
            errPrintln(message);
            errPrintln(ERR_ERROR_PARSING_ARGS.get(ae.getMessage()));
            errPrintln(argParser.getUsageMessage());
            return ResultCode.CLIENT_SIDE_PARAM_ERROR.intValue();
        }

        final String templatePath = argParser.getTrailingArguments().get(0);

        runner.generator =
            MakeLDIF.createGenerator(templatePath, resourcePathArg, randomSeedArg, constantsArg, false, this);
        Runtime.getRuntime().addShutdownHook(runner.newPurgerThread());

        return runner.run(connectionFactory);
    }

    private void addCommonArguments(ArgumentParser argParser) throws ArgumentException {
        StringArgument propertiesFileArgument = CommonArguments.getPropertiesFile();
    private void addCommonArguments(final ArgumentParser argParser) throws ArgumentException {
        final StringArgument propertiesFileArgument = CommonArguments.getPropertiesFile();
        argParser.addArgument(propertiesFileArgument);
        argParser.setFilePropertiesArgument(propertiesFileArgument);

        BooleanArgument noPropertiesFileArgument = CommonArguments.getNoPropertiesFile();
        final BooleanArgument noPropertiesFileArgument = CommonArguments.getNoPropertiesFile();
        argParser.addArgument(noPropertiesFileArgument);
        argParser.setNoPropertiesFileArgument(noPropertiesFileArgument);

        BooleanArgument showUsage = CommonArguments.getShowUsage();
        final BooleanArgument showUsage = CommonArguments.getShowUsage();
        argParser.addArgument(showUsage);
        argParser.setUsageArgument(showUsage, getOutputStream());


 opendj-ldap-toolkit/src/main/java/com/forgerock/opendj/ldap/tools/AuthRate.java

@@ -22,7 +22,7 @@
 *
 *
 *      Copyright 2010 Sun Microsystems, Inc.
 *      Portions Copyright 2011-2014 ForgeRock AS
 *      Portions Copyright 2011-2015 ForgeRock AS
 */
package com.forgerock.opendj.ldap.tools;

@@ -178,6 +178,7 @@
                    returnedPromise = performBind(connection, data);
                }

                incrementIterationCount();
                return returnedPromise.onSuccess(new UpdateStatsResultHandler<BindResult>(startTime)).onFailure(
                        new BindUpdateStatsResultHandler(startTime));
            }

 opendj-ldap-toolkit/src/main/java/com/forgerock/opendj/ldap/tools/ModRate.java

@@ -22,7 +22,7 @@
 *
 *
 *      Copyright 2010 Sun Microsystems, Inc.
 *      Portions Copyright 2011-2014 ForgeRock AS
 *      Portions Copyright 2011-2015 ForgeRock AS
 */
package com.forgerock.opendj.ldap.tools;

@@ -74,6 +74,7 @@
                mr = newModifyRequest(data);
                ResultHandler<Result> modRes = new UpdateStatsResultHandler<Result>(startTime);

                incrementIterationCount();
                return connection.modifyAsync(mr).onSuccess(modRes).onFailure(modRes);
            }


 opendj-ldap-toolkit/src/main/java/com/forgerock/opendj/ldap/tools/PerformanceRunner.java

@@ -22,7 +22,7 @@
 *
 *
 *      Copyright 2010 Sun Microsystems, Inc.
 *      Portions Copyright 2011-2014 ForgeRock AS.
 *      Portions Copyright 2011-2015 ForgeRock AS.
 */
package com.forgerock.opendj.ldap.tools;

@@ -440,6 +440,7 @@
            return EMPTY_STRINGS;
        }

        /** Resets both general and recent statistic indicators. */
        void resetStats() {
            failedCount = 0;
            operationCount = 0;
@@ -451,13 +452,17 @@
        }
    }

    private class TimerThread extends Thread {
    class TimerThread extends Thread {
        private final long timeToWait;

        public TimerThread(long timeToWait) {
        TimerThread(long timeToWait) {
            this.timeToWait = timeToWait;
        }

        void performStopOperations() {
            stopRequested = true;
        }

        @Override
        public void run() {
            try {
@@ -465,7 +470,7 @@
            } catch (InterruptedException e) {
                throw new IllegalStateException(e);
            } finally {
                stopRequested = true;
                performStopOperations();
            }
        }
    }
@@ -512,6 +517,7 @@
        private int count;
        private final Connection connection;
        private final ConnectionFactory connectionFactory;
        boolean localStopRequested;

        WorkerThread(final Connection connection, final ConnectionFactory connectionFactory) {
            super("Worker Thread");
@@ -526,11 +532,11 @@
        public void run() {
            Promise<?, LdapException> promise;
            Connection connection;

            final double targetTimeInMS = 1000.0 / (targetThroughput / (double) (numThreads * numConnections));
            double sleepTimeInMS = 0;
            long start;
            while (!stopRequested && !(maxIterations > 0 && count >= maxIterations)) {

            while (!stopRequested && !localStopRequested
                    && (maxIterations <= 0 || count < maxIterations)) {
                if (this.connection == null) {
                    try {
                        connection = connectionFactory.getConnectionAsync().getOrThrow();
@@ -565,15 +571,12 @@
                    }
                }

                start = System.nanoTime();
                long start = System.nanoTime();
                promise = performOperation(connection, dataSources.get(), start);
                operationRecentCount.getAndIncrement();
                count++;
                if (!isAsync) {
                    try {
                        if (promise != null) {
                            promise.getOrThrow();
                        }
                        promise.getOrThrow();
                    } catch (final InterruptedException e) {
                        // Ignore and check stop requested
                        continue;
@@ -607,6 +610,10 @@
                }
            }
        }

        void incrementIterationCount() {
            count++;
        }
    }

    private static final String[] EMPTY_STRINGS = new String[0];
@@ -636,10 +643,10 @@

    };

    private volatile boolean stopRequested;
    int numThreads;
    int numConnections;
    volatile boolean stopRequested;
    private volatile boolean isWarmingUp;
    private int numThreads;
    private int numConnections;
    private int targetThroughput;
    private int maxIterations;
    /** Warm-up duration time in ms. **/
@@ -650,7 +657,6 @@
    private boolean noRebind;
    private int statsInterval;
    private final IntegerArgument numThreadsArgument;
    private final IntegerArgument maxIterationsArgument;
    private final IntegerArgument maxDurationArgument;
    private final IntegerArgument statsIntervalArgument;
    private final IntegerArgument targetThroughputArgument;
@@ -660,8 +666,11 @@
    private final BooleanArgument noRebindArgument;
    private final BooleanArgument asyncArgument;
    private final StringArgument arguments;
    protected final IntegerArgument maxIterationsArgument;
    protected final IntegerArgument warmUpArgument;

    private final List<Thread> workerThreads = new ArrayList<Thread>();

    PerformanceRunner(final PerformanceRunnerOptions options) throws ArgumentException {
        ArgumentParser argParser = options.getArgumentParser();

@@ -686,8 +695,8 @@

        maxIterationsArgument =
                new IntegerArgument("maxIterations", 'm', "maxIterations", false, false, true,
                        LocalizableMessage.raw("{maxIterations}"), 0, null, LocalizableMessage
                                .raw("Max iterations, 0 for unlimited"));
                        LocalizableMessage.raw("{maxIterations}"), 0, null,
                        LocalizableMessage.raw("Max iterations, 0 for unlimited"));
        maxIterationsArgument.setPropertyName("maxIterations");
        argParser.addArgument(maxIterationsArgument);

@@ -828,19 +837,19 @@

    final DataSource[] getDataSources() {
        if (dataSourcePrototypes == null) {
            throw new IllegalStateException(
                    "dataSources are null - validate() must be called first");
            throw new IllegalStateException("dataSources are null - validate() must be called first");
        }
        return dataSourcePrototypes;
    }

    abstract WorkerThread newWorkerThread(final Connection connection,
            final ConnectionFactory connectionFactory);

    abstract WorkerThread newWorkerThread(final Connection connection, final ConnectionFactory connectionFactory);
    abstract StatsThread newStatsThread();

    TimerThread newEndTimerThread(final long timeTowait) {
        return new TimerThread(timeTowait);
    }

    final int run(final ConnectionFactory connectionFactory) {
        final List<Thread> threads = new ArrayList<Thread>();
        final List<Connection> connections = new ArrayList<Connection>();

        Connection connection = null;
@@ -854,13 +863,13 @@
                }
                for (int j = 0; j < numThreads; j++) {
                    final Thread thread = newWorkerThread(connection, connectionFactory);
                    threads.add(thread);
                    workerThreads.add(thread);
                    thread.start();
                }
            }

            if (maxDurationTime > 0) {
                new TimerThread(maxDurationTime).start();
                newEndTimerThread(maxDurationTime).start();
            }

            final StatsThread statsThread = newStatsThread();
@@ -873,11 +882,9 @@
                statsThread.resetStats();
                isWarmingUp = false;
            }
            statsThread.start();

            for (final Thread t : threads) {
                t.join();
            }
            statsThread.start();
            joinAllWorkerThreads();
            stopRequested = true;
            statsThread.join();
        } catch (final InterruptedException e) {
@@ -891,4 +898,10 @@

        return 0;
    }

    protected void joinAllWorkerThreads() throws InterruptedException {
        for (final Thread t : workerThreads) {
            t.join();
        }
    }
}

 opendj-ldap-toolkit/src/main/java/com/forgerock/opendj/ldap/tools/SearchRate.java

@@ -22,7 +22,7 @@
 *
 *
 *      Copyright 2010 Sun Microsystems, Inc.
 *      Portions Copyright 2011-2014 ForgeRock AS
 *      Portions Copyright 2011-2015 ForgeRock AS
 */
package com.forgerock.opendj.ldap.tools;

@@ -137,7 +137,7 @@
                }

                final SearchStatsHandler handler = new SearchStatsHandler(startTime);

                incrementIterationCount();
                return connection.searchAsync(sr, handler).onSuccess(handler).onFailure(handler);
            }
        }

 opendj-ldap-toolkit/src/main/resources/com/forgerock/opendj/ldap/tools/tools.properties

@@ -22,7 +22,7 @@
#
#
#      Copyright 2010 Sun Microsystems, Inc.
#      Portions copyright 2012-2014 ForgeRock AS.
#      Portions copyright 2012-2015 ForgeRock AS.
#
#
# Utility messages
@@ -536,3 +536,6 @@
 specified, but only only one at a time is supported
ERR_ADDRATE_DELMODE_RAND_THRESHOLD_AGE=A age based deletion threshold should not be used \
 with a random deletion mode
ERR_ADDRATE_SIZE_THRESHOLD_LOWER_THAN_ITERATIONS=The size threshold must be lower than \
 the maximum number of add operations
INFO_ADDRATE_DESCRIPTION_NOPURGE=Disable the purge phase when the tool stops.

 opendj-ldap-toolkit/src/test/java/com/forgerock/opendj/ldap/tools/AddRateITCase.java

@@ -21,55 +21,54 @@
 * CDDL HEADER END
 *
 *
 *      Copyright 2014 ForgeRock AS.
 *      Copyright 2014-2015 ForgeRock AS.
 */
package com.forgerock.opendj.ldap.tools;

import static com.forgerock.opendj.ldap.tools.ToolsMessages.ERR_ADDRATE_DELMODE_OFF_THRESHOLD_ON;
import static com.forgerock.opendj.ldap.tools.ToolsMessages.ERR_ADDRATE_DELMODE_RAND_THRESHOLD_AGE;
import static com.forgerock.opendj.ldap.tools.ToolsMessages.ERR_ADDRATE_THRESHOLD_SIZE_AND_AGE;
import static com.forgerock.opendj.ldap.tools.ToolsMessages.INFO_ADDRATE_TOOL_DESCRIPTION;
import static org.fest.assertions.Assertions.assertThat;
import static org.forgerock.util.Utils.closeSilently;
import static org.fest.assertions.Assertions.*;
import static org.forgerock.util.Utils.*;

import static com.forgerock.opendj.ldap.tools.ToolsMessages.*;

import java.io.PrintStream;

import org.forgerock.opendj.ldap.ByteStringBuilder;
import org.forgerock.opendj.ldap.TestCaseUtils;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;

public class AddRateITCase extends ToolsITCase {

    private static final String TEMPLATE_NAME = "addrate.template";
    private static final String ADD_PERCENT_TEXT = "Add%";
    private static final int THROUGHPUT_COLUMN = 1;
    private static final int ERR_PER_SEC_COLUMN_NUMBER = 8;

    @DataProvider
    public Object[][] addRateArgs() throws Exception {
        String[] commonArgs =
            args("-h", TestCaseUtils.getServerSocketAddress().getHostName(),
                "-p", Integer.toString(TestCaseUtils.getServerSocketAddress().getPort()),
                "-c", "1", "-t", "1", "-i", "1", "-m", "1000", "-S");
    private ByteStringBuilder out;
    private ByteStringBuilder err;
    private PrintStream outStream;
    private PrintStream errStream;

        return new Object[][] {
            // Check if the help message is correctly prompted
            { args("-H"), INFO_ADDRATE_TOOL_DESCRIPTION.get(), "" },
    @BeforeMethod
    private void refreshStreams() {
        out = new ByteStringBuilder();
        err = new ByteStringBuilder();
        outStream = new PrintStream(out.asOutputStream());
        errStream = new PrintStream(err.asOutputStream());
    }

            // Should report inconsistent use of options
            { args(commonArgs, "-C", "off", "-a", "3", TEMPLATE_NAME), "",
                ERR_ADDRATE_DELMODE_OFF_THRESHOLD_ON.get() },
            { args(commonArgs, "-C", "off", "-s", "30000", TEMPLATE_NAME), "",
                ERR_ADDRATE_DELMODE_OFF_THRESHOLD_ON.get() },
            { args(commonArgs, "-C", "fifo", "-a", "3", "-s", "20000", TEMPLATE_NAME), "",
                ERR_ADDRATE_THRESHOLD_SIZE_AND_AGE.get() },
            { args(commonArgs, "-C", "random", "-a", "3", TEMPLATE_NAME), "",
                ERR_ADDRATE_DELMODE_RAND_THRESHOLD_AGE.get() },
    @AfterMethod
    private void closeStreams() {
        closeSilently(outStream, errStream);
    }

            // Correct test case
            { args(commonArgs, "-s", "10", TEMPLATE_NAME), ADD_PERCENT_TEXT, "" },

        };
    private String[] commonsArgs() {
        return new String[] {
            "-h", TestCaseUtils.getServerSocketAddress().getHostName(),
            "-p", Integer.toString(TestCaseUtils.getServerSocketAddress().getPort()),
            "-c", "1", "-t", "1", "-i", "1", "-m", "100", "-S"};
    }

    public String[] args(String[] startLine, String... args) {
@@ -81,34 +80,51 @@
        return res;
    }

    @Test(dataProvider = "addRateArgs")
    public void testITAddRate(String[] arguments, Object expectedOut, Object expectedErr) throws Exception {
        ByteStringBuilder out = new ByteStringBuilder();
        ByteStringBuilder err = new ByteStringBuilder();
    @DataProvider
    public Object[][] invalidAddRateArgs() throws Exception {
        return new Object[][] {
            // Should report inconsistent use of options
            { args("-C", "off", "-a", "3", TEMPLATE_NAME), ERR_ADDRATE_DELMODE_OFF_THRESHOLD_ON.get() },
            { args("-C", "off", "-s", "30000", TEMPLATE_NAME), ERR_ADDRATE_DELMODE_OFF_THRESHOLD_ON.get() },
            { args("-C", "fifo", "-a", "3", "-s", "20000", TEMPLATE_NAME), ERR_ADDRATE_THRESHOLD_SIZE_AND_AGE.get() },
            { args("-C", "random", "-a", "3", TEMPLATE_NAME), ERR_ADDRATE_DELMODE_RAND_THRESHOLD_AGE.get() },
            { args("-s", "999", TEMPLATE_NAME), ERR_ADDRATE_SIZE_THRESHOLD_LOWER_THAN_ITERATIONS.get() }
        };
    }

        PrintStream outStream = new PrintStream(out.asOutputStream());
        PrintStream errStream = new PrintStream(err.asOutputStream());
    @Test(dataProvider = "invalidAddRateArgs")
    public void addRateExceptions(String[] arguments, Object expectedErr) throws Exception {
        AddRate addRate = new AddRate(outStream, errStream);
        int retCode = addRate.run(args(commonsArgs(), arguments));
        checkOuputStreams(out, err, "", expectedErr);
        assertThat(retCode).isNotEqualTo(0);
    }

        try {
            AddRate addRate = new AddRate(outStream, errStream);
            addRate.run(arguments);
    @Test
    public void addRateDisplaysHelp() throws Exception {
        AddRate addRate = new AddRate(outStream, errStream);
        int retCode = addRate.run(args("-H"));
        checkOuputStreams(out, err, INFO_ADDRATE_TOOL_DESCRIPTION.get(), "");
        assertThat(retCode).isEqualTo(0);
    }

            checkOuputStreams(out, err, expectedOut, expectedErr);
            String outContent = out.toString();
    @Test(timeOut = 10000)
    public void addRateSimpleRun() throws Exception {
        AddRate addRate = new AddRate(outStream, errStream);
        int retCode = addRate.run(args(commonsArgs(), "-s", "10", TEMPLATE_NAME));
        checkOuputStreams(out, err, ADD_PERCENT_TEXT, "");
        assertThat(retCode).isEqualTo(0);
        String outContent = out.toString();

            if (outContent.contains(ADD_PERCENT_TEXT)) {
                // Check that there was no error in search
                String[] addRateResLines = outContent.split(System.getProperty("line.separator"));
                // Skip header line
                for (int i = 1; i < addRateResLines.length; i++) {
                    String[] addhRateLineData = addRateResLines[i].split(",");
                    assertThat(addhRateLineData[ERR_PER_SEC_COLUMN_NUMBER].trim()).isEqualTo("0.0");
                    assertThat(addhRateLineData[THROUGHPUT_COLUMN].trim()).isNotEqualTo("0.0");
                }

        if (outContent.contains(ADD_PERCENT_TEXT)) {
            // Check that there was no error
            String[] addRateResLines = outContent.split(System.getProperty("line.separator"));
            // Skip header line
            for (int i = 1; i < addRateResLines.length; i++) {
                String[] lineData = addRateResLines[i].split(",");
                assertThat(lineData[ERR_PER_SEC_COLUMN_NUMBER].trim()).isEqualTo("0.0");
                assertThat(lineData[THROUGHPUT_COLUMN].trim()).isNotEqualTo("0.0");
            }
        } finally {
            closeSilently(outStream, errStream);
        }
    }


 opendj-ldap-toolkit/src/test/java/com/forgerock/opendj/ldap/tools/ToolsITCase.java

@@ -21,7 +21,7 @@
 * CDDL HEADER END
 *
 *
 *      Copyright 2014 ForgeRock AS.
 *      Copyright 2014-2015 ForgeRock AS.
 */
package com.forgerock.opendj.ldap.tools;

@@ -72,7 +72,7 @@
        checkOutputStream(err, expectedError);
    }

    private void checkOutputStream(ByteStringBuilder out, Object expectedOutput) {
    protected void checkOutputStream(ByteStringBuilder out, Object expectedOutput) {
        String lineSeparator = System.getProperty("line.separator");
        String toCompare = expectedOutput.toString();