Changes to support mmap reads on Broker

src/modules/dyad.c

@@ -159,16 +159,10 @@ dyad_fetch_request_cb (flux_t *h, flux_msg_handler_t *w, const flux_msg_t *msg,
 #endif  // DYAD_SPIN_WAIT
 
     FLUX_LOG_INFO (h, "Reading file %s for transfer", fullpath);
-    fd = open (fullpath, O_RDONLY);
-    if (fd < 0) {
-        FLUX_LOG_ERR (h, "DYAD_MOD: Failed to open file \"%s\".\n", fullpath);
-        goto fetch_error;
-    }
-    if ((inlen = read_all (fd, &inbuf)) < 0) {
+    if ((inlen = read_all (fullpath, &inbuf)) < 0) {
         FLUX_LOG_ERR (h, "DYAD_MOD: Failed to load file \"%s\".\n", fullpath);
         goto fetch_error;
     }
-    close (fd);
     FLUX_LOG_INFO (h, "Is inbuf NULL? -> %i\n", (int)(inbuf == NULL));
 
     FLUX_LOG_INFO (h, "Establish DTL connection with consumer");
@@ -188,7 +182,7 @@ dyad_fetch_request_cb (flux_t *h, flux_msg_handler_t *w, const flux_msg_t *msg,
         errno = ECOMM;
         goto fetch_error;
     }
-
+    munmap(inbuf, inlen);
     FLUX_LOG_INFO (h, "Close RPC message stream with an ENODATA (%d) message", ENODATA);
     if (flux_respond_error (h, msg, ENODATA, NULL) < 0) {
         FLUX_LOG_ERR (h,

src/utils/read_all.c

@@ -11,13 +11,22 @@
 #if HAVE_CONFIG_H
 #include "config.h"
 #endif  // HAVE_CONFIG_H
-
+#define _GNU_SOURCE
 #include <errno.h>
 #include <stdlib.h>
+#include <sys/types.h>
+#include <sys/stat.h>
 #include <unistd.h>
-
+#include <fcntl.h>
+#include <sys/mman.h>
 #include "read_all.h"
 
+void * allocate_aligned(size_t size) {
+  size_t alignment = getpagesize();
+  int malloc_offset = (size + alignment) % alignment; 
+  void* mem = malloc(size + alignment);
+  return (char*) mem + malloc_offset;
+}
 ssize_t write_all (int fd, const void *buf, size_t len)
 {
     ssize_t n = 0;
@@ -37,8 +46,13 @@ ssize_t write_all (int fd, const void *buf, size_t len)
     return count;
 }
 
-ssize_t read_all (int fd, void **bufp)
+ssize_t read_all (const char* filename, void **bufp)
 {
+    int fd = open (filename, O_RDONLY);
+    if (fd < 0) {
+        errno = EINVAL;
+        return -1;
+    }
     const ssize_t file_size = lseek (fd, 0, SEEK_END);
     if (file_size == 0) {
         errno = EINVAL;
@@ -49,16 +63,14 @@ ssize_t read_all (int fd, void **bufp)
         errno = EINVAL;
         return -1;
     }
-    *bufp = malloc (file_size);
+    *bufp = allocate_aligned (file_size);
     if (*bufp == NULL) {
         errno = EINVAL;
         return -1;
     }
-    ssize_t bytes_read = read (fd, *bufp, file_size);
-    if (bytes_read < file_size) {
-        // could not read all data
+    if ((*bufp = mmap (*bufp, file_size, PROT_READ, MAP_PRIVATE | MAP_FIXED, fd, 0)) == (caddr_t) -1) {
         errno = EINVAL;
-        return bytes_read;
+        return -1;
     }
     return file_size;
 }

src/utils/read_all.h

@@ -23,7 +23,7 @@ ssize_t write_all (int fd, const void *buf, size_t len);
 __attribute__ ((annotate ("@critical_path()")))
 #endif
 ssize_t
-read_all (int fd, void **bufp);
+read_all (const char* filename, void **bufp);
 
 #if defined(__cplusplus)
 };