// BSD License (http://www.galagosearch.org/license)
   package org.galagosearch.tupleflow;
   
   import java.io.File;
   import java.io.FileNotFoundException;
   import java.io.IOException;
   import java.lang.management.ManagementFactory;
   import java.lang.management.MemoryMXBean;
   import java.lang.management.MemoryNotificationInfo;
  import java.lang.management.MemoryPoolMXBean;
  import java.lang.management.MemoryType;
  import java.lang.management.MemoryUsage;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.Iterator;
  import java.util.List;
  import java.util.PriorityQueue;
  import java.util.logging.Logger;
  import javax.management.ListenerNotFoundException;
  import javax.management.Notification;
  import javax.management.NotificationEmitter;
  import javax.management.NotificationListener;
  import org.galagosearch.tupleflow.execution.ErrorHandler;
  import org.galagosearch.tupleflow.execution.Verification;
  
  /***
   * <p>
   * This class sorts an incoming stream of objects in some specified order.
   * When this object is closed (by calling the close method), the sorted
   * objects are then sent in sorted order to the next stage of the 
   * Processor chain.
   * </p>
   *
   * <p>
   * Since there may be many objects submitted to the Sorter (more than
   * will fit in main memory), the object may create temporary files
   * to store partially sorted results.  The path used for these temporary
   * files is specified in the TempPath Java preferences variable.
   * </p>
   *
   * <p>
   * In many instances, Sorters are used to generate streams of data that
   * are then used to create aggregate statistics.  For instance, suppose
   * we want to compute the monthly sales of a particular corporation, separated
   * by region.  We can feed a set of transactions to the Sorter, each containing
   * a dollar amount and the region it came from, e.g.:
   * <ul>
   *      <li>($5.39, South)</li>
   *      <li>($2.24, North)</li>
   *      <li>($1.50, South)</li>
   * </ul>
   * If we sort this list by region name:
   * <ul>
   *      <li>($2.24, North)</li>
   *      <li>($5.39, South)</li>
   *      <li>($1.50, South)</li>
   * </ul>
   * It's now very easy to add up totals for each region (since all data for each
   * region is adjacent in the list).</p>
   *
   * <p>
   * In these kinds of aggregate applications, it may be more efficient to 
   * provide the Sorter with a Reducer object.  A Reducer is an object that
   * transforms <i>n</i> sorted objects of type T into some (hopefully)
   * smaller number of objects, also of type T.  In the example above, we could
   * write a reducer that turned those three transactions into:
   * <ul>
   *      <li>($2.24, North)</li>
   *      <li>($6.89, South)</li>
   * </ul>
   * which would be equivalent for this application.  Using a Reducer allows
   * the application to buffer fewer items and hopefully reduce the reliance
   * on the disk during sorting.</p>
   *
   * @author Trevor Strohman
   */
  public class Sorter<T> extends StandardStep<T, T> implements NotificationListener {
      private int limit;
      private int fileLimit = 20;
      private volatile boolean flushRequested = false;
      private ArrayList<T> objects;
      private ArrayList<List<T>> runs;
      private long runsCount = 0;
      private Logger logger = Logger.getLogger(Sorter.class.toString());
      private ArrayList<File> temporaryFiles;
      private Order<T> order;
      private Comparator<T> lessThanCompare;
      private Reducer<T> reducer;
      private static int reduceInterval = 100 * 1000;
      private static int combineBufferSize = 100 * 1000;
      private static int defaultObjectLimit = 50 * 1000 * 1000;
      private Counter filesWritten = null;
      private Counter sorterCombineSteps;
  
      public Sorter(Order<T> order) {
          this(defaultObjectLimit, order, null, null);
      }
 
     public Sorter(Order<T> order, Reducer<T> reducer) {
         this(defaultObjectLimit, order, reducer, null);
     }
 
     public Sorter(int limit, Order<T> order) {
         this(limit, order, null, null);
     }
 
     public Sorter(int limit, Order<T> order, Reducer<T> reducer) {
         this(limit, order, reducer, null);
     }
 
     public Sorter(int limit, Order<T> order, Reducer<T> reducer, Processor<T> processor) {
         this.limit = limit;
         this.order = order;
         this.processor = processor;
         this.reducer = reducer;
         this.objects = new ArrayList<T>();
         this.runs = new ArrayList<List<T>>();
         this.temporaryFiles = new ArrayList<File>();
         this.lessThanCompare = order.lessThan();
         this.flushRequested = false;
 
         requestMemoryWarnings();
     }
 
     @SuppressWarnings("unchecked")
     public Sorter(TupleFlowParameters parameters)
             throws ClassNotFoundException, InstantiationException,
                    IllegalAccessException, IOException {
         String className = parameters.getXML().get("class");
         String[] orderSpec = parameters.getXML().get("order").split(" ");
 
         Class clazz = Class.forName(className);
         Type<T> typeInstance = (Type<T>) clazz.newInstance();
         this.order = typeInstance.getOrder(orderSpec);
         this.limit = (int) parameters.getXML().get("object-limit", defaultObjectLimit);
         this.reducer = null;
         this.flushRequested = false;
 
         if (parameters.getXML().containsKey("reducer")) {
             Class reducerClass = Class.forName(parameters.getXML().get("reducer"));
             this.reducer = (Reducer<T>) reducerClass.newInstance();
         }
 
         this.processor = null;
         this.objects = new ArrayList<T>();
         this.runs = new ArrayList<List<T>>();
         this.temporaryFiles = new ArrayList<File>();
         this.lessThanCompare = order.lessThan();
 
         this.filesWritten = parameters.getCounter("Sorter Files Written");
         this.sorterCombineSteps = parameters.getCounter("Sorter Combine Steps");
 
         requestMemoryWarnings();
     }
 
     public void requestMemoryWarnings() {
         List<MemoryPoolMXBean> pools = ManagementFactory.getMemoryPoolMXBeans();
         long maxPoolSize = 0;
         MemoryPoolMXBean biggestPool = null;
 
         for (MemoryPoolMXBean pool : pools) {
             if (pool.getType() != MemoryType.HEAP) {
                 continue;
             }
             MemoryUsage usage = pool.getUsage();
 
             if (pool.isUsageThresholdSupported() &&
                     usage.getMax() > maxPoolSize) {
                 maxPoolSize = usage.getMax();
                 biggestPool = pool;
             }
         }
 
         if (biggestPool != null) {
             biggestPool.setUsageThreshold((long) (maxPoolSize * 0.7));
             MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
             NotificationEmitter emitter = (NotificationEmitter) memoryBean;
             emitter.addNotificationListener(this, null, null);
         } else {
             throw new RuntimeException("Memory monitoring is not supported.");
         }
     }
 
     public void removeMemoryWarnings() {
         try {
             MemoryMXBean memoryBean = ManagementFactory.getMemoryMXBean();
             NotificationEmitter emitter = (NotificationEmitter) memoryBean;
             emitter.removeNotificationListener(this);
         } catch (ListenerNotFoundException e) {
             // do nothing
         }
     }
 
     public void handleNotification(Notification notification, Object handback) {
         if (notification.getType().equals(MemoryNotificationInfo.MEMORY_THRESHOLD_EXCEEDED)) {
             flushRequested = true;
             final Sorter f = this;
 
             Thread t = new Thread() {
                 @Override
                 public void run() {
                     try {
                         f.flush();
                     } catch (IOException e) {
                         logger.severe(e.toString());
                     }
                 }
             };
 
             t.start();
         }
     }
 
     public static void verify(TupleFlowParameters fullParameters, ErrorHandler handler) {
         Parameters parameters = fullParameters.getXML();
         String[] requiredParameters = {"order", "class"};
 
         if (!Verification.requireParameters(requiredParameters, parameters, handler)) {
             return;
         }
         String className = parameters.get("class");
         String[] orderSpec = parameters.get("order").split(" ");
 
         Verification.requireClass(className, handler);
         Verification.requireOrder(className, orderSpec, handler);
 
         if (parameters.containsKey("reducer")) {
             String reducerType = parameters.get("reducer");
             Verification.requireClass(reducerType, handler);
         }
     }
 
     public static String getInputClass(TupleFlowParameters parameters) {
         return parameters.getXML().get("class", "");
     }
 
     public static String getOutputClass(TupleFlowParameters parameters) {
         return parameters.getXML().get("class", "");
     }
 
     public static String[] getOutputOrder(TupleFlowParameters parameters) {
         String[] orderSpec = parameters.getXML().get("order", "").split(" ");
         return orderSpec;
     }
 
     @Override
     public String toString() {
         return order.getOrderedClass().getName() + " " + Arrays.asList(order.getOrderSpec());
     }
 
     public boolean needsFlush() {
         if (flushRequested) {
             return true;
         }
         return objects.size() > reduceInterval ||
                 objects.size() + runsCount > limit;
     }
 
     public synchronized void process(T object) throws IOException {
         objects.add(object);
         flushIfNecessary();
     }
 
     public synchronized void flushIfNecessary() throws IOException {
         if (needsFlush()) {
             reduce();
 
             if (needsFlush()) {
                 flush();
             }
         }
     }
 
     @Override
     /***
      * <p>Finishes sorting, then sends sorted output to later stages.</p>
      * 
      * <p>This method is intentionally unsynchronized, as synchronizing it 
      * tends to cause deadlock problems (since this method calls process on
      * later stages).</p>
      */
     public synchronized void close() throws IOException {
         // remove this object as quickly as possible from the alert queue
         // since we don't need to flush anymore
         removeMemoryWarnings();
 
         if (temporaryFiles.size() > 0) {
             combine();
         } else {
             reduce();
             combineRuns(processor);
         }
         processor.close();
     }
 
     /***
      * Reduces the number of buffered objects.  The recently buffered
      * objects are processed by a Reducer, if one was specified in the
      * constructor.  The resulting objects from this reduction are then
      * copied into a "reduced" set of buffered objects.  The process
      * resembles a generational garbage collector.
      * 
      * Even if no reducer exists, this sorts all the current objects and
      * sets them aside.  We perform this initial sort while the objects 
      * are still warm in the cache to get improved throughput overall.
      * 
      * Another benefit to reducing is the speed that we can respond to 
      * low memory events.  If a low memory event happens and the objects 
      * aren't sorted, we have to sort them first before writing them to disk.
      * The sorting process can take extra memory and time, which makes us
      * risk running out of RAM while trying to get data out onto the disk.
      */
     private synchronized void reduce() throws IOException {
         if (size() == 0) {
             return;
         }
         List<T> results = objects;
 
         if (reducer != null) {
             results = reducer.reduce(objects);
         }
         Collections.sort(results, lessThanCompare);
         runs.add(results);
         runsCount += results.size();
 
         objects = new ArrayList<T>();
     }
 
     /***
      * Returns the number of currently buffered objects.
      */
     private long size() {
         return runsCount + objects.size();
     }
 
     public synchronized void flush() throws IOException {
         if (size() == 0) {
             return;
         }
         reduce();
         assert objects.size() == 0;
 
         FileOrderedWriter<T> writer = getTemporaryWriter();
         combineRuns(writer);
         writer.close();
         if (filesWritten != null) filesWritten.increment();
 
         flushRequested = false;
     }
 
     private class RunWrapper<T> implements Comparable<RunWrapper<T>> {
         public Iterator<T> iterator;
         public T top;
         Comparator<T> lessThan;
 
         public RunWrapper(List<T> list, Comparator<T> lessThan) {
             iterator = list.iterator();
             this.lessThan = lessThan;
         }
 
         public int compareTo(RunWrapper<T> other) {
             T one = top;
             T two = other.top;
 
             int result = lessThan.compare(one, two);
             return result;
         }
 
         public boolean next() {
             if (iterator.hasNext()) {
                 top = iterator.next();
                 return true;
             } else {
                 top = null;
                 return false;
             }
         }
     }
 
     /***
      * Takes the sorted runs in the runs array, and combines them into a
      * single sorted list, which is processed by the processor called output.
      */
     private synchronized void combineRuns(Processor<T> output) throws IOException {
         PriorityQueue<RunWrapper<T>> queue = new PriorityQueue<RunWrapper<T>>();
 
         // make a run wrapper for each run we've got buffered, 
         // put it in the priority queue
         for (List<T> run : runs) {
             RunWrapper<T> wrapper = new RunWrapper<T>(run, lessThanCompare);
             if (wrapper.next()) {
                 queue.offer(wrapper);
             }
         }
 
         // we expect that some runs will have lots of contiguous tuples,
         // in the case where the input is already almost sorted.  This loop
         // is optimized for that case.
 
         while (queue.size() > 1) {
             RunWrapper<T> wrapper = queue.poll();
             RunWrapper<T> next = queue.peek();
 
             output.process(wrapper.top);
             wrapper.next();
 
             while (wrapper.top != null &&
                     lessThanCompare.compare(wrapper.top, next.top) <= 0) {
                 output.process(wrapper.top);
                 wrapper.next();
             }
 
             if (wrapper.top != null) {
                 queue.offer(wrapper);
             }
         }
 
         // process all objects from the final run
         if (queue.size() == 1) {
             RunWrapper<T> wrapper = queue.poll();
 
             do {
                 output.process(wrapper.top);
             } while (wrapper.next());
         }
 
         runs.clear();
         runsCount = 0;
     }
 
     private synchronized FileOrderedWriter<T> getTemporaryWriter(long fileSize) throws IOException, FileNotFoundException {
         File temporary = Utility.createTemporary(fileSize * 4);
         FileOrderedWriter<T> writer = new FileOrderedWriter<T>(temporary, order);
         temporaryFiles.add(temporary);
         return writer;
     }
 
     private synchronized FileOrderedWriter<T> getTemporaryWriter() throws IOException, FileNotFoundException {
         File temporary = Utility.createTemporary();
         FileOrderedWriter<T> writer = new FileOrderedWriter<T>(temporary, order);
         temporaryFiles.add(temporary);
         return writer;
     }
 
     private synchronized void combine() throws IOException {
         flush();
 
         if (temporaryFiles.size() == 0) {
             return;
         }
         while (temporaryFiles.size() > fileLimit) {
             // sort all the files so that small ones come first, since those
             // are the ones we want to combine together.
             Collections.sort(temporaryFiles, new Comparator<File>() {
                          public int compare(File one, File two) {
                              long oneLength = one.length();
                              long twoLength = two.length();
 
                              if (oneLength > twoLength) {
                                  return 1;
                              } else if (oneLength < twoLength) {
                                  return -1;
                              }
                              return 0;
                          }
                      });
 
             // pick a set of files to merge and remove them from the regular file set
             ArrayList<File> temporaryFileSet = new ArrayList<File>(temporaryFiles.subList(0,
                                                                                           fileLimit));
             temporaryFiles.subList(0, fileLimit).clear();
 
             // calculate the total amount of space we'll need for this
             long totalFileSize = 0;
 
             for (File f : temporaryFiles) {
                 totalFileSize += f.length();
             }
 
             // get a temporary writer that's big enough to handle all this data.
             // this adds the File to the end of the list of temporary files
             long oneGigabyte = 1024 * 1024 * 1024;
             FileOrderedWriter<T> writer = getTemporaryWriter(totalFileSize * 2 + oneGigabyte);
 
             // do the actual combination work
             combineStep(temporaryFileSet, writer);
 
             writer.close();
             temporaryFileSet.clear();
         }
 
         combineStep(temporaryFiles, processor);
         temporaryFiles.clear();
     }
 
     private synchronized void combineStep(List<File> files, Processor<T> output) throws FileNotFoundException, IOException {
         if (sorterCombineSteps != null) sorterCombineSteps.increment();
 
         ArrayList<String> filenames = new ArrayList<String>();
 
         for (File f : files) {
             filenames.add(f.getPath());
         }
         OrderedCombiner combiner = OrderedCombiner.combineFromFiles(filenames, order, output, false,
                                                                     combineBufferSize);
         combiner.run();
 
         for (File file : files) {
             file.delete();
         }
     }
 }