[Qemu-devel] [RFC PATCH] convert ram_list to RCU DQ

[Qemu-devel] [RFC PATCH] convert ram_list to RCU DQ

[Mike Day]

Allow "unlocked" reads of the ram_list by using an RCU-enabled
DQ. Most readers of the list no longer require holding the list mutex.

The ram_list now uses a QLIST instead of a QTAILQ. The difference is
minimal.

This patch has been built and make-checked for the x86_64, ppc64,
s390x, and arm targets. It has not been tested further than that at
this point.

To apply this patch, you must base upon Paolo Bonzini's rcu tree and
also apply the RCU DQ patch (below).

https://github.com/bonzini/qemu/tree/rcu
http://article.gmane.org/gmane.comp.emulators.qemu/230159/

Signed-off-by: Mike Day <ncmike@ncultra.org>
---
 arch_init.c               |  51 ++++++++++++---------
 exec.c                    | 111 +++++++++++++++++++++++++++++-----------------
 hw/9pfs/virtio-9p-synth.c |   2 +-
 include/exec/cpu-all.h    |   4 +-
 include/qemu/rcu_queue.h  |   8 ++++
 5 files changed, 111 insertions(+), 65 deletions(-)

diff --git a/arch_init.c b/arch_init.c
index 68a7ab7..5c7b284 100644
--- a/arch_init.c
+++ b/arch_init.c
@@ -49,6 +49,7 @@
 #include "trace.h"
 #include "exec/cpu-all.h"
 #include "hw/acpi/acpi.h"
+#include "qemu/rcu_queue.h"
 
 #ifdef DEBUG_ARCH_INIT
 #define DPRINTF(fmt, ...) \
@@ -397,8 +398,8 @@ static void migration_bitmap_sync(void)
 
     trace_migration_bitmap_sync_start();
     address_space_sync_dirty_bitmap(&address_space_memory);
-
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         for (addr = 0; addr < block->length; addr += TARGET_PAGE_SIZE) {
             if (memory_region_test_and_clear_dirty(block->mr,
                                                    addr, TARGET_PAGE_SIZE,
@@ -407,6 +408,7 @@ static void migration_bitmap_sync(void)
             }
         }
     }
+    rcu_read_unlock();
     trace_migration_bitmap_sync_end(migration_dirty_pages
                                     - num_dirty_pages_init);
     num_dirty_pages_period += migration_dirty_pages - num_dirty_pages_init;
@@ -457,8 +459,9 @@ static int ram_save_block(QEMUFile *f, bool last_stage)
     MemoryRegion *mr;
     ram_addr_t current_addr;
 
+    rcu_read_lock();
     if (!block)
-        block = QTAILQ_FIRST(&ram_list.blocks);
+        block = QLIST_FIRST_RCU(&ram_list.blocks);
 
     while (true) {
         mr = block->mr;
@@ -469,9 +472,9 @@ static int ram_save_block(QEMUFile *f, bool last_stage)
         }
         if (offset >= block->length) {
             offset = 0;
-            block = QTAILQ_NEXT(block, next);
+            block = QLIST_NEXT_RCU(block, next);
             if (!block) {
-                block = QTAILQ_FIRST(&ram_list.blocks);
+                block = QLIST_FIRST_RCU(&ram_list.blocks);
                 complete_round = true;
                 ram_bulk_stage = false;
             }
@@ -526,6 +529,7 @@ static int ram_save_block(QEMUFile *f, bool last_stage)
             }
         }
     }
+    rcu_read_unlock();
     last_seen_block = block;
     last_offset = offset;
 
@@ -565,10 +569,10 @@ uint64_t ram_bytes_total(void)
 {
     RAMBlock *block;
     uint64_t total = 0;
-
-    QTAILQ_FOREACH(block, &ram_list.blocks, next)
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next)
         total += block->length;
-
+    rcu_read_unlock();
     return total;
 }
 
@@ -631,7 +635,7 @@ static int ram_save_setup(QEMUFile *f, void *opaque)
     }
 
     qemu_mutex_lock_iothread();
-    qemu_mutex_lock_ramlist();
+    rcu_read_lock();
     bytes_transferred = 0;
     reset_ram_globals();
 
@@ -641,13 +645,13 @@ static int ram_save_setup(QEMUFile *f, void *opaque)
 
     qemu_put_be64(f, ram_bytes_total() | RAM_SAVE_FLAG_MEM_SIZE);
 
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         qemu_put_byte(f, strlen(block->idstr));
         qemu_put_buffer(f, (uint8_t *)block->idstr, strlen(block->idstr));
         qemu_put_be64(f, block->length);
     }
 
-    qemu_mutex_unlock_ramlist();
+    rcu_read_unlock();
 
     ram_control_before_iterate(f, RAM_CONTROL_SETUP);
     ram_control_after_iterate(f, RAM_CONTROL_SETUP);
@@ -664,8 +668,6 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
     int64_t t0;
     int total_sent = 0;
 
-    qemu_mutex_lock_ramlist();
-
     if (ram_list.version != last_version) {
         reset_ram_globals();
     }
@@ -701,8 +703,6 @@ static int ram_save_iterate(QEMUFile *f, void *opaque)
         i++;
     }
 
-    qemu_mutex_unlock_ramlist();
-
     /*
      * Must occur before EOS (or any QEMUFile operation)
      * because of RDMA protocol.
@@ -814,6 +814,7 @@ static inline void *host_from_stream_offset(QEMUFile *f,
     static RAMBlock *block = NULL;
     char id[256];
     uint8_t len;
+    void *ptr = NULL;
 
     if (flags & RAM_SAVE_FLAG_CONTINUE) {
         if (!block) {
@@ -828,13 +829,18 @@ static inline void *host_from_stream_offset(QEMUFile *f,
     qemu_get_buffer(f, (uint8_t *)id, len);
     id[len] = 0;
 
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
-        if (!strncmp(id, block->idstr, sizeof(id)))
-            return memory_region_get_ram_ptr(block->mr) + offset;
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
+        if (!strncmp(id, block->idstr, sizeof(id))) {
+            ptr = memory_region_get_ram_ptr(block->mr) + offset;
+            goto unlock_out;
+        }
     }
 
     fprintf(stderr, "Can't find block %s!\n", id);
-    return NULL;
+unlock_out:
+    rcu_read_unlock();
+    return ptr;
 }
 
 /*
@@ -889,8 +895,8 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
                     qemu_get_buffer(f, (uint8_t *)id, len);
                     id[len] = 0;
                     length = qemu_get_be64(f);
-
-                    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+                    rcu_read_lock();
+                    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
                         if (!strncmp(id, block->idstr, sizeof(id))) {
                             if (block->length != length) {
                                 fprintf(stderr,
@@ -898,12 +904,13 @@ static int ram_load(QEMUFile *f, void *opaque, int version_id)
                                         " in != " RAM_ADDR_FMT "\n", id, length,
                                         block->length);
                                 ret =  -EINVAL;
+                                rcu_read_unlock();
                                 goto done;
                             }
                             break;
                         }
                     }
-
+                    rcu_read_unlock();
                     if (!block) {
                         fprintf(stderr, "Unknown ramblock \"%s\", cannot "
                                 "accept migration\n", id);
diff --git a/exec.c b/exec.c
index 5eebcc1..47e3eab 100644
--- a/exec.c
+++ b/exec.c
@@ -46,7 +46,7 @@
 #endif
 #include "exec/cpu-all.h"
 #include "qemu/tls.h"
-
+#include "qemu/rcu_queue.h"
 #include "exec/cputlb.h"
 #include "translate-all.h"
 
@@ -57,7 +57,7 @@
 #if !defined(CONFIG_USER_ONLY)
 static int in_migration;
 
-RAMList ram_list = { .blocks = QTAILQ_HEAD_INITIALIZER(ram_list.blocks) };
+RAMList ram_list = { .blocks = QLIST_HEAD_INITIALIZER(ram_list.blocks) };
 
 static MemoryRegion *system_memory;
 static MemoryRegion *system_io;
@@ -1023,15 +1023,18 @@ static ram_addr_t find_ram_offset(ram_addr_t size)
 
     assert(size != 0); /* it would hand out same offset multiple times */
 
-    if (QTAILQ_EMPTY(&ram_list.blocks))
+    rcu_read_lock();
+    if (QLIST_EMPTY_RCU(&ram_list.blocks)) {
+        rcu_read_unlock();
         return 0;
+    }
 
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         ram_addr_t end, next = RAM_ADDR_MAX;
 
         end = block->offset + block->length;
 
-        QTAILQ_FOREACH(next_block, &ram_list.blocks, next) {
+        QLIST_FOREACH_RCU(next_block, &ram_list.blocks, next) {
             if (next_block->offset >= end) {
                 next = MIN(next, next_block->offset);
             }
@@ -1041,6 +1044,7 @@ static ram_addr_t find_ram_offset(ram_addr_t size)
             mingap = next - end;
         }
     }
+    rcu_read_unlock();
 
     if (offset == RAM_ADDR_MAX) {
         fprintf(stderr, "Failed to find gap of requested size: %" PRIu64 "\n",
@@ -1056,9 +1060,11 @@ ram_addr_t last_ram_offset(void)
     RAMBlock *block;
     ram_addr_t last = 0;
 
-    QTAILQ_FOREACH(block, &ram_list.blocks, next)
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         last = MAX(last, block->offset + block->length);
-
+    }
+    rcu_read_unlock();
     return last;
 }
 
@@ -1083,12 +1089,14 @@ void qemu_ram_set_idstr(ram_addr_t addr, const char *name, DeviceState *dev)
     RAMBlock *new_block, *block;
 
     new_block = NULL;
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         if (block->offset == addr) {
             new_block = block;
             break;
         }
     }
+    rcu_read_unlock();
     assert(new_block);
     assert(!new_block->idstr[0]);
 
@@ -1102,15 +1110,15 @@ void qemu_ram_set_idstr(ram_addr_t addr, const char *name, DeviceState *dev)
     pstrcat(new_block->idstr, sizeof(new_block->idstr), name);
 
     /* This assumes the iothread lock is taken here too.  */
-    qemu_mutex_lock_ramlist();
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         if (block != new_block && !strcmp(block->idstr, new_block->idstr)) {
             fprintf(stderr, "RAMBlock \"%s\" already registered, abort!\n",
                     new_block->idstr);
             abort();
         }
     }
-    qemu_mutex_unlock_ramlist();
+    rcu_read_unlock();
 }
 
 static int memory_try_enable_merging(void *addr, size_t len)
@@ -1126,7 +1134,7 @@ static int memory_try_enable_merging(void *addr, size_t len)
 ram_addr_t qemu_ram_alloc_from_ptr(ram_addr_t size, void *host,
                                    MemoryRegion *mr)
 {
-    RAMBlock *block, *new_block;
+    RAMBlock *block, *new_block, *last_block = 0;
 
     size = TARGET_PAGE_ALIGN(size);
     new_block = g_malloc0(sizeof(*new_block));
@@ -1165,15 +1173,20 @@ ram_addr_t qemu_ram_alloc_from_ptr(ram_addr_t size, void *host,
     new_block->length = size;
 
     /* Keep the list sorted from biggest to smallest block.  */
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
+        last_block = block;
         if (block->length < new_block->length) {
             break;
         }
     }
     if (block) {
-        QTAILQ_INSERT_BEFORE(block, new_block, next);
+        QLIST_INSERT_BEFORE_RCU(block, new_block, next);
     } else {
-        QTAILQ_INSERT_TAIL(&ram_list.blocks, new_block, next);
+        if (last_block) {
+            QLIST_INSERT_AFTER_RCU(last_block, new_block, next);
+        } else { /* list is empty */
+            QLIST_INSERT_HEAD_RCU(&ram_list.blocks, new_block, next);
+        }
     }
     ram_list.mru_block = NULL;
 
@@ -1206,9 +1219,9 @@ void qemu_ram_free_from_ptr(ram_addr_t addr)
 
     /* This assumes the iothread lock is taken here too.  */
     qemu_mutex_lock_ramlist();
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         if (addr == block->offset) {
-            QTAILQ_REMOVE(&ram_list.blocks, block, next);
+            QLIST_REMOVE_RCU(block, next);
             ram_list.mru_block = NULL;
             ram_list.version++;
             g_free(block);
@@ -1224,9 +1237,9 @@ void qemu_ram_free(ram_addr_t addr)
 
     /* This assumes the iothread lock is taken here too.  */
     qemu_mutex_lock_ramlist();
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         if (addr == block->offset) {
-            QTAILQ_REMOVE(&ram_list.blocks, block, next);
+            QLIST_REMOVE_RCU(block, next);
             ram_list.mru_block = NULL;
             ram_list.version++;
             if (block->flags & RAM_PREALLOC_MASK) {
@@ -1265,7 +1278,8 @@ void qemu_ram_remap(ram_addr_t addr, ram_addr_t length)
     int flags;
     void *area, *vaddr;
 
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         offset = addr - block->offset;
         if (offset < block->length) {
             vaddr = block->host + offset;
@@ -1313,9 +1327,11 @@ void qemu_ram_remap(ram_addr_t addr, ram_addr_t length)
                 memory_try_enable_merging(vaddr, length);
                 qemu_ram_setup_dump(vaddr, length);
             }
-            return;
+            goto unlock_out;
         }
     }
+unlock_out:
+    rcu_read_unlock();
 }
 #endif /* !_WIN32 */
 
@@ -1328,7 +1344,8 @@ static RAMBlock *qemu_get_ram_block(ram_addr_t addr)
     if (block && addr - block->offset < block->length) {
         goto found;
     }
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         if (addr - block->offset < block->length) {
             goto found;
         }
@@ -1338,6 +1355,7 @@ static RAMBlock *qemu_get_ram_block(ram_addr_t addr)
     abort();
 
 found:
+    rcu_read_unlock();
     ram_list.mru_block = block;
     return block;
 }
@@ -1377,9 +1395,11 @@ void *qemu_get_ram_ptr(ram_addr_t addr)
 static void *qemu_safe_ram_ptr(ram_addr_t addr)
 {
     RAMBlock *block;
+    void *ptr = NULL;
 
     /* The list is protected by the iothread lock here.  */
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         if (addr - block->offset < block->length) {
             if (xen_enabled()) {
                 /* We need to check if the requested address is in the RAM
@@ -1393,20 +1413,23 @@ static void *qemu_safe_ram_ptr(ram_addr_t addr)
                         xen_map_cache(block->offset, block->length, 1);
                 }
             }
-            return block->host + (addr - block->offset);
+            ptr =  block->host + (addr - block->offset);
+            goto unlock_out;
         }
     }
 
     fprintf(stderr, "Bad ram offset %" PRIx64 "\n", (uint64_t)addr);
     abort();
-
-    return NULL;
+unlock_out:
+    rcu_read_unlock();
+    return ptr;
 }
 
 /* Return a host pointer to guest's ram. Similar to qemu_get_ram_ptr
  * but takes a size argument */
 static void *qemu_ram_ptr_length(ram_addr_t addr, hwaddr *size)
 {
+    void *ptr = NULL;
     if (*size == 0) {
         return NULL;
     }
@@ -1414,18 +1437,22 @@ static void *qemu_ram_ptr_length(ram_addr_t addr, hwaddr *size)
         return xen_map_cache(addr, *size, 1);
     } else {
         RAMBlock *block;
-
-        QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+        rcu_read_lock();
+        QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
             if (addr - block->offset < block->length) {
                 if (addr - block->offset + *size > block->length)
                     *size = block->length - addr + block->offset;
-                return block->host + (addr - block->offset);
+                ptr = block->host + (addr - block->offset);
+                goto unlock_out;
             }
         }
 
         fprintf(stderr, "Bad ram offset %" PRIx64 "\n", (uint64_t)addr);
         abort();
     }
+unlock_out:
+    rcu_read_unlock();
+    return ptr;
 }
 
 /* Some of the softmmu routines need to translate from a host pointer
@@ -1434,32 +1461,35 @@ MemoryRegion *qemu_ram_addr_from_host(void *ptr, ram_addr_t *ram_addr)
 {
     RAMBlock *block;
     uint8_t *host = ptr;
+    MemoryRegion *mr = NULL;
 
     if (xen_enabled()) {
         *ram_addr = xen_ram_addr_from_mapcache(ptr);
         return qemu_get_ram_block(*ram_addr)->mr;
     }
-
+    rcu_read_lock();
     block = ram_list.mru_block;
     if (block && block->host && host - block->host < block->length) {
-        goto found;
+        *ram_addr = block->offset + (host - block->host);
+        mr = block->mr;
+        goto unlock_out;
     }
 
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         /* This case append when the block is not mapped. */
         if (block->host == NULL) {
             continue;
         }
         if (host - block->host < block->length) {
-            goto found;
+            *ram_addr = block->offset + (host - block->host);
+            mr = block->mr;
+            goto unlock_out;
         }
     }
 
-    return NULL;
-
-found:
-    *ram_addr = block->offset + (host - block->host);
-    return block->mr;
+unlock_out:
+    rcu_read_unlock();
+    return mr;
 }
 
 static void notdirty_mem_write(void *opaque, hwaddr ram_addr,
@@ -2709,9 +2739,10 @@ bool cpu_physical_memory_is_io(hwaddr phys_addr)
 void qemu_ram_foreach_block(RAMBlockIterFunc func, void *opaque)
 {
     RAMBlock *block;
-
-    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
+    rcu_read_lock();
+    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
         func(block->host, block->offset, block->length, opaque);
     }
+    rcu_read_unlock();
 }
 #endif
diff --git a/hw/9pfs/virtio-9p-synth.c b/hw/9pfs/virtio-9p-synth.c
index fdfea21..c2efaca 100644
--- a/hw/9pfs/virtio-9p-synth.c
+++ b/hw/9pfs/virtio-9p-synth.c
@@ -18,7 +18,7 @@
 #include "fsdev/qemu-fsdev.h"
 #include "virtio-9p-synth.h"
 #include "qemu/rcu.h"
-
+#include "qemu/rcu_queue.h"
 #include <sys/stat.h>
 
 /* Root node for synth file system */
diff --git a/include/exec/cpu-all.h b/include/exec/cpu-all.h
index e088089..9cd8a30 100644
--- a/include/exec/cpu-all.h
+++ b/include/exec/cpu-all.h
@@ -457,7 +457,7 @@ typedef struct RAMBlock {
     /* Reads can take either the iothread or the ramlist lock.
      * Writes must take both locks.
      */
-    QTAILQ_ENTRY(RAMBlock) next;
+    QLIST_ENTRY(RAMBlock) next;
 #if defined(__linux__) && !defined(TARGET_S390X)
     int fd;
 #endif
@@ -469,7 +469,7 @@ typedef struct RAMList {
     uint8_t *phys_dirty;
     RAMBlock *mru_block;
     /* Protected by the ramlist lock.  */
-    QTAILQ_HEAD(, RAMBlock) blocks;
+    QLIST_HEAD(, RAMBlock) blocks;
     uint32_t version;
 } RAMList;
 extern RAMList ram_list;
diff --git a/include/qemu/rcu_queue.h b/include/qemu/rcu_queue.h
index e2b8ba5..d159850 100644
--- a/include/qemu/rcu_queue.h
+++ b/include/qemu/rcu_queue.h
@@ -37,6 +37,14 @@
 extern "C" {
 #endif
 
+
+/*
+ * List access methods.
+ */
+#define QLIST_EMPTY_RCU(head) (atomic_rcu_read(&(head)->lh_first) == NULL)
+#define QLIST_FIRST_RCU(head) (atomic_rcu_read(&(head)->lh_first))
+#define QLIST_NEXT_RCU(elm, field) (atomic_rcu_read(&(elm)->field.le_next))
+
 /*
  * List functions.
  */
-- 
1.8.3.1

Re: [Qemu-devel] [RFC PATCH] convert ram_list to RCU DQ

[Paolo Bonzini]

Il 28/08/2013 18:02, Mike Day ha scritto:
> @@ -457,8 +459,9 @@ static int ram_save_block(QEMUFile *f, bool last_stage)
>      MemoryRegion *mr;
>      ram_addr_t current_addr;
>  
> +    rcu_read_lock();
>      if (!block)
> -        block = QTAILQ_FIRST(&ram_list.blocks);
> +        block = QLIST_FIRST_RCU(&ram_list.blocks);
>  
>      while (true) {
>          mr = block->mr;
> @@ -469,9 +472,9 @@ static int ram_save_block(QEMUFile *f, bool last_stage)
>          }
>          if (offset >= block->length) {
>              offset = 0;
> -            block = QTAILQ_NEXT(block, next);
> +            block = QLIST_NEXT_RCU(block, next);
>              if (!block) {
> -                block = QTAILQ_FIRST(&ram_list.blocks);
> +                block = QLIST_FIRST_RCU(&ram_list.blocks);
>                  complete_round = true;
>                  ram_bulk_stage = false;
>              }
> @@ -526,6 +529,7 @@ static int ram_save_block(QEMUFile *f, bool last_stage)
>              }
>          }
>      }
> +    rcu_read_unlock();

block lives across calls to ram_save_block, which is why the mutex was
locked in the caller (ram_save_iterate) rather than here.  For a first
conversion, keeping the long RCU critical sections is fine.  We don't
use RCU enough yet to care about delaying other call_rcu callbacks.

We can later check push the check for ram_list.version inside
ram_save_block, which should let us make the critical section smaller.
But that would be a bit tricky, so it's better to do it in a separate patch.

> @@ -828,13 +829,18 @@ static inline void *host_from_stream_offset(QEMUFile *f,
>      qemu_get_buffer(f, (uint8_t *)id, len);
>      id[len] = 0;
>  
> -    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
> -        if (!strncmp(id, block->idstr, sizeof(id)))
> -            return memory_region_get_ram_ptr(block->mr) + offset;
> +    rcu_read_lock();
> +    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
> +        if (!strncmp(id, block->idstr, sizeof(id))) {
> +            ptr = memory_region_get_ram_ptr(block->mr) + offset;
> +            goto unlock_out;
> +        }
>      }
>  
>      fprintf(stderr, "Can't find block %s!\n", id);
> -    return NULL;
> +unlock_out:
> +    rcu_read_unlock();
> +    return ptr;
>  }


Similarly, here the critical section includes the caller, and block is
living across calls to host.  Again, for now just put all of ram_load
under a huge RCU critical section.  Later we can use ram_list.version to
refresh the list and make the critical sections smaller.

Paolo

Re: [Qemu-devel] [RFC PATCH] convert ram_list to RCU DQ

[Paolo Bonzini]

Il 28/08/2013 18:02, Mike Day ha scritto:
> @@ -1102,15 +1110,15 @@ void qemu_ram_set_idstr(ram_addr_t addr, const char *name, DeviceState *dev)
>      pstrcat(new_block->idstr, sizeof(new_block->idstr), name);
>  
>      /* This assumes the iothread lock is taken here too.  */
> -    qemu_mutex_lock_ramlist();
> -    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
> +    rcu_read_lock();
> +    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
>          if (block != new_block && !strcmp(block->idstr, new_block->idstr)) {
>              fprintf(stderr, "RAMBlock \"%s\" already registered, abort!\n",
>                      new_block->idstr);
>              abort();
>          }
>      }
> -    qemu_mutex_unlock_ramlist();
> +    rcu_read_unlock();

Forgot about this.  Every time you see "This assumes the iothread lock
is taken here too", you're in the write side so you do not need
lock/unlock and you do not need...

>      /* This assumes the iothread lock is taken here too.  */
>      qemu_mutex_lock_ramlist();
> -    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
> +    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
>          if (addr == block->offset) {
> -            QTAILQ_REMOVE(&ram_list.blocks, block, next);
> +            QLIST_REMOVE_RCU(block, next);
>              ram_list.mru_block = NULL;
>              ram_list.version++;
>              g_free(block);

qemu_mutex_lock_ramlist/qemu_mutex_unlock_ramlist either.

Otherwise, the patch is pretty solid!  Thanks,

Paolo

Re: [Qemu-devel] [RFC PATCH] convert ram_list to RCU DQ

[Mike Day]

[-- Attachment #1: Type: text/plain, Size: 1276 bytes --]

On Wed, Aug 28, 2013 at 12:35 PM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
> > @@ -828,13 +829,18 @@ static inline void
*host_from_stream_offset(QEMUFile *f,
> >      qemu_get_buffer(f, (uint8_t *)id, len);
> >      id[len] = 0;
> >
> > -    QTAILQ_FOREACH(block, &ram_list.blocks, next) {
> > -        if (!strncmp(id, block->idstr, sizeof(id)))
> > -            return memory_region_get_ram_ptr(block->mr) + offset;
> > +    rcu_read_lock();
> > +    QLIST_FOREACH_RCU(block, &ram_list.blocks, next) {
> > +        if (!strncmp(id, block->idstr, sizeof(id))) {
> > +            ptr = memory_region_get_ram_ptr(block->mr) + offset;
> > +            goto unlock_out;
> > +        }
> >      }
> >
> >      fprintf(stderr, "Can't find block %s!\n", id);
> > -    return NULL;
> > +unlock_out:
> > +    rcu_read_unlock();
> > +    return ptr;
> >  }
>
>
> Similarly, here the critical section includes the caller, and block is
> living across calls to host.  Again, for now just put all of ram_load
> under a huge RCU critical section.  Later we can use ram_list.version to
> refresh the list and make the critical sections smaller.

And: rcu_read_lock() and rcu_read_unlock() can be called recursively, so I
can still leave the lock/unlock pair in host_from_stream_offset.

[-- Attachment #2: Type: text/html, Size: 1665 bytes --]

Re: [Qemu-devel] [RFC PATCH] convert ram_list to RCU DQ

[Paolo Bonzini]

Il 29/08/2013 21:18, Mike Day ha scritto:
> 
>> Similarly, here the critical section includes the caller, and block is
>> living across calls to host.  Again, for now just put all of ram_load
>> under a huge RCU critical section.  Later we can use ram_list.version to
>> refresh the list and make the critical sections smaller.
> 
> And: rcu_read_lock() and rcu_read_unlock() can be called recursively, so
> I can still leave the lock/unlock pair in host_from_stream_offset. 

I'm not sure about that; returning an RCU-protected variable after
rcu_read_unlock() seems wrong to me because the pointer may not be valid
at that point.  I suggest using a comment that asks to call
host_from_stream_offset within rcu_read_lock()/rcu_read_unlock().
However, if existing practice in the kernel is different, I'll bow to that.

Paolo

Re: [Qemu-devel] [RFC PATCH] convert ram_list to RCU DQ

[Mike Day]

[-- Attachment #1: Type: text/plain, Size: 849 bytes --]

On Fri, Aug 30, 2013 at 4:19 AM, Paolo Bonzini <pbonzini@redhat.com> wrote:
>
>
> I'm not sure about that; returning an RCU-protected variable after
> rcu_read_unlock() seems wrong to me because the pointer may not be valid
> at that point.  I suggest using a comment that asks to call
> host_from_stream_offset within rcu_read_lock()/rcu_read_unlock().
> However, if existing practice in the kernel is different, I'll bow to
that.

In this case the only caller is ram_load so I'm removing the critical
section from within host_from_stream_offset, and adding comments to note
that the ram_list needs to be protected by the caller. The current
docs/rcu.txt information indicates that rcu critical sections can be
"nested or overlapping." But your suggestion results in cleaner code - we
will have to go back to this later as you noted earlier.

Mike

[-- Attachment #2: Type: text/html, Size: 1052 bytes --]