src/ck-cp/arrayRedistributor.h

   1 /**
   2
   3     A system for exposing application and runtime "control points"
   4     to the dynamic optimization framework.
   5
   6 */
   7 #ifndef __ARRAYREDISTRIBUTOR_H__
   8 #define __ARRAYREDISTRIBUTOR_H__
   9
  10 #include <vector>
  11 #include <list>
  12 #include <map>
  13 #include <cmath>
  14 //#include "ControlPoints.decl.h"
  15
  16 #include<pup_stl.h>
  17
  18
  19 #if CMK_WITH_CONTROLPOINT
  20
  21
  22 /**
  23  * \addtogroup ControlPointFramework
  24  *   @{
  25  */
  26
  27
  28 /// A message containing a chunk of a data array used when redistributing to a different set of active chares
  29 class redistributor2DMsg : public CMessage_redistributor2DMsg {
  30  public:
  31   int top;
  32   int left;
  33   int height;
  34   int width;
  35   int new_chare_cols;
  36   int  new_chare_rows;
  37   int which_array;
  38   double *data;
  39 };
  40
  41
  42
  43 /// Integer Maximum
  44 static int maxi(int a, int b){
  45   if(a>b)
  46     return a;
  47   else
  48     return b;
  49 }
  50
  51 /// Integer Minimum
  52 static int mini(int a, int b){
  53   if(a<b)
  54     return a;
  55   else
  56     return b;
  57 }
  58
  59
  60 /// A chare group that can redistribute user data arrays. It is used by binding it to a user's Chare Array
  61 class redistributor2D: public CBase_redistributor2D {
  62  public:
  63
  64   std::map<int,double*> data_arrays;
  65   std::map<int,int> data_arrays_sizes;
  66
  67   /// The array associated with this data redistribution
  68   CProxyElement_ArrayElement associatedArray;
  69
  70   int incoming_count;
  71   std::map<int,double*> data_arrays_incoming;
  72   std::map<int,int> data_arrays_incoming_sizes;
  73
  74   /// Is this array element active
  75   bool thisElemActive;
  76
  77   bool resizeGranulesHasBeenCalled;
  78
  79   CkVec<redistributor2DMsg *> bufferedMsgs;
  80
  81  private:
  82
  83
  84   void *fakeMemoryUsage;
  85
  86
  87   CkCallback dataRedistributedCallback;
  88
  89   int x_chares; // number of active chares in x dimension
  90   int y_chares; // number of active chares in y dimension
  91
  92   int data_width;  // The width of the global array, not the local piece
  93   int data_height; // The height of the global array, not the local piece
  94
  95   int data_x_ghost; // The padding in the x dimension on each side of the data
  96   int data_y_ghost; // The padding in the y dimension on each side of the data
  97
  98
  99  public:
 100
 101   void pup(PUP::er &p) {
 102     CBase_redistributor2D::pup(p);
 103
 104     p | data_arrays_sizes;
 105     p | data_arrays_incoming_sizes;
 106     p | incoming_count;
 107     p | associatedArray;
 108     p | thisElemActive;
 109
 110     p | dataRedistributedCallback;
 111
 112     p | resizeGranulesHasBeenCalled;
 113
 114     p | x_chares;
 115     p | y_chares;
 116     p | data_width;
 117     p | data_height;
 118     p | data_x_ghost;
 119     p | data_y_ghost;
 120
 121     if(p.isPacking() && fakeMemoryUsage!=NULL)
 122       free(fakeMemoryUsage);
 123
 124     fakeMemoryUsage = NULL;
 125
 126     ////////////////////////////////
 127     // when packing, iterate through data_arrays
 128     // when unpacking
 129
 130     {
 131       std::map<int,int>::iterator iter;
 132       for(iter = data_arrays_sizes.begin(); iter != data_arrays_sizes.end(); iter++){
 133         int whichArray = iter->first;
 134         int arraySize = iter->second;
 135
 136         //      CkPrintf("Pupping data array %d\n",whichArray);
 137         p | whichArray;
 138
 139         if(p.isUnpacking())
 140           data_arrays[whichArray] = new double[arraySize];
 141
 142         PUParray(p,data_arrays[whichArray] ,arraySize);
 143
 144         if(p.isPacking())
 145           delete[] data_arrays[whichArray];
 146
 147       }
 148     }
 149
 150
 151     ///////////////////////////////
 152     {
 153       std::map<int,int>::iterator iter;
 154       for(iter = data_arrays_incoming_sizes.begin(); iter != data_arrays_incoming_sizes.end(); iter++){
 155         int whichArray = iter->first;
 156         int arraySize = iter->second;
 157
 158         //      CkPrintf("Pupping incoming array %d\n",whichArray);
 159         p | whichArray;
 160
 161         if(p.isUnpacking() && data_arrays_incoming_sizes[whichArray] > 0)
 162           data_arrays_incoming[whichArray] = new double[arraySize];
 163
 164         PUParray(p,data_arrays_incoming[whichArray],arraySize);
 165
 166         if(p.isPacking())
 167           delete[] data_arrays_incoming[whichArray];
 168
 169       }
 170     }
 171
 172     //    CkPrintf("pup redistributor2D\n");
 173   }
 174
 175
 176   void ckJustMigrated(){
 177     //   CkPrintf("redistributor element %02d %02d migrated to %d", thisIndex.x, thisIndex.y, CkMyPe());
 178   }
 179
 180
 181   // ------------ Some routines for computing the array bounds for this chare  ------------
 182
 183   // The index in the global array for my top row
 184   int top_data_idx();
 185
 186   int bottom_data_idx();
 187
 188   int left_data_idx();
 189
 190   int right_data_idx();
 191
 192   int top_neighbor();
 193
 194   int bottom_neighbor();
 195
 196   int left_neighbor();
 197
 198   int right_neighbor();
 199
 200
 201   /// the width of the non-ghost part of the local partition
 202   int mywidth();
 203
 204
 205   // the height of the non-ghost part of the local partition
 206   int myheight();
 207
 208
 209
 210   // ------------ Some routines for computing the array bounds for arbitrary chares  ------------
 211
 212   int top_data_idx(int y, int y_total){
 213     return (data_height * y) / y_total;
 214   }
 215
 216   int bottom_data_idx(int y, int y_total){
 217     return ((data_height * (y+1)) / y_total) - 1;
 218   }
 219
 220   int left_data_idx(int x, int x_total){
 221     return (data_width * x) / x_total;
 222   }
 223
 224   int right_data_idx(int x, int x_total){
 225     return ((data_width * (x+1)) / x_total) - 1;
 226   }
 227
 228
 229   int top_data_idx(int y){
 230     return (data_height * y) / y_chares;
 231   }
 232
 233   int bottom_data_idx(int y){
 234     return ((data_height * (y+1)) / y_chares) - 1;
 235   }
 236
 237   int left_data_idx(int x){
 238     return (data_width * x) / x_chares;
 239   }
 240
 241   int right_data_idx(int x){
 242     return ((data_width * (x+1)) / x_chares) - 1;
 243   }
 244
 245   /// Return which chare array element(x index) owns the global data item i
 246   int who_owns_idx_x(int i){
 247     int w=0;
 248     while(1){
 249       if( i >= left_data_idx(w) && i <= right_data_idx(w) ){
 250         return w;
 251       }
 252       w++;
 253     }
 254   }
 255
 256   /// Return which chare array element(y index) owns the global data item i
 257   int who_owns_idx_y(int i){
 258     int w=0;
 259     while(1){
 260       if( i >= top_data_idx(w) && i <= bottom_data_idx(w) ){
 261         return w;
 262       }
 263       w++;
 264     }
 265   }
 266
 267
 268
 269
 270
 271   // Convert a local column,row id (0 to mywidth()-1, 0 to myheight()-1) to the index in the padded array
 272   int local_to_padded(int x, int y){
 273     CkAssert(thisElemActive);
 274     CkAssert(x < (mywidth()+data_x_ghost) && x >= (0-data_x_ghost) && y < (myheight()+data_y_ghost) && y >= (0-data_y_ghost) );
 275     return (mywidth()+2*data_x_ghost)*(y+data_y_ghost)+x+data_x_ghost;
 276   }
 277
 278   // get a data value
 279   double data_local(int which, int x, int y){
 280     CkAssert(local_to_padded(x,y) < data_arrays_sizes[which]);
 281     return data_arrays[which][local_to_padded(x,y)];
 282   }
 283
 284
 285   // Convert a local column id (0 to mywidth-1) to the global column id (0 to data_width-1)
 286   int local_to_global_x(int x){
 287     return left_data_idx() + x;
 288   }
 289
 290   // Convert a local row id (0 to myheight-1) to the global row id (0 to data_height-1)
 291   int local_to_global_y(int y){
 292     return top_data_idx() + y;
 293   }
 294
 295   int global_array_width(){
 296     return data_width;
 297   }
 298
 299   int global_array_height(){
 300     return data_height;
 301   }
 302
 303   int global_array_size(){
 304     return global_array_width() * global_array_height();
 305   }
 306
 307   int my_array_width(){
 308     return mywidth()+2*data_x_ghost;
 309   }
 310
 311   int my_array_height(){
 312     return myheight()+2*data_y_ghost;
 313   }
 314
 315   // Total size of arrays including ghost layers
 316   int my_array_size(){
 317     return my_array_width() * my_array_height();
 318   }
 319
 320   /// Create an array. If multiple arrays are needed, each should have its own index
 321   template <typename t> t* createDataArray(int which=0) {
 322     t* data = new t[my_array_size()];
 323     data_arrays[which] = data;
 324     data_arrays_sizes[which] = my_array_size();
 325
 326     if(thisIndex.x==0 && thisIndex.y==0)
 327       CkPrintf("data_arrays_sizes[which] set to %d\n", data_arrays_sizes[which] );
 328
 329
 330     CkAssert(data_arrays[which] != NULL);
 331 #if DEBUG > 2
 332     CkPrintf("Allocated array of size %d at %p\n", my_array_size(), data_arrays[which] );
 333 #endif
 334     return data;
 335   }
 336
 337   template <typename t> t* getDataArray(int which=0) {
 338     return data_arrays[which];
 339   }
 340
 341   /// Constructor takes in the dimensions of the array, including any desired ghost layers
 342   /// The local part of the arrays will have (mywidth+x_ghosts*2)*(myheight+y_ghosts*2) elements
 343   void setInitialDimensions(int width, int height, int x_chares_, int y_chares_, int x_ghosts=0, int y_ghosts=0){
 344     data_width = width;      // These values cannot change after this method is called.
 345     data_height = height;
 346     data_x_ghost = x_ghosts;
 347     data_y_ghost = y_ghosts;
 348
 349     setDimensions(x_chares_, y_chares_);
 350
 351   }
 352
 353
 354   void setDimensions( int x_chares_, int y_chares_){
 355     x_chares = x_chares_;
 356     y_chares = y_chares_;
 357
 358
 359     if( thisIndex.x < x_chares && thisIndex.y < y_chares ){
 360       thisElemActive = true;
 361     } else {
 362       thisElemActive = false;
 363     }
 364
 365   }
 366
 367
 368   redistributor2D(){
 369     incoming_count = 0;
 370     fakeMemoryUsage = NULL;
 371     CkAssert(bufferedMsgs.size() == 0);
 372   }
 373
 374
 375   redistributor2D(CkMigrateMessage*){
 376     CkAssert(bufferedMsgs.size() == 0);
 377   }
 378
 379
 380   void startup(){
 381 #if DEBUG > 3
 382    CkPrintf("redistributor 2D startup %03d,%03d\n", thisIndex.x, thisIndex.y);
 383 #endif
 384
 385     contribute();
 386   }
 387
 388
 389   void printArrays(){
 390 #if DEBUG > 2
 391     CkAssert(data_arrays.size()==2);
 392     for(std::map<int,double*>::iterator diter = data_arrays.begin(); diter != data_arrays.end(); diter++){
 393       int which_array = diter->first;
 394       double *data = diter->second;
 395       CkPrintf("%d,%d data_arrays[%d] = %p\n", thisIndex.x, thisIndex.y, which_array, data);
 396     }
 397 #endif
 398   }
 399
 400
 401   // Called on all elements involved with the new granularity or containing part of the old data
 402   void resizeGranules(int new_active_chare_cols, int new_active_chare_rows){
 403 #if DEBUG>1
 404     CkPrintf("Resize Granules called for elem %d,%d\n", thisIndex.x, thisIndex.y);
 405 #endif
 406
 407     resizeGranulesHasBeenCalled = true;
 408
 409     const bool previouslyActive = thisElemActive;
 410     const int old_top = top_data_idx();
 411     const int old_left = left_data_idx();
 412     const int old_bottom = top_data_idx()+myheight()-1;
 413     const int old_right = left_data_idx()+mywidth()-1;
 414     const int old_myheight = myheight();
 415     const int old_mywidth = mywidth();
 416
 417     setDimensions(new_active_chare_cols, new_active_chare_rows); // update dimensions & thisElemActive
 418
 419     const int new_mywidth = mywidth();
 420     const int new_myheight = myheight();
 421
 422     // Transpose Data
 423     // Assume only one new owner of my data
 424
 425     if(previouslyActive){
 426
 427       // Send all my data to any blocks that will need it
 428
 429       int newOwnerXmin = who_owns_idx_x(old_left);
 430       int newOwnerXmax = who_owns_idx_x(old_right);
 431       int newOwnerYmin = who_owns_idx_y(old_top);
 432       int newOwnerYmax = who_owns_idx_y(old_bottom);
 433
 434       for(int newx=newOwnerXmin; newx<=newOwnerXmax; newx++){
 435         for(int newy=newOwnerYmin; newy<=newOwnerYmax; newy++){
 436
 437           // Determine overlapping region between my data and this destination
 438 #if DEBUG > 2
 439           CkPrintf("newy(%d)*new_myheight(%d)=%d, old_top=%d\n",newy,new_myheight,newy*new_myheight,old_top);
 440 #endif
 441           // global range for overlapping area
 442           int global_top = maxi(top_data_idx(newy),old_top);
 443           int global_left = maxi(left_data_idx(newx),old_left);
 444           int global_bottom = mini(bottom_data_idx(newy),old_bottom);
 445           int global_right = mini(right_data_idx(newx),old_right);
 446           int w = global_right-global_left+1;
 447           int h = global_bottom-global_top+1;
 448
 449           CkAssert(w*h>0);
 450
 451           int x_offset = global_left - old_left;
 452           int y_offset = global_top - old_top;
 453
 454 #if DEBUG > 2
 455           CkPrintf("w=%d h=%d x_offset=%d y_offset=%d\n", w, h, x_offset, y_offset);
 456 #endif
 457
 458           std::map<int,double*>::iterator diter;
 459           for(diter =data_arrays.begin(); diter != data_arrays.end(); diter++){
 460
 461             redistributor2DMsg* msg = new(w*h) redistributor2DMsg;
 462             //      CkPrintf("Created message msg %p\n", msg);
 463
 464             int which_array = diter->first;
 465             double *t = diter->second;
 466             int s = data_arrays_sizes[which_array];
 467
 468             for(int j=0; j<h; j++){
 469               for(int i=0; i<w; i++){
 470                 CkAssert(j*w+i < w*h);
 471                 CkAssert((data_x_ghost*2+old_mywidth)*(j+y_offset+data_y_ghost)+(i+ x_offset+data_x_ghost) < s);
 472                 msg->data[j*w+i] = t[(data_x_ghost*2+old_mywidth)*(j+y_offset+data_y_ghost)+(i+ x_offset+data_x_ghost)];
 473               }
 474             }
 475
 476             msg->top = global_top;
 477             msg->left = global_left;
 478             msg->height = h;
 479             msg->width = w;
 480             msg->new_chare_cols = new_active_chare_cols;
 481             msg->new_chare_rows = new_active_chare_rows;
 482             msg->which_array = which_array;
 483
 484             //      CkPrintf("Sending message msg %p\n", msg);
 485             thisProxy(newx, newy).receiveTransposeData(msg);
 486
 487           }
 488
 489         }
 490
 491
 492       }
 493     }
 494
 495     if(!thisElemActive){
 496 #if DEBUG > 2
 497       CkPrintf("Element %d,%d is no longer active\n", thisIndex.x, thisIndex.y);
 498 #endif
 499
 500       // Free my arrays
 501       for(std::map<int,double*>::iterator diter = data_arrays.begin(); diter != data_arrays.end(); diter++){
 502         int which_array = diter->first;
 503         delete data_arrays[which_array];
 504         data_arrays[which_array] = NULL;
 505         data_arrays_sizes[which_array] = 0;
 506       }
 507       continueToNextStep();
 508
 509     }
 510
 511
 512     // Call receiveTransposeData for any buffered messages.
 513     int size = bufferedMsgs.size();
 514     for(int i=0;i<size;i++){
 515       redistributor2DMsg *msg = bufferedMsgs[i];
 516       //     CkPrintf("Delivering buffered receiveTransposeData(msg=%p) i=%d\n", msg, i);
 517       receiveTransposeData(msg); // this will delete the message
 518     }
 519     bufferedMsgs.removeAll();
 520
 521     int newPe = (thisIndex.y * new_active_chare_cols + thisIndex.x) % CkNumPes();
 522     if(newPe == CkMyPe()){
 523       //      CkPrintf("Keeping %02d , %02d on PE %d\n", thisIndex.x, thisIndex.y, newPe);
 524     }
 525     else{
 526       // CkPrintf("Migrating %02d , %02d to PE %d\n", thisIndex.x, thisIndex.y, newPe);
 527       migrateMe(newPe);
 528     }
 529     // CANNOT CALL ANYTHING AFTER MIGRATE ME
 530   }
 531
 532
 533   void continueToNextStep(){
 534 #if DEBUG > 2
 535     CkPrintf("Elem %d,%d is ready to continue\n", thisIndex.x, thisIndex.y);
 536 #endif
 537
 538     resizeGranulesHasBeenCalled = false;
 539
 540     for(std::map<int,double*>::iterator diter =data_arrays.begin(); diter != data_arrays.end(); diter++){
 541       int which_array = diter->first;
 542       double *data = diter->second;
 543       if( ! ((data==NULL && !thisElemActive) || (data!=NULL && thisElemActive) )){
 544         CkPrintf("[%d] ERROR: ! ((data==NULL && !thisElemActive) || (data!=NULL && thisElemActive) )",CkMyPe());
 545         CkPrintf("[%d] ERROR: data=%p thisElemActive=%d  (perhaps continueToNextStep was called too soon)\n",CkMyPe(), data, (int)thisElemActive );
 546
 547         CkAbort("ERROR");
 548       }
 549     }
 550
 551
 552 #if USE_EXTRAMEMORY
 553 #error NO USE_EXTRAMEMORY ALLOWED YET
 554     if(thisElemActive){
 555
 556       long totalArtificialMemory = controlPoint("Artificial Memory Usage", 100, 500);
 557       long artificialMemoryPerChare = totalArtificialMemory *1024*1024 / x_chares / y_chares;
 558
 559       CkPrintf("Allocating fake memory of %d MB (of the total %d MB) (xchares=%d y_chares=%d)\n", artificialMemoryPerChare/1024/1024, totalArtificialMemory, x_chares, y_chares);
 560       free(fakeMemoryUsage);
 561       fakeMemoryUsage = malloc(artificialMemoryPerChare);
 562       CkAssert(fakeMemoryUsage != NULL);
 563     } else {
 564       free(fakeMemoryUsage);
 565       fakeMemoryUsage = NULL;
 566     }
 567 #endif
 568
 569
 570
 571     incoming_count = 0; // prepare for future granularity change
 572     contribute();
 573   }
 574
 575
 576
 577
 578
 579
 580
 581   void receiveTransposeData(redistributor2DMsg *msg){
 582
 583     // buffer this message until resizeGranules Has Been Called
 584     if(!resizeGranulesHasBeenCalled){
 585       bufferedMsgs.push_back(msg);
 586       //      CkPrintf("Buffering receiveTransposeData(msg=%p)\n", msg);
 587       return;
 588     }
 589
 590     CkAssert(resizeGranulesHasBeenCalled);
 591
 592     int top_new = top_data_idx(thisIndex.y, msg->new_chare_rows);
 593     int bottom_new = bottom_data_idx(thisIndex.y, msg->new_chare_rows);
 594     int left_new = left_data_idx(thisIndex.x, msg->new_chare_cols);
 595     int right_new = right_data_idx(thisIndex.x, msg->new_chare_cols);
 596
 597     int new_height = bottom_new - top_new + 1;
 598     int new_width = right_new - left_new + 1;
 599
 600     if(incoming_count == 0){
 601       // Allocate new arrays
 602       std::map<int,double*>::iterator diter;
 603       for(diter =data_arrays.begin(); diter != data_arrays.end(); diter++){
 604         int w = diter->first;
 605         data_arrays_incoming[w] = new double[(new_width+2*data_x_ghost)*(new_height+2*data_y_ghost)];
 606         data_arrays_incoming_sizes[w] = (new_width+2*data_x_ghost)*(new_height+2*data_y_ghost);
 607
 608         //      CkPrintf("data_arrays_incoming_sizes[%d] set to %d\n", w, data_arrays_incoming_sizes[w] );
 609
 610       }
 611     }
 612
 613
 614     // Copy values from the incoming array to the appropriate place in data_arrays_incoming
 615     // Current top left of my new array
 616
 617
 618     double *localData = data_arrays_incoming[msg->which_array];
 619     int s = data_arrays_incoming_sizes[msg->which_array];
 620
 621     //    CkPrintf("%d,%d data_arrays_incoming.size() = %d\n", thisIndex.x, thisIndex.y, data_arrays_incoming.size() );
 622     //    CkPrintf("msg->which_array=%d   localData=%p   s=%d\n", msg->which_array, localData, s);
 623     CkAssert(localData != NULL);
 624
 625     for(int j=0; j<msg->height; j++){
 626       for(int i=0; i<msg->width; i++){
 627
 628         if( (msg->top+j >= top_new) && (msg->top+j <= bottom_new) && (msg->left+i >= left_new) && (msg->left+i <= right_new) ) {
 629           CkAssert(j*msg->width+i<msg->height*msg->width);
 630           CkAssert((msg->top+j-top_new)*new_width+(msg->left+i-left_new) < new_width*new_height);
 631           CkAssert((msg->top+j-top_new)*new_width+(msg->left+i-left_new) >= 0);
 632
 633           CkAssert((msg->top+j-top_new+data_y_ghost)*(new_width+2*data_x_ghost)+(msg->left+i-left_new+data_x_ghost) < s);
 634           localData[(msg->top+j-top_new+data_y_ghost)*(new_width+2*data_x_ghost)+(msg->left+i-left_new+data_x_ghost)] = msg->data[j*msg->width+i];
 635           incoming_count++;
 636
 637         }
 638
 639       }
 640     }
 641
 642     //    CkPrintf("Deleting message msg %p\n", msg);
 643     delete msg;
 644
 645
 646     if(incoming_count == new_height*new_width*data_arrays.size()){
 647
 648       std::map<int,double*>::iterator diter;
 649       for(diter =data_arrays.begin(); diter != data_arrays.end(); diter++){
 650         int w = diter->first;
 651         delete[] data_arrays[w];
 652         data_arrays[w] = data_arrays_incoming[w];
 653         data_arrays_sizes[w] = data_arrays_incoming_sizes[w];
 654         data_arrays_incoming[w] = NULL;
 655         data_arrays_incoming_sizes[w] = 0;
 656
 657         //        if(thisIndex.x==0 && thisIndex.y==0)
 658           //          CkPrintf("data_arrays_incoming_sizes[%d] set to %d\n",w, data_arrays_incoming_sizes[w] );
 659
 660           //        if(thisIndex.x==0 && thisIndex.y==0)
 661           //  CkPrintf("data_arrays_sizes[%d] set to %d\n",w, data_arrays_sizes[w] );
 662
 663       }
 664
 665       continueToNextStep();
 666     }
 667
 668   }
 669 };
 670
 671 /** @} */
 672 #endif
 673 #endif