code/src/MPI_Main/slave.c

   1 #include "Util/types.h"
   2 #include "Util/utils.h"
   3 #include "TimeSeries/deserialize.h"
   4 #include "Algorithm/compute_coefficients.h"
   5 #include "Algorithm/get_dissimilarities.h"
   6 #include "Algorithm/pam.h"
   7 #include <mpi.h>
   8 #include <stdio.h>
   9 #include <math.h>
  10 #include <stdlib.h>
  11 #include <unistd.h>
  12 #include "MPI_Communication/unpack.h"
  13 #include "MPI_Communication/pack.h"
  14
  15 // main job done by a slave:
  16 static Result_t* do_work(Work_t* work)
  17 {
  18     // *** Step 1 ***
  19     // Obtain reduced coordinates matrix from time-series.
  20
  21     uint32_t nbSeries = work->nbSeries;
  22     uint32_t tsLength = get_tsLength(work->inputFileName);
  23     uint32_t nbValues = (tsLength - 4) / 3;
  24
  25     // nbReducedCoordinates = smallest power of 2 which is above nbValues
  26     uint32_t nbReducedCoordinates = (uint32_t)ceil(log2(nbValues));
  27     Real* reducedCoordinates = (Real*) malloc(nbSeries * nbReducedCoordinates * sizeof(Real));
  28
  29     // call preprocessing with the rows of raw power values matrix.
  30     // Keep the IDs in memory for further processing.
  31     uint32_t* IDs = (uint32_t*)malloc(nbSeries*sizeof(uint32_t));
  32     for (uint32_t i = 0; i < nbSeries; i+=CURVES_PER_REQUEST)
  33     {
  34         uint32_t nbCurvesRetrieved = CURVES_PER_REQUEST;
  35         if (i + nbCurvesRetrieved > nbSeries)
  36             nbCurvesRetrieved -= (i + nbCurvesRetrieved - nbSeries);
  37         PowerCurve* powerCurves =
  38             deserialize(work->inputFileName, NULL, work->ranks + i, nbCurvesRetrieved);
  39         compute_coefficients(powerCurves, nbCurvesRetrieved, nbValues,
  40             reducedCoordinates, i, nbReducedCoordinates);
  41         for (uint32_t ii=i; ii<i+nbCurvesRetrieved; ii++)
  42         {
  43             IDs[ii] = powerCurves[ii-i].ID;
  44             free(powerCurves[ii-i].values);
  45         }
  46         free(powerCurves);
  47     }
  48
  49     // *** Step 2 ***
  50     // Run PAM algorithm on the dissimilarity matrix computed from 'reducedCoordinates'.
  51
  52     Real* dissimilarities = get_dissimilarities_intra(
  53         reducedCoordinates, nbSeries, nbReducedCoordinates, work->p_for_dissims);
  54     free(reducedCoordinates);
  55
  56     uint32_t nbClusters = work->nbClusters;
  57     Result_t* result = (Result_t*) malloc(sizeof(Result_t));
  58     result->medoids_ID = (uint32_t*) malloc(nbClusters * sizeof(uint32_t));
  59     result->medoids_ranks = (uint32_t*) malloc(nbClusters * sizeof(uint32_t));
  60     result->nbClusters = nbClusters;
  61
  62     // Run PAM algorithm to fill result->medoids_ranks
  63     pam(dissimilarities, nbSeries, nbClusters, work->clustOnMedoids, NSTART, MAXITER, result);
  64     free(dissimilarities);
  65
  66     // Deduce medoids_IDs from indices
  67     for (uint32_t i = 0; i < nbClusters; i++)
  68         result->medoids_ID[i] = IDs[result->medoids_ranks[i]];
  69     free(IDs);
  70
  71     // return medoids IDs, indices and items labels (to be post-processed)
  72     return result;
  73 }
  74
  75 // code executed by slave process
  76 void slave_run(int myrank, uint32_t nbSeriesInChunk, uint32_t nbClusters)
  77 {
  78     Work_t* work;
  79     Result_t* result;
  80     MPI_Status status;
  81
  82     // Expected size of a Work message in bytes:
  83     uint32_t work_message_length = get_packedWork_length(nbSeriesInChunk);
  84     Byte packedWork[work_message_length];
  85
  86     // Expected size of a Result message in bytes: (uint32_t is on 4 bytes)
  87     uint32_t result_message_length = get_packedResult_length(nbClusters);
  88     Byte packedResult[result_message_length];
  89
  90     while (1)
  91     {
  92         // Receive a message from the master
  93         MPI_Recv(packedWork, work_message_length, MPI_BYTE, 0,
  94             MPI_ANY_TAG, MPI_COMM_WORLD, &status);
  95
  96         // Check the tag of the received message.
  97         if (status.MPI_TAG == DIETAG)
  98             return;
  99
 100         // Do the work
 101         work = unpack_work(packedWork, nbSeriesInChunk);
 102         fprintf(stdout, "%u / Slave pid=%u work on %s\n",myrank,getpid(),work->inputFileName);
 103         result = do_work(work);
 104         free_work(work);
 105
 106         // Send the result back
 107         pack_result(result, packedResult);
 108         free_result(result);
 109         MPI_Send(packedResult, result_message_length, MPI_BYTE, 0, WORKTAG, MPI_COMM_WORLD);
 110     }
 111 }