WorkerTeam

//******************************************************************************
//
// File:    WorkerTeam.java
// Package: edu.rit.pj
// Unit:    Class edu.rit.pj.WorkerTeam
//
// This Java source file is copyright (C) 2010 by Alan Kaminsky. All rights
// reserved. For further information, contact the author, Alan Kaminsky, at
// ark@cs.rit.edu.
//
// This Java source file is part of the Parallel Java Library ("PJ"). PJ is free
// software; you can redistribute it and/or modify it under the terms of the GNU
// General Public License as published by the Free Software Foundation; either
// version 3 of the License, or (at your option) any later version.
//
// PJ is distributed in the hope that it will be useful, but WITHOUT ANY
// WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
// A PARTICULAR PURPOSE. See the GNU General Public License for more details.
//
// Linking this library statically or dynamically with other modules is making a
// combined work based on this library. Thus, the terms and conditions of the GNU
// General Public License cover the whole combination.
//
// As a special exception, the copyright holders of this library give you
// permission to link this library with independent modules to produce an
// executable, regardless of the license terms of these independent modules, and
// to copy and distribute the resulting executable under terms of your choice,
// provided that you also meet, for each linked independent module, the terms
// and conditions of the license of that module. An independent module is a module
// which is not derived from or based on this library. If you modify this library,
// you may extend this exception to your version of the library, but you are not
// obligated to do so. If you do not wish to do so, delete this exception
// statement from your version.
//
// A copy of the GNU General Public License is provided in the file gpl.txt. You
// may also obtain a copy of the GNU General Public License on the World Wide
// Web at http://www.gnu.org/licenses/gpl.html.
//
//******************************************************************************
package edu.rit.pj;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Semaphore;

/**
 * Class WorkerTeam provides a team of threads, distributed across the processes
 * of a cluster parallel program, for executing a {@linkplain WorkerRegion} in
 * parallel.
 * <P>
 * A worker team uses a communicator for message passing. The communicator is
 * specified as a constructor argument; if not specified, the world communicator
 * is used. Every process that is part of the communicator must create the
 * worker team. In class WorkerTeam, there is one worker thread per process. (To
 * get more than one worker thread per process, use class {@linkplain
 * HybridTeam}.) Every worker thread in every process has a unique index, going
 * from index 0 for the first thread in the first process to index
 * <I>K</I>&minus;1 for the last thread in the last process, where <I>K</I> is
 * the total number of worker threads in all the processes. In process rank 0,
 * there is an additional master thread.
 * <P>
 * To execute a worker region, create a WorkerTeam object; create an instance of
 * a concrete subclass of class {@linkplain WorkerRegion}; and pass this
 * instance to the worker team's <code>execute()</code> method. For further
 * information, see class {@linkplain WorkerRegion}.
 *
 * @author Alan Kaminsky
 * @version 19-Jan-2010
 */
public class WorkerTeam {

// Hidden data members.
    // Number of worker threads in this process.
    int K;

    // Communicator for message passing, its size, this process's rank.
    Comm comm;
    int size;
    int rank;

    // Number of worker threads in all processes.
    int count;

    // Array of worker and master team threads in this process. There are K
    // worker threads. There is an additional master thread in the last process
    // of the communicator.
    WorkerTeamThread[] myThread;

    // Worker region being executed, or null if none is being executed.
    WorkerRegion myRegion;

    // Semaphore for synchronizing threads at the end of a worker region.
    Semaphore myRegionEndSemaphore = new Semaphore(0);

    // Exception map for worker region, or null if none is being executed.
    ConcurrentHashMap<Integer, Throwable> myExceptionMap;

// Hidden constructors.
    /**
     * Construct a new, uninitialized worker team.
     *
     * @param flag To distinguish this constructor from the others.
     */
    WorkerTeam(boolean flag) {
    }

// Exported constructors.
    /**
     * Construct a new worker team with one thread per process and using the
     * world communicator for message passing.
     */
    public WorkerTeam() {
        this(Comm.world());
    }

    /**
     * Construct a new worker team with one thread per process and using the
     * given communicator for message passing.
     *
     * @param comm Communicator to use for message passing.
     * @exception NullPointerException (unchecked exception) Thrown if
     * <code>comm</code> is null.
     */
    public WorkerTeam(Comm comm) {
        if (comm == null) {
            throw new NullPointerException("WorkerTeam(): comm is null");
        }
        initialize(/*K    */1,
                /*comm */ comm,
                /*size */ comm.size(),
                /*rank */ comm.rank(),
                /*count*/ comm.size(),
                /*wlb  */ comm.rank());
    }

// Hidden initializers.
    /**
     * Initialize a new worker team.
     *
     * @param K Number of worker threads in this process.
     * @param comm Communicator to use for message passing.
     * @param size Communicator's size.
     * @param rank This process's rank in the communicator.
     * @param count Number of worker threads in all processes.
     * @param wlb First worker index in this process.
     */
    void initialize(int K,
            Comm comm,
            int size,
            int rank,
            int count,
            int wlb) {
        // Record parameters.
        this.K = K;
        this.comm = comm;
        this.size = size;
        this.rank = rank;
        this.count = count;

        // Set up worker team threads. Additional master thread in process 0.
        int WM = K + (rank == 0 ? 1 : 0);
        myThread = new WorkerTeamThread[WM];
        for (int i = 0; i < K; ++i) {
            myThread[i] = new WorkerTeamThread(this, wlb + i);
        }
        if (WM > K) {
            myThread[K] = new WorkerTeamThread(this, -1);
        }
    }

// Exported operations.
    /**
     * Execute the given worker region.
     *
     * @param theRegion Worker region.
     * @exception NullPointerException (unchecked exception) Thrown if
     * <code>theRegion</code> is null.
     * @exception IllegalStateException (unchecked exception) Thrown if this
     * worker team is already executing a worker region. Thrown if
     * <code>theRegion</code> is already being executed by a worker team.
     * @exception Exception Exception thrown by the worker region's
     * <code>start()</code>,
     * <code>run()</code>, or <code>finish()</code> methods.
     * @throws java.lang.Exception if any.
     */
    public final void execute(WorkerRegion theRegion)
            throws Exception {
        // Verify preconditions.
        if (theRegion == null) {
            throw new NullPointerException("WorkerTeam.execute(): theRegion is null");
        }
        if (myRegion != null) {
            throw new IllegalStateException("WorkerTeam.execute(): Already executing a worker region");
        }
        if (theRegion.myTeam != null) {
            throw new IllegalStateException("WorkerTeam.execute(): theRegion already being executed by a worker team");
        }

        // Record worker region.
        myRegion = theRegion;
        myExceptionMap = new ConcurrentHashMap<Integer, Throwable>(K, 0.75f, K);
        theRegion.myTeam = this;

        try {
            // Perform the worker region's start() method. Any exception aborts
            // the execute() method.
            myRegion.start();

            // Release the team threads to perform the worker region's run()
            // method.
            for (WorkerTeamThread thread : myThread) {
                thread.myRegionBeginSemaphore.release();
            }

            // Wait until all team threads have returned from the worker
            // region's run() method.
            myRegionEndSemaphore.acquireUninterruptibly(myThread.length);

            // Propagate any exceptions thrown by the run() method.
            if (myExceptionMap.isEmpty()) {
            } else if (myExceptionMap.size() == 1) {
                rethrow(myExceptionMap.values().iterator().next());
            } else {
                throw new MultipleParallelException("WorkerTeam.execute(): Multiple threads threw exceptions",
                        myExceptionMap);
            }

            // Perform the worker region's finish() method. Any exception aborts
            // the execute() method.
            myRegion.finish();
        } finally {
            // Clean up.
            myRegion.myTeam = null;
            myExceptionMap = null;
            myRegion = null;
        }
    }

    /**
     * Determine if this worker team is executing a worker region.
     *
     * @return True if this worker team is executing a worker region, false
     * otherwise.
     */
    public final boolean isExecutingInParallel() {
        return myRegion != null;
    }

    /**
     * Returns the worker region of code that this worker team is executing.
     *
     * @return Worker region.
     * @exception IllegalStateException (unchecked exception) Thrown if this
     * worker team is not executing a worker region.
     */
    public final WorkerRegion region() {
        if (myRegion == null) {
            throw new IllegalStateException("WorkerTeam.region(): Not executing a worker region");
        }
        return myRegion;
    }

    /**
     * Determine the number of worker threads in this worker team in this
     * process. This does not include the master thread if any.
     *
     * @return Number of worker threads in this process.
     */
    public final int getThreadCount() {
        return K;
    }

    /**
     * Determine the total number of worker threads in this worker team in all
     * processes. This does not include the master thread.
     *
     * @return Number of worker threads in all processes.
     */
    public final int getTotalThreadCount() {
        return count;
    }

    /**
     * Determine the rank of the process that contains the master thread. At
     * present, this is always rank 0.
     *
     * @return Master process rank.
     */
    public int masterRank() {
        return 0;
    }

    /**
     * Determine the rank of the process that contains the worker thread with
     * the given index.
     *
     * @param w Worker index.
     * @return Worker process rank.
     * @exception IllegalArgumentException (unchecked exception) Thrown if
     * <code>w</code> is not in the range 0 ..
     * <code>getTotalThreadCount()</code>&minus;1.
     */
    public int workerRank(int w) {
        if (0 > w || w >= count) {
            throw new IllegalArgumentException("WorkerTeam.workerRank(): w (= " + w + ") illegal");
        }
        return w;
    }

// Hidden operations.
    /**
     * Re-throw the given object as a checked or unchecked exception. If the
     * given object is null or is not throwable, do nothing.
     */
    static void rethrow(Object exc)
            throws Exception {
        if (exc instanceof RuntimeException) {
            throw (RuntimeException) exc;
        } else if (exc instanceof Exception) {
            throw (Exception) exc;
        } else if (exc instanceof Error) {
            throw (Error) exc;
        }
    }

}