Changeset 149 in flair-src for trunk

Timestamp:

03/01/17 14:36:10 (7 years ago)

Author:

Bayard Gildas

Message:

SharedMem in producerConsumer mode

Location:

trunk/lib/FlairCore/src

Files:

• SharedMem.cpp (modified) (2 diffs)
• SharedMem.h (modified) (3 diffs)
• SharedMem_impl.cpp (modified) (6 diffs)
• unexported/SharedMem_impl.h (modified) (3 diffs)

trunk/lib/FlairCore/src/SharedMem.cpp

@@ -24 +24 @@
 namespace core {
 
-SharedMem::SharedMem(const Object *parent, string name, size_t size)
-    : Object(parent, name, "shmem") {
-  pimpl_ = new SharedMem_impl(this, name, size);
+SharedMem::SharedMem(const Object *parent, string name, size_t size, Type type)
+    : Object(parent, name, "shmem"), type(type) {
+  pimpl_ = new SharedMem_impl(this, name, size, type);
 }
 
@@ -35 +35 @@
 }
 
-void SharedMem::Read(char *buf, size_t size) const { pimpl_->Read(buf, size); }
+void SharedMem::Read(char *buf, size_t size) const {
+  pimpl_->Read(buf, size);
+}
+
+void SharedMem::ReaderReady() {
+  if (type==Type::producerConsumer) pimpl_->ReaderReady();
+  else Warn("Function called for a non producerConsumer\n");
+}
 
 } // end namespace core

trunk/lib/FlairCore/src/SharedMem.h

@@ -32 +32 @@
 class SharedMem : public Object {
 public:
+  enum class Type { mutex, producerConsumer };
   /*!
   * \brief Constructor
@@ -40 +41 @@
   * \param name name
   * \param size size of the shared memory
+  * \param blockOnRead if true reading will block if nothing written
   */
-  SharedMem(const Object *parent, std::string name, size_t size);
+  SharedMem(const Object *parent, std::string name, size_t size, Type type=Type::mutex);
 
   /*!
@@ -65 +67 @@
   void Read(char *buf, size_t size) const;
 
+
+  /*!
+  * \brief This function should be called when reader starts (in case of a SharedMem of type producerConsumer)
+  */
+   void ReaderReady();
 private:
   SharedMem_impl *pimpl_;
+  Type type;
 };
 

trunk/lib/FlairCore/src/SharedMem_impl.cpp

@@ -27 +27 @@
 
 SharedMem_impl::SharedMem_impl(const SharedMem *self, string name,
-                               size_t size) {
+                               size_t size, SharedMem::Type &type):type(type) {
   this->size = size;
   this->self = self;
@@ -72 +72 @@
   ftruncate(fd, size);
 
-  sem_name = "/" + name;
-  sem = sem_open(sem_name.c_str(), O_CREAT, 0666, 1);
-  if (sem == SEM_FAILED) {
-    self->Err("Error creating semaphore\n");
+  if (type==SharedMem::Type::mutex) {
+    sem_name="/" + name + "_mutex";
+    sem=sem_open(sem_name.c_str(), O_CREAT, 0666, 1);
+    if (sem==SEM_FAILED) {
+      self->Err("Error creating mutex semaphore\n");
+    }
+  } else if (type==SharedMem::Type::producerConsumer) {
+    sem_name_producer="/" + name + "_producer";
+    sem_producer=sem_open(sem_name_producer.c_str(), O_CREAT, 0666, 0);
+    if (sem==SEM_FAILED) {
+      self->Err("Error creating producer semaphore\n");
+    }
+Printf("semaphore %s opened\n",sem_name_producer.c_str());
+    sem_name_consumer="/" + name + "_consumer";
+    sem_consumer=sem_open(sem_name_consumer.c_str(), O_CREAT, 0666, 0);
+    if (sem==SEM_FAILED) {
+      self->Err("Error creating consumer semaphore\n");
+    }
+Printf("semaphore %s opened\n",sem_name_consumer.c_str());
   }
 
@@ -123 +138 @@
   }
 
-  // do not check erros as it can be done by another process
+  // do not check errors as it can be done by another process
   status = shm_unlink(shm_name.c_str()); /*
    if(status!=0)
@@ -130 +145 @@
    }
 */
-  // do not check erros as it can be done by another process
-  status = sem_unlink(sem_name.c_str()); /*
-   if(status!=0)
-   {
-       self->Err("Failed to unlink semaphore (%s)\n",strerror_r(-status, errorMsg, sizeof(errorMsg)));
-   }*/
-
-  status = sem_close(sem);
-  if (status != 0) {
-    self->Err("Failed to close semaphore (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
-  }
+  // do not check errors as it can be done by another process
+  if (type==SharedMem::Type::mutex) {
+    status = sem_unlink(sem_name.c_str());
+    status = sem_close(sem);
+    if (status != 0) {
+      self->Err("Failed to close mutex semaphore (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
+    }
+  } else if (type==SharedMem::Type::producerConsumer) {
+    status = sem_unlink(sem_name_producer.c_str());
+    status = sem_close(sem_producer);
+    if (status != 0) {
+      self->Err("Failed to close producer semaphore (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
+    }
+    status = sem_unlink(sem_name_consumer.c_str());
+    status = sem_close(sem_consumer);
+    if (status != 0) {
+      self->Err("Failed to close consumer semaphore (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
+    }
+  }
+#endif
+}
+
+void SharedMem_impl::ReaderReady() {
+#ifdef __XENO__
+//TODO
+#else
+  sem_post(sem_consumer);
 #endif
 }
@@ -158 +189 @@
         }
 #else
-  sem_wait(sem);
-  memcpy(mem_segment, buf, size);
-  sem_post(sem);
+  if (type==SharedMem::Type::mutex) {
+    sem_wait(sem);
+    memcpy(mem_segment, buf, size);
+    sem_post(sem);
+  } else if (type==SharedMem::Type::producerConsumer) {
+    //wait until consumer took the data away before writting a new one
+    //but should not block if nobody's there to read
+    if (sem_trywait(sem_consumer)==0) {
+      memcpy(mem_segment, buf, size);
+      sem_post(sem_producer);
+    }
+  }
 #endif
 }
@@ -178 +218 @@
         }
 #else
-  sem_wait(sem);
-  memcpy(buf, mem_segment, size);
-  sem_post(sem);
-#endif
-}
+  if (type==SharedMem::Type::mutex) {
+    sem_wait(sem);
+    memcpy(buf, mem_segment, size);
+    sem_post(sem);
+  } else if (type==SharedMem::Type::producerConsumer) {
+    sem_wait(sem_producer); //wait until some data is available
+    memcpy(buf, mem_segment, size);
+    sem_post(sem_consumer);
+  }
+#endif
+}

trunk/lib/FlairCore/src/unexported/SharedMem_impl.h

@@ -23 +23 @@
 #endif
 
-namespace flair {
-namespace core {
-class SharedMem;
-}
-}
+#include <SharedMem.h>
 
 /*! \class SharedMem_impl
@@ -37 +33 @@
 public:
   SharedMem_impl(const flair::core::SharedMem *self, std::string name,
-                 size_t size);
+                 size_t size, flair::core::SharedMem::Type &type);
   ~SharedMem_impl();
 
   void Write(const char *buf, size_t size);
   void Read(char *buf, size_t size);
+  void ReaderReady();
 
 private:
   const flair::core::SharedMem *self;
+  flair::core::SharedMem::Type type;
   size_t size;
   char *mem_segment;
@@ -54 +52 @@
 #else
   int fd;
-  sem_t *sem;
+  sem_t *sem,*sem_producer,*sem_consumer;
   std::string sem_name, shm_name;
+  std::string sem_name_producer,sem_name_consumer;
 #endif
 };