support for polling progress

This commit was SVN r3380.

src/mca/ptl/mx/ptl_mx_component.c

@@ -6,8 +6,7 @@
 #include "util/output.h"
 #include "threads/thread.h"
 #include "ptl_mx.h"
-#include "ptl_mx_recvfrag.h"
+#include "ptl_mx_module.h"
-#include "ptl_mx_sendfrag.h"
 #include "ptl_mx_peer.h"
 
 
@@ -104,9 +103,9 @@ int mca_ptl_mx_component_open(void)
     mca_ptl_mx_module.super.ptl_exclusivity =
         mca_ptl_mx_param_register_int("exclusivity", 0);
     mca_ptl_mx_module.super.ptl_first_frag_size =
-        mca_ptl_mx_param_register_int("first_frag_size", 64*1024);
+        mca_ptl_mx_param_register_int("first_frag_size", 32*1024);
     mca_ptl_mx_module.super.ptl_min_frag_size = 
-        mca_ptl_mx_param_register_int("min_frag_size", 64*1024);
+        mca_ptl_mx_param_register_int("min_frag_size", 32*1024);
     mca_ptl_mx_module.super.ptl_max_frag_size =
         mca_ptl_mx_param_register_int("max_frag_size", -1);
     return OMPI_SUCCESS;
@@ -215,6 +214,27 @@ int mca_ptl_mx_component_control(int param, void* value, size_t size)
 
 int mca_ptl_mx_component_progress(mca_ptl_tstamp_t tstamp)
 {
+    size_t i;
+    for(i=0; i<mca_ptl_mx_component.mx_num_ptls; i++) {
+        mca_ptl_mx_module_t* ptl = mca_ptl_mx_component.mx_ptls[i];
+        mx_request_t mx_request;
+        mx_return_t mx_return;
+        uint32_t mx_result;
+
+        /* poll for completion */
+        mx_return = mx_ipeek(
+            ptl->mx_endpoint,
+            &mx_request,
+            &mx_result);
+        if(mx_return != MX_SUCCESS) {
+            ompi_output(0, "mca_ptl_mx_component_progress: mx_ipeek() failed with status %d\n",
+                mx_return);
+            return OMPI_ERROR;
+        }
+        if(mx_result > 0) {
+            mca_ptl_mx_progress(ptl, mx_request);
+        }
+    }
     return OMPI_SUCCESS;
 }
 

src/mca/ptl/mx/ptl_mx_module.c

@@ -3,8 +3,7 @@
 #include "util/output.h"
 #include "ptl_mx_peer.h"
 #include "ptl_mx_proc.h"
-#include "ptl_mx_recvfrag.h"
+#include "ptl_mx_module.h"
-#include "ptl_mx_sendfrag.h"
 
 
 static mca_ptl_mx_module_t* mca_ptl_mx_create(uint64_t addr);
@@ -90,80 +89,6 @@ int mca_ptl_mx_module_init(void)
 }
 
 
-/**
- * Prepost recv buffers
- */
-
-static inline int mca_ptl_mx_post(mca_ptl_mx_module_t* ptl)
-{
-    mca_ptl_mx_recv_frag_t* frag;
-    mx_return_t status;
-    int rc;
-    /* post an additional recv */
-    MCA_PTL_MX_RECV_FRAG_ALLOC(frag, rc);
-    if(rc != OMPI_SUCCESS) {
-        ompi_output(0, "mca_ptl_mx_thread: unable to allocate recv frag\n");
-        return OMPI_ERR_OUT_OF_RESOURCE;
-    }
-    mca_ptl_mx_recv_frag_init(frag, ptl);
-    status = mx_irecv(
-        ptl->mx_endpoint,
-        frag->frag_segments,
-        frag->frag_segment_count,
-        1,
-        MX_MATCH_MASK_NONE,
-        frag,
-        &frag->frag_request);
-    if(status != MX_SUCCESS) {
-        ompi_output(0, "mca_ptl_mx_post: mx_irecv() failed with status=%d\n", status);
-        return OMPI_ERROR;
-    }
-    return OMPI_SUCCESS;
-}
-
-
-/**
- *  Routine to process complete request(s).
- */
-
-static inline void mca_ptl_mx_progress(mca_ptl_mx_module_t* ptl, mx_request_t mx_request)
-{
-    mx_return_t mx_return;
-    mx_status_t mx_status;
-    uint32_t mx_result;
-    mca_ptl_base_frag_t* frag;
-
-    mx_return = mx_test(
-        ptl->mx_endpoint, 
-        &mx_request, 
-        &mx_status,
-        &mx_result);
-    if(mx_return != MX_SUCCESS) {
-        ompi_output(0, "mca_ptl_mx_progress: mx_test() failed with status=%d\n", mx_return);
-        return;
-    }
-
-    frag = (mca_ptl_base_frag_t*)mx_status.context;
-    switch(frag->frag_type) {
-        case MCA_PTL_FRAGMENT_SEND: 
-        {
-            mca_ptl_mx_send_frag_handler((mca_ptl_mx_send_frag_t*)frag, ptl);
-            break;
-        }
-        case MCA_PTL_FRAGMENT_RECV:
-        {
-            mca_ptl_mx_recv_frag_handler((mca_ptl_mx_recv_frag_t*)frag, ptl);
-            mca_ptl_mx_post(ptl);
-            break;
-        }
-        default:
-        {
-            ompi_output(0, "mca_ptl_mx_progress: invalid request type: %d\n", frag->frag_type);
-            break;
-        }
-    }
-}
-
 /**
  * Thread to progress outstanding requests.
  */

src/mca/ptl/mx/ptl_mx_module.h

@@ -0,0 +1,93 @@
+/*
+ * $HEADER$
+ */
+/**
+ * @file
+ */
+#ifndef MCA_PTL_MX_MODULE_H
+#define MCA_PTL_MX_MODULE_H
+
+#include "ompi_config.h"
+#include "ptl_mx.h"
+#include "ptl_mx_recvfrag.h"
+#include "ptl_mx_sendfrag.h"
+
+
+
+/**
+ * Prepost recv buffers
+ */
+
+static inline int mca_ptl_mx_post(mca_ptl_mx_module_t* ptl)
+{
+    mca_ptl_mx_recv_frag_t* frag;
+    mx_return_t status;
+    int rc;
+    /* post an additional recv */
+    MCA_PTL_MX_RECV_FRAG_ALLOC(frag, rc);
+    if(rc != OMPI_SUCCESS) {
+        ompi_output(0, "mca_ptl_mx_thread: unable to allocate recv frag\n");
+        return OMPI_ERR_OUT_OF_RESOURCE;
+    }
+    mca_ptl_mx_recv_frag_init(frag, ptl);
+    status = mx_irecv(
+        ptl->mx_endpoint,
+        frag->frag_segments,
+        frag->frag_segment_count,
+        1,
+        MX_MATCH_MASK_NONE,
+        frag,
+        &frag->frag_request);
+    if(status != MX_SUCCESS) {
+        ompi_output(0, "mca_ptl_mx_post: mx_irecv() failed with status=%d\n", status);
+        return OMPI_ERROR;
+    }
+    return OMPI_SUCCESS;
+}
+
+
+/**
+ *  Routine to process complete request(s).
+ */
+                                                                                                  
+static inline void mca_ptl_mx_progress(mca_ptl_mx_module_t* ptl, mx_request_t mx_request)
+{
+    mx_return_t mx_return;
+    mx_status_t mx_status;
+    uint32_t mx_result;
+    mca_ptl_base_frag_t* frag;
+                                                                                                  
+    mx_return = mx_test(
+        ptl->mx_endpoint,
+        &mx_request,
+        &mx_status,
+        &mx_result);
+    if(mx_return != MX_SUCCESS) {
+        ompi_output(0, "mca_ptl_mx_progress: mx_test() failed with status=%d\n", mx_return);
+        return;
+    }
+                                                                                                  
+    frag = (mca_ptl_base_frag_t*)mx_status.context;
+    switch(frag->frag_type) {
+        case MCA_PTL_FRAGMENT_SEND:
+        {
+            mca_ptl_mx_send_frag_handler((mca_ptl_mx_send_frag_t*)frag, ptl);
+            break;
+        }
+        case MCA_PTL_FRAGMENT_RECV:
+        {
+            mca_ptl_mx_recv_frag_handler((mca_ptl_mx_recv_frag_t*)frag, ptl);
+            mca_ptl_mx_post(ptl);
+            break;
+        }
+        default:
+        {
+            ompi_output(0, "mca_ptl_mx_progress: invalid request type: %d\n", frag->frag_type);
+            break;
+        }
+    }
+}
+                                                                                                  
+
+#endif
+