package org.gridgain.examples.montecarlo.basic;

import java.io.*;
import java.util.*;
import org.gridgain.grid.*;
import org.gridgain.grid.gridify.*;

public class CreditRiskGridTask extends GridTaskSplitAdapter<GridifyArgument> {
    @Override
    protected Collection<? extends GridJob> split(int gridSize, final GridifyArgument arg) {
        // Split number of iterations.
        Integer iterations = ((Integer)arg.getMethodParameters()[2]);

        // Number of iterations should be done by each node.
        int iterPerNode = Math.round(iterations / (float)gridSize);

        // Number of iterations for the last/the only node.
        int lastNodeIter = iterations - (gridSize - 1) * iterPerNode;

        List<GridJobAdapter<Integer>> jobs = new ArrayList<GridJobAdapter<Integer>>(gridSize);

        // Note that for the purpose of this example we perform a simple homogeneous
        // (non weighted) split assuming that all computing resources in this split
        // will be identical. In real life scenarios when heterogeneous environment
        // is used a split that is weighted by, for example, CPU benchmarks of each
        // node in the split will be more efficient. It is fairly easy addition and
        // GridGain comes with convenient Spring-compatible benchmark that can be
        // used for weighted splits.
        for (int i = 0; i < gridSize; i++) {
            // Add new job reference to the split.
            // Add new job reference to the split.
            jobs.add(new GridJobAdapter<Integer>(i == gridSize - 1 ? lastNodeIter : iterPerNode) {
                public Serializable execute() throws GridException {
                    Object[] params = arg.getMethodParameters();

                    return new CreditRiskManager().calculateCreditRiskMonteCarlo((Credit[])params[0],
                        (Integer)params[1], getArgument(), (Double)params[3]);
                }
            });
        }

        // Trigger remote execution with given job references.
        return jobs;
    }

    public Serializable reduce(List<GridJobResult> results) throws GridException {
        double taskRes = 0;

        // Reduce results by summing them.
        for (GridJobResult res : results) {
            if (res.getException() != null) {
                throw res.getException();
            }

            // Note that since we use Gridify annotation for grid enabling
            // we know what type of result data to expect.
            taskRes += (Double)res.getData();
        }

        return taskRes / results.size();
    }
}