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.parallel.client.ClientDataServer;
import appl.parallel.client.DataServer;
import appl.parallel.data.PartitionDataHandler;
import appl.parallel.event.CommEvent.CommType;
import appl.parallel.event.CommEventSink;
import appl.parallel.event.TimeEvent;
import appl.parallel.event.TransferEvent;
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;
import de.appl.data.LoadingException;

/**
 * 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	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.parallel.client.ClientDataServer;
24	import appl.parallel.client.DataServer;
25	import appl.parallel.data.PartitionDataHandler;
26	import appl.parallel.event.CommEvent.CommType;
27	import appl.parallel.event.CommEventSink;
28	import appl.parallel.event.TimeEvent;
29	import appl.parallel.event.TransferEvent;
30	import appl.parallel.services.RemoteEventProxy;
31	import appl.parallel.spmd.MultiDataInfo;
32	import appl.parallel.spmd.MultiDataPartitionObject;
33	import appl.parallel.spmd.SPMDClientController;
34	import appl.parallel.spmd.split.DataPartition;
35	import appl.parallel.spmd.split.PartitionInfo;
36	import appl.parallel.spmd.split.SinglePartitionInfo;
37	import appl.parallel.spmd.split.SplitMap;
38	import appl.parallel.thread.DataServerThread;
39	import de.appl.data.LoadingException;
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	}