parallel/server/PartitionDataManager.java

package appl.parallel.server;

import java.awt.Rectangle;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.rmi.Naming;
import java.rmi.NotBoundException;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashMap;
import java.util.Hashtable;
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.data.LoadingException;
import appl.parallel.client.ClientDataServer;
import appl.parallel.client.DataServer;
import appl.parallel.data.PartitionDataHandler;
import appl.parallel.event.CommEventSink;
import appl.parallel.event.TimeEvent;
import appl.parallel.event.TransferEvent;
import appl.parallel.event.CommEvent.CommType;
import appl.parallel.services.RemoteEventProxy;
import appl.parallel.spmd.MultiDataInfo;
import appl.parallel.spmd.MultiDataPartitionObject;
import appl.parallel.spmd.SPMDClientController;
import appl.parallel.spmd.split.DataPartition;
import appl.parallel.spmd.split.PartitionInfo;
import appl.parallel.spmd.split.SinglePartitionInfo;
import appl.parallel.spmd.split.SplitMap;
import appl.parallel.thread.DataServerThread;

/**
 * Manages the all data for the {@link XuluServer}. This includes retrieval of
 * partitions from connected resources like the {@link DataServer} or, as part
 * of the update of neighborhood regions, from other
 * {@link XuluServer XuluServers}.
 *
 * @author Dominik Appl
 */
public class PartitionDataManager extends UnicastRemoteObject implements
                PartitionDataServer {

        /**
         * The Thread is used for updating the partitions neighborhood region from
         * the other servers.
         *
         * @author Dominik Appl
         */
        class UpdateThread extends DataServerThread {

                private final DataPartition thisPartition;

                private final Rectangle updateBounds;

                private final int id;

                private final String hostname;

                public UpdateThread(PartitionDataServer dataServer,
                                String ipOrHostname, DataPartition thisPartition,
                                Rectangle updateBounds, int id, CommEventSink sink) {
                        super(dataServer, null, null, CommType.REMOTE_UPDATE, sink);
                        hostname = ipOrHostname;
                        // TODO Auto-generated constructor stub
                        this.thisPartition = thisPartition;
                        this.updateBounds = updateBounds;
                        this.id = id;
                }

                /*
                 * (non-Javadoc)
                 *
                 * @see appl.parallel.thread.ExecutionThread#fireTimeEvents(long,
                 *      java.lang.Object)
                 */
                @Override
                protected void fireTimeEvents(long execTime, Object result) {
                        try {
                                eventSink.fireRemoteEvent(new TimeEvent(execTime,
                                                dataServerName, hostname, commType));
                        } catch (RemoteException e) {
                                e.printStackTrace();
                        }
                }

                /*
                 * (non-Javadoc)
                 *
                 * @see appl.parallel.thread.ExecutionThread#fireTransferEvent(java.lang.Object)
                 */
                @Override
                protected void fireTransferEvent(Object result) {
                        try {
                                eventSink.fireRemoteEvent(new TransferEvent(dataServerName,
                                                hostname, commType, new Object[] { result }));
                        } catch (RemoteException e) {
                                e.printStackTrace();
                        }
                }

                /*
                 * (non-Javadoc)
                 *
                 * @see appl.parallel.thread.ExecutionThread#run()
                 */
                @Override
                protected Object run() throws Exception {
                        try {
                                DataPartition updateData = getServer().getPartition(id,
                                                updateBounds);
                                // apply update:
                                this.thisPartition.setPartition(updateData, updateBounds);
                                return updateData;
                        } catch (Exception e) {
                                LOG.error("Error while retrieving partition " + updateBounds
                                                + "from PartitionDataServer with ip " + hostname);
                        }
                        return null;
                }
        }

        private final String bindingName = "PartitionDataServer";

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

        /**
         * all partition data. The key is the id. Is an Hashtable instead of
         * HashMasp for thread-safty.
         */
        private Hashtable<Integer, DataPartition> data = new Hashtable<Integer, DataPartition>(
                        15);

        /**
         * the names of the data objects
         */
        private HashMap<String, Integer> NameToIDMapping = new HashMap<String, Integer>(
                        15);

        private HashMap<Integer, SinglePartitionInfo> infos = new HashMap<Integer, SinglePartitionInfo>(
                        10);

        private HashMap<String, Object> baseParameters = new HashMap<String, Object>(
                        10);

        private final PartitionDataServer[] dataServers;

        private final ExecutorService executor;

        private final String[] IPs;

        private boolean dataServersInitialized = false;

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

        private final String dataServerName;

        private final CommEventSink remoteEventReceiver;

        private String[] hostnames;

        private ClientDataServer localSPMDClient;

        private final int registryPort;

        /**
         * Creates and inits a new PartitionDataserver.
         *
         * @param IPs
         *            the IPs of participating Servers for neighborhood updates(may
         *            include port specification)
         * @param remoteEventReceiver
         *            generated events are forwarded to the receiver
         * @param registryPort
         *            port of the registry to which the {@link PartitionDataManager}
         *            should be bound
         * @throws RemoteException
         *             if a connection to a participating server fails
         */
        public PartitionDataManager(String[] IPs,
                        CommEventSink remoteEventReceiver, int registryPort)
                        throws RemoteException {
                // create NameToID and IdToInfo-Mapping
                this.IPs = IPs;
                this.remoteEventReceiver = remoteEventReceiver;
                this.registryPort = registryPort;

                dataServers = new PartitionDataServer[IPs.length];
                startTheServer();

                // start Thread service
                executor = Executors.newCachedThreadPool();
                String name = "unknown PartitionDataServer";
                try {
                        name = InetAddress.getLocalHost().getHostName();
                } catch (UnknownHostException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
                dataServerName = name;
                // try to get the hostnames
                hostnames = new String[IPs.length];
                for (int i = 0; i < IPs.length; i++) {
                        // try {
                        hostnames[i] = IPs[i];
                        // hostnames[i] = InetAddress.getByName(IPs[i]).getHostName();
                        // } catch (UnknownHostException e) {
                        // LOG.warn("Exception while retrivieving hostname of address: " +
                        // IPs[i]);
                        // }
                }
        }

        /**
         * Same as {@link #PartitionDataManager(String[], CommEventSink, int)}, but uses
         * a local SPMDClient. <br>
         * This is used when a Server is running inside the Xulu-Client. It has
         * performance advantages (direct access - no TCP/IP)
         *
         * @param IPs
         *            the IPs of participating Servers for neighborhood updates(may
         *            include port specification)
         * @param remoteEventReceiver
         *            generated events are forwarded to the receiver
         * @param registryport
         *            port of the registry to which the {@link PartitionDataManager}
         *            should be bound
         * @param localSPMDClient
         *            a {@link ClientDataServer} for fast local access
         * @throws RemoteException
         */
        public PartitionDataManager(String[] IPs,
                        RemoteEventProxy remoteEventReceiver, int registryport,
                        ClientDataServer localSPMDClient) throws RemoteException {
                this(IPs, remoteEventReceiver, registryport);
                this.localSPMDClient = localSPMDClient;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.client.DataServer#addData(appl.parallel.spmd.split.DataPartition)
         */
        public void addData(DataPartition partition) throws RemoteException {
                if (partition == null) {
                        LOG.warn("To add partition was null - no data added");
                        return;
                }
                data.put(partition.getRootID(), partition);

        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#addMultiDataInfos(java.util.HashMap)
         */
        public void addMultiDataInfos(
                        HashMap<String, MultiDataInfo> newMultiDataInfos)
                        throws RemoteException {
                this.multiDataInfos.putAll(newMultiDataInfos);
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#addPartitionInfos(java.util.Vector)
         */
        public void addPartitionInfos(
                        Vector<SinglePartitionInfo> singlePartitionInfos)
                        throws RemoteException {
                for (SinglePartitionInfo singlePartitionInfo : singlePartitionInfos) {
                        NameToIDMapping.put(singlePartitionInfo.getBaseResourceName(),
                                        singlePartitionInfo.getBaseResourceID());
                        infos.put(singlePartitionInfo.getBaseResourceID(),
                                        singlePartitionInfo);
                }
        }

        /**
         * @param name
         * @param idx
         */
        public void destroyMultiPartition(String name, int idx) {

        }

        /**
         * returns the baseParameter by name
         *
         * @param name
         *            the name of the parameter
         * @return the parameter
         */
        public Object getBaseParameter(String name) {
                return baseParameters.get(name);
        }

        /**
         * Returns a a (remote) PartitionDataServer for an IP-Address. Looks up the
         * PartitionDataServer in the registry
         */
        private PartitionDataServer getConnection(String ip) {
                PartitionDataServer server;
                // lookup server
                try {
                        server = (PartitionDataServer) Naming.lookup("rmi://" + ip + "/"
                                        + bindingName);
                } catch (Exception e) {
                        LOG.error("bindig of " + ip + " failed" + e.getMessage(), e);
                        e.printStackTrace();
                        return null;
                }
                return server;
        }

        /**
         * Returns the {@link DataPartition} associated with the given ID
         *
         * @param id
         *            the id of the data
         * @return the data
         */
        public synchronized DataPartition getData(int id) throws RemoteException {
                // check if in hashtable already
                DataPartition partition = data.get(id);
                if (partition != null)
                        return partition;
                // if not in hashtable:
                // load the data and put it into hash
                // find the element with the ID:
                SinglePartitionInfo info = infos.get(id);

                if (info == null) {
                        LOG.warn("No partition with id " + id + " found!");
                        return null;
                }
                // retrieve data from client or other location
                try {
                        // if there is a local server try to get the local partition
                        PartitionDataHandler partitionDataHandler = info
                                        .getPartitionDataHandler();
                        if (localSPMDClient != null)
                                partitionDataHandler.setSPMDClient(localSPMDClient);
                        partition = partitionDataHandler.load();
                } catch (LoadingException e) {
                        // TODO Auto-generated catch block
                        LOG.error("Loading of partition with id '" + id
                                        + "' failed. Null returned!");
                        e.printStackTrace();
                        return null;
                }
                if (partition == null) {
                        LOG.error("Loading of partition with id '" + id
                                        + "' failed. Null returned!");
                        return null;
                }

                // enter data in partition table
                data.put(id, partition);
                return partition;
        }

        /**
         * Gets a partition by name
         *
         * @param name
         *            the name of the resource (probably given by the programmer)
         * @return the partition or null if not found
         */
        public DataPartition getData(String name) {
                Integer id = NameToIDMapping.get(name);
                if (id == null) {
                        LOG.warn("No data for name " + name + " found! Returning null");
                        return null;
                }
                try {
                        return getData(id);
                } catch (RemoteException e) {
                        // should never be reached (because the method call is local
                        e.printStackTrace();
                }
                return null;
        }

        /**
         * Returns the partition info with the specified id
         *
         * @param id
         *            the id of the partition
         * @return the info
         */
        public PartitionInfo getInfo(int id) {
                return infos.get(id);
        }

        /**
         * Gets a partition of a multi-data element. NOTE THAT ALL PARTITIONS OF THE
         * MULTIGRID WILL BE LOADED (watch out for loading performance!).
         *
         * @param name
         *            the name of the resource (probably given by the programmer)
         * @return the partition
         * @see SPMDClientController#addToMultiDataSplitControl(Object[], String)
         */
        public DataPartition[] getMultiData(String name) {
                // fill an array with the partitions
                DataPartition[] partitions = new DataPartition[multiDataInfos.get(name)
                                .getCount()];
                for (int i = 0; i < partitions.length; i++) {
                        partitions[i] = getMultiData(name, i);
                }
                return partitions;
        }

        /**
         * Gets a partition of a multi-data element. Only the resource at the given
         * position will be loaded.
         *
         * @param name
         *            the name of the resource (probably given by the programmer)
         * @param pos
         *            the index of the requested element
         * @return the partition
         * @see SPMDClientController#addToMultiDataSplitControl(Object[], String)
         */
        public DataPartition getMultiData(String name, int pos) {
                try {
                        return getData(multiDataInfos.get(name).getMultiID(pos));
                } catch (RemoteException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
                return null;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#getMultiDataInfo(java.lang.String)
         */
        public MultiDataInfo getMultiDataInfo(String name) throws RemoteException {
                return multiDataInfos.get(name);
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#getMultiDataObject(java.lang.String)
         */
        public MultiDataPartitionObject getMultiDataObject(String name)
                        throws RemoteException {
                return new MultiDataPartitionObject(multiDataInfos.get(name), this);
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#getPartition(int,
         *      java.awt.Rectangle)
         */
        public synchronized DataPartition getPartition(int id, Rectangle bounds)
                        throws RemoteException {
                DataPartition thisPartition = (DataPartition) data.get(id);
                if (thisPartition == null && LOG.isDebugEnabled()) {
                        LOG.error("Partition with id " + id + " and bounds " + bounds
                                        + "was requested but not found on DataManager");
                        return null;
                }

                if (LOG.isDebugEnabled())
                        LOG.debug("Data retrieved from DataManager: ID:" + id
                                        + " partition: " + bounds);
                return thisPartition.getPartition(bounds);
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#getPartitionInfo(int)
         */
        public SinglePartitionInfo getPartitionInfo(int rootID)
                        throws RemoteException {
                return infos.get(rootID);
        }

        /**
         * other dataservers may not be initialized on creation of this object.
         * Therefore the dataServers must be looked up later
         */
        private void initializeDataServers() {
                if (dataServersInitialized == true)
                        return;
                for (int i = 0; i < dataServers.length; i++) {
                        dataServers[i] = getConnection(IPs[i]);
                }
                dataServersInitialized = true;
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.client.DataServer#removeData(int)
         */
        public void removeData(int id) {
                data.remove(id);

        }

        /**
         * Removes the partition from the database
         *
         * @param name
         *            the name of the partition to remove
         */
        public void removeData(String name) {
                data.remove(NameToIDMapping.get(name));
        }

        /**
         * Removes the partition from the database
         *
         * @param partition
         *            the Partition to remove
         */
        public void removePartition(DataPartition partition) {
                data.remove(partition.getRootID());
        }

        /**
         * binds the server to the registry
         */
        private void startTheServer() throws RemoteException {
                Registry registry = LocateRegistry.getRegistry(registryPort);
                registry.rebind(bindingName, (PartitionDataServer) this);
                LOG.info("PartitionDataServer bound to port " + registryPort
                                + " and running....");
        }

        /**
         * unbinds the server from the registry, removes all data and runs the
         * garbage collector
         */
        public void stop() {
                data.clear();
                NameToIDMapping.clear();
                infos.clear();

                Registry registry;
                try {
                        registry = LocateRegistry.getRegistry();
                        registry.unbind(bindingName);
                        LOG.debug("Sucessfully removed " + bindingName + " from reggie");
                } catch (RemoteException e) {
                        LOG.warn("tried to unbind " + bindingName
                                        + " from reggie, but an exception occured: "
                                        + e.getMessage());
                } catch (NotBoundException e) {
                        LOG.warn("tried to unbind " + bindingName
                                        + " from reggie, but the name was not bound "
                                        + e.getMessage());
                }
                System.gc();
        }

        /**
         * @param rootID
         */
        public void unloadToSource(int rootID) throws RemoteException {
                SinglePartitionInfo info = infos.get(rootID);
                if (info == null) {
                        LOG.error("Could not unload partition with id " + rootID
                                        + ". The partitioninfo was not found!");
                        return;
                }
                // start unloading
                info.getPartitionDataHandler().setBasePartition(this.getData(rootID));
                info.getPartitionDataHandler().unload();
                // // remove data from memory and run garbage collection
                // data.remove(rootID);
                // System.gc();
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#updateBaseParameter(java.lang.Object[],
         *      java.lang.String[])
         */
        public void updateBaseParameter(HashMap<String, Object> newParameters)
                        throws RemoteException {
                // update the mapping. Old values are replaced automaticly
                baseParameters.putAll(newParameters);
        }

        /**
         * Updates the specified partition. For this the the neighborhood partitions
         * are updated from remote DataServers using the {@link SplitMap} of the
         * {@link SinglePartitionInfo}.
         *
         * @param id
         */
        public void updateFromNeighbors(int id) {
                initializeDataServers();
                // get the data to be updated:
                DataPartition thisPartition = data.get(id);
                PartitionInfo info = infos.get(id);
                if (thisPartition == null || info == null) {
                        LOG.error("Update with id " + id
                                        + " failed, because no data was found with this id");
                        return;
                }

                // get some other needed infos
                SplitMap map = info.getSplitMap();
                int mapPosition = info.getSplitMapPos();
                // check if something to do
                if (map.getNeighborhoodRange() == 0) {
                        LOG.debug("Update has nothing to do (neighborhoodrange is 0)");
                        return;
                }

                // calculate (for each neighbor) the needed partitions and get
                // the updates (implemented using Threads)
                int[] neighbors = map.getNeighborsForPosition(mapPosition);
                Future futures[] = new Future[neighbors.length];
                for (int j = 0; j < neighbors.length; j++) {
                        int neighborIdx = neighbors[j];
                        // get the overlap of the neighborhood area of the current partion
                        // with the calcultation area of the remote partition
                        Rectangle updateBounds = map.getPartitionNeighborhoodBounds(
                                        mapPosition).intersection(
                                        map.getPartitionCalculationBounds(neighborIdx));
                        // Get the partition from remote server:
                        DataPartition updateData;
                        futures[j] = executor.submit(new UpdateThread(
                                        dataServers[neighborIdx], hostnames[neighborIdx],
                                        thisPartition, updateBounds, id, remoteEventReceiver));
                }

                for (Future future : futures) {
                        try {
                                future.get();
                        } catch (InterruptedException e) {
                                e.printStackTrace();
                        } catch (ExecutionException e) {
                                LOG.error("The Remote update of a partition failed", e);
                        }
                }

        }

        /**
         * Does the same as {@link #updateFromNeighbors(int)}
         *
         * @param name
         *            the name of the resource
         */
        public void updateFromNeighbors(String name) {
                // get by ID
                updateFromNeighbors(NameToIDMapping.get(name));
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#setPartition(int,
         *      appl.parallel.spmd.split.DataPartition, java.awt.Rectangle)
         */
        public synchronized void updatePartition(int id, DataPartition partition,
                        Rectangle bounds) throws RemoteException {
                DataPartition thisPartition = (DataPartition) data.get(id);
                if (thisPartition == null && LOG.isDebugEnabled()) {
                        LOG
                                        .error("Partition of data with id " + id + " and bounds "
                                                        + bounds
                                                        + "should be set but was not found in DataManager");
                        return;
                }
                thisPartition.setPartition(partition, bounds);
                if (LOG.isDebugEnabled())
                        LOG.debug("Partition of data set on DataManager with id " + id
                                        + " and partition bounds: " + bounds);

        }

}
1	mojays	2	package appl.parallel.server;
2
3			import java.awt.Rectangle;
4			import java.net.InetAddress;
5			import java.net.UnknownHostException;
6			import java.rmi.Naming;
7			import java.rmi.NotBoundException;
8			import java.rmi.RemoteException;
9			import java.rmi.registry.LocateRegistry;
10			import java.rmi.registry.Registry;
11			import java.rmi.server.UnicastRemoteObject;
12			import java.util.HashMap;
13			import java.util.Hashtable;
14			import java.util.Vector;
15			import java.util.concurrent.ExecutionException;
16			import java.util.concurrent.ExecutorService;
17			import java.util.concurrent.Executors;
18			import java.util.concurrent.Future;
19
20			import org.apache.log4j.LogManager;
21			import org.apache.log4j.Logger;
22
23			import appl.data.LoadingException;
24	alfonx	78	import appl.parallel.client.ClientDataServer;
25	mojays	2	import appl.parallel.client.DataServer;
26			import appl.parallel.data.PartitionDataHandler;
27			import appl.parallel.event.CommEventSink;
28			import appl.parallel.event.TimeEvent;
29			import appl.parallel.event.TransferEvent;
30			import appl.parallel.event.CommEvent.CommType;
31			import appl.parallel.services.RemoteEventProxy;
32			import appl.parallel.spmd.MultiDataInfo;
33			import appl.parallel.spmd.MultiDataPartitionObject;
34			import appl.parallel.spmd.SPMDClientController;
35			import appl.parallel.spmd.split.DataPartition;
36			import appl.parallel.spmd.split.PartitionInfo;
37			import appl.parallel.spmd.split.SinglePartitionInfo;
38			import appl.parallel.spmd.split.SplitMap;
39			import appl.parallel.thread.DataServerThread;
40
41			/**
42			* Manages the all data for the {@link XuluServer}. This includes retrieval of
43			* partitions from connected resources like the {@link DataServer} or, as part
44			* of the update of neighborhood regions, from other
45			* {@link XuluServer XuluServers}.
46			*
47			* @author Dominik Appl
48			*/
49			public class PartitionDataManager extends UnicastRemoteObject implements
50			PartitionDataServer {
51
52			/**
53			* The Thread is used for updating the partitions neighborhood region from
54			* the other servers.
55			*
56			* @author Dominik Appl
57			*/
58			class UpdateThread extends DataServerThread {
59
60			private final DataPartition thisPartition;
61
62			private final Rectangle updateBounds;
63
64			private final int id;
65
66			private final String hostname;
67
68			public UpdateThread(PartitionDataServer dataServer,
69			String ipOrHostname, DataPartition thisPartition,
70			Rectangle updateBounds, int id, CommEventSink sink) {
71			super(dataServer, null, null, CommType.REMOTE_UPDATE, sink);
72			hostname = ipOrHostname;
73			// TODO Auto-generated constructor stub
74			this.thisPartition = thisPartition;
75			this.updateBounds = updateBounds;
76			this.id = id;
77			}
78
79			/*
80			* (non-Javadoc)
81			*
82			* @see appl.parallel.thread.ExecutionThread#fireTimeEvents(long,
83			* java.lang.Object)
84			*/
85			@Override
86			protected void fireTimeEvents(long execTime, Object result) {
87			try {
88			eventSink.fireRemoteEvent(new TimeEvent(execTime,
89			dataServerName, hostname, commType));
90			} catch (RemoteException e) {
91			e.printStackTrace();
92			}
93			}
94
95			/*
96			* (non-Javadoc)
97			*
98			* @see appl.parallel.thread.ExecutionThread#fireTransferEvent(java.lang.Object)
99			*/
100			@Override
101			protected void fireTransferEvent(Object result) {
102			try {
103			eventSink.fireRemoteEvent(new TransferEvent(dataServerName,
104			hostname, commType, new Object[] { result }));
105			} catch (RemoteException e) {
106			e.printStackTrace();
107			}
108			}
109
110			/*
111			* (non-Javadoc)
112			*
113			* @see appl.parallel.thread.ExecutionThread#run()
114			*/
115			@Override
116			protected Object run() throws Exception {
117			try {
118			DataPartition updateData = getServer().getPartition(id,
119			updateBounds);
120			// apply update:
121			this.thisPartition.setPartition(updateData, updateBounds);
122			return updateData;
123			} catch (Exception e) {
124			LOG.error("Error while retrieving partition " + updateBounds
125			+ "from PartitionDataServer with ip " + hostname);
126			}
127			return null;
128			}
129			}
130
131			private final String bindingName = "PartitionDataServer";
132
133			private final Logger LOG = LogManager.getLogger(this.getClass().getName());
134
135			/**
136			* all partition data. The key is the id. Is an Hashtable instead of
137			* HashMasp for thread-safty.
138			*/
139			private Hashtable<Integer, DataPartition> data = new Hashtable<Integer, DataPartition>(
140			15);
141
142			/**
143			* the names of the data objects
144			*/
145			private HashMap<String, Integer> NameToIDMapping = new HashMap<String, Integer>(
146			15);
147
148			private HashMap<Integer, SinglePartitionInfo> infos = new HashMap<Integer, SinglePartitionInfo>(
149			10);
150
151			private HashMap<String, Object> baseParameters = new HashMap<String, Object>(
152			10);
153
154			private final PartitionDataServer[] dataServers;
155
156			private final ExecutorService executor;
157
158			private final String[] IPs;
159
160			private boolean dataServersInitialized = false;
161
162			private HashMap<String, MultiDataInfo> multiDataInfos = new HashMap<String, MultiDataInfo>(
163			10);
164
165			private final String dataServerName;
166
167			private final CommEventSink remoteEventReceiver;
168
169			private String[] hostnames;
170
171			private ClientDataServer localSPMDClient;
172
173			private final int registryPort;
174
175			/**
176			* Creates and inits a new PartitionDataserver.
177			*
178			* @param IPs
179			* the IPs of participating Servers for neighborhood updates(may
180			* include port specification)
181			* @param remoteEventReceiver
182			* generated events are forwarded to the receiver
183			* @param registryPort
184			* port of the registry to which the {@link PartitionDataManager}
185			* should be bound
186			* @throws RemoteException
187			* if a connection to a participating server fails
188			*/
189			public PartitionDataManager(String[] IPs,
190			CommEventSink remoteEventReceiver, int registryPort)
191			throws RemoteException {
192			// create NameToID and IdToInfo-Mapping
193			this.IPs = IPs;
194			this.remoteEventReceiver = remoteEventReceiver;
195			this.registryPort = registryPort;
196
197			dataServers = new PartitionDataServer[IPs.length];
198			startTheServer();
199
200			// start Thread service
201			executor = Executors.newCachedThreadPool();
202			String name = "unknown PartitionDataServer";
203			try {
204			name = InetAddress.getLocalHost().getHostName();
205			} catch (UnknownHostException e) {
206			// TODO Auto-generated catch block
207			e.printStackTrace();
208			}
209			dataServerName = name;
210			// try to get the hostnames
211			hostnames = new String[IPs.length];
212			for (int i = 0; i < IPs.length; i++) {
213			// try {
214			hostnames[i] = IPs[i];
215			// hostnames[i] = InetAddress.getByName(IPs[i]).getHostName();
216			// } catch (UnknownHostException e) {
217			// LOG.warn("Exception while retrivieving hostname of address: " +
218			// IPs[i]);
219			// }
220			}
221			}
222
223			/**
224			* Same as {@link #PartitionDataManager(String[], CommEventSink, int)}, but uses
225			* a local SPMDClient. <br>
226			* This is used when a Server is running inside the Xulu-Client. It has
227			* performance advantages (direct access - no TCP/IP)
228			*
229			* @param IPs
230			* the IPs of participating Servers for neighborhood updates(may
231			* include port specification)
232			* @param remoteEventReceiver
233			* generated events are forwarded to the receiver
234			* @param registryport
235			* port of the registry to which the {@link PartitionDataManager}
236			* should be bound
237			* @param localSPMDClient
238			* a {@link ClientDataServer} for fast local access
239			* @throws RemoteException
240			*/
241			public PartitionDataManager(String[] IPs,
242			RemoteEventProxy remoteEventReceiver, int registryport,
243			ClientDataServer localSPMDClient) throws RemoteException {
244			this(IPs, remoteEventReceiver, registryport);
245			this.localSPMDClient = localSPMDClient;
246			}
247
248			/*
249			* (non-Javadoc)
250			*
251			* @see appl.parallel.client.DataServer#addData(appl.parallel.spmd.split.DataPartition)
252			*/
253			public void addData(DataPartition partition) throws RemoteException {
254			if (partition == null) {
255			LOG.warn("To add partition was null - no data added");
256			return;
257			}
258			data.put(partition.getRootID(), partition);
259
260			}
261
262			/*
263			* (non-Javadoc)
264			*
265			* @see appl.parallel.server.PartitionDataServer#addMultiDataInfos(java.util.HashMap)
266			*/
267			public void addMultiDataInfos(
268			HashMap<String, MultiDataInfo> newMultiDataInfos)
269			throws RemoteException {
270			this.multiDataInfos.putAll(newMultiDataInfos);
271			}
272
273			/*
274			* (non-Javadoc)
275			*
276			* @see appl.parallel.server.PartitionDataServer#addPartitionInfos(java.util.Vector)
277			*/
278			public void addPartitionInfos(
279			Vector<SinglePartitionInfo> singlePartitionInfos)
280			throws RemoteException {
281			for (SinglePartitionInfo singlePartitionInfo : singlePartitionInfos) {
282			NameToIDMapping.put(singlePartitionInfo.getBaseResourceName(),
283			singlePartitionInfo.getBaseResourceID());
284			infos.put(singlePartitionInfo.getBaseResourceID(),
285			singlePartitionInfo);
286			}
287			}
288
289			/**
290			* @param name
291			* @param idx
292			*/
293			public void destroyMultiPartition(String name, int idx) {
294
295			}
296
297			/**
298			* returns the baseParameter by name
299			*
300			* @param name
301			* the name of the parameter
302			* @return the parameter
303			*/
304			public Object getBaseParameter(String name) {
305			return baseParameters.get(name);
306			}
307
308			/**
309			* Returns a a (remote) PartitionDataServer for an IP-Address. Looks up the
310			* PartitionDataServer in the registry
311			*/
312			private PartitionDataServer getConnection(String ip) {
313			PartitionDataServer server;
314			// lookup server
315			try {
316			server = (PartitionDataServer) Naming.lookup("rmi://" + ip + "/"
317			+ bindingName);
318			} catch (Exception e) {
319			LOG.error("bindig of " + ip + " failed" + e.getMessage(), e);
320			e.printStackTrace();
321			return null;
322			}
323			return server;
324			}
325
326			/**
327			* Returns the {@link DataPartition} associated with the given ID
328			*
329			* @param id
330			* the id of the data
331			* @return the data
332			*/
333			public synchronized DataPartition getData(int id) throws RemoteException {
334			// check if in hashtable already
335			DataPartition partition = data.get(id);
336			if (partition != null)
337			return partition;
338			// if not in hashtable:
339			// load the data and put it into hash
340			// find the element with the ID:
341			SinglePartitionInfo info = infos.get(id);
342
343			if (info == null) {
344			LOG.warn("No partition with id " + id + " found!");
345			return null;
346			}
347			// retrieve data from client or other location
348			try {
349			// if there is a local server try to get the local partition
350			PartitionDataHandler partitionDataHandler = info
351			.getPartitionDataHandler();
352			if (localSPMDClient != null)
353			partitionDataHandler.setSPMDClient(localSPMDClient);
354			partition = partitionDataHandler.load();
355			} catch (LoadingException e) {
356			// TODO Auto-generated catch block
357			LOG.error("Loading of partition with id '" + id
358			+ "' failed. Null returned!");
359			e.printStackTrace();
360			return null;
361			}
362			if (partition == null) {
363			LOG.error("Loading of partition with id '" + id
364			+ "' failed. Null returned!");
365			return null;
366			}
367
368			// enter data in partition table
369			data.put(id, partition);
370			return partition;
371			}
372
373			/**
374			* Gets a partition by name
375			*
376			* @param name
377			* the name of the resource (probably given by the programmer)
378			* @return the partition or null if not found
379			*/
380			public DataPartition getData(String name) {
381			Integer id = NameToIDMapping.get(name);
382			if (id == null) {
383			LOG.warn("No data for name " + name + " found! Returning null");
384			return null;
385			}
386			try {
387			return getData(id);
388			} catch (RemoteException e) {
389			// should never be reached (because the method call is local
390			e.printStackTrace();
391			}
392			return null;
393			}
394
395			/**
396			* Returns the partition info with the specified id
397			*
398			* @param id
399			* the id of the partition
400			* @return the info
401			*/
402			public PartitionInfo getInfo(int id) {
403			return infos.get(id);
404			}
405
406			/**
407			* Gets a partition of a multi-data element. NOTE THAT ALL PARTITIONS OF THE
408			* MULTIGRID WILL BE LOADED (watch out for loading performance!).
409			*
410			* @param name
411			* the name of the resource (probably given by the programmer)
412			* @return the partition
413			* @see SPMDClientController#addToMultiDataSplitControl(Object[], String)
414			*/
415			public DataPartition[] getMultiData(String name) {
416			// fill an array with the partitions
417			DataPartition[] partitions = new DataPartition[multiDataInfos.get(name)
418			.getCount()];
419			for (int i = 0; i < partitions.length; i++) {
420			partitions[i] = getMultiData(name, i);
421			}
422			return partitions;
423			}
424
425			/**
426			* Gets a partition of a multi-data element. Only the resource at the given
427			* position will be loaded.
428			*
429			* @param name
430			* the name of the resource (probably given by the programmer)
431			* @param pos
432			* the index of the requested element
433			* @return the partition
434			* @see SPMDClientController#addToMultiDataSplitControl(Object[], String)
435			*/
436			public DataPartition getMultiData(String name, int pos) {
437			try {
438			return getData(multiDataInfos.get(name).getMultiID(pos));
439			} catch (RemoteException e) {
440			// TODO Auto-generated catch block
441			e.printStackTrace();
442			}
443			return null;
444			}
445
446			/*
447			* (non-Javadoc)
448			*
449			* @see appl.parallel.server.PartitionDataServer#getMultiDataInfo(java.lang.String)
450			*/
451			public MultiDataInfo getMultiDataInfo(String name) throws RemoteException {
452			return multiDataInfos.get(name);
453			}
454
455			/*
456			* (non-Javadoc)
457			*
458			* @see appl.parallel.server.PartitionDataServer#getMultiDataObject(java.lang.String)
459			*/
460			public MultiDataPartitionObject getMultiDataObject(String name)
461			throws RemoteException {
462			return new MultiDataPartitionObject(multiDataInfos.get(name), this);
463			}
464
465			/*
466			* (non-Javadoc)
467			*
468			* @see appl.parallel.server.PartitionDataServer#getPartition(int,
469			* java.awt.Rectangle)
470			*/
471			public synchronized DataPartition getPartition(int id, Rectangle bounds)
472			throws RemoteException {
473			DataPartition thisPartition = (DataPartition) data.get(id);
474			if (thisPartition == null && LOG.isDebugEnabled()) {
475			LOG.error("Partition with id " + id + " and bounds " + bounds
476			+ "was requested but not found on DataManager");
477			return null;
478			}
479
480			if (LOG.isDebugEnabled())
481			LOG.debug("Data retrieved from DataManager: ID:" + id
482			+ " partition: " + bounds);
483			return thisPartition.getPartition(bounds);
484			}
485
486			/*
487			* (non-Javadoc)
488			*
489			* @see appl.parallel.server.PartitionDataServer#getPartitionInfo(int)
490			*/
491			public SinglePartitionInfo getPartitionInfo(int rootID)
492			throws RemoteException {
493			return infos.get(rootID);
494			}
495
496			/**
497			* other dataservers may not be initialized on creation of this object.
498			* Therefore the dataServers must be looked up later
499			*/
500			private void initializeDataServers() {
501			if (dataServersInitialized == true)
502			return;
503			for (int i = 0; i < dataServers.length; i++) {
504			dataServers[i] = getConnection(IPs[i]);
505			}
506			dataServersInitialized = true;
507			}
508
509			/*
510			* (non-Javadoc)
511			*
512			* @see appl.parallel.client.DataServer#removeData(int)
513			*/
514			public void removeData(int id) {
515			data.remove(id);
516
517			}
518
519			/**
520			* Removes the partition from the database
521			*
522			* @param name
523			* the name of the partition to remove
524			*/
525			public void removeData(String name) {
526			data.remove(NameToIDMapping.get(name));
527			}
528
529			/**
530			* Removes the partition from the database
531			*
532			* @param partition
533			* the Partition to remove
534			*/
535			public void removePartition(DataPartition partition) {
536			data.remove(partition.getRootID());
537			}
538
539			/**
540			* binds the server to the registry
541			*/
542			private void startTheServer() throws RemoteException {
543			Registry registry = LocateRegistry.getRegistry(registryPort);
544			registry.rebind(bindingName, (PartitionDataServer) this);
545			LOG.info("PartitionDataServer bound to port " + registryPort
546			+ " and running....");
547			}
548
549			/**
550			* unbinds the server from the registry, removes all data and runs the
551			* garbage collector
552			*/
553			public void stop() {
554			data.clear();
555			NameToIDMapping.clear();
556			infos.clear();
557
558			Registry registry;
559			try {
560			registry = LocateRegistry.getRegistry();
561			registry.unbind(bindingName);
562			LOG.debug("Sucessfully removed " + bindingName + " from reggie");
563			} catch (RemoteException e) {
564			LOG.warn("tried to unbind " + bindingName
565			+ " from reggie, but an exception occured: "
566			+ e.getMessage());
567			} catch (NotBoundException e) {
568			LOG.warn("tried to unbind " + bindingName
569			+ " from reggie, but the name was not bound "
570			+ e.getMessage());
571			}
572			System.gc();
573			}
574
575			/**
576			* @param rootID
577			*/
578			public void unloadToSource(int rootID) throws RemoteException {
579			SinglePartitionInfo info = infos.get(rootID);
580			if (info == null) {
581			LOG.error("Could not unload partition with id " + rootID
582			+ ". The partitioninfo was not found!");
583			return;
584			}
585			// start unloading
586			info.getPartitionDataHandler().setBasePartition(this.getData(rootID));
587			info.getPartitionDataHandler().unload();
588			// // remove data from memory and run garbage collection
589			// data.remove(rootID);
590			// System.gc();
591			}
592
593			/*
594			* (non-Javadoc)
595			*
596			* @see appl.parallel.server.PartitionDataServer#updateBaseParameter(java.lang.Object[],
597			* java.lang.String[])
598			*/
599			public void updateBaseParameter(HashMap<String, Object> newParameters)
600			throws RemoteException {
601			// update the mapping. Old values are replaced automaticly
602			baseParameters.putAll(newParameters);
603			}
604
605			/**
606			* Updates the specified partition. For this the the neighborhood partitions
607			* are updated from remote DataServers using the {@link SplitMap} of the
608			* {@link SinglePartitionInfo}.
609			*
610			* @param id
611			*/
612			public void updateFromNeighbors(int id) {
613			initializeDataServers();
614			// get the data to be updated:
615			DataPartition thisPartition = data.get(id);
616			PartitionInfo info = infos.get(id);
617			if (thisPartition == null \|\| info == null) {
618			LOG.error("Update with id " + id
619			+ " failed, because no data was found with this id");
620			return;
621			}
622
623			// get some other needed infos
624			SplitMap map = info.getSplitMap();
625			int mapPosition = info.getSplitMapPos();
626			// check if something to do
627			if (map.getNeighborhoodRange() == 0) {
628			LOG.debug("Update has nothing to do (neighborhoodrange is 0)");
629			return;
630			}
631
632			// calculate (for each neighbor) the needed partitions and get
633			// the updates (implemented using Threads)
634			int[] neighbors = map.getNeighborsForPosition(mapPosition);
635			Future futures[] = new Future[neighbors.length];
636			for (int j = 0; j < neighbors.length; j++) {
637			int neighborIdx = neighbors[j];
638			// get the overlap of the neighborhood area of the current partion
639			// with the calcultation area of the remote partition
640			Rectangle updateBounds = map.getPartitionNeighborhoodBounds(
641			mapPosition).intersection(
642			map.getPartitionCalculationBounds(neighborIdx));
643			// Get the partition from remote server:
644			DataPartition updateData;
645			futures[j] = executor.submit(new UpdateThread(
646			dataServers[neighborIdx], hostnames[neighborIdx],
647			thisPartition, updateBounds, id, remoteEventReceiver));
648			}
649
650			for (Future future : futures) {
651			try {
652			future.get();
653			} catch (InterruptedException e) {
654			e.printStackTrace();
655			} catch (ExecutionException e) {
656			LOG.error("The Remote update of a partition failed", e);
657			}
658			}
659
660			}
661
662			/**
663			* Does the same as {@link #updateFromNeighbors(int)}
664			*
665			* @param name
666			* the name of the resource
667			*/
668			public void updateFromNeighbors(String name) {
669			// get by ID
670			updateFromNeighbors(NameToIDMapping.get(name));
671			}
672
673			/*
674			* (non-Javadoc)
675			*
676			* @see appl.parallel.server.PartitionDataServer#setPartition(int,
677			* appl.parallel.spmd.split.DataPartition, java.awt.Rectangle)
678			*/
679			public synchronized void updatePartition(int id, DataPartition partition,
680			Rectangle bounds) throws RemoteException {
681			DataPartition thisPartition = (DataPartition) data.get(id);
682			if (thisPartition == null && LOG.isDebugEnabled()) {
683			LOG
684			.error("Partition of data with id " + id + " and bounds "
685			+ bounds
686			+ "should be set but was not found in DataManager");
687			return;
688			}
689			thisPartition.setPartition(partition, bounds);
690			if (LOG.isDebugEnabled())
691			LOG.debug("Partition of data set on DataManager with id " + id
692			+ " and partition bounds: " + bounds);
693
694			}
695
696			}