Лабораторная работа 4

Задание

Разработка распределенного приложения с использованием платформы Apache Ignite. Необходимо разработать параллельный вариант алгоритма с применением подхода Grid Gain и платформа Apache Ignite, замерить время его работы.

Вариант: Определить минимальный элемент матрицы ниже главной диагонали.

Описание работы приложения

В решении используется Apache Ignite для распределения вычислений. Программа делает следующее:

Настраивает Ignite с нужной конфигурацией.
Генерирует квадратную матрицу 1000×1000, заполненную случайными числами от 0 до 9999.
Делит матрицу на 4 части по строкам. Для каждой части запускается ComputeTask, которая распределяет работу ComputeJob, ищущих локальный минимум среди элементов ниже главной диагонали.
Полученные локальные минимумы объединяются для нахождения глобального минимума, и замеряется общее время выполнения.

Объяснение

MatrixMinAlgorithm

package ru.kuznec;

import org.apache.ignite.Ignite;
import org.apache.ignite.Ignition;
import org.apache.ignite.configuration.IgniteConfiguration;
import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;

import java.util.Collections;
import java.util.Random;

public class MatrixMinAlgorithm {
    public static void main(String[] args) {
        IgniteConfiguration cfg = new IgniteConfiguration();
        cfg.setPeerClassLoadingEnabled(false);

        TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
        TcpDiscoveryVmIpFinder ipFinder = new TcpDiscoveryVmIpFinder();
        
        ipFinder.setAddresses(Collections.singletonList("127.0.0.1:47500..47509"));
        discoSpi.setIpFinder(ipFinder);
        cfg.setDiscoverySpi(discoSpi);

        try (Ignite ignite = Ignition.start(cfg)) {
            final int n = 1000;
            final int[][] matrix = new int[n][n];
            Random rnd = new Random();
            
            for (int i = 0; i < n; i++) {
                for (int j = 0; j < n; j++) {
                    matrix[i][j] = rnd.nextInt(10000);
                }
            }

            final int numTasks = 4;
            final int rowsPerTask = n / numTasks;
            int globalMin = Integer.MAX_VALUE;

            long startTime = System.currentTimeMillis();

            for (int task = 0; task < numTasks; task++) {
                int startRow = task * rowsPerTask;
                int endRow = (task == numTasks - 1) ? n : startRow + rowsPerTask;
                
                int localMin = ignite.compute().execute(new MatrixMinTask(matrix, startRow, endRow), null);
                globalMin = Math.min(globalMin, localMin);
            }

            long endTime = System.currentTimeMillis();

            System.out.println("Минимальный элемент ниже главной диагонали: " + globalMin);
            System.out.println("Общее время выполнения: " + (endTime - startTime) + " мс");
        }
    }
}

Объяснение: В этом файле происходит настройка Ignite и запуск узла. Затем генерируется матрица случайных чисел. Матрица делится на 4 части, и для каждой части запускается задача (MatrixMinTask), которая возвращает локальный минимум. В конце все локальные минимумы сравниваются, и выводится глобальный минимум, а также время работы.

MatrixMinTask.java

package ru.kuznec;

import org.apache.ignite.IgniteException;
import org.apache.ignite.cluster.ClusterNode;
import org.apache.ignite.compute.ComputeJob;
import org.apache.ignite.compute.ComputeJobResult;
import org.apache.ignite.compute.ComputeJobResultPolicy;
import org.apache.ignite.compute.ComputeTask;
import org.apache.ignite.compute.ComputeTaskName;

import java.util.Collections;
import java.util.List;
import java.util.Map;

@ComputeTaskName("MatrixMinTaskUser")
public class MatrixMinTask implements ComputeTask<Void, Integer> {
    private final int[][] matrix;
    private final int startRow;
    private final int endRow;

    public MatrixMinTask(int[][] matrix, int startRow, int endRow) {
        this.matrix = matrix;
        this.startRow = startRow;
        this.endRow = endRow;
    }

    @Override
    public Map<? extends ComputeJob, ClusterNode> map(List<ClusterNode> subgrid, Void arg) throws IgniteException {
        ComputeJob job = new MatrixMinJob(matrix, startRow, endRow);
        return Collections.singletonMap(job, subgrid.get(0));
    }

    @Override
    public ComputeJobResultPolicy result(ComputeJobResult res, List<ComputeJobResult> received) throws IgniteException {
        return ComputeJobResultPolicy.WAIT;
    }

    @Override
    public Integer reduce(List<ComputeJobResult> results) throws IgniteException {
        int globalMin = Integer.MAX_VALUE;
        
        for (ComputeJobResult res : results) {
            globalMin = Math.min(globalMin, res.getData());
        }

        return globalMin;
    }
}

Объяснение: Этот класс реализует интерфейс ComputeTask, который отвечает за распределение работы по узлам и объединение результатов.

В методе map() создается задание (ComputeJob) для заданного диапазона строк и привязывается к первому узлу в списке.
Метод result() всегда возвращает политику WAIT, чтобы дождаться выполнения всех заданий.
Метод reduce() проходит по всем полученным результатам и выбирает наименьшее значение.
Аннотация @ComputeTaskName("MatrixMinTaskUser") задаёт имя задачи, чтобы Ignite не считал её системной.

MatrixMinJob.java

package ru.kuznec;

import org.apache.ignite.IgniteException;
import org.apache.ignite.compute.ComputeJob;

public class MatrixMinJob implements ComputeJob {
    private final int[][] matrix;
    private final int startRow;
    private final int endRow;

    public MatrixMinJob(int[][] matrix, int startRow, int endRow) {
        this.matrix = matrix;
        this.startRow = startRow;
        this.endRow = endRow;
    }

    @Override
    public Object execute() throws IgniteException {
        int localMin = Integer.MAX_VALUE;

        for (int i = startRow; i < endRow; i++) {
            for (int j = 0; j < i && j < matrix[i].length; j++) {
                localMin = Math.min(localMin, matrix[i][j]);
            }
        }

        return localMin;
    }

    @Override
    public void cancel() {}
}

Объяснение:
Класс MatrixMinJob реализует ComputeJob и отвечает за выполнение вычислительной части задачи.
В методе execute() происходит проход по заданному диапазону строк матрицы. Для каждой строки ищутся элементы с индексами от 0 до i‑1 (то есть элементы, находящиеся ниже главной диагонали). Вычисляется локальный минимум, который затем возвращается. Метод cancel() оставлен пустым, так как отмена не требуется по заданию.

Запуск

Запустить jar можно командой (обязательно Java 8):

java -jar Lab-4-jar-with-dependencies.jar

При запуске Ignite запускает узел, распределяет задачу по 4 частям матрицы, и в конце выводится минимальный элемент и общее время выполнения.

README.md Unescape Escape