Re: [Qemu-devel] [RFC v2 1/6] bitmap: add atomic set functions

[Paolo Bonzini <pbonzini@redhat.com>]

On 02/12/2014 12:23, Stefan Hajnoczi wrote:
> Use atomic_or() for atomic bitmaps where several threads may set bits at
> the same time.  This avoids the race condition between threads loading
> an element, bitwise ORing, and then storing the element.
> 
> When setting all bits in a word we can avoid atomic ops and instead just
> use an smp_wmb() write barrier at the end.
> 
> Most bitmap users don't need atomicity so introduce new functions.
> 
> Signed-off-by: Stefan Hajnoczi <stefanha@redhat.com>
> ---
>  include/qemu/bitmap.h |  2 ++
>  include/qemu/bitops.h | 14 ++++++++++++++
>  util/bitmap.c         | 36 ++++++++++++++++++++++++++++++++++++
>  3 files changed, 52 insertions(+)
> 
> diff --git a/include/qemu/bitmap.h b/include/qemu/bitmap.h
> index f0273c9..3e0a4f3 100644
> --- a/include/qemu/bitmap.h
> +++ b/include/qemu/bitmap.h
> @@ -39,6 +39,7 @@
>   * bitmap_empty(src, nbits)			Are all bits zero in *src?
>   * bitmap_full(src, nbits)			Are all bits set in *src?
>   * bitmap_set(dst, pos, nbits)			Set specified bit area
> + * bitmap_set_atomic(dst, pos, nbits)   Set specified bit area with atomic ops
>   * bitmap_clear(dst, pos, nbits)		Clear specified bit area
>   * bitmap_find_next_zero_area(buf, len, pos, n, mask)	Find bit free area
>   */
> @@ -226,6 +227,7 @@ static inline int bitmap_intersects(const unsigned long *src1,
>  }
>  
>  void bitmap_set(unsigned long *map, long i, long len);
> +void bitmap_set_atomic(unsigned long *map, long i, long len);
>  void bitmap_clear(unsigned long *map, long start, long nr);
>  unsigned long bitmap_find_next_zero_area(unsigned long *map,
>                                           unsigned long size,
> diff --git a/include/qemu/bitops.h b/include/qemu/bitops.h
> index 181bd46..eda4132 100644
> --- a/include/qemu/bitops.h
> +++ b/include/qemu/bitops.h
> @@ -16,6 +16,7 @@
>  #include <assert.h>
>  
>  #include "host-utils.h"
> +#include "atomic.h"
>  
>  #define BITS_PER_BYTE           CHAR_BIT
>  #define BITS_PER_LONG           (sizeof (unsigned long) * BITS_PER_BYTE)
> @@ -39,6 +40,19 @@ static inline void set_bit(long nr, unsigned long *addr)
>  }
>  
>  /**
> + * set_bit_atomic - Set a bit in memory atomically
> + * @nr: the bit to set
> + * @addr: the address to start counting from
> + */
> +static inline void set_bit_atomic(long nr, unsigned long *addr)
> +{
> +    unsigned long mask = BIT_MASK(nr);
> +    unsigned long *p = addr + BIT_WORD(nr);
> +
> +    atomic_or(p, mask);
> +}
> +
> +/**
>   * clear_bit - Clears a bit in memory
>   * @nr: Bit to clear
>   * @addr: Address to start counting from
> diff --git a/util/bitmap.c b/util/bitmap.c
> index 9c6bb52..40db0d9 100644
> --- a/util/bitmap.c
> +++ b/util/bitmap.c
> @@ -11,6 +11,7 @@
>  
>  #include "qemu/bitops.h"
>  #include "qemu/bitmap.h"
> +#include "qemu/atomic.h"
>  
>  /*
>   * bitmaps provide an array of bits, implemented using an an
> @@ -177,6 +178,41 @@ void bitmap_set(unsigned long *map, long start, long nr)
>      }
>  }
>  
> +void bitmap_set_atomic(unsigned long *map, long start, long nr)
> +{
> +    unsigned long *p = map + BIT_WORD(start);
> +    const long size = start + nr;
> +    int bits_to_set = BITS_PER_LONG - (start % BITS_PER_LONG);
> +    unsigned long mask_to_set = BITMAP_FIRST_WORD_MASK(start);
> +
> +    /* First word */
> +    if (nr - bits_to_set > 0) {
> +        atomic_or(p, mask_to_set);
> +        nr -= bits_to_set;
> +        bits_to_set = BITS_PER_LONG;
> +        p++;
> +    }
> +
> +    /* Full words */
> +    while (nr - bits_to_set >= 0) {
> +        *p = ~0UL;
> +        nr -= bits_to_set;
> +        p++;
> +    }
> +
> +    /* Last word */
> +    if (nr) {
> +        mask_to_set &= BITMAP_LAST_WORD_MASK(size);
> +        atomic_or(p, mask_to_set);
> +    } else {
> +        /* If we avoided the full barrier in atomic_or() we should at least
> +         * issue a write barrier so that other threads using barriers see
> +         * up-to-date bitmap contents.
> +         */
> +        smp_wmb();

Why not smp_mb() since that's what atomic_or does?

You can also ensure that you always wrap the loop with atomic_ors (or do
one if setting a single word).  I think it can be like this:

    if (!nr) {
        return;
    }

    /* Always do one atomic_or at the beginning, except if one word.  */
    if (nr >= bits_to_set) {
        ...
    }

    /* Full words, leave the last word aside */
    while (nr - bits_to_set > BITS_PER_LONG) {
    }

    /* Last word */
    assert(nr > 0);
    ...

but maybe I messed up and there's some off-by-one somewhere in this sketch.

Paolo

> +    }
> +}
> +
>  void bitmap_clear(unsigned long *map, long start, long nr)
>  {
>      unsigned long *p = map + BIT_WORD(start);
>