parallel/client/ClientDataServer.java

package appl.parallel.client;

import java.awt.Rectangle;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.rmi.AccessException;
import java.rmi.NotBoundException;
import java.rmi.Remote;
import java.rmi.RemoteException;
import java.rmi.registry.LocateRegistry;
import java.rmi.registry.Registry;
import java.rmi.server.ServerNotActiveException;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashMap;
import java.util.Hashtable;

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

import appl.ext.XuluConfig;
import appl.parallel.event.CommEvent;
import appl.parallel.event.CommEventSink;
import appl.parallel.event.TransferEvent;
import appl.parallel.event.CommEvent.CommType;
import appl.parallel.spmd.split.DataPartition;
import appl.parallel.util.Helper;
// fuer Doku
import javax.activation.DataHandler;
import appl.parallel.server.PartitionDataServer;
import appl.parallel.server.XuluServer;
import appl.parallel.spmd.split.SplittableResource;

/**
 * This class is used for storing all data which is needed for parallelization
 * and may be requested remotely. On {@link DataPartition DataPartitions} the
 * request of partitions is supported. {@link PartitionDataServer}s running on
 * {@link XuluServer}s will typically use {@link DataHandler DataHandlers} for
 * retrieving the partitions.
 *
 * @see PartitionDataServer
 * @author Dominik Appl
 */
public class ClientDataServer extends UnicastRemoteObject implements DataServer {

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

        private final String bindingName = "DataServer";

        // for retrieval with the baseID
        private Hashtable<Integer, DataPartition> dataByID;

        HashMap<String, String> hostnames = new HashMap<String, String>(10);

        private int registryPort = 1099;

        private final CommEventSink eventSink;

        /**
         * The standard Constructor (binds itself to the running registry)
         *
         * @param eventSink
         *            the sink is used for generating events
         * @throws RemoteException
         *             if the connection fails
         */
        public ClientDataServer(CommEventSink eventSink) throws RemoteException {
                this.eventSink = eventSink;
                dataByID = new Hashtable<Integer, DataPartition>();

                // bind to the registry - create a new reg, if no reg is found
                try {
                        // try to get port from registry
                        int port = XuluConfig.getXuluConfig().getIntProperty(
                                        "XuluClient.registryport");
                        if (port != 0)
                                registryPort = port;
                        Helper.bind(bindingName, this, registryPort);
                } catch (Exception e) {
                        LOG.error("Error while binding SPMDClient to the registry."
                                        + e.getMessage());
                }
                LOG.info("DataServer up and running....");
        }

        /**
         * Returns the data identified by the given id
         *
         * @param baseID
         *            Id identifiying the data
         * @see SplittableResource#getRootID()
         */
        public DataPartition getData(int baseID) throws RemoteException {
                DataPartition result = dataByID.get(baseID);
                if (LOG.isDebugEnabled())
                        LOG.debug("Data retrieved from SPMDClient: ID:" + baseID);

                // generate transfer event. Notice that this is very slow and
                // will therefore only be created if a listener has registered
                if (eventSink.isTransferMonitoringEnabled()) {
                        try {
                                // try to get Hostname (for performance reasons this is done
                                // only one time)
                                String hostIP = getClientHost();
                                // check if already looked up;
                                String hostname = hostnames.get(hostIP);
                                if (hostname == null) {
                                        hostname = InetAddress.getByName(hostIP).getHostName();
                                        hostnames.put(hostIP, hostname);
                                }
                                eventSink.fireRemoteEvent(new TransferEvent("DataServer",
                                                hostname, CommType.TRANSFER_DATA, result));
                        } catch (ServerNotActiveException e) {
                                // if this is a purely local call no event
                                // should be generated.
                        } catch (UnknownHostException e) {
                                // TODO Auto-generated catch block
                                e.printStackTrace();
                        }
                }
                return result;
        }

        /**
         * DataPartitions can be added
         *
         * @see appl.parallel.client.DataServer#addData(appl.parallel.spmd.split.DataPartition)
         */
        public void addData(DataPartition splittable) {
                this.dataByID.put(splittable.getRootID(), splittable);
        }

        /**
         * stops the client and removes it from the registry
         */
        public void close() {
                Helper.unbind(bindingName);
        }

        /*
         * (non-Javadoc)
         *
         * @see appl.parallel.server.PartitionDataServer#getPartition(int,
         *      java.awt.Rectangle)
         */
        public DataPartition getPartition(int id, Rectangle bounds)
                        throws RemoteException {
                DataPartition data = (DataPartition) dataByID.get(id);
                if (data == null && LOG.isDebugEnabled()) {
                        LOG.error("Partition with id " + id + " and bounds " + bounds
                                        + "was requested but not found on SPMDClient");
                        return null;
                }
                DataPartition result = data.getPartition(bounds);
                if (eventSink.isTransferMonitoringEnabled()) {
                        try {
                                eventSink.fireRemoteEvent(new TransferEvent("DataServer",
                                                getClientHost(), CommType.TRANSFER_DATA, result));
                        } catch (ServerNotActiveException e) {
                                // TODO Auto-generated catch block
                                e.printStackTrace();
                        }
                }
                if (LOG.isDebugEnabled())
                        LOG.debug("Data retrieved from SPMDClient: ID:" + id
                                        + " partition: " + bounds);
                return result;
        }

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

        }

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

}
1	package appl.parallel.client;
2
3	import java.awt.Rectangle;
4	import java.net.InetAddress;
5	import java.net.UnknownHostException;
6	import java.rmi.AccessException;
7	import java.rmi.NotBoundException;
8	import java.rmi.Remote;
9	import java.rmi.RemoteException;
10	import java.rmi.registry.LocateRegistry;
11	import java.rmi.registry.Registry;
12	import java.rmi.server.ServerNotActiveException;
13	import java.rmi.server.UnicastRemoteObject;
14	import java.util.HashMap;
15	import java.util.Hashtable;
16
17	import org.apache.log4j.LogManager;
18	import org.apache.log4j.Logger;
19
20	import appl.ext.XuluConfig;
21	import appl.parallel.event.CommEvent;
22	import appl.parallel.event.CommEventSink;
23	import appl.parallel.event.TransferEvent;
24	import appl.parallel.event.CommEvent.CommType;
25	import appl.parallel.spmd.split.DataPartition;
26	import appl.parallel.util.Helper;
27	// fuer Doku
28	import javax.activation.DataHandler;
29	import appl.parallel.server.PartitionDataServer;
30	import appl.parallel.server.XuluServer;
31	import appl.parallel.spmd.split.SplittableResource;
32
33	/**
34	* This class is used for storing all data which is needed for parallelization
35	* and may be requested remotely. On {@link DataPartition DataPartitions} the
36	* request of partitions is supported. {@link PartitionDataServer}s running on
37	* {@link XuluServer}s will typically use {@link DataHandler DataHandlers} for
38	* retrieving the partitions.
39	*
40	* @see PartitionDataServer
41	* @author Dominik Appl
42	*/
43	public class ClientDataServer extends UnicastRemoteObject implements DataServer {
44
45	protected final Logger LOG = LogManager
46	.getLogger(this.getClass().getName());
47
48	private final String bindingName = "DataServer";
49
50	// for retrieval with the baseID
51	private Hashtable<Integer, DataPartition> dataByID;
52
53	HashMap<String, String> hostnames = new HashMap<String, String>(10);
54
55	private int registryPort = 1099;
56
57	private final CommEventSink eventSink;
58
59	/**
60	* The standard Constructor (binds itself to the running registry)
61	*
62	* @param eventSink
63	* the sink is used for generating events
64	* @throws RemoteException
65	* if the connection fails
66	*/
67	public ClientDataServer(CommEventSink eventSink) throws RemoteException {
68	this.eventSink = eventSink;
69	dataByID = new Hashtable<Integer, DataPartition>();
70
71	// bind to the registry - create a new reg, if no reg is found
72	try {
73	// try to get port from registry
74	int port = XuluConfig.getXuluConfig().getIntProperty(
75	"XuluClient.registryport");
76	if (port != 0)
77	registryPort = port;
78	Helper.bind(bindingName, this, registryPort);
79	} catch (Exception e) {
80	LOG.error("Error while binding SPMDClient to the registry."
81	+ e.getMessage());
82	}
83	LOG.info("DataServer up and running....");
84	}
85
86	/**
87	* Returns the data identified by the given id
88	*
89	* @param baseID
90	* Id identifiying the data
91	* @see SplittableResource#getRootID()
92	*/
93	public DataPartition getData(int baseID) throws RemoteException {
94	DataPartition result = dataByID.get(baseID);
95	if (LOG.isDebugEnabled())
96	LOG.debug("Data retrieved from SPMDClient: ID:" + baseID);
97
98	// generate transfer event. Notice that this is very slow and
99	// will therefore only be created if a listener has registered
100	if (eventSink.isTransferMonitoringEnabled()) {
101	try {
102	// try to get Hostname (for performance reasons this is done
103	// only one time)
104	String hostIP = getClientHost();
105	// check if already looked up;
106	String hostname = hostnames.get(hostIP);
107	if (hostname == null) {
108	hostname = InetAddress.getByName(hostIP).getHostName();
109	hostnames.put(hostIP, hostname);
110	}
111	eventSink.fireRemoteEvent(new TransferEvent("DataServer",
112	hostname, CommType.TRANSFER_DATA, result));
113	} catch (ServerNotActiveException e) {
114	// if this is a purely local call no event
115	// should be generated.
116	} catch (UnknownHostException e) {
117	// TODO Auto-generated catch block
118	e.printStackTrace();
119	}
120	}
121	return result;
122	}
123
124	/**
125	* DataPartitions can be added
126	*
127	* @see appl.parallel.client.DataServer#addData(appl.parallel.spmd.split.DataPartition)
128	*/
129	public void addData(DataPartition splittable) {
130	this.dataByID.put(splittable.getRootID(), splittable);
131	}
132
133	/**
134	* stops the client and removes it from the registry
135	*/
136	public void close() {
137	Helper.unbind(bindingName);
138	}
139
140	/*
141	* (non-Javadoc)
142	*
143	* @see appl.parallel.server.PartitionDataServer#getPartition(int,
144	* java.awt.Rectangle)
145	*/
146	public DataPartition getPartition(int id, Rectangle bounds)
147	throws RemoteException {
148	DataPartition data = (DataPartition) dataByID.get(id);
149	if (data == null && LOG.isDebugEnabled()) {
150	LOG.error("Partition with id " + id + " and bounds " + bounds
151	+ "was requested but not found on SPMDClient");
152	return null;
153	}
154	DataPartition result = data.getPartition(bounds);
155	if (eventSink.isTransferMonitoringEnabled()) {
156	try {
157	eventSink.fireRemoteEvent(new TransferEvent("DataServer",
158	getClientHost(), CommType.TRANSFER_DATA, result));
159	} catch (ServerNotActiveException e) {
160	// TODO Auto-generated catch block
161	e.printStackTrace();
162	}
163	}
164	if (LOG.isDebugEnabled())
165	LOG.debug("Data retrieved from SPMDClient: ID:" + id
166	+ " partition: " + bounds);
167	return result;
168	}
169
170	/*
171	* (non-Javadoc)
172	*
173	* @see appl.parallel.server.PartitionDataServer#setPartition(int,
174	* appl.parallel.spmd.split.DataPartition, java.awt.Rectangle)
175	*/
176	public void updatePartition(int id, DataPartition partition,
177	Rectangle bounds) throws RemoteException {
178	DataPartition data = (DataPartition) dataByID.get(id);
179	if (data == null && LOG.isDebugEnabled()) {
180	LOG.error("Partition of data with id " + id + " and bounds "
181	+ bounds + "should be set but was not found on SPMDClient");
182	return;
183	}
184	data.setPartition(partition, bounds);
185	if (LOG.isDebugEnabled())
186	LOG.debug("Partition set on SPMDClient with id " + id
187	+ " and partition bounds: " + bounds);
188
189	}
190
191	/*
192	* (non-Javadoc)
193	*
194	* @see appl.parallel.client.DataServer#removeData(int)
195	*/
196	public void removeData(int id) {
197	dataByID.remove(id);
198	}
199
200	}