fix: serialize shared replay fd map updates (task 478)

4 files changed, 16 insertions, 3 deletions

diff --git a/ioriot/src/replay/rioop.c b/ioriot/src/replay/rioop.c
index b5c360e..daaa7af 100644
--- a/ioriot/src/replay/rioop.c
+++ b/ioriot/src/replay/rioop.c
@@ -273,8 +273,11 @@ void rioop_open(rprocess_s *p, rthread_s *t, rtask_s *task, int flags_)
     }
 
     if (fd > 0 && ret > 0) {
+        rworker_s *w = t->worker;
         vfd_s *vfd = vfd_new(ret, fd, path);
+        pthread_mutex_lock(&w->fds_map_mutex);
         amap_set(p->fds_map, fd, vfd);
+        pthread_mutex_unlock(&w->fds_map_mutex);
 
 #ifdef THREAD_DEBUG
         fprintf(t->rthread_fd, "TRACE OPEN|open|%s|\n", path);
@@ -286,9 +289,16 @@ void rioop_open(rprocess_s *p, rthread_s *t, rtask_s *task, int flags_)
 void rioop_close(rprocess_s *p, rthread_s *t, rtask_s *task)
 {
     _Init_fd(3);
-    _Init_virtfd;
+    rworker_s *w = t->worker;
+    // Lookup lifetime still relies on the per-FD ownership invariant.
+    pthread_mutex_lock(&w->fds_map_mutex);
+    vfd_s *vfd = amap_get(p->fds_map, fd);
+    if (vfd != NULL)
+        amap_unset(p->fds_map, fd);
+    pthread_mutex_unlock(&w->fds_map_mutex);
+    if (vfd == NULL)
+        return;
 
-    amap_unset(p->fds_map, fd);
     if (vfd->dirfd) {
         closedir(vfd->dirfd);
 #ifdef THREAD_DEBUG
diff --git a/ioriot/src/replay/rprocess.h b/ioriot/src/replay/rprocess.h
index adba0f3..e8eaaa1 100644
--- a/ioriot/src/replay/rprocess.h
+++ b/ioriot/src/replay/rprocess.h
@@ -31,7 +31,7 @@ typedef struct rprocess_s_ {
     int pid; /**< The virtual process ID */
     unsigned long lineno; /**< Holding the current .replay line number */
     bool initm; /**< Indicates whether ioriot is in init mode or not */
-    amap_s *fds_map; /**< Holding all file descriptors */
+    amap_s *fds_map; /**< Shared worker-wide file descriptors map */
 } rprocess_s;
 
 rprocess_s* rprocess_new(const int pid, amap_s *fds_map);
diff --git a/ioriot/src/replay/rworker.c b/ioriot/src/replay/rworker.c
index 89bb49f..d4f45c4 100644
--- a/ioriot/src/replay/rworker.c
+++ b/ioriot/src/replay/rworker.c
@@ -65,6 +65,7 @@ rworker_s* rworker_new(const int rworker_num, amap_s *fds_map,
 
     pthread_mutex_init(&w->rthread_buffer_mutex, NULL);
     pthread_mutex_init(&w->task_buffer_mutex, NULL);
+    pthread_mutex_init(&w->fds_map_mutex, NULL);
 
     // TODO: Check in the program whether the ulimit is high enough
     // or not! (ulimit -n)
@@ -101,6 +102,7 @@ void rworker_destroy(rworker_s *w)
 
     pthread_mutex_destroy(&w->task_buffer_mutex);
     pthread_mutex_destroy(&w->rthread_buffer_mutex);
+    pthread_mutex_destroy(&w->fds_map_mutex);
 
 #ifdef THREAD_DEBUG
     if (w->rworker_fd)
diff --git a/ioriot/src/replay/rworker.h b/ioriot/src/replay/rworker.h
index 26a1300..f0528dc 100644
--- a/ioriot/src/replay/rworker.h
+++ b/ioriot/src/replay/rworker.h
@@ -34,6 +34,7 @@
 typedef struct {
     int rworker_num; /**< The current worker ID */
     amap_s* fds_map; /**< Holding all file descriptors */
+    pthread_mutex_t fds_map_mutex; /**< Sync access to the shared fds_map */
     amap_s* rprocess_map; /**< Holding all processes handled by this worker */
     amap_s* rthread_map; /**< Holding all threads handled by this worker */
     rbuffer_s *task_buffer; /**< Buffering thread tasks to be reused */