Changeset 203 in flair-src

Timestamp:

11/09/17 14:00:49 (6 years ago)

Author:

Bayard Gildas

Message:

Added timeout on SharedMem Read

Location:

trunk/lib/FlairCore/src

Files:

• Semaphore.cpp (modified) (1 diff)
• Semaphore.h (modified) (1 diff)
• Semaphore_impl.cpp (modified) (2 diffs)
• SharedMem.cpp (modified) (1 diff)
• SharedMem.h (modified) (2 diffs)
• SharedMem_impl.cpp (modified) (6 diffs)
• unexported/Semaphore_impl.h (modified) (1 diff)
• unexported/SharedMem_impl.h (modified) (3 diffs)

trunk/lib/FlairCore/src/Semaphore.cpp

@@ -31 +31 @@
 Semaphore::~Semaphore() { delete pimpl_; }
 
+bool Semaphore::TryGetSemaphore() const { return pimpl_->TryGetSemaphore(); }
 bool Semaphore::GetSemaphore(Time timeout) const { return pimpl_->GetSemaphore(timeout); }
 

trunk/lib/FlairCore/src/Semaphore.h

@@ -48 +48 @@
 
   /*!
+  * \brief TryGetSemaphore
+  *
+  * Lock the semaphore. If not possible immediately returns false
+  *
+  */
+  bool TryGetSemaphore() const;
+
+  /*!
   * \brief GetSemaphore
   *

trunk/lib/FlairCore/src/Semaphore_impl.cpp

@@ -53 +53 @@
 }
 
+bool Semaphore_impl::TryGetSemaphore() {
+#ifdef __XENO__
+  return !rt_sem_p(&semaphore, TM_NONBLOCK);
+#else
+  return !sem_trywait(&semaphore);
+#endif
+}
+
 bool Semaphore_impl::GetSemaphore(Time timeout) {
   int status;
@@ -75 +83 @@
         if (status != 0) {
                 if (errno == ETIMEDOUT) {
-                        self->Warn("warning : semaphore timedout\n");   
+#ifdef DEBUG
+                        self->Warn("warning : semaphore timedout\n");
+#endif
                 } else {
                                         char errorMsg[256];

trunk/lib/FlairCore/src/SharedMem.cpp

@@ -35 +35 @@
 }
 
-void SharedMem::Read(char *buf, size_t size) const {
-  pimpl_->Read(buf, size);
+bool SharedMem::Read(char *buf, size_t size, Time nsTimeout) const {
+  return pimpl_->Read(buf, size, nsTimeout);
 }
 

trunk/lib/FlairCore/src/SharedMem.h

@@ -27 +27 @@
 *
 * Shared memory is identified by its name so it can be accessed
-* by another processus using its name.
+* by another process using its name.
 */
 
@@ -65 +65 @@
   * \param size buffer size
   */
-  void Read(char *buf, size_t size) const;
+  bool Read(char *buf, size_t size, Time nsTimeout=TIME_INFINITE) const;
 
 
   /*!
-  * \brief This function should be called when reader starts (in case of a SharedMem of type producerConsumer)
+  * \brief This function should be called when reader is ready (in case of a SharedMem of type producerConsumer)
   */
    void ReaderReady();

trunk/lib/FlairCore/src/SharedMem_impl.cpp

@@ -26 +26 @@
 using namespace flair::core;
 
-SharedMem_impl::SharedMem_impl(const SharedMem *self, string name,
-                               size_t size, SharedMem::Type &type):type(type) {
-  this->size = size;
-  this->self = self;
-        char errorMsg[256];
+SharedMem_impl::SharedMem_impl(const SharedMem *self, string name, size_t size, SharedMem::Type &type):self(self),type(type),size(size),
+                               sem(self,1,"/" + name + "_mutex"), sem_producer(self,0,"/" + name + "_producer"),sem_consumer(self,0,"/" + name + "_consumer") {
+
+  char errorMsg[256];
 
 #ifdef __XENO__
@@ -52 +51 @@
   mem_segment = (char *)ptr;
 
-  mutex_binded = false;
-  string mutex_name = "mutex_" + name;
-  status = rt_mutex_create(&mutex, mutex_name.c_str());
-  if (status == -EEXIST) {
-    mutex_binded = true;
-    status = rt_mutex_bind(&mutex, mutex_name.c_str(), TM_INFINITE);
-  }
-  if (status != 0) {
-    self->Err("rt_mutex_create error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
-    return;
-  }
-
 #else
   shm_name = "/" + name;
@@ -72 +59 @@
   ftruncate(fd, size);
 
-  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");
-    }
-    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");
-    }
-  }
-
-  mem_segment =
-      (char *)mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
+  mem_segment = (char *)mmap(NULL, size, PROT_READ | PROT_WRITE, MAP_SHARED, fd, 0);
   if (mem_segment == MAP_FAILED) {
     self->Err("Failed to map memory\n");
@@ -104 +71 @@
         
 #ifdef __XENO__
-  /* unnecessary because heap is opened in H_SINGLE mode
-  status=rt_heap_free(&heap,mem_segment);
-  if(status!=0)
-  {
-      self->Err("rt_heap_free error (%s)\n",strerror_r(-status, errorMsg, sizeof(errorMsg)));
-  }
-  */
 
   if (heap_binded == false) {
@@ -119 +79 @@
   }
 
-  if (mutex_binded == false) {
-    status = rt_mutex_delete(&mutex);
-    if (status != 0) {
-      self->Err("error destroying mutex (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
-    }
-  }
 #else
   status = munmap(mem_segment, size);
@@ -143 +97 @@
    }
 */
-  // 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
+  sem_consumer.ReleaseSemaphore();
 }
 
 void SharedMem_impl::Write(const char *buf, size_t size) {
-#ifdef __XENO__
-  int status = rt_mutex_acquire(&mutex, TM_INFINITE);
-  if (status != 0) {
-                char errorMsg[256];
-    self->Err("error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
-        }
-  memcpy(mem_segment, buf, size);
-  status = rt_mutex_release(&mutex);
-  if (status != 0) {
-                char errorMsg[256];
-    self->Err("error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
-        }
-#else
   if (type==SharedMem::Type::mutex) {
-    sem_wait(sem);
+    sem.GetSemaphore();
     memcpy(mem_segment, buf, size);
-    sem_post(sem);
+    sem.ReleaseSemaphore();
   } 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) {
+    if (sem_consumer.TryGetSemaphore()) {
       memcpy(mem_segment, buf, size);
-      sem_post(sem_producer);
+      sem_producer.ReleaseSemaphore();
     }
   }
-#endif
 }
 
-void SharedMem_impl::Read(char *buf, size_t size) {
-#ifdef __XENO__
-  int status = rt_mutex_acquire(&mutex, TM_INFINITE);
-  if (status != 0) {
-                char errorMsg[256];
-    self->Err("error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
-        }
-  memcpy(buf, mem_segment, size);
-  status = rt_mutex_release(&mutex);
-  if (status != 0) {
-                char errorMsg[256];
-    self->Err("error (%s)\n", strerror_r(-status, errorMsg, sizeof(errorMsg)));
-        }
-#else
+bool SharedMem_impl::Read(char *buf, size_t size, Time nsTimeout) {
   if (type==SharedMem::Type::mutex) {
-    sem_wait(sem);
-    memcpy(buf, mem_segment, size);
-    sem_post(sem);
+    if (sem.GetSemaphore()) {
+      memcpy(buf, mem_segment, size);
+      sem.ReleaseSemaphore();
+      return true;
+    }
   } 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);
+      if (sem_producer.GetSemaphore(nsTimeout)) {
+        memcpy(buf, mem_segment, size);
+        sem_consumer.ReleaseSemaphore();
+      return true;
+    }
   }
-#endif
+  return false;
 }

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

@@ -31 +31 @@
   Semaphore_impl(flair::core::Semaphore *self, uint32_t initialValue);
   ~Semaphore_impl();
+  bool TryGetSemaphore();
   bool GetSemaphore(flair::core::Time timeout = TIME_INFINITE);
   bool ReleaseSemaphore(void);

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

@@ -18 +18 @@
 #ifdef __XENO__
 #include <native/heap.h>
-#include <native/mutex.h>
-#else
-#include <semaphore.h>
 #endif
+
+#include <Semaphore.h>
 
 #include <SharedMem.h>
@@ -37 +36 @@
 
   void Write(const char *buf, size_t size);
-  void Read(char *buf, size_t size);
+  bool Read(char *buf, size_t size, flair::core::Time nsTimeout);
   void ReaderReady();
 
@@ -45 +44 @@
   size_t size;
   char *mem_segment;
+  flair::core::Semaphore sem,sem_producer,sem_consumer;
 #ifdef __XENO__
   RT_HEAP heap;
-  RT_MUTEX mutex;
   bool heap_binded;
-  bool mutex_binded;
 #else
   int fd;
-  sem_t *sem,*sem_producer,*sem_consumer;
-  std::string sem_name, shm_name;
-  std::string sem_name_producer,sem_name_consumer;
+  std::string shm_name;
 #endif
 };