cpumap.c

  1// SPDX-License-Identifier: GPL-2.0-only
  2/* bpf/cpumap.c
  3 *
  4 * Copyright (c) 2017 Jesper Dangaard Brouer, Red Hat Inc.
  5 */
  6
  7/**
  8 * DOC: cpu map
  9 * The 'cpumap' is primarily used as a backend map for XDP BPF helper
 10 * call bpf_redirect_map() and XDP_REDIRECT action, like 'devmap'.
 11 *
 12 * Unlike devmap which redirects XDP frames out to another NIC device,
 13 * this map type redirects raw XDP frames to another CPU.  The remote
 14 * CPU will do SKB-allocation and call the normal network stack.
 15 */
 16/*
 17 * This is a scalability and isolation mechanism, that allow
 18 * separating the early driver network XDP layer, from the rest of the
 19 * netstack, and assigning dedicated CPUs for this stage.  This
 20 * basically allows for 10G wirespeed pre-filtering via bpf.
 21 */
 22#include <linux/bitops.h>
 23#include <linux/bpf.h>
 24#include <linux/filter.h>
 25#include <linux/ptr_ring.h>
 26#include <net/xdp.h>
 27#include <net/hotdata.h>
 28
 29#include <linux/sched.h>
 30#include <linux/workqueue.h>
 31#include <linux/kthread.h>
 32#include <linux/local_lock.h>
 33#include <linux/completion.h>
 34#include <trace/events/xdp.h>
 35#include <linux/btf_ids.h>
 36
 37#include <linux/netdevice.h>
 38#include <net/gro.h>
 39
 40/* General idea: XDP packets getting XDP redirected to another CPU,
 41 * will maximum be stored/queued for one driver ->poll() call.  It is
 42 * guaranteed that queueing the frame and the flush operation happen on
 43 * same CPU.  Thus, cpu_map_flush operation can deduct via this_cpu_ptr()
 44 * which queue in bpf_cpu_map_entry contains packets.
 45 */
 46
 47#define CPU_MAP_BULK_SIZE 8  /* 8 == one cacheline on 64-bit archs */
 48struct bpf_cpu_map_entry;
 49struct bpf_cpu_map;
 50
 51struct xdp_bulk_queue {
 52	void *q[CPU_MAP_BULK_SIZE];
 53	struct list_head flush_node;
 54	struct bpf_cpu_map_entry *obj;
 55	unsigned int count;
 56	local_lock_t bq_lock;
 57};
 58
 59/* Struct for every remote "destination" CPU in map */
 60struct bpf_cpu_map_entry {
 61	u32 cpu;    /* kthread CPU and map index */
 62	int map_id; /* Back reference to map */
 63
 64	/* XDP can run multiple RX-ring queues, need __percpu enqueue store */
 65	struct xdp_bulk_queue __percpu *bulkq;
 66
 67	/* Queue with potential multi-producers, and single-consumer kthread */
 68	struct ptr_ring *queue;
 69	struct task_struct *kthread;
 70
 71	struct bpf_cpumap_val value;
 72	struct bpf_prog *prog;
 73	struct gro_node gro;
 74
 75	struct completion kthread_running;
 76	struct rcu_work free_work;
 77};
 78
 79struct bpf_cpu_map {
 80	struct bpf_map map;
 81	/* Below members specific for map type */
 82	struct bpf_cpu_map_entry __rcu **cpu_map;
 83};
 84
 85static struct bpf_map *cpu_map_alloc(union bpf_attr *attr)
 86{
 87	u32 value_size = attr->value_size;
 88	struct bpf_cpu_map *cmap;
 89
 90	/* check sanity of attributes */
 91	if (attr->max_entries == 0 || attr->key_size != 4 ||
 92	    (value_size != offsetofend(struct bpf_cpumap_val, qsize) &&
 93	     value_size != offsetofend(struct bpf_cpumap_val, bpf_prog.fd)) ||
 94	    attr->map_flags & ~BPF_F_NUMA_NODE)
 95		return ERR_PTR(-EINVAL);
 96
 97	/* Pre-limit array size based on NR_CPUS, not final CPU check */
 98	if (attr->max_entries > NR_CPUS)
 99		return ERR_PTR(-E2BIG);
100
101	cmap = bpf_map_area_alloc(sizeof(*cmap), NUMA_NO_NODE);
102	if (!cmap)
103		return ERR_PTR(-ENOMEM);
104
105	bpf_map_init_from_attr(&cmap->map, attr);
106
107	/* Alloc array for possible remote "destination" CPUs */
108	cmap->cpu_map = bpf_map_area_alloc(cmap->map.max_entries *
109					   sizeof(struct bpf_cpu_map_entry *),
110					   cmap->map.numa_node);
111	if (!cmap->cpu_map) {
112		bpf_map_area_free(cmap);
113		return ERR_PTR(-ENOMEM);
114	}
115
116	return &cmap->map;
117}
118
119static void __cpu_map_ring_cleanup(struct ptr_ring *ring)
120{
121	/* The tear-down procedure should have made sure that queue is
122	 * empty.  See __cpu_map_entry_replace() and work-queue
123	 * invoked cpu_map_kthread_stop(). Catch any broken behaviour
124	 * gracefully and warn once.
125	 */
126	void *ptr;
127
128	while ((ptr = ptr_ring_consume(ring))) {
129		WARN_ON_ONCE(1);
130		if (unlikely(__ptr_test_bit(0, &ptr))) {
131			__ptr_clear_bit(0, &ptr);
132			kfree_skb(ptr);
133			continue;
134		}
135		xdp_return_frame(ptr);
136	}
137}
138
139static u32 cpu_map_bpf_prog_run_skb(struct bpf_cpu_map_entry *rcpu,
140				    void **skbs, u32 skb_n,
141				    struct xdp_cpumap_stats *stats)
142{
143	struct xdp_buff xdp;
144	u32 act, pass = 0;
145	int err;
146
147	for (u32 i = 0; i < skb_n; i++) {
148		struct sk_buff *skb = skbs[i];
149
150		act = bpf_prog_run_generic_xdp(skb, &xdp, rcpu->prog);
151		switch (act) {
152		case XDP_PASS:
153			skbs[pass++] = skb;
154			break;
155		case XDP_REDIRECT:
156			err = xdp_do_generic_redirect(skb->dev, skb, &xdp,
157						      rcpu->prog);
158			if (unlikely(err)) {
159				kfree_skb(skb);
160				stats->drop++;
161			} else {
162				stats->redirect++;
163			}
164			break;
165		default:
166			bpf_warn_invalid_xdp_action(NULL, rcpu->prog, act);
167			fallthrough;
168		case XDP_ABORTED:
169			trace_xdp_exception(skb->dev, rcpu->prog, act);
170			fallthrough;
171		case XDP_DROP:
172			napi_consume_skb(skb, true);
173			stats->drop++;
174			break;
175		}
176	}
177
178	stats->pass += pass;
179
180	return pass;
181}
182
183static int cpu_map_bpf_prog_run_xdp(struct bpf_cpu_map_entry *rcpu,
184				    void **frames, int n,
185				    struct xdp_cpumap_stats *stats)
186{
187	struct xdp_rxq_info rxq = {};
188	struct xdp_buff xdp;
189	int i, nframes = 0;
190
191	xdp.rxq = &rxq;
192
193	for (i = 0; i < n; i++) {
194		struct xdp_frame *xdpf = frames[i];
195		u32 act;
196		int err;
197
198		rxq.dev = xdpf->dev_rx;
199		rxq.mem.type = xdpf->mem_type;
200		/* TODO: report queue_index to xdp_rxq_info */
201
202		xdp_convert_frame_to_buff(xdpf, &xdp);
203
204		act = bpf_prog_run_xdp(rcpu->prog, &xdp);
205		switch (act) {
206		case XDP_PASS:
207			err = xdp_update_frame_from_buff(&xdp, xdpf);
208			if (err < 0) {
209				xdp_return_frame(xdpf);
210				stats->drop++;
211			} else {
212				frames[nframes++] = xdpf;
213			}
214			break;
215		case XDP_REDIRECT:
216			err = xdp_do_redirect(xdpf->dev_rx, &xdp,
217					      rcpu->prog);
218			if (unlikely(err)) {
219				xdp_return_frame(xdpf);
220				stats->drop++;
221			} else {
222				stats->redirect++;
223			}
224			break;
225		default:
226			bpf_warn_invalid_xdp_action(NULL, rcpu->prog, act);
227			fallthrough;
228		case XDP_DROP:
229			xdp_return_frame(xdpf);
230			stats->drop++;
231			break;
232		}
233	}
234
235	stats->pass += nframes;
236
237	return nframes;
238}
239
240#define CPUMAP_BATCH 8
241
242struct cpu_map_ret {
243	u32 xdp_n;
244	u32 skb_n;
245};
246
247static void cpu_map_bpf_prog_run(struct bpf_cpu_map_entry *rcpu, void **frames,
248				 void **skbs, struct cpu_map_ret *ret,
249				 struct xdp_cpumap_stats *stats)
250{
251	struct bpf_net_context __bpf_net_ctx, *bpf_net_ctx;
252
253	if (!rcpu->prog)
254		goto out;
255
256	rcu_read_lock();
257	bpf_net_ctx = bpf_net_ctx_set(&__bpf_net_ctx);
258	xdp_set_return_frame_no_direct();
259
260	ret->xdp_n = cpu_map_bpf_prog_run_xdp(rcpu, frames, ret->xdp_n, stats);
261	if (unlikely(ret->skb_n))
262		ret->skb_n = cpu_map_bpf_prog_run_skb(rcpu, skbs, ret->skb_n,
263						      stats);
264
265	if (stats->redirect)
266		xdp_do_flush();
267
268	xdp_clear_return_frame_no_direct();
269	bpf_net_ctx_clear(bpf_net_ctx);
270	rcu_read_unlock();
271
272out:
273	if (unlikely(ret->skb_n) && ret->xdp_n)
274		memmove(&skbs[ret->xdp_n], skbs, ret->skb_n * sizeof(*skbs));
275}
276
277static void cpu_map_gro_flush(struct bpf_cpu_map_entry *rcpu, bool empty)
278{
279	/*
280	 * If the ring is not empty, there'll be a new iteration soon, and we
281	 * only need to do a full flush if a tick is long (> 1 ms).
282	 * If the ring is empty, to not hold GRO packets in the stack for too
283	 * long, do a full flush.
284	 * This is equivalent to how NAPI decides whether to perform a full
285	 * flush.
286	 */
287	gro_flush_normal(&rcpu->gro, !empty && HZ >= 1000);
288}
289
290static int cpu_map_kthread_run(void *data)
291{
292	struct bpf_cpu_map_entry *rcpu = data;
293	unsigned long last_qs = jiffies;
294	u32 packets = 0;
295
296	complete(&rcpu->kthread_running);
297	set_current_state(TASK_INTERRUPTIBLE);
298
299	/* When kthread gives stop order, then rcpu have been disconnected
300	 * from map, thus no new packets can enter. Remaining in-flight
301	 * per CPU stored packets are flushed to this queue.  Wait honoring
302	 * kthread_stop signal until queue is empty.
303	 */
304	while (!kthread_should_stop() || !__ptr_ring_empty(rcpu->queue)) {
305		struct xdp_cpumap_stats stats = {}; /* zero stats */
306		unsigned int kmem_alloc_drops = 0, sched = 0;
307		struct cpu_map_ret ret = { };
308		void *frames[CPUMAP_BATCH];
309		void *skbs[CPUMAP_BATCH];
310		u32 i, n, m;
311		bool empty;
312
313		/* Release CPU reschedule checks */
314		if (__ptr_ring_empty(rcpu->queue)) {
315			set_current_state(TASK_INTERRUPTIBLE);
316			/* Recheck to avoid lost wake-up */
317			if (__ptr_ring_empty(rcpu->queue)) {
318				schedule();
319				sched = 1;
320				last_qs = jiffies;
321			} else {
322				__set_current_state(TASK_RUNNING);
323			}
324		} else {
325			rcu_softirq_qs_periodic(last_qs);
326			sched = cond_resched();
327		}
328
329		/*
330		 * The bpf_cpu_map_entry is single consumer, with this
331		 * kthread CPU pinned. Lockless access to ptr_ring
332		 * consume side valid as no-resize allowed of queue.
333		 */
334		n = __ptr_ring_consume_batched(rcpu->queue, frames,
335					       CPUMAP_BATCH);
336		for (i = 0; i < n; i++) {
337			void *f = frames[i];
338			struct page *page;
339
340			if (unlikely(__ptr_test_bit(0, &f))) {
341				struct sk_buff *skb = f;
342
343				__ptr_clear_bit(0, &skb);
344				skbs[ret.skb_n++] = skb;
345				continue;
346			}
347
348			frames[ret.xdp_n++] = f;
349			page = virt_to_page(f);
350
351			/* Bring struct page memory area to curr CPU. Read by
352			 * build_skb_around via page_is_pfmemalloc(), and when
353			 * freed written by page_frag_free call.
354			 */
355			prefetchw(page);
356		}
357
358		local_bh_disable();
359
360		/* Support running another XDP prog on this CPU */
361		cpu_map_bpf_prog_run(rcpu, frames, skbs, &ret, &stats);
362		if (!ret.xdp_n)
363			goto stats;
364
365		m = napi_skb_cache_get_bulk(skbs, ret.xdp_n);
366		if (unlikely(m < ret.xdp_n)) {
367			for (i = m; i < ret.xdp_n; i++)
368				xdp_return_frame(frames[i]);
369
370			if (ret.skb_n)
371				memmove(&skbs[m], &skbs[ret.xdp_n],
372					ret.skb_n * sizeof(*skbs));
373
374			kmem_alloc_drops += ret.xdp_n - m;
375			ret.xdp_n = m;
376		}
377
378		for (i = 0; i < ret.xdp_n; i++) {
379			struct xdp_frame *xdpf = frames[i];
380
381			/* Can fail only when !skb -- already handled above */
382			__xdp_build_skb_from_frame(xdpf, skbs[i], xdpf->dev_rx);
383		}
384
385stats:
386		/* Feedback loop via tracepoint.
387		 * NB: keep before recv to allow measuring enqueue/dequeue latency.
388		 */
389		trace_xdp_cpumap_kthread(rcpu->map_id, n, kmem_alloc_drops,
390					 sched, &stats);
391
392		for (i = 0; i < ret.xdp_n + ret.skb_n; i++)
393			gro_receive_skb(&rcpu->gro, skbs[i]);
394
395		/* Flush either every 64 packets or in case of empty ring */
396		packets += n;
397		empty = __ptr_ring_empty(rcpu->queue);
398		if (packets >= NAPI_POLL_WEIGHT || empty) {
399			cpu_map_gro_flush(rcpu, empty);
400			packets = 0;
401		}
402
403		local_bh_enable(); /* resched point, may call do_softirq() */
404	}
405	__set_current_state(TASK_RUNNING);
406
407	return 0;
408}
409
410static int __cpu_map_load_bpf_program(struct bpf_cpu_map_entry *rcpu,
411				      struct bpf_map *map, int fd)
412{
413	struct bpf_prog *prog;
414
415	prog = bpf_prog_get_type(fd, BPF_PROG_TYPE_XDP);
416	if (IS_ERR(prog))
417		return PTR_ERR(prog);
418
419	if (prog->expected_attach_type != BPF_XDP_CPUMAP ||
420	    !bpf_prog_map_compatible(map, prog)) {
421		bpf_prog_put(prog);
422		return -EINVAL;
423	}
424
425	rcpu->value.bpf_prog.id = prog->aux->id;
426	rcpu->prog = prog;
427
428	return 0;
429}
430
431static struct bpf_cpu_map_entry *
432__cpu_map_entry_alloc(struct bpf_map *map, struct bpf_cpumap_val *value,
433		      u32 cpu)
434{
435	int numa, err, i, fd = value->bpf_prog.fd;
436	gfp_t gfp = GFP_KERNEL | __GFP_NOWARN;
437	struct bpf_cpu_map_entry *rcpu;
438	struct xdp_bulk_queue *bq;
439
440	/* Have map->numa_node, but choose node of redirect target CPU */
441	numa = cpu_to_node(cpu);
442
443	rcpu = bpf_map_kmalloc_node(map, sizeof(*rcpu), gfp | __GFP_ZERO, numa);
444	if (!rcpu)
445		return NULL;
446
447	/* Alloc percpu bulkq */
448	rcpu->bulkq = bpf_map_alloc_percpu(map, sizeof(*rcpu->bulkq),
449					   sizeof(void *), gfp);
450	if (!rcpu->bulkq)
451		goto free_rcu;
452
453	for_each_possible_cpu(i) {
454		bq = per_cpu_ptr(rcpu->bulkq, i);
455		bq->obj = rcpu;
456		local_lock_init(&bq->bq_lock);
457	}
458
459	/* Alloc queue */
460	rcpu->queue = bpf_map_kmalloc_node(map, sizeof(*rcpu->queue), gfp,
461					   numa);
462	if (!rcpu->queue)
463		goto free_bulkq;
464
465	err = ptr_ring_init(rcpu->queue, value->qsize, gfp);
466	if (err)
467		goto free_queue;
468
469	rcpu->cpu    = cpu;
470	rcpu->map_id = map->id;
471	rcpu->value.qsize  = value->qsize;
472	gro_init(&rcpu->gro);
473
474	if (fd > 0 && __cpu_map_load_bpf_program(rcpu, map, fd))
475		goto free_ptr_ring;
476
477	/* Setup kthread */
478	init_completion(&rcpu->kthread_running);
479	rcpu->kthread = kthread_create_on_node(cpu_map_kthread_run, rcpu, numa,
480					       "cpumap/%d/map:%d", cpu,
481					       map->id);
482	if (IS_ERR(rcpu->kthread))
483		goto free_prog;
484
485	/* Make sure kthread runs on a single CPU */
486	kthread_bind(rcpu->kthread, cpu);
487	wake_up_process(rcpu->kthread);
488
489	/* Make sure kthread has been running, so kthread_stop() will not
490	 * stop the kthread prematurely and all pending frames or skbs
491	 * will be handled by the kthread before kthread_stop() returns.
492	 */
493	wait_for_completion(&rcpu->kthread_running);
494
495	return rcpu;
496
497free_prog:
498	if (rcpu->prog)
499		bpf_prog_put(rcpu->prog);
500free_ptr_ring:
501	gro_cleanup(&rcpu->gro);
502	ptr_ring_cleanup(rcpu->queue, NULL);
503free_queue:
504	kfree(rcpu->queue);
505free_bulkq:
506	free_percpu(rcpu->bulkq);
507free_rcu:
508	kfree(rcpu);
509	return NULL;
510}
511
512static void __cpu_map_entry_free(struct work_struct *work)
513{
514	struct bpf_cpu_map_entry *rcpu;
515
516	/* This cpu_map_entry have been disconnected from map and one
517	 * RCU grace-period have elapsed. Thus, XDP cannot queue any
518	 * new packets and cannot change/set flush_needed that can
519	 * find this entry.
520	 */
521	rcpu = container_of(to_rcu_work(work), struct bpf_cpu_map_entry, free_work);
522
523	/* kthread_stop will wake_up_process and wait for it to complete.
524	 * cpu_map_kthread_run() makes sure the pointer ring is empty
525	 * before exiting.
526	 */
527	kthread_stop(rcpu->kthread);
528
529	if (rcpu->prog)
530		bpf_prog_put(rcpu->prog);
531	gro_cleanup(&rcpu->gro);
532	/* The queue should be empty at this point */
533	__cpu_map_ring_cleanup(rcpu->queue);
534	ptr_ring_cleanup(rcpu->queue, NULL);
535	kfree(rcpu->queue);
536	free_percpu(rcpu->bulkq);
537	kfree(rcpu);
538}
539
540/* After the xchg of the bpf_cpu_map_entry pointer, we need to make sure the old
541 * entry is no longer in use before freeing. We use queue_rcu_work() to call
542 * __cpu_map_entry_free() in a separate workqueue after waiting for an RCU grace
543 * period. This means that (a) all pending enqueue and flush operations have
544 * completed (because of the RCU callback), and (b) we are in a workqueue
545 * context where we can stop the kthread and wait for it to exit before freeing
546 * everything.
547 */
548static void __cpu_map_entry_replace(struct bpf_cpu_map *cmap,
549				    u32 key_cpu, struct bpf_cpu_map_entry *rcpu)
550{
551	struct bpf_cpu_map_entry *old_rcpu;
552
553	old_rcpu = unrcu_pointer(xchg(&cmap->cpu_map[key_cpu], RCU_INITIALIZER(rcpu)));
554	if (old_rcpu) {
555		INIT_RCU_WORK(&old_rcpu->free_work, __cpu_map_entry_free);
556		queue_rcu_work(system_percpu_wq, &old_rcpu->free_work);
557	}
558}
559
560static long cpu_map_delete_elem(struct bpf_map *map, void *key)
561{
562	struct bpf_cpu_map *cmap = container_of(map, struct bpf_cpu_map, map);
563	u32 key_cpu = *(u32 *)key;
564
565	if (key_cpu >= map->max_entries)
566		return -EINVAL;
567
568	/* notice caller map_delete_elem() uses rcu_read_lock() */
569	__cpu_map_entry_replace(cmap, key_cpu, NULL);
570	return 0;
571}
572
573static long cpu_map_update_elem(struct bpf_map *map, void *key, void *value,
574				u64 map_flags)
575{
576	struct bpf_cpu_map *cmap = container_of(map, struct bpf_cpu_map, map);
577	struct bpf_cpumap_val cpumap_value = {};
578	struct bpf_cpu_map_entry *rcpu;
579	/* Array index key correspond to CPU number */
580	u32 key_cpu = *(u32 *)key;
581
582	memcpy(&cpumap_value, value, map->value_size);
583
584	if (unlikely(map_flags > BPF_EXIST))
585		return -EINVAL;
586	if (unlikely(key_cpu >= cmap->map.max_entries))
587		return -E2BIG;
588	if (unlikely(map_flags == BPF_NOEXIST))
589		return -EEXIST;
590	if (unlikely(cpumap_value.qsize > 16384)) /* sanity limit on qsize */
591		return -EOVERFLOW;
592
593	/* Make sure CPU is a valid possible cpu */
594	if (key_cpu >= nr_cpumask_bits || !cpu_possible(key_cpu))
595		return -ENODEV;
596
597	if (cpumap_value.qsize == 0) {
598		rcpu = NULL; /* Same as deleting */
599	} else {
600		/* Updating qsize cause re-allocation of bpf_cpu_map_entry */
601		rcpu = __cpu_map_entry_alloc(map, &cpumap_value, key_cpu);
602		if (!rcpu)
603			return -ENOMEM;
604	}
605	rcu_read_lock();
606	__cpu_map_entry_replace(cmap, key_cpu, rcpu);
607	rcu_read_unlock();
608	return 0;
609}
610
611static void cpu_map_free(struct bpf_map *map)
612{
613	struct bpf_cpu_map *cmap = container_of(map, struct bpf_cpu_map, map);
614	u32 i;
615
616	/* At this point bpf_prog->aux->refcnt == 0 and this map->refcnt == 0,
617	 * so the bpf programs (can be more than one that used this map) were
618	 * disconnected from events. Wait for outstanding critical sections in
619	 * these programs to complete. synchronize_rcu() below not only
620	 * guarantees no further "XDP/bpf-side" reads against
621	 * bpf_cpu_map->cpu_map, but also ensure pending flush operations
622	 * (if any) are completed.
623	 */
624	synchronize_rcu();
625
626	/* The only possible user of bpf_cpu_map_entry is
627	 * cpu_map_kthread_run().
628	 */
629	for (i = 0; i < cmap->map.max_entries; i++) {
630		struct bpf_cpu_map_entry *rcpu;
631
632		rcpu = rcu_dereference_raw(cmap->cpu_map[i]);
633		if (!rcpu)
634			continue;
635
636		/* Stop kthread and cleanup entry directly */
637		__cpu_map_entry_free(&rcpu->free_work.work);
638	}
639	bpf_map_area_free(cmap->cpu_map);
640	bpf_map_area_free(cmap);
641}
642
643/* Elements are kept alive by RCU; either by rcu_read_lock() (from syscall) or
644 * by local_bh_disable() (from XDP calls inside NAPI). The
645 * rcu_read_lock_bh_held() below makes lockdep accept both.
646 */
647static void *__cpu_map_lookup_elem(struct bpf_map *map, u32 key)
648{
649	struct bpf_cpu_map *cmap = container_of(map, struct bpf_cpu_map, map);
650	struct bpf_cpu_map_entry *rcpu;
651
652	if (key >= map->max_entries)
653		return NULL;
654
655	rcpu = rcu_dereference_check(cmap->cpu_map[key],
656				     rcu_read_lock_bh_held());
657	return rcpu;
658}
659
660static void *cpu_map_lookup_elem(struct bpf_map *map, void *key)
661{
662	struct bpf_cpu_map_entry *rcpu =
663		__cpu_map_lookup_elem(map, *(u32 *)key);
664
665	return rcpu ? &rcpu->value : NULL;
666}
667
668static int cpu_map_get_next_key(struct bpf_map *map, void *key, void *next_key)
669{
670	struct bpf_cpu_map *cmap = container_of(map, struct bpf_cpu_map, map);
671	u32 index = key ? *(u32 *)key : U32_MAX;
672	u32 *next = next_key;
673
674	if (index >= cmap->map.max_entries) {
675		*next = 0;
676		return 0;
677	}
678
679	if (index == cmap->map.max_entries - 1)
680		return -ENOENT;
681	*next = index + 1;
682	return 0;
683}
684
685static long cpu_map_redirect(struct bpf_map *map, u64 index, u64 flags)
686{
687	return __bpf_xdp_redirect_map(map, index, flags, 0,
688				      __cpu_map_lookup_elem);
689}
690
691static u64 cpu_map_mem_usage(const struct bpf_map *map)
692{
693	u64 usage = sizeof(struct bpf_cpu_map);
694
695	/* Currently the dynamically allocated elements are not counted */
696	usage += (u64)map->max_entries * sizeof(struct bpf_cpu_map_entry *);
697	return usage;
698}
699
700BTF_ID_LIST_SINGLE(cpu_map_btf_ids, struct, bpf_cpu_map)
701const struct bpf_map_ops cpu_map_ops = {
702	.map_meta_equal		= bpf_map_meta_equal,
703	.map_alloc		= cpu_map_alloc,
704	.map_free		= cpu_map_free,
705	.map_delete_elem	= cpu_map_delete_elem,
706	.map_update_elem	= cpu_map_update_elem,
707	.map_lookup_elem	= cpu_map_lookup_elem,
708	.map_get_next_key	= cpu_map_get_next_key,
709	.map_check_btf		= map_check_no_btf,
710	.map_mem_usage		= cpu_map_mem_usage,
711	.map_btf_id		= &cpu_map_btf_ids[0],
712	.map_redirect		= cpu_map_redirect,
713};
714
715static void bq_flush_to_queue(struct xdp_bulk_queue *bq)
716{
717	struct bpf_cpu_map_entry *rcpu = bq->obj;
718	unsigned int processed = 0, drops = 0;
719	const int to_cpu = rcpu->cpu;
720	struct ptr_ring *q;
721	int i;
722
723	lockdep_assert_held(&bq->bq_lock);
724
725	if (unlikely(!bq->count))
726		return;
727
728	q = rcpu->queue;
729	spin_lock(&q->producer_lock);
730
731	for (i = 0; i < bq->count; i++) {
732		struct xdp_frame *xdpf = bq->q[i];
733		int err;
734
735		err = __ptr_ring_produce(q, xdpf);
736		if (err) {
737			drops++;
738			xdp_return_frame_rx_napi(xdpf);
739		}
740		processed++;
741	}
742	bq->count = 0;
743	spin_unlock(&q->producer_lock);
744
745	__list_del_clearprev(&bq->flush_node);
746
747	/* Feedback loop via tracepoints */
748	trace_xdp_cpumap_enqueue(rcpu->map_id, processed, drops, to_cpu);
749}
750
751/* Runs under RCU-read-side, plus in softirq under NAPI protection.
752 * Thus, safe percpu variable access. PREEMPT_RT relies on
753 * local_lock_nested_bh() to serialise access to the per-CPU bq.
754 */
755static void bq_enqueue(struct bpf_cpu_map_entry *rcpu, struct xdp_frame *xdpf)
756{
757	struct xdp_bulk_queue *bq;
758
759	local_lock_nested_bh(&rcpu->bulkq->bq_lock);
760	bq = this_cpu_ptr(rcpu->bulkq);
761
762	if (unlikely(bq->count == CPU_MAP_BULK_SIZE))
763		bq_flush_to_queue(bq);
764
765	/* Notice, xdp_buff/page MUST be queued here, long enough for
766	 * driver to code invoking us to finished, due to driver
767	 * (e.g. ixgbe) recycle tricks based on page-refcnt.
768	 *
769	 * Thus, incoming xdp_frame is always queued here (else we race
770	 * with another CPU on page-refcnt and remaining driver code).
771	 * Queue time is very short, as driver will invoke flush
772	 * operation, when completing napi->poll call.
773	 */
774	bq->q[bq->count++] = xdpf;
775
776	if (!bq->flush_node.prev) {
777		struct list_head *flush_list = bpf_net_ctx_get_cpu_map_flush_list();
778
779		list_add(&bq->flush_node, flush_list);
780	}
781
782	local_unlock_nested_bh(&rcpu->bulkq->bq_lock);
783}
784
785int cpu_map_enqueue(struct bpf_cpu_map_entry *rcpu, struct xdp_frame *xdpf,
786		    struct net_device *dev_rx)
787{
788	/* Info needed when constructing SKB on remote CPU */
789	xdpf->dev_rx = dev_rx;
790
791	bq_enqueue(rcpu, xdpf);
792	return 0;
793}
794
795int cpu_map_generic_redirect(struct bpf_cpu_map_entry *rcpu,
796			     struct sk_buff *skb)
797{
798	int ret;
799
800	__skb_pull(skb, skb->mac_len);
801	skb_set_redirected(skb, false);
802	__ptr_set_bit(0, &skb);
803
804	ret = ptr_ring_produce(rcpu->queue, skb);
805	if (ret < 0)
806		goto trace;
807
808	wake_up_process(rcpu->kthread);
809trace:
810	trace_xdp_cpumap_enqueue(rcpu->map_id, !ret, !!ret, rcpu->cpu);
811	return ret;
812}
813
814void __cpu_map_flush(struct list_head *flush_list)
815{
816	struct xdp_bulk_queue *bq, *tmp;
817
818	list_for_each_entry_safe(bq, tmp, flush_list, flush_node) {
819		local_lock_nested_bh(&bq->obj->bulkq->bq_lock);
820		bq_flush_to_queue(bq);
821		local_unlock_nested_bh(&bq->obj->bulkq->bq_lock);
822
823		/* If already running, costs spin_lock_irqsave + smb_mb */
824		wake_up_process(bq->obj->kthread);
825	}
826}