Create a new opal_async_event_base and have the pmix/native and ORTE …

opal/mca/event/base/event_base_frame.c

@@ -1,6 +1,6 @@
 /*
  * Copyright (c) 2010      Cisco Systems, Inc.  All rights reserved.
- * Copyright (c) 2014      Intel, Inc. All rights reserved.
+ * Copyright (c) 2014-2015 Intel, Inc. All rights reserved.
  * $COPYRIGHT$
  *
  * Additional copyrights may follow
@@ -65,6 +65,7 @@ static int opal_event_base_close(void)
  * Globals
  */
 opal_event_base_t *opal_sync_event_base=NULL;
+opal_event_base_t *opal_async_event_base=NULL;
 
 static int opal_event_base_open(mca_base_open_flag_t flags)
 {

opal/mca/event/external/external.h

@@ -29,6 +29,7 @@ typedef struct event_base opal_event_base_t;
 typedef struct event opal_event_t;
 
 OPAL_DECLSPEC extern opal_event_base_t *opal_sync_event_base;
+OPAL_DECLSPEC extern opal_event_base_t *opal_async_event_base;
 
 #define OPAL_EV_TIMEOUT EV_TIMEOUT
 #define OPAL_EV_READ    EV_READ

opal/mca/event/libevent2022/libevent2022.h

@@ -68,6 +68,7 @@ typedef struct event_base opal_event_base_t;
 typedef struct event opal_event_t;
 
 OPAL_DECLSPEC extern opal_event_base_t *opal_sync_event_base;
+OPAL_DECLSPEC extern opal_event_base_t *opal_async_event_base;
 
 #define OPAL_EV_TIMEOUT EV_TIMEOUT
 #define OPAL_EV_READ    EV_READ

opal/mca/pmix/native/pmix_native.c

@@ -189,7 +189,7 @@ static int native_init(void)
         opal_argv_free(uri);
 
         /* create an event base and progress thread for us */
-        if (NULL == (mca_pmix_native_component.evbase = opal_start_progress_thread("pmix_native", true))) {
+        if (NULL == (mca_pmix_native_component.evbase = opal_start_progress_thread("opal_async", true))) {
             return OPAL_ERROR;
         }
     }
@@ -251,7 +251,7 @@ static int native_fini(void)
     }
 
     if (NULL != mca_pmix_native_component.evbase) {
-        opal_stop_progress_thread("pmix_native", true);
+        opal_stop_progress_thread("opal_async", true);
         mca_pmix_native_component.evbase = NULL;
     }
 

opal/runtime/opal_progress_threads.c

@@ -25,6 +25,7 @@
 /* create a tracking object for progress threads */
 typedef struct {
     opal_list_item_t super;
+    int refcount;
     char *name;
     opal_event_base_t *ev_base;
     volatile bool ev_active;
@@ -36,6 +37,7 @@ typedef struct {
 } opal_progress_tracker_t;
 static void trkcon(opal_progress_tracker_t *p)
 {
+    p->refcount = 1;  // start at one since someone created it
     p->name = NULL;
     p->ev_base = NULL;
     p->ev_active = true;
@@ -97,6 +99,21 @@ opal_event_base_t *opal_start_progress_thread(char *name,
     opal_progress_tracker_t *trk;
     int rc;
 
+    if (!inited) {
+        OBJ_CONSTRUCT(&tracking, opal_list_t);
+        inited = true;
+    }
+
+    /* check if we already have this thread */
+    OPAL_LIST_FOREACH(trk, &tracking, opal_progress_tracker_t) {
+        if (0 == strcmp(name, trk->name)) {
+            /* we do, so up the refcount on it */
+            ++trk->refcount;
+            /* return the existing base */
+            return trk->ev_base;
+        }
+    }
+
     trk = OBJ_NEW(opal_progress_tracker_t);
     trk->name = strdup(name);
     if (NULL == (trk->ev_base = opal_event_base_create())) {
@@ -136,10 +153,6 @@ opal_event_base_t *opal_start_progress_thread(char *name,
         OBJ_RELEASE(trk);
         return NULL;
     }
-    if (!inited) {
-        OBJ_CONSTRUCT(&tracking, opal_list_t);
-        inited = true;
-    }
     opal_list_append(&tracking, &trk->super);
     return trk->ev_base;
 }
@@ -166,6 +179,12 @@ void opal_stop_progress_thread(char *name, bool cleanup)
                 }
                 return;
             }
+            /* decrement the refcount */
+            --trk->refcount;
+            /* if we have reached zero, then it's time to stop it */
+            if (0 < trk->refcount) {
+                return;
+            }
             /* mark it as inactive */
             trk->ev_active = false;
             /* break the event loop - this will cause the loop to exit
@@ -207,6 +226,8 @@ int opal_restart_progress_thread(char *name)
                 OPAL_ERROR_LOG(OPAL_ERR_NOT_SUPPORTED);
                 return OPAL_ERR_NOT_SUPPORTED;
             }
+            /* up the refcount */
+            ++trk->refcount;
             /* ensure the block is set, if requested */
             if (0 <= trk->pipe[0] && !trk->block_active) {
                 opal_event_add(&trk->block, 0);

orte/mca/ess/base/ess_base_std_app.c

@@ -114,8 +114,9 @@ int orte_ess_base_app_setup(bool db_restrict_local)
         opal_proc_local_set(&orte_process_info.super);
     }
 
-    /* get a separate orte event base */
-    orte_event_base = opal_start_progress_thread("orte", true);
+    /* get an async event base - we use the opal_async one so
+     * we don't startup extra threads if not needed */
+    orte_event_base = opal_start_progress_thread("opal_async", true);
     progress_thread_running = true;
     /* open and setup the state machine */
     if (ORTE_SUCCESS != (ret = mca_base_framework_open(&orte_state_base_framework, 0))) {
@@ -337,13 +338,6 @@ int orte_ess_base_app_finalize(void)
 {
     orte_cr_finalize();
 
-    /* release the event base so we stop all potential
-     * race conditions in the messaging teardown */
-    if (progress_thread_running) {
-        opal_stop_progress_thread("orte", false);
-        progress_thread_running = false;
-    }
-
 #if OPAL_ENABLE_FT_CR == 1
     (void) mca_base_framework_close(&orte_snapc_base_framework);
     (void) mca_base_framework_close(&orte_sstore_base_framework);
@@ -365,8 +359,12 @@ int orte_ess_base_app_finalize(void)
 
     orte_session_dir_finalize(ORTE_PROC_MY_NAME);
 
-    /* free the event base to cleanup memory */
-    opal_stop_progress_thread("orte", true);
+    /* release the event base */
+    if (progress_thread_running) {
+        opal_stop_progress_thread("opal_async", true);
+        progress_thread_running = false;
+    }
+
     return ORTE_SUCCESS;
 }