Commit a01f1f76 authored by lykos98's avatar lykos98
Browse files

cleaned up heap exchange with all to all

parent c036a473

bb_plot.ipynb

+1 −1

File changed.

Preview size limit exceeded, changes collapsed.

src/tree/tree.c

+49 −68
Original line number Diff line number Diff line
@@ -1399,61 +1399,56 @@ void mpi_ngbh_search(global_context_t* ctx, datapoint_info_t* dp_info, top_kdtre 


   

 

    int* count_rcv_work_batches = (int*)malloc(ctx -> world_size * sizeof(int));

    float_t** rcv_work_batches = (float_t**)malloc(ctx -> world_size * sizeof(float_t*));

    for(int i = 0; i < ctx -> world_size; ++i) 

    {

        count_rcv_work_batches[i] = point_to_rcv_count[i];

        //rcv_work_batches[i]       = NULL;

        rcv_work_batches[i]       = __rcv_points + rcv_displ[i];

    }



//	for(int i = 0; i < ctx -> world_size; ++i)

//	{

//        if(point_to_snd_count[i] > 0)

//        {

//            MPI_Request request;

//            //MPI_Isend(const void *buf, int count, MPI_Datatype datatype, int dest, int tag, MPI_Comm comm, MPI_Request *request)

//            MPI_Isend( data_to_send_per_proc[i], 

//                    point_to_snd_count[i]*(1 + ctx -> dims) ,  /* 1 per max_dist + point*/

//                    MPI_MY_FLOAT, i, 0, ctx -> mpi_communicator, &request);

//        }

//	}

//    

//

	MPI_Status status;

	int flag;

//	int cc = 0;

//    MPI_Barrier(ctx -> mpi_communicator);

//	MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, ctx -> mpi_communicator, &flag, &status);

//	while(flag)

//	{

//		cc++;

//		MPI_Request request;

//        int count; 

//        MPI_Get_count(&status, MPI_MY_FLOAT, &count);

//        rcv_work_batches[status.MPI_SOURCE] = (float_t*)malloc(count * sizeof(float_t));

//        //MPI_Recv(void *buf, int count, MPI_Datatype datatype, int source, int tag, MPI_Comm comm, MPI_Status *status)

//

//		MPI_Recv(rcv_work_batches[status.MPI_SOURCE], count, MPI_MY_FLOAT, status.MPI_SOURCE, MPI_ANY_TAG, ctx -> mpi_communicator, &status);

//        count_rcv_work_batches[status.MPI_SOURCE] = count / (1 + ctx -> dims);

//		MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_SOURCE, ctx -> mpi_communicator, &flag, &status);

//

//	}



    /* prepare heap batches */



    int work_batch_stride = 1 + ctx -> dims;



    heap_node* __heap_batches_to_snd = (heap_node*)malloc(k * tot_points_rcv * sizeof(heap_node));

    heap_node* __heap_batches_to_rcv = (heap_node*)malloc(k * tot_points_snd * sizeof(heap_node));



    /*

     * need sizes in bytes

     */



    rcv_displ[0] = 0;

    snd_displ[0] = 0;

    rcv_count[0] = point_to_rcv_count[0] * k * sizeof(heap_node);

    snd_count[0] = point_to_snd_count[0] * k * sizeof(heap_node); 





    for(int i = 1; i < ctx -> world_size; ++i)

    {

        rcv_count[i] = point_to_rcv_count[i] * k * sizeof(heap_node); 

        snd_count[i] = point_to_snd_count[i] * k * sizeof(heap_node); 



        rcv_displ[i] = rcv_displ[i - 1] + rcv_count[i - 1];

        snd_displ[i] = snd_displ[i - 1] + snd_count[i - 1];

    }



    heap_node** heap_batches_per_node = (heap_node**)malloc(ctx -> world_size * sizeof(heap_node*));

    for(int p = 0; p < ctx -> world_size; ++p) heap_batches_per_node[p] = NULL;

    for(int p = 0; p < ctx -> world_size; ++p) 

    {

        //heap_batches_per_node[p] = NULL;

        heap_batches_per_node[p] = __heap_batches_to_snd + rcv_displ[p] / sizeof(heap_node);

    }



    /* compute everything */



    for(int p = 0; p < ctx -> world_size; ++p)

    {

        if(point_to_rcv_count[p] > 0)

        //if(count_rcv_work_batches[p] > 0)

        {

            heap_batches_per_node[p] = (heap_node*)malloc(k * count_rcv_work_batches[p] * sizeof(heap_node));

            //heap_batches_per_node[p] = (heap_node*)malloc(k * point_to_rcv_count[p] * sizeof(heap_node));

            for(int batch = 0; batch < point_to_rcv_count[p]; ++batch)

            {

                heap H;
@@ -1467,43 +1462,26 @@ void mpi_ngbh_search(global_context_t* ctx, datapoint_info_t* dp_info, top_kdtre 
        }

    }



    /* sendout */

    /* 

     * counts are inverted since I have to recieve as many batches as points I

     * Have originally sended

     */

    MPI_Alltoallv(__heap_batches_to_snd, rcv_count, rcv_displ, MPI_CHAR, 

                  __heap_batches_to_rcv, snd_count, snd_displ, MPI_CHAR, ctx -> mpi_communicator );



    /* 

     * send out heaps 

     * and rcv counterparts

     */

	MPI_Barrier(ctx -> mpi_communicator);



    heap_node** rcv_heap_batches = (heap_node**)malloc(ctx -> world_size * sizeof(heap_node*));

    for(int i = 0; i < ctx -> world_size; ++i) rcv_heap_batches[i] = NULL;



    for(int i = 0; i < ctx -> world_size; ++i)

    {

        if(count_rcv_work_batches[i] > 0)

        {

            MPI_Request request;

            //MPI_Isend(const void *buf, int count, MPI_Datatype datatype, int dest, int tag, MPI_Comm comm, MPI_Request *request)

            MPI_Isend( heap_batches_per_node[i], 

                    sizeof(heap_node) * k * count_rcv_work_batches[i],  /* 1 per max_dist + point*/

                    MPI_CHAR, i, 0, ctx -> mpi_communicator, &request);

        //rcv_heap_batches[i] = NULL;

        rcv_heap_batches[i] = __heap_batches_to_rcv + snd_displ[i] / sizeof(heap_node);

    }

	}





	MPI_Barrier(ctx -> mpi_communicator);

	MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, ctx -> mpi_communicator, &flag, &status);

    //DB_PRINT("%d %p %p\n",ctx -> mpi_rank, &flag, &status);

	while(flag)

	{

		MPI_Request request;

        int count; 

        MPI_Get_count(&status, MPI_CHAR, &count);

        rcv_heap_batches[status.MPI_SOURCE] = (heap_node*)malloc(count);



		MPI_Recv(rcv_heap_batches[status.MPI_SOURCE], count, MPI_CHAR, status.MPI_SOURCE, MPI_ANY_TAG, ctx -> mpi_communicator, &status);

		MPI_Iprobe(MPI_ANY_SOURCE, MPI_ANY_TAG, ctx -> mpi_communicator, &flag, &status);

	}



    MPI_Barrier(ctx -> mpi_communicator);

    /* merge old with new heaps */



	for(int i = 0; i < ctx -> world_size; ++i)
@@ -1550,17 +1528,16 @@ void mpi_ngbh_search(global_context_t* ctx, datapoint_info_t* dp_info, top_kdtre 
    

	for(int i = 0; i < ctx -> world_size; ++i)

	{

        if(heap_batches_per_node[i])  free(heap_batches_per_node[i]);

        //if(heap_batches_per_node[i])  free(heap_batches_per_node[i]);

		if(data_to_send_per_proc[i])  free(data_to_send_per_proc[i]);

		if(local_idx_of_the_point[i]) free(local_idx_of_the_point[i]);

        //if(rcv_work_batches[i])       free(rcv_work_batches[i]);

        if(rcv_heap_batches[i])       free(rcv_heap_batches[i]);

        //if(rcv_heap_batches[i])       free(rcv_heap_batches[i]);

	}



    free(heap_batches_per_node);

    free(rcv_heap_batches);

    free(rcv_work_batches);

    free(count_rcv_work_batches);

    free(point_to_rcv_count);

	free(point_to_snd_count);

	free(point_to_snd_capacity);
@@ -1569,6 +1546,10 @@ void mpi_ngbh_search(global_context_t* ctx, datapoint_info_t* dp_info, top_kdtre 
    free(snd_count);

    free(rcv_displ);

    free(snd_displ);

    free(__heap_batches_to_rcv);

    free(__heap_batches_to_snd);

    free(__rcv_points);

    free(__snd_points);

}



void test_the_idea(global_context_t* ctx)