parallel/spmd/AdvancedSPMDClientController.java

package appl.parallel.spmd;

import java.io.PrintStream;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Vector;

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

import appl.parallel.ComputingResourceContainer;
import appl.parallel.client.ClientDataServer;
import appl.parallel.event.CommEventSink;
import appl.parallel.thread.OneMethodThread;

/**
 * Performance optimizations on client side can be made with this class. Allows
 * heavy multithreading.<br>
 * <br>
 * Merging can be done in a separate thread (parallel to further computation)<br>
 * 
 * @see SPMDClientController for details on the usage of a client controller
 * 
 * @author Dominik Appl
 */
public class AdvancedSPMDClientController extends SPMDClientController
                implements AdvancedSPMDClientInterface {

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

        // used for synchronization
        private static final Object syncOperationMontitor = new Object();

        /**
         * For each {@link SyncPoint} a vector of threads associated with the point
         * is created
         */
        HashMap<SyncPoint, Vector<OneMethodThread>> vectorMap = new HashMap<SyncPoint, Vector<OneMethodThread>>();

        /**
         * Empty thread which is used as indicator for a already synchronized sync
         * point
         */
        private static final Vector<OneMethodThread> alreadySynchronized = new Vector<OneMethodThread>(
                        1);

        /**
         * same parameters as superclass
         * 
         * @see SPMDClientController#SPMDClientController(Vector, double[],
         *      ClientDataServer, CommEventSink)
         */
        public AdvancedSPMDClientController(
                        Vector<ComputingResourceContainer> computingResources,
                        double[] weights, ClientDataServer spmdClient,
                        CommEventSink eventProxy) {
                super(computingResources, weights, spmdClient, eventProxy);
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergePartition(java.lang.Object,
         *      appl.parallel.spmd.SyncPoint)
         */
        public void mergePartition(final Object partition, SyncPoint s,
                        final PrintStream stream, final String message) {
                checkSplittable(partition);
                // create a new thread
                OneMethodThread newThread = new OneMethodThread("mergeThread"
                                + s.getId(), s.getPriority()) {
                        @Override
                        public void run() {
                                mergePartition(partition);
                                if (stream != null) {
                                        stream.println(message);
                                }
                        }
                };
                mapThreadToSyncPoint(newThread, s);
                newThread.start();
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergePartition(java.lang.Object,
         *      appl.parallel.spmd.SyncPoint)
         */
        public synchronized void mergePartition(Object partition, SyncPoint s) {
                mergePartition(partition, s, null, null);
        }

        /**
         * Puts a Thread/Syncpoint pair into the Hashmap/Vector structure
         */
        private void mapThreadToSyncPoint(OneMethodThread thread, SyncPoint s) {
                // access to the syncpoints must be synchronized when working with
                // multithreading
                synchronized (syncOperationMontitor) {
                        Vector<OneMethodThread> threads = vectorMap.get(s);
                        // if this is the first thread for the syncPoint: create a new
                        // Vector
                        if (threads == null) {
                                threads = new Vector<OneMethodThread>(5);
                                vectorMap.put(s, threads);
                        }
                        threads.add(thread);
                }
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergeMultiData(appl.parallel.spmd.MultiDataObject,
         *      int, appl.parallel.spmd.SyncPoint, java.io.PrintStream,
         *      java.lang.String)
         */
        public synchronized void mergeMultiData(MultiDataObject multidata, int idx,
                        SyncPoint s, PrintStream stream, String message) {
                // create a new thread
                OneMethodThread newThread = new OneMethodThread("mergeThread-"
                                + s.getId(), s.getPriority(), multidata, idx, stream, message) {
                        @Override
                        public void run() {
                                mergeMultiData((MultiDataObject) getParameter(0),
                                                (Integer) getParameter(1));
                                if (getParameter(2) != null) {
                                        PrintStream stream = (PrintStream) getParameter(2);
                                        stream.println(getParameter(3));
                                }
                        }
                };
                mapThreadToSyncPoint(newThread, s);
                newThread.start();
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergeMultiData(appl.parallel.spmd.MultiDataObject,
         *      int, appl.parallel.spmd.SyncPoint)
         */
        public void mergeMultiData(MultiDataObject multidata, int idx, SyncPoint s) {
                mergeMultiData(multidata, idx, s, null, null);
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.spmd.AdvancedSPMDClientInterface#synchronizeToSyncPoint(appl.parallel.spmd.SyncPoint)
         */
        public void synchronizeToSyncPoint(SyncPoint s) {
                // access to the syncpoints must be synchronized when working with
                // multithreading
                synchronized (syncOperationMontitor) {
                        // join all threads associated with the given syncPoint
                        Vector<OneMethodThread> threads = vectorMap.get(s);
                        // Multithreading: check if another thread has already synchronized
                        // to
                        // the given point
                        if (threads == alreadySynchronized) {
                                if (LOG.isDebugEnabled())
                                        LOG.debug("SyncPoint " + s.getId()
                                                        + " was already synchronized");
                                return;
                        }
                        // if no thread was found: Some error occured.
                        if (threads == null) {
                                UnsupportedOperationException e = new UnsupportedOperationException(
                                                "The Thread with the SyncPoint '" + s.getId()
                                                                + "' does not exist. Synchronisation failed!!");
                                LOG.error("The Thread with the SyncPoint '" + s.getId()
                                                + "' was not found. Synchronisation failed!!", e);
                                throw e;
                        }
                        for (Iterator iter = threads.iterator(); iter.hasNext();) {
                                OneMethodThread thread = (OneMethodThread) iter.next();
                                thread.join();
                        }
                        // indicate as synchronized
                        vectorMap.put(s, alreadySynchronized);
                }
        }
}
1	package appl.parallel.spmd;
2
3	import java.io.PrintStream;
4	import java.util.HashMap;
5	import java.util.Iterator;
6	import java.util.Vector;
7
8	import org.apache.log4j.LogManager;
9	import org.apache.log4j.Logger;
10
11	import appl.parallel.ComputingResourceContainer;
12	import appl.parallel.client.ClientDataServer;
13	import appl.parallel.event.CommEventSink;
14	import appl.parallel.thread.OneMethodThread;
15
16	/**
17	* Performance optimizations on client side can be made with this class. Allows
18	* heavy multithreading.<br>
19	* <br>
20	* Merging can be done in a separate thread (parallel to further computation)<br>
21	*
22	* @see SPMDClientController for details on the usage of a client controller
23	*
24	* @author Dominik Appl
25	*/
26	public class AdvancedSPMDClientController extends SPMDClientController
27	implements AdvancedSPMDClientInterface {
28
29	private final Logger LOG = LogManager.getLogger(this.getClass().getName());
30
31	// used for synchronization
32	private static final Object syncOperationMontitor = new Object();
33
34	/**
35	* For each {@link SyncPoint} a vector of threads associated with the point
36	* is created
37	*/
38	HashMap<SyncPoint, Vector<OneMethodThread>> vectorMap = new HashMap<SyncPoint, Vector<OneMethodThread>>();
39
40	/**
41	* Empty thread which is used as indicator for a already synchronized sync
42	* point
43	*/
44	private static final Vector<OneMethodThread> alreadySynchronized = new Vector<OneMethodThread>(
45	1);
46
47	/**
48	* same parameters as superclass
49	*
50	* @see SPMDClientController#SPMDClientController(Vector, double[],
51	* ClientDataServer, CommEventSink)
52	*/
53	public AdvancedSPMDClientController(
54	Vector<ComputingResourceContainer> computingResources,
55	double[] weights, ClientDataServer spmdClient,
56	CommEventSink eventProxy) {
57	super(computingResources, weights, spmdClient, eventProxy);
58	}
59
60	/*
61	* (non-Javadoc)
62	*
63	* @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergePartition(java.lang.Object,
64	* appl.parallel.spmd.SyncPoint)
65	*/
66	public void mergePartition(final Object partition, SyncPoint s,
67	final PrintStream stream, final String message) {
68	checkSplittable(partition);
69	// create a new thread
70	OneMethodThread newThread = new OneMethodThread("mergeThread"
71	+ s.getId(), s.getPriority()) {
72	@Override
73	public void run() {
74	mergePartition(partition);
75	if (stream != null) {
76	stream.println(message);
77	}
78	}
79	};
80	mapThreadToSyncPoint(newThread, s);
81	newThread.start();
82	}
83
84	/*
85	* (non-Javadoc)
86	*
87	* @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergePartition(java.lang.Object,
88	* appl.parallel.spmd.SyncPoint)
89	*/
90	public synchronized void mergePartition(Object partition, SyncPoint s) {
91	mergePartition(partition, s, null, null);
92	}
93
94	/**
95	* Puts a Thread/Syncpoint pair into the Hashmap/Vector structure
96	*/
97	private void mapThreadToSyncPoint(OneMethodThread thread, SyncPoint s) {
98	// access to the syncpoints must be synchronized when working with
99	// multithreading
100	synchronized (syncOperationMontitor) {
101	Vector<OneMethodThread> threads = vectorMap.get(s);
102	// if this is the first thread for the syncPoint: create a new
103	// Vector
104	if (threads == null) {
105	threads = new Vector<OneMethodThread>(5);
106	vectorMap.put(s, threads);
107	}
108	threads.add(thread);
109	}
110	}
111
112	/*
113	* (non-Javadoc)
114	*
115	* @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergeMultiData(appl.parallel.spmd.MultiDataObject,
116	* int, appl.parallel.spmd.SyncPoint, java.io.PrintStream,
117	* java.lang.String)
118	*/
119	public synchronized void mergeMultiData(MultiDataObject multidata, int idx,
120	SyncPoint s, PrintStream stream, String message) {
121	// create a new thread
122	OneMethodThread newThread = new OneMethodThread("mergeThread-"
123	+ s.getId(), s.getPriority(), multidata, idx, stream, message) {
124	@Override
125	public void run() {
126	mergeMultiData((MultiDataObject) getParameter(0),
127	(Integer) getParameter(1));
128	if (getParameter(2) != null) {
129	PrintStream stream = (PrintStream) getParameter(2);
130	stream.println(getParameter(3));
131	}
132	}
133	};
134	mapThreadToSyncPoint(newThread, s);
135	newThread.start();
136	}
137
138	/*
139	* (non-Javadoc)
140	*
141	* @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergeMultiData(appl.parallel.spmd.MultiDataObject,
142	* int, appl.parallel.spmd.SyncPoint)
143	*/
144	public void mergeMultiData(MultiDataObject multidata, int idx, SyncPoint s) {
145	mergeMultiData(multidata, idx, s, null, null);
146	}
147
148	/*
149	* (non-Javadoc)
150	*
151	* @see appl.parallel.spmd.AdvancedSPMDClientInterface#synchronizeToSyncPoint(appl.parallel.spmd.SyncPoint)
152	*/
153	public void synchronizeToSyncPoint(SyncPoint s) {
154	// access to the syncpoints must be synchronized when working with
155	// multithreading
156	synchronized (syncOperationMontitor) {
157	// join all threads associated with the given syncPoint
158	Vector<OneMethodThread> threads = vectorMap.get(s);
159	// Multithreading: check if another thread has already synchronized
160	// to
161	// the given point
162	if (threads == alreadySynchronized) {
163	if (LOG.isDebugEnabled())
164	LOG.debug("SyncPoint " + s.getId()
165	+ " was already synchronized");
166	return;
167	}
168	// if no thread was found: Some error occured.
169	if (threads == null) {
170	UnsupportedOperationException e = new UnsupportedOperationException(
171	"The Thread with the SyncPoint '" + s.getId()
172	+ "' does not exist. Synchronisation failed!!");
173	LOG.error("The Thread with the SyncPoint '" + s.getId()
174	+ "' was not found. Synchronisation failed!!", e);
175	throw e;
176	}
177	for (Iterator iter = threads.iterator(); iter.hasNext();) {
178	OneMethodThread thread = (OneMethodThread) iter.next();
179	thread.join();
180	}
181	// indicate as synchronized
182	vectorMap.put(s, alreadySynchronized);
183	}
184	}
185	}