/*

  * 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 java.lang.reflect.Array;

 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.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.utils.ArrayWritable;

 import org.apache.giraph.utils.WritableUtils;

 import org.apache.giraph.worker.WorkerBroadcastUsage;

 import org.apache.hadoop.io.Writable;

 /**

  * One reducer representing reduction of array of individual values.

  * Elements are represented as object, and so BasicArrayReduce should be

  * used instead when elements are primitive types.

  *

  * @param <S> Single value type, objects passed on workers

  * @param <R> Reduced value type

  */

 public class ArrayReduce<S, R extends Writable>

     implements ReduceOperation<Pair<IntRef, S>, ArrayWritable<R>> {

   private int fixedSize;

   private ReduceOperation<S, R> elementReduceOp;

   private Class<R> elementClass;

   public ArrayReduce() {

   }

   /**

    * Create ReduceOperation that reduces arrays by reducing individual

    * elements.

    *

    * @param fixedSize Number of elements

    * @param elementReduceOp ReduceOperation for individual elements

    */

   public ArrayReduce(int fixedSize, ReduceOperation<S, R> elementReduceOp) {

     this.fixedSize = fixedSize;

     this.elementReduceOp = elementReduceOp;

     init();

   }

   /**

    * Registers one new reducer, that will reduce array of objects,

    * by reducing individual elements using {@code elementReduceOp}.

    *

    * This function will return ReducerArrayHandle to it, by which

    * individual elements can be manipulated separately.

    *

    * @param fixedSize Number of elements

    * @param elementReduceOp ReduceOperation for individual elements

    * @param createFunction Function for creating a reducer

    * @return Created ReducerArrayHandle

    */

   public static <S, T extends Writable>

   ReducerArrayHandle<S, T> createArrayHandles(

       final int fixedSize, ReduceOperation<S, T> elementReduceOp,

       CreateReducerFunctionApi createFunction) {

     final ReducerHandle<Pair<IntRef, S>, ArrayWritable<T>> reduceHandle =

         createFunction.createReducer(

             new ArrayReduce<>(fixedSize, elementReduceOp));

     final IntRef curIndex = new IntRef(0);

     final MutablePair<IntRef, S> reusablePair =

         MutablePair.of(new IntRef(0), null);

     final ReducerHandle<S, T> elementReduceHandle = new ReducerHandle<S, T>() {

       @Override

       public T getReducedValue(MasterGlobalCommUsage master) {

         ArrayWritable<T> result = reduceHandle.getReducedValue(master);

         return result.get()[curIndex.value];

       }

       @Override

       public void reduce(S valueToReduce) {

         reusablePair.getLeft().value = curIndex.value;

         reusablePair.setRight(valueToReduce);

         reduceHandle.reduce(reusablePair);

       }

       @Override

       public BroadcastHandle<T> broadcastValue(BlockMasterApi master) {

         throw new UnsupportedOperationException();

       }

     };

     return new ReducerArrayHandle<S, T>() {

       @Override

       public ReducerHandle<S, T> get(int index) {

         curIndex.value = index;

         return elementReduceHandle;

       }

       @Override

       public int getStaticSize() {

         return fixedSize;

       }

       @Override

       public int getReducedSize(BlockMasterApi master) {

         return getStaticSize();

       }

       @Override

       public BroadcastArrayHandle<T> broadcastValue(BlockMasterApi master) {

         final BroadcastHandle<ArrayWritable<T>> broadcastHandle =

             reduceHandle.broadcastValue(master);

         final IntRef curIndex = new IntRef(0);

         final BroadcastHandle<T>

         elementBroadcastHandle = new BroadcastHandle<T>() {

           @Override

           public T getBroadcast(WorkerBroadcastUsage worker) {

             ArrayWritable<T> result = broadcastHandle.getBroadcast(worker);

             return result.get()[curIndex.value];

           }

         };

         return new BroadcastArrayHandle<T>() {

           @Override

           public BroadcastHandle<T> get(int index) {

             curIndex.value = index;

             return elementBroadcastHandle;

           }

           @Override

           public int getStaticSize() {

             return fixedSize;

           }

           @Override

           public int getBroadcastedSize(WorkerBroadcastUsage worker) {

             return getStaticSize();

           }

         };

       }

     };

   }

   private void init() {

     elementClass = (Class<R>) elementReduceOp.createInitialValue().getClass();

   }

   @Override

   public ArrayWritable<R> createInitialValue() {

     R[] values = (R[]) Array.newInstance(elementClass, fixedSize);

     for (int i = 0; i < fixedSize; i++) {

       values[i] = elementReduceOp.createInitialValue();

     }

     return new ArrayWritable<>(elementClass, values);

   }

   @Override

   public ArrayWritable<R> reduce(

       ArrayWritable<R> curValue, Pair<IntRef, S> valueToReduce) {

     int index = valueToReduce.getLeft().value;

     curValue.get()[index] =

         elementReduceOp.reduce(curValue.get()[index], valueToReduce.getRight());

     return curValue;

   }

   @Override

   public ArrayWritable<R> reduceMerge(

       ArrayWritable<R> curValue, ArrayWritable<R> valueToReduce) {

     for (int i = 0; i < fixedSize; i++) {

       curValue.get()[i] =

           elementReduceOp.reduceMerge(

               curValue.get()[i], valueToReduce.get()[i]);

     }

     return curValue;

   }

   @Override

   public void write(DataOutput out) throws IOException {

     out.writeInt(fixedSize);

     WritableUtils.writeWritableObject(elementReduceOp, out);

   }

   @Override

   public void readFields(DataInput in) throws IOException {

     fixedSize = in.readInt();

     elementReduceOp = WritableUtils.readWritableObject(in, null);

     init();

   }

 }