text:
package org.apache.hadoop.streaming;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.net.URI;
import java.net.URLEncoder;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.ListIterator;
import java.util.Map;
import java.util.TreeMap;
import java.util.TreeSet;

import org.apache.commons.cli2.*;
import org.apache.commons.cli2.builder.ArgumentBuilder;
import org.apache.commons.cli2.builder.DefaultOptionBuilder;
import org.apache.commons.cli2.builder.GroupBuilder;
import org.apache.commons.cli2.commandline.Parser;
import org.apache.commons.cli2.option.PropertyOption;
import org.apache.commons.cli2.resource.ResourceConstants;
import org.apache.commons.cli2.util.HelpFormatter;
import org.apache.commons.cli2.validation.InvalidArgumentException;
import org.apache.commons.cli2.validation.Validator;
import org.apache.commons.logging.*;

import org.apache.hadoop.mapred.lib.aggregate.ValueAggregatorCombiner;
import org.apache.hadoop.mapred.lib.aggregate.ValueAggregatorReducer;
import org.apache.hadoop.conf.Configuration;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.fs.Path;

import org.apache.hadoop.mapred.FileAlreadyExistsException;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.InvalidJobConfException;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.JobClient;
import org.apache.hadoop.mapred.FileOutputFormat;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileAsTextInputFormat;
import org.apache.hadoop.mapred.TextInputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;

import org.apache.hadoop.filecache.*;
import org.apache.hadoop.util.*;

/** All the client-side work happens here.
 * (Jar packaging, MapRed job submission and monitoring)
 */
public class StreamJob {

  protected static final Log LOG = LogFactory.getLog(StreamJob.class.getName());
  final static String REDUCE_NONE = "NONE";

  /** -----------Streaming CLI Implementation  **/
  private DefaultOptionBuilder builder =
    new DefaultOptionBuilder("-","-", false);
  private ArgumentBuilder argBuilder = new ArgumentBuilder();
  private Parser parser = new Parser();
  private Group allOptions;
  HelpFormatter helpFormatter = new HelpFormatter("  ", "  ", "  ", 900);
  // need these two at class level to extract values later from
  // commons-cli command line
  private MultiPropertyOption jobconf = new MultiPropertyOption(
                                                                "-jobconf", "(n=v) Optional. Add or override a JobConf property.", 'D');
  private MultiPropertyOption cmdenv = new MultiPropertyOption(
                                                               "-cmdenv", "(n=v) Pass env.var to streaming commands.", 'E');

  public StreamJob(String[] argv, boolean mayExit) {
    setupOptions();
    argv_ = argv;
    mayExit_ = mayExit;
  }

  /**
   * This is the method that actually
   * intializes the job conf and submits the job
   * to the jobtracker
   * @throws IOException
   */
  public int go() throws IOException {
    init();

    preProcessArgs();
    parseArgv();
    postProcessArgs();

    setJobConf();
    return submitAndMonitorJob();
  }

  protected void init() {
    try {
      env_ = new Environment();
    } catch (IOException io) {
      throw new RuntimeException(io);
    }
  }

  void preProcessArgs() {
    verbose_ = false;
    addTaskEnvironment_ = "";
  }

  void postProcessArgs() throws IOException {
    if (cluster_ == null) {
      // hadoop-default.xml is standard, hadoop-local.xml is not.
      cluster_ = "default";
    }
    hadoopAliasConf_ = "hadoop-" + getClusterNick() + ".xml";
    if (inputSpecs_.size() == 0) {
      fail("Required argument: -input <name>");
    }
    if (output_ == null) {
      fail("Required argument: -output ");
    }
    msg("addTaskEnvironment=" + addTaskEnvironment_);

    Iterator it = packageFiles_.iterator();
    while (it.hasNext()) {
      File f = new File((String) it.next());
      if (f.isFile()) {
        shippedCanonFiles_.add(f.getCanonicalPath());
      }
    }
    msg("shippedCanonFiles_=" + shippedCanonFiles_);

    // careful with class names..
    mapCmd_ = unqualifyIfLocalPath(mapCmd_);
    comCmd_ = unqualifyIfLocalPath(comCmd_);
    redCmd_ = unqualifyIfLocalPath(redCmd_);
  }

  String unqualifyIfLocalPath(String cmd) throws IOException {
    if (cmd == null) {
      //
    } else {
      String prog = cmd;
      String args = "";
      int s = cmd.indexOf(" ");
      if (s != -1) {
        prog = cmd.substring(0, s);
        args = cmd.substring(s + 1);
      }
      String progCanon;
      try {
        progCanon = new File(prog).getCanonicalPath();
      } catch (IOException io) {
        progCanon = prog;
      }
      boolean shipped = shippedCanonFiles_.contains(progCanon);
      msg("shipped: " + shipped + " " + progCanon);
      if (shipped) {
        // Change path to simple filename.
        // That way when PipeMapRed calls Runtime.exec(),
        // it will look for the excutable in Task's working dir.
        // And this is where TaskRunner unjars our job jar.
        prog = new File(prog).getName();
        if (args.length() > 0) {
          cmd = prog + " " + args;
        } else {
          cmd = prog;
        }
      }
    }
    msg("cmd=" + cmd);
    return cmd;
  }

  String getHadoopAliasConfFile() {
    return new File(getHadoopClientHome() + "/conf", hadoopAliasConf_).getAbsolutePath();
  }

  void parseArgv(){
    CommandLine cmdLine = null;
    try{
      cmdLine = parser.parse(argv_);
    }catch(Exception oe){
      LOG.error(oe.getMessage());
      if (detailedUsage_) {
        exitUsage(true);
      } else {
        exitUsage(false);
      }
    }

    if (cmdLine != null){
      verbose_ =  cmdLine.hasOption("-verbose");
      detailedUsage_ = cmdLine.hasOption("-info");
      debug_ = cmdLine.hasOption("-debug")? debug_ + 1 : debug_;

      inputSpecs_.addAll(cmdLine.getValues("-input"));
      output_ = (String) cmdLine.getValue("-output");

      mapCmd_ = (String)cmdLine.getValue("-mapper");
      comCmd_ = (String)cmdLine.getValue("-combiner");
      redCmd_ = (String)cmdLine.getValue("-reducer");
      postMapCmd_ = (String)cmdLine.getValue("-postmapper");
      postRedCmd_ = (String)cmdLine.getValue("-postreducer");
      preRedCmd_ = (String)cmdLine.getValue("-prereducer");

      jobName_ = (String)cmdLine.getValue("-jobname");

      packageFiles_.addAll(cmdLine.getValues("-file"));

      cluster_ = (String)cmdLine.getValue("-cluster");

      configPath_.addAll(cmdLine.getValues("-config"));

      String fsName = (String)cmdLine.getValue("-dfs");
      if (null != fsName){
        userJobConfProps_.put("fs.default.name", fsName);
      }

      String jt = (String)cmdLine.getValue("mapred.job.tracker");
      if (null != jt){
        userJobConfProps_.put("fs.default.name", jt);
      }

      additionalConfSpec_ = (String)cmdLine.getValue("-additionalconfspec");
      inputFormatSpec_ = (String)cmdLine.getValue("-inputformat");
      outputFormatSpec_ = (String)cmdLine.getValue("-outputformat");
      numReduceTasksSpec_ = (String)cmdLine.getValue("-numReduceTasks");
      partitionerSpec_ = (String)cmdLine.getValue("-partitioner");
      inReaderSpec_ = (String)cmdLine.getValue("-inputreader");

      List<String> car = cmdLine.getValues("-cacheArchive");
      if (null != car){
        for(String s : car){
          cacheArchives = (cacheArchives == null)?s :cacheArchives + "," + s;
        }
      }

      List<String> caf = cmdLine.getValues("-cacheFile");
      if (null != caf){
        for(String s : caf){
          cacheFiles = (cacheFiles == null)?s :cacheFiles + "," + s;
        }
      }

      List<String> jobConfArgs = (List<String>)cmdLine.getValue(jobconf);
      List<String> envArgs = (List<String>)cmdLine.getValue(cmdenv);

      if (null != jobConfArgs){
        for(String s : jobConfArgs){
          String []parts = s.split("=", 2);
          userJobConfProps_.put(parts[0], parts[1]);
        }
      }
      if (null != envArgs){
        for(String s : envArgs){
          if (addTaskEnvironment_.length() > 0) {
            addTaskEnvironment_ += " ";
          }
          addTaskEnvironment_ += s;
        }
      }
    }else if (detailedUsage_) {
      exitUsage(true);
    }
  }

  protected void msg(String msg) {
    if (verbose_) {
      System.out.println("STREAM: " + msg);
    }
  }

  private Option createOption(String name, String desc,
                              String argName, int max, boolean required){
    Argument argument = argBuilder.
      withName(argName).
      withMinimum(1).
      withMaximum(max).
      create();
    return builder.
      withLongName(name).
      withArgument(argument).
      withDescription(desc).
      withRequired(required).
      create();
  }

  private Option createOption(String name, String desc,
                              String argName, int max, boolean required, Validator validator){

    Argument argument = argBuilder.
      withName(argName).
      withMinimum(1).
      withMaximum(max).
      withValidator(validator).
      create();

    return builder.
      withLongName(name).
      withArgument(argument).
      withDescription(desc).
      withRequired(required).
      create();
  }

  private Option createBoolOption(String name, String desc){
    return builder.withLongName(name).withDescription(desc).create();
  }

  private void setupOptions(){

    final Validator fileValidator = new Validator(){
        public void validate(final List values) throws InvalidArgumentException {
          // Note : This code doesnt belong here, it should be changed to
          // an can exec check in java 6
          for (String file : (List<String>)values) {
            File f = new File(file);
            if (!f.exists()) {
              throw new InvalidArgumentException("Argument : " +
                                                 f.getAbsolutePath() + " doesn't exist.");
            }
            if (!f.isFile()) {
              throw new InvalidArgumentException("Argument : " +
                                                 f.getAbsolutePath() + " is not a file.");
            }
            if (!f.canRead()) {
              throw new InvalidArgumentException("Argument : " +
                                                 f.getAbsolutePath() + " is not accessible");
            }
          }
        }
      };

    // Note: not extending CLI2's FileValidator, that overwrites
    // the String arg into File and causes ClassCastException
    // in inheritance tree.
    final Validator execValidator = new Validator(){
        public void validate(final List values) throws InvalidArgumentException {
          // Note : This code doesnt belong here, it should be changed to
          // an can exec check in java 6
          for (String file : (List<String>)values) {
            try{
              Runtime.getRuntime().exec("chmod 0777 " + (new File(file)).getAbsolutePath());
            }catch(IOException ioe){
              // ignore
            }
          }
          fileValidator.validate(values);
        }
      };

    Option input   = createOption("input",
                                  "DFS input file(s) for the Map step",
                                  "path",
                                  Integer.MAX_VALUE,
                                  true);

    Option output  = createOption("output",
                                  "DFS output directory for the Reduce step",
                                  "path", 1, true);
    Option mapper  = createOption("mapper",
                                  "The streaming command to run", "cmd", 1, false);
    Option combiner = createOption("combiner",
                                   "The streaming command to run", "cmd", 1, false);
    // reducer could be NONE
    Option reducer = createOption("reducer",
                                  "The streaming command to run", "cmd", 1, false);
    Option file = createOption("file",
                               "File/dir to be shipped in the Job jar file",
                               "file", Integer.MAX_VALUE, false, execValidator);
    Option dfs = createOption("dfs",
                              "Optional. Override DFS configuration", "<h:p>|local", 1, false);
    Option jt = createOption("jt",
                             "Optional. Override JobTracker configuration", "<h:p>|local", 1, false);
    Option additionalconfspec = createOption("additionalconfspec",
                                             "Optional.", "spec", 1, false);
    Option inputformat = createOption("inputformat",
                                      "Optional.", "spec", 1, false);
    Option outputformat = createOption("outputformat",
                                       "Optional.", "spec", 1, false);
    Option partitioner = createOption("partitioner",
                                      "Optional.", "spec", 1, false);
    Option numReduceTasks = createOption("numReduceTasks",
        "Optional.", "spec",1, false );
    Option inputreader = createOption("inputreader",
                                      "Optional.", "spec", 1, false);
    Option cacheFile = createOption("cacheFile",
                                    "File name URI", "fileNameURI", Integer.MAX_VALUE, false);
    Option cacheArchive = createOption("cacheArchive",
                                       "File name URI", "fileNameURI", 1, false);
    Option jobname = createOption("jobname",
                                  "Optional", "spec", 1, false);
    Option postmapper  = createOption("postmapper",
                                  "Optional", "spec", 1, false);
    Option postreducer  = createOption("postreducer",
                                  "Optional", "spec", 1, false);
    Option prereducer  = createOption("prereducer",
                                      "Optional", "spec", 1, false);

    // boolean properties

    Option verbose = createBoolOption("verbose", "print verbose output");
    Option info = createBoolOption("info", "print verbose output");
    Option help = createBoolOption("help", "print this help message");
    Option debug = createBoolOption("debug", "print debug output");
    Option inputtagged = createBoolOption("inputtagged", "inputtagged");

    allOptions = new GroupBuilder().
      withOption(input).
      withOption(output).
      withOption(mapper).
      withOption(postmapper).
      withOption(postreducer).
      withOption(prereducer).
      withOption(combiner).
      withOption(reducer).
      withOption(file).
      withOption(dfs).
      withOption(jt).
      withOption(additionalconfspec).
      withOption(inputformat).
      withOption(outputformat).
      withOption(partitioner).
      withOption(numReduceTasks).
      withOption(inputreader).
      withOption(jobconf).
      withOption(cmdenv).
      withOption(jobname).
      withOption(cacheFile).
      withOption(cacheArchive).
      withOption(verbose).
      withOption(info).
      withOption(debug).
      withOption(inputtagged).
      withOption(help).
      create();
    parser.setGroup(allOptions);

  }

  public void exitUsage(boolean detailed) {
    //         1         2         3         4         5         6         7
    //1234567890123456789012345678901234567890123456789012345678901234567890123456789
    if (!detailed) {
      System.out.println("Usage: $HADOOP_HOME/bin/hadoop [--config dir] jar \\");
      System.out.println("          $HADOOP_HOME/hadoop-streaming.jar [options]");
      System.out.println("Options:");
      System.out.println("  -input    <path>     DFS input file(s) for the Map step");
      System.out.println("  -output   <path>     DFS output directory for the Reduce step");
      System.out.println("  -mapper   <cmd|JavaClassName>      The streaming command to run");
      System.out.println("  -postmapper   <JavaClassName>      Map Class to post process streaming mapper");
      System.out.println("  -postreducer  <JavaClassName>   Map Class to post process streaming reducer");
      System.out.println("  -prereducer  <JavaClassName>   Reduce Class to process reduce input before streaming");
      System.out.println("  -combiner <JavaClassName> Combiner has to be a Java class");
      System.out.println("  -reducer  <cmd|JavaClassName>      The streaming command to run");
      System.out.println("  -file     <file>     File/dir to be shipped in the Job jar file");
      System.out.println("  -dfs    <h:p>|local  Optional. Override DFS configuration");
      System.out.println("  -jt     <h:p>|local  Optional. Override JobTracker configuration");
      System.out.println("  -additionalconfspec specfile  Optional.");
      System.out.println("  -inputformat TextInputFormat(default)|SequenceFileAsTextInputFormat|JavaClassName Optional.");
      System.out.println("  -outputformat TextOutputFormat(default)|JavaClassName  Optional.");
      System.out.println("  -partitioner JavaClassName  Optional.");
      System.out.println("  -numReduceTasks <num>  Optional.");
      System.out.println("  -inputreader <spec>  Optional.");
      System.out.println("  -jobconf  <n>=<v>    Optional. Add or override a JobConf property");
      System.out.println("  -cmdenv   <n>=<v>    Optional. Pass env.var to streaming commands");
      System.out.println("  -jobname  <name>     Optional. Set the name of the job");
      System.out.println("  -cacheFile fileNameURI");
      System.out.println("  -cacheArchive fileNameURI");
      System.out.println("  -verbose");
      System.out.println();
      System.out.println("For more details about these options:");
      System.out.println("Use $HADOOP_HOME/bin/hadoop jar build/hadoop-streaming.jar -info");
      fail("");
    }
    System.out.println("In -input: globbing on <path> is supported and can have multiple -input");
    System.out.println("Default Map input format: a line is a record in UTF-8");
    System.out.println("  the key part ends at first TAB, the rest of the line is the value");
    System.out.println("Custom input format: -inputformat package.MyInputFormat ");
    System.out.println("Map output format, reduce input/output format:");
    System.out.println("  Format defined by what the mapper command outputs. Line-oriented");
    System.out.println();
    System.out.println("The files or directories named in the -file argument[s] end up in the");
    System.out.println("  working directory when the mapper and reducer are run.");
    System.out.println("  The location of this working directory is unspecified.");
    System.out.println();
    System.out.println("To set the number of reduce tasks (num. of output files):");
    System.out.println("  -jobconf mapred.reduce.tasks=10");
    System.out.println("To skip the sort/combine/shuffle/sort/reduce step:");
    System.out.println("  Use -numReduceTasks 0");
    System.out
      .println("  A Task's Map output then becomes a 'side-effect output' rather than a reduce input");
    System.out
      .println("  This speeds up processing, This also feels more like \"in-place\" processing");
    System.out.println("  because the input filename and the map input order are preserved");
    System.out.println("  This equivalent -reducer NONE");
    System.out.println();
    System.out.println("To speed up the last reduces:");
    System.out.println("  -jobconf mapred.speculative.execution=true");
    System.out.println("To name the job (appears in the JobTracker Web UI):");
    System.out.println("  -jobconf mapred.job.name='My Job' ");
    System.out.println("To change the local temp directory:");
    System.out.println("  -jobconf dfs.data.dir=/tmp/dfs");
    System.out.println("  -jobconf stream.tmpdir=/tmp/streaming");
    System.out.println("Additional local temp directories with -cluster local:");
    System.out.println("  -jobconf mapred.local.dir=/tmp/local");
    System.out.println("  -jobconf mapred.system.dir=/tmp/system");
    System.out.println("  -jobconf mapred.temp.dir=/tmp/temp");
    System.out.println("Use a custom hadoopStreaming build along a standard hadoop install:");
    System.out.println("  $HADOOP_HOME/bin/hadoop jar /path/my-hadoop-streaming.jar [...]\\");
    System.out
      .println("    [...] -jobconf stream.shipped.hadoopstreaming=/path/my-hadoop-streaming.jar");
    System.out.println("For more details about jobconf parameters see:");
    System.out.println("  http://wiki.apache.org/lucene-hadoop/JobConfFile");
    System.out.println("To set an environement variable in a streaming command:");
    System.out.println("   -cmdenv EXAMPLE_DIR=/home/example/dictionaries/");
    System.out.println();
    System.out.println("Shortcut:");
    System.out
      .println("   setenv HSTREAMING \"$HADOOP_HOME/bin/hadoop jar $HADOOP_HOME/hadoop-streaming.jar\"");
    System.out.println();
    System.out.println("Example: $HSTREAMING -mapper \"/usr/local/bin/perl5 filter.pl\"");
    System.out.println("           -file /local/filter.pl -input \"/logs/0604*/*\" [...]");
    System.out.println("  Ships a script, invokes the non-shipped perl interpreter");
    System.out.println("  Shipped files go to the working directory so filter.pl is found by perl");
    System.out.println("  Input files are all the daily logs for days in month 2006-04");
    fail("");
  }

  public void fail(String message) {
    if (mayExit_) {
      System.err.println(message);
      throw new RuntimeException(message);
    } else {
      throw new IllegalArgumentException(message);
    }
  }

  // --------------------------------------------

  protected String getHadoopClientHome() {
    String h = env_.getProperty("HADOOP_HOME"); // standard Hadoop
    if (h == null) {
      //fail("Missing required environment variable: HADOOP_HOME");
      h = "UNDEF";
    }
    return h;
  }

  protected boolean isLocalHadoop() {
    boolean local;
    if (jobConf_ == null) {
      local = getClusterNick().equals("local");
    } else {
      local = StreamUtil.isLocalJobTracker(jobConf_);
    }
    return local;
  }

  protected String getClusterNick() {
    return cluster_;
  }

  /** @return path to the created Jar file or null if no files are necessary.
   */
  protected String packageJobJar() throws IOException {
    ArrayList unjarFiles = new ArrayList();

    // Runtime code: ship same version of code as self (job submitter code)
    // usually found in: build/contrib or build/hadoop-<version>-dev-streaming.jar

    // First try an explicit spec: it's too hard to find our own location in this case:
    // $HADOOP_HOME/bin/hadoop jar /not/first/on/classpath/custom-hadoop-streaming.jar
    // where findInClasspath() would find the version of hadoop-streaming.jar in $HADOOP_HOME
    String runtimeClasses = userJobConfProps_.get("stream.shipped.hadoopstreaming"); // jar or class dir
    System.out.println(runtimeClasses + "=@@@userJobConfProps_.get(stream.shipped.hadoopstreaming");

    if (runtimeClasses == null) {
      runtimeClasses = StreamUtil.findInClasspath(StreamJob.class.getName());
    }
    if (runtimeClasses == null) {
      throw new IOException("runtime classes not found: " + getClass().getPackage());
    } else {
      msg("Found runtime classes in: " + runtimeClasses);
    }
    if (isLocalHadoop()) {
      // don't package class files (they might get unpackaged in "." and then
      //  hide the intended CLASSPATH entry)
      // we still package everything else (so that scripts and executable are found in
      //  Task workdir like distributed Hadoop)
    } else {
      if (new File(runtimeClasses).isDirectory()) {
        packageFiles_.add(runtimeClasses);
      } else {
        unjarFiles.add(runtimeClasses);
      }
    }
    if (packageFiles_.size() + unjarFiles.size() == 0) {
      return null;
    }
    String tmp = jobConf_.get("stream.tmpdir"); //, "/tmp/${user.name}/"
    File tmpDir = (tmp == null) ? null : new File(tmp);
    // tmpDir=null means OS default tmp dir
    File jobJar = File.createTempFile("hive-streamjob", ".jar", tmpDir);
    System.out.println("packageJobJar: " + packageFiles_ + " " + unjarFiles + " " + jobJar
                       + " tmpDir=" + tmpDir);
    if (debug_ == 0) {
      jobJar.deleteOnExit();
    }
    JarBuilder builder = new JarBuilder();
    if (verbose_) {
      builder.setVerbose(true);
    }
    String jobJarName = jobJar.getAbsolutePath();
    builder.merge(packageFiles_, unjarFiles, jobJarName);
    return jobJarName;
  }

  /**
   * This method sets the user jobconf variable specified
   * by user using -jobconf key=value
   * @param doEarlyProps
   */
  protected void setUserJobConfProps(boolean doEarlyProps) {
    Iterator it = userJobConfProps_.keySet().iterator();
    while (it.hasNext()) {
      String key = (String) it.next();
      String val = (String)userJobConfProps_.get(key);
      boolean earlyName = key.equals("fs.default.name");
      earlyName |= key.equals("stream.shipped.hadoopstreaming");
      if (doEarlyProps == earlyName) {
        msg("xxxJobConf: set(" + key + ", " + val + ") early=" + doEarlyProps);
        jobConf_.set(key, val);
      }
    }
  }

  /**
   * get the uris of all the files/caches
   */
  protected void getURIs(String lcacheArchives, String lcacheFiles) {
    String archives[] = StringUtils.getStrings(lcacheArchives);
    String files[] = StringUtils.getStrings(lcacheFiles);
    fileURIs = StringUtils.stringToURI(files);
    archiveURIs = StringUtils.stringToURI(archives);
  }

  protected void setJobConf() throws IOException {
    msg("hadoopAliasConf_ = " + hadoopAliasConf_);
    config_ = new Configuration();
    if (!cluster_.equals("default")) {
      config_.addResource(new Path(getHadoopAliasConfFile()));
    } else {
      // use only defaults: hadoop-default.xml and hadoop-site.xml
    }
    System.out.println("additionalConfSpec_:" + additionalConfSpec_);
    if (additionalConfSpec_ != null) {
      config_.addResource(new Path(additionalConfSpec_));
    }
    Iterator it = configPath_.iterator();
    while (it.hasNext()) {
      String pathName = (String) it.next();
      config_.addResource(new Path(pathName));
    }

    // general MapRed job properties
    jobConf_ = new JobConf(config_);

    // All streaming jobs get the task timeout value
    // from the configuration settings.

    setUserJobConfProps(true);

    // The correct FS must be set before this is called!
    // (to resolve local vs. dfs drive letter differences)
    // (mapred.working.dir will be lazily initialized ONCE and depends on FS)
    for (int i = 0; i < inputSpecs_.size(); i++) {
      FileInputFormat.addInputPaths(jobConf_,
                        (String) inputSpecs_.get(i));
    }
    jobConf_.set("stream.numinputspecs", "" + inputSpecs_.size());

    String defaultPackage = this.getClass().getPackage().getName();
    Class c;
    Class fmt = null;
    if (inReaderSpec_ == null && inputFormatSpec_ == null) {
      fmt = TextInputFormat.class;
    } else if (inputFormatSpec_ != null) {
      if (inputFormatSpec_.equals(TextInputFormat.class.getName())
          || inputFormatSpec_.equals(TextInputFormat.class.getCanonicalName())) {
        fmt = TextInputFormat.class;
      } else if (inputFormatSpec_.equals(KeyValueTextInputFormat.class
          .getName())
          || inputFormatSpec_.equals(KeyValueTextInputFormat.class
              .getCanonicalName())) {
      } else if (inputFormatSpec_.equals(SequenceFileInputFormat.class
          .getName())
          || inputFormatSpec_
              .equals(org.apache.hadoop.mapred.SequenceFileInputFormat.class
                  .getCanonicalName())) {
          fmt = SequenceFileInputFormat.class;
      } else if (inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class
          .getName())
          || inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class
              .getCanonicalName())) {
        fmt = SequenceFileAsTextInputFormat.class;
      } else {
        c = StreamUtil.goodClassOrNull(inputFormatSpec_, defaultPackage);
        if (c != null) {
          fmt = c;
        } else {

        }
      }
    }
    if (fmt == null) {
      fmt = StreamInputFormat.class;
    }

    jobConf_.setInputFormat(fmt);

    jobConf_.setOutputKeyClass(Text.class);
    jobConf_.setOutputValueClass(Text.class);

    if (jobName_ != null) {
        jobConf_.setJobName(jobName_);
    }

    jobConf_.set("stream.addenvironment", addTaskEnvironment_);

    if (mapCmd_ != null) {
      c = StreamUtil.goodClassOrNull(mapCmd_, defaultPackage);
      if (c != null) {
        if(postMapCmd_ != null) {
          fail("-postmapper cannot be combined with a Java map class");
        }
        jobConf_.setMapperClass(c);
      } else {
        jobConf_.setMapperClass(PipeMapper.class);
        jobConf_.set("stream.map.streamprocessor",
                     URLEncoder.encode(mapCmd_, "UTF-8"));
      }
    }

    if (postMapCmd_ != null) {
      c = StreamUtil.goodClassOrNull(postMapCmd_, defaultPackage);
      if (c != null) {
          jobConf_.setClass("stream.map.posthook", c, Mapper.class);
      } else {
          fail("postmapper: "+postMapCmd_+" is not a valid Java Class");
      }
    }

    if (comCmd_ != null) {
      c = StreamUtil.goodClassOrNull(comCmd_, defaultPackage);
      if (c != null) {
        jobConf_.setCombinerClass(c);
      }
    }

    boolean reducerNone_ = false;
    if (redCmd_ != null) {
      reducerNone_ = redCmd_.equals(REDUCE_NONE);
      if (redCmd_.compareToIgnoreCase("aggregate") == 0) {
        if(postRedCmd_ != null) {
          fail("-postreducer cannot be combined with a Java reduce class");
        }
        if(preRedCmd_ != null) {
          fail("-prereducer cannot be combined with a Java reduce class");
        }
        jobConf_.setReducerClass(ValueAggregatorReducer.class);
        jobConf_.setCombinerClass(ValueAggregatorCombiner.class);
      } else {

        c = StreamUtil.goodClassOrNull(redCmd_, defaultPackage);
        if (c != null) {
          if(postRedCmd_ != null) {
              fail("-postreducer cannot be combined with a Java reduce class "+c.getName());
          }
          if(preRedCmd_ != null) {
              fail("-prereducer cannot be combined with a Java reduce class "+c.getName());
          }
          jobConf_.setReducerClass(c);
        } else {
          jobConf_.setReducerClass(PipeReducer.class);
          jobConf_.set("stream.reduce.streamprocessor", URLEncoder.encode(
              redCmd_, "UTF-8"));
        }
      }
    }

    if (postRedCmd_ != null) {
      c = StreamUtil.goodClassOrNull(postRedCmd_, defaultPackage);
      if (c != null) {
          jobConf_.setClass("stream.reduce.posthook", c, Mapper.class);
      } else {
          fail("postreducer: "+postRedCmd_+" is not a valid Java Class");
      }
    }

    if (preRedCmd_ != null) {
      c = StreamUtil.goodClassOrNull(preRedCmd_, defaultPackage);
      if (c != null) {
          jobConf_.setClass("stream.reduce.prehook", c, Reducer.class);
      } else {
          fail("postreducer: "+preRedCmd_+" is not a valid Java Class");
      }
    }

    if (inReaderSpec_ != null) {
      String[] args = inReaderSpec_.split(",");
      String readerClass = args[0];
      // this argument can only be a Java class
      c = StreamUtil.goodClassOrNull(readerClass, defaultPackage);
      if (c != null) {
        jobConf_.set("stream.recordreader.class", c.getName());
      } else {
        fail("-inputreader: class not found: " + readerClass);
      }
      for (int i = 1; i < args.length; i++) {
        String[] nv = args[i].split("=", 2);
        String k = "stream.recordreader." + nv[0];
        String v = (nv.length > 1) ? nv[1] : "";
        jobConf_.set(k, v);
      }
    }

    setUserJobConfProps(false);
    FileOutputFormat.setOutputPath(jobConf_,new Path(output_));
    fmt = null;
    if (outputFormatSpec_!= null) {
      c = StreamUtil.goodClassOrNull(outputFormatSpec_, defaultPackage);
      if (c != null) {
        fmt = c;
      }
    }
    if (fmt == null) {
      fmt = TextOutputFormat.class;
    }
    jobConf_.setOutputFormat(fmt);

    if (partitionerSpec_!= null) {
      c = StreamUtil.goodClassOrNull(partitionerSpec_, defaultPackage);
      if (c != null) {
        jobConf_.setPartitionerClass(c);
      }
    }

    if (numReduceTasksSpec_!= null) {
      int numReduceTasks = Integer.parseInt(numReduceTasksSpec_);
      jobConf_.setNumReduceTasks(numReduceTasks);
    }
    if (reducerNone_) {
      jobConf_.setNumReduceTasks(0);
    }

    // last, allow user to override anything
    // (although typically used with properties we didn't touch)

    jar_ = packageJobJar();
    if (jar_ != null) {
      jobConf_.setJar(jar_);
    }

    if ((cacheArchives != null) || (cacheFiles != null)){
      getURIs(cacheArchives, cacheFiles);
      boolean b = DistributedCache.checkURIs(fileURIs, archiveURIs);
      if (!b)
        fail(LINK_URI);
    }
    DistributedCache.createSymlink(jobConf_);
    // set the jobconf for the caching parameters
    if (cacheArchives != null)
      DistributedCache.setCacheArchives(archiveURIs, jobConf_);
    if (cacheFiles != null)
      DistributedCache.setCacheFiles(fileURIs, jobConf_);

    if (verbose_) {
      listJobConfProperties();
    }

    msg("submitting to jobconf: " + getJobTrackerHostPort());
  }

  /**
   * Prints out the jobconf properties on stdout
   * when verbose is specified.
   */
  protected void listJobConfProperties()
  {
    msg("==== JobConf properties:");
    Iterator it = jobConf_.iterator();
    TreeMap sorted = new TreeMap();
    while(it.hasNext()) {
      Map.Entry en = (Map.Entry)it.next();
      sorted.put(en.getKey(), en.getValue());
    }
    it = sorted.entrySet().iterator();
    while(it.hasNext()) {
      Map.Entry en = (Map.Entry)it.next();
      msg(en.getKey() + "=" + en.getValue());
    }
    msg("====");
  }

  protected String getJobTrackerHostPort() {
    return jobConf_.get("mapred.job.tracker");
  }

  protected void jobInfo() {
    if (isLocalHadoop()) {
      LOG.info("Job running in-process (local Hadoop)");
    } else {
      String hp = getJobTrackerHostPort();
      LOG.info("To kill this job, run:");
      LOG.info(getHadoopClientHome() + "/bin/hadoop job  -Dmapred.job.tracker=" + hp + " -kill "
               + jobId_);
      //LOG.info("Job file: " + running_.getJobFile());
      LOG.info("Tracking URL: " + StreamUtil.qualifyHost(running_.getTrackingURL()));
    }
  }

  // Based on JobClient
  public int submitAndMonitorJob() throws IOException {

    if (jar_ != null && isLocalHadoop()) {
      // getAbs became required when shell and subvm have different working dirs...
      File wd = new File(".").getAbsoluteFile();
      StreamUtil.unJar(new File(jar_), wd);
    }

    // if jobConf_ changes must recreate a JobClient
    jc_ = new JobClient(jobConf_);
    boolean error = true;
    running_ = null;
    String lastReport = null;
    try {
      running_ = jc_.submitJob(jobConf_);
      jobId_ = running_.getJobID();

      LOG.info("getLocalDirs(): " + Arrays.asList(jobConf_.getLocalDirs()));
      LOG.info("Running job: " + jobId_);
      jobInfo();

      while (!running_.isComplete()) {
        try {
          Thread.sleep(1000);
        } catch (InterruptedException e) {
        }
        running_ = jc_.getJob(jobId_);
        String report = null;
        report = " map " + Math.round(running_.mapProgress() * 100) + "%  reduce "
          + Math.round(running_.reduceProgress() * 100) + "%";

        if (!report.equals(lastReport)) {
          LOG.info(report);
          lastReport = report;
        }
      }
      if (!running_.isSuccessful()) {
        jobInfo();
	LOG.error("Job not Successful!");
	return 1;
      }
      LOG.info("Job complete: " + jobId_);
      LOG.info("Output: " + output_);
      error = false;
    } catch(FileNotFoundException fe) {
      LOG.error("Error launching job , bad input path : " + fe.getMessage());
      return 2;
    } catch(InvalidJobConfException je) {
      LOG.error("Error launching job , Invalid job conf : " + je.getMessage());
      return 3;
    } catch(FileAlreadyExistsException fae) {
      LOG.error("Error launching job , Output path already exists : "
                + fae.getMessage());
      return 4;
    } catch(IOException ioe) {
      LOG.error("Error Launching job : " + ioe.getMessage());
      return 5;
    } finally {
      if (error && (running_ != null)) {
        LOG.info("killJob...");
        running_.killJob();
      }
      jc_.close();
    }
    return 0;
  }
  /** Support -jobconf x=y x1=y1 type options **/
  class MultiPropertyOption extends PropertyOption{
    private String optionString;
    MultiPropertyOption(){
      super();
    }

    MultiPropertyOption(final String optionString,
                        final String description,
                        final int id){
      super(optionString, description, id);
      this.optionString = optionString;
    }

    public boolean canProcess(final WriteableCommandLine commandLine,
                              final String argument) {
      boolean ret = (argument != null) && argument.startsWith(optionString);

      return ret;
    }
    public void process(final WriteableCommandLine commandLine,
                        final ListIterator arguments) throws OptionException {
      final String arg = (String) arguments.next();

      if (!canProcess(commandLine, arg)) {
        throw new OptionException(this,
                                  ResourceConstants.UNEXPECTED_TOKEN, arg);
      }

      ArrayList properties = new ArrayList();
      String next = "";
      while(arguments.hasNext()){
        next = (String) arguments.next();
        if (!next.startsWith("-")){
          properties.add(next);
        }else{
          arguments.previous();
          break;
        }
      }

      // add to any existing values (support specifying args multiple times)
      List<String> oldVal = (List<String>)commandLine.getValue(this);
      if (oldVal == null){
        commandLine.addValue(this, properties);
      }else{
        oldVal.addAll(properties);
      }
    }
  }

  protected boolean mayExit_;
  protected String[] argv_;
  protected boolean verbose_;
  protected boolean detailedUsage_;
  protected int debug_;

  protected Environment env_;

  protected String jar_;
  protected boolean localHadoop_;
  protected Configuration config_;
  protected JobConf jobConf_;
  protected JobClient jc_;

  // command-line arguments
  protected ArrayList inputSpecs_ = new ArrayList(); // <String>
  protected TreeSet seenPrimary_ = new TreeSet(); // <String>
  protected boolean hasSimpleInputSpecs_;
  protected ArrayList packageFiles_ = new ArrayList(); // <String>
  protected ArrayList shippedCanonFiles_ = new ArrayList(); // <String>
  protected TreeMap<String, String> userJobConfProps_ = new TreeMap<String, String>();
  protected String output_;
  protected String mapCmd_;
  protected String comCmd_;
  protected String redCmd_;
  protected String jobName_;
  protected String cluster_;
  protected String cacheFiles;
  protected String cacheArchives;
  protected URI[] fileURIs;
  protected URI[] archiveURIs;
  protected ArrayList configPath_ = new ArrayList(); // <String>
  protected String hadoopAliasConf_;
  protected String inReaderSpec_;
  protected String inputFormatSpec_;
  protected String outputFormatSpec_;
  protected String partitionerSpec_;
  protected String numReduceTasksSpec_;
  protected String additionalConfSpec_;
  protected String postMapCmd_;
  protected String postRedCmd_;
  protected String preRedCmd_;

  // Use to communicate config to the external processes (ex env.var.HADOOP_USER)
  // encoding "a=b c=d"
  protected String addTaskEnvironment_;

  protected boolean outputSingleNode_;
  protected long minRecWrittenToEnableSkip_;

  protected RunningJob running_;
  protected String jobId_;
  protected static String LINK_URI = "You need to specify the uris as hdfs://host:port/#linkname," +
    "Please specify a different link name for all of your caching URIs";
}

redpajama_set_name: RedPajamaGithub


text:
#include "src/gpu/piet/Scene.h"

#include "include/core/SkTypes.h"

namespace skgpu::piet {
namespace {

bool get_piet_verb(SkPath::Verb verb, PgpuPathVerb* outVerb) {
    switch (verb) {
        case SkPath::kMove_Verb:
            *outVerb = PgpuPathVerb::MoveTo;
            break;
        case SkPath::kLine_Verb:
            *outVerb = PgpuPathVerb::LineTo;
            break;
        case SkPath::kQuad_Verb:
            *outVerb = PgpuPathVerb::QuadTo;
            break;
        case SkPath::kConic_Verb:
        case SkPath::kCubic_Verb:
            *outVerb = PgpuPathVerb::CurveTo;
            break;
        case SkPath::kClose_Verb:
            *outVerb = PgpuPathVerb::Close;
            break;
        case SkPath::kDone_Verb:
        default:
            return false;
    }
    return true;
}

PgpuFill get_fill_type(SkPathFillType fillType) {
    switch (fillType) {
        case SkPathFillType::kWinding:
            return PgpuFill::NonZero;
        case SkPathFillType::kEvenOdd:
            return PgpuFill::EvenOdd;
        default:
            // TODO(b/238756757): pgpu-render doesn't define fill types for kInverseWinding and
            // kInverseEvenOdd. This should be updated to support those cases.
            SkDebugf("piet: unsupported fill type\n");
            break;
    }
    return PgpuFill::NonZero;
}

}  // namespace

Scene::Scene() : Object(pgpu_scene_new()) { SkASSERT(this->get() != nullptr); }

void Scene::addPath(const SkPath& path, const Transform& transform) {
    this->initBuilderIfNecessary();

    SkPath::Iter pathIter(path, /*forceClose=*/false);
    PgpuPathIter iter;
    iter.context = &pathIter;
    iter.next_element = [](void* context, PgpuPathElement* outElem) {
        SkASSERT(outElem);
        SkASSERT(context);

        SkPoint points[4];
        SkPath::Verb verb = static_cast<SkPath::Iter*>(context)->next(points);
        if (!get_piet_verb(verb, &outElem->verb)) {
            return false;
        }

        for (int i = 0; i < 3; ++i) {
            const SkPoint& p = points[i];
            outElem->points[i] = {p.x(), p.y()};
        }

        return true;
    };

    pgpu_scene_builder_transform(fActiveBuilder->get(), &transform);
    pgpu_scene_builder_fill_path(
            fActiveBuilder->get(), get_fill_type(path.getFillType()), nullptr, nullptr, &iter);
}

bool Scene::finalize() {
    if (!fActiveBuilder) {
        return false;
    }
    fActiveBuilder.reset();
    return true;
}

void Scene::initBuilderIfNecessary() {
    if (!fActiveBuilder) {
        fActiveBuilder = Builder(pgpu_scene_builder_for_scene(this->get()));
        SkASSERT(fActiveBuilder);
    }
}

}  // namespace skgpu::piet


text:
<h2>{{title}} STOCK EXCHANGE</h2>
  <input  [(ngModel)] = "title" placeholder = "Customize the name">
  <nav>
  <button routerLink = "/display">SEARCH A STOCK</button>
  <button routerLink = "/analysis">TECHNICAL ANALYSIS</button>
  <button routerLink = "/compare">COMPARE STOCKS</button>
  <button routerLink = "/arithmetic">ARITHMETIC</button>
  </nav>
  <router-outlet></router-outlet>


text:
title: Add a Subscription
category: Subscriptions
order: 2
---
To add a new Subscription:

1) Select "Create" from the "Main Navigation" and choose "Subscription"

2) Select from the list of available "Platforms"

3) Enter the platform "Key Settings"

4) By default, "Enable Provisioning" is selected to allow you to create resources on the target platform. Unchecking this box will make the subscription read only, which may be useful for reporting purposes.

5)   Press "Add Subscription"


text:
class CreateCompetitionItemImages < ActiveRecord::Migration
  def change
    create_table :competition_item_images do |t|
      t.integer :order
      t.references :competition_item

      t.timestamps
    end

    add_attachment :competition_item_images, :image
  end
end


text:
var passport = require('passport');
var ldap = require('ldapjs');
var util = require('util');

function Strategy(options, verify) {

    if (!options || typeof options === 'function') {
        throw new Error('LDAP authentication strategy requires options');
    }

    passport.Strategy.call(this);

    this.name = 'ldapBind';
    this.options = options;
    this.verify = verify;
}

util.inherits(Strategy, passport.Strategy);

var verify = function() {
  return function(err, user, info) {
    if (err)   return this.error(err);
    if (!user) return this.fail(info);
    return this.success(user, info);
  }.bind(this);
};

Strategy.prototype.authenticate = function(req) {
    var self = this;

    if (!req.body.username || !req.body.password) {
        return self.fail(401);
    }

    var username = req.body.username;
    if (!username) {
        return retry('username required');
    }

    var password = req.body.password;
    if (!password) {
        return retry('password required');
    }

    var client = ldap.createClient(this.options.server);

    client.bind(username, password, function (err, user) {
        client.unbind();
        if (err) {
            if (err.name === 'InvalidCredentialsError' || err.name === 'NoSuchObjectError' || (typeof err === 'string' && err.match(/no such user/i))) {
                return self.fail('Invalid username/password');
            }
            return self.error(err);
        }
        else {
            if (self.verify) {
                if (self.options.passReqToCallback) {
                    return self.verify(req, username, verify.call(self));
                } else {
                    return self.verify(username, verify.call(self));
                }
            } else {
                return self.success(username);
            }
        }
    });
};

module.exports = Strategy;


text:
namespace Shapes
{
    public class Rectangle : Shape
    {
        private double height;
        private double width;

        public override double CalculateArea()
        {
            return this.height * this.width;
        }

        public override double CalculatePerimeter()
        {
            return 2 * (this.height + this.width);
        }

        public Rectangle(double height, double width)
        {
            this.height = height;
            this.width = width;
        }

        public override string Draw()
        {
            return base.Draw() + "Rectangle";
        }
    }
}


text:
<?php

class RsyncTest extends \Codeception\TestCase\Test
{
    /**
     * @var \CodeGuy
     */
    protected $guy;

    // tests
    public function testRsync()
    {
        verify(
            (new \Robo\Task\Remote\Rsync())
                ->fromPath('src/')
                ->toHost('localhost')
                ->toUser('dev')
                ->toPath('/var/www/html/app/')
                ->recursive()
                ->excludeVcs()
                ->checksum()
                ->wholeFile()
                ->verbose()
                ->progress()
                ->humanReadable()
                ->stats()
                ->getCommand()
        )->equals(
                'rsync --recursive --exclude .git --exclude .svn --exclude .hg --checksum --whole-file --verbose --progress --human-readable --stats src/ \'dev@localhost:/var/www/html/app/\''
        );

        // From the folder 'foo bar' (with space) in 'src' directory
        verify(
            (new \Robo\Task\Remote\Rsync())
                ->fromPath('src/foo bar/baz')
                ->toHost('localhost')
                ->toUser('dev')
                ->toPath('/var/path/with/a space')
                ->getCommand()
        )->equals(
                'rsync \'src/foo bar/baz\' \'dev@localhost:/var/path/with/a space\''
        );

        // Copy two folders, 'src/foo' and 'src/bar'
        verify(
            (new \Robo\Task\Remote\Rsync())
                ->fromPath(['src/foo', 'src/bar'])
                ->toHost('localhost')
                ->toUser('dev')
                ->toPath('/var/path/with/a space')
                ->getCommand()
        )->equals(
                'rsync src/foo src/bar \'dev@localhost:/var/path/with/a space\''
        );
    }
}


text:
import assert from 'power-assert';

describe('example', () => {
  it('should be example', () => assert('example' === 'example')); // eslint-disable-line no-self-compare
});


text:
<?php

/**
 * Class to manage command line arguments and options.
 *
 * @package    symfony
 * @subpackage command
 * @author     Fabien Potencier <fabien.potencier@symfony-project.com>
 * @version    SVN: $Id: sfCommandManager.class.php 21908 2009-09-11 12:06:21Z fabien $
 */
class sfCommandManager
{
  protected
    $arguments            = '',
    $errors               = array(),
    $optionSet            = null,
    $argumentSet          = array(),
    $optionValues         = array(),
    $argumentValues       = array(),
    $parsedArgumentValues = array();

  /**
   * Constructor.
   *
   * @param sfCommandArgumentSet $argumentSet A sfCommandArgumentSet object
   * @param sfCommandOptionSet   $optionSet   A setOptionSet object
   */
  public function __construct(sfCommandArgumentSet $argumentSet = null, sfCommandOptionSet $optionSet = null)
  {
    if (null === $argumentSet)
    {
      $argumentSet = new sfCommandArgumentSet();
    }
    $this->setArgumentSet($argumentSet);

    if (null === $optionSet)
    {
      $optionSet = new sfCommandOptionSet();
    }
    $this->setOptionSet($optionSet);
  }

  /**
   * Sets the argument set.
   *
   * @param sfCommandArgumentSet $argumentSet A sfCommandArgumentSet object
   */
  public function setArgumentSet(sfCommandArgumentSet $argumentSet)
  {
    $this->argumentSet = $argumentSet;
  }

  /**
   * Gets the argument set.
   *
   * @return sfCommandArgumentSet A sfCommandArgumentSet object
   */
  public function getArgumentSet()
  {
    return $this->argumentSet;
  }

  /**
   * Sets the option set.
   *
   * @param sfCommandOptionSet $optionSet A sfCommandOptionSet object
   */
  public function setOptionSet(sfCommandOptionSet $optionSet)
  {
    $this->optionSet = $optionSet;
  }

  /**
   * Gets the option set.
   *
   * @return sfCommandOptionSet A sfCommandOptionSet object
   */
  public function getOptionSet()
  {
    return $this->optionSet;
  }

  /**
   * Processes command line arguments.
   *
   * @param mixed $arguments A string or an array of command line parameters
   */
  public function process($arguments = null)
  {
    if (null === $arguments)
    {
      $arguments = $_SERVER['argv'];

      // we strip command line program
      if (isset($arguments[0]) && '-' != $arguments[0][0])
      {
        array_shift($arguments);
      }
    }
    else if (!is_array($arguments))
    {
      // hack to split arguments with spaces : --test="with some spaces"
      $arguments = preg_replace_callback('/(\'|")(.+?)\\1/', function($matches) {
        return str_replace(' ', '=PLACEHOLDER=', $matches[2]);
      }, $arguments);
      $arguments = preg_split('/\s+/', $arguments);
      $arguments = str_replace('=PLACEHOLDER=', ' ', $arguments);
    }

    $this->arguments            = $arguments;
    $this->optionValues         = $this->optionSet->getDefaults();
    $this->argumentValues       = $this->argumentSet->getDefaults();
    $this->parsedArgumentValues = array();
    $this->errors               = array();

    while (!in_array($argument = array_shift($this->arguments), array('', null)))
    {
      if ('--' == $argument)
      {
        // stop options parsing
        $this->parsedArgumentValues = array_merge($this->parsedArgumentValues, $this->arguments);
        break;
      }

      if ('--' == substr($argument, 0, 2))
      {
        $this->parseLongOption(substr($argument, 2));
      }
      else if ('-' == $argument[0])
      {
        $this->parseShortOption(substr($argument, 1));
      }
      else
      {
        $this->parsedArgumentValues[] = $argument;
      }
    }

    $position = 0;
    foreach ($this->argumentSet->getArguments() as $argument)
    {
      if (array_key_exists($position, $this->parsedArgumentValues))
      {
        if ($argument->isArray())
        {
          $this->argumentValues[$argument->getName()] = array_slice($this->parsedArgumentValues, $position);
          break;
        }
        else
        {
          $this->argumentValues[$argument->getName()] = $this->parsedArgumentValues[$position];
        }
      }
      ++$position;
    }

    $this->arguments = $arguments;

    if (count($this->parsedArgumentValues) < $this->argumentSet->getArgumentRequiredCount())
    {
      $this->errors[] = 'Not enough arguments.';
    }
    else if (count($this->parsedArgumentValues) > $this->argumentSet->getArgumentCount())
    {
      $this->errors[] = sprintf('Too many arguments ("%s" given).', implode(' ', $this->parsedArgumentValues));
    }
  }

  /**
   * Returns true if the current command line options validate the argument and option sets.
   *
   * @return true if there are some validation errors, false otherwise
   */
  public function isValid()
  {
    return count($this->errors) ? false : true;
  }

  /**
   * Gets the current errors.
   *
   * @return array An array of errors
   */
  public function getErrors()
  {
    return $this->errors;
  }

  /**
   * Returns the argument values.
   *
   * @return array An array of argument values
   */
  public function getArgumentValues()
  {
    return $this->argumentValues;
  }

  /**
   * Returns the argument value for a given argument name.
   *
   * @param string $name The argument name
   *
   * @return mixed The argument value
   */
  public function getArgumentValue($name)
  {
    if (!$this->argumentSet->hasArgument($name))
    {
      throw new sfCommandException(sprintf('The "%s" argument does not exist.', $name));
    }

    return $this->argumentValues[$name];
  }

  /**
   * Returns the options values.
   *
   * @return array An array of option values
   */
  public function getOptionValues()
  {
    return $this->optionValues;
  }

  /**
   * Returns the option value for a given option name.
   *
   * @param string $name The option name
   *
   * @return mixed The option value
   */
  public function getOptionValue($name)
  {
    if (!$this->optionSet->hasOption($name))
    {
      throw new sfCommandException(sprintf('The "%s" option does not exist.', $name));
    }

    return $this->optionValues[$name];
  }

  /**
   * Parses a short option.
   *
   * @param string $argument The option argument
   */
  protected function parseShortOption($argument)
  {
    // short option can be aggregated like in -vd (== -v -d)
    for ($i = 0, $count = strlen($argument); $i < $count; $i++)
    {
      $shortcut = $argument[$i];
      $value    = true;

      if (!$this->optionSet->hasShortcut($shortcut))
      {
        $this->errors[] = sprintf('The option "-%s" does not exist.', $shortcut);
        continue;
      }

      $option = $this->optionSet->getOptionForShortcut($shortcut);

      // required argument?
      if ($option->isParameterRequired())
      {
        if ($i + 1 < strlen($argument))
        {
          $value = substr($argument, $i + 1);
          $this->setOption($option, $value);
          break;
        }
        else
        {
          // take next element as argument (if it doesn't start with a -)
          if (count($this->arguments) && $this->arguments[0][0] != '-')
          {
            $value = array_shift($this->arguments);
            $this->setOption($option, $value);
            break;
          }
          else
          {
            $this->errors[] = sprintf('Option "-%s" requires an argument', $shortcut);
            $value = null;
          }
        }
      }
      else if ($option->isParameterOptional())
      {
        if (substr($argument, $i + 1) != '')
        {
          $value = substr($argument, $i + 1);
        }
        else
        {
          // take next element as argument (if it doesn't start with a -)
          if (count($this->arguments) && $this->arguments[0][0] != '-')
          {
            $value = array_shift($this->arguments);
          }
          else
          {
            $value = $option->getDefault();
          }
        }

        $this->setOption($option, $value);
        break;
      }

      $this->setOption($option, $value);
    }
  }

  /**
   * Parses a long option.
   *
   * @param string $argument The option argument
   */
  protected function parseLongOption($argument)
  {
    if (false !== strpos($argument, '='))
    {
      list($name, $value) = explode('=', $argument, 2);

      if (!$this->optionSet->hasOption($name))
      {
        $this->errors[] = sprintf('The "--%s" option does not exist.', $name);
        return;
      }

      $option = $this->optionSet->getOption($name);

      if (!$option->acceptParameter())
      {
        $this->errors[] = sprintf('Option "--%s" does not take an argument.', $name);
        $value = true;
      }
    }
    else
    {
      $name = $argument;

      if (!$this->optionSet->hasOption($name))
      {
        $this->errors[] = sprintf('The "--%s" option does not exist.', $name);
        return;
      }

      $option = $this->optionSet->getOption($name);

      if ($option->isParameterRequired())
      {
        $this->errors[] = sprintf('Option "--%s" requires an argument.', $name);
      }

      $value = $option->acceptParameter() ? $option->getDefault() : true;
    }

    $this->setOption($option, $value);
  }

  public function setOption(sfCommandOption $option, $value)
  {
    if ($option->isArray())
    {
      $this->optionValues[$option->getName()][] = $value;
    }
    else
    {
      $this->optionValues[$option->getName()] = $value;
    }
  }
}


text:
#pragma once

#include "../../Graphics/GPUObject.h"
#include "../../Graphics/GraphicsDefs.h"
#include "../../Math/Color.h"
#include "../../Resource/Resource.h"

namespace Atomic
{

static const int MAX_TEXTURE_QUALITY_LEVELS = 3;

class XMLElement;
class XMLFile;

/// Base class for texture resources.
class ATOMIC_API Texture : public Resource, public GPUObject
{
public:
    /// Construct.
    Texture(Context* context);
    /// Destruct.
    virtual ~Texture();

    /// Set number of requested mip levels. Needs to be called before setting size.
    void SetNumLevels(unsigned levels);
    /// Set filtering mode.
    void SetFilterMode(TextureFilterMode filter);
    /// Set addressing mode by texture coordinate.
    void SetAddressMode(TextureCoordinate coord, TextureAddressMode address);
    /// Set shadow compare mode.
    void SetShadowCompare(bool enable);
    /// Set border color for border addressing mode.
    void SetBorderColor(const Color& color);
    /// Set sRGB sampling and writing mode.
    void SetSRGB(bool enable);
    /// Set backup texture to use when rendering to this texture.
    void SetBackupTexture(Texture* texture);
    /// Set mip levels to skip on a quality setting when loading. Ensures higher quality levels do not skip more.
    void SetMipsToSkip(int quality, int mips);
    /// Dirty the parameters.
    void SetParametersDirty();
    /// Update changed parameters to OpenGL. Called by Graphics when binding the texture.
    void UpdateParameters();

    /// Return texture's OpenGL target.
    unsigned GetTarget() const { return target_; }

    /// Return texture format.
    unsigned GetFormat() const { return format_; }

    /// Return whether the texture format is compressed.
    bool IsCompressed() const;

    /// Return number of mip levels.
    unsigned GetLevels() const { return levels_; }

    /// Return width.
    int GetWidth() const { return width_; }

    /// Return height.
    int GetHeight() const { return height_; }

    /// Return height.
    int GetDepth() const { return depth_; }

    /// Return whether parameters are dirty.
    bool GetParametersDirty() const { return parametersDirty_; }

    /// Return filtering mode.
    TextureFilterMode GetFilterMode() const { return filterMode_; }

    /// Return addressing mode by texture coordinate.
    TextureAddressMode GetAddressMode(TextureCoordinate coord) const { return addressMode_[coord]; }

    /// Return whether shadow compare is enabled.
    bool GetShadowCompare() const { return shadowCompare_; }

    /// Return border color.
    const Color& GetBorderColor() const { return borderColor_; }

    /// Return whether is using sRGB sampling and writing.
    bool GetSRGB() const { return sRGB_; }

    /// Return backup texture.
    Texture* GetBackupTexture() const { return backupTexture_; }

    /// Return mip levels to skip on a quality setting when loading.
    int GetMipsToSkip(int quality) const;
    /// Return mip level width, or 0 if level does not exist.
    int GetLevelWidth(unsigned level) const;
    /// Return mip level width, or 0 if level does not exist.
    int GetLevelHeight(unsigned level) const;
    /// Return mip level depth, or 0 if level does not exist.
    int GetLevelDepth(unsigned level) const;

    /// Return texture usage type.
    TextureUsage GetUsage() const { return usage_; }

    /// Return data size in bytes for a rectangular region.
    unsigned GetDataSize(int width, int height) const;
    /// Return data size in bytes for a volume region.
    unsigned GetDataSize(int width, int height, int depth) const;
    /// Return data size in bytes for a pixel or block row.
    unsigned GetRowDataSize(int width) const;
    /// Return number of image components required to receive pixel data from GetData(), or 0 for compressed images.
    unsigned GetComponents() const;
    /// Return the non-internal texture format corresponding to an OpenGL internal format.
    static unsigned GetExternalFormat(unsigned format);
    /// Return the data type corresponding to an OpenGL internal format.
    static unsigned GetDataType(unsigned format);

    /// Set additional parameters from an XML file.
    void SetParameters(XMLFile* xml);
    /// Set additional parameters from an XML element.
    void SetParameters(const XMLElement& element);
    /// Return the corresponding SRGB texture format if supported. If not supported, return format unchanged.
    unsigned GetSRGBFormat(unsigned format);

protected:
    /// Check whether texture memory budget has been exceeded. Free unused materials in that case to release the texture references.
    void CheckTextureBudget(StringHash type);

    /// Create texture.
    virtual bool Create() { return true; }

    /// Texture usage type.
    TextureUsage usage_;
    /// OpenGL target.
    unsigned target_;
    /// Texture format.
    unsigned format_;
    /// Current mip levels.
    unsigned levels_;
    /// Requested mip levels.
    unsigned requestedLevels_;
    /// Texture width.
    int width_;
    /// Texture height.
    int height_;
    /// Texture depth.
    int depth_;
    /// Shadow compare mode.
    bool shadowCompare_;
    /// Parameters dirty flag.
    bool parametersDirty_;
    /// Filtering mode.
    TextureFilterMode filterMode_;
    /// Addressing mode.
    TextureAddressMode addressMode_[MAX_COORDS];
    /// Mip levels to skip when loading per texture quality setting.
    unsigned mipsToSkip_[MAX_TEXTURE_QUALITY_LEVELS];
    /// Border color.
    Color borderColor_;
    /// sRGB sampling and writing mode flag.
    bool sRGB_;
    /// Backup texture.
    SharedPtr<Texture> backupTexture_;
};

}


text:
Sprockets DynamoDB
==================
An asynchronous DynamoDB client and mixin for Tornado applications

|Version| |Status| |Coverage| |License|

Installation
------------
``sprockets-dynamodb`` is available on the Python package index and is installable via pip:

.. code:: bash

    pip install sprockets-dynamodb

Documentation
-------------
Documentation is available at `sprockets-dynamodb.readthedocs.io <https://sprockets-dynamodb.readthedocs.io>`_.

Configuration
-------------
The following table details the environment variable configuration options.

+----------------------------------+--------------------------------------------------------------------------+---------+
| Variable                         | Definition                                                               | Default |
+==================================+==========================================================================+=========+
| ``DYNAMODB_ENDPOINT``            | Override the default DynamoDB HTTP endpoint                              |         |
+----------------------------------+--------------------------------------------------------------------------+---------+
| ``DYNAMODB_MAX_CLIENTS``         | Maximum number of concurrent DynamoDB clients/requests per process       | ``100`` |
+----------------------------------+--------------------------------------------------------------------------+---------+
| ``DYNAMODB_MAX_RETRIES``         | Maximum number retries for transient errors                              | ``3``   |
+----------------------------------+--------------------------------------------------------------------------+---------+
| ``DYANMODB_NO_CREDS_RATE_LIMIT`` | If set to ``true``, a ``sprockets_dynamodb.NoCredentialsException`` will |         |
|                                  | return a 429 response when using ``sprockets_dynamodb.DynamoDBMixin``    |         |
+----------------------------------+--------------------------------------------------------------------------+---------+

Mixin Configuration
^^^^^^^^^^^^^^^^^^^
The ``sprockets_dynamodb.DynamoDBMixin`` class will automatically raise ``HTTPError``
responses for different classes of errors coming from DynamoDB. In addition it will attempt to
work with the `Sprockets InfluxDB <https://github.com/sprockets/sprockets-influxdb>`_ client
to instrument all DynamoDB requests, submitting per request measurements to InfluxDB. It will
attempt to automatically tag measurements with the application/service name if the ``SERVICE``
environment variable is set. It will also tag the measurement if the ``ENVIRONMENT`` environment
variable is set with the environment that the application is running in. Finally, if you are
using the `Sprockets Correlation Mixin <https://github.com/sprockets/sprockets.mixins.correlation>`_,
measurements will automatically be tagged with the correlation ID for a request.

Requirements
------------
-  `Tornado <https://tornadoweb.org>`_
-  `tornado-aws <https://pypi.python.org/pypi/tornado-aws>`_

Version History
---------------
Available at https://sprockets-dynamodb.readthedocs.org/en/latest/history.html

.. |Version| image:: https://img.shields.io/pypi/v/sprockets-dynamodb.svg?
   :target: https://pypi.python.org/pypi/sprockets-dynamodb

.. |Status| image:: https://img.shields.io/travis/sprockets/sprockets-dynamodb.svg?
   :target: https://travis-ci.org/sprockets/sprockets-dynamodb

.. |Coverage| image:: https://img.shields.io/codecov/c/github/sprockets/sprockets-dynamodb.svg?
   :target: https://codecov.io/github/sprockets/sprockets-dynamodb?branch=master

.. |License| image:: https://img.shields.io/pypi/l/sprockets-dynamodb.svg?
   :target: https://sprockets-dynamodb.readthedocs.org


text:
// Copyright (c) Microsoft Corporation. All rights reserved.
// Licensed under the MIT License.
// Code generated by Microsoft (R) AutoRest Code Generator.

package com.azure.resourcemanager.network.fluent.models;

import com.azure.core.annotation.Fluent;
import com.azure.resourcemanager.network.models.AddressSpace;
import com.azure.resourcemanager.network.models.BgpSettings;
import com.azure.resourcemanager.network.models.ProvisioningState;
import com.fasterxml.jackson.annotation.JsonProperty;

/** LocalNetworkGateway properties. */
@Fluent
public final class LocalNetworkGatewayPropertiesFormat {
    /*
     * Local network site address space.
     */
    @JsonProperty(value = "localNetworkAddressSpace")
    private AddressSpace localNetworkAddressSpace;

    /*
     * IP address of local network gateway.
     */
    @JsonProperty(value = "gatewayIpAddress")
    private String gatewayIpAddress;

    /*
     * FQDN of local network gateway.
     */
    @JsonProperty(value = "fqdn")
    private String fqdn;

    /*
     * Local network gateway's BGP speaker settings.
     */
    @JsonProperty(value = "bgpSettings")
    private BgpSettings bgpSettings;

    /*
     * The resource GUID property of the local network gateway resource.
     */
    @JsonProperty(value = "resourceGuid", access = JsonProperty.Access.WRITE_ONLY)
    private String resourceGuid;

    /*
     * The provisioning state of the local network gateway resource.
     */
    @JsonProperty(value = "provisioningState", access = JsonProperty.Access.WRITE_ONLY)
    private ProvisioningState provisioningState;

    /** Creates an instance of LocalNetworkGatewayPropertiesFormat class. */
    public LocalNetworkGatewayPropertiesFormat() {
    }

    /**
     * Get the localNetworkAddressSpace property: Local network site address space.
     *
     * @return the localNetworkAddressSpace value.
     */
    public AddressSpace localNetworkAddressSpace() {
        return this.localNetworkAddressSpace;
    }

    /**
     * Set the localNetworkAddressSpace property: Local network site address space.
     *
     * @param localNetworkAddressSpace the localNetworkAddressSpace value to set.
     * @return the LocalNetworkGatewayPropertiesFormat object itself.
     */
    public LocalNetworkGatewayPropertiesFormat withLocalNetworkAddressSpace(AddressSpace localNetworkAddressSpace) {
        this.localNetworkAddressSpace = localNetworkAddressSpace;
        return this;
    }

    /**
     * Get the gatewayIpAddress property: IP address of local network gateway.
     *
     * @return the gatewayIpAddress value.
     */
    public String gatewayIpAddress() {
        return this.gatewayIpAddress;
    }

    /**
     * Set the gatewayIpAddress property: IP address of local network gateway.
     *
     * @param gatewayIpAddress the gatewayIpAddress value to set.
     * @return the LocalNetworkGatewayPropertiesFormat object itself.
     */
    public LocalNetworkGatewayPropertiesFormat withGatewayIpAddress(String gatewayIpAddress) {
        this.gatewayIpAddress = gatewayIpAddress;
        return this;
    }

    /**
     * Get the fqdn property: FQDN of local network gateway.
     *
     * @return the fqdn value.
     */
    public String fqdn() {
        return this.fqdn;
    }

    /**
     * Set the fqdn property: FQDN of local network gateway.
     *
     * @param fqdn the fqdn value to set.
     * @return the LocalNetworkGatewayPropertiesFormat object itself.
     */
    public LocalNetworkGatewayPropertiesFormat withFqdn(String fqdn) {
        this.fqdn = fqdn;
        return this;
    }

    /**
     * Get the bgpSettings property: Local network gateway's BGP speaker settings.
     *
     * @return the bgpSettings value.
     */
    public BgpSettings bgpSettings() {
        return this.bgpSettings;
    }

    /**
     * Set the bgpSettings property: Local network gateway's BGP speaker settings.
     *
     * @param bgpSettings the bgpSettings value to set.
     * @return the LocalNetworkGatewayPropertiesFormat object itself.
     */
    public LocalNetworkGatewayPropertiesFormat withBgpSettings(BgpSettings bgpSettings) {
        this.bgpSettings = bgpSettings;
        return this;
    }

    /**
     * Get the resourceGuid property: The resource GUID property of the local network gateway resource.
     *
     * @return the resourceGuid value.
     */
    public String resourceGuid() {
        return this.resourceGuid;
    }

    /**
     * Get the provisioningState property: The provisioning state of the local network gateway resource.
     *
     * @return the provisioningState value.
     */
    public ProvisioningState provisioningState() {
        return this.provisioningState;
    }

    /**
     * Validates the instance.
     *
     * @throws IllegalArgumentException thrown if the instance is not valid.
     */
    public void validate() {
        if (localNetworkAddressSpace() != null) {
            localNetworkAddressSpace().validate();
        }
        if (bgpSettings() != null) {
            bgpSettings().validate();
        }
    }
}


text:
/*
 *  Thread: The thread class
 *
 *  Written by:
 *   Author: Steve Parker
 *   Department of Computer Science
 *   University of Utah
 *   Date: June 1997
 *
 *  Copyright (C) 1997 SCI Group
 */

#include <Core/Thread/Thread.h>
#include <Core/Exceptions/Exception.h>
#include <Core/Thread/Parallel.h>
#include <Core/Thread/Runnable.h>
#include <Core/Thread/ThreadError.h>
#include <Core/Thread/ThreadGroup.h>
#include <Core/Thread/Time.h>
#include <Core/Containers/StringUtil.h>
#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string>

#if defined(_AIX)
// Needed for StrCaseCmp on aix 4.3 (on 5.1 we don't need this.)
// currently blue is 4.3.
#  include <strings.h>
#endif

#include <string.h>
#include <sys/types.h>
#include <unistd.h>

#ifdef _WIN32
#define WIN32_LEAN_AND_MEAN
#include <windows.h>
#include <winnt.h>
#include <io.h>
#include <process.h>
#include <imagehlp.h>
#include <psapi.h>

#elif defined(__GNUC__) && defined(__linux)
#include <execinfo.h>
#endif

// provide "C" interface to exitAll
extern "C" {
void exit_all_threads(int rc) {
  Manta::Thread::exitAll(rc);
}
}
namespace Manta {

class ParallelHelper : public Runnable {
  ParallelBase &helper_;
  int proc_;
public:
  ParallelHelper(ParallelBase& helper, int proc)
    : helper_(helper), proc_(proc) {}
  virtual ~ParallelHelper() {}
  virtual void run() {
    helper_.run(proc_);
  }
};

bool Thread::initialized = false;
const char* Thread::defaultAbortMode = "ask";
bool Thread::callExit = true;

bool Thread::isInitialized()
{
  return initialized;
}

void Thread::setDefaultAbortMode(const char* abortMode)
{
  defaultAbortMode = abortMode;
}

Thread::~Thread()
{
    if(runner_){
        runner_->my_thread_=0;
        if(runner_->delete_on_exit)
          delete runner_;
    }
    free(const_cast<char *>(threadname_));
}

Thread::Thread(ThreadGroup* g, const char* name)
{
    group_=g;
    g->addme(this);
    threadname_=strdup(name);
    daemon_=false;
    detached_=false;
    runner_=0;
    cpu_=-1;
    stacksize_ = Thread::DEFAULT_STACKSIZE;
}

void
Thread::run_body()
{
    try {
      runner_->run();
    } catch(const ThreadError& e){
      fprintf(stderr, "Caught unhandled Thread error:\n%s\n",
              e.message());
      Thread::niceAbort();
    } catch(const Exception& e){
      fprintf(stderr, "Caught unhandled exception:\n%s\n",e.message());
        const char *trace = e.stackTrace();
        if (trace)
          fprintf(stderr, "Exception %s", trace);
        Thread::niceAbort();
    } catch(const std::string &e){
      fprintf(stderr, "Caught unhandled string exception:\n%s\n", e.c_str());
      Thread::niceAbort();
    } catch(const char *&e){
      fprintf(stderr, "Caught unhandled char exception:\n%s\n", e);
      Thread::niceAbort();
#ifndef _MSC_VER
    // catch these differently with MS compiler, we can get the whole stack trace, but it must be done with
    // an MS-specific exception handler in a different function
    } catch(...){
      fprintf(stderr, "Caught unhandled exception of unknown type\n");
      Thread::niceAbort();
#endif
    }
}

Thread::Thread(Runnable* runner, const char* name,
               ThreadGroup* group, ActiveState state,
               unsigned long stacksize)
    : runner_(runner),
      threadname_(strdup(name)),
      group_(group),
      stacksize_(stacksize),
      daemon_(false),
      detached_(false),
      cpu_(-1)
{
    if(group_ == 0){
        if(!ThreadGroup::s_default_group)
          Thread::initialize();
        group_=ThreadGroup::s_default_group;
    }

    runner_->my_thread_=this;
    group_->addme(this);
    switch(state){
    case Activated:
      os_start(false);
      activated_=true;
      break;
    case Stopped:
      os_start(true);
      activated_=true;
      break;
    case NotActivated:
      activated_=false;
      priv_=0;
      break;
    }
}

void
Thread::activate(bool stopped)
{
    if(activated_)
      throw ThreadError("Thread is already activated");
    activated_=true;
    os_start(stopped);
}

ThreadGroup*
Thread::getThreadGroup()
{
    return group_;
}

Runnable*
Thread::getRunnable()
{
    return runner_;
}

void
Thread::setDaemon(bool to)
{
    daemon_=to;
    checkExit();
}

bool
Thread::isDaemon() const
{
    return daemon_;
}

bool
Thread::isDetached() const
{
    return detached_;
}

const char*
Thread::getThreadName() const
{
    return threadname_;
}

ThreadGroup*
Thread::parallel(ParallelBase& helper, int nthreads,
                 bool block, ThreadGroup* threadGroup)
{
  if (block && nthreads <= 1)
  {
    helper.run(0);
    return 0;
  }

  ThreadGroup* newgroup=new ThreadGroup("Parallel group", threadGroup);
  if(!block){
    // Extra synchronization to make sure that helper doesn't
    // get destroyed before the threads actually start
    helper.wait_=new Semaphore("Thread::parallel startup wait", 0);
  }
  for(int i=0;i<nthreads;i++){
    char buf[50];
    sprintf(buf, "Parallel thread %d of %d", i, nthreads);
    new Thread(new ParallelHelper(helper, i), buf,
               newgroup, Thread::Stopped);
  }
  newgroup->gangSchedule();
  newgroup->resume();
  if(block){
    newgroup->join();
    delete newgroup;
    return 0;
  } else {
    helper.wait_->down(nthreads);
    delete helper.wait_;
    newgroup->detach();
  }
  return newgroup;
}

void
Thread::niceAbort(void* context /* = 0 */)
{
  fprintf(stderr, getStackTrace(context).c_str());
  const char* smode = getenv("SCI_SIGNALMODE");
  if (!smode)
    smode = defaultAbortMode; //"e";

  Thread* s=Thread::self();
  print_threads();
  fprintf(stderr, "\n");
  fprintf(stderr, "Abort signalled by pid: %d\n", getpid());
  if(s)
    fprintf(stderr, "Occured for thread: \"%s\"\n", s->threadname_);
  else
    fprintf(stderr, "With NULL thread pointer.\n");

  for (;;) {
    if (StrCaseCmp(smode, "ask") == 0) {
      char buf[100];
      fprintf(stderr, "resume(r)/gdb(d)/kill thread(k)/exit(e)? ");
      fflush(stderr);
      while(read(fileno(stdin), buf, 100) <= 0){
        if(errno != EINTR){
          fprintf(stderr, "\nCould not read response, sleeping for 20 seconds.\n");
          Time::waitFor(20.0);
          buf[0]='e';
          exitAll(1);
        }
      }
      switch (buf[0]) {
      case 'r': case 'R':
        smode = "resume";
        break;
      case 'd': case 'D':
        smode = "dbx";
        break;
      case 'k': case 'K':
        smode = "kill";
        break;
      case 'e': case 'E':
        smode = "exit";
        break;
      default:
        break;
      }
    }

    if (StrCaseCmp(smode, "resume") == 0) {
      return;
    } else if (StrCaseCmp(smode, "dbx") == 0) {

#if defined( REDSTORM )
      printf("Error: running debugger at exception is not supported on RedStorm\n");
#else
      char command[500];
      if(getenv("SCI_DBXCOMMAND")){
        sprintf(command, getenv("SCI_DBXCOMMAND"), getpid());
      } else {
#ifdef __sgi
        sprintf(command, "winterm -c dbx -p %d &", getpid());
#else
        sprintf(command, "xterm -e gdb -c %d &", getpid());
#endif
      }
      system(command);
      smode = "ask";
#endif
    } else if (StrCaseCmp(smode, "kill") == 0) {
      exit();
    } else if (StrCaseCmp(smode, "exit") == 0) {
      exitAll(1);
    } else {
      fprintf(stderr, "Unrecognized option, exiting\n");
      smode = "exit";
    }
  }
}

int
Thread::couldBlock(const char* why)
{
  if(!initialized)
    Thread::initialize();
  Thread_private* p=Thread::self()->priv_;
  return push_bstack(p, BLOCK_ANY, why);
}

void
Thread::couldBlockDone(int restore)
{
  Thread_private* p=Thread::self()->priv_;
  pop_bstack(p, restore);
}

unsigned long
Thread::getStackSize() const
{
  return stacksize_;
}

void
Thread::setStackSize(unsigned long stacksize)
{
  if(activated_)
    throw ThreadError("Cannot change stack size on a running thread");
  stacksize_=stacksize;
}

/*
 * Return the statename for p
 */
const char*
Thread::getStateString(ThreadState state)
{
  switch(state) {
  case STARTUP:
    return "startup";
  case RUNNING:
    return "running";
  case IDLE:
    return "idle";
  case SHUTDOWN:
    return "shutting down";
  case BLOCK_SEMAPHORE:
    return "blocking on semaphore";
  case PROGRAM_EXIT:
    return "waiting for program exit";
  case JOINING:
    return "joining with thread";
  case BLOCK_MUTEX:
    return "blocking on mutex";
  case BLOCK_ANY:
    return "blocking";
  case DIED:
    return "died";
  case BLOCK_BARRIER:
    return "spinning in barrier";
  default:
    return "UNKNOWN";
  }
}

} // End namespace Manta


text:
module(..., package.seeall)

local constants = require("apps.lwaftr.constants")
local fragmentv6 = require("apps.lwaftr.fragmentv6")
local fragv6_h = require("apps.lwaftr.fragmentv6_hardened")
local ndp = require("apps.lwaftr.ndp")
local lwutil = require("apps.lwaftr.lwutil")
local icmp = require("apps.lwaftr.icmp")
local lwcounter = require("apps.lwaftr.lwcounter")

local ethernet = require("lib.protocol.ethernet")
local ipv6 = require("lib.protocol.ipv6")
local checksum = require("lib.checksum")
local packet = require("core.packet")
local counter = require("core.counter")
local lib = require("core.lib")
local link = require("core.link")
local engine = require("core.app")

local bit = require("bit")
local ffi = require("ffi")

local receive, transmit = link.receive, link.transmit
local wr16 = lwutil.wr16
local is_ipv6, is_ipv6_fragment = lwutil.is_ipv6, lwutil.is_ipv6_fragment
local htons = lib.htons

local ipv6_fixed_header_size = constants.ipv6_fixed_header_size

local proto_icmpv6 = constants.proto_icmpv6
local ethernet_header_size = constants.ethernet_header_size
local o_icmpv6_header = ethernet_header_size + ipv6_fixed_header_size
local o_icmpv6_msg_type = o_icmpv6_header + constants.o_icmpv6_msg_type
local o_icmpv6_checksum = o_icmpv6_header + constants.o_icmpv6_checksum
local icmpv6_echo_request = constants.icmpv6_echo_request
local icmpv6_echo_reply = constants.icmpv6_echo_reply
local ehs = constants.ethernet_header_size

ReassembleV6 = {}
Fragmenter = {}
NDP = {}
ICMPEcho = {}

function ReassembleV6:new(conf)
   local max_ipv6_reassembly_packets = conf.max_ipv6_reassembly_packets
   local max_fragments_per_reassembly_packet = conf.max_fragments_per_reassembly_packet
   local o = {
      counters = lwcounter.init_counters(),
      ctab = fragv6_h.initialize_frag_table(max_ipv6_reassembly_packets,
         max_fragments_per_reassembly_packet),
   }
   counter.set(o.counters["memuse-ipv6-frag-reassembly-buffer"],
               o.ctab:get_backing_size())
   return setmetatable(o, {__index = ReassembleV6})
end

function ReassembleV6:cache_fragment(fragment)
   return fragv6_h.cache_fragment(self.ctab, fragment)
end

function ReassembleV6:push ()
   local input, output = self.input.input, self.output.output
   local errors = self.output.errors

   for _ = 1, link.nreadable(input) do
      local pkt = receive(input)
      if is_ipv6_fragment(pkt) then
         counter.add(self.counters["in-ipv6-frag-needs-reassembly"])
         local status, maybe_pkt, ejected = self:cache_fragment(pkt)
         if ejected then
            counter.add(self.counters["drop-ipv6-frag-random-evicted"])
         end

         if status == fragv6_h.REASSEMBLY_OK then
            counter.add(self.counters["in-ipv6-frag-reassembled"])
            transmit(output, maybe_pkt)
         elseif status == fragv6_h.FRAGMENT_MISSING then
            -- Nothing useful to be done yet, continue
         elseif status == fragv6_h.REASSEMBLY_INVALID then
            counter.add(self.counters["drop-ipv6-frag-invalid-reassembly"])
            if maybe_pkt then -- This is an ICMP packet
               transmit(errors, maybe_pkt)
            end
         else -- unreachable
            packet.free(pkt)
         end
      else
         -- Forward all packets that aren't IPv6 fragments.
         counter.add(self.counters["in-ipv6-frag-reassembly-unneeded"])
         transmit(output, pkt)
      end
   end
end

function Fragmenter:new(conf)
   local o = {
      counters = lwcounter.init_counters(),
      mtu = assert(conf.mtu),
   }
   return setmetatable(o, {__index=Fragmenter})
end

function Fragmenter:push ()
   local input, output = self.input.input, self.output.output

   local mtu = self.mtu

   for _ = 1, link.nreadable(input) do
      local pkt = receive(input)
      if pkt.length > mtu + ehs and is_ipv6(pkt) then
         -- It's possible that the IPv6 packet has an IPv4 packet as
         -- payload, and that payload has the Don't Fragment flag set.
         -- However ignore this; the fragmentation policy of the L3
         -- protocol (in this case, IP) doesn't affect the L2 protocol.
         -- We always fragment.
         local unfragmentable_header_size = ehs + ipv6_fixed_header_size
         local pkts = fragmentv6.fragment(pkt, unfragmentable_header_size, mtu)
         for i=1,#pkts do
            counter.add(self.counters["out-ipv6-frag"])
            transmit(output, pkts[i])
         end
      else
         counter.add(self.counters["out-ipv6-frag-not"])
         transmit(output, pkt)
      end
   end
end

function NDP:new(conf)
   local o = setmetatable({}, {__index=NDP})
   o.conf = conf
   -- TODO: verify that the src and dst ipv6 addresses and src mac address
   -- have been provided, in pton format.
   if not conf.dst_eth then
      self.ns_pkt = ndp.form_ns(conf.src_eth, conf.src_ipv6, conf.dst_ipv6)
      self.ns_interval = 3 -- Send a new NS every three seconds.
   end
   o.dst_eth = conf.dst_eth -- Intentionally nil if to request by NS
   return o
end

function NDP:maybe_send_ns_request (output)
   if self.dst_eth then return end
   self.next_ns_time = self.next_ns_time or engine.now()
   if self.next_ns_time <= engine.now() then
      print(("NDP: Resolving '%s'"):format(ipv6:ntop(self.conf.dst_ipv6)))
      self:send_ns(output)
      self.next_ns_time = engine.now() + self.ns_interval
   end
end

function NDP:send_ns (output)
   transmit(output, packet.clone(self.ns_pkt))
end

function NDP:push()
   local isouth, osouth = self.input.south, self.output.south
   local inorth, onorth = self.input.north, self.output.north

   -- TODO: do unsolicited neighbor advertisement on start and on
   -- configuration reloads?
   -- This would be an optimization, not a correctness issue
   self:maybe_send_ns_request(osouth)

   for _ = 1, link.nreadable(isouth) do
      local p = receive(isouth)
      if ndp.is_ndp(p) then
         if not self.dst_eth and ndp.is_solicited_neighbor_advertisement(p) then
            local dst_ethernet = ndp.get_dst_ethernet(p, {self.conf.dst_ipv6})
            if dst_ethernet then
               print(("NDP: '%s' resolved (%s)"):format(ipv6:ntop(self.conf.dst_ipv6),
                                                        ethernet:ntop(dst_ethernet)))
               self.dst_eth = dst_ethernet
            end
            packet.free(p)
         elseif ndp.is_neighbor_solicitation_for_addr(p, self.conf.src_ipv6) then
            local snap = ndp.form_nsolicitation_reply(self.conf.src_eth, self.conf.src_ipv6, p)
            if snap then
               transmit(osouth, snap)
            end
            packet.free(p)
         else -- TODO? incoming NDP that we don't handle; drop it silently
            packet.free(p)
         end
      else
         transmit(onorth, p)
      end
   end

   for _ = 1, link.nreadable(inorth) do
      local p = receive(inorth)
      if not self.dst_eth then
         -- drop all southbound packets until the next hop's ethernet address is known
          packet.free(p)
      else
          lwutil.set_dst_ethernet(p, self.dst_eth)
          transmit(osouth, p)
      end
   end
end

function ICMPEcho:new(conf)
   local addresses = {}
   if conf.address then
      addresses[ffi.string(conf.address, 16)] = true
   end
   if conf.addresses then
      for _, v in ipairs(conf.addresses) do
         addresses[ffi.string(v, 16)] = true
      end
   end
   return setmetatable({addresses = addresses}, {__index = ICMPEcho})
end

function ICMPEcho:push()
   local l_in, l_out, l_reply = self.input.south, self.output.north, self.output.south

   for _ = 1, link.nreadable(l_in) do
      local out, pkt = l_out, receive(l_in)

      if icmp.is_icmpv6_message(pkt, icmpv6_echo_request, 0) then
         local pkt_ipv6 = ipv6:new_from_mem(pkt.data + ethernet_header_size,
                                            pkt.length - ethernet_header_size)
         local pkt_ipv6_dst = ffi.string(pkt_ipv6:dst(), 16)
         if self.addresses[pkt_ipv6_dst] then
            ethernet:new_from_mem(pkt.data, ethernet_header_size):swap()

            -- Swap IP source/destination
            pkt_ipv6:dst(pkt_ipv6:src())
            pkt_ipv6:src(pkt_ipv6_dst)

            -- Change ICMP message type
            pkt.data[o_icmpv6_msg_type] = icmpv6_echo_reply

            -- Recalculate checksums
            wr16(pkt.data + o_icmpv6_checksum, 0)
            local ph_len = pkt.length - o_icmpv6_header
            local ph = pkt_ipv6:pseudo_header(ph_len, proto_icmpv6)
            local csum = checksum.ipsum(ffi.cast("uint8_t*", ph), ffi.sizeof(ph), 0)
            csum = checksum.ipsum(pkt.data + o_icmpv6_header, 4, bit.bnot(csum))
            csum = checksum.ipsum(pkt.data + o_icmpv6_header + 4,
                                  pkt.length - o_icmpv6_header - 4,
                                  bit.bnot(csum))
            wr16(pkt.data + o_icmpv6_checksum, htons(csum))

            out = l_reply
         end
      end

      transmit(out, pkt)
   end

   l_in, l_out = self.input.north, self.output.south
   for _ = 1, link.nreadable(l_in) do
      transmit(l_out, receive(l_in))
   end
end


text:
Given /^I have no images$/ do
  Image.destroy_all
end

When /^I attach the image at "([^\"]*)"$/ do |file_path|
  attach_file('image_image', File.join(File.expand_path('../../uploads/', __FILE__), file_path))
end

Then /^the image "([^\"]*)" should have uploaded successfully$/ do |file_name|
  Image.find_by_image_name(file_name).nil?.should == false
end

Then /^the image should have size "([^\"]*)"$/ do |size|
  Image.first.size.should == size.to_i
end

Then /^the image should have width "([^\"]*)"$/ do |width|
  Image.first.width.should == width.to_i
end

Then /^the image should have height "([^\"]*)"$/ do |height|
  Image.first.height.should == height.to_i
end

Then /^the image should have mime_type "([^\"]*)"$/ do |mime_type|
  Image.first.mime_type.should == mime_type.to_s
end

Then /^I should have ([0-9]+) images?$/ do |number|
  Image.count.should == number.to_i
end

When /^I upload the image at "([^\"]*)"$/ do |file_path|
  original_stderr = $stderr.dup
  $stderr.reopen(Tempfile.new('stderr'))
  visit new_admin_image_path
  attach_file('image_image', File.join(File.expand_path('../../uploads/', __FILE__), file_path))
  click_button 'Save'
  $stderr.reopen(original_stderr)
end


text:
import time
from collections import namedtuple
from datetime import timedelta

import requests
from django.contrib.sessions.models import Session
from django.db import connection
from django.db.models import Count
from django.db.models import Sum
from django.db.utils import OperationalError
from django.utils import timezone

from kolibri.core.analytics import SUPPORTED_OS
from kolibri.core.content.models import ChannelMetadata
from kolibri.core.logger.models import ContentSessionLog
from kolibri.core.logger.models import UserSessionLog
from kolibri.utils.server import NotRunning
from kolibri.utils.server import PID_FILE

try:
    import kolibri.utils.pskolibri as psutil
except NotImplementedError:
    # This module can't work on this OS
    psutil = None

def get_db_info():
    """
    Returns information about the sessions and users the current
    Kolibri server has in use

    """
    # Users information
    active_sessions = "unknown"
    active_users = active_users_minute = None
    try:
        connection.ensure_connection()
        # Sessions active in the last 10 minutes (includes guest accesses):
        active_sessions = str(
            Session.objects.filter(expire_date__gte=timezone.now()).count()
        )
        last_ten_minutes = timezone.now() - timedelta(minutes=10)
        last_minute = timezone.now() - timedelta(minutes=1)
        # Active logged users:
        active_users = str(
            UserSessionLog.objects.filter(
                last_interaction_timestamp__gte=last_ten_minutes
            ).count()
        )
        # Logged users with activity in the last minute:
        active_users_minute = str(
            UserSessionLog.objects.filter(
                last_interaction_timestamp__gte=last_minute
            ).count()
        )
    except OperationalError:
        print("Database unavailable, impossible to retrieve users and sessions info")

    return (active_sessions, active_users, active_users_minute)

def get_channels_usage_info():
    """
    Scan the channels Kolibri has installed, getting information on how many times
    their resources have been accessed and how long they have been used
    :returns: List containing namedtuples, with each channel: id, name, accesses and time spent
    """
    channels_info = []
    ChannelsInfo = namedtuple("ChannelsInfo", "id name accesses time_spent")

    try:
        connection.ensure_connection()
        channels = ChannelMetadata.objects.values("id", "name")
        channel_stats = ContentSessionLog.objects.values("channel_id").annotate(
            time_spent=Sum("time_spent"), total=Count("channel_id")
        )
        for channel in channels:
            stats = channel_stats.filter(channel_id=channel["id"])
            if stats:
                channels_info.append(
                    ChannelsInfo(
                        id=channel["id"],
                        name=channel["name"],
                        accesses=str(stats[0]["total"]),
                        time_spent="{:.2f} s".format(stats[0]["time_spent"]),
                    )
                )
            else:
                channels_info.append(
                    ChannelsInfo(
                        id=channel["id"],
                        name=channel["name"],
                        accesses="0",
                        time_spent="0.00 s",
                    )
                )
    except OperationalError:
        print("Database unavailable, impossible to retrieve channels usage info")
    return channels_info

def get_requests_info():
    """
    Returns timing information on some Kolibri pages that can be hit without credentials
    :returns: tuple of strings containing time in seconds when requesting
              - Kolibri homepage
              - Kolibri recommended channels
              - Kolibri channels list
    """

    def format_url(url, base_url):
        formatted = "{base_url}{url}&contentCacheKey={cache}".format(
            base_url=base_url, url=url, cache=time.time()
        )
        return formatted

    _, port = get_kolibri_process_info()
    if port:
        base_url = "http://localhost:{}".format(port)
        homepage_time = "{:.2f} s".format(
            requests.get(base_url).elapsed.total_seconds()
        )
        recommended_url = format_url(
            "/api/content/contentnode_slim/popular/?include_coach_content=false",
            base_url,
        )
        recommended_time = "{:.2f} s".format(
            requests.get(recommended_url).elapsed.total_seconds()
        )
        channels_url = format_url("/api/content/channel/?available=true", base_url)
        channels_time = "{:.2f} s".format(
            requests.get(channels_url).elapsed.total_seconds()
        )
    else:
        homepage_time = recommended_time = channels_time = None

    return (homepage_time, recommended_time, channels_time)

def get_machine_info():
    """
    Gets information on the memory, cpu and processes in the server
    :returns: tuple of strings containing cpu percentage, used memory, free memory and number of active processes
    """
    if not SUPPORTED_OS:
        return (None, None, None, None)
    used_cpu = str(psutil.cpu_percent())
    used_memory = str(psutil.virtual_memory().used / pow(10, 6))  # In Megabytes
    total_memory = str(psutil.virtual_memory().total / pow(10, 6))  # In Megabytes
    total_processes = str(len(psutil.pids()))

    return (used_cpu, used_memory, total_memory, total_processes)

def get_kolibri_process_info():
    """
    Return information on the Kolibri process running in the machine
    :returns: tuple of integers containing PID and TCP Port of
              the running (if any) Kolibri server in this same machine
    """
    kolibri_pid = None
    kolibri_port = None
    try:
        with open(PID_FILE, "r") as f:
            kolibri_pid = int(f.readline())
            kolibri_port = int(f.readline())
    except IOError:
        pass  # Kolibri PID file does not exist
    except ValueError:
        pass  # corrupted Kolibri PID file
    return (kolibri_pid, kolibri_port)

def get_kolibri_process_cmd():
    """
    Retrieve from the OS the command line executed to run Kolibri server
    :returns: tuple with command line and its arguments
    """
    if not SUPPORTED_OS:
        return None
    kolibri_pid, _ = get_kolibri_process_info()
    try:
        kolibri_proc = psutil.Process(kolibri_pid)
    except psutil.NoSuchProcess:
        # Kolibri server is not running
        raise NotRunning(0)
    return kolibri_proc.cmdline()

def get_kolibri_use(development=False):
    """
    Gets information on the memory and cpu usage of the current Kolibri process
    :returns: tuple of strings containing cpu percentage and virtual memory used (in Mb)
    """
    if not SUPPORTED_OS:
        return (None, None)
    kolibri_mem = kolibri_cpu = "None"
    kolibri_pid, _ = get_kolibri_process_info()

    if kolibri_pid:
        try:
            kolibri_proc = psutil.Process(kolibri_pid)
            kolibri_mem = str(kolibri_proc.memory_info().rss / pow(10, 6))
            kolibri_cpu = str(kolibri_proc.cpu_percent())
        except psutil.NoSuchProcess:
            # Kolibri server is not running
            raise NotRunning(0)

    return (kolibri_cpu, kolibri_mem)