parallel/spmd/SPMDClientController.java

package appl.parallel.spmd;

import java.rmi.RemoteException;
import java.util.HashMap;
import java.util.Vector;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import appl.ext.XuluConfig;
import appl.parallel.ComputingResourceContainer;
import appl.parallel.ComputingResourceProperties;
import appl.parallel.client.ClientDataServer;
import appl.parallel.client.RemoteEventHandler;
import appl.parallel.data.PartitionDataHandler;
import appl.parallel.data.PartitionHandlerFactory;
import appl.parallel.event.CommEvent.CommType;
import appl.parallel.event.CommEventSink;
import appl.parallel.model.AbstractParallelStepModel;
import appl.parallel.server.PartitionDataServer;
import appl.parallel.server.SPMDResource;
import appl.parallel.spmd.split.AbstractSplitMap;
import appl.parallel.spmd.split.AbstractSplitMap.NeighborhoodBoxingMode;
import appl.parallel.spmd.split.SinglePartitionInfo;
import appl.parallel.spmd.split.SplitMap;
import appl.parallel.spmd.split.SplittableResource;
import appl.parallel.thread.ComputingResourceThread;
import appl.parallel.thread.DataServerThread;
import edu.bonn.xulu.plugin.data.grid.MultiGrid;

/**
 * This class controls all the parallelization action on the client side and is
 * the counterpart to {@link SPMDServerController}. It is accessed by the model
 * developer by retrieving the {@link SPMDClientInterface} from the
 * {@link AbstractParallelStepModel}.
 *
 * @author Dominik Appl
 */
public class SPMDClientController implements SPMDClientInterface {

        /**
         * There are two states: <br>
         * <br>
         * <b>STATE.INIT</b> is the initializing state. In this state it is
         * allowed:<br>
         * to add resources to split control<br>
         * change neighborhodRange/boxing modes/reference resource<br>
         * <br>
         * All other methods are disabled and will throw
         * {@link UnsupportedOperationException}<br>
         * <br>
         * <b>STATE.RUN</b> is the running stage. All methods of {@link STATE#INIT}
         * are disabled and will throw {@link UnsupportedOperationException}. This
         * mode is automatically set by the first call to
         * {@link SPMDClientController#runSPMDModelTask(SPMDTask, Object...)}
         */
        public enum STATE {
                INIT, RUN
        }

        private STATE state = STATE.INIT;

        private final String splitMapClassPropertyName = "Parallel.splitmapforclass";

        private final Logger LOG = LogManager.getLogger(this.getClass().getName());

        /**
         * participating servers
         */
        private final SPMDResource[] servers;

        private final ComputingResourceProperties[] serverInfos;

        /**
         * stores the {@link PartitionDataServer DataServers} of the resources. They
         * are retrieved, after all resources are submitted to split control.
         */
        private final PartitionDataServer[] dataServers;;

        /**
         * number of partitions (== no of participating resources)
         */
        private final int noOfPartitions;

        /**
         * contains the {@link SinglePartitionInfo} of the
         * {@link SplittableResource splitted resources} for each server
         */
        private final Vector<SinglePartitionInfo>[] singlePartitionInfos;

        private final Vector<SinglePartitionInfo>[] toTransferPartitionInfos;

        private HashMap<String, Object> toTransferBaseParameters;

        private final HashMap<String, MultiDataInfo> multiDataInfos = new HashMap<String, MultiDataInfo>(
                        10);

        private final HashMap<String, MultiDataInfo> toTransferMultiDataObjects = new HashMap<String, MultiDataInfo>(
                        10);

        /** The IP-Addresses of the resources */
        private String[] IPs;

        /***************************************************************************
         * This Vector contains all IDs of the resources, which are currently under
         * splitControl
         *
         * @see #addToSplitControl(SplittableResource, String)
         * @see #mergePartition(SplittableResource)
         **************************************************************************/
        private Vector<Integer> listOfAllActivelyControlledData = new Vector<Integer>();

        /**
         * @see AbstractSplitMap.NeighborhoodBoxingMode
         */
        private NeighborhoodBoxingMode boxingMode = AbstractSplitMap.NeighborhoodBoxingMode.inBoxing;

        /** current neighborhood range, default is 0 */
        private int neighborhoodRange = 0;

        private final ClientDataServer spmdClient;

        /**
         * Local calculation bounds on server side are calculated using the
         * reference resource This is the ID of that resource
         */
        protected int referenceResourceID = -1;

        /** Thread execution pool */
        private final ExecutorService executor;

        /** true if there is data which has to be transfered to the servers */
        private boolean dataToTransfer = false;

        private SplitMap splitMap;

        private final CommEventSink eventProxy;

        private final double[] weights;

        /**
         * Creates a new Client controller.
         *
         * @param computingResources
         *            the resources which are used by this controller
         * @param spmdClient
         *            the spmd client responsible for the data retrieval
         * @param weights
         *            the weights for the distribution over the computing resources
         *            (values with a sum of 1) or null (distribution will be
         *            average)
         * @param eventProxy
         *            a {@link RemoteEventHandler} for eventHandling
         */
        @SuppressWarnings("unchecked")
        public SPMDClientController(
                        Vector<ComputingResourceContainer> computingResources,
                        double[] weights, ClientDataServer spmdClient,
                        CommEventSink eventProxy) {
                if (weights == null)
                        weights = averageWeights(computingResources.size());
                this.weights = weights;

                this.eventProxy = eventProxy;
                this.noOfPartitions = computingResources.size();
                if (noOfPartitions == 0)
                        throw new UnsupportedOperationException(
                                        "No Computing Ressources found");

                this.spmdClient = spmdClient;
                dataServers = new PartitionDataServer[noOfPartitions];
                serverInfos = new ComputingResourceProperties[noOfPartitions];
                servers = new SPMDResource[noOfPartitions];
                for (int i = 0; i < computingResources.size(); i++) {
                        serverInfos[i] = computingResources.get(i).getInformation();
                        servers[i] = (SPMDResource) computingResources.get(i).getResource();

                }
                // extract the IPs
                IPs = new String[noOfPartitions];
                for (int i = 0; i < IPs.length; i++) {
                        IPs[i] = serverInfos[i].getIP();
                        IPs[i] = serverInfos[i].getPort() == null ? IPs[i] : IPs[i] + ":"
                                        + serverInfos[i].getPort();

                        if (IPs[i] == null)
                                LOG
                                                .fatal("The IP-Information of a computingRessource was NULL. "
                                                                + "This will result in failure of the computation!");
                }

                // initialize the singlePartitionInfos and tasks. Notice that the array
                // singlePartitionInfos is final.
                // Each vector of infos is permanently associated with one server.
                this.singlePartitionInfos = new Vector[noOfPartitions];
                this.toTransferPartitionInfos = new Vector[noOfPartitions];
                this.toTransferBaseParameters = new HashMap<String, Object>();
                for (int i = 0; i < noOfPartitions; i++) {
                        singlePartitionInfos[i] = new Vector<SinglePartitionInfo>();
                        toTransferPartitionInfos[i] = new Vector<SinglePartitionInfo>();
                }

                // initialize executor
                executor = Executors.newCachedThreadPool();

                // connect to all participating servers
                connectAll();
                // initialize DataServers (must be after connect)
                for (int i = 0; i < noOfPartitions; i++)
                        try {
                                dataServers[i] = servers[i].createDataServer(IPs);
                        } catch (RemoteException e) {
                                LOG.fatal(e);
                                e.printStackTrace();
                        }
        }

        private double[] averageWeights(int noResources) {
                // create weighted rating with the same weight
                double[] weights = new double[noResources];
                for (int i = 0; i < weights.length; i++) {
                        weights[i] = 1 / noResources;
                }
                return weights;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#addBaseParameter(java.lang.Object,
         *      java.lang.String)
         */
        public void addBaseParameter(Object parameter, String parameterName) {
                toTransferBaseParameters.put(parameterName, parameter);
                dataToTransfer = true;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#addBaseParameters(java.lang.Object[],
         *      java.lang.String[])
         */
        public void addBaseParameters(Object[] parameters, String[] parameterNames) {
                for (int i = 0; i < parameterNames.length; i++) {
                        addBaseParameter(parameters[i], parameterNames[i]);
                }
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#addToMultiDataSplitControl(java.lang.Object[],
         *      java.lang.String)
         */
        public MultiDataObject addToMultiDataSplitControl(
                        Object splittableResources[], String name) {
                if (splittableResources.length == 0) {
                        throw new UnsupportedOperationException(
                                        "There must be at least one resource to create a MultiData Element");
                }
                // Check if adding is allowed:
                checkState(STATE.INIT);
                SplittableResource[] resources = checkSplittableArray(splittableResources);

                // add each element to splitcontrol. The constant gives each element a
                // unique name and identifies it on the server side as belonging to a multisplit

                MultiDataInfo multi = new MultiDataInfo(new int[0], name);
                for (int i = 0; i < resources.length; i++) {
                        multi.addElement(resources[i].getRootID());
                        addToSplitControl(resources[i], MultiDataInfo.getNameWithIdx(i,
                                        name));
                }
                toTransferMultiDataObjects.put(name, multi);
                multiDataInfos.put(name, multi);
                dataToTransfer = true;
                return new MultiDataObject(multi, spmdClient);
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#addToMultiDataSplitControl(edu.bonn.xulu.plugin.data.grid.MultiGrid,
         *      java.lang.String)
         */
        public MultiDataObject addToMultiDataSplitControl(MultiGrid multiGrid,
                        String name) {
                MultiDataObject dataObject = addToMultiDataSplitControl(multiGrid
                                .toArray(), name);
                dataObject.setManagedGrid(multiGrid);
                return dataObject;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#addToSplitControl(java.lang.Object,
         *      java.lang.String)
         */
        public void addToSplitControl(Object splittableResource, String name) {

                SplittableResource resource = checkSplittable(splittableResource);
                // Check if adding is allowed:
                checkState(STATE.INIT);
                // if first call, than this is the reference for now - a map will be
                // generated
                if (referenceResourceID == -1)
                        setReferenceResource(resource);

                // add the data to the SPMDClient for server retrieval
                spmdClient.addData(resource);
                // add the the resource to the controlled resources
                listOfAllActivelyControlledData.add(resource.getRootID());
                // make the singlePartitionInfos for each participating server and store
                // the
                // info for later use

                SplitMap map = getSplitMap();
                // create a partition info for each Server (only the
                // splitMapPosition differs)
                for (int i = 0; i < singlePartitionInfos.length; i++) {
                        PartitionDataHandler loader = PartitionHandlerFactory.newInstance(
                                        resource.getRootID(), spmdClient,
                                        map.getPartitionBounds(i), map
                                                        .getPartitionCalculationBounds(i));
                        SinglePartitionInfo info = new SinglePartitionInfo(resource
                                        .getRootID(), name, loader, map, i);
                        singlePartitionInfos[i].add(info);
                        toTransferPartitionInfos[i].add(info);
                        dataToTransfer = true;
                }
        }

        /**
         * Checks if the given object is an instance of {@link SplittableResource}
         * and gives it back as splittable.
         *
         * @param splittableResource
         * @throws UnsupportedOperationException
         *             if not instance of {@link SplittableResource}
         * @return the object as {@link SplittableResource}
         */
        protected SplittableResource checkSplittable(Object splittableResource) {
                if (!(splittableResource instanceof SplittableResource))
                        throw new UnsupportedOperationException(
                                        "Operation failed: the argument for 'addToSplitControl' \n "
                                                        + "must be an instance of SplittableResource! (like e.g. SplittableLLProxyGrid).\n "
                                                        + "You can add non splittable Objects as Parameters of the SPMDTasks! For arrays"
                                                        + "of SplittableResources use addToMultiSplitControl");
                return (SplittableResource) splittableResource;
        }

        /**
         * checks if every object of the array is an instance of
         * {@link SplittableResource} and gives it back as splittable
         *
         * @param splittableResources
         *            an array of (hopefully)
         *            {@link SplittableResource SplittableResources}
         * @throws UnsupportedOperationException
         *             if not all elements are instances of
         *             {@link SplittableResource} or if the splitHeights or
         *             SplitWidths do not match
         * @return the object as {@link SplittableResource} array
         */
        protected SplittableResource[] checkSplittableArray(
                        Object[] splittableResources) {
                SplittableResource[] res = new SplittableResource[splittableResources.length];
                for (int i = 0; i < splittableResources.length; i++) {
                        if (!(splittableResources[i] instanceof SplittableResource))
                                throw new UnsupportedOperationException(
                                                "Operation failed: the argument must be an instance "
                                                                + "of SplittableResource! (like e.g. SplittableLLProxyGrid).\n "
                                                                + "You can add non splittable Objects as Parameters of the SPMDTasks!");
                        res[i] = (SplittableResource) splittableResources[i];
                }
                // check if the splitLengths match
                if (res.length == 0)
                        return res;
                int splitHeight = res[0].getSplitHeight();
                int splitWidth = res[0].getSplitWidth();
                for (SplittableResource resource : res) {
                        if (resource.getSplitHeight() != splitHeight
                                        || resource.getSplitWidth() != splitWidth)
                                throw new UnsupportedOperationException(
                                                "Operation Failed: Splitvalues (height/width) of the array elements do not match!");
                }
                return res;
        }

        /**
         * Throws a {@link UnsupportedOperationException} if the state does not
         * match the required state
         *
         * @param requiredState
         *            the required state
         */
        private void checkState(STATE requiredState) {
                if (requiredState != state)
                        throw new UnsupportedOperationException(
                                        "This Operation is not available: " + "You are in state"
                                                        + state + " but you should be in state "
                                                        + requiredState + ". See documentation of "
                                                        + state.getClass().getName()
                                                        + " for more information ");
        }

        /**
         * disconnects from all servers
         */
        public void close() {
                disconnectAll();
        }

        /**
         * connects to all servers
         */
        @SuppressWarnings("unchecked")
        private void connectAll() {
                checkState(STATE.INIT);
                long l = System.currentTimeMillis();
                Future[] futureResults = new Future[servers.length];
                for (int i = 0; i < servers.length; i++) {
                        futureResults[i] = executor.submit(new ComputingResourceThread(
                                        servers[i], serverInfos[i], null, CommType.CONNECT,
                                        eventProxy, true) {
                                public Object run() throws Exception {
                                        getServer().connect();
                                        return null;
                                }
                        });
                }
                try {
                        // wait for threads to finish
                        for (Future future : futureResults) {
                                future.get();
                        }
                } catch (Exception e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
                System.out.println("Connect time: " + (System.currentTimeMillis() - l)
                                + " ms");
        }

        /**
         * disconnect from all servers
         */
        private void disconnectAll() {
                Future[] futureResults = new Future[servers.length];
                for (int i = 0; i < servers.length; i++) {
                        futureResults[i] = executor.submit(new ComputingResourceThread(
                                        servers[i], serverInfos[i], null, CommType.DISCONNECT,
                                        eventProxy, true) {
                                public Object run() throws Exception {
                                        getServer().disconnect();
                                        return null;
                                }
                        });
                }
                try {
                        // wait threads to finish
                        for (Future future : futureResults) {
                                future.get();
                        }
                } catch (Exception e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }

        }

        /**
         * @return the actual splitmap. The referenceResource must be set before
         *         call!
         */
        private SplitMap getSplitMap() {
                if (referenceResourceID == -1) {
                        LOG.error("NO MAP CREATED YET!");
                        return null;
                } else
                        return splitMap;
        }

        /**
         * @return the current state
         */
        public STATE getState() {
                return state;
        }

        /**
         * Creates a {@link SplitMap} for the specified resource. Use
         * {@link #getSplitMap()} for retrieval.
         *
         * @param splittable
         *            the resource, for which the {@link SplitMap} is created
         */
        private void makeSplitMap(SplittableResource splittable) {
                SplitMap map = null;

                // get splitMap implementation for this splittable from XuluConfig
                String classname = XuluConfig.getXuluConfig().getProperty(
                                splitMapClassPropertyName + "."
                                                + splittable.getClass().getSimpleName());
                // if no entry was found lookup default splitter
                if (classname == null)
                        classname = XuluConfig.getXuluConfig().getProperty(
                                        splitMapClassPropertyName + "." + "default");
                try {
                        map = (SplitMap) Class.forName(classname).newInstance();
                        map.setParameters(splittable, neighborhoodRange, noOfPartitions,
                                        boxingMode);
                        map.setWeights(weights);
                        map.makeMap();
                } catch (Exception e) {
                        String error = "Could not create Splitmap from classname : '"
                                        + classname + "' out of property '"
                                        + splitMapClassPropertyName + "'. Nested errormessage is :"
                                        + e.getMessage();
                        LOG.fatal(error, e);
                        throw new UnsupportedOperationException(error);

                }
                this.splitMap = map;

        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#mergeAllPartitions()
         */
        @SuppressWarnings("unchecked")
        public synchronized void mergeAllPartitions() {
                // clone the list first, because mergePartition(int) removes elements
                // from the list, which causes
                // problems with the for-each loop
                Vector<Integer> activeIDs = (Vector<Integer>) listOfAllActivelyControlledData
                                .clone();
                // Vector<Future> futures = new Vector<Future>();
                // for (Integer id : activeIDs) {
                // futures.add(executor.submit(new DataServerThread(null, id) {
                // public Object call() throws Exception {
                // mergePartition(getIntArgument());
                // return null;
                // }
                // }));
                // }
                // // wait on tasks to finish
                // for (Future future : futures) {
                // try {
                // future.get();
                // } catch (Exception e) {
                // e.printStackTrace();
                // }
                //
                // }
                for (Integer id : activeIDs) {
                        mergePartition((int) id);
                }
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#mergeMultiData(appl.parallel.spmd.MultiDataObject)
         */
        public synchronized void mergeMultiData(MultiDataObject multidata) {
                // the problem is that perhaps on serverside grids were added, but not
                // on client side. It is assumed that all servers have created the
                // same number of grids with the same names (which is assured by
                // multiDataInfo)

                // first lookup the local multiData info and the info of any
                // (here the first) dataserver multiInfo - as i said: all should be the
                // same
                MultiDataInfo localInfo = multidata.getMultiInfo();
                String name = multidata.getName();
                MultiDataInfo remoteInfo;
                try {
                        // all remote infos should be the same, so it is enough to retrieve
                        // the first one
                        remoteInfo = dataServers[0].getMultiDataInfo(name);
                        if (localInfo == null || remoteInfo == null) {
                                LOG.error("Could not lookup MultidataInfo with name " + name
                                                + ". localInfo was " + localInfo + " remote info was "
                                                + remoteInfo);
                                return;
                        }
                        // merge the ids which are there on both sides:
                        int i = 0;
                        for (; i < localInfo.getCount(); i++)
                                mergePartition(localInfo.getMultiID(i));
                        // merge the new partitions
                        for (; i < remoteInfo.getCount(); i++) {
                                // get the new ID (the same as on server side)
                                int newID = remoteInfo.getMultiID(i);
                                // first create the new data element out of the first element in
                                // the list:
                                multidata.addElement(newID);
                                SplittableResource newResource = (SplittableResource) multidata
                                                .getElement(multidata.getCount() - 1);
                                spmdClient.addData(newResource);
                                // now merge
                                mergePartition(newID);
                        }
                } catch (RemoteException e) {
                        LOG.error("ERROR while trying to merge multi data: "
                                        + e.getMessage(), e);
                }
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#mergeMultiData(appl.parallel.spmd.MultiDataObject,
         *      int)
         */
        public void mergeMultiData(MultiDataObject multidata, int idx) {
                // the problem is that perhaps on serverside grids were added, but not
                // on client side. It is assumed that all servers have created the
                // same number of grids with the same names (which is assured by
                // multiDataInfo)

                // first lookup the local multiData info and the info of any
                // (here the first) dataserver multiInfo - as i said: all should be the
                // same
                MultiDataInfo localInfo = multidata.getMultiInfo();
                String name = multidata.getName();
                MultiDataInfo remoteInfo;
                try {
                        // all remote infos should be the same, so it is enough to retrieve
                        // the first one
                        remoteInfo = dataServers[0].getMultiDataInfo(name);
                        if (localInfo == null || remoteInfo == null) {
                                LOG.error("Could not lookup MultidataInfo with name " + name
                                                + ". localInfo was " + localInfo + " remote info was "
                                                + remoteInfo);
                                return;
                        }
                        // merge the grid with the given index
                        // for a local idx:
                        if (idx < localInfo.getCount())
                                mergePartition(localInfo.getMultiID(idx));
                        // else create a new grid
                        else {
                                // get the new ID (the same as on server side)
                                int newID = remoteInfo.getMultiID(idx);
                                if (newID == 0)
                                        throw new UnsupportedOperationException(
                                                        "For the requested index (" + idx
                                                                        + ") was no grid found on the servers");
                                // first create the new data element out of the first element in
                                // the list:
                                multidata.addElement(newID);
                                SplittableResource newResource = (SplittableResource) multidata
                                                .getElement(multidata.getCount() - 1);
                                spmdClient.addData(newResource);
                                // now merge the serverdata into the new clientgrid
                                mergePartition(newID);
                        }
                } catch (RemoteException e) {
                        LOG.error("ERROR while trying to merge multi data: "
                                        + e.getMessage(), e);
                }
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#mergePartition(int)
         */
        public synchronized void mergePartition(int rootID) {
                checkState(STATE.RUN);
                Future[] futures = new Future[noOfPartitions];
                for (int i = 0; i < noOfPartitions; i++)
                        futures[i] = executor.submit(new DataServerThread(dataServers[i],
                                        serverInfos[i], rootID, CommType.CLIENT_MERGE, eventProxy) {
                                public Object run() throws Exception {
                                        getServer().unloadToSource(getIntArgument());
                                        return null;
                                }
                        });
                // wait on tasks to finish:
                try {
                        // wait threads to finish
                        for (Future future : futures) {
                                future.get();
                        }
                } catch (Exception e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
                listOfAllActivelyControlledData.remove((Integer) rootID);
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#mergePartition(java.lang.Object)
         */
        public void mergePartition(Object splittableResource) {
                checkSplittable(splittableResource);
                checkState(STATE.RUN);
                mergePartition(((SplittableResource) splittableResource).getRootID());
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#runSPMDModelTask(appl.parallel.spmd.SPMDTask,
         *      java.lang.Object[])
         */
        @SuppressWarnings("unchecked")
        public Object[] runSPMDModelTask(final SPMDTask task, Object... parameters)
                        throws Throwable {
                state = STATE.RUN;
                // transfer needed data
                if (dataToTransfer) {
                        transferDataToServers();
                }
                // submit task and arguments to the resources
                // each task is calculated in its own thread;
                // create tasks
                parameters = (parameters == null) ? new Object[0] : parameters;
                try {
                        Future futureResults[] = new Future[servers.length];
                        for (int i = 0; i < servers.length; i++) {
                                futureResults[i] = executor.submit(new ComputingResourceThread(
                                                servers[i], serverInfos[i], new Object[] { task,
                                                                referenceResourceID, parameters },
                                                CommType.CLIENT_EXECUTION, eventProxy) {
                                        public Object run() throws Exception {
                                                SPMDResource server = (SPMDResource) getServer();
                                                return server.runSPMDModelTask(task.getClass()
                                                                .getName(),
                                                                (Integer) getObjectArrayArgument()[1],
                                                                (Object[]) getObjectArrayArgument()[2]);
                                        }
                                });
                        }

                        Vector<Object> results = new Vector<Object>(15);

                        // aim of the next loop is to get single Object-array containing all
                        // results!
                        // Notice that every server result is a result array.
                        // This is because serverexecution can happen in multiple threads
                        // (when multiple processors are available) each serverthread will
                        // produce its own result
                        for (int i = 0; i < noOfPartitions; i++) {
                                // wait for threads finished and collect results
                                Object[] result = (Object[]) futureResults[i].get();
                                for (int j = 0; j < result.length; j++)
                                        results.add(result[j]);
                        }
                        return results.toArray();

                } catch (ExecutionException e) {
                        LOG.fatal("Error while trying to execute task "
                                        + task.getClass().getName() + ": " + e.getMessage(), e);
                        throw e.getCause();
                } catch (Exception e) {
                        LOG.error("Error while trying to execute task "
                                        + task.getClass().getName() + ": " + e.getMessage(), e);
                        e.printStackTrace();
                }
                return null;

        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#setBoxingMode(appl.parallel.spmd.split.AbstractSplitMap.NeighborhoodBoxingMode)
         */
        public void setBoxingMode(AbstractSplitMap.NeighborhoodBoxingMode boxingMode) {
                checkState(STATE.INIT);
                this.boxingMode = boxingMode;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#setNeighborhoodRange(int)
         */
        public void setNeighborhoodRange(int neighborhoodRange) {
                checkState(STATE.INIT);
                this.neighborhoodRange = neighborhoodRange;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#setReferenceResource(java.lang.Object)
         */
        public void setReferenceResource(Object splittableResource) {
                checkSplittable(splittableResource);
                SplittableResource res = (SplittableResource) splittableResource;
                checkState(STATE.INIT);
                if (res instanceof SplittableResource) {
                        referenceResourceID = res.getRootID();
                        makeSplitMap(res);
                } else
                        LOG
                                        .warn("Set reference failed: given resource was not an instance of SplittableResource!");

        }

        /**
         * transfers all collected data to the servers. (The data is collected to
         * save communication time)
         */
        private void transferDataToServers() {
                long time = System.currentTimeMillis();
                Vector<Future> futures = new Vector<Future>();
                if (toTransferPartitionInfos[0].size() > 0)
                        for (int i = 0; i < servers.length; i++) {
                                futures.add(executor.submit(new DataServerThread(
                                                dataServers[i], serverInfos[i],
                                                toTransferPartitionInfos[i],
                                                CommType.TRANSFER_METADATA, eventProxy, true) {
                                        public Object run() throws Exception {
                                                getServer()
                                                                .addPartitionInfos(
                                                                                (Vector<SinglePartitionInfo>) getObjectArgument());
                                                return null;
                                        }
                                }));
                        }
                if (toTransferMultiDataObjects.size() > 0)
                        for (int i = 0; i < servers.length; i++) {
                                futures.add(executor.submit(new DataServerThread(
                                                dataServers[i], serverInfos[i],
                                                toTransferMultiDataObjects, CommType.TRANSFER_METADATA,
                                                eventProxy, true) {
                                        public Object run() throws Exception {
                                                getServer()
                                                                .addMultiDataInfos(
                                                                                (HashMap<String, MultiDataInfo>) getObjectArgument());
                                                return null;
                                        }
                                }));
                        }
                if (toTransferBaseParameters.size() > 0)
                        for (int i = 0; i < servers.length; i++) {
                                futures.add(executor.submit(new DataServerThread(
                                                dataServers[i], serverInfos[i],
                                                toTransferBaseParameters, CommType.TRANSFER_PARAMETERS,
                                                eventProxy) {
                                        public Object run() throws Exception {
                                                getServer().updateBaseParameter(
                                                                (HashMap) getObjectArgument());
                                                return null;
                                        }
                                }));
                        }
                // wait on threads to finish
                for (Future future : futures) {
                        try {
                                future.get();
                        } catch (InterruptedException e) {
                                LOG.error(e);
                                e.printStackTrace();
                        } catch (ExecutionException e) {
                                LOG.error(e);
                                e.printStackTrace();
                        }
                }
                dataToTransfer = false;
                toTransferBaseParameters.clear();
                toTransferBaseParameters.clear();
                toTransferMultiDataObjects.clear();
                if (LOG.isDebugEnabled())
                        LOG.debug("Transfered Data to clients in "
                                        + (System.currentTimeMillis() - time) / 1000000 + " ms");
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#updateNeighborhood(java.lang.Object)
         */
        public void updateNeighborhood(Object splittableResource) {
                checkSplittable(splittableResource);
                SplittableResource res = (SplittableResource) splittableResource;
                updateNeighborhood(res.getRootID());
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.spmd.SPMDClientInterface#updateNeighborhood(appl.parallel.spmd.MultiDataObject,
         *      int)
         */
        public void updateNeighborhood(MultiDataObject multiDataObject, int index) {
                this.updateNeighborhood(multiDataObject.getMultiInfo()
                                .getMultiID(index));
        }

        /**
         * updates the neighborhood of the splittable with the given id on the
         * servers
         */
        private void updateNeighborhood(int rootID) {
                checkState(STATE.RUN);
                Future[] futureResults = new Future[servers.length];
                for (int i = 0; i < servers.length; i++) {
                        futureResults[i] = executor.submit(new DataServerThread(
                                        dataServers[i], serverInfos[i], rootID,
                                        CommType.CLIENT_UPDATE, eventProxy) {
                                public Object run() throws Exception {
                                        getServer().updateFromNeighbors(getIntArgument());
                                        return null;
                                }
                        });
                }
                try {
                        // wait threads to finish
                        for (Future future : futureResults) {
                                future.get();
                        }
                } catch (Exception e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
        }
}