Программа где-то застряла

Я пытаюсь отправить шаблон реализации мастер-подчиненный, в котором мастер имеет массив (действует как очередь заданий) и отправляет данные подчиненным процессорам. На основе данных, полученных от ведущего, ведомые устройства вычисляют результаты и возвращают ответы ведущему. Мастер при получении результатов определяет ранг подчиненного, от которого было получено сообщение, и затем отправляет следующее задание этому подчиненному устройству.

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

        if (my_rank != 0) 
        {
            MPI_Recv(&seed, 1, MPI_FLOAT, 0, tag, MPI_COMM_WORLD, &status);

                    //.. some processing 

            MPI_Send(&message, 100, MPI_FLOAT, 0, my_rank, MPI_COMM_WORLD);
        } 
        else 
        {
            for (i = 1; i < p; i++) {
                MPI_Send(&A[i], 1, MPI_FLOAT, i, tag, MPI_COMM_WORLD);
            }

            for (i = p; i <= S; i++) {
                MPI_Recv(&buf, 100, MPI_FLOAT, MPI_ANY_SOURCE, MPI_ANY_TAG,
                        MPI_COMM_WORLD, &status);
                //.. processing to find out free slave rank from which above msg was received (y)
                MPI_Send(&A[i], 1, MPI_FLOAT, y, tag, MPI_COMM_WORLD);
            }

            for (i = 1; i < p; i++) {
                MPI_Recv(&buf, 100, MPI_FLOAT, MPI_ANY_SOURCE, MPI_ANY_TAG,MPI_COMM_WORLD, &status);

                // .. more processing 
            }

        }

Если я использую 4 процессора; 1 - главный, 3 - подчиненные; программа отправляет и принимает сообщения для первых 3 заданий в очереди заданий, но после этого программа зависает. В чем может быть проблема?


parallel-processing mpi openmpi
person Kany    schedule 05.03.2015    source источник
comment
Похоже, что один из процессов умирает до того, как отправит ответ. Выясните, какой процесс не отправляет ответ основному процессу. Здесь может быть полезен некоторый отладочный код.   -  person tdbeckett    schedule 05.03.2015
comment
Это очень неполно.   -  person Vladimir F    schedule 05.03.2015
comment
^ Это единственный код MPI, в котором я отправляю и получаю. все остальное мне кажется нормальным.   -  person Kany    schedule 05.03.2015

Ответы (1)


arrow_upward
0
arrow_downward

Если это весь ваш код на основе MPI, то похоже, что вам не хватает цикла while вне клиентского кода. Я делал это раньше, и я обычно разбиваю это как TaskMaster и подчиненные

в taskMaster:

 for (int i = 0; i < commSize; ++i){
    if (i == commRank){ // commRank doesn't have to be 0
        continue;
    }

    if (taskNum < taskCount){
        // tasks is vector<Task>, where I have crated a Task 
        // class and send it as a stream of bytes
        toSend = tasks.at(taskNum);  
        jobList.at(i) = taskNum;  // so we no which rank has which task
        taskNum += 1;
        activePeons += 1;
    } else{
        // stopTask is a flag value to stop receiving peon
        toSend = stopTask;
        allTasksDistributed = true;
    }

    // send the task, with the size of the task as the tag
    taskSize = sizeof(toSend);
    MPI_Send(&toSend, taskSize, MPI_CHAR, i, taskSize, MPI_COMM_WORLD);
}   

MPI_Status status;

while (activePeons > 0){ 
    // get the results from a peon (but figure out who it is coming from and what the size is)
    MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);
    MPI_Recv(   &toSend,                    // receive the incoming task (with result data)
                status.MPI_TAG,             // Tag holds number of bytes
                MPI_CHAR,                   // type, may need to be more robust later
                status.MPI_SOURCE,          // source of send
                MPI_ANY_TAG,                // tag
                MPI_COMM_WORLD,             // COMM
                &status);                   // status

    // put the result from that task into the results vector
    results[jobList[status.MPI_SOURCE]] = toSend.getResult();

    // if there are more tasks to send, distribute the next one
    if (taskNum < taskCount ){
        toSend = tasks.at(taskNum);
        jobList[status.MPI_SOURCE] = taskNum;
        taskNum += 1;
    } else{ // otherwise send the stop task and decrement activePeons
        toSend = stopTask;
        activePeons -= 1;
    }

    // send the task, with the size of the task as the tag
    taskSize = sizeof(toSend);
    MPI_Send(&toSend, taskSize, MPI_CHAR, status.MPI_SOURCE, taskSize, MPI_COMM_WORLD);
}

в функции пеона:

while (running){
    MPI_Probe(MPI_ANY_SOURCE, MPI_ANY_TAG, MPI_COMM_WORLD, &status);    // tag holds data size

    incoming = (Task *) malloc(status.MPI_TAG);

    MPI_Recv(   incoming,           // memory location of input
                status.MPI_TAG,     // tag holds data size
                MPI_CHAR,           // type of data
                status.MPI_SOURCE,  // source is from distributor
                MPI_ANY_TAG,        // tag
                MPI_COMM_WORLD,     // comm
                &status);           // status

    task = Task(*incoming);

    if (task.getFlags() == STOP_FLAG){
        running = false;
        continue;
    }

    task.run();   // my task class has a "run" method
    MPI_Send(   &task,                  // string to send back
                status.MPI_TAG,         // size in = size out
                MPI_CHAR,               // data type
                status.MPI_SOURCE,      // destination
                status.MPI_TAG,         // tag doesn't matter
                MPI_COMM_WORLD);        // comm

    free(incoming);
}

Есть несколько значений bool и int, которые должны быть назначены (и, как я уже сказал, у меня есть класс Task), но это дает базовую структуру для того, что, я думаю, вы хотите сделать.

person Dr.Tower    schedule 06.03.2015
comment
Я, вероятно, придираюсь к этому вопросу, но почему вы отправляете размер данных в виде тега, когда есть MPI_GET_COUNT, который можно использовать для получения количества элементов в (просмотренном) сообщении с учетом объекта MPI_Status? - person Hristo Iliev; 11.03.2015
comment
Честно говоря, я раньше не слышал об этой функции. Я, вероятно, по-прежнему предпочитаю использовать status.count, а не выполнять вызов функции со статусом, который у меня уже есть, но это то, что у меня хорошо сработало. - person Dr.Tower; 11.03.2015
comment
Если вы это сделаете, вы должны иметь в виду, что стандарт MPI требует, чтобы максимальное значение тега было не менее 32767. Реализации могут предоставлять гораздо большее пространство тегов, и большинство из них это делает, но если вы неявно полагаетесь на предположение, что это всегда верно, и установите для тега значения больше 32767 (например, ваши сообщения длинные), ваши программы MPI не будут переносимы. - person Hristo Iliev; 12.03.2015