Changeset 4931

Timestamp:

07/09/09 15:00:30 (8 months ago)

Author:

dkim

Message:

Code cleanup.

Location:

mpich2/branches/dev/dkim/src/pm/hydra

Files:

• include/hydra_utils.h (modified) (1 diff)
• pm/pmiserv/pmi_proxy_cb.c (modified) (2 diffs)
• pm/pmiserv/pmi_proxy_utils.c (modified) (4 diffs)
• ui/mpiexec/callback.c (modified) (2 diffs)
• ui/mpiexec/mpiexec.c (modified) (4 diffs)
• ui/mpiexec/mpiexec.h (modified) (1 diff)
• ui/mpiexec/utils.c (modified) (8 diffs)
• ui/utils/uiu.c (modified) (1 diff)
• utils/sock/sock.c (modified) (4 diffs)

mpich2/branches/dev/dkim/src/pm/hydra/include/hydra_utils.h

@@ -230 +230 @@
 HYD_Status HYDU_sock_set_nonblock(int fd);
 HYD_Status HYDU_sock_set_cloexec(int fd);
-HYD_Status HYDU_sock_stdout_cb(int fd, HYD_Event_t events, int stdout_fd, int *closed);
-HYD_Status HYDU_sock_stdout_file_cb(int fd, HYD_Event_t events, int stdout_fd, int *closed, int message_length);
-/*HYD_Status HYDU_sock_stderr_file_cb(int fd, HYD_Event_t events, int stdout_fd, int *closed, int message_length);*/
+HYD_Status HYDU_sock_stdout_cb(int fd, HYD_Event_t events, int stdout_fd, int *closed, int message_length);
 HYD_Status HYDU_sock_stdin_cb(int fd, HYD_Event_t events, int stdin_fd, char *buf,
                               int *buf_count, int *buf_offset, int *closed);

mpich2/branches/dev/dkim/src/pm/hydra/pm/pmiserv/pmi_proxy_cb.c

@@ -128 +128 @@
     HYDU_ERR_POP(status, "make header and message error\n");
 
-    if (pmi_header.message_length >= 0)
-        {
-            status = HYD_PMCD_pmi_proxy_write_header(HYD_PMCD_pmi_proxy_params.upstream.out, &pmi_header);
-            HYDU_ERR_POP(status, "header write error\n");
-            status = HYD_PMCD_pmi_proxy_write_message(HYD_PMCD_pmi_proxy_params.upstream.out, buf, &pmi_header);
-            HYDU_ERR_POP(status, "message write error\n");
-        }
+    status = HYD_PMCD_pmi_proxy_write_header(HYD_PMCD_pmi_proxy_params.upstream.out, &pmi_header);
+    HYDU_ERR_POP(status, "header write error\n");
+    
+    status = HYD_PMCD_pmi_proxy_write_message(HYD_PMCD_pmi_proxy_params.upstream.out, buf, &pmi_header);
+    HYDU_ERR_POP(status, "message write error\n");
 
     if (pmi_header.message_length == 0)
         closed = 1;
-
-    /*status = HYDU_sock_stdout_cb(fd, events, HYD_PMCD_pmi_proxy_params.upstream.out, &closed);
-      HYDU_ERR_POP(status, "stdout callback error\n");*/
 
     if (closed) {
@@ -183 +178 @@
     HYDU_ERR_POP(status, "make header and message error\n");
 
-    if (pmi_header.message_length >= 0)
-        {
-            status = HYD_PMCD_pmi_proxy_write_header(HYD_PMCD_pmi_proxy_params.upstream.err, &pmi_header);
-            HYDU_ERR_POP(status, "header write error\n");
-            status = HYD_PMCD_pmi_proxy_write_message(HYD_PMCD_pmi_proxy_params.upstream.err, buf, &pmi_header);
-            HYDU_ERR_POP(status, "message write error\n");
-        }
+    status = HYD_PMCD_pmi_proxy_write_header(HYD_PMCD_pmi_proxy_params.upstream.err, &pmi_header);
+    HYDU_ERR_POP(status, "header write error\n");
+
+    status = HYD_PMCD_pmi_proxy_write_message(HYD_PMCD_pmi_proxy_params.upstream.err, buf, &pmi_header);
+    HYDU_ERR_POP(status, "message write error\n");
 
     if (pmi_header.message_length == 0)
         closed = 1;
-
-    /*status = HYDU_sock_stdout_cb(fd, events, HYD_PMCD_pmi_proxy_params.upstream.err, &closed);
-      HYDU_ERR_POP(status, "stdout callback error\n");*/
 
     if (closed) {

mpich2/branches/dev/dkim/src/pm/hydra/pm/pmiserv/pmi_proxy_utils.c

@@ -671 +671 @@
 
     if (i == HYD_PMCD_pmi_proxy_params.exec_proc_count)
-        pmi_header->pmi_id = -1;
+        HYDU_ERR_SETANDJUMP(status, HYD_INVALID_PARAM, "no matching fd\n");
 
     pmi_header->pmi_id = HYDU_local_to_global_id(i, HYD_PMCD_pmi_proxy_params.partition_core_count, HYD_PMCD_pmi_proxy_params.segment_list, HYD_PMCD_pmi_proxy_params.global_core_count);
@@ -699 +699 @@
 {
     int i, written;
+    const int intsize = sizeof(int);
     HYD_Status status = HYD_SUCCESS;
 
@@ -704 +705 @@
 
     i = 0;
-    while (i != sizeof(int))
+    while (i != intsize)
         {
-            written = write(upstream_fd, (char *)&(pmi_header->pmi_id) + i, sizeof(int) - i);
-
-            if (written < 0 && errno != EAGAIN)
-                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "write error on %d (%s)\n", upstream_fd, HYDU_strerror(errno));
+            written = write(upstream_fd, (char *)&(pmi_header->pmi_id) + i, intsize - i);
 
             if (written > 0)
                 i += written;
+            else if (written < 0 && errno != EAGAIN)
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "write error on %d (%s)\n", upstream_fd, HYDU_strerror(errno));
         }
 
     i = 0;
-    while (i != sizeof(int))
+    while (i != intsize)
         {
-            written = write(upstream_fd, (char *)&(pmi_header->message_length) + i, sizeof(int) - i);
-
-            if (written < 0 && errno != EAGAIN)
-                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "write error on %d (%s)\n", upstream_fd, HYDU_strerror(errno));
+            written = write(upstream_fd, (char *)&(pmi_header->message_length) + i, intsize - i);
 
             if (written > 0)
                 i += written;
+            else if (written < 0 && errno != EAGAIN)
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "write error on %d (%s)\n", upstream_fd, HYDU_strerror(errno));
         }
 
@@ -747 +746 @@
             written = write(upstream_fd, buf + i, pmi_header->message_length - i);
 
-            if (written < 0 && errno != EAGAIN)
+            if (written > 0)
+                {
+                    i +=written;
+                }
+            else if (written < 0 && errno != EAGAIN)
                 {
                     HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "write error on %d (%s)\n", upstream_fd, HYDU_strerror(errno));
                 }
-            else if (written == 0)
-                {
-                    break;
-                }
-            else
-                i += written;
         }
 

mpich2/branches/dev/dkim/src/pm/hydra/ui/mpiexec/callback.c

@@ -50 +50 @@
     HYDU_FUNC_ENTER();
 
-    HYD_UII_mpx_read_header(fd, &rank, &message_length);
+    status = HYD_UII_mpx_read_header(fd, &rank, &message_length);
+    HYDU_ERR_SETANDJUMP2(status, status, "header read error on fd %d: %s\n", fd, HYDU_strerror(errno));
 
     if (rank < 0 || rank > HYD_handle.process_count)
-        {
-            goto fn_fail;
-        }
+        HYDU_ERR_SETANDJUMP2(status, HYD_INVALID_PARAM, "invalid rank %d returned on fd %d\n", rank, fd);
 
-    /* Write output to stdout fd */
+    /* Write output to appropriate stdout fd */
     if (HYD_handle.stdout_file && HYD_handle.multiple_stdout_files)
-        status = HYDU_sock_stdout_file_cb(fd, events, HYD_handle.stdout_file_fds[rank], &closed, message_length);
+        status = HYDU_sock_stdout_cb(fd, events, HYD_handle.stdout_file_fds[rank], &closed, message_length);
     else if (HYD_handle.stdout_file)
-        status = HYDU_sock_stdout_file_cb(fd, events, HYD_handle.stdout_file_fds[0], &closed, message_length);
+        status = HYDU_sock_stdout_cb(fd, events, HYD_handle.stdout_file_fds[0], &closed, message_length);
     else
-        status = HYDU_sock_stdout_file_cb(fd, events, 1, &closed, message_length);
+        status = HYDU_sock_stdout_cb(fd, events, 1, &closed, message_length);
 
     HYDU_ERR_SETANDJUMP2(status, status, "stdout callback error on fd %d: %s\n",
@@ -90 +89 @@
     HYDU_FUNC_ENTER();
 
-    HYD_UII_mpx_read_header(fd, &rank, &message_length);
+    status = HYD_UII_mpx_read_header(fd, &rank, &message_length);
+    HYDU_ERR_SETANDJUMP2(status, status, "header read error on fd %d: %s\n", fd, HYDU_strerror(errno));
 
     if (rank < 0 || rank > HYD_handle.process_count)
-        {
-            goto fn_fail;
-        }
+        HYDU_ERR_SETANDJUMP2(status, HYD_INVALID_PARAM, "invalid rank %d returned on fd %d\n", rank, fd);
 
-    /* Write output to stderr fd */
+    /* Write output to appropriate stderr fd */
     if (HYD_handle.stderr_file && HYD_handle.multiple_stderr_files)
-        status = HYDU_sock_stdout_file_cb(fd, events, HYD_handle.stderr_file_fds[rank], &closed, message_length);
+        status = HYDU_sock_stdout_cb(fd, events, HYD_handle.stderr_file_fds[rank], &closed, message_length);
     else if (HYD_handle.stderr_file)
-        status = HYDU_sock_stdout_file_cb(fd, events, HYD_handle.stderr_file_fds[0], &closed, message_length);
+        status = HYDU_sock_stdout_cb(fd, events, HYD_handle.stderr_file_fds[0], &closed, message_length);
     else
-        status = HYDU_sock_stdout_file_cb(fd, events, 2, &closed, message_length);
+        status = HYDU_sock_stdout_cb(fd, events, 2, &closed, message_length);
 
     HYDU_ERR_SETANDJUMP2(status, status, "stdout callback error on %d (%s)\n",

mpich2/branches/dev/dkim/src/pm/hydra/ui/mpiexec/mpiexec.c

@@ -247 +247 @@
     /* If output from ALL processes are dumped into a single file */
     else if (HYD_handle.stdout_file) {
+        partition = HYD_handle.partition_list;
         HYD_handle.stdout_file_fds = (int *) malloc(sizeof(int));
-        HYD_handle.stdout_file_fds[0] = -1;
+
+        name = make_filename(HYD_handle.stdout_file, partition->base->name, partition->base->partition_id, HYD_handle.process_count);
+        HYD_handle.stdout_file_fds[0] = open(name, O_RDWR | O_CREAT, 0644);
+        HYDU_FREE(name);
     }
 
@@ -271 +275 @@
     /* If error from ALL processes are dumped into a single file */
     else if (HYD_handle.stderr_file) {
+        partition = HYD_handle.partition_list;
         HYD_handle.stderr_file_fds = (int *) malloc(sizeof(int));
-        HYD_handle.stderr_file_fds[0] = -1;
+
+        name = make_filename(HYD_handle.stderr_file, partition->base->name, partition->base->partition_id, HYD_handle.process_count);
+        HYD_handle.stderr_file_fds[0] = open(name, O_RDWR | O_CREAT, 0644);
+        HYDU_FREE(name);
     }
 
     FORALL_ACTIVE_PARTITIONS(partition, HYD_handle.partition_list) {
         if (partition->base->out != -1) {
-            /* If output from ALL processes are dumped into a single file */
-            if (HYD_handle.stdout_file && HYD_handle.stdout_file_fds[0] == -1)
-                {
-                    name = make_filename(HYD_handle.stdout_file, partition->base->name, partition->base->partition_id, HYD_handle.process_count);
-
-                    close(1);
-                    HYD_handle.stdout_file_fds[0] = open(name, O_RDWR | O_CREAT, 0644);
-                    HYDU_FREE(name);
-                }
-
             status = HYD_DMX_register_fd(1, &partition->base->out, HYD_STDOUT, NULL,
                                          HYD_UII_mpx_stdout_cb);
@@ -293 +291 @@
 
         if (partition->base->err != -1) {
-            /* If error from ALL processes are dumped into a single file */
-            if (HYD_handle.stderr_file && HYD_handle.stderr_file_fds[0] == -1)
-                {
-                    name = make_filename(HYD_handle.stderr_file, partition->base->name, partition->base->partition_id, HYD_handle.process_count);
-
-                    close(2);
-                    HYD_handle.stderr_file_fds[0] = open(name, O_RDWR | O_CREAT, 0644);
-                    HYDU_FREE(name);
-                }
-
             status = HYD_DMX_register_fd(1, &partition->base->err, HYD_STDOUT, NULL,
                                          HYD_UII_mpx_stderr_cb);
@@ -346 +334 @@
     HYDU_ERR_POP(status, "process manager error on finalize\n");
 
-    if (HYD_handle.stdout_file_fds)
-        HYDU_FREE(HYD_handle.stdout_file_fds);
-
-    if (HYD_handle.stderr_file_fds)
-        HYDU_FREE(HYD_handle.stderr_file_fds);
-
     /* Free the mpiexec params */
     HYD_UIU_free_params();

mpich2/branches/dev/dkim/src/pm/hydra/ui/mpiexec/mpiexec.h

@@ -15 +15 @@
 HYD_Status HYD_UII_mpx_stdin_cb(int fd, HYD_Event_t events, void *userp);
 char *make_filename(char *input, char *hostname, int rank, int proc_count);
-void HYD_UII_mpx_read_header(int fd, int *rank, int *message_length);
+HYD_Status HYD_UII_mpx_read_header(int fd, int *rank, int *message_length);
 
 #endif /* MPIEXEC_H_INCLUDED */

mpich2/branches/dev/dkim/src/pm/hydra/ui/mpiexec/utils.c

@@ -40 +40 @@
 }
 
+/* parses filename argument and returns the final formatted filename */
 char *make_filename(char *input, char *hostname, int rank, int proc_count)
 {
@@ -47 +48 @@
     char *hostnameptr = NULL;
     char *temp = NULL;
+    const int charsize = sizeof(char);
     HYD_Status status = HYD_SUCCESS;
 
@@ -53 +55 @@
     hostnameptr = strstr(input, "%q");
 
-    /* find how many digits proc_count is */  /* is this necessary? */
+    /* find how many digits proc_count is */
     divten = proc_count / 10;
 
@@ -62 +64 @@
         }
 
+    difference = num_digits - strlen(rankstr);
+
     /* create a string for rank which contains leading zeroes *
-     * ie - 001 instead of just 1 for a job running 100-1000 *
-     * processes.  This makes reading through the list of    *
-     * output files much easier as they are sorted correctly */
-    difference = num_digits - strlen(rankstr);
-
+     * ie - 001 instead of just 1 for a job running 1000      *
+     * processes.  This makes reading through the list of     *
+     * output files much easier as they are sorted correctly  */
     if (difference > 0)
         {
-            temp = (char *) calloc(num_digits + 1, sizeof(char));
+            temp = (char *) calloc(num_digits + 1, charsize);
 
             i = 0;
@@ -92 +94 @@
 
     /* allocate - use calloc to initialize to zero */
-    char *filename = (char *) calloc(size, sizeof(char));
+    char *filename = (char *) calloc(size, charsize);
 
     /* read each character from the input one at a time and put into         *
@@ -101 +103 @@
     while (input[i + l] && i + l < size)
         {
+            /* if current position in array does not match a flag location */
             if (&input[i + l] != rankptr && &input[i + l] != hostnameptr)
                 {
                     filename[i + k] = input[i + l];
                 }
+            /* if current position matches location of '%p' */
             else if (&input[i + l] == rankptr)
                 {
@@ -115 +119 @@
                     l++;
                 }
+            /* current position matches '%q' */
             else
                 {
@@ -133 +138 @@
 }
 
-void HYD_UII_mpx_read_header(int fd, int *rank, int *message_length)
+HYD_Status HYD_UII_mpx_read_header(int fd, int *rank, int *message_length)
 {
     int i, ret;
+    const int intsize = sizeof(int);
+    HYD_Status status = HYD_SUCCESS;
+
+    HYDU_FUNC_ENTER()
 
     i = 0;
-    while (i != sizeof(int))
+    while (i != intsize)
         {
-            ret = read(fd, (char *)rank + i, sizeof(int) - i);
-
-            /*if (ret < 0 && errno != EAGAIN)
-                {
-                    /* ERROR MESSAGE HERE */
-            /*break;*/
-                    /*}*/
+            ret = read(fd, (char *)rank + i, intsize - i);
 
             if (ret > 0)
-                i += ret;
-            else
-                break;
+                {
+                    i += ret;
+                }
+            else if (ret < 0 && errno != EAGAIN)
+                {
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "read error on %d (%s)\n", fd, HYDU_strerror(errno));
+                }
+            else if (ret == 0 && i != intsize)
+                {
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "socket closed before reading %d expected bytes on fd %d\n", intsize, fd);
+                }
         }
 
     i = 0;
-    while (i != sizeof(int))
+    while (i != intsize)
         {
-            ret = read(fd, (char *)message_length + i, sizeof(int) - i);
-
-            /*if (ret < 0 && errno != EAGAIN)
-                {
-                    /* ERROR MESSAGE HERE */
-            /*break;*/
-                    /*}*/
+            ret = read(fd, (char *)message_length + i, intsize - i);
 
             if (ret > 0)
-                i += ret;
-            else
-                break;
+                {
+                    i += ret;
+                }
+            else if (ret < 0 && errno != EAGAIN)
+                {
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "read error on %d (%s)\n", fd, HYDU_strerror(errno));
+                }
+            else if (ret == 0 && i != intsize)
+                {
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "socket closed before reading %d expected bytes on fd %d\n", intsize, fd);
+                }
         }
+
+ fn_exit:
+    HYDU_FUNC_EXIT();
+    return status;
+
+ fn_fail:
+    goto fn_exit;
 }
 

mpich2/branches/dev/dkim/src/pm/hydra/ui/utils/uiu.c

@@ -66 +66 @@
 void HYD_UIU_free_params(void)
 {
+    if (HYD_handle.stdout_file_fds)
+        {
+            if (HYD_handle.multiple_stdout_files)
+                {
+                    int i;
+                    for (i = 0; i < HYD_handle.process_count; i++)
+                        if (HYD_handle.stdout_file_fds[i] != -1)
+                            close(HYD_handle.stdout_file_fds[i]);
+                }
+            else if (HYD_handle.stdout_file_fds[0] != -1)
+                close(HYD_handle.stdout_file_fds[0]);
+                                  
+            HYDU_FREE(HYD_handle.stdout_file_fds);
+        }
+
+    if (HYD_handle.stderr_file_fds)
+        {
+            if (HYD_handle.multiple_stderr_files)
+                {
+                    int i;
+                    for (i = 0; i < HYD_handle.process_count; i++)
+                        if (HYD_handle.stderr_file_fds[i] != -1)
+                            close(HYD_handle.stderr_file_fds[i]);
+                }
+            else if (HYD_handle.stderr_file_fds[0] != -1)
+                close(HYD_handle.stderr_file_fds[0]);
+                                  
+            HYDU_FREE(HYD_handle.stderr_file_fds);
+        }
+
     if (HYD_handle.stdout_file)
         HYDU_FREE(HYD_handle.stdout_file);

mpich2/branches/dev/dkim/src/pm/hydra/utils/sock/sock.c

@@ -376 +376 @@
 }
 
-
-HYD_Status HYDU_sock_stdout_cb(int fd, HYD_Event_t events, int stdout_fd, int *closed)
+HYD_Status HYDU_sock_stdout_cb(int fd, HYD_Event_t events, int stdout_fd, int *closed, int message_length)
 {
     int count, written, ret;
@@ -390 +389 @@
         HYDU_ERR_SETANDJUMP(status, HYD_INTERNAL_ERROR, "stdout handler got stdin event\n");
 
-    count = read(fd, buf, HYD_TMPBUF_SIZE);
-    if (count < 0) {
-        HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "read error on %d (%s)\n",
-                             fd, HYDU_strerror(errno));
-    }
-    else if (count == 0) {
-        /* The connection has closed */
-        *closed = 1;
-        goto fn_exit;
-    }
-
-    written = 0;
-    while (written != count) {
-        ret = write(stdout_fd, buf + written, count - written);
-        if (ret < 0 && errno != EAGAIN)
-            HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "write error on %d (%s)\n",
-                                 stdout_fd, HYDU_strerror(errno));
-        if (ret > 0)
-            written += ret;
-    }
-
-  fn_exit:
-    HYDU_FUNC_EXIT();
-    return status;
-
-  fn_fail:
-    goto fn_exit;
-}
-
-HYD_Status HYDU_sock_stdout_file_cb(int fd, HYD_Event_t events, int stdout_fd, int *closed, int message_length)
-{
-    int count, written, ret;
-    char buf[HYD_TMPBUF_SIZE];
-    HYD_Status status = HYD_SUCCESS;
-
-    HYDU_FUNC_ENTER();
-
-    *closed = 0;
-
-    if (events & HYD_STDIN)
-        HYDU_ERR_SETANDJUMP(status, HYD_INTERNAL_ERROR, "stdout handler got stdin event\n");
-
     if (stdout_fd < 0)
-        goto fn_exit;
+        HYDU_ERR_SETANDJUMP1(status, HYD_INVALID_PARAM, "callback received invalid output fd %d\n", stdout_fd);
 
     /* Connection has closed */
+    /* Knowing the connection has closed is no longer given by a read *
+     * and checking if the read returned 0.  Headers are ALWAYS sent  *
+     * and ALWAYS received.  message_length ensures messages/headers  *
+     * are not cannabalized by other reads.                           */
     if (message_length == -1)
         {
@@ -442 +403 @@
         }
 
+    /* Header should have already been read at this point.  We now *
+     * read for the message.                                       */
     count = 0;
     do
@@ -453 +416 @@
             else if (ret < 0 && errno != EAGAIN)
                 {
-                    *closed = 1;
-                    break;
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "read error on %d (%s)\n", fd, HYDU_strerror(errno));
+                }
+            else if (ret == 0 && count != message_length)
+                {
+                    HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "socket closed before reading %d expected bytes on fd %d\n", message_length, fd);
                 }
         } while(count != message_length);
 
-    if (count < 0) {
-        HYDU_ERR_SETANDJUMP2(status, HYD_SOCK_ERROR, "read error on %d (%s)\n",
-                             fd, HYDU_strerror(errno));
-    }
-
+
+    /* Write the message out.  If message_length == 0, nothing is written */
     written = 0;
     while (written != count) {