Changeset 3296

Timestamp:

11/06/09 12:37:19 (2 weeks ago)

Author:

tautges

Message:

Couple of improvements and bug fixes to parallel ghost exchange code.

- add debug-only code to check for duplicate procs in proc list for entities
- add a little MPE stuff, commented out
- in build_shared_hps list, terminate list of procs/handles
- in a couple places, print error and return failure instead of asserting
- couple of improvements to print_buffer to make it more informative
- adding a pcomm_serial test, which does the equivalent of ghost exchange but

completely in serial

Passes make check, except for iMeshP tests.

Location:

MOAB/trunk/parallel

Files:

• MBParallelComm.cpp (modified) (29 diffs)
• Makefile.am (modified) (2 diffs)
• pcomm_serial.cpp (added)
• pcomm_unit.cpp (modified) (3 diffs)

MOAB/trunk/parallel/MBParallelComm.cpp

@@ -48 +48 @@
 
 #undef DEBUG_MPE
+//#define DEBUG_MPE 1
 #ifdef DEBUG_MPE
 #include "mpe.h"
@@ -54 +55 @@
 int SHAREDV_START, SHAREDV_END;
 int RESOLVE_START, RESOLVE_END;
+int ENTITIES_START, ENTITIES_END;
+int RHANDLES_START, RHANDLES_END;
+
 #endif
 #undef DEBUG_COMM
+//#define DEBUG_COMM 1
 #undef DEBUG_PACKING
 #undef DEBUG_MSGS
@@ -489 +494 @@
                         MPI_UNSIGNED_CHAR, to_proc, 
                         mesg_tag+1, procConfig.proc_comm(), &send_req2);
+      // put this inside so we can stop on completion in the debugger
     if (success != MPI_SUCCESS) return MB_FAILURE;
   }
@@ -868 +874 @@
       PACK_INTS(buff_ptr, tmp_procs, num_ents);
       PACK_EH(buff_ptr, tmp_handles, num_ents);
+
+#ifndef NDEBUG
+        // check for duplicates in proc list
+      unsigned int dp = 0;
+      for (; dp < MAX_SHARING_PROCS && -1 != tmp_procs[dp]; dp++)
+        dumprocs.insert(tmp_procs[dp]);
+      assert(dumprocs.size() == dp);
+      dumprocs.clear();
+#endif      
     }
   }
@@ -1016 +1031 @@
   }
 
+    // put -1 after procs and 0 after handles
+  if (MAX_SHARING_PROCS > num_ents) {
+    tmp_procs[num_ents] = -1;
+    tmp_handles[num_ents] = 0;
+  }
+  
   return MB_SUCCESS;
 }
@@ -1312 +1333 @@
     for (i = 0; i < num_ents; i++) {
       UNPACK_INT(buff_ptr, j);
-      assert(j >= 0 && "Should be non-negative # proc/handles.");
+      if (j < 0) {
+        std::cout << "Should be non-negative # proc/handles.";
+        return MB_FAILURE;
+      }
+      
       buff_ptr += j * (sizeof(int)+sizeof(MBEntityHandle));
     }
@@ -1355 +1380 @@
           // pointers to other procs/handles
         UNPACK_INT(buff_save, num_ps);
-        assert("Shouldn't ever be fewer than 1 procs here." && 0 < num_ps);
+        if (0 >= num_ps) {
+          std::cout << "Shouldn't ever be fewer than 1 procs here." << std::endl;
+          return MB_FAILURE;
+        }
+        
         UNPACK_INTS(buff_save, &ps[0], num_ps);
         UNPACK_EH(buff_save, &hs[0], num_ps);
@@ -1441 +1470 @@
                                      (created_here ? (PSTATUS_GHOST | PSTATUS_NOT_OWNED) : 0)));
         RRA("");
-      
+
           // need to send this new handle to all sharing procs
         if (!is_iface) {
@@ -1507 +1536 @@
 }
 
-MBErrorCode MBParallelComm::print_buffer(unsigned char *buff_ptr, int mesg_tag, 
+MBErrorCode MBParallelComm::print_buffer(unsigned char *buff_ptr, 
+                                         int mesg_tag, 
                                          int from_proc, bool sent) 
 {
@@ -1516 +1546 @@
             << " to/from proc " << from_proc << "; contents:" << std::endl;
 
+  int msg_length;
+  unsigned char *orig_ptr = buff_ptr;
+  UNPACK_INT(buff_ptr, msg_length);
+  std::cout << msg_length << " bytes..." << std::endl;
+
   if (MB_MESG_ENTS == mesg_tag) {
-    int total_size;
-    UNPACK_INT(buff_ptr, total_size);
-    std::cout << total_size << " entities..." << std::endl;
 
       // 1. # entities = E
@@ -1533 +1565 @@
     for (i = 0; i < num_ents; i++) {
       UNPACK_INT(buff_ptr, j);
+      if (0 > j) return MB_FAILURE;
       ps.resize(j);
       hs.resize(j);
-      std::cout << "Entity " << i << ": # procs = " << j << std::endl;
+      std::cout << "Entity " << i << ", # procs = " << j << std::endl;
       UNPACK_INTS(buff_ptr, &ps[0], j);
       UNPACK_EH(buff_ptr, &hs[0], j);
@@ -1544 +1577 @@
       for (k = 0; k < j; k++) std::cout << hs[k] << " ";
       std::cout << std::endl;
-    }
-  
-  while (true) {
-    MBEntityType this_type = MBMAXTYPE;
-    UNPACK_TYPE(buff_ptr, this_type);
-    assert(this_type != MBENTITYSET);
+
+      if (buff_ptr-orig_ptr > msg_length) {
+        std::cout << "End of buffer..." << std::endl;
+        return MB_FAILURE;
+      }
+    }
+  
+    while (true) {
+      MBEntityType this_type = MBMAXTYPE;
+      UNPACK_TYPE(buff_ptr, this_type);
+      assert(this_type != MBENTITYSET);
 
         // MBMAXTYPE signifies end of entities data
@@ -1588 +1626 @@
           std::cout << std::endl;
         }
+
+        if (buff_ptr-orig_ptr > msg_length) {
+          std::cout << "End of buffer..." << std::endl;
+          return MB_FAILURE;
+        }
       }
     }
@@ -1593 +1636 @@
   
   else if (MB_MESG_REMOTE_HANDLES) {
+    int num_bytes;
+    UNPACK_INT(buff_ptr, num_bytes);
+    std::cout << num_bytes << " bytes..." << std::endl;
     int num_ents;
     UNPACK_INT(buff_ptr, num_ents);
@@ -1607 +1653 @@
                 << L1p[i] << std::endl;
     }
+
+    if (buff_ptr-orig_ptr > msg_length) {
+      std::cout << "End of buffer..." << std::endl;
+      return MB_FAILURE;
+    }
+
   }
   else if (MB_MESG_TAGS) {
@@ -1706 +1758 @@
   RRA("");
   
+#ifndef NDEBUG
+  {
+      // check for duplicates in proc list
+    std::set<unsigned int> dumprocs;
+    unsigned int dp = 0;
+    for (; (int) dp < num_ps && -1 != ps[dp]; dp++)
+      dumprocs.insert(ps[dp]);
+    assert(dp == dumprocs.size());
+  }
+#endif      
+
     // add any new sharing data
   bool changed = false;
@@ -1812 +1875 @@
     RRA("Couldn't set sharedhs tag.");
     pstat |= (PSTATUS_MULTISHARED | PSTATUS_SHARED);
+
+#ifndef NDEBUG
+    {
+        // check for duplicates in proc list
+      std::set<unsigned int> dumprocs;
+      unsigned int dp = 0;
+      for (; dp < num_exist && -1 != tag_ps[dp]; dp++)
+        dumprocs.insert(tag_ps[dp]);
+      assert(dp == dumprocs.size());
+    }
+#endif      
   }
   else if (num_exist == 2 || num_exist == 1) {
     if (tag_ps[0] == (int) procConfig.proc_rank()) {
-      assert(2 == num_exist);
+      assert(2 == num_exist && tag_ps[1] != (int) procConfig.proc_rank());
       tag_ps[0] = tag_ps[1];
       tag_hs[0] = tag_hs[1];
@@ -2931 +3005 @@
   MPE_Log_get_state_eventIDs( &SHAREDV_START, &SHAREDV_END);
   MPE_Log_get_state_eventIDs( &RESOLVE_START, &RESOLVE_END);
+  MPE_Log_get_state_eventIDs( &ENTITIES_START, &ENTITIES_END);
+  MPE_Log_get_state_eventIDs( &RHANDLES_START, &RHANDLES_END);
   success = MPE_Describe_state(IFACE_START, IFACE_END, "Resolve interface ents", "green");
   success = MPE_Describe_state(GHOST_START, GHOST_END, "Exchange ghost ents", "red");
   success = MPE_Describe_state(SHAREDV_START, SHAREDV_END, "Resolve interface vertices", "blue");
   success = MPE_Describe_state(RESOLVE_START, RESOLVE_END, "Resolve shared ents", "purple");
+  success = MPE_Describe_state(ENTITIES_START, ENTITIES_END, "Exchange shared ents", "yellow");
+  success = MPE_Describe_state(RHANDLES_START, RHANDLES_END, "Remote handles", "cyan");
 #endif
 }
@@ -3709 +3787 @@
     // post ghost irecv's for ghost entities from all communicating procs
     //===========================================
+#ifdef DEBUG_MPE
+    MPE_Log_event(ENTITIES_START, procConfig.proc_rank(), "Starting entity exchange.");
+#endif
     // index reqs the same as buffer/sharing procs indices
   std::vector<MPI_Request> recv_reqs(buffProcs.size(), MPI_REQUEST_NULL);
@@ -3764 +3845 @@
                          sendReqs[ind], sendReqs[ind+buffProcs.size()]);
     RRA("Failed to Isend in ghost exchange.");
+    
+//    if (1 == num_layers) 
+//      print_buffer(&ownerSBuffs[ind][0], MB_MESG_ENTS, *proc_it, true);
   }
 
@@ -3795 +3879 @@
       
       std::cerr << "Received from " << status[0].MPI_SOURCE
-                << ": count = " << this_count << ", tag = " << status[0].MPI_TAG;
+                << ", count = " << this_count << ", tag = " << status[0].MPI_TAG;
       if (MB_MESG_ENTS+1 == status[0].MPI_TAG) std::cerr << " (second)";
       std::cerr << std::endl;
@@ -3811 +3895 @@
     
     if (done) {
+#ifdef DEBUG_MSGS
+      print_buffer(&ghostRBuffs[ind][0], MB_MESG_ENTS, buffProcs[ind], false);
+#endif  
       unsigned char *buff_ptr = &ghostRBuffs[ind][sizeof(int)];
-#ifdef DEBUG_MSGS
-      print_buffer(buff_ptr-sizeof(int), MB_MESG_ENTS, buffProcs[ind], false);
-#endif  
       result = unpack_entities(buff_ptr,
                                store_remote_handles, ind, is_iface,
                                L1hloc, L1hrem, L1p, L2hloc, L2hrem, L2p, new_ents);
-      RRA("Failed to unpack entities.");
+      if (MB_SUCCESS != result) {
+        std::cout << "Failed to unpack entities.  Buffer contents:" << std::endl;
+        print_buffer(&ghostRBuffs[ind][0], MB_MESG_ENTS, buffProcs[ind], false);
+        return result;
+      }
 
       if (recv_reqs.size() != buffProcs.size()) {
@@ -3835 +3923 @@
   }
     
+#ifdef DEBUG_MPE
+    MPE_Log_event(ENTITIES_END, procConfig.proc_rank(), "Ending entity exchange.");
+#endif
+
   if (is_iface) {
       // need to check over entities I sent and make sure I received 
@@ -3880 +3972 @@
     // post recvs for remote handles of my sent ents
     //===========================================
+#ifdef DEBUG_MPE
+    MPE_Log_event(RHANDLES_START, procConfig.proc_rank(), "Starting remote handles.");
+#endif
   for (ind = 0, proc_it = buffProcs.begin(); 
        proc_it != buffProcs.end(); proc_it++, ind++) {
@@ -3942 +4037 @@
       
       std::cerr << "Received from " << status[0].MPI_SOURCE
-                << ": count = " << this_count << ", tag = " << status[0].MPI_TAG;
+                << ", count = " << this_count << ", tag = " << status[0].MPI_TAG;
       if (MB_MESG_REMOTE_HANDLES_SECOND == status[0].MPI_TAG) 
         std::cerr << " (second)";
@@ -3955 +4050 @@
     if (done) {
         // incoming remote handles
+#ifdef DEBUG_MSGS
+      print_buffer(&ghostRBuffs[ind][0], MB_MESG_REMOTE_HANDLES, buffProcs[ind], false);
+#endif  
       buff_ptr = &ghostRBuffs[ind][sizeof(int)];
-#ifdef DEBUG_MSGS
-      print_buffer(buff_ptr, MB_MESG_REMOTE_HANDLES, buffProcs[ind], false);
-#endif  
       result = unpack_remote_handles(buffProcs[ind], buff_ptr,
                                      L2hloc, L2hrem, L2p);
@@ -3969 +4064 @@
   }
     
+#ifdef DEBUG_MPE
+    MPE_Log_event(RHANDLES_END, procConfig.proc_rank(), "Ending remote handles.");
+#endif
 #ifdef DEBUG_MPE
       MPE_Log_event(GHOST_END, procConfig.proc_rank(), 
@@ -4418 +4516 @@
       else hpair[0] = 0;
     }
-    assert(hpair[0] && hpair[1]);
+    if (!(hpair[0] && hpair[1])) return MB_FAILURE;
     int this_proc = from_proc;
     result = update_remote_data(hpair[0], &this_proc, hpair+1, 1, 0);
@@ -5710 +5808 @@
   
   if (!bad_ents.empty()) {
-    std::cout << "Found bad entities in check_local_shared:" << std::endl;
+    std::cout << "Found bad entities in check_local_shared, proc rank "
+              << procConfig.proc_rank() << "," << std::endl;
     std::vector<std::string>::iterator vit;
     for (rit = bad_ents.begin(), vit = errors.begin(); rit != bad_ents.end(); rit++, vit++) {

MOAB/trunk/parallel/Makefile.am

@@ -47 +47 @@
      crystal.h errmem.h types.h
 
-  MOAB_PARALLEL_TEST += pcomm_unit parallel_unit_tests uber_parallel_test scdtest
+  MOAB_PARALLEL_TEST += pcomm_unit parallel_unit_tests uber_parallel_test scdtest pcomm_serial
 
 if PARALLEL_HDF5
@@ -89 +89 @@
 uber_parallel_test_SOURCES = uber_parallel_test.cpp
 uber_parallel_test_LDADD = ../libMOAB.la
+pcomm_serial_SOURCES = pcomm_serial.cpp
+pcomm_serial_LDADD = ../libMOAB.la
 
 scdtest_SOURCES = scdtest.cpp

MOAB/trunk/parallel/pcomm_unit.cpp

@@ -49 +49 @@
 /** Test pack/unpack of shared entities in 3d*/
 void test_pack_shared_entities_3d();
-/** Test pack/unpack of arbitrary mesh file */
-void test_pack_shared_arbitrary();
 /** Test filter_pstatus function*/
 void test_filter_pstatus();
@@ -77 +75 @@
   num_err += RUN_TEST( test_pack_shared_entities_2d );
   num_err += RUN_TEST( test_pack_shared_entities_3d );
-  num_err += RUN_TEST( test_pack_shared_arbitrary );
   num_err += RUN_TEST( test_filter_pstatus );
   
@@ -1863 +1860 @@
 
   for (unsigned int i = 0; i < 4; i++)
-    delete pc[i];
-}
-
-void test_pack_shared_arbitrary()
-{
-#define NP 3
-  MBCore moab[NP];
-  MBParallelComm *pc[NP];
-  for (unsigned int i = 0; i < NP; i++) {
-    pc[i] = new MBParallelComm(&moab[i]);
-    pc[i]->set_rank(i);
-    pc[i]->set_size(NP);
-  }
-
-  std::vector<int> pa_vec;
-  pa_vec.push_back(ReadParallel::PA_READ);
-  pa_vec.push_back(ReadParallel::PA_GET_FILESET_ENTS);
-  pa_vec.push_back(ReadParallel::PA_DELETE_NONLOCAL);
-  MBErrorCode rval;
-  std::vector<int> partition_tag_vals;
-  bool partition_distrib = false;
-
-#ifdef SRCDIR
-  const char *fnames[] = {STRINGIFY(SRCDIR) "/ptest.cub"};
-#else
-  const char *fnames[] = {"./ptest.cub"};
-#endif
-  
-  std::string ptag_name("GEOM_DIMENSION");
-  partition_tag_vals.push_back(3);
-  partition_distrib = true;
-  
-    //std::string ptag_name("MATERIAL_SET");
-    //partition_distrib = true;
-  
-  FileOptions fopts(NULL);
-  
-  for (unsigned int i = 0; i < NP; i++) {
-    ReadParallel rp(moab+i, pc[i]);
-    MBEntityHandle tmp_set = 0;
-    rval = rp.load_file(fnames, 1, tmp_set, ReadParallel::POPT_READ_DELETE,
-                        ptag_name, 
-                        partition_tag_vals, partition_distrib, false, pa_vec, 
-                        fopts, NULL, 0, NULL, i, false, -1, -1, -1, -1, 0);
-    CHECK_ERR(rval);
-  }
-  
-  rval = MBParallelComm::resolve_shared_ents(pc, NP, 3);
-  CHECK_ERR(rval);
-
-    // exchange interface cells
-  rval = MBParallelComm::exchange_ghost_cells(pc, NP, -1, -1, 0, true);
-  CHECK_ERR(rval);
-  
-    // now 1 layer of hex ghosts
-  rval = MBParallelComm::exchange_ghost_cells(pc, NP, 3, 0, 1, true);
-  CHECK_ERR(rval);
-
-  for (unsigned int i = 0; i < NP; i++)
     delete pc[i];
 }