*
* BASIC CONCEPTS *
* A cluster parallel computer typically consists of a frontend * processor and a number of backend processors connected via a * dedicated high-speed network. A user logs into the frontend processor and * runs a PJ program there. The PJ message passing middleware causes copies of * the PJ program to run in a number of separate processes, each process on a * different backend processor. The backend processes run the PJ program, using * the PJ middleware to send messages amongst themselves. The PJ middleware * redirects the backend processes' standard output and standard error streams * to the frontend process. The frontend process does not actually execute the * PJ program, but merely displays all the backend processes' standard output * and standard error streams on the frontend process's own standard output and * standard error. *
* For the PJ message passing middleware to work, certain server processes must * be running. See package {@linkplain edu.rit.pj.cluster} for further * information. *
* To initialize the PJ message passing middleware, the program must first call * the static Comm.init() method, passing in the command line * arguments. *
* A communicator is associated with a group of backend processes. The * communicator's size is the number of processes in the group. Each * process in the communicator has a different rank in the range 0 .. * size-1. A process may obtain the size and rank by calling the * communicator's size() and rank() methods. *
* There is one predefined communicator, the world communicator, * consisting of all the backend processes in the parallel program. A process * may obtain a reference to the world communicator by calling the static * Comm.world() method. Typically, the first few lines in a PJ cluster * parallel program look like this: *
* public class AParallelProgram
* {
* public static void main
* (String[] args)
* throws Exception
* {
* Comm.init (args);
* Comm world = Comm.world();
* int size = world.size();
* int rank = world.rank();
* . . .
* * The number of processes in the parallel program -- that is, the size of the * world communicator -- is specified by the "pj.np" property, which * must be an integer greater than or equal to 1. You can specify the number of * processes on the Java command line like this: *
* java -Dpj.np=4 . . .*
* If the "pj.np" property is not specified, the default is 1. *
* The PJ program will run on the specified number of backend processors as * described above. To determine which backend processors to use, the PJ program * interacts with a Job Scheduler server process running on the frontend * processor. When the PJ program starts and calls the Comm.init() * method, the middleware first prints the job number on the standard error. The * middleware then waits until the required number of backend processors are * ready to run a job. As each backend processor becomes ready, the middleware * prints on the standard error the name of each backend processor assigned to * the job. Once all are ready, the PJ program starts running on those backend * processors, and all further output comes from the PJ program. Since each PJ * program interacts with the Job Scheduler, the Job Scheduler can ensure that * each backend processor is running a backend process for only one job at a * time. *
* Depending on the system load, your PJ program may have to wait in the Job * Scheduler's queue for a while until enough backend processors become ready. * If you get tired of waiting, you can kill your PJ program (e.g., by typing * CTRL-C), which will remove your PJ program from the Job Scheduler's queue. *
* The Job Scheduler has a web interface that lets you examine the cluster * status. Just point your web browser at this URL: *
* http://<hostname>:8080/ *
* where <hostname> is replaced by the host name of the frontend * processor. The default port for the cluster status web interface is port * 8080. The Job Scheduler can be configured to use a different port. For * further information, see package {@linkplain edu.rit.pj.cluster}. *
* If the PJ program is executed on a host where no Job Scheduler is running, * the PJ program will run in one process on that host (i.e., the machine * you're logged into), rather than on the backend processors. The message * passing methods in class Comm will still work, though. This option can be * useful for debugging a PJ program's logic on a non-parallel machine before * running the PJ program on a cluster. *
*
* MESSAGE PASSING *
* PJ provides two categories of communication, point-to-point * communication and collective communication. The following methods * of class Comm are used for point-to-point communication: *
* In addition, you can create a new communicator consisting of all, or a subset * of, the processes in an existing communicator. Message passing in the new * communicator is completely independent of message passing in the original * communicator. The following method creates a new communicator: *
*
* POINT-TO-POINT COMMUNICATION *
* One process in a PJ cluster parallel program, the source process, may * use a communicator to send a message to another process in the program, the * destination process. This is called a point-to-point * communication because just the two processes are involved (as opposed to * a collective communication, which involves all the processes). Five * point-to-point communication methods are available in this release: send, * receive, send-receive, flood-send, and flood-receive. *
* Send and Receive *
* To do a point-to-point communication, the source process calls the * send() method on a certain communicator, such as the world * communicator. The source process specifies the destination process's rank, * the tag for the message, and a buffer containing the data items * to be sent (type {@linkplain edu.rit.mp.Buf}). Likewise, the destination * process calls the receive() method on the same communicator as the * source process. The destination process specifies the source process's rank, * the message tag which must be the same as in the source process, and the * buffer for the data items to be received. *
* A send() method call and a receive() method call are said * to match if (a) the rank passed to the send() method equals * the rank of the process calling receive(), (b) the rank passed to * the receive() method equals the rank of the process calling * send(), (c) the item data type in the source buffer is the same as * the item data type in the destination buffer, and (d) the send message tag * equals the receive message tag. A receive() method call will block * until a matching send() method call occurs. If more than one * send() method call matches a receive() method call, one of * the matching send() method calls is picked in an unspecified manner. * A send() method call may block until a matching * receive() method call occurs due to flow control in the underlying * network communication. *
* The message tag can be used to distinguish different kinds of messages. A * receive() method call will only match a send() method call * with the same tag. If there is no need to distinguish different kinds of * messages, omit the tag (it will default to 0). *
* Once a send() method call and a receive() method call have * been matched together, the actual message data transfer takes place. Each * item in the source buffer, starting at index 0 and continuing for the entire * length of the source buffer, is written to the message. At the other end, * each item in the destination buffer, starting at index 0, is read from the * message. *
* The receive() method returns a {@linkplain CommStatus} object. The * status object gives the actual rank of the process that sent the message, the * actual message tag that was received, and the actual number of data items in * the message. If the actual number of data items in the message is less than * the length of the destination buffer, nothing is stored into the extra data * items at the end of the destination buffer. If the actual number of data * items in the message is greater than the length of the destination buffer, * the extra data items at the end of the message are discarded. *
* The send() method does not return until all the message elements * have been written from the source buffer. Likewise, the receive() * method does not return until all the message elements have been read into the * destination buffer. However, you cannot assume that because the * send() method has returned, the matching receive() method * has also returned. Because of buffering in the underlying network * communication, not all the destination items might have been received even * though all the source items have been sent. *
* The destination process, instead of specifying a particular source process, * can declare that it will receive a message from any source process by * specifying null for the source process rank in the receive() method * call. This is called a wildcard source. In this case the * receive() method call's returned status object will indicate the * actual source process that sent the message. *
* The destination process, instead of specifying a particular message tag, can * declare that it will receive a message with any tag by specifying null for * the tag in the receive() method call. This is called a wildcard * tag. Alternatively, the destination process can specify a range of * message tags, and it will receive a message with any tag in the given range. * In these cases the receive() method call's returned status object * will indicate the actual message tag that was sent. *
* A process can send a message to itself. In this case one thread must call * send() (specifying the process's own rank as the destination) and a * different thread must call receive() (specifying the process's own * rank as the source), otherwise a deadlock will ensue. *
* Send-Receive *
* By calling the sendReceive() method, a process can send a buffer of * outgoing message items to a destination process while simultaneously * receiving a buffer of incoming message items from a source process. The * destination process may be the same as the source process, or different from * the source process. The outgoing message items must come from a different * place than where the incoming message items will be stored, otherwise the * incoming message items may overwrite the outgoing message items before they * can be sent. When the sendReceive() method returns, the outgoing * message items have been fully sent, but they may not yet have been fully * received; and the incoming message items have been fully received. *
* With the sendReceive() method, a process cannot receive a message * from a wildcard source, and a process cannot receive a message with a * wildcard tag or a range of tags. The process calling sendReceive() * must know the rank of the source process and the message tag (if not 0). The * sendReceive() method does return a status object giving the outcome * of the receive half of the send-receive operation, just as the * receive() method does. *
* A process can send-receive messages with itself. In this case one thread must * call sendReceive() (specifying the process's own rank as the source * and destination) and a different thread must also call sendReceive() * (specifying the process's own rank as the source and destination), otherwise * a deadlock will ensue. *
* Non-Blocking Communication *
* The send(), receive(), and sendReceive() methods * each have a non-blocking version. A non-blocking communication method * initiates the communication operation and immediately returns, storing the * state of the communication operation in a {@linkplain CommRequest} object. * The communicator then performs the communication operation in a separate * thread. This allows the calling thread to do other work while the * communication operation is in progress. To wait for the send and receive * operations to finish, call the CommRequest object's waitForFinish() * method. *
* Flood-Send and Flood-Receive *
* Any process can send a message to all processes in the communicator. This is * called "flooding" the message. First, all processes must start a * flood-receive operation, either by calling the non-blocking * floodReceive() method, or by having a separate thread call the * blocking floodReceive() method. Then, one process (any process) must * call the floodSend() method. The data items in the flood-send * operation's outgoing buffer are copied into the flood-receive operation's * incoming buffer in all processes. *
* Message flooding is similar to the "broadcast" collective communication * operation (see below). The differences are these: Broadcasting combines * sending and receiving in a single operation; flooding uses separate send and * receive operations. For broadcasting, every process must know which process * is sending the outgoing data items; for flooding, the receiving processes do * not need to know which process is sending (any process can send). *
*
* COLLECTIVE COMMUNICATION *
* A PJ cluster parallel program may use a communicator to send a message among * all the processes in the program at the same time. This is called a * collective communication because all the processes in the communicator * are involved (as opposed to a point-to-point communication). Ten collective * communication methods are available in this release: broadcast, scatter, * gather, all-gather, reduce, all-reduce, all-to-all, scan, exclusive-scan, and * barrier. Further collective communication methods will be added to class Comm * in a later release. *
* Broadcast *
* One process in the communicator, the root process, has a source buffer * (type {@linkplain edu.rit.mp.Buf Buf}) filled with data. The other processes * in the communicator each have a destination buffer with the same length and * the same item data type as the source buffer. Each process calls the * communicator's broadcast() method. Afterwards, all the destination * buffers contain the same data as the source buffer. *
*
* Before:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | | | | | | * | 2 | | | | | | | * | 3 | | | | | | | * | 4 | | | | | | | * | 5 | | | | | | | * | 6 | | | | | | | * | 7 | | | | | | | * | 8 | | | | | | | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 1 | | 1 | | 1 | * | 2 | | 2 | | 2 | | 2 | * | 3 | | 3 | | 3 | | 3 | * | 4 | | 4 | | 4 | | 4 | * | 5 | | 5 | | 5 | | 5 | * | 6 | | 6 | | 6 | | 6 | * | 7 | | 7 | | 7 | | 7 | * | 8 | | 8 | | 8 | | 8 | * +----+ +----+ +----+ +----+* * |
*
* Scatter *
* One process in the communicator, the root process, has K source * buffers (type {@linkplain edu.rit.mp.Buf Buf}) filled with data, where * K is the size of the communicator. For example, the source buffers * could be different portions of an array. Each process in the communicator * (including the root process) has a destination buffer with the same length * and the same item data type as the corresponding source buffer. Each process * calls the communicator's scatter() method. Afterwards, each * process's destination buffer contains the same data as the corresponding * source buffer in the root process. *
*
* Before:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ * | 1 | * | 2 | * +----+ * | 3 | * | 4 | * +----+ * | 5 | * | 6 | * +----+ * | 7 | * | 8 | * +----+ * * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ * | 1 | * | 2 | * +----+ * | 3 | * | 4 | * +----+ * | 5 | * | 6 | * +----+ * | 7 | * | 8 | * +----+ * * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+* * |
*
*
* Before:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ * | 1 | * | 2 | * +----+ +----+ * | 3 | | | * | 4 | | | * +----+ +----+ +----+ * | 5 | | | * | 6 | | | * +----+ +----+ +----+ * | 7 | | | * | 8 | | | * +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ * | 1 | * | 2 | * +----+ +----+ * | 3 | | 3 | * | 4 | | 4 | * +----+ +----+ +----+ * | 5 | | 5 | * | 6 | | 6 | * +----+ +----+ +----+ * | 7 | | 7 | * | 8 | | 8 | * +----+ +----+* * |
*
* Gather *
* Gather is the opposite of scatter. One process in the communicator, the root * process, has K destination buffers (type {@linkplain edu.rit.mp.Buf * Buf}), where K is the size of the communicator. For example, the * destination buffers could be different portions of an array. Each process in * the communicator (including the root process) has a source buffer with the * same length and the same item data type as the corresponding destination * buffer, filled with data. Each process calls the communicator's * gather() method. Afterwards, each destination buffer in the root * process contains the same data as the corresponding source buffer. *
*
* Before:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+ * * +----+ * | | * | | * +----+ * | | * | | * +----+ * | | * | | * +----+ * | | * | | * +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+ * * +----+ * | 1 | * | 2 | * +----+ * | 3 | * | 4 | * +----+ * | 5 | * | 6 | * +----+ * | 7 | * | 8 | * +----+* * |
*
*
* Before:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ * | 1 | * | 2 | * +----+ +----+ * | | | 3 | * | | | 4 | * +----+ +----+ +----+ * | | | 5 | * | | | 6 | * +----+ +----+ +----+ * | | | 7 | * | | | 8 | * +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ * | 1 | * | 2 | * +----+ +----+ * | 3 | | 3 | * | 4 | | 4 | * +----+ +----+ +----+ * | 5 | | 5 | * | 6 | | 6 | * +----+ +----+ +----+ * | 7 | | 7 | * | 8 | | 8 | * +----+ +----+* * |
*
* All-Gather *
* All-gather is the same as gather, except that every process has an array of * destination buffers, and every process receives the results of the gather. * Each process in the communicator has a source buffer (type {@linkplain * edu.rit.mp.Buf Buf}) filled with data. Each process in the communicator has * K destination buffers, where K is the size of the communicator. * For example, the destination buffers could be different portions of an array. * Each destination buffer has the same length and the same item data type as * the corresponding source buffer. Each process calls the communicator's * allGather() method. Afterwards, each destination buffer in every * process contains the same data as the corresponding source buffer. *
*
* Before:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+ * * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+ * * +----+ +----+ +----+ +----+ * | 1 | | 1 | | 1 | | 1 | * | 2 | | 2 | | 2 | | 2 | * +----+ +----+ +----+ +----+ * | 3 | | 3 | | 3 | | 3 | * | 4 | | 4 | | 4 | | 4 | * +----+ +----+ +----+ +----+ * | 5 | | 5 | | 5 | | 5 | * | 6 | | 6 | | 6 | | 6 | * +----+ +----+ +----+ +----+ * | 7 | | 7 | | 7 | | 7 | * | 8 | | 8 | | 8 | | 8 | * +----+ +----+ +----+ +----+* * |
*
*
* Before:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | | | | | | * | 2 | | | | | | | * +----+ +----+ +----+ +----+ * | | | 3 | | | | | * | | | 4 | | | | | * +----+ +----+ +----+ +----+ * | | | | | 5 | | | * | | | | | 6 | | | * +----+ +----+ +----+ +----+ * | | | | | | | 7 | * | | | | | | | 8 | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 1 | | 1 | | 1 | * | 2 | | 2 | | 2 | | 2 | * +----+ +----+ +----+ +----+ * | 3 | | 3 | | 3 | | 3 | * | 4 | | 4 | | 4 | | 4 | * +----+ +----+ +----+ +----+ * | 5 | | 5 | | 5 | | 5 | * | 6 | | 6 | | 6 | | 6 | * +----+ +----+ +----+ +----+ * | 7 | | 7 | | 7 | | 7 | * | 8 | | 8 | | 8 | | 8 | * +----+ +----+ +----+ +----+* * |
*
* Reduce *
* Reduce is like gather, except the buffers' contents are combined together * instead of just copied. Each process in the communicator has a buffer (type * {@linkplain edu.rit.mp.Buf Buf}) filled with data. Each process calls the * communicator's reduce() method, specifying some binary operation * (type {@linkplain edu.rit.pj.reduction.Op Op}) for combining the data. * Afterwards, each element of the buffer in the root process contains the * result of combining all the corresponding elements in all the buffers using * the specified binary operation. For example, if the operation is addition, * each buffer element in the root process ends up being the sum of the * corresponding buffer elements in all the processes. In the non-root * processes, the buffers' contents may be changed from their original contents. *
*
* Before:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 (root) 1 2 3 * +----+ +----+ +----+ +----+ * | 16 | | ?? | | ?? | | ?? | * | 20 | | ?? | | ?? | | ?? | * +----+ +----+ +----+ +----+* * |
*
* All-Reduce *
* All-reduce is the same as reduce, except that every process receives the * results of the reduction. Each process in the communicator has a buffer (type * {@linkplain edu.rit.mp.Buf Buf}) filled with data. Each process calls the * communicator's allReduce() method, specifying some binary operation * (type {@linkplain edu.rit.pj.reduction.Op Op}) for combining the data. * Afterwards, each element of the buffer in each process contains the result of * combining all the corresponding elements in all the buffers using the * specified binary operation. For example, if the operation is addition, each * buffer element ends up being the sum of the corresponding buffer elements. *
*
* Before:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 16 | | 16 | | 16 | | 16 | * | 20 | | 20 | | 20 | | 20 | * +----+ +----+ +----+ +----+* * |
*
* All-to-All *
* Every process in the communicator has K source buffers (type * {@linkplain edu.rit.mp.Buf Buf}) filled with data, where K is the size * of the communicator. Every process in the communicator also has K * destination buffers (type {@linkplain edu.rit.mp.Buf Buf}). The source * buffers and the destination buffers must refer to different storage. For * example, the source buffers could be portions of an array, and the * destination buffers could be portions of a different array. Each process * calls the communicator's allToAll() method. Afterwards, for each * process rank k, 0 <= k <= K-1, and each buffer * index i, 0 <= i <= K-1, destination buffer * i in process k contains the same data as source buffer k * in process i. *
*
* Before:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 9 | | 17 | | 25 | * | 2 | | 10 | | 18 | | 26 | * +----+ +----+ +----+ +----+ * | 3 | | 11 | | 19 | | 27 | * | 4 | | 12 | | 20 | | 28 | * +----+ +----+ +----+ +----+ * | 5 | | 13 | | 21 | | 29 | * | 6 | | 14 | | 22 | | 30 | * +----+ +----+ +----+ +----+ * | 7 | | 15 | | 23 | | 31 | * | 8 | | 16 | | 24 | | 32 | * +----+ +----+ +----+ +----+ * * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+ * | | | | | | | | * | | | | | | | | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 9 | | 17 | | 25 | * | 2 | | 10 | | 18 | | 26 | * +----+ +----+ +----+ +----+ * | 3 | | 11 | | 19 | | 27 | * | 4 | | 12 | | 20 | | 28 | * +----+ +----+ +----+ +----+ * | 5 | | 13 | | 21 | | 29 | * | 6 | | 14 | | 22 | | 30 | * +----+ +----+ +----+ +----+ * | 7 | | 15 | | 23 | | 31 | * | 8 | | 16 | | 24 | | 32 | * +----+ +----+ +----+ +----+ * * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+ * | 9 | | 11 | | 13 | | 15 | * | 10 | | 12 | | 14 | | 16 | * +----+ +----+ +----+ +----+ * | 17 | | 19 | | 21 | | 23 | * | 18 | | 20 | | 22 | | 24 | * +----+ +----+ +----+ +----+ * | 25 | | 27 | | 29 | | 31 | * | 26 | | 28 | | 30 | | 32 | * +----+ +----+ +----+ +----+* * |
*
* Scan *
* Each process in the communicator has a buffer (type {@linkplain * edu.rit.mp.Buf Buf}) filled with data. Each process calls the communicator's * scan() method, specifying some binary operation (type {@linkplain * edu.rit.pj.reduction.Op Op}) for combining the data. Afterwards, each element * of the buffer in a particular process contains the result of combining all * the corresponding elements in its own and all lower-ranked processes' buffers * using the specified binary operation. For example, if the operation is * addition, each buffer element ends up being the sum of its own and all * lower-ranked processes' buffer elements. *
*
* Before:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 4 | | 9 | | 16 | * | 2 | | 6 | | 12 | | 20 | * +----+ +----+ +----+ +----+* * |
*
* Exclusive-Scan *
* The exclusive-scan operation is a variation of the scan operation. Each * process in the communicator has a buffer (type {@linkplain edu.rit.mp.Buf * Buf}) filled with data. Each process calls the communicator's * exclusiveScan() method, specifying some binary operation (type * {@linkplain edu.rit.pj.reduction.Op Op}) for combining the data, and * specifying an initial data value. Afterwards, each element of the buffer in a * particular process contains the result of combining all the corresponding * elements in all lower-ranked processes' buffers using the specified binary * operation, except in process 0 each element of the buffer contains the * initial data value. For example, if the operation is addition and the initial * data value is 0, each buffer element ends up being the sum of all * lower-ranked processes' buffer elements. *
*
* Before:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 1 | | 3 | | 5 | | 7 | * | 2 | | 4 | | 6 | | 8 | * +----+ +----+ +----+ +----+* * |
* * |
* After:
*
* * Process Process Process Process * 0 1 2 3 * +----+ +----+ +----+ +----+ * | 0 | | 1 | | 4 | | 9 | * | 0 | | 2 | | 6 | | 12 | * +----+ +----+ +----+ +----+* * |
*
* Barrier *
* The barrier operation causes all the processes to synchronize with each
* other. Each process calls the communicator's barrier() method. The
* calling thread blocks until all processes in the communicator have called the
* barrier() method. Then the calling thread unblocks and returns from
* the barrier() method call.
*
* @author Alan Kaminsky
* @version 21-Jan-2009
*/
public class Comm
{
// Hidden data members.
// Predefined communicators.
private static Comm theWorldCommunicator;
private static Comm theFrontendCommunicator;
// This communicator's size, rank, and host.
private int mySize;
private int myRank;
private String myHost;
// The largest power of 2 less than or equal to this communicator's size.
private int mySizePowerOf2;
// Channel group for message passing in this communicator.
private ChannelGroup myChannelGroup;
// Map from rank (array index) to channel group address (array element).
private InetSocketAddress[] myAddressForRank;
// Map from rank (array index) to channel for communicating with the process
// at that rank (array element).
private Channel[] myChannelForRank;
// Broadcast trees for flood-send, flood-receive, broadcast, and reduce
// operations, indexed by root.
private int[][] myBroadcastTree;
// Hidden constructors.
/**
* Construct a new communicator.
*
* @param size
* Communicator's size.
* @param rank
* Current process's rank in the communicator.
* @param host
* Host name.
* @param channelgroup
* Channel group for message passing in this communicator.
* @param address
* Map from rank (array index) to channel group address (array element).
*/
private Comm
(int size,
int rank,
String host,
ChannelGroup channelgroup,
InetSocketAddress[] address)
{
// Record size, rank, channel group.
mySize = size;
myRank = rank;
myHost = host;
myChannelGroup = channelgroup;
// Determine the largest power of 2 less than or equal to this
// communicator's size.
int p2 = 1;
while (p2 <= size) p2 <<= 1;
mySizePowerOf2 = p2 >>> 1;
// Set channel group ID equal to rank.
myChannelGroup.setChannelGroupId (rank);
// Set up connect listener.
myChannelGroup.setConnectListener (new ConnectListener()
{
public void nearEndConnected
(ChannelGroup theChannelGroup,
Channel theChannel)
throws IOException
{
}
public void farEndConnected
(ChannelGroup theChannelGroup,
Channel theChannel)
throws IOException
{
doFarEndConnected (theChannel);
}
});
// Record socket address for each process rank.
myAddressForRank = address;
// Set up channel for each process rank.
myChannelForRank = new Channel [size];
// Populate channel at my own rank with the loopback channel.
myChannelForRank[myRank] = channelgroup.loopbackChannel();
// If there's more than one process, start listening for incoming
// connections.
if (mySize > 1) myChannelGroup.startListening();
}
// Exported operations.
/**
* Initialize the PJ message passing middleware. Certain Java system
* properties specify the middleware's behavior; these properties are
* typically given on the Java command line with the "-D" flag.
* For further information, see class {@linkplain PJProperties}.
*
* @param args Command line arguments.
*
* @exception NullPointerException
* (unchecked exception) Thrown if args is null.
* @exception IllegalArgumentException
* (unchecked exception) Thrown if the value of one of the Java system
* properties is illegal.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public static void init
(String[] args)
throws IOException
{
// Verify preconditions.
if (args == null)
{
throw new NullPointerException ("Comm.init(): args is null");
}
// Get the job backend object.
JobBackend backend = JobBackend.getJobBackend();
if (backend == null)
{
// We're running on the frontend processor.
// Prepare constructor arguments for the Job Frontend object.
String username = System.getProperty ("user.name");
int Nn = PJProperties.getPjNn();
int Np = PJProperties.getPjNp();
int Nt = PJProperties.getPjNt();
boolean hasFrontendComm = false;
// Examine the call stack to find the main program class name.
StackTraceElement[] stack =
Thread.currentThread().getStackTrace();
StackTraceElement bottom = stack[stack.length-1];
if (! bottom.getMethodName().equals ("main"))
{
throw new IllegalStateException
("Comm.init(): Not called from main program");
}
String mainClassName = bottom.getClassName();
// Set up the Job Frontend object.
JobFrontend frontend = null;
try
{
frontend =
new JobFrontend
(username, Nn, Np, Nt, hasFrontendComm, mainClassName,
args);
// We were able to contact the Job Scheduler.
// Run the job frontend in this process, then exit.
frontend.run();
System.exit (0);
}
catch (JobSchedulerException exc)
{
// We were not able to contact the Job Scheduler.
System.err.println
("No Job Scheduler at " +
PJProperties.getPjHost() + ":" +
PJProperties.getPjPort() +
", running in this (one) process");
// Set up world communicator.
theWorldCommunicator =
new Comm
(/*size */ 1,
/*rank */ 0,
/*host */ "
* In processes participating in the new communicator, the new communicator
* is returned. The participating processes appear in the same order by rank
* in the new communicator as in this communicator. The process can call the
* new communicator's rank() method to determine the process's rank
* in the new communicator.
*
* In processes not participating in the new communicator, null is returned.
*
* @param participate True if this process will participate in the new
* communicator; false otherwise.
*
* @return New communicator if this process will participate in the new
* communicator; null otherwise.
*
* @exception IOException
* Thrown if an I/O error occurred.
*/
public Comm createComm
(boolean participate)
throws IOException
{
return createComm (participate, 0);
}
/**
* Create a new communicator. Every process in this communicator must
* call the createComm() method. Each process passes true or false
* for the participate argument to state whether the process will
* participate in the new communicator. At least one process must
* participate in the new communicator. Messages to set up the new
* communicator are sent to all processes in this communicator, using the
* given message tag.
*
* In processes participating in the new communicator, the new communicator
* is returned. The participating processes appear in the same order by rank
* in the new communicator as in this communicator. The process can call the
* new communicator's rank() method to determine the process's rank
* in the new communicator.
*
* In processes not participating in the new communicator, null is returned.
*
* @param participate True if this process will participate in the new
* communicator; false otherwise.
* @param tag Message tag.
*
* @return New communicator if this process will participate in the new
* communicator; null otherwise.
*
* @exception IOException
* Thrown if an I/O error occurred.
*/
public Comm createComm
(boolean participate,
int tag)
throws IOException
{
// Set up array of socket addresses for all processes.
InetSocketAddress[] address = new InetSocketAddress [mySize];
ObjectBuf
* A process can send a message to itself; in this case a different thread
* must call the receive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param buffer Buffer of data items to be sent.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void send
(int toRank,
Buf buffer)
throws IOException
{
send (toRank, 0, buffer);
}
/**
* Send a message to the process at the given rank in this communicator with
* the given message tag. The message items come from the given buffer. To
* receive the message, the destination process must call the
* receive() method. When the send() method returns, the
* message has been fully sent, but it may not yet have been fully received.
*
* A process can send a message to itself; in this case a different thread
* must call the receive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param tag Message tag.
* @param buffer Buffer of data items to be sent.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void send
(int toRank,
int tag,
Buf buffer)
throws IOException
{
myChannelGroup.send (getChannel (toRank), tag, buffer);
}
/**
* Send a message to the process at the given rank in this communicator
* (non-blocking). A message tag of 0 is used. The message items come from
* the given buffer. To receive the message, the destination process must
* call the receive() method.
*
* The send() method initiates the send operation and immediately
* returns a {@linkplain CommRequest} object. The send operation is
* performed by a separate thread. To wait for the send operation to finish,
* call the returned {@linkplain CommRequest} object's
* waitForFinish() method. When that method returns, the message
* has been fully sent, but it may not yet have been fully received.
*
* A process can send a message to itself; in this case a different thread
* must call the receive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param buffer Buffer of data items to be sent.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest send
(int toRank,
Buf buffer,
CommRequest request)
throws IOException
{
return send (toRank, 0, buffer, request);
}
/**
* Send a message to the process at the given rank in this communicator with
* the given message tag (non-blocking). The message items come from the
* given buffer. To receive the message, the destination process must call
* the receive() method.
*
* The send() method initiates the send operation and immediately
* returns a {@linkplain CommRequest} object. The send operation is
* performed by a separate thread. To wait for the send operation to finish,
* call the returned {@linkplain CommRequest} object's
* waitForFinish() method. When that method returns, the message
* has been fully sent, but it may not yet have been fully received.
*
* A process can send a message to itself; in this case a different thread
* must call the receive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param tag Message tag.
* @param buffer Buffer of data items to be sent.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest send
(int toRank,
int tag,
Buf buffer,
CommRequest request)
throws IOException
{
// Set up CommRequest object.
CommRequest req = request == null ? new CommRequest() : request;
// Send message (non-blocking).
req.mySendRequest = new IORequest();
req.myRecvRequest = null;
myChannelGroup.sendNoWait
(getChannel (toRank), tag, buffer, req.mySendRequest);
// Return CommRequest object.
return req;
}
/**
* Receive a message from the process at the given rank in this
* communicator. If rank is null, a message will be received from
* any process in this communicator. The message must have a tag of 0. The
* received message items are stored in the given buffer. To send the
* message, the source process must call the send() method. When
* the receive() method returns, the message has been fully
* received.
*
* A {@linkplain CommStatus} object is returned. The status object gives the
* actual rank of the process that sent the message, the actual message tag
* that was received, and the actual number of data items in the message. If
* the actual number of data items in the message is less than the length of
* the buffer, nothing is stored into the extra data items at the end of the
* buffer. If the actual number of data items in the message is greater than
* the length of the buffer, the extra data items at the end of the message
* are discarded.
*
* A process can receive a message from itself; in this case a different
* thread must call the send() method on this communicator.
*
* @param fromRank Source process's rank in this communicator, or null to
* receive from any process.
* @param buffer Buffer of data items to be received.
*
* @return Status object giving the outcome of the message reception.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if fromRank is not null and is
* not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommStatus receive
(Integer fromRank,
Buf buffer)
throws IOException
{
return receive (fromRank, 0, buffer);
}
/**
* Receive a message from the process at the given rank in this communicator
* with the given message tag. If rank is null, a message will be
* received from any process in this communicator. The received message
* items are stored in the given buffer. To send the message, the source
* process must call the send() method. When the receive()
* method returns, the message has been fully received.
*
* A {@linkplain CommStatus} object is returned. The status object gives the
* actual rank of the process that sent the message, the actual message tag
* that was received, and the actual number of data items in the message. If
* the actual number of data items in the message is less than the length of
* the buffer, nothing is stored into the extra data items at the end of the
* buffer. If the actual number of data items in the message is greater than
* the length of the buffer, the extra data items at the end of the message
* are discarded.
*
* A process can receive a message from itself; in this case a different
* thread must call the send() method on this communicator.
*
* @param fromRank Source process's rank in this communicator, or null to
* receive from any process.
* @param tag Message tag.
* @param buffer Buffer of data items to be received.
*
* @return Status object giving the outcome of the message reception.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if fromRank is not null and is
* not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommStatus receive
(Integer fromRank,
int tag,
Buf buffer)
throws IOException
{
Status status;
// If source is a wildcard, ensure a channel to every process, then
// receive from any process.
if (fromRank == null)
{
for (int src = 0; src < mySize; ++ src) ensureChannel (src);
status = myChannelGroup.receive (null, tag, buffer);
}
// If source is not a wildcard, receive from that process.
else
{
status =
myChannelGroup.receive (getChannel (fromRank), tag, buffer);
}
// Return CommStatus object.
return new CommStatus
(getFarRank (status.channel),
status.tag,
status.length);
}
/**
* Receive a message from the process at the given rank in this communicator
* with the given message tag range. If rank is null, a message
* will be received from any process in this communicator. If
* tagRange is null, a message will be received with any tag. If
* tagRange is not null, a message will be received with any tag in
* the given range. The received message items are stored in the given
* buffer. To send the message, the source process must call the
* send() method. When the receive() method returns, the
* message has been fully received.
*
* A {@linkplain CommStatus} object is returned. The status object gives the
* actual rank of the process that sent the message, the actual message tag
* that was received, and the actual number of data items in the message. If
* the actual number of data items in the message is less than the length of
* the buffer, nothing is stored into the extra data items at the end of the
* buffer. If the actual number of data items in the message is greater than
* the length of the buffer, the extra data items at the end of the message
* are discarded.
*
* A process can receive a message from itself; in this case a different
* thread must call the send() method on this communicator.
*
* @param fromRank Source process's rank in this communicator, or null to
* receive from any process.
* @param tagRange Message tag range, or null to receive any tag.
* @param buffer Buffer of data items to be received.
*
* @return Status object giving the outcome of the message reception.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if fromRank is not null and is
* not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommStatus receive
(Integer fromRank,
Range tagRange,
Buf buffer)
throws IOException
{
Status status;
// If source is a wildcard, ensure a channel to every process, then
// receive from any process.
if (fromRank == null)
{
for (int src = 0; src < mySize; ++ src) ensureChannel (src);
status = myChannelGroup.receive (null, tagRange, buffer);
}
// If source is not a wildcard, receive from that process.
else
{
status = myChannelGroup.receive
(getChannel (fromRank), tagRange, buffer);
}
// Return CommStatus object.
return new CommStatus
(getFarRank (status.channel),
status.tag,
status.length);
}
/**
* Receive a message from the process at the given rank in this communicator
* (non-blocking). If rank is null, a message will be received from
* any process in this communicator. The message must have a tag of 0. The
* received message items are stored in the given buffer. To send the
* message, the source process must call the send() method.
*
* The receive() method initiates the receive operation and
* immediately returns a {@linkplain CommRequest} object. The receive
* operation is performed by a separate thread. To wait for the receive
* operation to finish, call the returned {@linkplain CommRequest} object's
* waitForFinish() method. When that method returns, the incoming
* message items have been fully received.
*
* A process can receive a message from itself; in this case a different
* thread must call the send() method on this communicator.
*
* @param fromRank Source process's rank in this communicator, or null to
* receive from any process.
* @param buffer Buffer of data items to be received.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if fromRank is not null and is
* not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest receive
(Integer fromRank,
Buf buffer,
CommRequest request)
throws IOException
{
return receive (fromRank, 0, buffer, request);
}
/**
* Receive a message from the process at the given rank in this communicator
* with the given message tag (non-blocking). If rank is null, a
* message will be received from any process in this communicator. If
* tag is null, a message will be received with any tag. The
* received message items are stored in the given buffer. To send the
* message, the source process must call the send() method.
*
* The receive() method initiates the receive operation and
* immediately returns a {@linkplain CommRequest} object. The receive
* operation is performed by a separate thread. To wait for the receive
* operation to finish, call the returned {@linkplain CommRequest} object's
* waitForFinish() method. When that method returns, the incoming
* message items have been fully received.
*
* A process can receive a message from itself; in this case a different
* thread must call the send() method on this communicator.
*
* @param fromRank Source process's rank in this communicator, or null to
* receive from any process.
* @param tag Message tag.
* @param buffer Buffer of data items to be received.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if fromRank is not null and is
* not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest receive
(Integer fromRank,
int tag,
Buf buffer,
CommRequest request)
throws IOException
{
// Set up CommRequest object.
CommRequest req = request == null ? new CommRequest() : request;
req.mySendRequest = null;
req.myRecvRequest = new IORequest();
// If source is a wildcard, ensure a channel to every process, then
// receive (non-blocking) from any process.
if (fromRank == null)
{
for (int src = 0; src < mySize; ++ src) ensureChannel (src);
myChannelGroup.receiveNoWait
(null, tag, buffer, req.myRecvRequest);
}
// If source is not a wildcard, receive (non-blocking) from that
// process.
else
{
myChannelGroup.receiveNoWait
(getChannel (fromRank), tag, buffer, req.myRecvRequest);
}
// Return CommRequest object.
return req;
}
/**
* Receive a message from the process at the given rank in this communicator
* with the given message tag range (non-blocking). If rank is
* null, a message will be received from any process in this communicator.
* If tagRange is null, a message will be received with any tag. If
* tagRange is not null, a message will be received with any tag in
* the given range. The received message items are stored in the given
* buffer. To send the message, the source process must call the
* send() method.
*
* The receive() method initiates the receive operation and
* immediately returns a {@linkplain CommRequest} object. The receive
* operation is performed by a separate thread. To wait for the receive
* operation to finish, call the returned {@linkplain CommRequest} object's
* waitForFinish() method. When that method returns, the incoming
* message items have been fully received.
*
* A process can receive a message from itself; in this case a different
* thread must call the send() method on this communicator.
*
* @param fromRank Source process's rank in this communicator, or null to
* receive from any process.
* @param tagRange Message tag range, or null to receive any tag.
* @param buffer Buffer of data items to be received.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if fromRank is not null and is
* not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest receive
(Integer fromRank,
Range tagRange,
Buf buffer,
CommRequest request)
throws IOException
{
// Set up CommRequest object.
CommRequest req = request == null ? new CommRequest() : request;
req.mySendRequest = null;
req.myRecvRequest = new IORequest();
// If source is a wildcard, ensure a channel to every process, then
// receive (non-blocking) from any process.
if (fromRank == null)
{
for (int src = 0; src < mySize; ++ src) ensureChannel (src);
myChannelGroup.receiveNoWait
(null, tagRange, buffer, req.myRecvRequest);
}
// If source is not a wildcard, receive (non-blocking) from that
// process.
else
{
myChannelGroup.receiveNoWait
(getChannel (fromRank), tagRange, buffer, req.myRecvRequest);
}
// Return CommRequest object.
return req;
}
/**
* Send a message to the process at the given rank in this communicator, and
* receive a message from the process at the given rank in this
* communicator. A message tag of 0 is used. The outgoing message items come
* from the buffer sendbuf. The incoming message items go into the
* buffer recvbuf. The outgoing message items must come from a
* different place than where the incoming message items will be stored. The
* destination process (process toRank) must call a method to
* receive this process's outgoing message items. The source process
* (process fromRank) must call a method to send this process's
* incoming message items. When the sendReceive() method returns,
* the outgoing message items have been fully sent, but they may not yet
* have been fully received; and the incoming message items have been fully
* received.
*
* A {@linkplain CommStatus} object is returned giving the results of the
* receive half of the operation. The status object gives the rank of the
* process that sent the incoming message, the message tag that was
* received, and the actual number of data items in the message. If the
* actual number of data items in the message is less than the length of the
* receive buffer, nothing is stored into the extra data items at the end of
* the receive buffer. If the actual number of data items in the message is
* greater than the length of the receive buffer, the extra data items at
* the end of the message are discarded.
*
* A process can send-receive messages with itself; in this case a different
* thread must call the sendReceive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param sendBuf Buffer of data items to be sent.
* @param fromRank Source process's rank in this communicator.
* @param recvBuf Buffer of data items to be received.
*
* @return Status object giving the outcome of the message reception.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank or fromRank
* is not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if sendBuf or recvBuf
* is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommStatus sendReceive
(int toRank,
Buf sendBuf,
int fromRank,
Buf recvBuf)
throws IOException
{
return sendReceive (toRank, 0, sendBuf, fromRank, 0, recvBuf);
}
/**
* Send a message to the process at the given rank in this communicator with
* the given message tag, and receive a message from the process at the
* given rank in this communicator with the given message tag. The outgoing
* message items come from the buffer sendbuf. The incoming message
* items go into the buffer recvbuf. The outgoing message items
* must come from a different place than where the incoming message items
* will be stored. The destination process (process toRank) must
* call a method to receive this process's outgoing message items. The
* source process (process fromRank) must call a method to send
* this process's incoming message items. When the sendReceive()
* method returns, the outgoing message items have been fully sent, but they
* may not yet have been fully received; and the incoming message items have
* been fully received.
*
* A {@linkplain CommStatus} object is returned giving the results of the
* receive half of the operation. The status object gives the rank of the
* process that sent the incoming message, the message tag that was
* received, and the actual number of data items in the message. If the
* actual number of data items in the message is less than the length of the
* receive buffer, nothing is stored into the extra data items at the end of
* the receive buffer. If the actual number of data items in the message is
* greater than the length of the receive buffer, the extra data items at
* the end of the message are discarded.
*
* A process can send-receive messages with itself; in this case a different
* thread must call the sendReceive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param sendTag Message tag for outgoing message.
* @param sendBuf Buffer of data items to be sent.
* @param fromRank Source process's rank in this communicator.
* @param recvTag Message tag for incoming message.
* @param recvBuf Buffer of data items to be received.
*
* @return Status object giving the outcome of the message reception.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank or fromRank
* is not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if sendBuf or recvBuf
* is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommStatus sendReceive
(int toRank,
int sendTag,
Buf sendBuf,
int fromRank,
int recvTag,
Buf recvBuf)
throws IOException
{
// Send the outgoing message (non-blocking).
IORequest sendRequest = new IORequest();
myChannelGroup.sendNoWait
(getChannel (toRank), sendTag, sendBuf, sendRequest);
// Receive the outgoing message (non-blocking).
IORequest recvRequest = new IORequest();
myChannelGroup.receiveNoWait
(getChannel (fromRank), recvTag, recvBuf, recvRequest);
// Wait for both messages to finish.
sendRequest.waitForFinish();
Status status = recvRequest.waitForFinish();
return new CommStatus
(getFarRank (status.channel),
status.tag,
status.length);
}
/**
* Send a message to the process at the given rank in this communicator, and
* receive a message from the process at the given rank in this communicator
* (non-blocking). A message tag of 0 is used. The outgoing message items
* come from the buffer sendbuf. The incoming message items go into
* the buffer recvbuf. The outgoing message items must come from a
* different place than where the incoming message items will be stored. The
* destination process (process toRank) must call a method to
* receive this process's outgoing message items. The source process
* (process fromRank) must call a method to send this process's
* incoming message items.
*
* The sendReceive() method initiates the send and receive
* operations and immediately returns a {@linkplain CommRequest} object. The
* send and receive operations are performed by a separate thread. To wait
* for the send and receive operations to finish, call the returned
* {@linkplain CommRequest} object's waitForFinish() method. When
* that method returns, the outgoing message items have been fully sent, but
* they may not yet have been fully received; and the incoming message items
* have been fully received.
*
* A process can send-receive messages with itself; in this case a different
* thread must call the sendReceive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param sendBuf Buffer of data items to be sent.
* @param fromRank Source process's rank in this communicator.
* @param recvBuf Buffer of data items to be received.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank or fromRank
* is not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if sendBuf or recvBuf
* is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest sendReceive
(int toRank,
Buf sendBuf,
int fromRank,
Buf recvBuf,
CommRequest request)
throws IOException
{
return sendReceive (toRank, 0, sendBuf, fromRank, 0, recvBuf, request);
}
/**
* Send a message to the process at the given rank in this communicator with
* the given message tag, and receive a message from the process at the
* given rank in this communicator with the given message tag
* (non-blocking). The outgoing message items come from the buffer
* sendbuf. The incoming message items go into the buffer
* recvbuf. The outgoing message items must come from a different
* place than where the incoming message items will be stored. The
* destination process (process toRank) must call a method to
* receive this process's outgoing message items. The source process
* (process fromRank) must call a method to send this process's
* incoming message items.
*
* The sendReceive() method initiates the send and receive
* operations and immediately returns a {@linkplain CommRequest} object. The
* send and receive operations are performed by a separate thread. To wait
* for the send and receive operations to finish, call the returned
* {@linkplain CommRequest} object's waitForFinish() method. When
* that method returns, the outgoing message items have been fully sent, but
* they may not yet have been fully received; and the incoming message items
* have been fully received.
*
* A process can send-receive messages with itself; in this case a different
* thread must call the sendReceive() method on this communicator.
*
* @param toRank Destination process's rank in this communicator.
* @param sendTag Message tag for outgoing message.
* @param sendBuf Buffer of data items to be sent.
* @param fromRank Source process's rank in this communicator.
* @param recvTag Message tag for incoming message.
* @param recvBuf Buffer of data items to be received.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if toRank or fromRank
* is not in the range 0 .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if sendBuf or recvBuf
* is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest sendReceive
(int toRank,
int sendTag,
Buf sendBuf,
int fromRank,
int recvTag,
Buf recvBuf,
CommRequest request)
throws IOException
{
// Set up CommRequest object.
CommRequest req = request == null ? new CommRequest() : request;
// Send the outgoing message (non-blocking).
req.mySendRequest = new IORequest();
myChannelGroup.sendNoWait
(getChannel (toRank), sendTag, sendBuf, req.mySendRequest);
// Receive the outgoing message (non-blocking).
req.myRecvRequest = new IORequest();
myChannelGroup.receiveNoWait
(getChannel (fromRank), recvTag, recvBuf, req.myRecvRequest);
// Return CommRequest object.
return req;
}
/**
* Flood-send a message to all processes in this communicator. The message
* uses a tag of 0. The message items come from the given buffer. To receive
* the message, every process (including the sending process) must call the
* floodReceive() method. When the floodSend() method
* returns, the message has been fully sent, but it may not yet have been
* fully received in all processes.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param buffer Buffer of data items to be sent.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void floodSend
(Buf buffer)
throws IOException
{
floodSend (0, buffer, null) .waitForFinish();
}
/**
* Flood-send a message to all processes in this communicator with the given
* message tag. The message items come from the given buffer. To receive the
* message, every process (including the sending process) must call the
* floodReceive() method. When the floodSend() method
* returns, the message has been fully sent, but it may not yet have been
* fully received in all processes.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param tag Message tag.
* @param buffer Buffer of data items to be sent.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void floodSend
(int tag,
Buf buffer)
throws IOException
{
floodSend (tag, buffer, null) .waitForFinish();
}
/**
* Flood-send a message to all processes in this communicator
* (non-blocking). A message tag of 0 is used. The message items come from
* the given buffer. To receive the message, every process (including the
* sending process) must call the floodReceive() method.
*
* The floodSend() method initiates the flood-send operation and
* immediately returns a {@linkplain CommRequest} object. The flood-send
* operation is performed by a separate thread. To wait for the flood-send
* operation to finish, call the returned {@linkplain CommRequest} object's
* waitForFinish() method. When that method returns, the message
* has been fully sent, but it may not yet have been fully received in all
* processes.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param buffer Buffer of data items to be sent.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest floodSend
(Buf buffer,
CommRequest request)
throws IOException
{
return floodSend (0, buffer, request);
}
/**
* Flood-send a message to all processes in this communicator with the given
* message tag (non-blocking). The message items come from the given buffer.
* To receive the message, every process (including the sending process)
* must call the floodReceive() method.
*
* The floodSend() method initiates the flood-send operation and
* immediately returns a {@linkplain CommRequest} object. The flood-send
* operation is performed by a separate thread. To wait for the flood-send
* operation to finish, call the returned {@linkplain CommRequest} object's
* waitForFinish() method. When that method returns, the message
* has been fully sent, but it may not yet have been fully received in all
* processes.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param tag Message tag.
* @param buffer Buffer of data items to be sent.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest floodSend
(int tag,
Buf buffer,
CommRequest request)
throws IOException
{
// Set up CommRequest object.
CommRequest req = request == null ? new CommRequest() : request;
req.mySendRequest = new IORequest();
req.myRecvRequest = null;
// Send data to process 0. Process 0's flood-receive I/O request object
// will forward the data to all the processes.
myChannelGroup.sendNoWait
(getChannel (0), tag, buffer, req.mySendRequest);
// Return CommRequest object.
return req;
}
/**
* Flood-receive a message from any process in this communicator. The
* message must have a tag of 0. The received message items are stored in
* the given buffer. To send the message, the source process must call the
* floodSend() method. When the floodReceive() method
* returns, the message has been fully received.
*
* A {@linkplain CommStatus} object is returned. The status object gives the
* actual rank of the process that sent the message, the actual message tag
* that was received, and the actual number of data items in the message.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param buffer Buffer of data items to be received.
*
* @return Status object giving the outcome of the message reception.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommStatus floodReceive
(Buf buffer)
throws IOException
{
return floodReceive (0, buffer, null) .waitForFinish();
}
/**
* Flood-receive a message from any process in this communicator with the
* given message tag. If tag is null, a message will be received
* with any tag. The received message items are stored in the given buffer.
* To send the message, the source process must call the
* floodSend() method. When the floodReceive() method
* returns, the message has been fully received.
*
* A {@linkplain CommStatus} object is returned. The status object gives the
* actual rank of the process that sent the message, the actual message tag
* that was received, and the actual number of data items in the message.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param tag Message tag, or null to receive any tag.
* @param buffer Buffer of data items to be received.
*
* @return Status object giving the outcome of the message reception.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommStatus floodReceive
(Integer tag,
Buf buffer)
throws IOException
{
return floodReceive (tag, buffer, null) .waitForFinish();
}
/**
* Flood-receive a message from any process in this communicator
* (non-blocking). A message tag of 0 is used. If tag is null, a
* message will be received with any tag. The received message items are
* stored in the given buffer. To send the message, the source process must
* call the floodSend() method.
*
* The floodReceive() method initiates the flood-receive operation
* and immediately returns a {@linkplain CommRequest} object. The
* flood-receive operation is performed by a separate thread. To wait for
* the flood-receive operation to finish, call the returned {@linkplain
* CommRequest} object's waitForFinish() method. When that method
* returns, the incoming message items have been fully received.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param buffer Buffer of data items to be received.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest floodReceive
(Buf buffer,
CommRequest request)
throws IOException
{
return floodReceive (0, buffer, request);
}
/**
* Flood-receive a message from any process in this communicator with the
* given message tag (non-blocking). If tag is null, a message will
* be received with any tag. The received message items are stored in the
* given buffer. To send the message, the source process must call the
* floodSend() method.
*
* The floodReceive() method initiates the flood-receive operation
* and immediately returns a {@linkplain CommRequest} object. The
* flood-receive operation is performed by a separate thread. To wait for
* the flood-receive operation to finish, call the returned {@linkplain
* CommRequest} object's waitForFinish() method. When that method
* returns, the incoming message items have been fully received.
*
* Note: The length of the incoming buffer in the
* floodReceive() method call must be the same as the length of the
* outgoing buffer in the floodSend() method call.
*
* @param tag Message tag, or null to receive any tag.
* @param buffer Buffer of data items to be received.
* @param request CommRequest object to use to wait for the operation to
* finish; in this case request is returned. If
* request is null, a new CommRequest object is
* created and returned.
*
* @return CommRequest object to use to wait for the operation to finish.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public CommRequest floodReceive
(Integer tag,
Buf buffer,
CommRequest request)
throws IOException
{
// Get broadcast tree for root=0.
int[] tree = getBroadcastTree (0);
// Set up CommRequest object with a special I/O request object that
// forwards the message down the broadcast tree.
CommRequest req = request == null ? new CommRequest() : request;
req.mySendRequest = null;
req.myRecvRequest = new FloodReceiveIORequest (tree);
// In process 0, ensure a channel to every process, then receive
// (non-blocking) a message from any process.
if (myRank == 0)
{
for (int src = 0; src < mySize; ++ src) ensureChannel (src);
myChannelGroup.receiveNoWait
(null, tag, buffer, req.myRecvRequest);
}
// In other processes, ensure a channel to the child processes in the
// broadcast tree, then receive (non-blocking) a message from the parent
// process in the broadcast tree.
else
{
for (int i = 1; i < tree.length; ++ i) ensureChannel (tree[i]);
myChannelGroup.receiveNoWait
(getChannel (tree[0]), tag, buffer, req.myRecvRequest);
}
// Return CommRequest object.
return req;
}
/**
* Class FloodReceiveIORequest overrides the methods of class IORequest with
* additional processing to forward the message when a message is received.
*/
private class FloodReceiveIORequest
extends IORequest
{
// Broadcast tree.
private int[] tree;
// List of zero or more additional I/O requests to forward copies of the
// received message.
private LinkedList
* The root process (the process whose rank in this communicator is
* root) sends the message items. The message items come from the
* given buffer. When the broadcast() method returns, the message
* has been fully sent, but it may not yet have been fully received by all
* processes.
*
* Each non-root process receives the message items. The message items are
* stored in the given buffer. When the broadcast() method returns,
* the message has been fully received.
*
* @param root Root process's rank in this communicator.
* @param buffer Buffer of data items to be sent (root process) or
* received (non-root processes).
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void broadcast
(int root,
Buf buffer)
throws IOException
{
broadcast (root, 0, buffer);
}
/**
* Broadcast a message to all processes in this communicator using the given
* message tag. All processes must call broadcast() with the same
* values for root and tag and with a buffer of the same
* length and the same item data type.
*
* The root process (the process whose rank in this communicator is
* root) sends the message items. The message items come from the
* given buffer. When the broadcast() method returns, the message
* has been fully sent, but it may not yet have been fully received by all
* processes.
*
* Each non-root process receives the message items. The message items are
* stored in the given buffer. When the broadcast() method returns,
* the message has been fully received.
*
* @param root Root process's rank in this communicator.
* @param tag Message tag.
* @param buffer Buffer of data items to be sent (root process) or
* received (non-root processes).
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buffer is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void broadcast
(int root,
int tag,
Buf buffer)
throws IOException
{
// Verify preconditions.
if (0 > root || root >= mySize)
{
throw new IndexOutOfBoundsException
("Comm.broadcast(): root = " + root + " out of bounds");
}
// Early return if only one process.
if (mySize == 1) return;
// A broadcast is done as a series of point-to-point messages. The
// messages are organized into rounds. The number of rounds is
// ceil(log_2(mySize)). In each round, processes send messages to other
// processes in parallel. Here is the message pattern for a communicator
// with 8 processes doing a broadcast from root process 0:
//
// Process
// 0 1 2 3 4 5 6 7
// | | | | | | | |
// |---->| | | | | | | Round 1
// | | | | | | | |
// |---------->| | | | | | Round 2
// | |---------->| | | | |
// | | | | | | | |
// |---------------------->| | | | Round 3
// | |---------------------->| | |
// | | |---------------------->| |
// | | | |---------------------->|
// | | | | | | | |
//
// If a process other than process 0 is the root, the message pattern is
// the same, except the process ranks are circularly rotated.
// Get array of process ranks in the broadcast tree.
int[] broadcasttree = getBroadcastTree (root);
int n = broadcasttree.length;
// Receive data from parent if any (blocking).
int parent = broadcasttree[0];
if (parent != -1)
{
myChannelGroup.receive (getChannel (parent), tag, buffer);
}
// Send data to children if any (non-blocking).
IORequest[] iorequest = new IORequest [n];
for (int i = 1; i < n; ++ i)
{
int child = broadcasttree[i];
iorequest[i] = new IORequest();
myChannelGroup.sendNoWait
(getChannel (child), tag, buffer, iorequest[i]);
}
// Wait for sends to finish if any.
for (int i = 1; i < n; ++ i)
{
iorequest[i].waitForFinish();
}
}
/**
* Scatter messages to all processes in this communicator. The scatter
* uses a message tag of 0. All processes must call scatter()
* with the same value for root.
*
* The root process (the process whose rank in this communicator is
* root) sends the message items. The message items sent to process
* i come from the source buffer at index i in the given array
* of source buffers. When the scatter() method returns, the
* messages have been fully sent, but they may not yet have been fully
* received by all processes.
*
* Each process, including the root process, receives the message items. The
* message items are stored in the given destination buffer. This must have
* the same length and the same item data type as the corresponding source
* buffer. When the scatter() method returns, the message has been
* fully received.
*
* In the non-root processes, the source buffer array is ignored and may be
* null.
*
* @param root Root process's rank in this communicator.
* @param srcarray Array of source buffers to be sent by the root process.
* Ignored in the non-root processes.
* @param dst Destination buffer to be received.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1. Thrown if srcarray's length does not
* equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if srcarray or any element
* thereof is null. Thrown if dst is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void scatter
(int root,
Buf[] srcarray,
Buf dst)
throws IOException
{
scatter (root, 0, srcarray, dst);
}
/**
* Scatter messages to all processes in this communicator using the given
* message tag. All processes must call scatter() with the same
* values for root and tag.
*
* The root process (the process whose rank in this communicator is
* root) sends the message items. The message items sent to process
* i come from the source buffer at index i in the given array
* of source buffers. When the scatter() method returns, the
* messages have been fully sent, but they may not yet have been fully
* received by all processes.
*
* Each process, including the root process, receives the message items. The
* message items are stored in the given destination buffer. This must have
* the same length and the same item data type as the corresponding source
* buffer. When the scatter() method returns, the message has been
* fully received.
*
* In the non-root processes, the source buffer array is ignored and may be
* null.
*
* @param root Root process's rank in this communicator.
* @param tag Message tag.
* @param srcarray Array of source buffers to be sent by the root process.
* Ignored in the non-root processes.
* @param dst Destination buffer to be received.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1. Thrown if srcarray's length does not
* equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if srcarray or any element
* thereof is null. Thrown if dst is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void scatter
(int root,
int tag,
Buf[] srcarray,
Buf dst)
throws IOException
{
// Verify preconditions.
if (0 > root || root >= mySize)
{
throw new IndexOutOfBoundsException
("Comm.scatter(): root = " + root + " out of bounds");
}
// A scatter is done as a series of point-to-point messages. The root
// process sends a separate message to every other process. Here is the
// message pattern for a communicator with 8 processes scattering from
// root process 0:
//
// Process
// 0 1 2 3 4 5 6 7
// | | | | | | | |
// |---->| | | | | | |
// | | | | | | | |
// |---------->| | | | | |
// | | | | | | | |
// |---------------->| | | | |
// | | | | | | | |
// |---------------------->| | | |
// | | | | | | | |
// |---------------------------->| | |
// | | | | | | | |
// |---------------------------------->| |
// | | | | | | | |
// |---------------------------------------->|
// | | | | | | | |
// Root process sends all messages.
if (myRank == root)
{
// Array of IORequest objects for non-blocking sends.
IORequest[] iorequest = new IORequest [mySize];
// Initiate sends to lower-ranked processes.
for (int rank = 0; rank < myRank; ++ rank)
{
iorequest[rank] = new IORequest();
myChannelGroup.sendNoWait
(getChannel (rank), tag, srcarray[rank], iorequest[rank]);
}
// Initiate sends to higher-ranked processes.
for (int rank = myRank+1; rank < mySize; ++ rank)
{
iorequest[rank] = new IORequest();
myChannelGroup.sendNoWait
(getChannel (rank), tag, srcarray[rank], iorequest[rank]);
}
// Copy to itself.
dst.copy (srcarray[myRank]);
// Wait for completion of sends to lower-ranked processes.
for (int rank = 0; rank < myRank; ++ rank)
{
iorequest[rank].waitForFinish();
}
// Wait for completion of sends to higher-ranked processes.
for (int rank = myRank+1; rank < mySize; ++ rank)
{
iorequest[rank].waitForFinish();
}
}
// Non-root process receives one message.
else
{
myChannelGroup.receive (getChannel (root), tag, dst);
}
}
/**
* Gather messages from all processes in this communicator. The gather uses
* a message tag of 0. All processes must call gather() with the
* same value for root.
*
* The root process (the process whose rank in this communicator is
* root) receives the message items. The message items received
* from process i are stored in the destination buffer at index
* i in the given array of destination buffers. When the
* gather() method returns, all the messages have been fully
* received.
*
* Each process, including the root process, sends the message items. The
* message items come from the given source buffer. This must have the same
* length and the same item data type as the corresponding destination
* buffer. When the gather() method returns, the message has been
* fully sent, but it may not yet have been fully received by the root
* process.
*
* In the non-root processes, the destination buffer array is ignored and
* may be null.
*
* @param root Root process's rank in this communicator.
* @param src Source buffer to be sent.
* @param dstarray Array of destination buffers to be received by the root
* process. Ignored in the non-root processes.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1. Thrown if dstarray's length does not
* equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if dstarray or any element
* thereof is null. Thrown if src is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void gather
(int root,
Buf src,
Buf[] dstarray)
throws IOException
{
gather (root, 0, src, dstarray);
}
/**
* Gather messages from all processes in this communicator using the given
* message tag. All processes must call gather() with the same
* values for root and tag.
*
* The root process (the process whose rank in this communicator is
* root) receives the message items. The message items received
* from process i are stored in the destination buffer at index
* i in the given array of destination buffers. When the
* gather() method returns, all the messages have been fully
* received.
*
* Each process, including the root process, sends the message items. The
* message items come from the given source buffer. This must have the same
* length and the same item data type as the corresponding destination
* buffer. When the gather() method returns, the message has been
* fully sent, but it may not yet have been fully received by the root
* process.
*
* In the non-root processes, the destination buffer array is ignored and
* may be null.
*
* @param root Root process's rank in this communicator.
* @param tag Message tag.
* @param src Source buffer to be sent.
* @param dstarray Array of destination buffers to be received by the root
* process. Ignored in the non-root processes.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1. Thrown if dstarray's length does not
* equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if dstarray or any element
* thereof is null. Thrown if src is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void gather
(int root,
int tag,
Buf src,
Buf[] dstarray)
throws IOException
{
// Verify preconditions.
if (0 > root || root >= mySize)
{
throw new IndexOutOfBoundsException
("Comm.gather(): root = " + root + " out of bounds");
}
// A gather is done as a series of point-to-point messages. The root
// process receives a separate message from every other process. Here is
// the message pattern for a communicator with 8 processes gathering
// into root process 0:
//
// Process
// 0 1 2 3 4 5 6 7
// | | | | | | | |
// |<----| | | | | | |
// | | | | | | | |
// |<----------| | | | | |
// | | | | | | | |
// |<----------------| | | | |
// | | | | | | | |
// |<----------------------| | | |
// | | | | | | | |
// |<----------------------------| | |
// | | | | | | | |
// |<----------------------------------| |
// | | | | | | | |
// |<----------------------------------------|
// | | | | | | | |
// Root process receives all messages.
if (myRank == root)
{
// Array of IORequest objects for non-blocking receives.
IORequest[] iorequest = new IORequest [mySize];
// Initiate receives from lower-ranked processes.
for (int rank = 0; rank < myRank; ++ rank)
{
iorequest[rank] = new IORequest();
myChannelGroup.receiveNoWait
(getChannel (rank), tag, dstarray[rank], iorequest[rank]);
}
// Initiate receives from higher-ranked processes.
for (int rank = myRank+1; rank < mySize; ++ rank)
{
iorequest[rank] = new IORequest();
myChannelGroup.receiveNoWait
(getChannel (rank), tag, dstarray[rank], iorequest[rank]);
}
// Copy to itself.
dstarray[myRank].copy (src);
// Wait for completion of receives from lower-ranked processes.
for (int rank = 0; rank < myRank; ++ rank)
{
iorequest[rank].waitForFinish();
}
// Wait for completion of receives from higher-ranked processes.
for (int rank = myRank+1; rank < mySize; ++ rank)
{
iorequest[rank].waitForFinish();
}
}
// Non-root process sends one message.
else
{
myChannelGroup.send (getChannel (root), tag, src);
}
}
/**
* All-gather messages from each process to all processes in this
* communicator. A message tag of 0 is used. All processes must call
* allGather().
*
* Each process sends the message items in the given source buffer. When the
* allGather() method returns, the source buffer has been fully
* sent.
*
* Each process receives message items from the other processes. The message
* items received from process i are stored in the destination buffer
* at index i in the given array of destination buffers. This
* destination buffer must have the same length and the same item data type
* as the source buffer in process i. When the allGather()
* method returns, all the destination buffers have been fully received.
*
* All-gather is the same as gather, except that every process has an array
* of destination buffers, and every process receives the results of the
* gather.
*
* @param src Source buffer to be sent.
* @param dstarray Array of destination buffers to be received.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if dstarray's length does not
* equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if dstarray or any element
* thereof is null. Thrown if src is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void allGather
(Buf src,
Buf[] dstarray)
throws IOException
{
allGather (0, src, dstarray);
}
/**
* All-gather messages from each process to all processes in this
* communicator using the given message tag. All processes must call
* allGather() with the same value for tag.
*
* Each process sends the message items in the given source buffer. When the
* allGather() method returns, the source buffer has been fully
* sent.
*
* Each process receives message items from the other processes. The message
* items received from process i are stored in the destination buffer
* at index i in the given array of destination buffers. This
* destination buffer must have the same length and the same item data type
* as the source buffer in process i. When the allGather()
* method returns, all the destination buffers have been fully received.
*
* All-gather is the same as gather, except that every process has an array
* of destination buffers, and every process receives the results of the
* gather.
*
* @param tag Message tag.
* @param src Source buffer to be sent.
* @param dstarray Array of destination buffers to be received.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if dstarray's length does not
* equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if dstarray or any element
* thereof is null. Thrown if src is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void allGather
(int tag,
Buf src,
Buf[] dstarray)
throws IOException
{
// Get ranks of predecessor and successor processes.
int pred = (myRank - 1 + mySize) % mySize;
int succ = (myRank + 1) % mySize;
// Copy source buffer into destination buffer at my own rank.
dstarray[myRank].copy (src);
// Do (mySize-1) message rounds. Messages are sent in a pipelined
// fashion from each process to its predecessor until each process's
// source data has arrived in every process. Each outgoing message is
// overlapped with an incoming message.
for (int i = 1; i < mySize; ++ i)
{
sendReceive
(/*toRank */ pred,
/*sendTag */ tag,
/*sendBuf */ dstarray[(myRank+i-1) % mySize],
/*fromRank*/ succ,
/*recvTag */ tag,
/*recvBuf */ dstarray[(myRank+i) % mySize]);
}
}
/**
* Perform a reduction on all processes in this communicator. The reduction
* uses a message tag of 0. All processes must call reduce() with
* the same value for root, with a buffer of the same length and
* the same item data type, and with the same binary operation (class
* {@linkplain edu.rit.pj.reduction.Op Op}).
*
* Before calling reduce(), each process has a buffer filled with
* data items. After reduce() returns, each data item in the root
* process's buffer has been set to the reduction of the
* corresponding data items in all the processes' buffers. The reduction is
* calculated by this formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process rank 0, 1, 2, and so on. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* In the root process, the reduce operation always changes the buffer's
* contents as described above. In the non-root processes, the reduce
* operation may or may not change the buffer's contents; the final contents
* of the buffer in the non-root processes is not specified.
*
* When the reduce() method returns in the root process, the
* reduction has been fully performed as described above. When the
* reduce() method returns in a non-root process, the non-root
* process has sent all its data items into the reduction, but the reduction
* may not be fully complete in the root process yet.
*
* @param root Root process's rank in this communicator.
* @param buffer Buffer of data items to be reduced.
* @param op Binary operation.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void reduce
(int root,
Buf buffer,
Op op)
throws IOException
{
reduce (root, 0, buffer, op);
}
/**
* Perform a reduction on all processes in this communicator using the given
* message tag. All processes must call reduce() with the same
* value for root, with a buffer of the same length and the same
* item data type, and with the same binary operation (class {@linkplain
* edu.rit.pj.reduction.Op Op}).
*
* Before calling reduce(), each process has a buffer filled with
* data items. After reduce() returns, each data item in the root
* process's buffer has been set to the reduction of the
* corresponding data items in all the processes' buffers. The reduction is
* calculated by this formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process rank 0, 1, 2, and so on. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* In the root process, the reduce operation always changes the buffer's
* contents as described above. In the non-root processes, the reduce
* operation may or may not change the buffer's contents; the final contents
* of the buffer in the non-root processes is not specified.
*
* When the reduce() method returns in the root process, the
* reduction has been fully performed as described above. When the
* reduce() method returns in a non-root process, the non-root
* process has sent all its data items into the reduction, but the reduction
* may not be fully complete in the root process yet.
*
* @param root Root process's rank in this communicator.
* @param tag Message tag.
* @param buffer Buffer of data items to be reduced.
* @param op Binary operation.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if root is not in the range 0
* .. size()-1.
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void reduce
(int root,
int tag,
Buf buffer,
Op op)
throws IOException
{
// Verify preconditions.
if (0 > root || root >= mySize)
{
throw new IndexOutOfBoundsException
("Comm.reduce(): root = " + root + " out of bounds");
}
// Early return if only one process.
if (mySize == 1) return;
// A reduction is done as a series of point-to-point messages. The
// messages are organized into rounds. The number of rounds is
// ceil(log_2(mySize)). The message pattern is the reverse of the
// broadcast message pattern. In each round, processes receive messages
// from other processes and reduce the data items into their accumulator
// buffers in parallel. When a process has received all messages, it
// sends the reduced results on. Here is the message pattern for a
// communicator with 8 processes doing a reduction into root process 0:
//
// Process
// 0 1 2 3 4 5 6 7
// | | | | | | | |
// |<----------------------| | | | Round 1
// | |<----------------------| | |
// | | |<----------------------| |
// | | | |<----------------------|
// | | | | | | | |
// |<----------| | | | | | Round 2
// | |<----------| | | | |
// | | | | | | | |
// |<----| | | | | | | Round 3
// | | | | | | | |
//
// If a process other than process 0 is the root, the message pattern is
// the same, except the process ranks are circularly rotated.
// Get array of process ranks in the broadcast tree.
int[] broadcasttree = getBroadcastTree (root);
int n = broadcasttree.length;
// Set up reduction buffer on top of source buffer.
Buf reductionbuf = buffer.getReductionBuf (op);
// Receive data from children if any, one at a time in reverse order.
for (int i = n-1; i >= 1; -- i)
{
int child = broadcasttree[i];
myChannelGroup.receive
(getChannel (child), tag, reductionbuf);
}
// Send data to parent if any.
int parent = broadcasttree[0];
if (parent != -1)
{
myChannelGroup.send (getChannel (parent), tag, buffer);
}
}
/**
* Perform an all-reduce on all processes in this communicator. The
* all-reduce uses a message tag of 0. All processes must call
* allReduce() with a buffer of the same length and the same item
* data type, and with the same binary operation (class {@linkplain
* edu.rit.pj.reduction.Op Op}).
*
* Before calling allReduce(), each process has a buffer filled
* with data items. After allReduce() returns, each data item in
* the calling process's buffer has been set to the reduction of the
* corresponding data items in all the processes' buffers. The reduction is
* calculated by this formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process rank 0, 1, 2, and so on. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* The allReduce() method is similar to the reduce()
* method, except the results are stored in all the processes' buffers, not
* just the one root process's buffer.
*
* @param buffer Buffer of data items to be reduced.
* @param op Binary operation.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void allReduce
(Buf buffer,
Op op)
throws IOException
{
allReduce (0, buffer, op);
}
/**
* Perform an all-reduce on all processes in this communicator using the
* given message tag. All processes must call allReduce() with a
* buffer of the same length and the same item data type, and with the same
* binary operation (class {@linkplain edu.rit.pj.reduction.Op Op}).
*
* Before calling allReduce(), each process has a buffer filled
* with data items. After allReduce() returns, each data item in
* the calling process's buffer has been set to the reduction of the
* corresponding data items in all the processes' buffers. The reduction is
* calculated by this formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process rank 0, 1, 2, and so on. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* The allReduce() method is similar to the reduce()
* method, except the results are stored in all the processes' buffers, not
* just the one root process's buffer.
*
* @param tag Message tag.
* @param buffer Buffer of data items to be reduced.
* @param op Binary operation.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void allReduce
(int tag,
Buf buffer,
Op op)
throws IOException
{
// An all-reduce is done using a "butterfly" message passing pattern.
// Consider the case of K=8 processes. In the first round, processes one
// rank apart exchange data, then each processes accumulates the data
// from the other process using the reduction operator. In the second
// round, processes two ranks apart exchange and accumulate data. In the
// third round, processes four ranks apart exchange and accumulate data.
//
// Process
// 0 1 2 3 4 5 6 7
// | | | | | | | |
// |<--->| |<--->| |<--->| |<--->| Round 1
// | | | | | | | |
// |<--------->| | |<--------->| | Round 2
// | |<--------->| | |<--------->|
// | | | | | | | |
// |<--------------------->| | | | Round 3
// | |<--------------------->| | |
// | | |<--------------------->| |
// | | | |<--------------------->|
// | | | | | | | |
//
// The butterfly pattern works only if the number of processes is a
// power of two. If this is not the case, there are two extra message
// rounds. Each process outside the butterfly pattern sends its data to
// its counterpart inside the butterfly pattern, which accumulates the
// data using the reduction operator. Then the butterfly pattern takes
// place. Afterwards, each process outside the butterfly pattern
// receives the final result from its counterpart inside the butterfly
// pattern. For the case of K=10 processes:
//
// Process
// 0 1 2 3 4 5 6 7 8 9
// | | | | | | | | | |
// |<----------------------------------------------| | Pre
// | |<----------------------------------------------|
// | | | | | | | | | |
// |<--->| |<--->| |<--->| |<--->| | | Round 1
// | | | | | | | | | |
// |<--------->| | |<--------->| | | | Round 2
// | |<--------->| | |<--------->| | |
// | | | | | | | | | |
// |<--------------------->| | | | | | Round 3
// | |<--------------------->| | | | |
// | | |<--------------------->| | | |
// | | | |<--------------------->| | |
// | | | | | | | | | |
// |---------------------------------------------->| | Post
// | |---------------------------------------------->|
// | | | | | | | | | |
//
// If K is a power of two, the all-reduce takes (log_2 K) rounds. If K
// is not a power of two, the all-reduce takes floor(log_2 K)+2 rounds.
// Early exit if only one process.
if (mySize == 1) return;
// Determine the highest power of 2 less than or equal to this
// communicator's size. Processes at this rank and above will be outside
// the butterfly message passing pattern.
int outside = mySizePowerOf2;
// For processes outside the butterfly:
if (myRank >= outside)
{
int insideRank = myRank - outside;
// Send initial data to counterpart inside.
send (insideRank, tag, buffer);
// Receive reduced result from counterpart inside.
receive (insideRank, tag, buffer);
}
// For processes inside the butterfly:
else
{
// Set up temporary receive buffer.
Buf receiveBuf = buffer.getTemporaryBuf();
// Set up reduction buffer on top of data buffer.
Buf reductionBuf = buffer.getReductionBuf (op);
// If there is a counterpart process outside, receive and accumulate
// its initial data.
int outsideRank = myRank + outside;
if (outsideRank < mySize)
{
receive (outsideRank, tag, reductionBuf);
}
// Perform butterfly message passing rounds with counterpart
// processes inside.
int round = 1;
while (round < outside)
{
int otherRank = myRank ^ round;
sendReceive
(otherRank, tag, buffer, otherRank, tag, receiveBuf);
reductionBuf.copy (receiveBuf);
round <<= 1;
}
// If there is a counterpart process outside, send the reduced
// result.
if (outsideRank < mySize)
{
send (outsideRank, tag, buffer);
}
}
}
/**
* Do an all-to-all among all processes in this communicator. A message tag
* of 0 is used.
*
* srcarray must be an array of K buffers, where K is
* the size of this communicator. dstarray must be an array of
* K buffers referring to different storage from the source buffers.
* For each process rank k, 0 <= k <= K, and each
* buffer index i, 0 <= i <= K, the contents of
* srcarray[k] in process i are sent to dstarray[i]
* in process k.
*
* @param srcarray Array of source buffers to be sent by this process.
* @param dstarray Array of destination buffers to be received by this
* process.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if srcarray's length does not
* equal the size of this communicator. Thrown if dstarray's
* length does not equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if srcarray or any element
* thereof is null. Thrown if dstarray or any element thereof
* is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void allToAll
(Buf[] srcarray,
Buf[] dstarray)
throws IOException
{
allToAll (0, srcarray, dstarray);
}
/**
* Do an all-to-all among all processes in this communicator using the given
* message tag. All processes must call allToAll() with the same
* value for tag.
*
* srcarray must be an array of K buffers, where K is
* the size of this communicator. dstarray must be an array of
* K buffers referring to different storage from the source buffers.
* For each process rank k, 0 <= k <= K, and each
* buffer index i, 0 <= i <= K, the contents of
* srcarray[k] in process i are sent to dstarray[i]
* in process k.
*
* @param tag Message tag.
* @param srcarray Array of source buffers to be sent by this process.
* @param dstarray Array of destination buffers to be received by this
* process.
*
* @exception IndexOutOfBoundsException
* (unchecked exception) Thrown if srcarray's length does not
* equal the size of this communicator. Thrown if dstarray's
* length does not equal the size of this communicator.
* @exception NullPointerException
* (unchecked exception) Thrown if srcarray or any element
* thereof is null. Thrown if dstarray or any element thereof
* is null.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void allToAll
(int tag,
Buf[] srcarray,
Buf[] dstarray)
throws IOException
{
// An all-to-all is done as a series of send-receives. Each process
// sends the appropriate buffer to the process one ahead and receives
// the appropriate buffer from the process one behind. Then each process
// sends the appropriate buffer to the process two ahead and receives
// the appropriate buffer from the process two behind. And so on. Here
// is the message pattern for a communicator with 4 processes doing an
// all-to-all:
//
// Process
// 0 1 2 3
// | | | |
// |---->| | | Round 1
// | |---->| |
// | | |---->|
// - -->| | | |--- -
// | | | |
// | | | |
// |---------->| | Round 2
// | |---------->|
// - -->| | |--------- -
// - -------->| | |--- -
// | | | |
// | | | |
// |---------------->| Round 3
// - -->| |--------------- -
// - -------->| |--------- -
// - -------------->| |--- -
// | | | |
// Copy source to destination at this process's own rank.
dstarray[myRank].copy (srcarray[myRank]);
// Initiate K-1 non-blocking send-receives.
CommRequest[] commrequest = new CommRequest [mySize];
for (int i = 1; i < mySize; ++ i)
{
int toRank = (myRank + i) % mySize;
int fromRank = (myRank - i + mySize) % mySize;
commrequest[i] =
sendReceive
(toRank, tag, srcarray[toRank],
fromRank, tag, dstarray[fromRank],
(CommRequest) null);
}
// Wait for completion of all send-receives.
for (int i = 1; i < mySize; ++ i)
{
commrequest[i].waitForFinish();
}
}
/**
* Perform a scan on all processes in this communicator. A message tag of 0
* is used. All processes must call scan() with a buffer of the
* same length and the same item data type, and with the same binary
* operation (class {@linkplain edu.rit.pj.reduction.Op Op}).
*
* Before calling scan(), each process has a buffer filled with
* data items. After scan() returns, each data item in the buffer
* of process rank i has been set to the reduction of the
* corresponding data items in the buffers of process ranks 0 through
* i. The reduction is calculated by this formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process ranks 0 through i. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* @param buf Buffer of data items to be scanned.
* @param op Binary operation.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void scan
(Buf buf,
Op op)
throws IOException
{
scan (0, buf, op);
}
/**
* Perform a scan on all processes in this communicator using the given
* message tag. All processes must call scan() with the same value
* for tag, with a buffer of the same length and the same item data
* type, and with the same binary operation (class {@linkplain
* edu.rit.pj.reduction.Op Op}).
*
* Before calling scan(), each process has a buffer filled with
* data items. After scan() returns, each data item in the buffer
* of process rank i has been set to the reduction of the
* corresponding data items in the buffers of process ranks 0 through
* i. The reduction is calculated by this formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process ranks 0 through i. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* @param tag Message tag.
* @param buf Buffer of data items to be scanned.
* @param op Binary operation.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void scan
(int tag,
Buf buf,
Op op)
throws IOException
{
// Early return if only one process.
if (mySize == 1) return;
// A scan is done as a series of point-to-point messages. The messages
// are organized into rounds. The number of rounds is
// ceil(log_2(mySize)). In the first round, each process sends its data
// to the process one rank ahead, and the incoming data is combined with
// the process's data using the reduction operator. In the second round,
// each process sends its data to the process two ranks ahead. In the
// third round, each process sends its data to process four ranks ahead.
// And so on. Here is the message pattern for a communicator with 8
// processes:
//
// Process
// 0 1 2 3 4 5 6 7
// | | | | | | | |
// |---->|---->|---->|---->|---->|---->|---->| Round 1
// | | | | | | | |
// |---------->| | | | | | Round 2
// | |---------->| | | | |
// | | |---------->| | | |
// | | | |---------->| | |
// | | | | |---------->| |
// | | | | | |---------->|
// | | | | | | | |
// |---------------------->| | | | Round 3
// | |---------------------->| | |
// | | |---------------------->| |
// | | | |---------------------->|
//
// Get temporary buffer for holding incoming data items.
Buf tempbuf = buf.getTemporaryBuf();
// Get reduction buffer for combining data items.
Buf reductionbuf = buf.getReductionBuf (op);
// Do rounds of message passing and reduction.
int skip = 1;
for (;;)
{
int toRank = myRank + skip;
int fromRank = myRank - skip;
boolean toExists = 0 <= toRank && toRank < mySize;
boolean fromExists = 0 <= fromRank && fromRank < mySize;
if (toExists && fromExists)
{
sendReceive (toRank, tag, buf, fromRank, tag, tempbuf);
reductionbuf.copy (tempbuf);
}
else if (fromExists)
{
receive (fromRank, tag, reductionbuf);
}
else if (toExists)
{
send (toRank, tag, buf);
}
else break;
skip <<= 1;
}
}
/**
* Perform an exclusive scan on all processes in this communicator. A
* message tag of 0 is used. All processes must call
* exclusiveScan() with a buffer of the same length and the same
* item data type, with the same binary operation (class {@linkplain
* edu.rit.pj.reduction.Op Op}), and with the same initial data value.
*
* Before calling exclusiveScan(), each process has a buffer filled
* with data items. After exclusiveScan() returns, each data item
* in the buffer of process rank i > 0 has been set to the
* reduction of the corresponding data items in the buffers of
* process ranks 0 through i-1. The reduction is calculated by this
* formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process ranks 0 through i-1. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* In process 0, each data item in the buffer has been set to the initial
* data value using the buffer's fill() method.
*
* If the buffer's item data type is a primitive type, the item
* must be an instance of the corresponding primitive wrapper class -- class
* Integer for type int, class Double for type double, and
* so on. If the item is null, the item data type's default initial
* value is assigned to each element in the buffer.
*
* If the buffer's item data type is a nonprimitive type, the item
* must be an instance of the item class or a subclass thereof. The
* item may be null. Note that since item is
* assigned to every buffer element, every buffer element ends up
* referring to the same item.
*
* @param buf Buffer of data items to be scanned.
* @param op Binary operation.
* @param item Initial data value.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void exclusiveScan
(Buf buf,
Op op,
Object item)
throws IOException
{
exclusiveScan (0, buf, op, item);
}
/**
* Perform an exclusive scan on all processes in this communicator using the
* given message tag. All processes must call exclusiveScan() with
* the same value for tag, with a buffer of the same length and the
* same item data type, with the same binary operation (class {@linkplain
* edu.rit.pj.reduction.Op Op}), and with the same initial data value.
*
* Before calling exclusiveScan(), each process has a buffer filled
* with data items. After exclusiveScan() returns, each data item
* in the buffer of process rank i > 0 has been set to the
* reduction of the corresponding data items in the buffers of
* process ranks 0 through i-1. The reduction is calculated by this
* formula:
*
* item0 op
* item1 op item2 op . . .
*
* where op is the binary operation passed in as an argument and
* item0, item1,
* item2, and so on are the data items in the buffers of
* process ranks 0 through i-1. However, the order in which the data
* items actually are combined is not specified. Therefore, the binary
* operation must be such that the answer will be the same regardless of the
* order in which the data items are combined; that is, the binary operation
* must be commutative and associative.
*
* In process 0, each data item in the buffer has been set to the initial
* data value using the buffer's fill() method.
*
* If the buffer's item data type is a primitive type, the item
* must be an instance of the corresponding primitive wrapper class -- class
* Integer for type int, class Double for type double, and
* so on. If the item is null, the item data type's default initial
* value is assigned to each element in the buffer.
*
* If the buffer's item data type is a nonprimitive type, the item
* must be an instance of the item class or a subclass thereof. The
* item may be null. Note that since item is
* assigned to every buffer element, every buffer element ends up
* referring to the same item.
*
* @param tag Message tag.
* @param buf Buffer of data items to be scanned.
* @param op Binary operation.
* @param item Initial data value.
*
* @exception NullPointerException
* (unchecked exception) Thrown if buf is null or op
* is null.
* @exception ClassCastException
* (unchecked exception) Thrown if buf and op do not
* use the same item data type.
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void exclusiveScan
(int tag,
Buf buf,
Op op,
Object item)
throws IOException
{
// An exclusive scan begins with each process sending its buffer to the
// next higher process. Then process 0 fills its buffer with the initial
// data value, while processes 1 and higher do an inclusive scan.
int toRank;
int fromRank;
boolean toExists;
boolean fromExists;
// Process 0 does this.
if (myRank == 0)
{
// Send buffer to next higher process.
toRank = 1;
toExists = toRank < mySize;
if (toExists)
{
send (toRank, tag, buf);
}
// Fill buffer with initial data value.
buf.fill (item);
}
// Processes 1 and higher do this.
else
{
// Get temporary buffer for holding incoming data items.
Buf tempbuf = buf.getTemporaryBuf();
// Get reduction buffer for combining data items.
Buf reductionbuf = buf.getReductionBuf (op);
// Send buffer to next higher process.
toRank = myRank + 1;
fromRank = myRank - 1;
toExists = 0 <= toRank && toRank < mySize;
if (toExists)
{
sendReceive (toRank, tag, buf, fromRank, tag, tempbuf);
buf.copy (tempbuf);
}
else
{
receive (fromRank, tag, buf);
}
// Do rounds of message passing and reduction.
int skip = 1;
for (;;)
{
toRank = myRank + skip;
fromRank = myRank - skip;
toExists = 1 <= toRank && toRank < mySize;
fromExists = 1 <= fromRank && fromRank < mySize;
if (toExists && fromExists)
{
sendReceive (toRank, tag, buf, fromRank, tag, tempbuf);
reductionbuf.copy (tempbuf);
}
else if (fromExists)
{
receive (fromRank, tag, reductionbuf);
}
else if (toExists)
{
send (toRank, tag, buf);
}
else break;
skip <<= 1;
}
}
}
/**
* Cause all processes in this communicator to wait at a barrier. The
* barrier uses a message tag of 0. All processes must call
* barrier(). The calling thread blocks until every process has
* called barrier(), then the calling thread unblocks and returns
* from the barrier() call.
*
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void barrier()
throws IOException
{
barrier (0);
}
/**
* Cause all processes in this communicator to wait at a barrier, using the
* given message tag. All processes must call barrier() with the
* same tag. The calling thread blocks until every process has called
* barrier(), then the calling thread unblocks and returns from the
* barrier() call.
*
* @param tag Message tag.
*
* @exception IOException
* Thrown if an I/O error occurred.
*/
public void barrier
(int tag)
throws IOException
{
// A barrier is done as an all-reduce of an empty buffer.
allReduce (tag, IntegerBuf.emptyBuffer(), IntegerOp.SUM);
}
/**
* Returns a string version of this communicator. The string includes the
* communicator's size, the current process's rank, and the host and port of
* each backend process.
*
* @return String version.
*/
public String toString()
{
StringBuilder buf = new StringBuilder();
buf.append ("Comm(size=");
buf.append (mySize);
buf.append (",rank=");
buf.append (myRank);
buf.append (",backend");
for (int i = 0; i < mySize; ++ i)
{
if (i > 0) buf.append (',');
buf.append ('[');
buf.append (i);
buf.append ("]=");
buf.append (myAddressForRank[i]);
}
buf.append (')');
return buf.toString();
}
/**
* Dump the state of this communicator on the given print stream. For
* debugging.
*
* @param out Print stream.
* @param prefix String to print at the beginning of each line.
*/
public void dump
(PrintStream out,
String prefix)
{
out.println();
out.println (prefix+getClass().getName()+"@"+Integer.toHexString(System.identityHashCode(this)));
out.println (prefix+"mySize = "+mySize);
out.println (prefix+"myRank = "+myRank);
out.println (prefix+"myHost = "+myHost);
out.println (prefix+"mySizePowerOf2 = "+mySizePowerOf2);
out.println (prefix+"myChannelGroup = "+myChannelGroup);
out.println (prefix+"myAddressForRank:");
for (int i = 0; i < myAddressForRank.length; ++ i)
{
out.println (prefix+"\t["+i+"] "+myAddressForRank[i]);
}
out.println (prefix+"myChannelForRank:");
for (int i = 0; i < myChannelForRank.length; ++ i)
{
out.println (prefix+"\t["+i+"] "+myChannelForRank[i]);
}
out.println (prefix+"myBroadcastTree:");
for (int i = 0; i < myBroadcastTree.length; ++ i)
{
out.print (prefix+"\t["+i+"]");
int[] tree = myBroadcastTree[i];
if (tree == null)
{
out.print (" null");
}
else
{
for (int j = 0; j < tree.length; ++ j)
{
out.print (" "+tree[j]);
}
}
out.println();
}
out.println();
myChannelGroup.dump (out, prefix);
}
// Hidden operations.
/**
* Notify that another process connected a channel to this process.
*
* @param theChannel Channel.
*
* @exception IOException
* Thrown if an I/O error occurred.
*/
private synchronized void doFarEndConnected
(Channel theChannel)
throws IOException
{
// Record channel and rank.
myChannelForRank[getFarRank(theChannel)] = theChannel;
// Notify any threads waiting in getChannel().
notifyAll();
}
/**
* Ensure that a channel for communicating with the process at the given
* rank is or will be set up.
*
* @param farrank Rank of far end process.
*
* @exception IOException
* Thrown if an I/O error occurred.
*/
private synchronized void ensureChannel
(int farrank)
throws IOException
{
// Get channel from channel array.
Channel channel = myChannelForRank[farrank];
// If the channel does not exist:
if (channel == null)
{
// If this is the lower-ranked process, set up the connection.
if (myRank < farrank)
{
myChannelForRank[farrank] =
myChannelGroup.connect (myAddressForRank[farrank]);
}
// If this is the higher-ranked process, the lower-ranked process
// will set up the connection.
}
}
/**
* Get the channel for communicating with the process at the given rank.
*
* @param farrank Rank of far end process.
*
* @return Channel.
*
* @exception IOException
* Thrown if an I/O error occurred.
*/
private synchronized Channel getChannel
(int farrank)
throws IOException
{
// Get channel from channel array.
Channel channel = myChannelForRank[farrank];
// If the channel does not exist:
if (channel == null)
{
// If this is the lower-ranked process, set up the connection.
if (myRank < farrank)
{
channel = myChannelGroup.connect (myAddressForRank[farrank]);
myChannelForRank[farrank] = channel;
}
// If this is the higher-ranked process, wait for the lower-ranked
// process to set up the connection.
else
{
try
{
while (channel == null)
{
wait();
channel = myChannelForRank[farrank];
}
}
catch (InterruptedException exc)
{
IOException exc2 = new InterruptedIOException();
exc2.initCause (exc);
throw exc2;
}
}
}
return channel;
}
/**
* Get the rank of the process at the far end of the given channel.
*
* @param channel Channel.
*
* @return Far end process rank.
*/
static int getFarRank
(Channel channel)
{
return channel.farEndChannelGroupId();
}
/**
* Get an array of process ranks in the broadcast tree for the given root.
* The broadcast tree is cached in the field myBroadcastTree for later use.
*
* @param root Root process's rank.
*
* @return Broadcast tree.
*/
private synchronized int[] getBroadcastTree
(int root)
{
if (myBroadcastTree == null)
{
myBroadcastTree = new int [mySize] [];
}
int[] broadcasttree = myBroadcastTree[root];
if (broadcasttree == null)
{
broadcasttree = CommPattern.broadcastPattern (mySize, myRank, root);
myBroadcastTree[root] = broadcasttree;
}
return broadcasttree;
}
}