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	mojays	2	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			}