/*
    * 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.giraph.block_app.reducers.array;
  
  import java.io.DataInput;
  import java.io.DataOutput;
  import java.io.IOException;
  
  import org.apache.commons.lang3.tuple.MutablePair;
  import org.apache.commons.lang3.tuple.Pair;
  import org.apache.giraph.block_app.framework.api.BlockMasterApi;
  import org.apache.giraph.block_app.framework.api.CreateReducersApi;
  import org.apache.giraph.block_app.framework.api.CreateReducersApi.CreateReducerFunctionApi;
  import org.apache.giraph.block_app.framework.piece.global_comm.BroadcastHandle;
  import org.apache.giraph.block_app.framework.piece.global_comm.ReducerHandle;
  import org.apache.giraph.block_app.framework.piece.global_comm.array.BroadcastArrayHandle;
  import org.apache.giraph.block_app.framework.piece.global_comm.array.ReducerArrayHandle;
  import org.apache.giraph.function.primitive.PrimitiveRefs.IntRef;
  import org.apache.giraph.master.MasterGlobalCommUsage;
  import org.apache.giraph.reducers.ReduceOperation;
  import org.apache.giraph.types.ops.PrimitiveTypeOps;
  import org.apache.giraph.types.ops.TypeOpsUtils;
  import org.apache.giraph.types.ops.collections.array.WArrayList;
  import org.apache.giraph.utils.WritableUtils;
  import org.apache.giraph.worker.WorkerBroadcastUsage;
  import org.apache.hadoop.io.Writable;
  
  /**
   * Efficient generic primitive array reduce operation.
   *
   * Allows two modes - fixed size, and infinite size
   * (with keeping only actually used elements and resizing)
   *
   * @param <S> Single value type
   * @param <R> Reduced value type
   */
  public class BasicArrayReduce<S, R extends Writable>
      implements ReduceOperation<Pair<IntRef, S>, WArrayList<R>> {
    private int fixedSize;
    private PrimitiveTypeOps<R> typeOps;
    private ReduceOperation<S, R> elementReduceOp;
    private R initialElement;
    private R reusable;
    private R reusable2;
  
    public BasicArrayReduce() {
    }
  
  
    /**
     * Create ReduceOperation that reduces BasicArrays by reducing individual
     * elements, with predefined size.
     *
     * @param fixedSize Number of elements
     * @param typeOps TypeOps of individual elements
     * @param elementReduceOp ReduceOperation for individual elements
     */
    public BasicArrayReduce(
        int fixedSize,
        PrimitiveTypeOps<R> typeOps,
        ReduceOperation<S, R> elementReduceOp) {
      this.fixedSize = fixedSize;
      this.typeOps = typeOps;
      this.elementReduceOp = elementReduceOp;
      init();
    }
  
  
    /**
     * Create ReduceOperation that reduces BasicArrays by reducing individual
     * elements, with unbounded size.
     *
     * @param typeOps TypeOps of individual elements
     * @param elementReduceOp ReduceOperation for individual elements
     */
    public BasicArrayReduce(
        PrimitiveTypeOps<R> typeOps, ReduceOperation<S, R> elementReduceOp) {
      this(-1, typeOps, elementReduceOp);
    }
  
  
    /**
     * Registers one new local reducer, that will reduce BasicArray,
     * by reducing individual elements using {@code elementReduceOp},
    * with unbounded size.
    *
    * This function will return ReducerArrayHandle, by which
    * individual elements can be manipulated separately.
    *
    * @param typeOps TypeOps of individual elements
    * @param elementReduceOp ReduceOperation for individual elements
    * @param reduceApi API for creating reducers
    * @return Created ReducerArrayHandle
    */
   public static <S, R extends Writable>
   ReducerArrayHandle<S, R> createLocalArrayHandles(
       PrimitiveTypeOps<R> typeOps, ReduceOperation<S, R> elementReduceOp,
       CreateReducersApi reduceApi) {
     return createLocalArrayHandles(-1, typeOps, elementReduceOp, reduceApi);
   }
 
   /**
    * Registers one new local reducer, that will reduce BasicArray,
    * by reducing individual elements using {@code elementReduceOp},
    * with predefined size.
    *
    * This function will return ReducerArrayHandle, by which
    * individual elements can be manipulated separately.
    *
    * @param fixedSize Number of elements
    * @param typeOps TypeOps of individual elements
    * @param elementReduceOp ReduceOperation for individual elements
    * @param reduceApi API for creating reducers
    * @return Created ReducerArrayHandle
    */
   public static <S, R extends Writable>
   ReducerArrayHandle<S, R> createLocalArrayHandles(
       int fixedSize, PrimitiveTypeOps<R> typeOps,
       ReduceOperation<S, R> elementReduceOp,
       final CreateReducersApi reduceApi) {
     return createArrayHandles(fixedSize, typeOps, elementReduceOp,
         new CreateReducerFunctionApi() {
           @Override
           public <S, R extends Writable> ReducerHandle<S, R> createReducer(
               ReduceOperation<S, R> reduceOp) {
             return reduceApi.createLocalReducer(reduceOp);
           }
         });
   }
 
   /**
    * Registers one new reducer, that will reduce BasicArray,
    * by reducing individual elements using {@code elementReduceOp},
    * with unbounded size.
    *
    * This function will return ReducerArrayHandle, by which
    * individual elements can be manipulated separately.
    *
    * @param typeOps TypeOps of individual elements
    * @param elementReduceOp ReduceOperation for individual elements
    * @param createFunction Function for creating a reducer
    * @return Created ReducerArrayHandle
    */
   public static <S, R extends Writable>
   ReducerArrayHandle<S, R> createArrayHandles(
       PrimitiveTypeOps<R> typeOps, ReduceOperation<S, R> elementReduceOp,
       CreateReducerFunctionApi createFunction) {
     return createArrayHandles(-1, typeOps, elementReduceOp, createFunction);
   }
 
   /**
    * Registers one new reducer, that will reduce BasicArray,
    * by reducing individual elements using {@code elementReduceOp},
    * with predefined size.
    *
    * This function will return ReducerArrayHandle, by which
    * individual elements can be manipulated separately.
    *
    * @param fixedSize Number of elements
    * @param typeOps TypeOps of individual elements
    * @param elementReduceOp ReduceOperation for individual elements
    * @param createFunction Function for creating a reducer
    * @return Created ReducerArrayHandle
    */
   public static <S, R extends Writable>
   ReducerArrayHandle<S, R> createArrayHandles(
       final int fixedSize, final PrimitiveTypeOps<R> typeOps,
       ReduceOperation<S, R> elementReduceOp,
       CreateReducerFunctionApi createFunction) {
     final ReducerHandle<Pair<IntRef, S>, WArrayList<R>> reduceHandle =
         createFunction.createReducer(
             new BasicArrayReduce<>(fixedSize, typeOps, elementReduceOp));
     final IntRef curIndex = new IntRef(0);
     final R reusableValue = typeOps.create();
     final R initialValue = elementReduceOp.createInitialValue();
     final MutablePair<IntRef, S> reusablePair =
         MutablePair.of(new IntRef(0), null);
     final ReducerHandle<S, R> elementReduceHandle = new ReducerHandle<S, R>() {
       @Override
       public R getReducedValue(MasterGlobalCommUsage master) {
         WArrayList<R> result = reduceHandle.getReducedValue(master);
         if (fixedSize == -1 && curIndex.value >= result.size()) {
           typeOps.set(reusableValue, initialValue);
         } else {
           result.getIntoW(curIndex.value, reusableValue);
         }
         return reusableValue;
       }
 
       @Override
       public void reduce(S valueToReduce) {
         reusablePair.getLeft().value = curIndex.value;
         reusablePair.setRight(valueToReduce);
         reduceHandle.reduce(reusablePair);
       }
 
       @Override
       public BroadcastHandle<R> broadcastValue(BlockMasterApi master) {
         throw new UnsupportedOperationException();
       }
     };
 
     return new ReducerArrayHandle<S, R>() {
       @Override
       public ReducerHandle<S, R> get(int index) {
         curIndex.value = index;
         return elementReduceHandle;
       }
 
       @Override
       public int getStaticSize() {
         if (fixedSize == -1) {
           throw new UnsupportedOperationException(
               "Cannot call size, when one is not specified upfront");
         }
         return fixedSize;
       }
 
       @Override
       public int getReducedSize(BlockMasterApi master) {
         return reduceHandle.getReducedValue(master).size();
       }
 
       @Override
       public BroadcastArrayHandle<R> broadcastValue(BlockMasterApi master) {
         final BroadcastHandle<WArrayList<R>> broadcastHandle =
             reduceHandle.broadcastValue(master);
         final IntRef curIndex = new IntRef(0);
         final R reusableValue = typeOps.create();
         final BroadcastHandle<R>
         elementBroadcastHandle = new BroadcastHandle<R>() {
           @Override
           public R getBroadcast(WorkerBroadcastUsage worker) {
             WArrayList<R> result = broadcastHandle.getBroadcast(worker);
             if (fixedSize == -1 && curIndex.value >= result.size()) {
               typeOps.set(reusableValue, initialValue);
             } else {
               result.getIntoW(curIndex.value, reusableValue);
             }
             return reusableValue;
           }
         };
         return new BroadcastArrayHandle<R>() {
           @Override
           public BroadcastHandle<R> get(int index) {
             curIndex.value = index;
             return elementBroadcastHandle;
           }
 
           @Override
           public int getStaticSize() {
             if (fixedSize == -1) {
               throw new UnsupportedOperationException(
                   "Cannot call size, when one is not specified upfront");
             }
             return fixedSize;
           }
 
           @Override
           public int getBroadcastedSize(WorkerBroadcastUsage worker) {
             return broadcastHandle.getBroadcast(worker).size();
           }
         };
       }
     };
   }
 
 
   private void init() {
     initialElement = elementReduceOp.createInitialValue();
     reusable = typeOps.create();
     reusable2 = typeOps.create();
   }
 
   @Override
   public WArrayList<R> createInitialValue() {
     if (fixedSize != -1) {
       WArrayList<R> list = typeOps.createArrayList(fixedSize);
       fill(list, fixedSize);
       return list;
     } else {
       return typeOps.createArrayList(1);
     }
   }
 
   private void fill(WArrayList<R> list, int newSize) {
     if (fixedSize != -1 && newSize > fixedSize) {
       throw new IllegalArgumentException(newSize + " larger then " + fixedSize);
     }
 
     if (list.capacity() < newSize) {
       list.setCapacity(newSize);
     }
     while (list.size() < newSize) {
       list.addW(initialElement);
     }
   }
 
   @Override
   public WArrayList<R> reduce(
       WArrayList<R> curValue, Pair<IntRef, S> valueToReduce) {
     int index = valueToReduce.getLeft().value;
     fill(curValue, index + 1);
     curValue.getIntoW(index, reusable);
     R result = elementReduceOp.reduce(reusable, valueToReduce.getRight());
     curValue.setW(index, result);
     return curValue;
   }
 
   @Override
   public WArrayList<R> reduceMerge(
       WArrayList<R> curValue, WArrayList<R> valueToReduce) {
     fill(curValue, valueToReduce.size());
     for (int i = 0; i < valueToReduce.size(); i++) {
       valueToReduce.getIntoW(i, reusable2);
       curValue.getIntoW(i, reusable);
       R result = elementReduceOp.reduceMerge(reusable, reusable2);
       curValue.setW(i, result);
     }
 
     return curValue;
   }
 
   @Override
   public void write(DataOutput out) throws IOException {
     out.writeInt(fixedSize);
     TypeOpsUtils.writeTypeOps(typeOps, out);
     WritableUtils.writeWritableObject(elementReduceOp, out);
   }
 
   @Override
   public void readFields(DataInput in) throws IOException {
     fixedSize = in.readInt();
     typeOps = TypeOpsUtils.readTypeOps(in);
     elementReduceOp = WritableUtils.readWritableObject(in, null);
     init();
   }
 }