Changeset 203 in flair-src for trunk/lib/FlairCore/src/SharedMem_impl.cpp

Timestamp:

Nov 9, 2017, 2:00:49 PM (7 years ago)

Author:

Bayard Gildas

Message:

Added timeout on SharedMem Read

File:

• trunk/lib/FlairCore/src/SharedMem_impl.cpp (modified) (6 diffs)

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;
 }