/*
* 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.
*/

import edu.purdue.cs.aggr.NOAHReduceFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;

/**
 * Estimates the value of Pi using the Monte Carlo method.
 * The area of a circle is Pi * R^2, R being the radius of the circle
 * The area of a square is 4 * R^2, where the length of the square's edge is 2*R.
 *
 * Thus Pi = 4 * (area of circle / area of square).
 *
 * The idea is to find a way to estimate the circle to square area ratio.
 * The Monte Carlo method suggests collecting random points (within the square)
 * and then counting the number of points that fall within the circle
 *
 * <pre>
 * {@code
 * x = Math.random()
 * y = Math.random()
 *
 * x * x + y * y < 1
 * }
 * </pre>
 */
@SuppressWarnings("serial")
public class RandomSummer implements java.io.Serializable {

  public static void main(String[] args) throws Exception {

    final long numSamples = args.length > 1 ? Long.parseLong(args[1]) : 1000000;

    final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

    // count how many of the samples would randomly fall into
    // the unit circle
    DataSet<CustomLong> count =
        env.generateSequence(1, numSamples)
            .map(new Sampler())
            .reduce(new SumReducer());

    count.print();

    env.execute("RandomSummer");
  }

  //*************************************************************************
  //     USER FUNCTIONS
  //*************************************************************************

	public static class CustomLong {

		// fields
		private Long number;

		// constructors
		public CustomLong() {}

		public CustomLong(Long n) {
			this.number = n;
		}

		// getters setters
		public Long getNumber() {
			return number;
		}
		public void setNumber(Long n) {
			this.number = n;
		}

		@Override
		public String toString() {
			return number.toString();
		}
	}

  /**
   * Sampler randomly emits points that fall within a square of edge x * y.
   * It calculates the distance to the center of a virtually centered circle of radius x = y = 1
   * If the distance is less than 1, then and only then does it returns a 1.
   */
  public static class Sampler implements MapFunction<Long, CustomLong> {

    @Override
    public CustomLong map(Long value) throws Exception{
      double x = Math.random();
      double y = Math.random();
      return (x * x + y * y) < 1 ? new CustomLong(1L) : new CustomLong(0L);
    }
  }


  /**
   * Simply sums up all long values.
   */
  public static final class SumReducer implements
			NOAHReduceFunction<CustomLong> {

    @Override
    public CustomLong reduce(CustomLong value1, CustomLong value2) throws Exception {
			value1.setNumber(value1.getNumber() + value2.getNumber());
      return value1;
    }

		@Override
		public float getInputOutputRatio(int numParitions) {
			return 1.0f;
		}
	}

}