//******************************************************************************
//
// File: ParallelTeam.java
// Package: edu.rit.pj
// Unit: Class edu.rit.pj.ParallelTeam
//
// This Java source file is copyright (C) 2008 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.
//
// 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;
import java.util.concurrent.atomic.AtomicInteger;
/**
* Class ParallelTeam provides a team of threads for executing a {@linkplain
* ParallelRegion} in parallel.
*
* To execute a parallel region, create a ParallelTeam object; create an
* instance of a concrete subclass of class {@linkplain ParallelRegion}; and
* pass this instance to the parallel team's execute() method. For
* further information, see class {@linkplain ParallelRegion}.
*
* @author Alan Kaminsky
* @version 19-May-2008
*/
public class ParallelTeam
{
// Hidden data members.
// Number of threads.
int K;
// Array of threads in the team.
ParallelTeamThread[] myThread;
// Parallel region being executed, or null if none is being executed.
ParallelRegion myRegion;
// Semaphore for synchronizing threads at the end of a parallel region.
Semaphore myRegionEndSemaphore = new Semaphore (0);
// Exception map for parallel region, or null if none is being executed.
ConcurrentHashMap myExceptionMap;
// Team barrier flag. Used by the ParallelRegion.barrier() method.
volatile int myBarrierFlag;
// Parallel construct counter. Counts how many parallel constructs have been
// encountered.
AtomicInteger myConstructCount = new AtomicInteger (0);
// Exported constructors.
/**
* Construct a new parallel team with the default number of threads. If the
* "pj.nt" Java property is specified, that property gives the
* default number of threads, which must be an integer greater than or equal
* to 1. If the "pj.nt" Java property is not specified, the default
* number of threads is the value returned by the
* Runtime.availableProcessors() method. You can specify the
* default number of threads on the Java command line like this:
*
* java -Dpj.nt=4 . . .
*
*
* @exception IllegalArgumentException
* (unchecked exception) Thrown if the "pj.nt" property value
* is not an integer greater than or equal to 1.
*/
public ParallelTeam()
{
this (getDefaultThreadCount());
}
/**
* Construct a new parallel team with the given number of threads.
*
* @param K Number of threads.
*
* @exception IllegalArgumentException
* (unchecked exception) Thrown if K is less than 1.
*/
public ParallelTeam
(int K)
{
if (K < 1)
{
throw new IllegalArgumentException
("ParallelTeam(): K must be >= 1");
}
this.K = K;
myThread = new ParallelTeamThread [K];
myThread[0] = new ParallelTeamThread_0 (this, 0);
for (int i = 1; i < K; ++ i)
{
myThread[i] = new ParallelTeamThread (this, i);
}
}
// Exported operations.
/**
* Execute the given parallel region.
*
* @param theRegion Parallel region.
*
* @exception NullPointerException
* (unchecked exception) Thrown if theRegion is null.
* @exception IllegalStateException
* (unchecked exception) Thrown if this parallel team is already
* executing a parallel region. Thrown if theRegion is already
* being executed by a parallel team.
* @exception Exception
* Exception thrown by the parallel region's start(),
* run(), or finish() methods.
*/
public final void execute
(ParallelRegion theRegion)
throws Exception
{
// Verify preconditions.
if (theRegion == null)
{
throw new NullPointerException
("ParallelTeam.execute(): theRegion is null");
}
if (myRegion != null)
{
throw new IllegalStateException
("ParallelTeam.execute(): Already executing a parallel region");
}
if (theRegion.myTeam != null)
{
throw new IllegalStateException
("ParallelTeam.execute(): theRegion already being executed by a parallel team");
}
// Record parallel region.
myRegion = theRegion;
myExceptionMap = new ConcurrentHashMap (K, 0.75f, K);
theRegion.myTeam = this;
try
{
// Perform the parallel region's start() method. Any exception
// aborts the execute() method.
myRegion.start();
// Release the team threads to perform the parallel region's run()
// method.
for (ParallelTeamThread thread : myThread)
{
thread.myRegionBeginSemaphore.release();
}
// Wait until all team threads have returned from the parallel
// region's run() method.
myRegionEndSemaphore.acquireUninterruptibly (K);
// 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
("ParallelTeam.execute(): Multiple threads threw exceptions",
myExceptionMap);
}
// Perform the parallel 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 parallel team is executing a parallel region.
*
* @return True if this parallel team is executing a parallel region, false
* otherwise.
*/
public final boolean isExecutingInParallel()
{
return myRegion != null;
}
/**
* Returns the parallel region of code that this parallel team is executing.
*
* @return Parallel region.
*
* @exception IllegalStateException
* (unchecked exception) Thrown if this parallel team is not executing a
* parallel region.
*/
public final ParallelRegion region()
{
if (myRegion == null)
{
throw new IllegalStateException
("ParallelTeam.region(): Not executing a parallel region");
}
return myRegion;
}
/**
* Determine the number of threads in this parallel team.
*
* @return Number of threads in the team.
*/
public final int getThreadCount()
{
return K;
}
/**
* Determine the default number of threads for a parallel team. If the
* "pj.nt" Java property is specified, that property gives the
* default number of threads, which must be an integer greater than or equal
* to 1. If the "pj.nt" Java property is not specified, the default
* number of threads is the value returned by the
* Runtime.availableProcessors() method. You can specify the
* default number of threads on the Java command line like this:
*
* java -Dpj.nt=4 . . .
*
*
* @return Default number of threads for a parallel team.
*
* @exception IllegalArgumentException
* (unchecked exception) Thrown if the "pj.nt" property value
* is not an integer greater than or equal to 1.
*/
public static int getDefaultThreadCount()
{
int k = PJProperties.getPjNt();
if (k == 0) k = Runtime.getRuntime().availableProcessors();
return k;
}
// Hidden operations.
/**
* Do the thread-0 portion of a barrier with no barrier action. This method
* is called by thread 0 of the parallel team.
*/
void barrier()
{
// Get the new team barrier flag.
int newBarrierFlag = myBarrierFlag ^ 1;
// Wait until each team thread 1 .. K-1 has switched to the new
// barrier flag.
for (int i = 1; i < K; ++ i)
{
ParallelTeamThread thread_i = myThread[i];
if (thread_i.myBarrierFlag != newBarrierFlag)
{
Spinner spinner = new Spinner();
while (thread_i.myBarrierFlag != newBarrierFlag)
{
spinner.spin();
}
}
}
// Switch to the new team barrier flag.
myBarrierFlag = newBarrierFlag;
}
/**
* Do the thread-0 portion of a barrier with a barrier action. This method
* is called by thread 0 of the parallel team.
*
* @param action Barrier action.
*
* @exception Exception
* Thrown if the action's run() method throws an
* exception.
*/
void barrier
(BarrierAction action)
throws Exception
{
// Get the new team barrier flag.
int newBarrierFlag = myBarrierFlag ^ 1;
// Wait until each team thread 1 .. K-1 has switched to the new
// barrier flag.
for (int i = 1; i < K; ++ i)
{
ParallelTeamThread thread_i = myThread[i];
if (thread_i.myBarrierFlag != newBarrierFlag)
{
Spinner spinner = new Spinner();
while (thread_i.myBarrierFlag != newBarrierFlag)
{
spinner.spin();
}
}
}
try
{
// Do the barrier action.
action.myTeam = this;
action.run();
}
finally
{
action.myTeam = null;
// Switch to the new team barrier flag.
myBarrierFlag = newBarrierFlag;
}
}
/**
* 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;
}
}
}
Ads help maintain this website.