bpf: smp_wmb before bpf_ringbuf really commit

Commit Message

zhongjinji Oct. 31, 2024, 8:42 a.m. UTC

From: zhongjinji <[e-mail]>

To guarantee visibility of writing ringbuffer,it is necessary
to call smp_wmb before ringbuffer really commit.
for instance, when updating the data of buffer in cpu1,
it is not visible for the cpu2 which may be accessing buffer. This may
lead to the consumer accessing a incorrect data. using the smp_wmb
before commmit will guarantee that the consume can access the correct data.

CPU1:
    struct mem_event_t* data = bpf_ringbuf_reserve();
    data->type = MEM_EVENT_KSWAPD_WAKE;
    data->event_data.kswapd_wake.node_id = args->nid;
    bpf_ringbuf_commit(data);

CPU2:
    cons_pos = smp_load_acquire(r->consumer_pos);
    len_ptr = r->data + (cons_pos & r->mask);
    sample = (void *)len_ptr + BPF_RINGBUF_HDR_SZ;
    access to sample

Signed-off-by: zhongjinji <[e-mail]>
---
 kernel/bpf/ringbuf.c | 4 ++++
 1 file changed, 4 insertions(+)

Comments

Hou Tao Oct. 31, 2024, 11:03 a.m. UTC | #1

Hi,

On 10/31/2024 4:42 PM, [e-mail] wrote:
> From: zhongjinji <[e-mail]>
>
> To guarantee visibility of writing ringbuffer,it is necessary
> to call smp_wmb before ringbuffer really commit.
> for instance, when updating the data of buffer in cpu1,
> it is not visible for the cpu2 which may be accessing buffer. This may
> lead to the consumer accessing a incorrect data. using the smp_wmb
> before commmit will guarantee that the consume can access the correct data.
>
> CPU1:
>     struct mem_event_t* data = bpf_ringbuf_reserve();
>     data->type = MEM_EVENT_KSWAPD_WAKE;
>     data->event_data.kswapd_wake.node_id = args->nid;
>     bpf_ringbuf_commit(data);
>
> CPU2:
>     cons_pos = smp_load_acquire(r->consumer_pos);
>     len_ptr = r->data + (cons_pos & r->mask);
>     sample = (void *)len_ptr + BPF_RINGBUF_HDR_SZ;
>     access to sample

It seems you didn't use the ringbuf related APIs (e.g.,
ring_buffer__consume()) in libbpf to access the ring buffer. In my
understanding, the "xchg(&hdr->len, new_len)" in bpf_ringbuf_commit()
works as the barrier to ensure the order of committed data and hdr->len,
and accordingly the "smp_load_acquire(len_ptr)" in
ringbuf_process_ring() in libbpf works as the paired barrier to ensure
the order of hdr->len and the committed data. So I think the extra
smp_wmb() in the kernel is not necessary here. Instead, you should fix
your code in the userspace.
>
> Signed-off-by: zhongjinji <[e-mail]>
> ---
>  kernel/bpf/ringbuf.c | 4 ++++
>  1 file changed, 4 insertions(+)
>
> diff --git a/kernel/bpf/ringbuf.c b/kernel/bpf/ringbuf.c
> index e1cfe890e0be..a66059e2b0d6 100644
> --- a/kernel/bpf/ringbuf.c
> +++ b/kernel/bpf/ringbuf.c
> @@ -508,6 +508,10 @@ static void bpf_ringbuf_commit(void *sample, u64 flags, bool discard)
>  	rec_pos = (void *)hdr - (void *)rb->data;
>  	cons_pos = smp_load_acquire(&rb->consumer_pos) & rb->mask;
>  
> +	/* Make sure the modification of data is visible on other CPU's
> +	 * before consume the event
> +	 */
> +	smp_wmb();
>  	if (flags & BPF_RB_FORCE_WAKEUP)
>  		irq_work_queue(&rb->work);
>  	else if (cons_pos == rec_pos && !(flags & BPF_RB_NO_WAKEUP))

zhongjinji Oct. 31, 2024, 4:19 p.m. UTC | #2

>It seems you didn't use the ringbuf related APIs (e.g.,
>ring_buffer__consume()) in libbpf to access the ring buffer. In my
>understanding, the "xchg(&hdr->len, new_len)" in bpf_ringbuf_commit()
>works as the barrier to ensure the order of committed data and hdr->len,
>and accordingly the "smp_load_acquire(len_ptr)" in
>ringbuf_process_ring() in libbpf works as the paired barrier to ensure
>the order of hdr->len and the committed data. So I think the extra
>smp_wmb() in the kernel is not necessary here. Instead, you should fix
>your code in the userspace.
>>
>> Signed-off-by: zhongjinji <[e-mail]>
>> ---
>>  kernel/bpf/ringbuf.c | 4 ++++
>>  1 file changed, 4 insertions(+)
>>
>> diff --git a/kernel/bpf/ringbuf.c b/kernel/bpf/ringbuf.c
>> index e1cfe890e0be..a66059e2b0d6 100644
>> --- a/kernel/bpf/ringbuf.c
>> +++ b/kernel/bpf/ringbuf.c
>> @@ -508,6 +508,10 @@ static void bpf_ringbuf_commit(void *sample, u64 flags, bool discard)
>>  	rec_pos = (void *)hdr - (void *)rb->data;
>>  	cons_pos = smp_load_acquire(&rb->consumer_pos) & rb->mask;
>>  
>> +	/* Make sure the modification of data is visible on other CPU's
>> +	 * before consume the event
>> +	 */
>> +	smp_wmb();
>>  	if (flags & BPF_RB_FORCE_WAKEUP)
>>  		irq_work_queue(&rb->work);
>>  	else if (cons_pos == rec_pos && !(flags & BPF_RB_NO_WAKEUP))

well, I missed the implementation of __smp_store_release in the x86 architecture,
it is not necessary because __smp_store_release has a memory barrier in x86,
but it can't guarantee the order of committed data in arm64 because
__smp_store_release does not include a memory barrier, and it just ensure
the variable is visible. The implementation of ARM64 is as follows,
in common/arch/arm64/include/asm/barrier.h

#define __smp_load_acquire(p)						\
({									\
	union { __unqual_scalar_typeof(*p) __val; char __c[1]; } __u;	\
	typeof(p) __p = (p);						\
	compiletime_assert_atomic_type(*p);				\
	kasan_check_read(__p, sizeof(*p));				\
	switch (sizeof(*p)) {						\
	case 1:								\
		asm volatile ("ldarb %w0, %1"				\
			: "=r" (*(__u8 *)__u.__c)			\
			: "Q" (*__p) : "memory");			\
		break;							\
	case 2:								\
		asm volatile ("ldarh %w0, %1"				\
			: "=r" (*(__u16 *)__u.__c)			\
			: "Q" (*__p) : "memory");			\
		break;							\
	case 4:								\
		asm volatile ("ldar %w0, %1"				\
			: "=r" (*(__u32 *)__u.__c)			\
			: "Q" (*__p) : "memory");			\
		break;							\
	case 8:								\
		asm volatile ("ldar %0, %1"				\
			: "=r" (*(__u64 *)__u.__c)			\
			: "Q" (*__p) : "memory");			\
		break;							\
	}								\
	(typeof(*p))__u.__val;						\
})

Patch

diff --git a/kernel/bpf/ringbuf.c b/kernel/bpf/ringbuf.c
index e1cfe890e0be..a66059e2b0d6 100644
--- a/kernel/bpf/ringbuf.c
+++ b/kernel/bpf/ringbuf.c
@@ -508,6 +508,10 @@  static void bpf_ringbuf_commit(void *sample, u64 flags, bool discard)
 	rec_pos = (void *)hdr - (void *)rb->data;
 	cons_pos = smp_load_acquire(&rb->consumer_pos) & rb->mask;
 
+	/* Make sure the modification of data is visible on other CPU's
+	 * before consume the event
+	 */
+	smp_wmb();
 	if (flags & BPF_RB_FORCE_WAKEUP)
 		irq_work_queue(&rb->work);
 	else if (cons_pos == rec_pos && !(flags & BPF_RB_NO_WAKEUP))