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).
#	User	Rev	Content
1	tdb	1.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	tdb	1.4	* 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	tdb	1.1	*
18	tdb	1.2	* @author $Author: tdb1 $
19	tdb	1.12	* @version $Id: PacketSorter.java,v 1.11 2001/03/01 17:29:46 tdb1 Exp $
20	tdb	1.1	*/
21			class PacketSorter extends Thread {
22
23			//---FINAL ATTRIBUTES---
24
25			/**
26			* The current CVS revision of this class
27			*/
28	tdb	1.12	public final String REVISION = "$Revision: 1.11 $";
29	tdb	1.1
30			//---STATIC METHODS---
31
32			//---CONSTRUCTORS---
33
34			/**
35			* Creates a new PacketSorter.
36	tdb	1.6	*
37			* @param queueMonitorInterval The interval at which to monitor the Queue
38	tdb	1.1	*/
39	tdb	1.6	public PacketSorter(int queueMonitorInterval) {
40	tdb	1.1	_queue = new Queue();
41	tdb	1.5	// startup a monitor on this queue, every minute
42	tdb	1.6	String queueName = _name + " PacketSorterQueue";
43			_queue.startMonitor(queueMonitorInterval*1000, queueName);
44	tdb	1.11	_hostMap = new HashMap();
45			_allHostDataList = new LinkedList();
46			_allHostsList = new LinkedList();
47	tdb	1.1	_logger.write(toString(), Logger.SYSINIT, "created");
48			}
49
50			//---PUBLIC METHODS---
51
52	tdb	1.4	/**
53			* Method to start the PacketSorter running. This method will
54			* loop forever processing and sending data.
55			*/
56	tdb	1.1	public void run() {
57			int qID = _queue.getQueue();
58			while(true) {
59	tdb	1.4	// attempt to get some data from the Queue
60	tdb	1.1	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	tdb	1.10	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	tdb	1.1
79	tdb	1.8	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	tdb	1.11	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	tdb	1.8	}
94			}
95
96			// any handler in this list wants all packets, so send
97			// it on to them regardless
98	tdb	1.11	synchronized(this) {
99			Iterator i = _allHostDataList.iterator();
100			while(i.hasNext()) {
101			((Queue) i.next()).add(xml);
102			}
103			}
104	tdb	1.1	}
105	tdb	1.8	else {
106	tdb	1.10	// always send this packet to all hosts, because it's
107			// "extra" data, not host data
108	tdb	1.11	synchronized(this) {
109			Iterator i = _allHostsList.iterator();
110			while(i.hasNext()) {
111			((Queue) i.next()).add(xml);
112			}
113			}
114	tdb	1.2	}
115	tdb	1.1	}
116			}
117
118	tdb	1.4	/**
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	tdb	1.12	public void register(Queue dhQueue, String hostList) {
132	tdb	1.4	// check to see if we want all hosts
133	tdb	1.2	if(hostList.equals("")) {
134	tdb	1.12	synchronized(this) {
135			_allHostDataList.add(dhQueue);
136			}
137	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for all hosts");
138			}
139			else {
140	tdb	1.4	// go through the list of hosts
141	tdb	1.2	StringTokenizer st = new StringTokenizer(hostList, ";");
142			while(st.hasMoreTokens()) {
143			String host = st.nextToken();
144	tdb	1.12	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	tdb	1.2	}
158	tdb	1.1	}
159	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "registered DataHandler for hosts: "+hostList);
160	tdb	1.1	}
161	tdb	1.10	// always add host to our complete host list
162	tdb	1.12	synchronized(this) {
163			_allHostsList.add(dhQueue);
164			}
165	tdb	1.1	}
166
167	tdb	1.4	/**
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	tdb	1.12	public void deregister(Queue dhQueue, String hostList) {
179	tdb	1.4	// go through the list of hosts
180	tdb	1.2	if(hostList.equals("")) {
181	tdb	1.12	synchronized(this) {
182			_allHostDataList.remove(dhQueue);
183			}
184	tdb	1.2	_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	tdb	1.12	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	tdb	1.2	}
201	tdb	1.1	}
202			}
203	tdb	1.2	_logger.write(toString(), Logger.SYSMSG, "deregistered DataHandler for hosts: "+hostList);
204	tdb	1.1	}
205	tdb	1.10	// always remove host from our complete host list
206	tdb	1.12	synchronized(this) {
207			_allHostsList.remove(dhQueue);
208			}
209	tdb	1.1	}
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	tdb	1.4
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	tdb	1.1	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	tdb	1.4	/**
262			* A reference to the Queue we're using.
263			*/
264	tdb	1.1	private Queue _queue;
265
266	tdb	1.4	/**
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	tdb	1.11	private HashMap _hostMap;
271	tdb	1.4
272			/**
273			* A list specifically for a Queue's associated with DataHandlers
274			* that want all host information.
275	tdb	1.10	*/
276	tdb	1.11	private LinkedList _allHostDataList;
277	tdb	1.10
278			/**
279			* A list of all hosts.
280	tdb	1.4	*/
281	tdb	1.11	private LinkedList _allHostsList;
282	tdb	1.1
283			//---STATIC ATTRIBUTES---
284
285			}