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