Problem with collective communication in parallel threads.

[hujie-frank]

Excuse me, does MPI can call collective communication concurrently in multi-threads? I want to do collective communication like MPI_Allreduce in parallel threads to deal with different tasks for some optimizations. I initialize with MPI_THREAD_MULTIPLE in MPI_Init_thread(), but it does not work ("MPI_ERR_TRUNCATE: message truncated"). I found the reason in the website, in Open MPI, each collective communication uses internally the same message tag values and hence different communication can interfere with each other in not issued in sequence. But I also try to use asynchronous communication interface MPI_Iallreduce which obey MPI-3.0 standard to tackle this problem, there are still some errors occurred("MPI Error in MPI_Testall() (18)"). Moreover, if communicate with same size message all the time, MPI_Allreduce and MPI_Iallreduce will operate without checking correctness of communication results.

[bosilca]

From the MPI perspective it is incorrect to start multiple collective communications on the same communicator in same time (for more info please read the Section 5.13 from the MPI standard, especially the page 217).

If you launch the collectives on different communicators, I can't see how the OMPI collective code today can lead to the type of errors you mention. Can you post a reproducer ?

[ggouaillardet]

@hujiePKU as pointed by George, you cannot use the same communicator in collectives with more one threads at a time.
The issue might be related to ddt (e.g. many elements on one hand, one large ddt on the other hand.
So you might wanna try

mpirun --mca coll ^tuned ...

And see if it helps

[hujie-frank]

I read the Section 5.13 and more Sections about multi-threads, communicators and MPI environment from the MPI standard carefully. I modified my code to launch a communicator duplicated from MPI_COMM_WORLD for each thread and call non-blocking collective communication and MPI_Wait subsequently in each thread. The program can run without any error or warning, except deadlocking at MPI_Wait sometimes. I use the different communicators all come from MPI_COMM_WORLD, is it responsible for the deadlock? What is the thread-safe approach to concurrently call collective communication in multi-threads?

[ggouaillardet]

Can you post a trimmed version of your code that exhibits the hang ?
Btw, you configure'd with --enable-mpi-thread-multiple , right ?
Which version are you running ?

[hujie-frank]

Yes, I configure with --enable-mpi-thread-multiple and call MPI_Init_thread(). The MPI version is 2.0.1. The attachment is the trimmed code.
test_mpi_multithread.cpp.txt

[ggouaillardet]

@hujiePKU your program invokes MPI_Comm_dup on MPI_COMM_WORLD in both threads and at the same time.
i do not know whether this is allowed by the MPI standard. @bosilca any thoughts about that ?

in Open MPI, i believe MPI_Comm_dup involves collective operations, so i do not think we support that.

here is a rewritten version of your program that works fine for me
(i fixed the MPI_Comm_dup thing and remove CUDA related stuff, since i do not believe this is relevant here, plus i have no NVIDIA hardware to test with)

#include <vector>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <assert.h>
#include <mpi.h>
#include <iostream>

#define ENV_LOCAL_RANK "OMPI_COMM_WORLD_LOCAL_RANK"

void do_allreduce(std::vector<float*> &data_cpu_vec, std::vector<int> count_vec, MPI_Comm comm) {
  MPI_Request req;
  for (int i = 0; i < count_vec.size(); ++i) {
    // copy device --> host

    // call Iallreduce
    MPI_Iallreduce(MPI_IN_PLACE, (void*)data_cpu_vec[i], count_vec[i], MPI_FLOAT, MPI_SUM, comm, &req);
    MPI_Wait(&req, MPI_STATUS_IGNORE);

    // copy host --> device
  }
} 

int main(int argc, char** argv) {
// Initialize MPI

  int provided_thread_support;
  MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided_thread_support);
  assert(MPI_THREAD_MULTIPLE == provided_thread_support);

  int rank, size;
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  MPI_Comm_size(MPI_COMM_WORLD, &size);

  MPI_Comm comm1, comm2;
  MPI_Comm_dup(MPI_COMM_WORLD, &comm1); 
  MPI_Comm_dup(MPI_COMM_WORLD, &comm2); 

  const int data_count(atoi(argv[1]));
  // for thread1
  std::vector<float*> data_cpu_vec1(data_count, NULL);
  std::vector<int> count_vec1(data_count, 0);

  // for thread2
  std::vector<float*> data_cpu_vec2(data_count, NULL);
  std::vector<int> count_vec2(data_count, 0);

  for (int i = 0; i < data_count; ++i) {
    count_vec1[i] = int(rand() % 10000) + 100;
    count_vec2[i] = int(rand() % 10000) + 100;
  }

  // Brodcast the random count to all processor.
  MPI_Bcast(count_vec1.data(), data_count, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(count_vec2.data(), data_count, MPI_INT, 0, MPI_COMM_WORLD);

  // Allocate host and device memory
  for (int i = 0; i < data_count; ++i) {
    // malloc for thread1
    data_cpu_vec1[i] = (float *)malloc(sizeof(float) * count_vec1[i]);

    // malloc for thread2
    data_cpu_vec2[i] = (float *)malloc(sizeof(float) * count_vec2[i]);
  }

  MPI_Barrier(MPI_COMM_WORLD);
  double t1, t2;
  t1 = MPI_Wtime();

  // start two threads
  boost::shared_ptr<boost::thread> thread1(new boost::thread(boost::bind(&do_allreduce, data_cpu_vec1, count_vec1, comm1)));
  boost::shared_ptr<boost::thread> thread2(new boost::thread(boost::bind(&do_allreduce, data_cpu_vec2, count_vec2, comm2)));

  thread1->join();
  thread2->join();

  t2 = MPI_Wtime();
  std::cout << "communication time " << t2-t1 << " second.";

  // release memory and handle

  MPI_Comm_free(&comm1);
  MPI_Comm_free(&comm2);

  for (int i = 0; i < data_count; ++i) {
    free(data_cpu_vec1[i]); 
    free(data_cpu_vec2[i]); 
  }

  MPI_Finalize();
}

can you please try it and let me know how things go ?

[hujie-frank]

@ggouaillardet I complied and ran your code without any modification. I executed the script "mpirun -np 2 ./test 1000" more than 10 times, but procedure still occurred deadlock frequently. If I run with smaller data_count as input by script "mpirun -np 2 ./test 100", the probability of deadlock will decrease.

[ggouaillardet]

i got confirmation today that MPI_Comm_dup and friends are considered as collective operations, and hence MPI_Comm_dup(MPI_COMM_WORLD, ...) cannot be invoked simultaneously from different threads.

i found an issue related to thread safety in coll/libnbc

can you please manually apply the inline patch and sees if it works for you ?

diff --git a/ompi/mca/coll/libnbc/coll_libnbc.h b/ompi/mca/coll/libnbc/coll_libnbc.h
index 724b86d..b66ce60 100644
--- a/ompi/mca/coll/libnbc/coll_libnbc.h
+++ b/ompi/mca/coll/libnbc/coll_libnbc.h
@@ -13,7 +13,7 @@
  * Copyright (c) 2008      Cisco Systems, Inc.  All rights reserved.
  * Copyright (c) 2013-2015 Los Alamos National Security, LLC. All rights
  *                         reserved.
- * Copyright (c) 2014-2015 Research Organization for Information Science
+ * Copyright (c) 2014-2016 Research Organization for Information Science
  *                         and Technology (RIST). All rights reserved.
  * Copyright (c) 2016      IBM Corporation.  All rights reserved.
  * $COPYRIGHT$
@@ -76,6 +76,7 @@ struct ompi_coll_libnbc_component_t {
     opal_list_t active_requests;
     int32_t active_comms;
     opal_atomic_lock_t progress_lock;
+    opal_mutex_t lock;
 };
 typedef struct ompi_coll_libnbc_component_t ompi_coll_libnbc_component_t;

diff --git a/ompi/mca/coll/libnbc/coll_libnbc_component.c b/ompi/mca/coll/libnbc/coll_libnbc_component.c
index 8670181..fe8bdcc 100644
--- a/ompi/mca/coll/libnbc/coll_libnbc_component.c
+++ b/ompi/mca/coll/libnbc/coll_libnbc_component.c
@@ -91,6 +91,7 @@ libnbc_open(void)

     OBJ_CONSTRUCT(&mca_coll_libnbc_component.requests, opal_free_list_t);
     OBJ_CONSTRUCT(&mca_coll_libnbc_component.active_requests, opal_list_t);
+    OBJ_CONSTRUCT(&mca_coll_libnbc_component.lock, opal_mutex_t);
     ret = opal_free_list_init (&mca_coll_libnbc_component.requests,
                                sizeof(ompi_coll_libnbc_request_t), 8,
                                OBJ_CLASS(ompi_coll_libnbc_request_t),
@@ -115,6 +116,7 @@ libnbc_close(void)

     OBJ_DESTRUCT(&mca_coll_libnbc_component.requests);
     OBJ_DESTRUCT(&mca_coll_libnbc_component.active_requests);
+    OBJ_DESTRUCT(&mca_coll_libnbc_component.lock);

     return OMPI_SUCCESS;
 }
@@ -263,13 +265,17 @@ ompi_coll_libnbc_progress(void)

     if (opal_atomic_trylock(&mca_coll_libnbc_component.progress_lock)) return 0;

+    OPAL_THREAD_LOCK(&mca_coll_libnbc_component.lock);
     OPAL_LIST_FOREACH_SAFE(request, next, &mca_coll_libnbc_component.active_requests,
                            ompi_coll_libnbc_request_t) {
+        OPAL_THREAD_UNLOCK(&mca_coll_libnbc_component.lock);
         res = NBC_Progress(request);
         if( NBC_CONTINUE != res ) {
             /* done, remove and complete */
+            OPAL_THREAD_LOCK(&mca_coll_libnbc_component.lock);
             opal_list_remove_item(&mca_coll_libnbc_component.active_requests,
                                   &request->super.super.super);
+            OPAL_THREAD_UNLOCK(&mca_coll_libnbc_component.lock);

             if( OMPI_SUCCESS == res || NBC_OK == res || NBC_SUCCESS == res ) {
                 request->super.req_status.MPI_ERROR = OMPI_SUCCESS;
@@ -281,7 +287,9 @@ ompi_coll_libnbc_progress(void)
             ompi_request_complete(&request->super, true);
             OPAL_THREAD_UNLOCK(&ompi_request_lock);
         }
+        OPAL_THREAD_LOCK(&mca_coll_libnbc_component.lock);
     }
+    OPAL_THREAD_UNLOCK(&mca_coll_libnbc_component.lock);

     opal_atomic_unlock(&mca_coll_libnbc_component.progress_lock);

diff --git a/ompi/mca/coll/libnbc/nbc.c b/ompi/mca/coll/libnbc/nbc.c
index 430b506..7949fe1 100644
--- a/ompi/mca/coll/libnbc/nbc.c
+++ b/ompi/mca/coll/libnbc/nbc.c
@@ -702,7 +702,9 @@ int NBC_Start(NBC_Handle *handle, NBC_Schedule *schedule) {
   if (OPAL_UNLIKELY(OMPI_SUCCESS != res)) {
     return res;
   }
+  OPAL_THREAD_LOCK(&mca_coll_libnbc_component.lock);
   opal_list_append(&mca_coll_libnbc_component.active_requests, &(handle->super.super.super));
+  OPAL_THREAD_UNLOCK(&mca_coll_libnbc_component.lock);

   return OMPI_SUCCESS;
 }

if you are still facing issues, could you please give a try to the modified test program below
(i replaced non blocking MPI_Iallreduce(); MPI_Wait() with the blocking MPI_Allreduce())

#include <vector>
#include <boost/shared_ptr.hpp>
#include <boost/thread.hpp>
#include <assert.h>
#include <mpi.h>
#include <iostream>

#define ENV_LOCAL_RANK "OMPI_COMM_WORLD_LOCAL_RANK"

void do_allreduce(std::vector<float*> &data_cpu_vec, std::vector<int> count_vec, MPI_Comm comm) {
  MPI_Request req;
  for (int i = 0; i < count_vec.size(); ++i) {
    // copy device --> host

    // call Iallreduce
    MPI_Allreduce(MPI_IN_PLACE, (void*)data_cpu_vec[i], count_vec[i], MPI_FLOAT, MPI_SUM, comm);

    // copy host --> device
  }
} 

int main(int argc, char** argv) {
// Initialize MPI

  int provided_thread_support;
  MPI_Init_thread(&argc, &argv, MPI_THREAD_MULTIPLE, &provided_thread_support);
  assert(MPI_THREAD_MULTIPLE == provided_thread_support);

  int rank, size;
  MPI_Comm_rank(MPI_COMM_WORLD, &rank);
  MPI_Comm_size(MPI_COMM_WORLD, &size);

  MPI_Comm comm1, comm2;
  MPI_Comm_dup(MPI_COMM_WORLD, &comm1); 
  MPI_Comm_dup(MPI_COMM_WORLD, &comm2); 

  const int data_count(atoi(argv[1]));
  // for thread1
  std::vector<float*> data_cpu_vec1(data_count, NULL);
  std::vector<int> count_vec1(data_count, 0);

  // for thread2
  std::vector<float*> data_cpu_vec2(data_count, NULL);
  std::vector<int> count_vec2(data_count, 0);

  for (int i = 0; i < data_count; ++i) {
    count_vec1[i] = int(rand() % 10000) + 100;
    count_vec2[i] = int(rand() % 10000) + 100;
  }

  // Brodcast the random count to all processor.
  MPI_Bcast(count_vec1.data(), data_count, MPI_INT, 0, MPI_COMM_WORLD);
  MPI_Bcast(count_vec2.data(), data_count, MPI_INT, 0, MPI_COMM_WORLD);

  // Allocate host and device memory
  for (int i = 0; i < data_count; ++i) {
    // malloc for thread1
    data_cpu_vec1[i] = (float *)malloc(sizeof(float) * count_vec1[i]);

    // malloc for thread2
    data_cpu_vec2[i] = (float *)malloc(sizeof(float) * count_vec2[i]);
  }

  MPI_Barrier(MPI_COMM_WORLD);
  double t1, t2;
  t1 = MPI_Wtime();

  // start two threads
  boost::shared_ptr<boost::thread> thread1(new boost::thread(boost::bind(&do_allreduce, data_cpu_vec1, count_vec1, comm1)));
  boost::shared_ptr<boost::thread> thread2(new boost::thread(boost::bind(&do_allreduce, data_cpu_vec2, count_vec2, comm2)));

  thread1->join();
  thread2->join();

  t2 = MPI_Wtime();
  std::cout << "communication time " << t2-t1 << " second.";

  // release memory and handle

  MPI_Comm_free(&comm1);
  MPI_Comm_free(&comm2);

  for (int i = 0; i < data_count; ++i) {
    free(data_cpu_vec1[i]); 
    free(data_cpu_vec2[i]); 
  }

  MPI_Finalize();
}

[hujie-frank]

I replaced non blocking MPI_Iallreduce(), MPI_Wait() with the blocking MPI_Allreduce(), then the procedure work normally without any deadlock. But when I test the speedup ratio and parallel efficiency, I am surprised to find that calling MPI_Allreduce serially is faster than spliting the task into two concurrent threads to call MPI_Allreduce individually. The time consumption of serial call is just half of concurrent call whatever how much data_count is given. What happened? Part of code changes to test time consumption is in the following.

  void do_allreduce(std::vector<float*> &data_cpu_vec, std::vector<int> count_vec, MPI_Comm comm) {
    for (int i = 0; i < count_vec.size(); ++i) {
      MPI_Allreduce(MPI_IN_PLACE, (void*)data_cpu_vec[i], count_vec[i], MPI_FLOAT, MPI_SUM, comm);
    }
  }

int main() {
... 
  // --------------------------------- concurrent call --------------------------------------
  double t1, t2;
  MPI_Barrier(MPI_COMM_WORLD);
  t1 = MPI_Wtime();

  boost::shared_ptr<boost::thread> thread1(new boost::thread(boost::bind(&do_allreduce, data_cpu_vec1, count_vec1, comm1)));
  boost::shared_ptr<boost::thread> thread2(new boost::thread(boost::bind(&do_allreduce, data_cpu_vec2, count_vec2, comm2)));

  thread1->join();
  thread2->join();

  t2 = MPI_Wtime();
  std::cout << "strategy1 communication time: " << t2-t1 << " second." << std::endl;
// ------------------------------------  serial call ------------------------------------------------
  MPI_Barrier(MPI_COMM_WORLD);
  t1 = MPI_Wtime();
  do_allreduce(data_cpu_vec1, count_vec1, comm1);
  do_allreduce(data_cpu_vec2, count_vec2, comm2);
  t2 = MPI_Wtime();
  std::cout << "strategy2 communication time strategy2:  " << t2-t1 << " second." << std::endl;
...
}

[ggouaillardet]

@hujiePKU first things first, did you apply the patch and did it fix the hang ?

if i understand correctly C++, both threads invoke MPI_Allreduce the same number of times, but quantity of data to be transferred vary from one thread to an other, so you cannot expect the 2 threads version takes half the time of the serial one.

keep in mind there is some overhead creating/joining threads.
then you should double check your MPI tasks are bound to 2 cores each, otherwise you end up doing time sharing (time sharing plus overhead will definitely make the threaded version slower than the serial one)

last but not least, MPI_THREAD_MULTIPLE has some inherent overhead (e.g. locking). since this test is I/O intensive, threads will likely spend some time waiting for locks, and hence slowing things down. messages might not be received by the thread waiting for them, and this impacts performance.

you might want to compare MPI_THREAD_MULTIPLE 2 tasks and 2 tasks per thread performance with 4 tasks performance and no MPI_THREAD_MULTIPLE (and once again, double check bindings is done in such a way that performance comparison is fair)

[hujie-frank]

@ggouaillardet Yes, I have fixed the hang( non-blocking collective communication in multi-threads) by applying your patch, and I have seen your commitment in master. Thank you very much for your remarkable contributions! Another question, I found that processors are bound by default after version 1.7.4. When I start two threads in each processor, how to be bound these two threads to two cores respectively? Can you support some usages or relevant information available?

[ggouaillardet]

@hujiePKU Open MPI only bind MPI tasks.
then it is up to the programmer/runtime to correctly bind threads.

all you can ask Open MPI is to bound a task on two cores.
Generally speaking, the OpenMP runtime can bind the OpenMP threads on the cores each MPI task was assigned.

in your example, you create the threads manually ... so you have to bind them manually.
note your application really uses 3 threads

• the main threads (that does not perform any real work)
• two worker threads.

so at first, you might want to

• spawn only one thread working on data_cpu_vec2
• work on data_cpu_vec1 in the main thread, and then join the previously spawned thread

for better performance, it is up to you to bind the spawned thread to the second core assigned to the MPI task, and the main thread to the first core assigned to the MPI task.
that can be achieved via the sched_getaffinity() and sched_setaffinity() Linux system calls, or by using the hwloc library.
if you plan to use the hwloc library (this is more portable), then you should probably build it externally, and then configure Open MPI with --with-hwloc=... (e.g. do not use the embedded library)
you might also be interested in the OMPI_Affinity_str extension (this works only with Open MPI, and hence is not portable) to figure out how a given task was bound.

[hujie-frank]

@bosilca @ggouaillardet The real problem I face is that there are some calculations to be done in thousands of iterations(for Deep learning training) on Centos system equiped with 32 cpu cores and 8 Tesla M40. In each iteration, I do some calculations on GPU in main thread while invoking collective operations in a spawned thread. The next iteration will never start until all the operations in last iteration have been done. But collective operations is slower than GPU calculations in each iteration when I start procedure with 8 processors. So I want to spawn one more thread to alleviate burden of former one spawned thread, then reduce the collective time in single iteration. I am frustrated that it doesn't work and even increase the time. A series of Q&A talked above are used to solve this core problem, but still not be tackled. Can you give me some advice?

[bosilca]

Expecting a non-blocking allreduce (a complex operation that involves a lot of computations) to perform correctly when no thread is ensuring MPI progress is not reasonable. I understand when one expect non-blocking point-to-point to provide async progress, as up to a certain degree this can be provided by the hardware. But doing reductions on the hardware, even on contiguous data, is not yet a mainstream operation.

So, as you have a "communication" thread, I would go for using blocking collectives. And then depending on the amount of data involved in this collective, I would try to optimize the collective algorithm used by the MPI library. How large is the data involved in the Allreduce ?