parallel/services/RemoteEventProxy.java

package appl.parallel.services;

import java.rmi.RemoteException;
import java.util.Vector;
import java.util.concurrent.ConcurrentLinkedQueue;

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

import appl.parallel.event.CommEventSink;
import appl.parallel.event.RemoteEvent;
import appl.parallel.event.RemoteEventSink;
import appl.parallel.event.TimeEvent;
import appl.parallel.event.TransferEvent;

/**
 * Delays events and fires them after the delay all <b>at once</b> in one
 * object (this will save communication time)
 * 
 * @author Dominik Appl
 */
public class RemoteEventProxy implements CommEventSink, Service {

        /**
         * This Thread submits the events after the delay
         * 
         * @author Dominik Appl
         */
        private class SubmitThread extends Thread {

                private boolean exit = false;

                private final ConcurrentLinkedQueue<RemoteEvent> queue2;

                private final CommEventSink sink;

                private final long delay;

                SubmitThread(ConcurrentLinkedQueue<RemoteEvent> queue,
                                CommEventSink sink, long delay) {
                        queue2 = queue;
                        this.sink = sink;
                        this.delay = Math.max(100, delay);
                }

                /*
                 * (non-Javadoc)
                 * 
                 * @see java.lang.Thread#run()
                 */
                public void run() {
                        while (!exit) {
                                try {
                                        this.sleep(delay);
                                } catch (InterruptedException e) {
                                        // TODO Auto-generated catch block
                                        e.printStackTrace();
                                }
                                // add all new events to a vector submit them at once
                                Vector<RemoteEvent> events = new Vector<RemoteEvent>();
                                while (queue.peek() != null)
                                        events.add(queue.poll());
                                try {
                                        if (events.size() > 0)
                                                sink.fireRemoteEvents((RemoteEvent[]) events
                                                                .toArray(new RemoteEvent[events.size()]));
                                } catch (RemoteException e) {
                                        // TODO Auto-generated catch block
                                        LOG.warn("Could not transmit events." + e.getMessage(), e);
                                        events.clear();
                                }
                        }
                }

                public void startThread() {
                        exit = false;
                        this.start();
                }

                public void stopThread() {
                        exit = true;
                }
        }

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

        private boolean timeMonitoring = false;

        private boolean transferMonitoring = false;

        private boolean isRunning;

        SubmitThread submitThread;

        ConcurrentLinkedQueue<RemoteEvent> queue;

        private final CommEventSink eventSink;

        private final long delay;

        /**
         * Creates a new proxy
         * 
         * @param targetSink
         *            the sink where the events should be submitted to
         * @param delay
         *            the delay (in milliseconds)
         */
        public RemoteEventProxy(CommEventSink targetSink, long delay) {
                this.eventSink = targetSink;
                this.delay = delay;
                queue = new ConcurrentLinkedQueue<RemoteEvent>();

        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.event.RemoteEventSink#fireRemoteEvent(appl.parallel.event.RemoteEvent)
         */
        public void fireRemoteEvent(RemoteEvent e) throws RemoteException {
                queue.add(e);
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.event.RemoteEventSink#fireRemoteEvents(appl.parallel.event.RemoteEvent[])
         */
        public void fireRemoteEvents(RemoteEvent[] e) throws RemoteException {
                for (RemoteEvent event : e) {
                        fireRemoteEvent(event);
                }
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#isRunning()
         */
        public boolean isRunning() {
                return isRunning;
        }

        /**
         * @return true if the service is running and time monitoring is enabled
         */
        public boolean isTimeMonitoringEnabled() throws RemoteException {
                return timeMonitoring;
        }

        /**
         * @return true if the service is running and tansfer monitoring is enabled
         */
        public boolean isTransferMonitoringEnabled() throws RemoteException {
                return transferMonitoring;

        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#startService()
         */
        public void startService() {
                isRunning = true;
                if (submitThread != null)
                        submitThread.stopThread();
                submitThread = new SubmitThread(queue, eventSink, delay);
                submitThread.startThread();
                try {
                        this.timeMonitoring = eventSink.isTimeMonitoringEnabled();
                        this.transferMonitoring = eventSink.isTransferMonitoringEnabled();
                } catch (RemoteException e) {
                        LOG.error("Could not connect to CommEventSink!!!");
                }

        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.services.Service#stopService()
         */
        public void stopService() {
                isRunning = false;
                if (submitThread != null)
                        submitThread.stopThread();
        }
}
1	package appl.parallel.services;
2
3	import java.rmi.RemoteException;
4	import java.util.Vector;
5	import java.util.concurrent.ConcurrentLinkedQueue;
6
7	import org.apache.log4j.LogManager;
8	import org.apache.log4j.Logger;
9
10	import appl.parallel.event.CommEventSink;
11	import appl.parallel.event.RemoteEvent;
12	import appl.parallel.event.RemoteEventSink;
13	import appl.parallel.event.TimeEvent;
14	import appl.parallel.event.TransferEvent;
15
16	/**
17	* Delays events and fires them after the delay all <b>at once</b> in one
18	* object (this will save communication time)
19	*
20	* @author Dominik Appl
21	*/
22	public class RemoteEventProxy implements CommEventSink, Service {
23
24	/**
25	* This Thread submits the events after the delay
26	*
27	* @author Dominik Appl
28	*/
29	private class SubmitThread extends Thread {
30
31	private boolean exit = false;
32
33	private final ConcurrentLinkedQueue<RemoteEvent> queue2;
34
35	private final CommEventSink sink;
36
37	private final long delay;
38
39	SubmitThread(ConcurrentLinkedQueue<RemoteEvent> queue,
40	CommEventSink sink, long delay) {
41	queue2 = queue;
42	this.sink = sink;
43	this.delay = Math.max(100, delay);
44	}
45
46	/*
47	* (non-Javadoc)
48	*
49	* @see java.lang.Thread#run()
50	*/
51	public void run() {
52	while (!exit) {
53	try {
54	this.sleep(delay);
55	} catch (InterruptedException e) {
56	// TODO Auto-generated catch block
57	e.printStackTrace();
58	}
59	// add all new events to a vector submit them at once
60	Vector<RemoteEvent> events = new Vector<RemoteEvent>();
61	while (queue.peek() != null)
62	events.add(queue.poll());
63	try {
64	if (events.size() > 0)
65	sink.fireRemoteEvents((RemoteEvent[]) events
66	.toArray(new RemoteEvent[events.size()]));
67	} catch (RemoteException e) {
68	// TODO Auto-generated catch block
69	LOG.warn("Could not transmit events." + e.getMessage(), e);
70	events.clear();
71	}
72	}
73	}
74
75	public void startThread() {
76	exit = false;
77	this.start();
78	}
79
80	public void stopThread() {
81	exit = true;
82	}
83	}
84
85	private final Logger LOG = LogManager.getLogger(this.getClass().getName());
86
87	private boolean timeMonitoring = false;
88
89	private boolean transferMonitoring = false;
90
91	private boolean isRunning;
92
93	SubmitThread submitThread;
94
95	ConcurrentLinkedQueue<RemoteEvent> queue;
96
97	private final CommEventSink eventSink;
98
99	private final long delay;
100
101	/**
102	* Creates a new proxy
103	*
104	* @param targetSink
105	* the sink where the events should be submitted to
106	* @param delay
107	* the delay (in milliseconds)
108	*/
109	public RemoteEventProxy(CommEventSink targetSink, long delay) {
110	this.eventSink = targetSink;
111	this.delay = delay;
112	queue = new ConcurrentLinkedQueue<RemoteEvent>();
113
114	}
115
116	/*
117	* (non-Javadoc)
118	*
119	* @see appl.parallel.event.RemoteEventSink#fireRemoteEvent(appl.parallel.event.RemoteEvent)
120	*/
121	public void fireRemoteEvent(RemoteEvent e) throws RemoteException {
122	queue.add(e);
123	}
124
125	/*
126	* (non-Javadoc)
127	*
128	* @see appl.parallel.event.RemoteEventSink#fireRemoteEvents(appl.parallel.event.RemoteEvent[])
129	*/
130	public void fireRemoteEvents(RemoteEvent[] e) throws RemoteException {
131	for (RemoteEvent event : e) {
132	fireRemoteEvent(event);
133	}
134	}
135
136	/*
137	* (non-Javadoc)
138	*
139	* @see appl.parallel.services.Service#isRunning()
140	*/
141	public boolean isRunning() {
142	return isRunning;
143	}
144
145	/**
146	* @return true if the service is running and time monitoring is enabled
147	*/
148	public boolean isTimeMonitoringEnabled() throws RemoteException {
149	return timeMonitoring;
150	}
151
152	/**
153	* @return true if the service is running and tansfer monitoring is enabled
154	*/
155	public boolean isTransferMonitoringEnabled() throws RemoteException {
156	return transferMonitoring;
157
158	}
159
160	/*
161	* (non-Javadoc)
162	*
163	* @see appl.parallel.services.Service#startService()
164	*/
165	public void startService() {
166	isRunning = true;
167	if (submitThread != null)
168	submitThread.stopThread();
169	submitThread = new SubmitThread(queue, eventSink, delay);
170	submitThread.startThread();
171	try {
172	this.timeMonitoring = eventSink.isTimeMonitoringEnabled();
173	this.transferMonitoring = eventSink.isTransferMonitoringEnabled();
174	} catch (RemoteException e) {
175	LOG.error("Could not connect to CommEventSink!!!");
176	}
177
178	}
179
180	/*
181	* (non-Javadoc)
182	*
183	* @see appl.parallel.services.Service#stopService()
184	*/
185	public void stopService() {
186	isRunning = false;
187	if (submitThread != null)
188	submitThread.stopThread();
189	}
190	}