API Addition: Added pgm_init_process_local()

Added new PGM init API pgm_init_process_local(). This
initializes the PGM^RT backend to support only process-local
graphs. The PGM^RT runtime does not need to create any
directories or files when initialized into process-local mode.
However, the user cannot use FIFO-based edges in this mode.

3 files changed, 115 insertions, 62 deletions

diff --git a/include/pgm.h b/include/pgm.h
index d570faf..0db7e01 100644
--- a/include/pgm.h
+++ b/include/pgm.h
@@ -126,6 +126,17 @@ static const edge_attr_t default_edge = {
 };
 
 /*
+   Initialize the PGM runtime in the application for cases where
+   graph nodes are always (1) process-local (threads within the same
+   process) and (2) FIFO-based edges are never used.
+
+   Use pgm_init() if nodes are cross-process, or FIFO edges
+   are needed. If just FIFO edges are needed and nodes are not
+   cross-process, pass "use_shared_mem=0" to pgm_init().
+*/
+int pgm_init_process_local(void);
+
+/*
    Initialize the PGM runtime in the application.
      [in] dir: Directory where PGM data is stored.
      [in] create:
diff --git a/src/pgm.cpp b/src/pgm.cpp
index 8057c36..a9816ea 100644
--- a/src/pgm.cpp
+++ b/src/pgm.cpp
@@ -46,9 +46,7 @@ using namespace boost::filesystem;
 
 
 // TODO LIST:
-//  * In-memory buffers that can be passed from producer to consumer
-//    without any copies.
-//      BONUS: Shared memory support.
+//  * Ring buffers in shared memory.
 
 #ifndef PGM_CONFIG
 #error "pgm/include/config.h not included!"
@@ -420,19 +418,28 @@ static int fifo_create(pgm_graph* g,
                                 pgm_edge* edge)
 {
         int ret = -1;
-        string fifoName(fifo_name(g, producer, consumer, edge));
 
-        path fifoPath(gGraphPath);
-        fifoPath /= fifoName;
+        if(gGraphPath.string().empty())
+        {
+                F("Graph directory not set. Did you call pgm_init_process_local() "
+                  "insted of pgm_init()?\n");
+        }
+        else
+        {
+                string fifoName(fifo_name(g, producer, consumer, edge));
 
-        // Remove any old FIFO that may exist.
-        remove(fifoPath);
+                path fifoPath(gGraphPath);
+                fifoPath /= fifoName;
 
-        // TODO: See what boost can do here.
-        ret = mkfifo(fifoPath.string().c_str(), S_IRUSR | S_IWUSR);
-        if(0 != ret)
-        {
-                F("Failed to make FIFO %s\n", fifoPath.string().c_str());
+                // Remove any old FIFO that may exist.
+                remove(fifoPath);
+
+                // TODO: See what boost can do here.
+                ret = mkfifo(fifoPath.string().c_str(), S_IRUSR | S_IWUSR);
+                if(0 != ret)
+                {
+                        F("Failed to make FIFO %s\n", fifoPath.string().c_str());
+                }
         }
         return ret;
 }
@@ -441,18 +448,28 @@ static int fifo_open_consumer(pgm_graph* g,
                                 pgm_node* producer, pgm_node* consumer,
                                 pgm_edge* edge)
 {
-        path fifoPath(gGraphPath);
-        fifoPath /= fifo_name(g, producer, consumer, edge);
-        edge->fd_in = open(fifoPath.string().c_str(), O_RDONLY | O_NONBLOCK);
-        if(edge->fd_in == -1)
+        int ret = -1;
+        if(gGraphPath.string().empty())
         {
-                F("Could not open inbound edge %s/%s (FIFO)\n", g->name, edge->name);
-                return -1;
+                F("Graph directory not set. Did you call pgm_init_process_local() "
+                  "insted of pgm_init()?\n");
         }
+        else
+        {
+                path fifoPath(gGraphPath);
+                fifoPath /= fifo_name(g, producer, consumer, edge);
+                edge->fd_in = open(fifoPath.string().c_str(), O_RDONLY | O_NONBLOCK);
 
-        edge->buf_in = __pgm_malloc_edge_buf(g, edge, false);
-
-        return 0;
+                if(edge->fd_in != -1)
+                {
+                        edge->buf_in = __pgm_malloc_edge_buf(g, edge, false);
+                }
+                else
+                {
+                        F("Could not open inbound edge %s/%s (FIFO)\n", g->name, edge->name);
+                }
+        }
+        return ret;
 }
 
 static int fifo_open_producer(pgm_graph* g,
@@ -460,41 +477,54 @@ static int fifo_open_producer(pgm_graph* g,
                                 pgm_edge* edge)
 {
         int ret = -1;
-        path fifoPath(gGraphPath);
-        fifoPath /= fifo_name(g, producer, consumer, edge);
 
-        const int timeout = 60;
-        const int start_time = time(0);
-        __sync_synchronize();
-        do
+        if(gGraphPath.string().empty())
         {
-                edge->fd_out = open(fifoPath.string().c_str(), O_WRONLY | O_NONBLOCK);
-                if(edge->fd_out == -1)
+                F("Graph directory not set. Did you call pgm_init_process_local() "
+                  "insted of pgm_init()?\n");
+        }
+        else
+        {
+                path fifoPath(gGraphPath);
+                fifoPath /= fifo_name(g, producer, consumer, edge);
+
+                const int timeout = 60;
+                const int start_time = time(0);
+                __sync_synchronize();
+                do
                 {
-                        if(errno != ENXIO)
-                        {
-                                F("Could not open outbound edge %s/%s (FIFO)\n", g->name, edge->name);
-                                break;
-                        }
-                        else
+                        edge->fd_out = open(fifoPath.string().c_str(),
+                                O_WRONLY | O_NONBLOCK);
+                        if(edge->fd_out == -1)
                         {
-                                if(time(0) - start_time > timeout)
+                                if(errno != ENXIO)
                                 {
-                                        F("Could not open outbound edge %s/%s (FIFO)\n", g->name, edge->name);
+                                        F("Could not open outbound edge %s/%s (FIFO)\n",
+                                          g->name, edge->name);
                                         break;
                                 }
-                                usleep(1000); // wait for a millisecond
+                                else
+                                {
+                                        if(time(0) - start_time > timeout)
+                                        {
+                                                F("Could not open outbound edge %s/%s (FIFO)\n",
+                                                   g->name, edge->name);
+                                                break;
+                                        }
+                                        usleep(1000); // wait for a millisecond
+                                }
                         }
-                }
-                else
+                        else
+                        {
+                                ret = 0;
+                        }
+                }while(ret == -1);
+
+                if(!ret)
                 {
-                        ret = 0;
+                        edge->buf_out = __pgm_malloc_edge_buf(g, edge, true);
                 }
-        }while(ret == -1);
-
-        if(!ret)
-                edge->buf_out = __pgm_malloc_edge_buf(g, edge, true);
-
+        }
         return ret;
 }
 
@@ -527,26 +557,32 @@ static int fifo_destroy(pgm_graph* g,
                                 pgm_edge* edge)
 {
         int ret = -1;
-        string fifoName(fifo_name(g, producer, consumer, edge));
 
-        path fifoPath(gGraphPath);
-        fifoPath /= fifoName;
-
-        if(edge->buf_in != 0 || edge->buf_out != 0)
+        if(gGraphPath.string().empty())
         {
-                W("Edge has not been closed: (producer:%s, consumer:%s)!\n",
-                        (edge->buf_out != 0) ? "open" : "closed",
-                        (edge->buf_in != 0) ? "open" : "closed");
+                F("Graph directory not set. Did you call pgm_init_process_local() "
+                  "insted of pgm_init()?\n");
         }
+        else
+        {
+                string fifoName(fifo_name(g, producer, consumer, edge));
+                path fifoPath(gGraphPath);
+                fifoPath /= fifoName;
 
-        if(!exists(fifoPath))
-                goto out;
-        if(!remove(fifoPath))
-                goto out;
+                if(edge->buf_in != 0 || edge->buf_out != 0)
+                {
+                        W("Edge has not been closed: (producer:%s, consumer:%s)!\n",
+                                (edge->buf_out != 0) ? "open" : "closed",
+                                (edge->buf_in != 0) ? "open" : "closed");
+                }
 
-        ret = 0;
+                if(exists(fifoPath) &&
+                   remove(fifoPath))
+                {
+                        ret = 0;
+                }
+        }
 
-out:
         return ret;
 }
 
@@ -1569,6 +1605,12 @@ static int prepare_graph_private_mem(void)
         return ret;
 }
 
+int pgm_init_process_local(void)
+{
+        int ret = prepare_graph_private_mem();
+        return ret;
+}
+
 int pgm_init(const char* dir, int create, int use_shared_mem)
 {
         int ret = -1;
diff --git a/tools/ringtest.cpp b/tools/ringtest.cpp
index d6850ed..3f2bbb5 100644
--- a/tools/ringtest.cpp
+++ b/tools/ringtest.cpp
@@ -130,7 +130,7 @@ int main(void)
         ring_attr.nr_threshold = sizeof(uint32_t);
         ring_attr.nmemb = 32;
 
-        CheckError(pgm_init("/tmp/graphs", 1));
+        CheckError(pgm_init_process_local());
         CheckError(pgm_init_graph(&g, "demo"));
 
         CheckError(pgm_init_node(&n0, g, "n0"));