package edu.washington.escience.myria.parallel;

import edu.washington.escience.myria.DbException;
import edu.washington.escience.myria.TupleBatch;
import edu.washington.escience.myria.operator.Operator;

/**
 * The producer part of the Shuffle Exchange operator.
 * 
 * ShuffleProducer distributes tuples to the workers according to some partition function (provided as a
 * PartitionFunction object during the ShuffleProducer's instantiation).
 * 
 */
public class LocalShuffleProducer extends Producer {

  /** Required for Java serialization. */
  private static final long serialVersionUID = 1L;

  /**
   * the partition function.
   * */
  private final PartitionFunction<?, ?> partitionFunction;

  /**
   * @param child the child who provides data for this producer to distribute.
   * @param operatorIDs destination operators the data goes
   * @param pf the partition function
   * */
  public LocalShuffleProducer(final Operator child, final ExchangePairID[] operatorIDs, final PartitionFunction<?, ?> pf) {
    super(child, operatorIDs);
    partitionFunction = pf;
  }

  /**
   * @return the partition function I'm using.
   * */
  public final PartitionFunction<?, ?> getPartitionFunction() {
    return partitionFunction;
  }

  @Override
  protected final void consumeTuples(final TupleBatch tup) throws DbException {
    TupleBatch[] tbs = tup.partition(partitionFunction);
    for (int p = 0; p < tbs.length; p++) {
      if (tbs[p] != null) {
        writeMessage(p, tbs[p]);
      }
    }
  }

  @Override
  protected final void childEOS() throws DbException {
    for (int i = 0; i < numChannels(); i++) {
      super.channelEnds(i);
    }
  }

  @Override
  protected final void childEOI() throws DbException {
    TupleBatch eoiTB = TupleBatch.eoiTupleBatch(getSchema());
    for (int i = 0; i < numChannels(); i++) {
      writeMessage(i, eoiTB);
    }
  }
}

@@ -42,16 +42,62 @@
 allel;%0A%0A
+import com.google.common.base.Preconditions;%0A%0A
 import e
@@ -125,27 +125,33 @@
 e.myria.
-DbException
+operator.Operator
 ;%0Aimport
@@ -181,26 +181,46 @@
 e.myria.
-TupleBatch
+parallel.ipc.IPCConnectionPool
 ;%0Aimport
@@ -250,33 +250,36 @@
 e.myria.
-operator.Operator
+util.MyriaArrayUtils
 ;%0A%0A/**%0A 
@@ -564,16 +564,30 @@
 extends 
+GenericShuffle
 Producer
@@ -688,111 +688,8 @@
 L;%0A%0A
-  /**%0A   * the partition function.%0A   * */%0A  private final PartitionFunction%3C?, ?%3E partitionFunction;%0A%0A
   /*
@@ -1031,838 +1031,206 @@
 rIDs
-);%0A    partitionFunction = pf;%0A  %7D%0A%0A  /**%0A   * @return the partition function I'm using.%0A   * */%0A  public final PartitionFunction%3C?, ?%3E getPartitionFunction() %7B%0A    return partitionFunction;%0A  %7D%0A%0A  @Override%0A  protected final void consumeTuples(final TupleBatch tup) throws DbException %7B%0A    TupleBatch%5B%5D tbs = tup.partition(partitionFunction);%0A    for (int p = 0; p %3C tbs.length; p++) %7B%0A      if (tbs%5Bp%5D != null) %7B%0A        writeMessage(p, tbs%5Bp%5D);%0A      %7D%0A    %7D%0A  %7D%0A%0A  @Override%0A  protected final void childEOS() throws DbException %7B%0A    for (int i = 0; i %3C numChannels(); i++) %7B%0A      super.channelEnds(i);%0A    %7D%0A  %7D%0A%0A  @Override%0A  protected final void childEOI() throws DbException %7B%0A    TupleBatch eoiTB = TupleBatch.eoiTupleBatch(getSchema());%0A    for (int i = 0; i %3C numChannels(); i++) %7B%0A      writeMessage(i, eoiTB);%0A    %7D
+, MyriaArrayUtils.create2DVerticalIndex(pf.numPartition()),%0A        new int%5B%5D %7B IPCConnectionPool.SELF_IPC_ID %7D, pf, false);%0A    Preconditions.checkArgument(operatorIDs.length == pf.numPartition());
 %0A  %7D

make LocalShuffleProducer as a subclass of GenericShuffleProducer.