Wrong usage of upcxx Constructs ?

[NecOnIce]

Hi,

i have another example program which performs an KMeans on a set of Points. This program always fails with segmentation faults or other memory related errors like sigabrt... Have i used one of the upcxx Constructs in a wrong way or something like this ?

//

#define GASNET_SEQ

#include <iostream>
#include <cmath>
#include <memory>
#include <random>
#include "hcupc_spmd.h"
#include <chrono>
#include <iomanip>
#include <mutex>

#define CLUSTERS 5
#define DIMENSION 4
#define NUM_POINTS 200000
#define ITERATIONS 50
#define THRESHOLD 2500;

using namespace std;

upcxx::global_ptr<float> *globalPoints;
upcxx::global_ptr<float> *newCenters;
upcxx::global_ptr<int> *newCounts;
upcxx::global_ptr<float> globalCurrentClusters;
upcxx::shared_lock centerLock;
upcxx::shared_var<int> numberOfPointsPerPlace;
std::mutex centerMutex;

void calculateClusterCenters(size_t startIndex, size_t endIndex);

int main(int argc, char **argv) {

    // start HabaneroUPC++ Runtime
    hupcpp::launch(&argc, &argv, [] () {

        numberOfPointsPerPlace.put(NUM_POINTS / upcxx::ranks());

        cout << "Resilient K-Means: " << CLUSTERS << " clusters, " << NUM_POINTS
        << " points, " << DIMENSION << " dimensions, " << upcxx::ranks()
        << " places, " << hupcpp::numWorkers() << " threads" << endl;

        globalPoints = new upcxx::global_ptr<float> [upcxx::ranks()];
        newCenters = new upcxx::global_ptr<float> [upcxx::ranks()];
        newCounts = new upcxx::global_ptr<int> [upcxx::ranks()];
        for (size_t i = 0; i < upcxx::ranks(); i++) {
            globalPoints[i] = upcxx::allocate<float>(i, numberOfPointsPerPlace * DIMENSION);
            newCenters[i] = upcxx::allocate<float>(i, CLUSTERS*DIMENSION);
            newCounts[i] = upcxx::allocate<int>(i, CLUSTERS);
        }

        // initialize Points on each Place
        hupcpp::finish_spmd([] () {

            for (size_t i = 0; i < upcxx::ranks(); i++) {

                if (upcxx::myrank() == 0) {

                    hupcpp::asyncAt(i, [] () {

                        // create random generator
                        default_random_engine generator;
                        uniform_real_distribution<float> distribution(0.0f, 1.0f);

                        float *localPoints = globalPoints[upcxx::myrank()].raw_ptr();

//                        cout << "init points " << upcxx::myrank() << endl;

                        // fill the localPoints
                        for (size_t j = 0; j < numberOfPointsPerPlace; ++j) {
                            for (size_t k = 0; k < DIMENSION; ++k) {
                                localPoints[j*DIMENSION+k] = distribution(generator);
                            }
                        }
                    });
                }
            }
        });

        // initialize centers and counters on rank 0
        globalCurrentClusters = upcxx::allocate<float>(0, CLUSTERS*DIMENSION);

        if (upcxx::myrank() == 0) {

            float *localPoints = globalPoints[upcxx::myrank()].raw_ptr();

            for (size_t i = 0; i < CLUSTERS; i++) {
                for (size_t j = 0; j< DIMENSION; j++) {
                    globalCurrentClusters[i*DIMENSION+j] = localPoints[i*DIMENSION+j];
                }
            }
         }

        auto startTime = chrono::high_resolution_clock::now();
        for (size_t iter = 0; iter < ITERATIONS; ++iter) {

            hupcpp::finish_spmd([] () {

                for (size_t i = 0; i < upcxx::ranks(); i++) {

                    if (upcxx::myrank() == 0) {

                        hupcpp::asyncAt(i, [] () {

                            float *localCurrentClusters = new float[CLUSTERS*DIMENSION];
                            upcxx::copy(globalCurrentClusters, localCurrentClusters, CLUSTERS*DIMENSION);
                            float *localClusterCenters = newCenters[upcxx::myrank()].raw_ptr();
                            float *localPoints = globalPoints[upcxx::myrank()].raw_ptr();
                            int *localClusterCounts = newCounts[upcxx::myrank()].raw_ptr();

                            for (int c = 0; c < CLUSTERS; c++) {
                                localClusterCounts[c] = 0;
                                for (int d = 0; d < DIMENSION; d++) {
                                    localClusterCenters[c*DIMENSION+d] = 0.0f;
                                }
                            }

                            hupcpp::finish([] () {
                               calculateClusterCenters(0, numberOfPointsPerPlace);
                            });

                            // compute new clusters and counters
//                            for (size_t j = 0; j < numberOfPointsPerPlace; ++j) {
//
//                                int closest = -1; // closest Cluster number
//                                float closestDist = numeric_limits<float>::max(); // closest Distance to Cluster
//
//                                for (int k = 0; k < CLUSTERS; ++k) {
//
//                                    float dist = 0; // currentDistance
//
//                                    // quadratic distance from central Cluster to current Point
//                                    for (size_t d = 0; d < DIMENSION; ++d) {
//                                        const float tmp = localPoints[j*DIMENSION+d]
//                                                          - localCurrentClusters[k*DIMENSION+d];
//                                        dist += tmp * tmp;
//                                    }
//
//                                    // if the current distance is the smallest,
//                                    // associate the current Point to the current Cluster
//                                    if (dist < closestDist) {
//                                        closestDist = dist;
//                                        closest = k;
//                                    }
//                                }
//
//                                // sum up the Points to the nearest Cluster to build the new Cluster center
//                                for (int d = 0; d < DIMENSION; ++d) {
//                                    localClusterCenters[closest*DIMENSION+d] +=
//                                            localPoints[j*DIMENSION+d];
//                                }
//
//                                // increment the counter for the closest Cluster
//                                localClusterCounts[closest]++;
//                            }

                            delete[](localCurrentClusters);
                        });
                    }

                }
            });

            if (upcxx::myrank() == 0) {

                int clustersSize = CLUSTERS*DIMENSION;
                float *tempClusters = new float[upcxx::ranks()*clustersSize];
                int * tempCounts = new int[upcxx::ranks()*CLUSTERS];

                // copy the newClusters and counts to local
                for (size_t i = 0; i < upcxx::ranks(); i++) {

                    upcxx::async_copy(newCenters[i], tempClusters+i*clustersSize, clustersSize);
                    upcxx::async_copy(newCounts[i], tempCounts+i*CLUSTERS, CLUSTERS);
                }

                upcxx::async_copy_fence();

//                for (size_t i = 0; i  < upcxx::ranks(); i++) {
//                    cout << "centers of rank: " << i << endl;
//                    // print the result
//                    for (int c = 0; c < CLUSTERS; c++) {
//                        for (int j = 0; j < DIMENSION; j++) {
//                            if (j > 0) {
//                                cout << " ";
//                            }
//                            cout <<  tempClusters[i*clustersSize+c*DIMENSION+j];
//                        }
//                        cout << endl;
//                    }
//                }

                // reduce all values to the array of rank 0
                for (size_t i = 1; i < upcxx::ranks(); i++) {

                    for (size_t c = 0; c < CLUSTERS; c++) {
                        for (size_t d = 0; d < DIMENSION; d++) {
                            tempClusters[c*DIMENSION+d] += tempClusters[i*clustersSize+c*DIMENSION+d];
                        }
                        tempCounts[c] += tempCounts[i*CLUSTERS+c];
                    }
                }

                // normalize the new cluster centers and set the new globalCurrentClusters
                for (size_t c = 0; c < CLUSTERS; c++) {
                    for (size_t d = 0; d < DIMENSION; d++) {
                        globalCurrentClusters[c*DIMENSION+d] = tempClusters[c*DIMENSION+d] / tempCounts[c];
                    }
                }

                delete[](tempClusters);
                delete[](tempCounts);
            }
        }

        auto endTime = chrono::high_resolution_clock::now();

        if (upcxx::myrank() == 0) {

            // print the result
            for (int i = 0; i < CLUSTERS; i++) {
                for (int j = 0; j < DIMENSION; j++) {
                    if (j > 0) {
                        cout << " ";
                    }
                    cout <<  globalCurrentClusters[i*DIMENSION+j];
                }
                cout << endl;
            }

            cout << "Wall clock time passed: " << chrono::duration<double, milli>(endTime - startTime).count()/ITERATIONS
            << " ms" << endl;
        }
    });

    return 0;
}

void calculateClusterCenters(size_t startIndex, size_t endIndex) {

    size_t nPoints = endIndex - startIndex;
    size_t thresh = THRESHOLD;
    if (nPoints < thresh) {

        float *currentClusters = new float[CLUSTERS*DIMENSION];
        upcxx::copy(globalCurrentClusters, currentClusters, CLUSTERS*DIMENSION);
        float *points = globalPoints[upcxx::myrank()].raw_ptr();

        float *newClustersTmp = new float[CLUSTERS*DIMENSION];
        int *clusterCountsTmp = new int[CLUSTERS];

        for (size_t i = 0; i < CLUSTERS; i++) {
            clusterCountsTmp[i] = 0;
            for (size_t j = 0; j < DIMENSION; j++) {
                newClustersTmp[i*DIMENSION+j] = 0;
            }
        }

        // compute new clusters and counters
        for (size_t j = startIndex; j < endIndex; ++j) {

            int closest = -1; // closest Cluster number
            float closestDist = numeric_limits<float>::max(); // closest Distance to Cluster
            for (int k = 0; k < CLUSTERS; ++k) {
                float dist = 0; // currentDistance
                // quadratic distance from central Cluster to current Point
                for (size_t d = 0; d < DIMENSION; ++d) {
                    const float tmp = points[j*DIMENSION+d]
                                      - currentClusters[k*DIMENSION+d];
                    dist += tmp * tmp;
                }

                // if the current distance is the smallest,
                // associate the current Point to the current Cluster
                if (dist < closestDist) {
                    closestDist = dist;
                    closest = k;
                }
            }

            // sum up the Points to the nearest Cluster to build the new Cluster center
            for (int d = 0; d < DIMENSION; ++d) {
                newClustersTmp[closest*DIMENSION+d] += points[j*DIMENSION+d];
            }
            // increment the counter for the closest Cluster
            clusterCountsTmp[closest]++;
        }

        lock_guard<std::mutex> lock(centerMutex);
        float *newClusters = newCenters[upcxx::myrank()].raw_ptr();
        int *clusterCounts = newCounts[upcxx::myrank()].raw_ptr();
        for (int i = 0; i < CLUSTERS; i++) {
            clusterCounts[i] += clusterCountsTmp[i];
            for (int j = 0; j < DIMENSION; j++) {
                newClusters[i*DIMENSION+j] += newClustersTmp[i*DIMENSION+j];
            }
        }

    } else {

        size_t mid = nPoints/2 + startIndex;
        hupcpp::async([ startIndex, mid] () {
            calculateClusterCenters(startIndex, mid);
        });
        hupcpp::async([mid, endIndex] () {
            calculateClusterCenters(mid, endIndex);
        });
    }

}

[NecOnIce]

In Addition, these error only occure if i use more than one node e.g. with mpirun -np 2. If i use mpirun -np 1 the Program runs fine and without errors.

[agrippa]

@NecOnIce This may be a result of using unsupported UPCXX constructs from inside asynchronous tasks, but we believe there is also an issue in the interaction between UPCXX and HClib that could cause similar behavior. I'm working on a fix for the latter, so I'd like to get that fix in first and then verify that we still see this problem before saying it is the former.

[NecOnIce]

My Programs are working now correctly by avoiding upcxx Constructs inside the asynchronous tasks.