parallel/thread/ExecutionThread.java

package appl.parallel.thread;

import java.nio.channels.Pipe.SinkChannel;
import java.rmi.RemoteException;
import java.util.concurrent.Callable;

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