[mpich-commits] [mpich] MPICH primary repository branch, master, updated. v3.2b1-78-g17e31e5
Service Account
noreply at mpich.org
Fri Apr 17 19:17:49 CDT 2015
This is an automated email from the git hooks/post-receive script. It was
generated because a ref change was pushed to the repository containing
the project "MPICH primary repository".
The branch, master has been updated
via 17e31e59b491fc1383653ddd41f8264e8c5c69e6 (commit)
from 73e3211228c525b08221fbfcb161f88878f8cd24 (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
- Log -----------------------------------------------------------------
http://git.mpich.org/mpich.git/commitdiff/17e31e59b491fc1383653ddd41f8264e8c5c69e6
commit 17e31e59b491fc1383653ddd41f8264e8c5c69e6
Author: Halim Amer <aamer at anl.gov>
Date: Fri Apr 17 19:17:23 2015 -0500
Applied the PPoPP patch
diff --git a/src/include/mpiimplthreadpost.h b/src/include/mpiimplthreadpost.h
index be4578f..60a0882 100644
--- a/src/include/mpiimplthreadpost.h
+++ b/src/include/mpiimplthreadpost.h
@@ -191,7 +191,13 @@ MPIU_Thread_CS_yield_lockname_recursive_impl_(enum MPIU_Nest_mutexes kind,
MPID_Thread_mutex_unlock(mutex);
MPID_Thread_yield();
- MPID_Thread_mutex_lock(mutex);
+
+/* Use low priority here because this thread has a lower probability
+ * to do useful work compared to others outside the progress engine.
+ * This is only effective with the priority lock.
+ * The other lock types implement a plain lock underneath.
+ */
+ MPID_Thread_mutex_lock_low(mutex);
}
/* undef for safety, this is a commonly-included header */
diff --git a/src/include/thread/mpiu_thread_posix_funcs.h b/src/include/thread/mpiu_thread_posix_funcs.h
index 10af9fc..facea60 100644
--- a/src/include/thread/mpiu_thread_posix_funcs.h
+++ b/src/include/thread/mpiu_thread_posix_funcs.h
@@ -8,8 +8,10 @@
/*
* Threads
*/
-
+#include <limits.h>
#include "mpiu_process_wrappers.h" /* for MPIU_PW_Sched_yield */
+#include "mpidbg.h"
+#include "opa_primitives.h"
/*
One of PTHREAD_MUTEX_RECURSIVE_NP and PTHREAD_MUTEX_RECURSIVE seem to be
@@ -53,144 +55,443 @@ do { \
MPIU_DBG_MSG(THREAD,VERBOSE,"exit MPIU_Thread_yield"); \
} while (0)
+/*----------------------*/
+/* Ticket Lock Routines */
+/*----------------------*/
-/*
- * Mutexes
- */
+static inline int ticket_lock_init(ticket_lock_t *lock)
+{
+ OPA_store_int(&lock->next_ticket, 0);
+ OPA_store_int(&lock->now_serving, 0);
+ return 0;
+}
+
+/* Atomically increment the nex_ticket counter and get my ticket.
+ Then spin on now_serving until it equals my ticket.
+ */
+static inline int ticket_acquire_lock(ticket_lock_t* lock)
+{
+ int my_ticket = OPA_fetch_and_add_int(&lock->next_ticket, 1);
+ while(OPA_load_int(&lock->now_serving) != my_ticket)
+ ;
+ return 0;
+}
+
+/* Release the lock
+ */
+static inline int ticket_release_lock(ticket_lock_t* lock)
+{
+ /* Avoid compiler reordering before releasing the lock*/
+ OPA_compiler_barrier();
+ OPA_incr_int(&lock->now_serving);
+ return 0;
+}
+
+/*------------------------*/
+/* Priority Lock Routines */
+/*------------------------*/
+
+static inline int priority_lock_init(priority_lock_t *lock)
+{
+ OPA_store_int(&lock->next_ticket_H, 0);
+ OPA_store_int(&lock->now_serving_H, 0);
+ OPA_store_int(&lock->next_ticket_L, 0);
+ OPA_store_int(&lock->now_serving_L, 0);
+ OPA_store_int(&lock->next_ticket_B, 0);
+ OPA_store_int(&lock->now_serving_B, 0);
+ lock->already_blocked = 0;
+ return 0;
+}
+
+/* First wait my turn in this priority level, Then if I am the first
+ one to block the LPRs, wait for the last LPR to terminate.
+ */
+static inline int priority_acquire_lock(priority_lock_t* lock)
+{
+ int my_ticket = OPA_fetch_and_add_int(&lock->next_ticket_H, 1);
+ while(OPA_load_int(&lock->now_serving_H) != my_ticket)
+ ;
+ int B_ticket;
+ if(!lock->already_blocked)
+ {
+ B_ticket = OPA_fetch_and_add_int(&lock->next_ticket_B, 1);
+ while(OPA_load_int(&lock->now_serving_B) != B_ticket)
+ ;
+ lock->already_blocked = 1;
+ }
+ lock->last_acquisition_priority = HIGH_PRIORITY;
+ return 0;
+}
+static inline int priority_acquire_lock_low(priority_lock_t* lock)
+{
+ int my_ticket = OPA_fetch_and_add_int(&lock->next_ticket_L, 1);
+ while(OPA_load_int(&lock->now_serving_L) != my_ticket)
+ ;
+ int B_ticket = OPA_fetch_and_add_int(&lock->next_ticket_B, 1);
+ while(OPA_load_int(&lock->now_serving_B) != B_ticket)
+ ;
+ lock->last_acquisition_priority = LOW_PRIORITY;
+ return 0;
+}
-/* FIXME: mutex creation and destruction should be implemented as routines
- because there is no reason to use macros (these are not on the performance
- critical path). Making these macros requires that any code that might use
- these must load all of the pthread.h (or other thread library) support.
+/* Release the lock
+ */
+static inline int priority_release_lock(priority_lock_t* lock)
+{
+ int err=0;
+ if(lock->last_acquisition_priority==HIGH_PRIORITY)
+ err = priority_release_lock_high(lock);
+ else
+ err = priority_release_lock_low(lock);
+ return err;
+}
+
+static inline int priority_release_lock_high(priority_lock_t* lock)
+{
+ /* Avoid compiler reordering before releasing the lock*/
+ OPA_compiler_barrier();
+ /* Only me in the HPRs queue -> let the LPRs pass */
+ if(OPA_load_int(&lock->now_serving_H) == OPA_load_int(&lock->next_ticket_H) - 1)
+ {
+ lock->already_blocked = 0;
+ OPA_incr_int(&lock->now_serving_B);
+ }
+ OPA_incr_int(&lock->now_serving_H);
+ return 0;
+}
+
+static inline int priority_release_lock_low(priority_lock_t* lock)
+{
+ /* Avoid compiler reordering before releasing the lock*/
+ OPA_compiler_barrier();
+ OPA_incr_int(&lock->now_serving_B);
+ OPA_incr_int(&lock->now_serving_L);
+ return 0;
+}
+
+/*
+ * MPIU Mutexes: encapsulate lower level locks like pthread mutexes
+ * and ticket locks
*/
-/* FIXME: using constant initializer if available */
#if !defined(MPICH_DEBUG_MUTEX) || !defined(PTHREAD_MUTEX_ERRORCHECK_VALUE)
-#define MPIU_Thread_mutex_create(mutex_ptr_, err_ptr_) \
-do { \
- int err__; \
- \
- err__ = pthread_mutex_init((mutex_ptr_), NULL); \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
- *(int *)(err_ptr_) = err__; \
- MPIU_DBG_MSG_P(THREAD,TYPICAL,"Created MPIU_Thread_mutex %p", (mutex_ptr_)); \
-} while (0)
+static inline void MPIU_Thread_mutex_create(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_init(&mutex_ptr_->pthread_lock, NULL);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_lock_init(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_lock_init(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+ MPIU_DBG_MSG_P(THREAD,TYPICAL,"Created MPIU_Thread_mutex %p", (mutex_ptr_));
+}
#else /* MPICH_DEBUG_MUTEX */
-#define MPIU_Thread_mutex_create(mutex_ptr_, err_ptr_) \
-do { \
- int err__; \
- pthread_mutexattr_t attr__; \
- \
- /* FIXME this used to be PTHREAD_MUTEX_ERRORCHECK_NP, but we had to change
- it for the thread granularity work when we needed recursive mutexes. We
- should go through this code and see if there's any good way to implement
- error checked versions with the recursive mutexes. */ \
- pthread_mutexattr_init(&attr__); \
- pthread_mutexattr_settype(&attr__, PTHREAD_MUTEX_ERRORCHECK_VALUE); \
- err__ = pthread_mutex_init((mutex_ptr_), &attr__); \
- if (err__) \
+static inline void MPIU_Thread_mutex_create(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ pthread_mutexattr_t attr__;
+ /* FIXME this used to be PTHREAD_MUTEX_ERRORCHECK_NP, but we had to change
+ it for the thread granularity work when we needed recursive mutexes. We
+ should go through this code and see if there's any good way to implement
+ error checked versions with the recursive mutexes. */
+ pthread_mutexattr_init(&attr__);
+ pthread_mutexattr_settype(&attr__, PTHREAD_MUTEX_ERRORCHECK_VALUE);
+ err__ = pthread_mutex_init(&mutex_ptr_->pthread_lock, &attr__);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_lock_init(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_lock_init(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+ if (err__)
MPIU_Internal_sys_error_printf("pthread_mutex_init", err__, \
- " %s:%d\n", __FILE__, __LINE__);\
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
- *(int *)(err_ptr_) = err__; \
- MPIU_DBG_MSG_P(THREAD,TYPICAL,"Created MPIU_Thread_mutex %p", (mutex_ptr_)); \
-} while (0)
+ " %s:%d\n", __FILE__, __LINE__);
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+ MPIU_DBG_MSG_P(THREAD,TYPICAL,"Created MPIU_Thread_mutex %p", (mutex_ptr_));
+}
#endif
-#define MPIU_Thread_mutex_destroy(mutex_ptr_, err_ptr_) \
-do { \
- int err__; \
- \
- MPIU_DBG_MSG_P(THREAD,TYPICAL,"About to destroy MPIU_Thread_mutex %p", (mutex_ptr_)); \
- err__ = pthread_mutex_destroy(mutex_ptr_); \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
- *(int *)(err_ptr_) = err__; \
-} while (0)
+static inline void MPIU_Thread_mutex_destroy(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+
+ MPIU_DBG_MSG_P(THREAD,TYPICAL,"About to destroy MPIU_Thread_mutex %p", (mutex_ptr_));
+
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_destroy(&mutex_ptr_->pthread_lock);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ /* FIXME Should we do something here ?*/
+ err__ = 0;
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ /* FIXME Should we do something here ?*/
+ err__ = 0;
+ break;
+ }
+ }
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+}
#ifndef MPICH_DEBUG_MUTEX
-#define MPIU_Thread_mutex_lock(mutex_ptr_, err_ptr_) \
-do { \
- int err__; \
- MPIU_DBG_MSG_P(THREAD,VERBOSE,"enter MPIU_Thread_mutex_lock %p", (mutex_ptr_)); \
- err__ = pthread_mutex_lock(mutex_ptr_); \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
- *(int *)(err_ptr_) = err__; \
- MPIU_DBG_MSG_P(THREAD,VERBOSE,"exit MPIU_Thread_mutex_lock %p", (mutex_ptr_)); \
-} while (0)
+static inline void MPIU_Thread_mutex_lock(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"enter MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_lock(&mutex_ptr_->pthread_lock);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_acquire_lock(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_acquire_lock(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"exit MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+}
#else /* MPICH_DEBUG_MUTEX */
-#define MPIU_Thread_mutex_lock(mutex_ptr_, err_ptr_) \
-do { \
- int err__; \
- MPIU_DBG_MSG_P(THREAD,VERBOSE,"enter MPIU_Thread_mutex_lock %p", (mutex_ptr_)); \
- err__ = pthread_mutex_lock(mutex_ptr_); \
- if (err__) \
- { \
- MPIU_DBG_MSG_S(THREAD,TERSE," mutex lock error: %s", MPIU_Strerror(err__)); \
+static inline void MPIU_Thread_mutex_lock(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"enter MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_lock(&mutex_ptr_->pthread_lock);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_acquire_lock(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_acquire_lock(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+
+ if (err__)
+ {
+ MPIU_DBG_MSG_S(THREAD,TERSE," mutex lock error: %s", MPIU_Strerror(err__));
MPIU_Internal_sys_error_printf("pthread_mutex_lock", err__,\
- " %s:%d\n", __FILE__, __LINE__);\
- } \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
- *(int *)(err_ptr_) = err__; \
- MPIU_DBG_MSG_P(THREAD,VERBOSE,"exit MPIU_Thread_mutex_lock %p", (mutex_ptr_)); \
-} while (0)
+ " %s:%d\n", __FILE__, __LINE__);
+ }
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"exit MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+}
#endif
#ifndef MPICH_DEBUG_MUTEX
-#define MPIU_Thread_mutex_unlock(mutex_ptr_, err_ptr_) \
-do { \
- int err__; \
- \
- MPIU_DBG_MSG_P(THREAD,TYPICAL,"MPIU_Thread_mutex_unlock %p", (mutex_ptr_)); \
- err__ = pthread_mutex_unlock(mutex_ptr_); \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
- *(int *)(err_ptr_) = err__; \
-} while (0)
+static inline void MPIU_Thread_mutex_lock_low(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"enter MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_lock(&mutex_ptr_->pthread_lock);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_acquire_lock(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_acquire_lock_low(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"exit MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+}
#else /* MPICH_DEBUG_MUTEX */
-#define MPIU_Thread_mutex_unlock(mutex_ptr_, err_ptr_) \
-do { \
- int err__; \
- \
- MPIU_DBG_MSG_P(THREAD,VERBOSE,"MPIU_Thread_mutex_unlock %p", (mutex_ptr_)); \
- err__ = pthread_mutex_unlock(mutex_ptr_); \
- if (err__) \
- { \
- MPIU_DBG_MSG_S(THREAD,TERSE," mutex unlock error: %s", MPIU_Strerror(err__)); \
- MPIU_Internal_sys_error_printf("pthread_mutex_unlock", err__, \
- " %s:%d\n", __FILE__, __LINE__); \
- } \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
- *(int *)(err_ptr_) = err__; \
-} while (0)
+static inline void MPIU_Thread_mutex_lock_low(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"enter MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_lock(&mutex_ptr_->pthread_lock);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_acquire_lock(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_acquire_lock_low(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+
+ if (err__)
+ {
+ MPIU_DBG_MSG_S(THREAD,TERSE," mutex lock error: %s", MPIU_Strerror(err__));
+ MPIU_Internal_sys_error_printf("pthread_mutex_lock", err__,\
+ " %s:%d\n", __FILE__, __LINE__);
+ }
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"exit MPIU_Thread_mutex_lock %p", (mutex_ptr_));
+}
#endif
#ifndef MPICH_DEBUG_MUTEX
-#define MPIU_Thread_mutex_trylock(mutex_ptr_, flag_ptr_, err_ptr_) \
-do { \
- int err__; \
- \
- err__ = pthread_mutex_trylock(mutex_ptr_); \
- *(flag_ptr_) = (err__ == 0) ? TRUE : FALSE; \
- MPIU_DBG_MSG_FMT(THREAD,VERBOSE,(MPIU_DBG_FDEST, "MPIU_Thread_mutex_trylock mutex=%p result=%s", (mutex_ptr_), (*(flag_ptr_) ? "success" : "failure"))); \
- *(int *)(err_ptr_) = (err__ == EBUSY) ? MPIU_THREAD_SUCCESS : err__; \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
-} while (0)
+static inline void MPIU_Thread_mutex_unlock(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+
+ MPIU_DBG_MSG_P(THREAD,TYPICAL,"MPIU_Thread_mutex_unlock %p", (mutex_ptr_));
+
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_unlock(&mutex_ptr_->pthread_lock);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_release_lock(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_release_lock(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+}
#else /* MPICH_DEBUG_MUTEX */
-#define MPIU_Thread_mutex_trylock(mutex_ptr_, flag_ptr_, err_ptr_) \
-do { \
- int err__; \
- \
- err__ = pthread_mutex_trylock(mutex_ptr_); \
- if (err__ && err__ != EBUSY) \
- { \
- MPIU_DBG_MSG_S(THREAD,TERSE," mutex trylock error: %s", MPIU_Strerror(err__)); \
+static inline void MPIU_Thread_mutex_unlock(MPIU_Thread_mutex_t* mutex_ptr_, int* err_ptr_)
+{
+ int err__=0;
+
+ MPIU_DBG_MSG_P(THREAD,VERBOSE,"MPIU_Thread_mutex_unlock %p", (mutex_ptr_));
+
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ err__ = pthread_mutex_unlock(&mutex_ptr_->pthread_lock);
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ err__ = ticket_release_lock(&mutex_ptr_->ticket_lock);
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ err__ = priority_release_lock(&mutex_ptr_->priority_lock);
+ break;
+ }
+ }
+ if (err__)
+ {
+ MPIU_DBG_MSG_S(THREAD,TERSE," mutex unlock error: %s", MPIU_Strerror(err__));
+ MPIU_Internal_sys_error_printf("pthread_mutex_unlock", err__,\
+ " %s:%d\n", __FILE__, __LINE__);
+ }
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+ *(int *)(err_ptr_) = err__;
+}
+#endif
+
+#ifndef MPICH_DEBUG_MUTEX
+static inline void MPIU_Thread_mutex_trylock(MPIU_Thread_mutex_t* mutex_ptr_, int* flag_ptr_, int* err_ptr_)
+{
+ int err__=0;
+
+ if(MPIU_lock_type == MPIU_MUTEX)
+ err__ = pthread_mutex_trylock(&mutex_ptr_->pthread_lock);
+ else
+ /*No trylock routine for ticket-based locks*/
+ err__ = 1;
+
+ *(flag_ptr_) = (err__ == 0) ? TRUE : FALSE;
+ MPIU_DBG_MSG_FMT(THREAD,VERBOSE,(MPIU_DBG_FDEST, "MPIU_Thread_mutex_trylock mutex=%p result=%s", (mutex_ptr_), (*(flag_ptr_) ? "success" : "failure")));
+ *(int *)(err_ptr_) = (err__ == EBUSY) ? MPIU_THREAD_SUCCESS : err__;
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+}
+#else /* MPICH_DEBUG_MUTEX */
+static inline void MPIU_Thread_mutex_trylock(MPIU_Thread_mutex_t* mutex_ptr_, int* flag_ptr_, int* err_ptr_)
+{
+ int err__=0;
+
+ if(MPIU_lock_type == MPIU_MUTEX)
+ err__ = pthread_mutex_trylock(&mutex_ptr_->pthread_lock);
+ else
+ /*No trylock routine for ticket-based locks*/
+ err__ = 1;
+
+ if (err__ && err__ != EBUSY)
+ {
+ MPIU_DBG_MSG_S(THREAD,TERSE," mutex trylock error: %s", MPIU_Strerror(err__));
MPIU_Internal_sys_error_printf("pthread_mutex_trylock", err__,\
- " %s:%d\n", __FILE__, __LINE__);\
- } \
- *(flag_ptr_) = (err__ == 0) ? TRUE : FALSE; \
- MPIU_DBG_MSG_FMT(THREAD,VERBOSE,(MPIU_DBG_FDEST, "MPIU_Thread_mutex_trylock mutex=%p result=%s", (mutex_ptr_), (*(flag_ptr_) ? "success" : "failure"))); \
- *(int *)(err_ptr_) = (err__ == EBUSY) ? MPIU_THREAD_SUCCESS : err__; \
- /* FIXME: convert error to an MPIU_THREAD_ERR value */ \
-} while (0)
+ " %s:%d\n", __FILE__, __LINE__);
+ }
+ *(flag_ptr_) = (err__ == 0) ? TRUE : FALSE;
+ MPIU_DBG_MSG_FMT(THREAD,VERBOSE,(MPIU_DBG_FDEST, "MPIU_Thread_mutex_trylock mutex=%p result=%s", (mutex_ptr_), (*(flag_ptr_) ? "success" : "failure")));
+ *(int *)(err_ptr_) = (err__ == EBUSY) ? MPIU_THREAD_SUCCESS : err__;
+ /* FIXME: convert error to an MPIU_THREAD_ERR value */
+}
#endif
/*
diff --git a/src/include/thread/mpiu_thread_posix_types.h b/src/include/thread/mpiu_thread_posix_types.h
index b38f7a1..c163b67 100644
--- a/src/include/thread/mpiu_thread_posix_types.h
+++ b/src/include/thread/mpiu_thread_posix_types.h
@@ -7,8 +7,59 @@
#include <errno.h>
#include <pthread.h>
+#include "opa_primitives.h"
+
+/* Define lock types here */
+typedef enum{
+ MPIU_MUTEX,
+ MPIU_TICKET,
+ MPIU_PRIORITY
+}MPIU_Thread_lock_impl_t;
+
+MPIU_Thread_lock_impl_t MPIU_lock_type;
+
+/*----------------------------*/
+/* Ticket lock data structure */
+/*----------------------------*/
+
+/* Define the lock as a structure of two counters */
+typedef struct ticket_lock_t{
+ OPA_int_t next_ticket;
+ OPA_int_t now_serving;
+} ticket_lock_t;
+
+/*------------------------------*/
+/* Priority lock data structure */
+/*------------------------------*/
+
+#define HIGH_PRIORITY 1
+#define LOW_PRIORITY 2
+
+ /* We only define for now 2 levels of priority: */
+ /* The high priority requests (HPRs) */
+ /* The low priority requests (LPRs) */
+typedef struct priority_lock_t{
+ OPA_int_t next_ticket_H __attribute__((aligned(64)));
+ OPA_int_t now_serving_H;
+ OPA_int_t next_ticket_L __attribute__((aligned(64)));
+ OPA_int_t now_serving_L;
+ /* In addition we include two other counters */
+ /* so high priority requests block the lower ones */
+ OPA_int_t next_ticket_B __attribute__((aligned(64)));
+ OPA_int_t now_serving_B;
+ /* This is to allow high priority requests know */
+ /* that low priority requests are already blocked */
+ /* by another high*/
+ unsigned already_blocked;
+ unsigned last_acquisition_priority;
+} priority_lock_t;
+
+typedef struct MPIU_Thread_mutex_t{
+ pthread_mutex_t pthread_lock;
+ ticket_lock_t ticket_lock;
+ priority_lock_t priority_lock;
+} MPIU_Thread_mutex_t;
-typedef pthread_mutex_t MPIU_Thread_mutex_t;
typedef pthread_cond_t MPIU_Thread_cond_t;
typedef pthread_t MPIU_Thread_id_t;
typedef pthread_key_t MPIU_Thread_tls_t;
diff --git a/src/mpi/init/async.c b/src/mpi/init/async.c
index c003827..a95e648 100644
--- a/src/mpi/init/async.c
+++ b/src/mpi/init/async.c
@@ -61,9 +61,10 @@ static void progress_fn(void * data)
MPIU_Thread_mutex_unlock(&progress_mutex, &mpi_errno);
MPIU_Assert(!mpi_errno);
-
- MPIU_Thread_cond_signal(&progress_cond, &mpi_errno);
- MPIU_Assert(!mpi_errno);
+ if(MPIU_lock_type==MPIU_MUTEX){
+ MPIU_Thread_cond_signal(&progress_cond, &mpi_errno);
+ MPIU_Assert(!mpi_errno);
+ } /* Else: busy loop will automatically break*/
MPIU_THREAD_CS_EXIT(ALLFUNC,);
@@ -137,16 +138,21 @@ int MPIR_Finalize_async_thread(void)
/* XXX DJG why is this unlock/lock necessary? Should we just YIELD here or later? */
MPIU_THREAD_CS_EXIT(ALLFUNC,);
- MPIU_Thread_mutex_lock(&progress_mutex, &mpi_errno);
- MPIU_Assert(!mpi_errno);
-
- while (!progress_thread_done) {
- MPIU_Thread_cond_wait(&progress_cond, &progress_mutex, &mpi_errno);
- MPIU_Assert(!mpi_errno);
- }
-
- MPIU_Thread_mutex_unlock(&progress_mutex, &mpi_errno);
- MPIU_Assert(!mpi_errno);
+ if(MPIU_lock_type==MPIU_MUTEX){
+ MPIU_Thread_mutex_lock(&progress_mutex, &mpi_errno);
+ MPIU_Assert(!mpi_errno);
+
+ while (!progress_thread_done) {
+ MPIU_Thread_cond_wait(&progress_cond, &progress_mutex.pthread_lock, &mpi_errno);
+ MPIU_Assert(!mpi_errno);
+ }
+
+ MPIU_Thread_mutex_unlock(&progress_mutex, &mpi_errno);
+ MPIU_Assert(!mpi_errno);
+ }
+ else
+ while (!progress_thread_done) ; /* busy loop */
+ /* No need to unlock the mutex */
mpi_errno = MPIR_Comm_free_impl(progress_comm_ptr);
MPIU_Assert(!mpi_errno);
diff --git a/src/mpi/init/initthread.c b/src/mpi/init/initthread.c
index aaf20c8..2b5db44 100644
--- a/src/mpi/init/initthread.c
+++ b/src/mpi/init/initthread.c
@@ -190,6 +190,16 @@ static int MPIR_Thread_CS_Init( void )
int err;
MPIU_THREADPRIV_DECL;
+ MPIU_lock_type = MPIU_MUTEX;
+ char *s;
+ s = getenv( "MPICH_LOCK_TYPE" );
+ if(s){
+ if(strcmp( "ticket", s ) == 0)
+ MPIU_lock_type = MPIU_TICKET;
+ else if (strcmp( "priority", s ) == 0)
+ MPIU_lock_type = MPIU_PRIORITY;
+ }
+
MPIU_Assert(MPICH_MAX_LOCKS >= MPIU_Nest_NUM_MUTEXES);
/* we create this at all granularities right now */
@@ -558,6 +568,30 @@ int MPIR_Init_thread(int * argc, char ***argv, int required, int * provided)
if (mpi_errno == MPI_SUCCESS)
mpi_errno = MPID_InitCompleted();
+ const char *s;
+
+ switch(MPIU_lock_type){
+ case MPIU_MUTEX:
+ {
+ s = "mutex";
+ break;
+ }
+ case MPIU_TICKET:
+ {
+ s = "ticket";
+ break;
+ }
+ case MPIU_PRIORITY:
+ {
+ s = "priority";
+ break;
+ }
+ }
+
+ if(MPIR_Process.comm_world->rank==0){
+ printf("\n[MPICH INFO] Critical section(s) based on %s \n\n", s);
+ }
+
fn_exit:
MPIU_THREAD_CS_EXIT(INIT,required);
/* Make fields of MPIR_Process global visible and set mpich_state
@@ -626,6 +660,7 @@ int MPI_Init_thread( int *argc, char ***argv, int required, int *provided )
{
int mpi_errno = MPI_SUCCESS;
int rc ATTRIBUTE((unused)), reqd = required;
+
MPID_MPI_INIT_STATE_DECL(MPID_STATE_MPI_INIT_THREAD);
rc = MPID_Wtime_init();
diff --git a/src/mpid/ch3/channels/nemesis/src/ch3_progress.c b/src/mpid/ch3/channels/nemesis/src/ch3_progress.c
index f022651..8dda2cd 100644
--- a/src/mpid/ch3/channels/nemesis/src/ch3_progress.c
+++ b/src/mpid/ch3/channels/nemesis/src/ch3_progress.c
@@ -563,13 +563,24 @@ static int MPIDI_CH3I_Progress_delay(unsigned int completion_count)
/* FIXME should be appropriately abstracted somehow */
# if defined(MPICH_IS_THREADED) && (MPIU_THREAD_GRANULARITY == MPIU_THREAD_GRANULARITY_GLOBAL)
{
+ if(MPIU_lock_type==MPIU_MUTEX){
while (1)
{
if (completion_count != OPA_load_int(&MPIDI_CH3I_progress_completion_count) ||
MPIDI_CH3I_progress_blocked != TRUE)
break;
- MPID_Thread_cond_wait(&MPIDI_CH3I_progress_completion_cond, &MPIR_ThreadInfo.global_mutex/*MPIDCOMM*/);
+ MPID_Thread_cond_wait(&MPIDI_CH3I_progress_completion_cond, &MPIR_ThreadInfo.global_mutex.pthread_lock/*MPIDCOMM*/);
}
+ }
+ else{
+ /* First release the lock and then enter a busy loop*/
+ MPID_Thread_mutex_unlock(&MPIR_ThreadInfo.global_mutex/*MPIDCOMM*/);
+ while (completion_count == OPA_load_int(&MPIDI_CH3I_progress_completion_count) &&
+ MPIDI_CH3I_progress_blocked == TRUE)
+ ; /* wait in a busy loop*/
+ /* Hold the lock again*/
+ MPID_Thread_mutex_lock(&MPIR_ThreadInfo.global_mutex/*MPIDCOMM*/);
+ }
}
# endif
@@ -593,7 +604,9 @@ static int MPIDI_CH3I_Progress_continue(unsigned int completion_count/*unused*/)
# if defined(MPICH_IS_THREADED) && (MPIU_THREAD_GRANULARITY == MPIU_THREAD_GRANULARITY_GLOBAL)
{
/* we currently hold the MPIDCOMM CS */
+ if(MPIU_lock_type==MPIU_MUTEX)
MPID_Thread_cond_broadcast(&MPIDI_CH3I_progress_completion_cond);
+ /* Else, the condition shoul be satisfied to break the busy loop*/
}
# endif
diff --git a/src/mpid/common/thread/mpid_thread.h b/src/mpid/common/thread/mpid_thread.h
index 8386cf2..df9c355 100644
--- a/src/mpid/common/thread/mpid_thread.h
+++ b/src/mpid/common/thread/mpid_thread.h
@@ -314,6 +314,14 @@ do { \
("mutex_unlock failed, err_=%d (%s)",err_,MPIU_Strerror(err_))); \
} while (0)
+#define MPID_Thread_mutex_lock_low(mutex_) \
+do { \
+ int err_; \
+ MPIU_Thread_mutex_lock_low((mutex_), &err_); \
+ MPIU_Assert_fmt_msg(err_ == MPIU_THREAD_SUCCESS, \
+ ("mutex_lock failed, err_=%d (%s)",err_,MPIU_Strerror(err_))); \
+} while (0)
+
#define MPID_Thread_mutex_trylock(mutex_, flag_) \
do { \
int err_; \
-----------------------------------------------------------------------
Summary of changes:
src/include/mpiimplthreadpost.h | 8 +-
src/include/thread/mpiu_thread_posix_funcs.h | 535 +++++++++++++++++-----
src/include/thread/mpiu_thread_posix_types.h | 53 +++-
src/mpi/init/async.c | 32 +-
src/mpi/init/initthread.c | 35 ++
src/mpid/ch3/channels/nemesis/src/ch3_progress.c | 15 +-
src/mpid/common/thread/mpid_thread.h | 8 +
7 files changed, 553 insertions(+), 133 deletions(-)
hooks/post-receive
--
MPICH primary repository
More information about the commits
mailing list