<commit_before>/*
 * Copyright (c) 2008-2015, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hazelcast.simulator.coordinator;

import com.hazelcast.simulator.agent.workerjvm.WorkerJvmSettings;
import com.hazelcast.simulator.common.AgentsFile;
import com.hazelcast.simulator.common.GitInfo;
import com.hazelcast.simulator.common.SimulatorProperties;
import com.hazelcast.simulator.coordinator.remoting.AgentsClient;
import com.hazelcast.simulator.protocol.connector.CoordinatorConnector;
import com.hazelcast.simulator.protocol.registry.AgentData;
import com.hazelcast.simulator.protocol.registry.ComponentRegistry;
import com.hazelcast.simulator.provisioner.Bash;
import com.hazelcast.simulator.test.TestCase;
import com.hazelcast.simulator.test.TestPhase;
import com.hazelcast.simulator.test.TestSuite;
import com.hazelcast.simulator.utils.CommandLineExitException;
import com.hazelcast.simulator.utils.ThreadSpawner;
import org.apache.log4j.Logger;

import java.io.File;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static com.hazelcast.simulator.coordinator.CoordinatorHelper.assignDedicatedMemberMachines;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.createAddressConfig;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.findNextAgentLayout;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.getMaxTestCaseIdLength;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.initAgentMemberLayouts;
import static com.hazelcast.simulator.protocol.configuration.Ports.AGENT_PORT;
import static com.hazelcast.simulator.utils.CloudProviderUtils.isEC2;
import static com.hazelcast.simulator.utils.CommonUtils.exitWithError;
import static com.hazelcast.simulator.utils.CommonUtils.getSimulatorVersion;
import static com.hazelcast.simulator.utils.CommonUtils.secondsToHuman;
import static com.hazelcast.simulator.utils.CommonUtils.sleepSeconds;
import static com.hazelcast.simulator.utils.ExecutorFactory.createFixedThreadPool;
import static com.hazelcast.simulator.utils.FileUtils.ensureExistingFile;
import static com.hazelcast.simulator.utils.FileUtils.getFilesFromClassPath;
import static com.hazelcast.simulator.utils.FileUtils.getSimulatorHome;
import static com.hazelcast.simulator.utils.HarakiriMonitorUtils.getStartHarakiriMonitorCommandOrNull;
import static java.lang.String.format;

@SuppressWarnings("checkstyle:classdataabstractioncoupling")
public final class Coordinator {

    static final File SIMULATOR_HOME = getSimulatorHome();
    static final File WORKING_DIRECTORY = new File(System.getProperty("user.dir"));
    static final File UPLOAD_DIRECTORY = new File(WORKING_DIRECTORY, "upload");

    private static final int COOLDOWN_SECONDS = 10;
    private static final int TEST_CASE_RUNNER_SLEEP_PERIOD = 30;
    private static final int EXECUTOR_TERMINATION_TIMEOUT_SECONDS = 10;

    private static final Logger LOGGER = Logger.getLogger(Coordinator.class);

    // options
    boolean monitorPerformance;
    boolean verifyEnabled = true;
    String workerClassPath;
    Integer testStopTimeoutMs;
    File agentsFile;
    TestSuite testSuite;
    int dedicatedMemberMachineCount;
    boolean parallel;
    TestPhase lastTestPhaseToSync;
    WorkerJvmSettings workerJvmSettings;
    int cooldownSeconds = COOLDOWN_SECONDS;
    int testCaseRunnerSleepPeriod = TEST_CASE_RUNNER_SLEEP_PERIOD;

    // internal state
    final SimulatorProperties props = new SimulatorProperties();

    AgentsClient agentsClient;
    CoordinatorConnector coordinatorConnector;
    FailureMonitor failureMonitor;
    PerformanceMonitor performanceMonitor;

    private final ComponentRegistry componentRegistry = new ComponentRegistry();

    private Bash bash;
    private ExecutorService parallelExecutor;

    private void run() throws Exception {
        try {
            bash = new Bash(props);

            initAgents();

            startWorkers();

            failureMonitor = new FailureMonitor(agentsClient, testSuite.id);
            failureMonitor.start();

            performanceMonitor = new PerformanceMonitor(agentsClient);

            runTestSuite();

            logFailureInfo();
        } finally {
            if (performanceMonitor != null) {
                performanceMonitor.stop();
            }

            if (failureMonitor != null) {
                failureMonitor.stop();
            }

            if (parallelExecutor != null) {
                parallelExecutor.shutdown();
                parallelExecutor.awaitTermination(EXECUTOR_TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS);
            }

            if (agentsClient != null) {
                agentsClient.terminateWorkers();
                agentsClient.stop();
            }

            if (coordinatorConnector != null) {
                coordinatorConnector.shutdown();
            }

            killAgents();
        }
    }

    private void initAgents() {
        populateComponentRegister();

        startAgents();
        try {
            startCoordinatorConnector();
        } catch (Exception e) {
            throw new CommandLineExitException("Could not start CoordinatorConnector", e);
        }

        agentsClient = new AgentsClient(componentRegistry.getAgents());
        agentsClient.start();

        initMemberWorkerCount(workerJvmSettings);
        initHzConfig(workerJvmSettings);
        initClientHzConfig(workerJvmSettings);

        int agentCount = agentsClient.getAgentCount();
        LOGGER.info(format("Performance monitor enabled: %s", monitorPerformance));
        LOGGER.info(format("Total number of agents: %s", agentCount));
        LOGGER.info(format("Total number of Hazelcast member workers: %s", workerJvmSettings.memberWorkerCount));
        LOGGER.info(format("Total number of Hazelcast client workers: %s", workerJvmSettings.clientWorkerCount));

        try {
            agentsClient.initTestSuite(testSuite);<commit_msg>Changed order of connection to Agents in Coordinator.<commit_after>/*
 * Copyright (c) 2008-2015, Hazelcast, Inc. All Rights Reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.hazelcast.simulator.coordinator;

import com.hazelcast.simulator.agent.workerjvm.WorkerJvmSettings;
import com.hazelcast.simulator.common.AgentsFile;
import com.hazelcast.simulator.common.GitInfo;
import com.hazelcast.simulator.common.SimulatorProperties;
import com.hazelcast.simulator.coordinator.remoting.AgentsClient;
import com.hazelcast.simulator.protocol.connector.CoordinatorConnector;
import com.hazelcast.simulator.protocol.registry.AgentData;
import com.hazelcast.simulator.protocol.registry.ComponentRegistry;
import com.hazelcast.simulator.provisioner.Bash;
import com.hazelcast.simulator.test.TestCase;
import com.hazelcast.simulator.test.TestPhase;
import com.hazelcast.simulator.test.TestSuite;
import com.hazelcast.simulator.utils.CommandLineExitException;
import com.hazelcast.simulator.utils.ThreadSpawner;
import org.apache.log4j.Logger;

import java.io.File;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import static com.hazelcast.simulator.coordinator.CoordinatorHelper.assignDedicatedMemberMachines;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.createAddressConfig;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.findNextAgentLayout;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.getMaxTestCaseIdLength;
import static com.hazelcast.simulator.coordinator.CoordinatorHelper.initAgentMemberLayouts;
import static com.hazelcast.simulator.protocol.configuration.Ports.AGENT_PORT;
import static com.hazelcast.simulator.utils.CloudProviderUtils.isEC2;
import static com.hazelcast.simulator.utils.CommonUtils.exitWithError;
import static com.hazelcast.simulator.utils.CommonUtils.getSimulatorVersion;
import static com.hazelcast.simulator.utils.CommonUtils.secondsToHuman;
import static com.hazelcast.simulator.utils.CommonUtils.sleepSeconds;
import static com.hazelcast.simulator.utils.ExecutorFactory.createFixedThreadPool;
import static com.hazelcast.simulator.utils.FileUtils.ensureExistingFile;
import static com.hazelcast.simulator.utils.FileUtils.getFilesFromClassPath;
import static com.hazelcast.simulator.utils.FileUtils.getSimulatorHome;
import static com.hazelcast.simulator.utils.HarakiriMonitorUtils.getStartHarakiriMonitorCommandOrNull;
import static java.lang.String.format;

@SuppressWarnings("checkstyle:classdataabstractioncoupling")
public final class Coordinator {

    static final File SIMULATOR_HOME = getSimulatorHome();
    static final File WORKING_DIRECTORY = new File(System.getProperty("user.dir"));
    static final File UPLOAD_DIRECTORY = new File(WORKING_DIRECTORY, "upload");

    private static final int COOLDOWN_SECONDS = 10;
    private static final int TEST_CASE_RUNNER_SLEEP_PERIOD = 30;
    private static final int EXECUTOR_TERMINATION_TIMEOUT_SECONDS = 10;

    private static final Logger LOGGER = Logger.getLogger(Coordinator.class);

    // options
    boolean monitorPerformance;
    boolean verifyEnabled = true;
    String workerClassPath;
    Integer testStopTimeoutMs;
    File agentsFile;
    TestSuite testSuite;
    int dedicatedMemberMachineCount;
    boolean parallel;
    TestPhase lastTestPhaseToSync;
    WorkerJvmSettings workerJvmSettings;
    int cooldownSeconds = COOLDOWN_SECONDS;
    int testCaseRunnerSleepPeriod = TEST_CASE_RUNNER_SLEEP_PERIOD;

    // internal state
    final SimulatorProperties props = new SimulatorProperties();

    AgentsClient agentsClient;
    CoordinatorConnector coordinatorConnector;
    FailureMonitor failureMonitor;
    PerformanceMonitor performanceMonitor;

    private final ComponentRegistry componentRegistry = new ComponentRegistry();

    private Bash bash;
    private ExecutorService parallelExecutor;

    private void run() throws Exception {
        try {
            bash = new Bash(props);

            initAgents();

            startWorkers();

            failureMonitor = new FailureMonitor(agentsClient, testSuite.id);
            failureMonitor.start();

            performanceMonitor = new PerformanceMonitor(agentsClient);

            runTestSuite();

            logFailureInfo();
        } finally {
            if (performanceMonitor != null) {
                performanceMonitor.stop();
            }

            if (failureMonitor != null) {
                failureMonitor.stop();
            }

            if (parallelExecutor != null) {
                parallelExecutor.shutdown();
                parallelExecutor.awaitTermination(EXECUTOR_TERMINATION_TIMEOUT_SECONDS, TimeUnit.SECONDS);
            }

            if (agentsClient != null) {
                agentsClient.terminateWorkers();
                agentsClient.stop();
            }

            if (coordinatorConnector != null) {
                coordinatorConnector.shutdown();
            }

            killAgents();
        }
    }

    private void initAgents() {
        populateComponentRegister();

        startAgents();

        agentsClient = new AgentsClient(componentRegistry.getAgents());
        agentsClient.start();

        try {
            startCoordinatorConnector();
        } catch (Exception e) {
            throw new CommandLineExitException("Could not start CoordinatorConnector", e);
        }

        initMemberWorkerCount(workerJvmSettings);
        initHzConfig(workerJvmSettings);
        initClientHzConfig(workerJvmSettings);

        int agentCount = agentsClient.getAgentCount();
        LOGGER.info(format("Performance monitor enabled: %s", monitorPerformance));
        LOGGER.info(format("Total number of agents: %s", agentCount));
        LOGGER.info(format("Total number of Hazelcast member workers: %s", workerJvmSettings.memberWorkerCount));
        LOGGER.info(format("Total number of Hazelcast client workers: %s", workerJvmSettings.clientWorkerCount));

        try {
            agentsClient.initTestSuite(testSuite);
{'fraction_non_alphanumeric': 0.08008483873326995, 'fraction_numerical': 0.002925473561032692, 'mean_word_length': 3.943600867678959, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4743080', 'n_tokens_mistral': 3781, 'n_tokens_neox': 3682, 'n_words': 794}