/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.examples.pi;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileWriter;
import java.io.IOException;
import java.io.InputStreamReader;
import java.io.PrintStream;
import java.io.PrintWriter;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;

import org.apache.hadoop.examples.pi.DistSum.Machine;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.util.ToolRunner;

/** Utility methods */
public class Util {
  /** Output stream */
  public static final PrintStream out = System.out; 
  /** Error stream */
  public static final PrintStream err = System.out; 

  /** Timer */
  public static class Timer {
    private final boolean isAccumulative;
    private final long start = System.currentTimeMillis();
    private long previous = start;
  
    /** Timer constructor
     * @param isAccumulative  Is accumulating the time duration?
     */
    public Timer(boolean isAccumulative) {
      this.isAccumulative = isAccumulative;
      final StackTraceElement[] stack = Thread.currentThread().getStackTrace();
      final StackTraceElement e = stack[stack.length - 1];
      out.println(e + " started at " + new Date(start));
    }
  
    /** Same as tick(null). */
    public long tick() {return tick(null);}

    /**
     * Tick
     * @param s Output message.  No output if it is null.
     * @return delta
     */
    public synchronized long tick(String s) {
      final long t = System.currentTimeMillis();
      final long delta = t - (isAccumulative? start: previous);
      if (s != null) {
        out.format("%15dms (=%-15s: %s\n", delta, millis2String(delta) + ")", s);
        out.flush();
      }
      previous = t;
      return delta;
    }
  }
    
  /** Covert milliseconds to a String. */
  public static String millis2String(long n) {
    if (n < 0)
      return "-" + millis2String(-n);
    else if (n < 1000)
      return n + "ms";

    final StringBuilder b = new StringBuilder();
    final int millis = (int)(n % 1000L);
    if (millis != 0)
      b.append(String.format(".%03d", millis)); 
    if ((n /= 1000) < 60)
      return b.insert(0, n).append("s").toString();

    b.insert(0, String.format(":%02d", (int)(n % 60L)));
    if ((n /= 60) < 60)
      return b.insert(0, n).toString();

    b.insert(0, String.format(":%02d", (int)(n % 60L)));
    if ((n /= 60) < 24)
      return b.insert(0, n).toString();

    b.insert(0, n % 24L);
    final int days = (int)((n /= 24) % 365L);
    b.insert(0, days == 1? " day ": " days ").insert(0, days);
    if ((n /= 365L) > 0)
      b.insert(0, n == 1? " year ": " years ").insert(0, n);

    return b.toString();
  }

  /** Covert a String to a long.  
   * This support comma separated number format.
   */
  public static long string2long(String s) {
    return Long.parseLong(s.trim().replace(",", ""));
  }

  /** Covert a long to a String in comma separated number format. */  
  public static String long2string(long n) {
    if (n < 0)
      return "-" + long2string(-n);
    
    final StringBuilder b = new StringBuilder();
    for(; n >= 1000; n = n/1000)
      b.insert(0, String.format(",%03d", n % 1000));
    return n + b.toString();    
  }

  /** Parse a variable. */  
  public static long parseLongVariable(final String name, final String s) {
    return string2long(parseStringVariable(name, s));
  }

  /** Parse a variable. */  
  public static String parseStringVariable(final String name, final String s) {
    if (!s.startsWith(name + '='))
      throw new IllegalArgumentException("!s.startsWith(name + '='), name="
          + name + ", s=" + s);
    return s.substring(name.length() + 1);
  }

  /** Execute the callables by a number of threads */
  public static <T, E extends Callable<T>> void execute(int nThreads, List<E> callables
      ) throws InterruptedException, ExecutionException {
    final ExecutorService executor = Executors.newFixedThreadPool(nThreads); 
    final List<Future<T>> futures = executor.invokeAll(callables);
    for(Future<T> f : futures)
      f.get();
  }

  /** Print usage messages */
  public static int printUsage(String[] args, String usage) {
    err.println("args = " + Arrays.asList(args));
    err.println();
    err.println("Usage: java " + usage);
    err.println();
    ToolRunner.printGenericCommandUsage(err);
    return -1;
  }

  /** Combine a list of items. */
  public static <T extends Combinable<T>> List<T> combine(Collection<T> items) {
    final List<T> sorted = new ArrayList<T>(items);
    if (sorted.size() <= 1)
      return sorted;

    Collections.sort(sorted);
    final List<T> combined = new ArrayList<T>(items.size());
    T prev = sorted.get(0);
    for(int i = 1; i < sorted.size(); i++) {
      final T curr = sorted.get(i);
      final T c = curr.combine(prev);

      if (c != null)
        prev = c;
      else {
        combined.add(prev);
        prev = curr;
      }
    }
    combined.add(prev);
    return combined;
  }

  /** Check local directory. */
  public static void checkDirectory(File dir) {
    if (!dir.exists())
      if (!dir.mkdirs())
        throw new IllegalArgumentException("!dir.mkdirs(), dir=" + dir);
    if (!dir.isDirectory())
      throw new IllegalArgumentException("dir (=" + dir + ") is not a directory.");
  }

  private static final SimpleDateFormat DATE_FORMAT = new SimpleDateFormat("-yyyyMMdd-HHmmssSSS");
  /** Create a writer of a local file. */
  public static PrintWriter createWriter(File dir, String prefix) throws IOException {
    checkDirectory(dir);

    for(;;) {
      final File f = new File(dir,
          prefix + DATE_FORMAT.format(new Date(System.currentTimeMillis())) + ".txt");
      if (!f.exists())
        return new PrintWriter(new FileWriter(f));

      try {Thread.sleep(10);} catch (InterruptedException e) {}
    }
  }

  /** Print a "bits skipped" message. */
  public static void printBitSkipped(final long b) {
    out.println();
    out.println("b = " + long2string(b)
        + " (" + (b < 2? "bit": "bits") + " skipped)");
  }

  /** Convert a pi value to a String. */
  public static String pi2string(final double pi, final long terms) {
    final long value = (long)(pi * (1L << DOUBLE_PRECISION));
    final int acc_bit = accuracy(terms, false);
    final int acc_hex = acc_bit/4;
    final int shift = DOUBLE_PRECISION - acc_bit;
    return String.format("%0" + acc_hex + "X %0" + (13-acc_hex) + "X (%d hex digits)",
        value >> shift, value & ((1 << shift) - 1), acc_hex);
  }

  static final int DOUBLE_PRECISION = 52; //mantissa size
  static final int MACHEPS_EXPONENT = DOUBLE_PRECISION + 1;
  /** Estimate accuracy. */
  public static int accuracy(final long terms, boolean print) {
    final double error = terms <= 0? 2: (Math.log(terms) / Math.log(2)) / 2;
    final int bits = MACHEPS_EXPONENT - (int)Math.ceil(error);
    if (print)
      out.println("accuracy: bits=" + bits + ", terms=" + long2string(terms) + ", error exponent=" + error);
    return bits - bits%4;
  }

  private static final String JOB_SEPARATION_PROPERTY = "pi.job.separation.seconds";
  private static final Semaphore JOB_SEMAPHORE = new Semaphore(1);

  /** Run a job. */
  static void runJob(String name, Job job, Machine machine, String startmessage, Util.Timer timer) {
    JOB_SEMAPHORE.acquireUninterruptibly();
    Long starttime = null;
    try {
      try {
        starttime = timer.tick("starting " + name + " ...\n  " + startmessage);

        //initialize and submit a job
        machine.init(job);
        job.submit();
  
        // Separate jobs
        final long sleeptime = 1000L * job.getConfiguration().getInt(JOB_SEPARATION_PROPERTY, 10);
        if (sleeptime > 0) {
          Util.out.println(name + "> sleep(" + Util.millis2String(sleeptime) + ")");
          Thread.sleep(sleeptime);
        }
      } finally {
        JOB_SEMAPHORE.release();
      }
  
      if (!job.waitForCompletion(false))
        throw new RuntimeException(name + " failed.");
    } catch(Exception e) {
      throw e instanceof RuntimeException? (RuntimeException)e: new RuntimeException(e);
    } finally {
      if (starttime != null)
        timer.tick(name + "> timetaken=" + Util.millis2String(timer.tick() - starttime));
    }
  }

  /** Read job outputs */
  static List<TaskResult> readJobOutputs(FileSystem fs, Path outdir) throws IOException {
    final List<TaskResult> results = new ArrayList<TaskResult>();
    for(FileStatus status : fs.listStatus(outdir)) {
      if (status.getPath().getName().startsWith("part-")) {
        final BufferedReader in = new BufferedReader(new InputStreamReader(fs.open(status.getPath())));
        try {
          for(String line; (line = in.readLine()) != null; )
            results.add(TaskResult.valueOf(line));
        }
        finally {
          in.close();
        }
      }
    }
    if (results.isEmpty())
      throw new IOException("Output not found");
    return results;
  }
  
  /** Write results */
  static void writeResults(String name, List<TaskResult> results, FileSystem fs, String dir) throws IOException {
    final Path outfile = new Path(dir, name + ".txt");
    Util.out.println(name + "> writing results to " + outfile);
    final PrintStream out = new PrintStream(fs.create(outfile), true);
    try {
      for(TaskResult r : results)
        out.println(r);
    }
    finally {
      out.close();
    }
  }

  /** Create a directory. */
  static boolean createNonexistingDirectory(FileSystem fs, Path dir) throws IOException {
    if (fs.exists(dir)) {
      Util.err.println("dir (= " + dir + ") already exists.");
      return false;
    } else if (!fs.mkdirs(dir)) {
      throw new IOException("Cannot create working directory " + dir);
    }
    fs.setPermission(dir, new FsPermission((short)0777));
    return true;
  }
}