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

Timestamp:

Mar 1, 2017, 2:36:10 PM (8 years ago)

Author:

Bayard Gildas

Message:

SharedMem in producerConsumer mode

File:

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

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