parallel/client/ClientDataServer.java

package appl.parallel.client;

import java.awt.Rectangle;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.rmi.RemoteException;
import java.rmi.server.ServerNotActiveException;
import java.rmi.server.UnicastRemoteObject;
import java.util.HashMap;
import java.util.Hashtable;

import javax.activation.DataHandler;

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

import appl.ext.XuluConfig;
import appl.parallel.event.CommEventSink;
import appl.parallel.event.TransferEvent;
import appl.parallel.event.CommEvent.CommType;
import appl.parallel.server.PartitionDataServer;
import appl.parallel.server.XuluServer;
import appl.parallel.spmd.split.DataPartition;
import appl.parallel.spmd.split.SplittableResource;
import appl.parallel.util.Helper;

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