server/clientinterface/PacketSorter.java

//---PACKAGE DECLARATION---
package uk.ac.ukc.iscream.clientinterface;

//---IMPORTS---
import uk.ac.ukc.iscream.util.*;
import uk.ac.ukc.iscream.componentmanager.*;
import uk.ac.ukc.iscream.core.*;
import java.util.*;

/**
 * Receives data from the incoming CORBA servant, places
 * it in a Queue, and then arranges distribution to the
 * DataHandlers.
 * Has extra functionality to send data to DataHandlers
 * on a per-host basis - ie. the Client can request which
 * hosts it would like to listen for.
 *
 * @author  $Author: tdb1 $
 * @version $Id: PacketSorter.java,v 1.11 2001/03/01 17:29:46 tdb1 Exp $
 */
class PacketSorter extends Thread {

//---FINAL ATTRIBUTES---

    /**
     * The current CVS revision of this class
     */
    public final String REVISION = "$Revision: 1.11 $";
    
//---STATIC METHODS---

//---CONSTRUCTORS---

    /**
     * Creates a new PacketSorter.
     *
     * @param queueMonitorInterval The interval at which to monitor the Queue
     */
    public PacketSorter(int queueMonitorInterval) {
        _queue = new Queue();
        // startup a monitor on this queue, every minute
        String queueName = _name + " PacketSorterQueue";
        _queue.startMonitor(queueMonitorInterval*1000, queueName);
        _hostMap = new HashMap();
        _allHostDataList = new LinkedList();
        _allHostsList = new LinkedList();
        _logger.write(toString(), Logger.SYSINIT, "created");
    }
    
//---PUBLIC METHODS---

    /**
     * Method to start the PacketSorter running. This method will
     * loop forever processing and sending data.
     */
    public void run() {
        int qID = _queue.getQueue();
        while(true) {
            // attempt to get some data from the Queue
            String xml = "";
            try {
                xml = (String) _queue.get(qID);
            }
            catch(InvalidQueueException e) {
                _logger.write(toString(), Logger.ERROR, "Queue failure: "+e);
            }
            
            XMLPacket packet = null;
            
            try {
                XMLPacketMaker xmlPacketMaker = new XMLPacketMaker(xml);
                packet = xmlPacketMaker.createXMLPacket();
            } catch(InvalidXMLException e) {
                _logger.write(toString(), Logger.ERROR, "Invalid XML: "+e);
                // skip the rest of this loop iteration
                continue;
            }
            
            String packetType = packet.getParam("packet.attributes.type");
            // check if we need to send it regardless
            if(packetType.equals("data") || packetType.equals("heartbeat")) {
                String host = packet.getParam("packet.attributes.machine_name");
                
                // look in the hostMap to see if anyone wants this data
                synchronized(this) {
                    if(_hostMap.containsKey(host)) {
                        LinkedList list = (LinkedList) _hostMap.get(host);
                        Iterator i = list.iterator();
                        // push the data to the listening Handler's queue
                        while(i.hasNext()) {
                            ((Queue) i.next()).add(xml);
                        }
                    }
                }
                
                // any handler in this list wants all packets, so send
                // it on to them regardless
                synchronized(this) {
                    Iterator i = _allHostDataList.iterator();
                    while(i.hasNext()) {
                        ((Queue) i.next()).add(xml);
                    }
                }
            }
            else {
                // always send this packet to all hosts, because it's
                // "extra" data, not host data
                synchronized(this) {
                    Iterator i = _allHostsList.iterator();
                    while(i.hasNext()) {
                        ((Queue) i.next()).add(xml);
                    }
                }
            }
        }
    }
    
    /**
     * Register a DataHandler in the system. This method
     * actually takes a reference to a Queue, which should be
     * a Queue that the DataHandler is making use of.
     * It also takes a hostList, this being a semi-colon
     * seperated list of hosts that the Client the DataHandler
     * is serving has requested. If this list is simply an empty
     * String, it is assumed the Client wants to listen to all
     * host information.
     *
     * @param dhQueue a Queue being used by the DataHandler that is registering
     * @param hostList a semi-colon seperated list of hosts
     */
    public void register(Queue dhQueue, String hostList) {
        // check to see if we want all hosts
        if(hostList.equals("")) {
            synchronized(this) {
                _allHostDataList.add(dhQueue);
            }
            _logger.write(toString(), Logger.SYSMSG, "registered DataHandler for all hosts");
        }
        else {
            // go through the list of hosts
            StringTokenizer st = new StringTokenizer(hostList, ";");
            while(st.hasMoreTokens()) {
                String host = st.nextToken();
                synchronized(this) {
                    // see if we already have a list in the map for this host
                    if(_hostMap.containsKey(host)) {
                        // we do, so add to it
                        List list = (List) _hostMap.get(host);
                        list.add(dhQueue);
                    }
                    else {
                        // we don't, so create a list and put it in the map
                        LinkedList list = new LinkedList();
                        list.add(dhQueue);
                        _hostMap.put(host, list);
                    }
                }
            }
            _logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
        }
        // always add host to our complete host list
        synchronized(this) {
            _allHostsList.add(dhQueue);
        }
    }
    
    /**
     * Deregister a DataHandler. The DataHandler should give a reference
     * to the Queue it's using, and the *same* hostList it gave when it
     * register. It is imperative that the hostList is the same, otherwise
     * there will be all sorts of problems with lists getting out of sync.
     *
     * NB: Possible future addition would be recording of hostList's.
     *
     * @param dhQueue a Queue being used by the DataHandler that is deregistering
     * @param hostList a semi-colon seperated list of hosts
     */
    public void deregister(Queue dhQueue, String hostList) {
        // go through the list of hosts
        if(hostList.equals("")) {
            synchronized(this) {
                _allHostDataList.remove(dhQueue);
            }
            _logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for all hosts");
        }
        else {
            StringTokenizer st = new StringTokenizer(hostList, ";");
            while(st.hasMoreTokens()) {
                String host = st.nextToken();
                synchronized(this) {
                    // this should in reality always be true, but best check
                    if(_hostMap.containsKey(host)) {
                        // get the list and remove the host in question
                        LinkedList list = (LinkedList) _hostMap.get(host);
                        list.remove(dhQueue);
                        // if the list is now empty, we might as well remove it
                        if(list.size()==0) {
                            _hostMap.remove(host);
                        }
                    }
                }
            }
            _logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for hosts: "+hostList);
        }
        // always remove host from our complete host list
        synchronized(this) {
            _allHostsList.remove(dhQueue);
        }
    }
    
    /**
     * Overrides the {@link java.lang.Object#toString() Object.toString()}
     * method to provide clean logging (every class should have this).
     *
     * This uses the uk.ac.ukc.iscream.util.NameFormat class
     * to format the toString()
     *
     * @return the name of this class and its CVS revision
     */
    public String toString() {
        return FormatName.getName(
            _name,
            getClass().getName(),
            REVISION);
    }

//---PRIVATE METHODS---

//---ACCESSOR/MUTATOR METHODS---
    
    /**
     * Accessor to return a reference to the Queue object. This
     * is needed so the ClientInterfaceServant can get add data
     * easily.
     *
     * @return a reference to our Queue object.
     */
    public Queue getQueue() {
        return _queue;
    }

//---ATTRIBUTES---

    /**
     * This is the friendly identifier of the
     * component this class is running in.
     * eg, a Filter may be called "filter1",
     * If this class does not have an owning
     * component,  a name from the configuration
     * can be placed here.  This name could also
     * be changed to null for utility classes.
     */
    private String _name = ClientInterfaceMain.NAME;

    /**
     * This holds a reference to the
     * system logger that is being used.
     */
    private Logger _logger = ReferenceManager.getInstance().getLogger();
    
    /**
     * A reference to the Queue we're using.
     */
    private Queue _queue;
    
    /**
     * A HashMap to store lists of Queue's (in the DataHandlers)
     * in a way that can be easily accessed when data comes in.
     */
    private HashMap _hostMap;
    
    /**
     * A list specifically for a Queue's associated with DataHandlers
     * that want all host information.
     */
    private LinkedList _allHostDataList;
    
    /**
     * A list of all hosts.
     */
    private LinkedList _allHostsList;
    
//---STATIC ATTRIBUTES---

}
Revision:	1.12
Committed:	Thu Mar 1 23:57:09 2001 UTC (23 years, 3 months ago) by tdb
Branch:	MAIN
Changes since 1.11:	+39 -27 lines
Log Message:	Further attempts to sort this class out. This has gone very much to the extreme now. Every access to the lists and hostmap is synchronized on this object. This means that only one of the bits may be executed at once. Whilst this may clearly solve some problems, it does open this class to becoming a major bottleneck, and could possibly lead to deadlock externally (although I think our Queue's avoid this issue).
#	Content
1	//---PACKAGE DECLARATION---
2	package uk.ac.ukc.iscream.clientinterface;
3
4	//---IMPORTS---
5	import uk.ac.ukc.iscream.util.*;
6	import uk.ac.ukc.iscream.componentmanager.*;
7	import uk.ac.ukc.iscream.core.*;
8	import java.util.*;
9
10	/**
11	* Receives data from the incoming CORBA servant, places
12	* it in a Queue, and then arranges distribution to the
13	* DataHandlers.
14	* Has extra functionality to send data to DataHandlers
15	* on a per-host basis - ie. the Client can request which
16	* hosts it would like to listen for.
17	*
18	* @author $Author: tdb1 $
19	* @version $Id: PacketSorter.java,v 1.11 2001/03/01 17:29:46 tdb1 Exp $
20	*/
21	class PacketSorter extends Thread {
22
23	//---FINAL ATTRIBUTES---
24
25	/**
26	* The current CVS revision of this class
27	*/
28	public final String REVISION = "$Revision: 1.11 $";
29
30	//---STATIC METHODS---
31
32	//---CONSTRUCTORS---
33
34	/**
35	* Creates a new PacketSorter.
36	*
37	* @param queueMonitorInterval The interval at which to monitor the Queue
38	*/
39	public PacketSorter(int queueMonitorInterval) {
40	_queue = new Queue();
41	// startup a monitor on this queue, every minute
42	String queueName = _name + " PacketSorterQueue";
43	_queue.startMonitor(queueMonitorInterval*1000, queueName);
44	_hostMap = new HashMap();
45	_allHostDataList = new LinkedList();
46	_allHostsList = new LinkedList();
47	_logger.write(toString(), Logger.SYSINIT, "created");
48	}
49
50	//---PUBLIC METHODS---
51
52	/**
53	* Method to start the PacketSorter running. This method will
54	* loop forever processing and sending data.
55	*/
56	public void run() {
57	int qID = _queue.getQueue();
58	while(true) {
59	// attempt to get some data from the Queue
60	String xml = "";
61	try {
62	xml = (String) _queue.get(qID);
63	}
64	catch(InvalidQueueException e) {
65	_logger.write(toString(), Logger.ERROR, "Queue failure: "+e);
66	}
67
68	XMLPacket packet = null;
69
70	try {
71	XMLPacketMaker xmlPacketMaker = new XMLPacketMaker(xml);
72	packet = xmlPacketMaker.createXMLPacket();
73	} catch(InvalidXMLException e) {
74	_logger.write(toString(), Logger.ERROR, "Invalid XML: "+e);
75	// skip the rest of this loop iteration
76	continue;
77	}
78
79	String packetType = packet.getParam("packet.attributes.type");
80	// check if we need to send it regardless
81	if(packetType.equals("data") \|\| packetType.equals("heartbeat")) {
82	String host = packet.getParam("packet.attributes.machine_name");
83
84	// look in the hostMap to see if anyone wants this data
85	synchronized(this) {
86	if(_hostMap.containsKey(host)) {
87	LinkedList list = (LinkedList) _hostMap.get(host);
88	Iterator i = list.iterator();
89	// push the data to the listening Handler's queue
90	while(i.hasNext()) {
91	((Queue) i.next()).add(xml);
92	}
93	}
94	}
95
96	// any handler in this list wants all packets, so send
97	// it on to them regardless
98	synchronized(this) {
99	Iterator i = _allHostDataList.iterator();
100	while(i.hasNext()) {
101	((Queue) i.next()).add(xml);
102	}
103	}
104	}
105	else {
106	// always send this packet to all hosts, because it's
107	// "extra" data, not host data
108	synchronized(this) {
109	Iterator i = _allHostsList.iterator();
110	while(i.hasNext()) {
111	((Queue) i.next()).add(xml);
112	}
113	}
114	}
115	}
116	}
117
118	/**
119	* Register a DataHandler in the system. This method
120	* actually takes a reference to a Queue, which should be
121	* a Queue that the DataHandler is making use of.
122	* It also takes a hostList, this being a semi-colon
123	* seperated list of hosts that the Client the DataHandler
124	* is serving has requested. If this list is simply an empty
125	* String, it is assumed the Client wants to listen to all
126	* host information.
127	*
128	* @param dhQueue a Queue being used by the DataHandler that is registering
129	* @param hostList a semi-colon seperated list of hosts
130	*/
131	public void register(Queue dhQueue, String hostList) {
132	// check to see if we want all hosts
133	if(hostList.equals("")) {
134	synchronized(this) {
135	_allHostDataList.add(dhQueue);
136	}
137	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for all hosts");
138	}
139	else {
140	// go through the list of hosts
141	StringTokenizer st = new StringTokenizer(hostList, ";");
142	while(st.hasMoreTokens()) {
143	String host = st.nextToken();
144	synchronized(this) {
145	// see if we already have a list in the map for this host
146	if(_hostMap.containsKey(host)) {
147	// we do, so add to it
148	List list = (List) _hostMap.get(host);
149	list.add(dhQueue);
150	}
151	else {
152	// we don't, so create a list and put it in the map
153	LinkedList list = new LinkedList();
154	list.add(dhQueue);
155	_hostMap.put(host, list);
156	}
157	}
158	}
159	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
160	}
161	// always add host to our complete host list
162	synchronized(this) {
163	_allHostsList.add(dhQueue);
164	}
165	}
166
167	/**
168	* Deregister a DataHandler. The DataHandler should give a reference
169	* to the Queue it's using, and the same hostList it gave when it
170	* register. It is imperative that the hostList is the same, otherwise
171	* there will be all sorts of problems with lists getting out of sync.
172	*
173	* NB: Possible future addition would be recording of hostList's.
174	*
175	* @param dhQueue a Queue being used by the DataHandler that is deregistering
176	* @param hostList a semi-colon seperated list of hosts
177	*/
178	public void deregister(Queue dhQueue, String hostList) {
179	// go through the list of hosts
180	if(hostList.equals("")) {
181	synchronized(this) {
182	_allHostDataList.remove(dhQueue);
183	}
184	_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for all hosts");
185	}
186	else {
187	StringTokenizer st = new StringTokenizer(hostList, ";");
188	while(st.hasMoreTokens()) {
189	String host = st.nextToken();
190	synchronized(this) {
191	// this should in reality always be true, but best check
192	if(_hostMap.containsKey(host)) {
193	// get the list and remove the host in question
194	LinkedList list = (LinkedList) _hostMap.get(host);
195	list.remove(dhQueue);
196	// if the list is now empty, we might as well remove it
197	if(list.size()==0) {
198	_hostMap.remove(host);
199	}
200	}
201	}
202	}
203	_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for hosts: "+hostList);
204	}
205	// always remove host from our complete host list
206	synchronized(this) {
207	_allHostsList.remove(dhQueue);
208	}
209	}
210
211	/**
212	* Overrides the {@link java.lang.Object#toString() Object.toString()}
213	* method to provide clean logging (every class should have this).
214	*
215	* This uses the uk.ac.ukc.iscream.util.NameFormat class
216	* to format the toString()
217	*
218	* @return the name of this class and its CVS revision
219	*/
220	public String toString() {
221	return FormatName.getName(
222	_name,
223	getClass().getName(),
224	REVISION);
225	}
226
227	//---PRIVATE METHODS---
228
229	//---ACCESSOR/MUTATOR METHODS---
230
231	/**
232	* Accessor to return a reference to the Queue object. This
233	* is needed so the ClientInterfaceServant can get add data
234	* easily.
235	*
236	* @return a reference to our Queue object.
237	*/
238	public Queue getQueue() {
239	return _queue;
240	}
241
242	//---ATTRIBUTES---
243
244	/**
245	* This is the friendly identifier of the
246	* component this class is running in.
247	* eg, a Filter may be called "filter1",
248	* If this class does not have an owning
249	* component, a name from the configuration
250	* can be placed here. This name could also
251	* be changed to null for utility classes.
252	*/
253	private String _name = ClientInterfaceMain.NAME;
254
255	/**
256	* This holds a reference to the
257	* system logger that is being used.
258	*/
259	private Logger _logger = ReferenceManager.getInstance().getLogger();
260
261	/**
262	* A reference to the Queue we're using.
263	*/
264	private Queue _queue;
265
266	/**
267	* A HashMap to store lists of Queue's (in the DataHandlers)
268	* in a way that can be easily accessed when data comes in.
269	*/
270	private HashMap _hostMap;
271
272	/**
273	* A list specifically for a Queue's associated with DataHandlers
274	* that want all host information.
275	*/
276	private LinkedList _allHostDataList;
277
278	/**
279	* A list of all hosts.
280	*/
281	private LinkedList _allHostsList;
282
283	//---STATIC ATTRIBUTES---
284
285	}