/*
 * mBench: The Open Source Micro Benchmark Tool
 *
 * Distributable under GPL license.
 * See terms of license at gnu.org.
 *
 * Copyright (C) 2005 Sumisho Computer Systems Corp.
 */
package jp.co.scs.mbench;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.net.UnknownHostException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Calendar;
import java.util.Date;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import javax.xml.parsers.ParserConfigurationException;

import org.xml.sax.SAXException;

/**
 *      The Benchmark Manager Class.
 *
 *      @author Tetsuro Ikeda
 *      @author Masato Koga
 */
public class BenchmarkManager extends Thread {
    /**
     *  The ID for this manager instance (default 1).
     *  This is for multiple node benchmarking.
     */
    private Integer managerID = new Integer(1);

    /**
     *  Current state of benchmark
     */
    protected String currentState = null;

    /**
     *  Node type.
     *  "single" for single node benchmark.
     *  "primary" for multile node benchmark and this manager is primary node.
     *  "secondary" for multiple node benchmark and this manager
     *  isn't primary node.
     */
    private String nodeType = MultiNodeInfo.NODE_SINGLE;

    /**
     *  The benchmark name.
     */
    private String benchmarkName = null;

    /**
     *  Options for benchmark.
     */
    private Map optionMap = null;

    /**
     *  Benchmark Infomation.
     */
    private BenchmarkInfo info = null;

    /**
     *  Benchmark Containers (each container is thread).
     */
    private BenchmarkContainer[] containers = null;

    /**
     *  Containers which are managed by this manager.
     *  The container is removed from this list when the container's
     *  execution is finished.
     */
    private List managedContainers = null;

    /**
     *  Containers which couldn't finished normally.
     *  The cause may be timeout, throwing exception.
     */
    private List terminatedContainers = new ArrayList();

    /**
     *  The instance to synchronize.
     */
    private Synchronizer synchronizer = null;

    /**
     *  The socket used when this manager is secondary node.
     */
    private Socket socket = null;

    /**
     *  The sockets used when this manager is primary node.
     */
    private Socket[] socketArray = null;

    /**
     *  The interval (milliseconds) for checking containers.
     */
    private final long watchInterval = 100;

    /**
     *  The timeout (milliseconds) for benchmark timeout.
     */
    private long benchmarkTimeout = 300 * 1000;

    /**
     *  The timeout (milliseconds) for timeout
     *  in communicating with other nodes.
     *  This is used when this manager
     *  is not "single" node. (primary or secondary)
     */
    private int socketTimeout = 60 * 100;

    /**
     *  System logging handler
     */
    private LogWriter logWriter = null;

    /**
     *  Data logging handler
     */
    private DataWriter dataWriter = null;

    /**
     *  Number of executed all transaction
     */
    protected static int allIterationCount = 0;

    /**
     *  Creates a benchmark manager instance with benchmark name.
     * 
     *  @param benchmarkName benchmark name
     */
    public BenchmarkManager(String benchmarkName) {
        this.benchmarkName = benchmarkName;
    }
    
    /**
     *  Gets the ID of this manager.
     * 
     *  @return manager ID
     */
    public Integer getManagerID() {
        return this.managerID;
    }
    
    /**
     *  Sets the ID of this manager.
     * 
     *  @param managerID manager ID
     */
    public void setManagerID(Integer managerID) {
        this.managerID = managerID;
    }
    
    /**
     *  Gets the benchmark name.
     * 
     *  @return benchmark name
     */
    public String getBenchmarkName() {
        return this.benchmarkName;
    }
    
    /**
     *  Sets the benchmark name
     * 
     *  @param benchmarkName benchmark name
     */
    public void setBenchmarkName(String benchmarkName) {
        this.benchmarkName = benchmarkName;
    }
    
    /**
     *  Gets the options
     * 
     *  @return options
     */
    public Map getOptionMap() {
        return this.optionMap;
    }
    
    /**
     *  Sets the options.<br>
     *  <br>
     *  This options is defined by command line argments and not by configulation file.
     *  The porpse of these options are used for override the setting on configulation file
     *  or default value. (e.g. the number of threads/containers).
     * 
     *  @param optionMap options
     */
    public void setOptionMap(Map optionMap) {
        this.optionMap = optionMap;
    }
    
    /**
     *  Gets the benchmark information.
     * 
     *  @return benchmark information
     */
    public BenchmarkInfo getBenchmarkInfo() {
        return this.info;
    }
    
    /**
     *  Sets the benchmark information.
     * 
     *  @param info benchmark information
     */
    public void setBenchmarkInfo(BenchmarkInfo info) {
        this.info = info;
    }
    
    /**
     *  Gets the current state
     * 
     *  @return current state
     */
    public String getCurrentState() {
        return this.currentState;
    }
    
    /**
     *  Sets the current state
     * 
     *  @param currentState current state
     */
    public void setCurrentState(String currentState) {
        this.currentState = currentState;
    }
    
    /**
     *  Sets the timeout (milliseconds) for communicating in multiple node benchmark.
     * 
     *  @param socketTimeout timeout (milliseconds)
     */
    public void setSocketTimeout(int socketTimeout) {
        this.socketTimeout = socketTimeout;
    }
    
    /**
     *  Gets the timeout (milliseconds) for communicating in multiple node benchmark.
     * 
     *  @return timeout (milliseconds)
     */
    public int getSocketTimeout() {
        return this.socketTimeout;
    }

    /**
     *  Sets the system log handler.
     *  Note: This method should be called before the benchmark manager
     *  is started. If any system log handler isn't set, the default handler is used.
     * 
     *  @param logWriter system log handler
     */
    public void setLogWriter(LogWriter logWriter) {
        this.logWriter = logWriter;
    }

    /**
     *  Gets the system log handler.
     * 
     *  @return system log handler
     */
    public LogWriter getLogWriter() {
        return this.logWriter;
    }
    
    /**
     *  Sets the data log handler.
     *  Note: This method should be called before the benchmark manager
     *  is started. If any data log handler isn't set, the default handler is used.
     * 
     *  @param dataWriter data log handler
     */
    public void setDataWriter(DataWriter dataWriter) {
        this.dataWriter = dataWriter;
    }
    
    /**
     *  Gets the data log handler.
     * 
     *  @return data log handler
     */
    public DataWriter getDataWriter() {
        return this.dataWriter;
    }
    
    /**
     *  Sets the synchronizer
     * 
     *  @param synchronizer synchronizer
     */
    public void setSynchronizer(Synchronizer synchronizer) {
        this.synchronizer = synchronizer;
    }
    
    /**
     *  Gets the synchronizer
     * 
     *  @return synchronizer
     */
    public Synchronizer getSynchronizer() {
        return this.synchronizer;
    }
    
    /**
     *  Sets the benchmark containers.
     *  Note: This method should be called before manager is started,
     *  or the default containers are created and used.
     * 
     *  @param containers benchmark containers
     */
    public void setBenchmarkContainers(BenchmarkContainer[] containers) {
        this.containers = containers;
    }
    
    /**
     *  Gets the benchmark containers.
     * 
     *  @return benchmark containers
     */
    public BenchmarkContainer[] getBenchmarkContainers() {
        return this.containers;
    }
    
    /**
     *  Adds the containers to list which are not finished normally
     *  for timeout or throwing exception.
     * 
     *  @param container container
     */
    public void addTerminatedContainer(BenchmarkContainer container) {
        this.terminatedContainers.add(container);
    }
    
    /**
     *  Gets the containers list which are not finished normally
     *  for timeout or throwing exception.
     * 
     *  @return container
     */
    public List getTerminatedContainers() {
        return this.terminatedContainers;
    }
    
    /**
     *  Sets the benchmark timeout (milliseconds).
     *  The default is 100sec (100,000 miliseconds)
     * 
     *  @param benchmarkTimeout timeout (milliseconds)
     */
    public void setBenchmarkTimeout(long benchmarkTimeout) {
        this.benchmarkTimeout = benchmarkTimeout;
    }
    
    /**
     *  Gets the benchmark timeout (milliseconds)
     * 
     *  @return timeout (milliseconds)
     */
    public long getBenchmarkTimeout() {
        return this.benchmarkTimeout;
    }
    
    /**
     *  Run the benchmark.<br>
     *  <br>
     *  Note: The benchmark name must be set before this method is called.
     */
    public void run() {
        this.currentState = BenchmarkState.CREATED;
        
        Date currentDate = Calendar.getInstance().getTime();
        String pattern = "yyyyMMdd_HHmmss";
        SimpleDateFormat dateFormat = new SimpleDateFormat(pattern);
        String suffix =  dateFormat.format(currentDate);
        try {
            if (this.logWriter == null) {
                String logName = "log_" + suffix + ".log";
                this.logWriter = new LogWriter(logName);
            }
            this.logWriter.init();
            
            if (this.dataWriter == null) {
                String dataName = "data_" + suffix + ".csv";
                this.dataWriter = new CSVAsynchronousDataWriter(dataName);
            }
            
            this.logWriter.write(LogWriter.INFO, MessageResources.getMessage(
                    "BenchmarkManager.start", this.managerID));
        } catch (BenchmarkTerminateException ex) {
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            throw new BenchmarkAbortException(ex);
        }
        
        this.currentState = BenchmarkState.INITIALIZING;
        
        try {
            // initialization
            doInit();
        // if exception is thrown in this manager's containers
        } catch (BenchmarkTerminateException ex) {
            // notification to other nodes.
            // this notification is received at the next check point.
            if (this.nodeType.equals(MultiNodeInfo.NODE_PRIMARY)) {
                // notify to other secondary nodes
                notifyInitFailureAsPrimary();
            } else if (this.nodeType.equals(MultiNodeInfo.NODE_SECONDARY)) {
                // notify to the primary node.
                notifyInitFailureAsSecondary();
            }
            // call finialization and end
            this.currentState = BenchmarkState.CLEANING;
            doClean();
            this.logWriter.write(LogWriter.INFO, MessageResources.getMessage(
                    "BenchmarkManager.ng", this.managerID));
            this.logWriter.clean();
            this.dataWriter.clean();
            this.currentState = BenchmarkState.CLEANED;
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            return;
        }

        // initialization of this node is succeeded.
        this.currentState = BenchmarkState.INITIALIZED;

        try {
            // check if other nodes got failed of initialization.
            if (this.nodeType.equals(MultiNodeInfo.NODE_PRIMARY)) {
                // check from secondary nodes and send "ok" if not any errors. 
                checkInitAsPrimary();
            } else if (this.nodeType.equals(MultiNodeInfo.NODE_SECONDARY)) {
                // check from primary node and send "ok" if not any errors.
                checkInitAsSecondary();
            }
        // if other node failed
        } catch (BenchmarkTerminateException ex) {
            // call finalization and end
            this.currentState = BenchmarkState.CLEANING;                    
            doClean();
            this.logWriter.write(LogWriter.INFO, MessageResources.getMessage(
                    "BenchmarkManager.ng", this.managerID));
            this.logWriter.clean();
            this.dataWriter.clean();
            this.currentState = BenchmarkState.CLEANED;
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            return;
        }
            
        this.currentState = BenchmarkState.RUNNING;
        
        try {
            // synchronize if this benchmark is running as multiple node
            if (this.nodeType.equals(MultiNodeInfo.NODE_PRIMARY)) {
                synchronizeMultiNodeAsPrimary();
            } else if (this.nodeType.equals(MultiNodeInfo.NODE_SECONDARY)){
                synchronizeMultiNodeAsSecondary();   
            }

            // start the benchmark
            doRun();
        // go in this block if any managed container become "BENCHMARK_TERMINATED",
        // or other node (manager) got same status.
        } catch (BenchmarkTerminateException ex) { 
            // kill all threads(containers) still running
            killManagedContainers();
            // call finialization and end
            this.currentState = BenchmarkState.CLEANING;                    
            doClean();
            this.logWriter.write(LogWriter.INFO, MessageResources.getMessage(
                    "BenchmarkManager.ng", this.managerID));
            this.logWriter.clean();
            this.dataWriter.clean();
            this.currentState = BenchmarkState.CLEANED;
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            return;
        }

        // executing of benchmark is succeeded in this manager's containers
        this.currentState = BenchmarkState.FINISHED;

        // cleaning
        this.currentState = BenchmarkState.CLEANING;
        doClean();

        this.logWriter.write(LogWriter.INFO, MessageResources.getMessage(
                "BenchmarkManager.ok", this.managerID));
        
        this.logWriter.clean();
        this.dataWriter.clean();
        
        try {
            if (this.socketArray != null) {
                for (int i = 0; i < this.socketArray.length; i++) {
                    this.socketArray[i].close();
                }
            }
            if (this.socket != null) {
                this.socket.close();
            }
        } catch (IOException ex) {
            // ignore
        }
        this.currentState = BenchmarkState.CREATED;
        this.currentState = BenchmarkState.BENCHMARK_SUCCESS;
    }
    
    /**
     *  Initializes the benchmark.
     *  If exception is thrown while initializing, finalization is called for
     *  the containers that have already finished the initialization.
     * 
     *  @throws BenchmarkTerminateException Benchmark should be terminated
     */
    protected void doInit() throws BenchmarkTerminateException {
        try {
            // get the information from configuration file.
            if (this.info == null) {
                this.info = DocumentUtil.getBenchmarkInfo(this.benchmarkName);
            }
            // multiple node configuration, initialization
            if (this.optionMap != null && this.optionMap.get("multi-client") != null) {
                checkMultiNodeConnect((String) this.optionMap.get("multi-client"));
            }
            // override the benchmark information.
            overRideInfo(this.optionMap);
            
            Object[] items = {
                new Integer(this.info.getThreadNumber()),
                new Integer(this.info.getRepeatNumber()),
                new Integer(this.info.getTransactionNumber()),
                new Long(this.info.getBenchmarkEndTime()),
                this.info.getClassName(),
                this.nodeType
            };
            this.logWriter.write(LogWriter.INFO, MessageResources.getMessage(
                    "BenchmarkManager.configuration", items));
            
            this.dataWriter.init();
            try {
                Thread.sleep(100);
            } catch (InterruptedException ex) {
                // ignore
            }

            // print the label of benchmark data.
            Class benchmarkClass = Class.forName(info.getClassName());
            Object benchmarkObject = 
                benchmarkClass.newInstance();
            String[] userLabels =
                ((Benchmark) benchmarkObject).getOptionLabels();
            if (userLabels != null) {
                String[] resultLabels = new String[
                    BenchmarkContainer.SYSTEM_LABELS.length + userLabels.length];
                for (int i = 0; i < BenchmarkContainer.SYSTEM_LABELS.length; i++) {
                    resultLabels[i] = BenchmarkContainer.SYSTEM_LABELS[i];
                }
                for (int i = 0; i < userLabels.length; i++) {
                    resultLabels[BenchmarkContainer.SYSTEM_LABELS.length + i] =
                        userLabels[i];
                }
                this.dataWriter.write(resultLabels);
            } else {
                this.dataWriter.write(BenchmarkContainer.SYSTEM_LABELS);
            }

            // create the synchronizer instance
            if (this.synchronizer == null) {
                this.synchronizer = new Synchronizer();
            }

            // create the containers
            if (this.containers == null) {
                this.containers =
                    new BenchmarkContainer[this.info.getThreadNumber()];
                
                for (int i = 0; i < this.containers.length; i++) {
                    int containerID = i + 1;
                    this.containers[i] = new BenchmarkContainer(
                            containerID, this.info, this.synchronizer,
                            this.logWriter, this.dataWriter);
                    
                }
            }

            // initialize the containers
            for (int i = 0; i < this.containers.length; i++) {
                this.containers[i].initTarget();
            }
            
            // create the synchronizer thread and lock it
            this.synchronizer.start();

            // all containers starts, but wait because of synchronizer lock
            for (int i = 0; i < this.containers.length; i++) {
                this.containers[i].start();
            }
        // terminate whole benchmark if any exception is thrown in this sequence.
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        } catch (ParserConfigurationException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        } catch (SAXException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        } catch (IllegalArgumentException ex) {
            logWriter.write(LogWriter.ERROR,
                    "[BenchmarkManager] " + ex.getMessage());
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        } catch (ClassNotFoundException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        } catch (InstantiationException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        } catch (IllegalAccessException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        } catch (BenchmarkTerminateException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw ex;
        }
    }
    
    /**
     *  Run the benchmark
     * 
     *  @throws BenchmarkTerminateException benchmark should be terminated
     */
    protected void doRun() throws BenchmarkTerminateException {
        // get the grobal start time
        this.info.setBenchmarkStartTime(System.currentTimeMillis());

        // release the synchronizer lock, all containers start actually
        this.synchronizer.interrupt();

        // make a managed container's list
        this.managedContainers = new LinkedList();
        for (int i = 0; i < this.containers.length; i++) {
            this.managedContainers.add(this.containers[i]);
        }

        // check repeat until number of managed containers become 0.
        while (this.managedContainers.size() > 0) {
            for (int i = 0; i < this.managedContainers.size(); i++) {
                BenchmarkContainer container =
                    (BenchmarkContainer) this.managedContainers.get(i);
                // check this container status
                if (container.getCurrentState().equals(BenchmarkState.RUNNING)) {
                    if (this.benchmarkTimeout > 0) {
                        // check if this container should be determined as timeout
                        long currentTime = System.currentTimeMillis();
                        long progressTime = currentTime - container.getStartTime();
                        if (progressTime > this.benchmarkTimeout) {
                            // check if this container should be determined as timeout
                            container.stop();
                            addTerminatedContainer(container);
                            this.managedContainers.remove(container);
                            // collect the information from this container
                            logWriter.write(LogWriter.ERROR, MessageResources.getMessage(
                                    "BenchmarkManager.timeout",
                                    new Integer(container.getContainerID()),
                                    new Long(this.benchmarkTimeout)));
                        }
                    }
                } else if (container.getCurrentState().equals(BenchmarkState.FINISHED)) {
                    // remove this container from managed container list
                    // because this container finished normally.
                    this.managedContainers.remove(container);
                } else if (container.getCurrentState().equals(BenchmarkState.THREAD_TERMINATED)) {
                    // remove this container from managed container list
                    // because this container stopped by itself
                    this.managedContainers.remove(container);
                } else if (container.getCurrentState().equals(BenchmarkState.BENCHMARK_TERMINATED)) {
                    this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
                    // bad case. this container status became BENCHMARL_TERMINATED,
                    // so must notify to other nodes.
                    if (this.nodeType.equals(MultiNodeInfo.NODE_PRIMARY)) {
                        notifyRunFailureAsPrimary();
                    } else if (this.nodeType.equals(MultiNodeInfo.NODE_SECONDARY)){
                        notifyRunFailureAsSecondary();   
                    }
                    throw new BenchmarkTerminateException();
                } else {
                    // ignore
                }
            }
            // check if the other nodes notify the benchmark termination
            if (this.nodeType.equals(MultiNodeInfo.NODE_PRIMARY)) {
                checkRunFailureAsPrimary();
            } else if (this.nodeType.equals(MultiNodeInfo.NODE_SECONDARY)){
                checkRunFailureAsSecondary();   
            }
            // Good.
            // all managed container is going well.
            // get a bit sleep and will check again.
            if (this.managedContainers.size() > 0){
                try {
                    Thread.sleep(this.watchInterval);
                } catch (InterruptedException ex) {
                    // ignore
                }
            }
        }
    }
    
    /**
     *  clean all components
     */
    protected void doClean() {
        if (this.containers != null) {
            // for each container
            for (int i = 0; i < this.containers.length; i++) {
                // call finalization if initialization was done
                if (this.containers[i].isInitialized()) {
                    this.containers[i].cleanTarget();
                }
            }
        }
    }
    
    /**
     *  initializes for multiple nodes
     *  
     *  @param name definition name for multiple node in configuration file.
     */
    protected void checkMultiNodeConnect(String name) {
        try {
            MultiNodeInfo info = DocumentUtil.getMultiNodeInfo(name);
            this.nodeType = info.getNodeType();
            // polymorphism
            if (this.nodeType.equalsIgnoreCase(MultiNodeInfo.NODE_PRIMARY)) {
                checkConnectAsPrimary(info);  
            } else if (this.nodeType.equalsIgnoreCase(MultiNodeInfo.NODE_SECONDARY)) {
                checkConnectAsSecondary(info);
            }             
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException(ex);
        } catch (ParserConfigurationException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException(ex);
        } catch (SAXException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException(ex);
        }
    }

    /**
     *  check if all communication with secondary nodes are good,
     *  then order to do the initialization.
     *
     *  @param info multinode information
     */
    protected void checkConnectAsPrimary(MultiNodeInfo info) {
        try {
            int[] targetPort = info.getSecondaryPort();
            String[] targetHost = info.getSecondaryHost();
            info.getLocalPort();
            InetAddress.getByName(info.getLocalHost());
            
            this.socketArray = new Socket[targetHost.length];
            int maxRetry = 10;
            int successSocketCount = 0;
            for (int i = 0; i < maxRetry; i++) {
                for (int j = 0; j < info.getSecondaryHost().length; j++) {
                    if (this.socketArray[j] == null) {
                        try {
                            this.socketArray[j] = new Socket(
                                targetHost[j], targetPort[j]/*, localAddress, localPort*/);
                            this.socketArray[j].setSoTimeout(this.socketTimeout);
                            successSocketCount++;
                        } catch (IOException ex) {
                            // ignore
                        }
                    }
                }
                if (successSocketCount == targetHost.length) {
                    break;
                }
                try {
                    Thread.sleep(this.socketTimeout / maxRetry);
                } catch (InterruptedException ex) {
                    // ignore
                }
            }
            try {
                // check if all sockets to secondary nodes are fine.
                if (successSocketCount == targetHost.length) {
                    int managerIDOffset = this.managerID.intValue();
                    for (int i = 0; i < successSocketCount; i++) {
                        OutputStream os = this.socketArray[i].getOutputStream();
                        managerIDOffset += this.info.getThreadNumber();
                        String message = "doInit" + managerIDOffset;
                        os.write(new String(message).getBytes());
                    }
                // if any problem, notify the termination of benchmark to communicatable nodes.
                } else {
                    for (int i = 0; i < targetHost.length; i++) {
                        if (this.socketArray[i] != null) {
                            OutputStream os = this.socketArray[i].getOutputStream();
                            os.write(new String("doAbort").getBytes());
                        }
                    }
                    this.currentState = BenchmarkState.BENCHMARK_FAIL;
                    throw new BenchmarkAbortException();
                }
            } catch (IOException ex) {
                // if any problem and also can't communicate with other nodes.
                this.currentState = BenchmarkState.BENCHMARK_FAIL;
                throw new BenchmarkAbortException(ex);
            }
        } catch (UnknownHostException ex) {
            // if IP address can't be look up by host name.
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            throw new BenchmarkAbortException(ex);
        }
    }
    
    /**
     *  accept the message from primary node check if initialization should be started.
     * 
     *  @param info multinode information
     */
    protected void checkConnectAsSecondary(MultiNodeInfo info) {
        try {
            int localPort = info.getLocalPort();
            int backlog = 1;
            InetAddress localAddress = InetAddress.getByName(info.getLocalHost());
            ServerSocket server = new ServerSocket(localPort, backlog, localAddress);

            // accept the message from primary node
            server.setSoTimeout(this.socketTimeout);
            this.socket = server.accept();
            this.socket.setSoTimeout(this.socketTimeout);
            byte[] buffer = new byte[20];
            InputStream is = this.socket.getInputStream();
            is.read(buffer);
 
            String message = new String(trimBlankByte(buffer));

            // continue if the message is "doInit"
            if (message.startsWith("doInit")) {
                String managerIDString = message.substring(6, message.length());
                int newManagerID = Integer.parseInt(managerIDString);
                this.managerID = new Integer(newManagerID);
                return;
                // abort if the message isn't "doInit".
            } else {
                String errMsg = "aborting";
                this.currentState = BenchmarkState.BENCHMARK_FAIL;
                throw new BenchmarkAbortException(errMsg);   
            }
        } catch (UnknownHostException ex) {
            ex.printStackTrace();
            // can't lookup my IP address from hostname.
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            throw new BenchmarkAbortException(ex);
        } catch (SocketTimeoutException ex) {
            // timeout of communicate with primary node
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            throw new BenchmarkAbortException(ex);
        } catch (IOException ex) {
            // abort 
            this.currentState = BenchmarkState.BENCHMARK_FAIL;
            throw new BenchmarkAbortException(ex);
        }
    }
    
    /**
     *  check if all initialization of secondary nodes were succeeded,
     *  and then notify the order of preparing to start benchmark. 
     * 
     *  @throws BenchmarkTerminateException benchmark should be terminated
     */
    protected void checkInitAsPrimary() throws BenchmarkTerminateException {
        try {
            // chek if all secondary nodes succeeded their initialization.
            boolean allSecondariesOK = true;
            for (int i = 0; i < this.socketArray.length; i++) {
                InputStream is = this.socketArray[i].getInputStream();
                byte[] buffer = new byte[20];
                // may wait for 60 seconds (max).
                is.read(buffer);
                buffer = trimBlankByte(buffer);
                String message = new String(trimBlankByte(buffer));
                // the message should be "doInit OK" if the initialization was succeeded.
                // if not, the message should be "ERROR:init"
                if (!message.equals("doInit OK")) {
                    allSecondariesOK = false;
                    break;
                }
            }
            // if all nodes succeeded.
            if (allSecondariesOK) {
                for (int i = 0; i < this.socketArray.length; i++) {
                    OutputStream os = this.socketArray[i].getOutputStream();
                    os.write(new String("doRun").getBytes());
                }
                return;
            // if any node failed
            } else {
                for (int i = 0; i < this.socketArray.length; i++) {
                    OutputStream os = this.socketArray[i].getOutputStream();
                    os.write(new String("doTerminate").getBytes());
                }
                this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
                throw new BenchmarkTerminateException("terminating");
            }
        } catch(IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException("terminating");
        }
    }
    
    /**
     *  notify the result (initialization was succeeded) to the primary node,
     *  and wait the order of preparing to start benchmark.
     * 
     *  @throws BenchmarkTerminateException benchmark should be terminated
     */
    protected void checkInitAsSecondary() throws BenchmarkTerminateException {
        try {
            OutputStream os = this.socket.getOutputStream();
            os.write(new String("doInit OK").getBytes());
            // may wait for 60sec max.
            InputStream is = this.socket.getInputStream();
            byte[] buffer = new byte[20];
            is.read(buffer);
            String message = new String(trimBlankByte(buffer));
            // continue if the message is "doRun"
            if (message.equals("doRun")) {
                return;
            } else {
                throw new BenchmarkTerminateException(message);   
            }
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        }
    }
    
    /**
     *  notify the termination of benchmark because of initialization error
     */
    protected void notifyInitFailureAsSecondary() {
        try {
            OutputStream os = this.socket.getOutputStream();
            os.write(new String("ERROR:init").getBytes());
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException();
        }
    }
    
    /**
     *  notify the termination of benchmark because of initialization error
     */
    protected void notifyInitFailureAsPrimary() {
        try {
            for (int i = 0; i < this.socketArray.length; i++) {
                OutputStream os = this.socketArray[i].getOutputStream();
                os.write(new String("ERROR:init").getBytes());
            }
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException();
        }
    }
    
    /**
     * notify the termination of benchmark because of error while benchmark is running
     */
    protected void notifyRunFailureAsSecondary() {
        try {
            OutputStream os = this.socket.getOutputStream();
            os.write(new String("ERROR:run").getBytes());
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException();
        }
    }
    
    /**
     * notify the termination of benchmark because of error while benchmark is running
     */
    protected void notifyRunFailureAsPrimary() {
        try {
            for (int i = 0; i < this.socketArray.length; i++) {
                OutputStream os = this.socketArray[i].getOutputStream();
                os.write(new String("ERROR:run").getBytes());
            }
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException();
        }
    }
    
    /**
     *  synchronize as secondary node
     * 
     *  @throws BenchmarkTerminateException benchmark should be terminated
     */
    protected void synchronizeMultiNodeAsSecondary() throws BenchmarkTerminateException {
        try {
            InputStream is = this.socket.getInputStream();
            byte[] readBuffer = new byte[10];
            // トリガー
            is.read(readBuffer);
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkTerminateException(ex);
        }
    }
    
    /**
     *  synchronize as primary node
     */
    protected void synchronizeMultiNodeAsPrimary() {
        class Trigger extends Thread {
            private Synchronizer sync = null;
            private Socket socket = null;
            public Trigger(Synchronizer sync, Socket socket) {
                this.sync = sync;
                this.socket = socket;
            }
            public void run() {
                try {
                    OutputStream os = this.socket.getOutputStream();
                    byte[] writeBuffer = new String("start").getBytes();
                    // トリガー
                    this.sync.doSynchronize();
                    os.write(writeBuffer);
                } catch (IOException ex) {
                    throw new BenchmarkAbortException();
                }    
            }
        }
        Synchronizer lock = new Synchronizer();
        lock.start();
        for (int i = 0; i < this.socketArray.length; i++) {
            new Trigger(lock, this.socketArray[i]).start();
        }
        try {
            Thread.sleep(100);
        } catch (InterruptedException ex) {
            // ignore
        }
        lock.interrupt();
    }
    
    /**
     *  Checks if any error is notified by secondary nodes.
     */
    protected void checkRunFailureAsPrimary() throws BenchmarkTerminateException {
        try {
            for (int i = 0; i < this.socketArray.length; i++) {
                byte[] readBuffer = new byte[20];
                InputStream is = this.socketArray[i].getInputStream();
                this.socketArray[i].setSoTimeout(1);
                try {
                    is.read(readBuffer);
                } catch (SocketTimeoutException ex) {
                    // ignore
                }
                this.socketArray[i].setSoTimeout(this.socketTimeout);
                String message = new String(readBuffer);
                if (message.equals("ERROR:run")) {
                    throw new BenchmarkTerminateException();
                }   
            }
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException();
        }
    }
    
    /**
     *  Checkes if any error is notified by primary node.
     */
    protected void checkRunFailureAsSecondary() throws BenchmarkTerminateException {
        try {
            byte[] readBuffer = new byte[20];
            InputStream is = this.socket.getInputStream();
            this.socket.setSoTimeout(1);
            try {
                is.read(readBuffer);
                
            } catch (SocketTimeoutException ex) {
                // ignore
            }
            this.socket.setSoTimeout(this.socketTimeout);
            String message = new String(trimBlankByte(readBuffer));
            if (message.equals("ERROR:run")) {
                throw new BenchmarkTerminateException();
            }  
        } catch (IOException ex) {
            this.currentState = BenchmarkState.BENCHMARK_TERMINATED;
            throw new BenchmarkAbortException();
        }
    }
    
    /**
     *  Kills all managed containers.
     */
    private void killManagedContainers() {
        for (int j = 0; j < this.managedContainers.size(); j++) {
            BenchmarkContainer leftContainer =
                (BenchmarkContainer) this.managedContainers.get(j);
            leftContainer.stop();
            this.addTerminatedContainer(leftContainer);
            // get the information from container
            logWriter.write(LogWriter.INFO, MessageResources.getMessage(
                "BenchmarkManager.terminate", new Integer(leftContainer.getContainerID())));
        }
        // clear the list
        this.managedContainers.clear();
    }
    
    /**
     *  override the benchmark information by option map
     * 
     *  @param map option map
     */
    private void overRideInfo(Map optionMap) {
        if (optionMap == null || optionMap.size() == 0) {
            return;
        }
        // number of threads(containers)
        String threadNumber = (String) optionMap.get(
                BenchmarkMain.THREAD_NUMBER);
        if (threadNumber != null) {
            this.info.setThreadNumber(Integer.parseInt(threadNumber));
        }
        
        // number of repeats
        String repeatNumber = (String) optionMap.get(
                BenchmarkMain.REPEAT_NUMBER);
        if (repeatNumber != null) {
            this.info.setRepeatNumber(Integer.parseInt(repeatNumber));
        }
        
        // number of all transaction
        String txNumber = (String) optionMap.get(
                BenchmarkMain.TX_NUMBER);
        if (txNumber != null) {
            this.info.setTransactionNumber(Integer.parseInt(txNumber));
        }
        
        // time of benchmark end
        String benchmarkEndTime = (String) optionMap.get(
                BenchmarkMain.END_TIME);
        if(benchmarkEndTime != null) {
            this.info.setBenchmarkEndTime(Long.parseLong(benchmarkEndTime));
        }
    }
    
    /**
     *  clean the byte array to get the string
     */
    private byte[] trimBlankByte(byte[] buffer) {
        int count = 0;
        for (int i = 0; i < buffer.length; i++) {
            if (buffer[i] == 0) {
                break;
            } else {
                count++;
            }
        }
        byte[] newBuffer = new byte[count];
        for (int i = 0; i < count; i++) {
            newBuffer[i] = buffer[i];
        }
        return newBuffer;
    }
}
