OpenMPI Java Bindings Behavior Разница для блокировки, накопления, получения - PullRequest
0 голосов
/ 30 мая 2018

У меня есть случай, когда мне нужно использовать Java в нашем исследовательском кластере с MPI.Одна конкретная функция, которая мне нужна, хорошо описана в этом вопросе (код C ++ включен в связанный ответ).Я создал код C ++, и он работает точно так, как задумано.

Я пытался создать Java-эквивалент этого кода, но с треском провалился.Несмотря на то, что функционально я повторил то, что делает код C ++, версия Java не всегда возвращает желаемые результаты.

mpiexec --oversubscribe -n 4 ./test

0 got counter 1
2 got counter 2
1 got counter 3
3 got counter 4
 1  1  1  1 

(Запуск с --oversubscribe на моем локальном ноутбуке.)

Когда я запускаю свой эквивалент Java, я не получаю ничего похожего:

mpirun --oversubscribe -n 4 java -cp .:/usr/local/lib/openmpi/mpi.jar CounterTest

0 got counter 1
3 got counter 1
1 got counter 3
2 got counter 2
 1  1  1  1 

Я ожидаю, что каждый ранг получает один и только один счетчик.В этом прогоне счетчик 1 использовался дважды.Однажды в голубой луне я могу заставить его доставить мне 1 - 4 (порядок не важен; уникальный счетчик).

Мы запускаем версию 2.1.0 на нашем кластере.На моем локальном ноутбуке у меня установлены OpenMPI 2.1.0 и 3.1.0 (текущий), и я могу воспроизвести правильное поведение программы на C ++ и неправильное поведение программы на Java в любой версии.

Вот счетчиккласс, который я создал:

import java.nio.ByteBuffer;
import java.util.ArrayList;

import mpi.MPI;
import mpi.MPIException;
import mpi.Win;

public class Counter {
    Win win;
    int hostRank;
    int myVal;
    ByteBuffer data;
    int rank;
    int size;

    public Counter(int hostRank) throws MPIException {
        this.setHostRank(hostRank);
        this.setSize(MPI.COMM_WORLD.getSize());
        this.setRank(MPI.COMM_WORLD.getRank());

        if (this.getRank() == hostRank) {
//          this.setData(MPI.newByteBuffer(this.getSize() * Integer.BYTES));
            this.setData(ByteBuffer.allocateDirect(this.getSize() * Integer.BYTES));
            for (int i = 0; i < this.getData().capacity(); i += Integer.BYTES)
                this.getData().putInt(i, 0);
        } else {
//          this.setData(MPI.newByteBuffer(0));
            this.setData(ByteBuffer.allocateDirect(0));
        }   

        this.setWin(new Win(this.getData(), this.getData().capacity(), Integer.BYTES,
                MPI.INFO_NULL, MPI.COMM_WORLD));

        this.setMyVal(0);
    }

    public int increment(int increment) throws MPIException {

        // A list to store all of the values we pull
        ArrayList<Integer> vals = new ArrayList<Integer>();
        for (int i = 0; i < this.getSize(); i++)
            vals.add(i, 0);

        // Need to convert the increment to a buffer
        ByteBuffer incrbuff = ByteBuffer.allocateDirect(Integer.BYTES);
        incrbuff.putInt(increment);

        // Our values are returned to us in a byte buffer
        ByteBuffer valBuff = ByteBuffer.allocateDirect(Integer.BYTES);

//      System.out.printf("Data for RANK %d: ", this.getRank());
        this.getWin().lock(MPI.LOCK_EXCLUSIVE, this.getHostRank(), 0);
        for (int i = 0; i < this.getSize(); i++) {
            // Always ensure that we're at the top of the buffer
            valBuff.position(0);
            if (i == this.getRank()) {
                this.getWin().accumulate(incrbuff, 1, MPI.INT, this.getHostRank(), i, 1, MPI.INT, MPI.SUM);
                // Without this, it comes back all 1s 
                this.getWin().flushLocalAll();
//              System.out.printf(" [%d] ", this.getMyVal() + increment);
            } else {
                this.getWin().get(valBuff, 1, MPI.INT, this.getHostRank(), i, 1, MPI.INT);
                vals.set(i, valBuff.getInt(0));
//              System.out.printf("  %d  ", vals.get(i))
            }
        }
        this.getWin().unlock(this.getHostRank());

        this.setMyVal(this.getMyVal() + increment);
        vals.set(this.getRank(), this.getMyVal());

//      System.out.printf(" <<%d>> \n", vals.stream().mapToInt(Integer::intValue).sum());
//      this.getWin().unlock(this.getHostRank());

        return vals.stream().mapToInt(Integer::intValue).sum();

    }

    public void printCounter() {
        if (this.getRank() == this.getHostRank()) {
            for (int i = 0; i < this.getSize(); i++) {
                System.out.printf(" %d ", this.getData().getInt());
            }
            System.out.println("");
        }
    }

    public void delete() throws MPIException {
        this.getWin().detach(this.getData());
        this.getWin().free();

        this.setData(null);
        this.setHostRank(0);
        this.setMyVal(0);
        this.setRank(0);
        this.setSize(0);
        this.setWin(null);

    }

    private Win getWin() {
        return win;
    }

    private void setWin(Win win) {
        this.win = win;
    }

    private int getHostRank() {
        return hostRank;
    }

    private void setHostRank(int hostrank) {
        this.hostRank = hostrank;
    }

    private int getMyVal() {
        return myVal;
    }

    private void setMyVal(int myval) {
        this.myVal = myval;
    }

    private ByteBuffer getData() {
        return data;
    }

    private void setData(ByteBuffer data) {
        this.data = data;
    }

    private int getRank() {
        return rank;
    }

    private void setRank(int rank) {
        this.rank = rank;
    }

    private int getSize() {
        return size;
    }

    private void setSize(int size) {
        this.size = size;
    }

}

Следует также отметить, что код Java включает в себя то, чего нет в коде C ++:

this.getWin().flushLocalAll();

Без этого счетчик будет равен "1" длякаждый ранг.

Здесь также находится первая часть тестового класса:

import java.util.Random;

import mpi.*;

public class CounterTest {

    public static void main(String[] args) {

        try {
            MPI.Init(args);
        } catch (MPIException e1) {
            // TODO Auto-generated catch block
            e1.printStackTrace();
        }

        try {
            test1();
//          test2();
        } catch (MPIException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

        try {
            MPI.Finalize();
        } catch (MPIException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }

    }

    public static void test1 () throws MPIException {
        Counter c = new Counter(0);
        int rank = MPI.COMM_WORLD.getRank();
        int size = MPI.COMM_WORLD.getSize();

        int result = c.increment(1);
        System.out.printf("%d got counter %d\n", rank, result);

        MPI.COMM_WORLD.barrier();
        c.printCounter();
        c.delete();
        c = null;                       

    }
}

Я пробовал различные другие методы, с точки зрения попытки ограбления, использования группы для использования MPI_Win_start () и MPI_Win_complete (), но безрезультатно.Я чувствую, что это настолько близко к истинному представлению исходного кода C ++, который я могу получить.

Чего мне не хватает?Почему это не ведет себя так же, как собственный код C ++?

РЕДАКТИРОВАТЬ: Я также считаю, что мне нужно добавить это при запуске этого против фактического кластера (он был отключен для обслуживанияпоследние два дня):

this.getWin().flush(0);

1 Ответ

0 голосов
/ 31 мая 2018

Я думаю, что проблема с этими строками

this.getWin().get(valBuff, 1, MPI.INT, this.getHostRank(), i, 1, MPI.INT);
vals.set(i, valBuff.getInt(0));

Насколько я понимаю, вы не можете предположить, что содержание valBuff является правильным до того, как MPI_Win_unlock() был вызван.

Я переписал подпрограмму, используя несколько буферов, и установил vals после MPI_Win_unlock() и смог получить правильный вывод.

public int increment(int increment) throws MPIException {

    // A list to store all of the values we pull
    ArrayList<Integer> vals = new ArrayList<Integer>();
    for (int i = 0; i < this.getSize(); i++)
        vals.add(i, 0);

    // Need to convert the increment to a buffer
    ByteBuffer incrbuff = ByteBuffer.allocateDirect(Integer.BYTES);
    incrbuff.putInt(increment);

    // Our values are returned to us in several byte buffers
    ByteBuffer valBuff[] = new ByteBuffer[this.getSize()];

    this.getWin().lock(MPI.LOCK_EXCLUSIVE, this.getHostRank(), 0);
    for (int i = 0; i < this.getSize(); i++) {
        // Always ensure that we're at the top of the buffer
        if (i == this.getRank()) {
            this.getWin().accumulate(incrbuff, 1, MPI.INT, this.getHostRank(), i, 1, MPI.INT, MPI.SUM);
        } else {
            valBuff[i] = ByteBuffer.allocateDirect(Integer.BYTES);
            valBuff[i].position(0);
            this.getWin().get(valBuff[i], 1, MPI.INT, this.getHostRank(), i, 1, MPI.INT);
        }
    }
    this.getWin().unlock(this.getHostRank());
    for (int i = 0; i < this.getSize(); i++) {
        if (i != this.getRank()) {
            vals.set(i, valBuff[i].getInt(0));
        }
    }

    this.setMyVal(this.getMyVal() + increment);
    vals.set(this.getRank(), this.getMyVal());

    return vals.stream().mapToInt(Integer::intValue).sum();

}

Примечание больше не существуетпотребность в

this.getWin().flushLocalAll();
this.getWin().flush(0);

FWIW, я пытался использовать один буфер из this.getSize() целых чисел, но не смог заставить что-то работать.

...