rwlock rules list

diff --git a/src/rxpd.h b/src/rxpd.h
index b8885df..e8dd20c 100644 (file)
--- a/src/rxpd.h
+++ b/src/rxpd.h
@@ -149,6 +149,7 @@ struct rxpd_file
   const char* filename; // full filename
   //TODO later     struct stat last_stat;
   struct rxpd_base* base;
+  pth_rwlock_t lock;
   llist rules;
 };
 
@@ -167,6 +168,9 @@ rxpd_file_load (struct rxpd_file* self);
 int
 rxpd_file_save (struct rxpd_file* self);
 
+struct rxpd_file*
+rxpd_file_dump (struct rxpd_file* self, struct rxpd_buffer* out);
+
 int
 rxpd_file_cmp (const void* A, const void* B);
 

diff --git a/src/rxpd_connection_cmd.c b/src/rxpd_connection_cmd.c
index caebb9e..f1a2c67 100644 (file)
--- a/src/rxpd_connection_cmd.c
+++ b/src/rxpd_connection_cmd.c
@@ -44,8 +44,8 @@ rxpd_connection_cmd_CHECK (struct rxpd_connection* self)
         }
       else
         {
-          // TODO lock the list, printf schedules!
           // TODO implement RXPD_YIELD_EVERY
+          pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RD, FALSE, NULL);
           LLIST_FOREACH (&self->file->rules, n)
             {
               struct rxpd_rule* rule = (struct rxpd_rule*)n;
@@ -61,6 +61,7 @@ rxpd_connection_cmd_CHECK (struct rxpd_connection* self)
                     }
                 }
             }
+          pth_rwlock_release (&self->file->lock);
         }
     }
 
@@ -99,8 +100,12 @@ rxpd_connection_cmd_APPEND (struct rxpd_connection* self)
 {
   RXPD_FILENAME_REQUIRED;
 
+  pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RW, FALSE, NULL);
+
   rxpd_connection_APPEND_PREPEND_helper (self);
   llist_insertlist_prev (&self->file->rules, &self->tmp_list);
+
+  pth_rwlock_release (&self->file->lock);
 }
 
 void
@@ -108,8 +113,12 @@ rxpd_connection_cmd_PREPEND (struct rxpd_connection* self)
 {
   RXPD_FILENAME_REQUIRED;
 
+  pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RW, FALSE, NULL);
+
   rxpd_connection_APPEND_PREPEND_helper (self);
   llist_insertlist_next (&self->file->rules, &self->tmp_list);
+
+  pth_rwlock_release (&self->file->lock);
 }
 
 void
@@ -120,6 +129,7 @@ rxpd_connection_cmd_REMOVE (struct rxpd_connection* self)
   char* line;
   while ((line = rxpd_buffer_readline (&self->in)))
     {
+      pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RW, FALSE, NULL);
       LLIST_FOREACH (&self->file->rules, n)
         {
           struct rxpd_rule* rule = (struct rxpd_rule*)n;
@@ -134,7 +144,7 @@ rxpd_connection_cmd_REMOVE (struct rxpd_connection* self)
         }
       rxpd_buffer_printf (&self->out, "#ERROR: line not found\n");
     done:
-      ;
+      pth_rwlock_release (&self->file->lock);
     }
 
   if (rxpd_buffer_state (&self->in) == RXPD_ERROR)
@@ -147,17 +157,23 @@ static int
 rxpd_connection_do_REPLACE (struct rxpd_connection* self)
 {
   struct rxpd_rule* rule;
+
+  pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RW, FALSE, NULL);
+
   LLIST_FOREACH (&self->file->rules, n)
     {
       rule = (struct rxpd_rule*)n;
       if (strcmp (rule->string, self->tmp_str) == 0)
         goto found;
     }
+  pth_rwlock_release (&self->file->lock);
   return 0;
 
  found:
   llist_insertlist_next (&rule->node, &self->tmp_list);
   rxpd_rule_delete (rule);
+  pth_rwlock_release (&self->file->lock);
+
   free (self->tmp_str);
   self->tmp_str = NULL;
   return 1;
@@ -168,6 +184,8 @@ rxpd_connection_cmd_REPLACE (struct rxpd_connection* self)
 {
   RXPD_FILENAME_REQUIRED;
 
+  pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RW, FALSE, NULL);
+
   char* line;
   while ((line = rxpd_buffer_readline (&self->in)))
     {
@@ -188,6 +206,8 @@ rxpd_connection_cmd_REPLACE (struct rxpd_connection* self)
         self->tmp_str = rxpd_strdup (line);
     }
 
+  pth_rwlock_release (&self->file->lock);
+
   if (rxpd_buffer_state (&self->in) == RXPD_ERROR)
     rxpd_buffer_printf (&self->out, "#ERROR:\n");
 
@@ -241,6 +261,8 @@ rxpd_connection_cmd_FETCH (struct rxpd_connection* self)
   line = rxpd_buffer_readline (&self->in);
   if (*line)
     {
+      pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RW, FALSE, NULL);
+
       rxpd_file_rules_delete (self->file);
 
       char* list = strrchr (line, '/');
@@ -279,6 +301,10 @@ rxpd_connection_cmd_FETCH (struct rxpd_connection* self)
       // add recieving event
 
       freeaddrinfo (addrs);
+
+
+      pth_rwlock_release (&self->file->lock);
+
     }
 }
 
@@ -286,22 +312,7 @@ void
 rxpd_connection_cmd_DUMP (struct rxpd_connection* self)
 {
   RXPD_FILENAME_REQUIRED;
-
-  if (llist_is_empty (&self->file->rules))
-    rxpd_buffer_printf (&self->out, "#OK:\n");
-  else
-    {
-      LLIST_FOREACH (&self->file->rules, n)
-        {
-          struct rxpd_rule* rule = (struct rxpd_rule*)n;
-          if (rule->atime != (time_t)-1)
-            rxpd_buffer_printf (&self->out, "%ld:%s\n", rule->atime, rule->string);
-          else if (*rule->string != '#')
-            rxpd_buffer_printf (&self->out, ":%s\n", rule->string);
-          else
-            rxpd_buffer_printf (&self->out, "%s\n", rule->string);
-        }
-    }
+  rxpd_file_dump (self->file, &self->out);
 }
 
 
@@ -334,11 +345,13 @@ void
 rxpd_connection_cmd_SHUTDOWN (struct rxpd_connection* self)
 {
   struct rxpd_base* base = self->socket->base;
+
   LLIST_FOREACH (&base->sockets, n)
     {
       struct rxpd_socket* socket = (struct rxpd_socket*)n;
       rxpd_socket_cancel (socket);
     }
+
   rxpd_buffer_printf (&self->out, "#OK:\n");
 }
 
@@ -381,6 +394,8 @@ rxpd_connection_cmd_EXPIRE (struct rxpd_connection* self)
       rxpd_log (base, LOG_INFO, "expire all entries in '%s' since %ld\n",
                 (const char*) self->file->node.key, since);
 
+      pth_rwlock_acquire (&self->file->lock, PTH_RWLOCK_RW, FALSE, NULL);
+
       LLIST_FOREACH (&self->file->rules, n)
         {
           struct rxpd_rule* rule = (struct rxpd_rule*)n;
@@ -392,6 +407,8 @@ rxpd_connection_cmd_EXPIRE (struct rxpd_connection* self)
               rxpd_rule_delete (rule);
             }
         }
+
+      pth_rwlock_release (&self->file->lock);
     }
   else
     rxpd_buffer_printf (&self->out, "#ERROR: no age given\n");

diff --git a/src/rxpd_file.c b/src/rxpd_file.c
index d0341b1..09f0dc9 100644 (file)
--- a/src/rxpd_file.c
+++ b/src/rxpd_file.c
@@ -42,6 +42,8 @@ rxpd_file_new (struct rxpd_base* base, const char* filename)
   self = rxpd_malloc (sizeof (struct rxpd_file));
   self->filename = filename;
   self->base = base;
+  pth_rwlock_init (&self->lock);
+
   const char* basename = strrchr (filename, '/');
   if (basename)
     ++basename;
@@ -73,11 +75,13 @@ rxpd_file_rules_delete (struct rxpd_file* self)
 {
   if (self)
     {
+      pth_rwlock_acquire (&self->lock, PTH_RWLOCK_RW, FALSE, NULL);
       LLIST_WHILE_HEAD (&self->rules, n)
         {
           struct rxpd_rule* node = (struct rxpd_rule*)n;
           rxpd_rule_delete (node);
         }
+      pth_rwlock_release (&self->lock);
     }
   return self;
 }
@@ -89,6 +93,8 @@ rxpd_file_load (struct rxpd_file* self)
   // TODO error handling
   if (f)
     {
+      pth_rwlock_acquire (&self->lock, PTH_RWLOCK_RW, FALSE, NULL);
+
       /* First purge old rules */
       LLIST_WHILE_HEAD (&self->rules, n)
         {
@@ -117,6 +123,7 @@ rxpd_file_load (struct rxpd_file* self)
           llist_insert_tail (&self->rules, &rule->node);
         }
 
+      pth_rwlock_release (&self->lock);
       fclose (f);
       return 1;
     }
@@ -134,6 +141,8 @@ rxpd_file_save (struct rxpd_file* self)
   // TODO error handling
   if (f)
     {
+      pth_rwlock_acquire (&self->lock, PTH_RWLOCK_RD, FALSE, NULL);
+
       LLIST_FOREACH (&self->rules, n)
         {
           struct rxpd_rule* node = (struct rxpd_rule*)n;
@@ -145,6 +154,7 @@ rxpd_file_save (struct rxpd_file* self)
             fprintf (f, "%s\n", node->string);
         }
 
+      pth_rwlock_release (&self->lock);
       fclose (f);
       rxpd_log (self->base, LOG_NOTICE, "saved '%s'\n", self->filename);
       return 1;
@@ -156,6 +166,34 @@ rxpd_file_save (struct rxpd_file* self)
     }
 }
 
+struct rxpd_file*
+rxpd_file_dump (struct rxpd_file* self, struct rxpd_buffer* out)
+{
+  if (self)
+    {
+      if (llist_is_empty (&self->rules))
+        rxpd_buffer_printf (out, "#OK:\n");
+      else
+        {
+          pth_rwlock_acquire (&self->lock, PTH_RWLOCK_RD, FALSE, NULL);
+
+          LLIST_FOREACH (&self->rules, n)
+            {
+              struct rxpd_rule* rule = (struct rxpd_rule*)n;
+              if (rule->atime != (time_t)-1)
+            rxpd_buffer_printf (out, "%ld:%s\n", rule->atime, rule->string);
+              else if (*rule->string != '#')
+                rxpd_buffer_printf (out, ":%s\n", rule->string);
+              else
+                rxpd_buffer_printf (out, "%s\n", rule->string);
+            }
+          pth_rwlock_release (&self->lock);
+        }
+    }
+
+  return self;
+}
+
 int
 rxpd_file_cmp (const void* A, const void* B)
 {