RE: [RFC] rwlock: prevent readers from starving writers

["Morten Brørup" <mb@smartsharesystems.com>]

> From: Honnappa Nagarahalli [mailto:Honnappa.Nagarahalli@arm.com]
> Sent: Friday, 8 July 2022 21.22
> 
> <snip>
> >
> > The original reader/writer lock in DPDK can cause a stream of readers
> to
> > starve writers.
> >
> > The new version uses an additional bit to indicate that a writer is
> waiting and
> > which keeps readers from starving the writer.
> This addition makes sense.
> I am wondering if we should create a new lock. Is it possible that some
> applications are dependent on the current behavior?

Any reader risks having to wait a while for a writer to finish its work.

In my opinion, this implementation only increases the probability of that risk occurring, but it doesn't change the writer's impact on the readers. Therefore, I think this improved implementation can replace the old rwlock.

> 
> >
> > Signed-off-by: Stephen Hemminger <stephen@networkplumber.org>
> > ---
> > Would like this to be in 22.11, but needs some more review
> >
> >  lib/eal/include/generic/rte_rwlock.h | 93 ++++++++++++++++++--------
> --
> >  1 file changed, 61 insertions(+), 32 deletions(-)
> >
> > diff --git a/lib/eal/include/generic/rte_rwlock.h
> > b/lib/eal/include/generic/rte_rwlock.h
> > index da9bc3e9c0e2..725cd19ffb27 100644
> > --- a/lib/eal/include/generic/rte_rwlock.h
> > +++ b/lib/eal/include/generic/rte_rwlock.h
> > @@ -13,7 +13,7 @@
> >   * This file defines an API for read-write locks. The lock is used
> to
> >   * protect data that allows multiple readers in parallel, but only
> >   * one writer. All readers are blocked until the writer is finished
> > - * writing.
> > + * writing. This version will not starve writers.
> >   *
> >   */
> >
> > @@ -28,10 +28,17 @@ extern "C" {
> >  /**
> >   * The rte_rwlock_t type.
> >   *
> > - * cnt is -1 when write lock is held, and > 0 when read locks are
> held.
> > + * Readers increment the counter by RW_READ (4)
> > + * Writers set the RWLOCK_WRITE bit when lock is held
> > + *     and set the RWLOCK_WAIT bit while waiting.
> >   */
> > +
> > +#define RTE_RWLOCK_WAIT	 0x1	/* Writer is waiting */
> > +#define RTE_RWLOCK_WRITE 0x2	/* Writer has the lock */
> > +#define RTE_RWLOCK_READ	 0x4	/* Reader increment */
> > +
> >  typedef struct {
> > -	volatile int32_t cnt; /**< -1 when W lock held, > 0 when R locks
> held.
> > */
> > +	volatile int32_t cnt;

Not signed anymore, so consider uint32_t. Suggest also rename to cnt_state or similar, since it is not just a counter anymore.

> >  } rte_rwlock_t;
> >
> >  /**
> > @@ -61,17 +68,24 @@ static inline void
> >  rte_rwlock_read_lock(rte_rwlock_t *rwl)  {
> >  	int32_t x;
> > -	int success = 0;
> >
> > -	while (success == 0) {
> > +	while (1) {
> >  		x = __atomic_load_n(&rwl->cnt, __ATOMIC_RELAXED);
> >  		/* write lock is held */

Held -> Held or pending, not just held. Add question mark, or move inside the if block.

> > -		if (x < 0) {
> > +		if (x & (RTE_RWLOCK_WAIT | RTE_RWLOCK_WRITE)) {
> >  			rte_pause();
> >  			continue;
> >  		}
> > -		success = __atomic_compare_exchange_n(&rwl->cnt, &x, x
> > + 1, 1,
> > -					__ATOMIC_ACQUIRE,
> > __ATOMIC_RELAXED);
> > +
> > +		/* Try to get read lock */
> > +		x = __atomic_add_fetch(&rwl->cnt, RTE_RWLOCK_READ,
> > +				       __ATOMIC_ACQUIRE);
> > +		if (!(x & (RTE_RWLOCK_WAIT | RTE_RWLOCK_WRITE)))
> > +			return;
> > +
> > +		/* Undo */

Undo -> Unable, so release the read lock.

> > +		__atomic_fetch_sub(&rwl->cnt, RTE_RWLOCK_READ,
> > +				   __ATOMIC_RELEASE);
> >  	}
> >  }
> >
> > @@ -93,17 +107,23 @@ static inline int
> >  rte_rwlock_read_trylock(rte_rwlock_t *rwl)  {
> >  	int32_t x;
> > -	int success = 0;
> >
> > -	while (success == 0) {
> > -		x = __atomic_load_n(&rwl->cnt, __ATOMIC_RELAXED);
> > -		/* write lock is held */
> > -		if (x < 0)
> > -			return -EBUSY;
> > -		success = __atomic_compare_exchange_n(&rwl->cnt, &x, x
> > + 1, 1,
> > -					__ATOMIC_ACQUIRE,
> > __ATOMIC_RELAXED);
> > -	}
> > +	x = __atomic_load_n(&rwl->cnt, __ATOMIC_RELAXED);
> > +
> > +	/* write lock is held */

Same comment as above.

> > +	if (x & (RTE_RWLOCK_WAIT | RTE_RWLOCK_WRITE))
> > +		return -EBUSY;
> > +
> > +	/* Try to get read lock */
> > +	x = __atomic_add_fetch(&rwl->cnt, RTE_RWLOCK_READ,
> > +			       __ATOMIC_ACQUIRE);
> > +
> > +	if (x & (RTE_RWLOCK_WAIT | RTE_RWLOCK_WRITE)) {

Add a comment, e.g.: Unable, so release the read lock.

> > +		__atomic_fetch_sub(&rwl->cnt, RTE_RWLOCK_READ,
> > +				   __ATOMIC_RELEASE);
> >
> > +		return -EBUSY;
> > +	}
> >  	return 0;
> >  }
> >
> > @@ -116,7 +136,7 @@ rte_rwlock_read_trylock(rte_rwlock_t *rwl)
> static
> > inline void  rte_rwlock_read_unlock(rte_rwlock_t *rwl)  {
> > -	__atomic_fetch_sub(&rwl->cnt, 1, __ATOMIC_RELEASE);
> > +	__atomic_fetch_sub(&rwl->cnt, RTE_RWLOCK_READ,
> > __ATOMIC_RELEASE);
> >  }
> >
> >  /**
> > @@ -139,11 +159,12 @@ rte_rwlock_write_trylock(rte_rwlock_t *rwl)
> >  	int32_t x;
> >
> >  	x = __atomic_load_n(&rwl->cnt, __ATOMIC_RELAXED);
> > -	if (x != 0 || __atomic_compare_exchange_n(&rwl->cnt, &x, -1, 1,
> > -			      __ATOMIC_ACQUIRE, __ATOMIC_RELAXED) == 0)
> > +	if (x < RTE_RWLOCK_WRITE &&
> > +	    __atomic_compare_exchange_n(&rwl->cnt, &x, x +
> > RTE_RWLOCK_WRITE,
> > +					1, __ATOMIC_ACQUIRE,
> > __ATOMIC_RELAXED))
> > +		return 0;
> > +	else
> >  		return -EBUSY;
> > -
> > -	return 0;
> >  }
> >
> >  /**
> > @@ -156,18 +177,26 @@ static inline void
> > rte_rwlock_write_lock(rte_rwlock_t *rwl)  {
> >  	int32_t x;
> > -	int success = 0;
> >
> > -	while (success == 0) {
> > +	while (1) {
> >  		x = __atomic_load_n(&rwl->cnt, __ATOMIC_RELAXED);
> > -		/* a lock is held */
> > -		if (x != 0) {
> > -			rte_pause();
> > -			continue;
> > +
> > +		/* No readers or writers */

Add question mark, or move inside if block.

> > +		if (x < RTE_RWLOCK_WRITE) {
> > +			/* Turn off RTE_RWLOCK_WAIT, turn on
> > RTE_RWLOCK_WRITE */
> > +			if (__atomic_compare_exchange_n(&rwl->cnt, &x,
> > RTE_RWLOCK_WRITE, 1,
> > +
> > 	__ATOMIC_ACQUIRE, __ATOMIC_RELAXED))
> > +				return;
> >  		}
> > -		success = __atomic_compare_exchange_n(&rwl->cnt, &x, -
> > 1, 1,
> > -					__ATOMIC_ACQUIRE,
> > __ATOMIC_RELAXED);
> > -	}
> > +
> > +		/* Turn on writer wait bit */
> > +		if (!(x & RTE_RWLOCK_WAIT))
> > +			__atomic_fetch_or(&rwl->cnt, RTE_RWLOCK_WAIT,
> > __ATOMIC_RELAXED);
> > +
> > +		/* Wait until can try to take the lock */
> > +		while (__atomic_load_n(&rwl->cnt, __ATOMIC_RELAXED) >
> > RTE_RWLOCK_WAIT)
> > +			rte_pause();
> > +    }
> >  }
> >
> >  /**
> > @@ -179,7 +208,7 @@ rte_rwlock_write_lock(rte_rwlock_t *rwl)  static
> > inline void  rte_rwlock_write_unlock(rte_rwlock_t *rwl)  {
> > -	__atomic_store_n(&rwl->cnt, 0, __ATOMIC_RELEASE);
> > +	__atomic_fetch_sub(&rwl->cnt, RTE_RWLOCK_WRITE,
> > __ATOMIC_RELEASE);
> >  }
> >
> >  /**
> > --
> > 2.35.1
> 

Always the creative mind, Stephen. :-)

You might consider adding/updating even more comments.

Acked-by: Morten Brørup <mb@smartsharesystems.com>