<commit_before>/*
This file is part of Bohrium and copyright (c) 2012 the Bohrium
team <http://www.bh107.org>.

Bohrium is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as
published by the Free Software Foundation, either version 3
of the License, or (at your option) any later version.

Bohrium is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the
GNU Lesser General Public License along with Bohrium.

If not, see <http://www.gnu.org/licenses/>.
*/

#include <bh.h>
#include <cassert>
#include <map>
#include "pgrid.h"
#include "array.h"
#include "exec.h"
#include "except.h"
#include "batch.h"
#include "tmp.h"


/* Gather or scatter the global base array between processes.
 * NB: this is a collective operation.
 *
 * @scatter If true we scatter else we gather
 * @global_ary Global base array
 */
static void comm_gather_scatter(int scatter, bh_base *global_ary)
{
    bh_error err;
    bh_base *local_ary = array_get_local(global_ary);
    bh_intp totalsize = global_ary->nelem;

    if(totalsize <= 0)
        return;

    //Find the local size for all processes
    int sendcnts[pgrid_worldsize], displs[pgrid_worldsize];
    {
        bh_intp s = totalsize / pgrid_worldsize;//local size for all but the last process
        s *= bh_type_size(global_ary->type);
        for(int i=0; i<pgrid_worldsize; ++i)
        {
            sendcnts[i] = s;
            displs[i] = s * i;
        }
        //The last process gets the rest
        sendcnts[pgrid_worldsize-1] += totalsize % pgrid_worldsize * bh_type_size(global_ary->type);
    }

    int e;
    if(scatter)
    {
        //The slave-processes may need to allocate memory
        if(sendcnts[pgrid_myrank] > 0 && local_ary->data == NULL)
        {
            if((err = bh_data_malloc(local_ary)) != BH_SUCCESS)
                EXCEPT_OUT_OF_MEMORY();
        }
        //The master-process MUST have allocated memory already
        assert(pgrid_myrank != 0 || global_ary->data != NULL);

        //Scatter from master to slaves
        e = MPI_Scatterv(global_ary->data, sendcnts, displs, MPI_BYTE,
                         local_ary->data, sendcnts[pgrid_myrank], MPI_BYTE,
                         0, MPI_COMM_WORLD);
    }
    else
    {
        //The master-processes may need to allocate memory
        if(pgrid_myrank == 0 && global_ary->data == NULL)
        {
            if((err = bh_data_malloc(global_ary)) != BH_SUCCESS)
                EXCEPT_OUT_OF_MEMORY();
        }

        //We will always allocate the local array when gathering because
        //only the last process knows if the array has been initiated.
        if((err = bh_data_malloc(local_ary)) != BH_SUCCESS)
            EXCEPT_OUT_OF_MEMORY();

        assert(sendcnts[pgrid_myrank] == 0 || local_ary->data != NULL);

        //Gather from the slaves to the master
        e = MPI_Gatherv(local_ary->data, sendcnts[pgrid_myrank], MPI_BYTE,
                        global_ary->data, sendcnts, displs, MPI_BYTE,
                        0, MPI_COMM_WORLD);
    }
    if(e != MPI_SUCCESS)
        EXCEPT_MPI(e);
}


/* Distribute the global base array data to all slave processes.
 * The master-process MUST have allocated the @global_ary data.
 * NB: this is a collective operation.
 *
 * @global_ary Global base array
 */
void comm_master2slaves(bh_base *global_ary)
{
    batch_flush();
    comm_gather_scatter(1, global_ary);
}


/* Gather the global base array data at the master processes.
 * NB: this is a collective operation.
 *
 * @global_ary Global base array
 */
void comm_slaves2master(bh_base *global_ary)
{
    batch_flush();
    comm_gather_scatter(0, global_ary);
}


/* Communicate array data such that the processes can apply local computation.
 * NB: The process that owns the data and the process where the data is located
 *     must both call this function.
 *
 * @chunk          The local array chunk to communicate
 * @sending_rank   The rank of the sending process
 * @receiving_rank The rank of the receiving process, e.g. the process that should
 *                 apply the computation
 */
void comm_array_data(const bh_view &chunk, int sending_rank, int receiving_rank)
{
    //Check if communication is even necessary
    if(sending_rank == receiving_rank)
        return;

    if(pgrid_myrank == receiving_rank)
    {
        //Schedule the receive message
        batch_schedule_comm(0, sending_rank, chunk);
    }
    else if(pgrid_myrank == sending_rank)
    {
        //We need to copy the local array view into a base array.
        bh_view tmp_view = chunk;
        tmp_view.base = tmp_get_ary(bh_base_array(&chunk)->type, bh_nelements(chunk.ndim, chunk.shape));

        //Tell the VEM to do the data copy.
        bh_view ops[] = {tmp_view, chunk};
        batch_schedule_inst(BH_IDENTITY, ops, NULL);

        //Schedule the send message
        batch_schedule_comm(1, receiving_rank, tmp_view);

        //Cleanup the local arrays
        batch_schedule_inst(BH_FREE, bh_base_array(&chunk));
        batch_schedule_inst(BH_DISCARD, bh_base_array(&chunk));
    }
}


/* Communicate array data such that the processes can apply local computation.
 * This function may reshape the input array chunk.
 * NB: The process that owns the data and the process where the data is located
 *     must both call this function.
 *
 * @chunk The local array chunk to communicate
 * @receiving_rank The rank of the receiving process, e.g. the process that should
 *                 apply the computation
 */
void comm_array_data(const ary_chunk &chunk, int receiving_rank)
{<commit_msg>fixed cleanup bug in the communication<commit_after>/*
This file is part of Bohrium and copyright (c) 2012 the Bohrium
team <http://www.bh107.org>.

Bohrium is free software: you can redistribute it and/or modify
it under the terms of the GNU Lesser General Public License as
published by the Free Software Foundation, either version 3
of the License, or (at your option) any later version.

Bohrium is distributed in the hope that it will be useful,
but WITHOUT ANY WARRANTY; without even the implied warranty of
MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
GNU General Public License for more details.

You should have received a copy of the
GNU Lesser General Public License along with Bohrium.

If not, see <http://www.gnu.org/licenses/>.
*/

#include <bh.h>
#include <cassert>
#include <map>
#include "pgrid.h"
#include "array.h"
#include "exec.h"
#include "except.h"
#include "batch.h"
#include "tmp.h"


/* Gather or scatter the global base array between processes.
 * NB: this is a collective operation.
 *
 * @scatter If true we scatter else we gather
 * @global_ary Global base array
 */
static void comm_gather_scatter(int scatter, bh_base *global_ary)
{
    bh_error err;
    bh_base *local_ary = array_get_local(global_ary);
    bh_intp totalsize = global_ary->nelem;

    if(totalsize <= 0)
        return;

    //Find the local size for all processes
    int sendcnts[pgrid_worldsize], displs[pgrid_worldsize];
    {
        bh_intp s = totalsize / pgrid_worldsize;//local size for all but the last process
        s *= bh_type_size(global_ary->type);
        for(int i=0; i<pgrid_worldsize; ++i)
        {
            sendcnts[i] = s;
            displs[i] = s * i;
        }
        //The last process gets the rest
        sendcnts[pgrid_worldsize-1] += totalsize % pgrid_worldsize * bh_type_size(global_ary->type);
    }

    int e;
    if(scatter)
    {
        //The slave-processes may need to allocate memory
        if(sendcnts[pgrid_myrank] > 0 && local_ary->data == NULL)
        {
            if((err = bh_data_malloc(local_ary)) != BH_SUCCESS)
                EXCEPT_OUT_OF_MEMORY();
        }
        //The master-process MUST have allocated memory already
        assert(pgrid_myrank != 0 || global_ary->data != NULL);

        //Scatter from master to slaves
        e = MPI_Scatterv(global_ary->data, sendcnts, displs, MPI_BYTE,
                         local_ary->data, sendcnts[pgrid_myrank], MPI_BYTE,
                         0, MPI_COMM_WORLD);
    }
    else
    {
        //The master-processes may need to allocate memory
        if(pgrid_myrank == 0 && global_ary->data == NULL)
        {
            if((err = bh_data_malloc(global_ary)) != BH_SUCCESS)
                EXCEPT_OUT_OF_MEMORY();
        }

        //We will always allocate the local array when gathering because
        //only the last process knows if the array has been initiated.
        if((err = bh_data_malloc(local_ary)) != BH_SUCCESS)
            EXCEPT_OUT_OF_MEMORY();

        assert(sendcnts[pgrid_myrank] == 0 || local_ary->data != NULL);

        //Gather from the slaves to the master
        e = MPI_Gatherv(local_ary->data, sendcnts[pgrid_myrank], MPI_BYTE,
                        global_ary->data, sendcnts, displs, MPI_BYTE,
                        0, MPI_COMM_WORLD);
    }
    if(e != MPI_SUCCESS)
        EXCEPT_MPI(e);
}


/* Distribute the global base array data to all slave processes.
 * The master-process MUST have allocated the @global_ary data.
 * NB: this is a collective operation.
 *
 * @global_ary Global base array
 */
void comm_master2slaves(bh_base *global_ary)
{
    batch_flush();
    comm_gather_scatter(1, global_ary);
}


/* Gather the global base array data at the master processes.
 * NB: this is a collective operation.
 *
 * @global_ary Global base array
 */
void comm_slaves2master(bh_base *global_ary)
{
    batch_flush();
    comm_gather_scatter(0, global_ary);
}


/* Communicate array data such that the processes can apply local computation.
 * NB: The process that owns the data and the process where the data is located
 *     must both call this function.
 *
 * @chunk          The local array chunk to communicate
 * @sending_rank   The rank of the sending process
 * @receiving_rank The rank of the receiving process, e.g. the process that should
 *                 apply the computation
 */
void comm_array_data(const bh_view &chunk, int sending_rank, int receiving_rank)
{
    //Check if communication is even necessary
    if(sending_rank == receiving_rank)
        return;

    if(pgrid_myrank == receiving_rank)
    {
        //Schedule the receive message
        batch_schedule_comm(0, sending_rank, chunk);
    }
    else if(pgrid_myrank == sending_rank)
    {
        //We need to copy the local array view into a base array.
        bh_view tmp_view = chunk;
        tmp_view.base = tmp_get_ary(bh_base_array(&chunk)->type,
                                    bh_nelements(chunk.ndim, chunk.shape));

        //Tell the VEM to do the data copy.
        bh_view ops[] = {tmp_view, chunk};
        batch_schedule_inst(BH_IDENTITY, ops, NULL);

        //Schedule the send message
        batch_schedule_comm(1, receiving_rank, tmp_view);

        //Cleanup the local arrays
        batch_schedule_inst(BH_FREE, bh_base_array(&tmp_view));
        batch_schedule_inst(BH_DISCARD, bh_base_array(&tmp_view));
    }
}


/* Communicate array data such that the processes can apply local computation.
 * This function may reshape the input array chunk.
 * NB: The process that owns the data and the process where the data is located
 *     must both call this function.
 *
 * @chunk The local array chunk to communicate
 * @receiving_rank The rank of the receiving process, e.g. the process that should
 *                 apply the computation
 */
void comm_array_data(const ary_chunk &chunk, int receiving_rank)
{
{'fraction_non_alphanumeric': 0.08578494073505585, 'fraction_numerical': 0.003922571842756033, 'mean_word_length': 2.766217084136159, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 39, 'https://': 0, 'lorem ipsum': 0, 'www.': 4, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5176918', 'n_tokens_mistral': 3610, 'n_tokens_neox': 3465, 'n_words': 1187}