//******************************************************************************
//
// File:    Runner.java
// Package: edu.rit.pj.job
// Unit:    Class edu.rit.pj.job.Runner
//
// 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.
//
// 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.job;

import edu.rit.mp.BooleanBuf;

import edu.rit.mp.buf.BooleanItemBuf;

import edu.rit.pj.Comm;
import edu.rit.pj.WorkerIteration;
import edu.rit.pj.WorkerRegion;
import edu.rit.pj.WorkerTeam;

import edu.rit.util.Instance;

import java.io.File;
import java.io.IOException;
import java.io.PrintStream;

import java.util.Date;
import java.util.HashSet;
import java.util.Scanner;

/**
 * Class Runner is a parallel program that runs, in parallel, a group of
 * {@linkplain Job}s created by a {@linkplain JobGenerator}. The job generator
 * is specified on the command line as a constructor expression. An instance of
 * the class specified in the constructor expression is constructed, with the
 * constructor arguments specified in the constructor expression. For further
 * information, see class {@linkplain edu.rit.util.Instance Instance}.
 * 

 * The Runner program is targeted at three use cases:
 * 

 * 

 * All these processes form a worker team. The Runner program uses the
 * job generator specified on the command line to create jobs and sends each job
 * to a worker team process to be executed.
 * 

 * When the Runner program starts, it prints the job generator constructor
 * expression on the standard output. Whenever a job starts or finishes, the
 * Runner program prints a log message on the standard output consisting of the
 * job's number and description.
 * 

 * Checkpointing. It is recommended to redirect the Runner program's
 * standard output into a checkpoint file. If a failure occurs before the
 * Runner program finishes running all the jobs, the checkpoint file contains a
 * record of the job generator that was used as well as which jobs did and did
 * not finish. To resume the Runner program where it left off, specify the
 * checkpoint file name on the command line instead of a job generator
 * constructor expression. The Runner program reads the checkpoint file to
 * determine the job generator and the jobs that finished. The Runner program
 * then generates and runs the jobs that did not finish.
 * 

 * Usage: java edu.rit.pj.job.Runner { generator | file }
 * 
generator = Job generator constructor expression
 * 
file = Checkpoint file name
 *
 * @author  Alan Kaminsky
 * @version 22-Oct-2010
 */
public class Runner
	{

// Prevent construction.

	private Runner()
		{
		}

// Global variables.

	private static PrintStream stdout = System.out;
	private static PrintStream stderr = System.err;

	private static Comm world;
	private static int rank;

	private static WorkerTeam team;

	private static String generatorExpression;
	private static HashSet omitted;
	private static JobGenerator generator;

// Main program.

	/**
	 * Main program.
	 */
	public static void main
		(String[] args)
		throws Exception
		{
		if (args.length != 1) usage();

		// Initialize world communicator.
		Comm.init (args);
		world = Comm.world();
		rank = world.rank();

		// Set up worker team.
		team = new WorkerTeam();

		// Master process sets up job generator.
		if (rank == team.masterRank())
			{
			omitted = new HashSet();

			// Assume argument is a checkpoint file name and try to read it.
			Scanner scanner = null;
			try
				{
				scanner = new Scanner (new File (args[0]));
				}
			catch (IOException exc)
				{
				}

			// Read checkpoint file.
			if (scanner != null)
				{
				while (scanner.hasNextLine())
					{
					Scanner linescanner = new Scanner (scanner.nextLine());
					String word;
					int jobnum;
					if (! linescanner.hasNext()) continue;
					word = linescanner.next();
					if (! word.equals ("***")) continue;
					if (! linescanner.hasNext()) continue;
					word = linescanner.next();
					if (word.equals ("Generator"))
						{
						if (! linescanner.hasNext()) continue;
						if (generatorExpression == null)
							{
							generatorExpression = linescanner.next();
							}
						}
					else if (word.equals ("Job"))
						{
						if (! linescanner.hasNextInt()) continue;
						jobnum = linescanner.nextInt();
						if (! linescanner.hasNext()) continue;
						word = linescanner.next();
						if (word.equals ("finished")) omitted.add (jobnum);
						}
					}
				scanner.close();
				}

			// Assume argument is a job generator constructor expression.
			else
				{
				generatorExpression = args[0];
				}

			// Create job generator.
			if (generatorExpression == null)
				{
				stderr.printf
					("Runner: No job generator in checkpoint file %s%n",
					 args[0]);
				}
			else
				{
				try
					{
					generator = (JobGenerator)
						Instance.newInstance (generatorExpression);
					stdout.printf
						("*** Generator %s%n", generatorExpression);
					generator.omit (omitted);
					}
				catch (Throwable exc)
					{
					stderr.printf
						("Runner: Could not create job generator %s%n",
						 generatorExpression);
					exc.printStackTrace (stderr);
					}
				}
			}

		// Abort every process if job generator was not created.
		BooleanItemBuf buf = BooleanBuf.buffer (generator != null);
		world.broadcast (team.masterRank(), buf);
		if (! buf.item) System.exit (1);

		// Generate and run jobs.
		team.execute (new WorkerRegion()
			{
			public void run() throws Exception
				{
				execute (generator, new WorkerIteration()
					{
					public void sendTaskInput
						(Job job, Comm comm, int wRank, int tag)
						{
						stdout.printf
							("*** Job %d started %s %s%n",
							 job.getJobNumber(),
							 new Date(),
							 job.getDescription());
						}

					public void run (Job job)
						{
						job.run();
						}

					public void receiveTaskOutput
						(Job job, Comm comm, int wRank, int tag)
						{
						stdout.printf
							("*** Job %d finished %s%n",
							 job.getJobNumber(),
							 new Date());
						}
					});
				}
			});

		if (rank == team.masterRank())
			{
			stdout.printf ("*** All jobs finished%n");
			}
		}

// Hidden operations.

	/**
	 * Print a usage message and exit.
	 */
	private static void usage()
		{
		stderr.println ("Usage: java edu.rit.pj.job.Runner {|}");
		stderr.println (" = Job generator constructor expression");
		stderr.println (" = Checkpoint file name");
		System.exit (1);
		}

	}

Ads help maintain this website.