[RFC,v2,1/6] bitmap: add atomic set functions

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(+)

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);
>