parallel/spmd/AdvancedSPMDServerController.java

package appl.parallel.spmd;

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.server.PartitionDataManager;
import appl.parallel.spmd.split.DataPartition;
import appl.parallel.thread.OneMethodThread;

/**
 * Performance optimizations can be made with this class. Allows heavy
 * multithreading. Because multiple threads may use this class concurrently
 * nearly all additional methods of this class are internally synchronized using a static variable.
 * Also allows preloading of partitions using {@link SyncPoint}s.
 * 
 * @author Dominik Appl
 */
public class AdvancedSPMDServerController extends SPMDServerController
                implements AdvancedSPMDServerInterface {

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

        private static final HashMap<String, String> partitionPreloadStatus = 
                                                                    new HashMap<String, String>();

        // private static final HashMap<Integer,String> multiPartitionPreloadStatus
        // = new HashMap<Integer, String>();
        /**
         * For each {@link SyncPoint} a vector of threads associated with the id of
         * a SyncPoint is created. The {@link HashMap} is static so all possible threads use
         * the same map. Access to this map must be synchronized
         */
        private static final HashMap<Integer, Vector<OneMethodThread>> vectors =
                                                              new HashMap<Integer, Vector<OneMethodThread>>();

        private boolean isMaster = true;

        /**
         * same parameters as superclass
         * 
         * @see SPMDServerController#SPMDServerController(PartitionDataManager, int)
         */
        public AdvancedSPMDServerController(PartitionDataManager dataManager,
                        int referenceResouceID) {
                super(dataManager, referenceResouceID);
                // TODO Auto-generated constructor stub
        }

        /**
         * The partition is loaded in a separate thread into the local
         * {@link PartitionDataManager}<br>
         * You must specify the partition name and a {@link SyncPoint}.<br>
         * You can specify a priority for the thread in the {@link SyncPoint}.
         * <br>
         * <br>
         * <b>Preloading will only be done one time independent of the number of 
         * parallel threads calling this method. (synchronized)<br>
         * </b>
         * 
         * @param partition
         *            the name of the partition for identification
         * @param s
         *            the associated {@link SyncPoint}
         * @see SPMDServerController#getPartition(String)
         * @see #isMasterThread()
         */
        public void preloadPartition(String partition, SyncPoint s) {
                // create a new thread only if some other thread has not already
                // triggered the preloading
                synchronized (vectors) {
                        if (partitionPreloadStatus.get(partition) == null) {
                                partitionPreloadStatus.put(partition, "inProgress");
                                OneMethodThread newThread = new OneMethodThread(
                                                "getPartitionThread " + s.getId(), s.getPriority(),
                                                partition) {
                                        @Override
                                        public void run() {
                                                // retrieving the partition will as a sideeffect trigger
                                                // loading the
                                                // partition into the PartitionDataManager
                                                getPartition((String) getParameter(0));
                                        }
                                };
                                mapThreadToSyncPoint(newThread, s);
                                newThread.start();
                        }
                }
        }

        /**
         * The partition is loaded in a separate thread into the local
         * {@link PartitionDataManager} You must specify the partition name and a
         * {@link SyncPoint}. You can specify a priority for the thread in the
         * {@link SyncPoint} <br>
         * <b>Preloading will only be done one time independent of the number of 
         * parallel threads calling this method. (synchronized)<br>
         * </b>
         * 
         * @param multiPartition
         *            the name of the partition for identification
         * @param idx
         *            the index of the partition in the multidata object
         * @param s
         *            the associated {@link SyncPoint}
         * @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergePartition(java.lang.Object,
         *      appl.parallel.spmd.SyncPoint)
         * @see SPMDServerController#getPartition(String)
         */
        public void preloadMultiPartition(String multiPartition, int idx,
                        SyncPoint s) {
                synchronized (vectors) {
                        // create a new thread only if some other thread has not already
                        // triggered the preloading
                        if (partitionPreloadStatus.get(multiPartition
                                        + MultiDataInfo.MULTISPLITCONSTANT + idx) == null) {
                                partitionPreloadStatus.put(multiPartition
                                                + MultiDataInfo.MULTISPLITCONSTANT + idx, "inProgress");

                                OneMethodThread newThread = new OneMethodThread(
                                                "Preload thread - getMultiPartition SID:" + s.getId(),
                                                s.getPriority(), multiPartition, idx) {
                                        @Override
                                        public void run() {
                                                // retrieving the partition will also trigger loading
                                                // the
                                                // partition into the PartitionDataManager
                                                getMultiPartition((String) getParameter(0),
                                                                (Integer) getParameter(1));
                                        }
                                };
                                mapThreadToSyncPoint(newThread, s);
                                newThread.start();
                        }
                }
        }

        /*
         * (non-Javadoc)
         * 
         * @see appl.parallel.spmd.AdvancedSPMDServerInterface#synchronizeToSyncPoint(appl.parallel.spmd.SyncPoint)
         */
        public void synchronizeToSyncPoint(SyncPoint s) {
                // join all threads associated with the given syncPoint
                synchronized (vectors) {
                        Vector<OneMethodThread> threads = vectors.get(s.getId());
                        if (threads == null) {
                                UnsupportedOperationException e = new UnsupportedOperationException(
                                                "The Thread with the SyncPoint '"
                                                                + s.getId()
                                                                + "' was not found or was earlier synchronized. 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();

                        }
                }
        }

        /**
         * You should call this method from time to time to clear all syncpoints and
         * the associated threads out of the memory
         */
        public void clearSyncData() {
                synchronized (vectors) {
                        vectors.clear();
                        partitionPreloadStatus.clear();
                }
        }

        /**
         * Puts a thread/Syncpoint pair into the Hashmap/Vector structure
         */
        private static void mapThreadToSyncPoint(OneMethodThread thread, SyncPoint s) {
                Vector<OneMethodThread> threads = vectors.get(s.getId());
                // if this is the first thread for the syncPoint: create a new
                // Vector
                if (threads == null) {
                        threads = new Vector<OneMethodThread>(5);
                        vectors.put(s.getId(), threads);
                }
                threads.add(thread);

        }

        /**
         * If there are multiple processors on the machine and the current task
         * {@link SPMDTask#supportsMultiThreading() supports multithreading} there
         * are several threads created. Each thread uses a different
         * {@link #getLocalBounds(DataPartition) calculation area}. Now there might
         * be parts of the task code were multithreading is undesired, e.g. when
         * operations affect the whole partition. In this case you should use this
         * method to guarantee that only one thread has access to a code block.
         * 
         * @return whether this thread is the "master" thread, which means whether
         *         this thread refers to the first task created.
         */
        public boolean isMasterThread() {
                return isMaster;

        }

        /**
         * @param isMaster
         *            true, if this should be the master thread
         * @see #isMasterThread()
         */
        public void setMasterThread(boolean isMaster) {
                this.isMaster = isMaster;
        }
}
1	package appl.parallel.spmd;
2
3	import java.util.HashMap;
4	import java.util.Iterator;
5	import java.util.Vector;
6
7	import org.apache.log4j.LogManager;
8	import org.apache.log4j.Logger;
9
10	import appl.parallel.server.PartitionDataManager;
11	import appl.parallel.spmd.split.DataPartition;
12	import appl.parallel.thread.OneMethodThread;
13
14	/**
15	* Performance optimizations can be made with this class. Allows heavy
16	* multithreading. Because multiple threads may use this class concurrently
17	* nearly all additional methods of this class are internally synchronized using a static variable.
18	* Also allows preloading of partitions using {@link SyncPoint}s.
19	*
20	* @author Dominik Appl
21	*/
22	public class AdvancedSPMDServerController extends SPMDServerController
23	implements AdvancedSPMDServerInterface {
24
25	private final Logger LOG = LogManager.getLogger(this.getClass().getName());
26
27	private static final HashMap<String, String> partitionPreloadStatus =
28	new HashMap<String, String>();
29
30	// private static final HashMap<Integer,String> multiPartitionPreloadStatus
31	// = new HashMap<Integer, String>();
32	/**
33	* For each {@link SyncPoint} a vector of threads associated with the id of
34	* a SyncPoint is created. The {@link HashMap} is static so all possible threads use
35	* the same map. Access to this map must be synchronized
36	*/
37	private static final HashMap<Integer, Vector<OneMethodThread>> vectors =
38	new HashMap<Integer, Vector<OneMethodThread>>();
39
40	private boolean isMaster = true;
41
42	/**
43	* same parameters as superclass
44	*
45	* @see SPMDServerController#SPMDServerController(PartitionDataManager, int)
46	*/
47	public AdvancedSPMDServerController(PartitionDataManager dataManager,
48	int referenceResouceID) {
49	super(dataManager, referenceResouceID);
50	// TODO Auto-generated constructor stub
51	}
52
53	/**
54	* The partition is loaded in a separate thread into the local
55	* {@link PartitionDataManager}<br>
56	* You must specify the partition name and a {@link SyncPoint}.<br>
57	* You can specify a priority for the thread in the {@link SyncPoint}.
58	* <br>
59	* <br>
60	* <b>Preloading will only be done one time independent of the number of
61	* parallel threads calling this method. (synchronized)<br>
62	* </b>
63	*
64	* @param partition
65	* the name of the partition for identification
66	* @param s
67	* the associated {@link SyncPoint}
68	* @see SPMDServerController#getPartition(String)
69	* @see #isMasterThread()
70	*/
71	public void preloadPartition(String partition, SyncPoint s) {
72	// create a new thread only if some other thread has not already
73	// triggered the preloading
74	synchronized (vectors) {
75	if (partitionPreloadStatus.get(partition) == null) {
76	partitionPreloadStatus.put(partition, "inProgress");
77	OneMethodThread newThread = new OneMethodThread(
78	"getPartitionThread " + s.getId(), s.getPriority(),
79	partition) {
80	@Override
81	public void run() {
82	// retrieving the partition will as a sideeffect trigger
83	// loading the
84	// partition into the PartitionDataManager
85	getPartition((String) getParameter(0));
86	}
87	};
88	mapThreadToSyncPoint(newThread, s);
89	newThread.start();
90	}
91	}
92	}
93
94	/**
95	* The partition is loaded in a separate thread into the local
96	* {@link PartitionDataManager} You must specify the partition name and a
97	* {@link SyncPoint}. You can specify a priority for the thread in the
98	* {@link SyncPoint} <br>
99	* <b>Preloading will only be done one time independent of the number of
100	* parallel threads calling this method. (synchronized)<br>
101	* </b>
102	*
103	* @param multiPartition
104	* the name of the partition for identification
105	* @param idx
106	* the index of the partition in the multidata object
107	* @param s
108	* the associated {@link SyncPoint}
109	* @see appl.parallel.spmd.AdvancedSPMDClientInterface#mergePartition(java.lang.Object,
110	* appl.parallel.spmd.SyncPoint)
111	* @see SPMDServerController#getPartition(String)
112	*/
113	public void preloadMultiPartition(String multiPartition, int idx,
114	SyncPoint s) {
115	synchronized (vectors) {
116	// create a new thread only if some other thread has not already
117	// triggered the preloading
118	if (partitionPreloadStatus.get(multiPartition
119	+ MultiDataInfo.MULTISPLITCONSTANT + idx) == null) {
120	partitionPreloadStatus.put(multiPartition
121	+ MultiDataInfo.MULTISPLITCONSTANT + idx, "inProgress");
122
123	OneMethodThread newThread = new OneMethodThread(
124	"Preload thread - getMultiPartition SID:" + s.getId(),
125	s.getPriority(), multiPartition, idx) {
126	@Override
127	public void run() {
128	// retrieving the partition will also trigger loading
129	// the
130	// partition into the PartitionDataManager
131	getMultiPartition((String) getParameter(0),
132	(Integer) getParameter(1));
133	}
134	};
135	mapThreadToSyncPoint(newThread, s);
136	newThread.start();
137	}
138	}
139	}
140
141	/*
142	* (non-Javadoc)
143	*
144	* @see appl.parallel.spmd.AdvancedSPMDServerInterface#synchronizeToSyncPoint(appl.parallel.spmd.SyncPoint)
145	*/
146	public void synchronizeToSyncPoint(SyncPoint s) {
147	// join all threads associated with the given syncPoint
148	synchronized (vectors) {
149	Vector<OneMethodThread> threads = vectors.get(s.getId());
150	if (threads == null) {
151	UnsupportedOperationException e = new UnsupportedOperationException(
152	"The Thread with the SyncPoint '"
153	+ s.getId()
154	+ "' was not found or was earlier synchronized. Synchronisation failed!!");
155	LOG.error("The Thread with the SyncPoint '" + s.getId()
156	+ "' was not found. Synchronisation failed!!", e);
157	throw e;
158	}
159	for (Iterator iter = threads.iterator(); iter.hasNext();) {
160	OneMethodThread thread = (OneMethodThread) iter.next();
161	thread.join();
162
163	}
164	}
165	}
166
167	/**
168	* You should call this method from time to time to clear all syncpoints and
169	* the associated threads out of the memory
170	*/
171	public void clearSyncData() {
172	synchronized (vectors) {
173	vectors.clear();
174	partitionPreloadStatus.clear();
175	}
176	}
177
178	/**
179	* Puts a thread/Syncpoint pair into the Hashmap/Vector structure
180	*/
181	private static void mapThreadToSyncPoint(OneMethodThread thread, SyncPoint s) {
182	Vector<OneMethodThread> threads = vectors.get(s.getId());
183	// if this is the first thread for the syncPoint: create a new
184	// Vector
185	if (threads == null) {
186	threads = new Vector<OneMethodThread>(5);
187	vectors.put(s.getId(), threads);
188	}
189	threads.add(thread);
190
191	}
192
193	/**
194	* If there are multiple processors on the machine and the current task
195	* {@link SPMDTask#supportsMultiThreading() supports multithreading} there
196	* are several threads created. Each thread uses a different
197	* {@link #getLocalBounds(DataPartition) calculation area}. Now there might
198	* be parts of the task code were multithreading is undesired, e.g. when
199	* operations affect the whole partition. In this case you should use this
200	* method to guarantee that only one thread has access to a code block.
201	*
202	* @return whether this thread is the "master" thread, which means whether
203	* this thread refers to the first task created.
204	*/
205	public boolean isMasterThread() {
206	return isMaster;
207
208	}
209
210	/**
211	* @param isMaster
212	* true, if this should be the master thread
213	* @see #isMasterThread()
214	*/
215	public void setMasterThread(boolean isMaster) {
216	this.isMaster = isMaster;
217	}
218	}