Source code

001/*
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *   http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing,
013 * software distributed under the License is distributed on an
014 * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
015 * KIND, either express or implied.  See the License for the
016 * specific language governing permissions and limitations
017 * under the License.
018 */
019package org.apache.reef.io.network.group.api.operators;
020
021import org.apache.reef.exception.evaluator.NetworkException;
022import org.apache.reef.io.network.group.impl.operators.ReduceReceiver;
023import org.apache.reef.io.network.group.impl.operators.ReduceSender;
024import org.apache.reef.tang.annotations.DefaultImplementation;
025import org.apache.reef.wake.Identifier;
026
027import java.util.List;
028
029/**
030 * MPI Reduce operator.
031 * <p>
032 * This is another operator with root being receiver All senders send an element
033 * to the receiver. These elements are passed through a reduce function and its
034 * result is made available at the root
035 */
036public interface Reduce {
037
038  /**
039   * Receiver or Root.
040   */
041  @DefaultImplementation(ReduceReceiver.class)
042  interface Receiver<T> extends GroupCommOperator {
043
044    /**
045     * Receive values sent by senders and pass them through the reduce
046     * function in default order.
047     *
048     * @return Result of applying reduce function on the elements gathered in default order.
049     */
050    T reduce() throws InterruptedException, NetworkException;
051
052    /**
053     * Receive values sent by senders and pass them through the reduce
054     * function in specified order.
055     *
056     * @return Result of applying reduce function on the elements gathered in specified order.
057     */
058    T reduce(List<? extends Identifier> order) throws InterruptedException, NetworkException;
059
060    /**
061     * The reduce function to be applied on the set of received values.
062     *
063     * @return {@link ReduceFunction}
064     */
065    Reduce.ReduceFunction<T> getReduceFunction();
066  }
067
068  /**
069   * Senders or non roots.
070   */
071  @DefaultImplementation(ReduceSender.class)
072  interface Sender<T> extends GroupCommOperator {
073
074    /**
075     * Send the element to the root.
076     */
077    void send(T element) throws NetworkException, InterruptedException;
078
079    /**
080     * The {@link ReduceFunction} to be applied on the set of received values.
081     *
082     * @return {@link ReduceFunction}
083     */
084    Reduce.ReduceFunction<T> getReduceFunction();
085  }
086
087  /**
088   * Interface for a Reduce Function takes in an {@link Iterable} returns an.
089   * aggregate value computed from the {@link Iterable}
090   */
091  interface ReduceFunction<T> {
092    /**
093     * Apply the function on elements.
094     *
095     * @return aggregate value computed from elements.
096     */
097    T apply(Iterable<T> elements);
098  }
099}