parallel/thread/ExecutionThread.java

package appl.parallel.thread;

import java.rmi.RemoteException;
import java.util.concurrent.Callable;

import appl.parallel.ComputingResourceProperties;
import appl.parallel.event.CommEvent;
import appl.parallel.event.CommEventSink;
import appl.parallel.event.TimeEvent;
import appl.parallel.event.TransferEvent;
import appl.parallel.event.CommEvent.CommType;

/**
 * A simple helper class for more easy Thread handling with anonymous classes.
 *
 * @see DataServerThread
 * @see ComputingResourceThread
 *
 * @author Dominik Appl
 */
public abstract class ExecutionThread implements Callable {

        protected final Object server;

        protected final Object argument;

        protected final CommType commType;

        protected final ComputingResourceProperties serverInfos;

        protected final CommEventSink eventSink;

        protected boolean disableTransferEvents=false;

        /**
         * creates a new thread
         *
         * @param server
         *            the server which is accessed with this thread
         * @param serverInfos the serverinfos (used for event info)
         * @param argument
         *            an argument which can be used inside the anonymous class
         * @param type the type of the communication
         * @param sink events are submitted to this sink
         * @param disableTransferEvents says whether to ignore transfer events (if the transfer volume is to low to be counted))
         */
        public ExecutionThread(Object server,
                        ComputingResourceProperties serverInfos, Object argument,
                        CommType type, CommEventSink sink, boolean disableTransferEvents) {
                this.server = server;
                this.serverInfos = serverInfos;
                this.argument = argument;
                this.commType = type;
                this.eventSink = sink;
                this.disableTransferEvents = disableTransferEvents;
        }

        /**
         * @return the server given the constructor (should be overwrited by a
         *         subclass)
         */
        protected Object getServer() {
                return server;
        }

        /**
         * @return the argument given in the constructor as int value
         */
        protected int getIntArgument() {
                return (Integer) argument;
        }

        /**
         * @return the argument given in the constructor as object value
         */
        protected Object getObjectArgument() {
                return argument;
        }

        /**
         * @return the argument given in the constructor as array value
         */
        protected Object[] getObjectArrayArgument() {
                return (Object[]) argument;
        }

        /**
         * Sends standard {@link CommEvent CommEvents} to the {@link CommEventSink} given with the
         * constructor. Executes the {@link #run()} method and returns its result.
         *
         * @return the result of {@link #run()}.
         *
         * @see java.util.concurrent.Callable#call()
         */
        public Object call() throws Exception {
                long l = System.nanoTime();
                Object result = run();
                if (eventSink.isTimeMonitoringEnabled())
                        fireTimeEvents((System.nanoTime() - l), result);
                if (eventSink.isTransferMonitoringEnabled())
                        fireTransferEvent(result);
                return result;

        }

        /**
         * Sends the time events for this execution. Method generates only the event of
         * this execution. If you want more events (e.g. details from a remote
         * execution) you can overwrite this method. This method is only called if time
         * monitoring is enabled
         *
         * @param execTime
         *            the time of THIS execution
         * @param result the result of the computation (may be of use when overwriting)
         */
        protected void fireTimeEvents(long execTime, Object result) {
                // localhost as String for performance reasons
                try {
                        eventSink.fireRemoteEvent(new TimeEvent(execTime, "localhost",
                                        serverInfos.getName(), commType));
                } catch (RemoteException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                }
        }

        /**
         * The TransferEvents for this execution. Method generates only the event
         * out of the size of {@link #getObjectArgument()}.If you want more events
         * (e.g. details from a remote execution) you can overwrite this method.
         * This method is only called if time monitoring is enabled.
         *  Notice that this will be very slow for large Data!!
         *
         * @param result
         *            the result of the execution (per default not needed)
         */
        protected void fireTransferEvent(Object result) {
                if(!disableTransferEvents)
                // client as String for performance reasons
                try {
                        eventSink.fireRemoteEvent(new TransferEvent("Xulu/V-Client",
                                        serverInfos.getName(), commType, argument));
                } catch (RemoteException e) {
                        e.printStackTrace();
                }

        }

        /**
         * This method should implement the real running code.
         * @return the result you want to retrieve later
         * @throws Exception the exceptions thrown by your code
         */
        protected abstract Object run() throws Exception;

}
1	package appl.parallel.thread;
2
3	import java.rmi.RemoteException;
4	import java.util.concurrent.Callable;
5
6	import appl.parallel.ComputingResourceProperties;
7	import appl.parallel.event.CommEvent;
8	import appl.parallel.event.CommEventSink;
9	import appl.parallel.event.TimeEvent;
10	import appl.parallel.event.TransferEvent;
11	import appl.parallel.event.CommEvent.CommType;
12
13	/**
14	* A simple helper class for more easy Thread handling with anonymous classes.
15	*
16	* @see DataServerThread
17	* @see ComputingResourceThread
18	*
19	* @author Dominik Appl
20	*/
21	public abstract class ExecutionThread implements Callable {
22
23	protected final Object server;
24
25	protected final Object argument;
26
27	protected final CommType commType;
28
29	protected final ComputingResourceProperties serverInfos;
30
31	protected final CommEventSink eventSink;
32
33	protected boolean disableTransferEvents=false;
34
35	/**
36	* creates a new thread
37	*
38	* @param server
39	* the server which is accessed with this thread
40	* @param serverInfos the serverinfos (used for event info)
41	* @param argument
42	* an argument which can be used inside the anonymous class
43	* @param type the type of the communication
44	* @param sink events are submitted to this sink
45	* @param disableTransferEvents says whether to ignore transfer events (if the transfer volume is to low to be counted))
46	*/
47	public ExecutionThread(Object server,
48	ComputingResourceProperties serverInfos, Object argument,
49	CommType type, CommEventSink sink, boolean disableTransferEvents) {
50	this.server = server;
51	this.serverInfos = serverInfos;
52	this.argument = argument;
53	this.commType = type;
54	this.eventSink = sink;
55	this.disableTransferEvents = disableTransferEvents;
56	}
57
58	/**
59	* @return the server given the constructor (should be overwrited by a
60	* subclass)
61	*/
62	protected Object getServer() {
63	return server;
64	}
65
66	/**
67	* @return the argument given in the constructor as int value
68	*/
69	protected int getIntArgument() {
70	return (Integer) argument;
71	}
72
73	/**
74	* @return the argument given in the constructor as object value
75	*/
76	protected Object getObjectArgument() {
77	return argument;
78	}
79
80	/**
81	* @return the argument given in the constructor as array value
82	*/
83	protected Object[] getObjectArrayArgument() {
84	return (Object[]) argument;
85	}
86
87	/**
88	* Sends standard {@link CommEvent CommEvents} to the {@link CommEventSink} given with the
89	* constructor. Executes the {@link #run()} method and returns its result.
90	*
91	* @return the result of {@link #run()}.
92	*
93	* @see java.util.concurrent.Callable#call()
94	*/
95	public Object call() throws Exception {
96	long l = System.nanoTime();
97	Object result = run();
98	if (eventSink.isTimeMonitoringEnabled())
99	fireTimeEvents((System.nanoTime() - l), result);
100	if (eventSink.isTransferMonitoringEnabled())
101	fireTransferEvent(result);
102	return result;
103
104	}
105
106	/**
107	* Sends the time events for this execution. Method generates only the event of
108	* this execution. If you want more events (e.g. details from a remote
109	* execution) you can overwrite this method. This method is only called if time
110	* monitoring is enabled
111	*
112	* @param execTime
113	* the time of THIS execution
114	* @param result the result of the computation (may be of use when overwriting)
115	*/
116	protected void fireTimeEvents(long execTime, Object result) {
117	// localhost as String for performance reasons
118	try {
119	eventSink.fireRemoteEvent(new TimeEvent(execTime, "localhost",
120	serverInfos.getName(), commType));
121	} catch (RemoteException e) {
122	// TODO Auto-generated catch block
123	e.printStackTrace();
124	}
125	}
126
127	/**
128	* The TransferEvents for this execution. Method generates only the event
129	* out of the size of {@link #getObjectArgument()}.If you want more events
130	* (e.g. details from a remote execution) you can overwrite this method.
131	* This method is only called if time monitoring is enabled.
132	* Notice that this will be very slow for large Data!!
133	*
134	* @param result
135	* the result of the execution (per default not needed)
136	*/
137	protected void fireTransferEvent(Object result) {
138	if(!disableTransferEvents)
139	// client as String for performance reasons
140	try {
141	eventSink.fireRemoteEvent(new TransferEvent("Xulu/V-Client",
142	serverInfos.getName(), commType, argument));
143	} catch (RemoteException e) {
144	e.printStackTrace();
145	}
146
147	}
148
149	/**
150	* This method should implement the real running code.
151	* @return the result you want to retrieve later
152	* @throws Exception the exceptions thrown by your code
153	*/
154	protected abstract Object run() throws Exception;
155
156	}