// /linux/net/core/sock_reuseport.c
253  /**
254	 *  reuseport_select_sock - Select a socket from an SO_REUSEPORT group.
255	 *  @sk: First socket in the group.
256	 *  @hash: When no BPF filter is available, use this hash to select.
257	 *  @skb: skb to run through BPF filter.
258	 *  @hdr_len: BPF filter expects skb data pointer at payload data.  If
259	 *    the skb does not yet point at the payload, this parameter represents
260	 *    how far the pointer needs to advance to reach the payload.
261	 *  Returns a socket that should receive the packet (or NULL on error).
262	 */
263	struct sock *reuseport_select_sock(struct sock *sk,
264					   u32 hash,
265					   struct sk_buff *skb,
266					   int hdr_len)
267	{
268		struct sock_reuseport *reuse;
269		struct bpf_prog *prog;
270		struct sock *sk2 = NULL;
271		u16 socks;
272	
273		rcu_read_lock();
274		reuse = rcu_dereference(sk->sk_reuseport_cb);
275	
276		/* if memory allocation failed or add call is not yet complete */
277		if (!reuse)
278			goto out;
279	
280		prog = rcu_dereference(reuse->prog);
281		socks = READ_ONCE(reuse->num_socks);
282		if (likely(socks)) {
283			/* paired with smp_wmb() in reuseport_add_sock() */
284			smp_rmb();
285	
286			if (!prog || !skb)
287				goto select_by_hash;
288	
289			if (prog->type == BPF_PROG_TYPE_SK_REUSEPORT)
290				sk2 = bpf_run_sk_reuseport(reuse, sk, prog, skb, hash);
291			else
292				sk2 = run_bpf_filter(reuse, socks, prog, skb, hdr_len);
293	
294	select_by_hash:
295			/* no bpf or invalid bpf result: fall back to hash usage */
296			if (!sk2)
297				sk2 = reuse->socks[reciprocal_scale(hash, socks)];
298		}
299	
300	out:
301		rcu_read_unlock();
302		return sk2;
303	}

// linux/include/net/sock_reuseport.h
13	struct sock_reuseport {
14		struct rcu_head		rcu;
15	
16		u16			max_socks;	/* length of socks */
17		u16			num_socks;	/* elements in socks */
18		/* The last synq overflow event timestamp of this
19		 * reuse->socks[] group.
20		 */
21		unsigned int		synq_overflow_ts;
22		/* ID stays the same even after the size of socks[] grows. */
23		unsigned int		reuseport_id;
24		bool			bind_inany;
25		struct bpf_prog __rcu	*prog;		/* optional BPF sock selector */
26		struct sock		*socks[0];	/* array of sock pointers */
27	};

// /linux/net/ipv4/udp.c
442     if (sk->sk_reuseport) {
443					hash = udp_ehashfn(net, daddr, hnum,
444							   saddr, sport);
445					result = reuseport_select_sock(sk, hash, skb,
446								sizeof(struct udphdr));
447					if (result)
448						return result;
449				}

// /linux/net/ipv4/udp.c
412	static u32 udp_ehashfn(const struct net *net, const __be32 laddr,
413			       const __u16 lport, const __be32 faddr,
414			       const __be16 fport)
415	{
416		static u32 udp_ehash_secret __read_mostly;
417	
418		net_get_random_once(&udp_ehash_secret, sizeof(udp_ehash_secret));
419	
420		return __inet_ehashfn(laddr, lport, faddr, fport,
421				      udp_ehash_secret + net_hash_mix(net));
422	}

struct inet_hashinfo {
	/* This is for sockets with full identity only.  Sockets here will
	 * always be without wildcards and will have the following invariant:
	 *          TCP_ESTABLISHED <= sk->sk_state < TCP_CLOSE
	 */
	struct inet_ehash_bucket	*ehash;
	spinlock_t			*ehash_locks;
	unsigned int			ehash_mask;
	unsigned int			ehash_locks_mask;
	/* Ok, let's try this, I give up, we do need a local binding
	 * TCP hash as well as the others for fast bind/connect.
	 */
	struct kmem_cache		*bind_bucket_cachep;
	struct inet_bind_hashbucket	*bhash;
	unsigned int			bhash_size;
  
	/* The 2nd listener table hashed by local port and address */
	unsigned int			lhash2_mask;
	struct inet_listen_hashbucket	*lhash2;
	/* All the above members are written once at bootup and
	 * never written again _or_ are predominantly read-access.
	 *
	 * Now align to a new cache line as all the following members
	 * might be often dirty.
	 */
	/* All sockets in TCP_LISTEN state will be in listening_hash.
	 * This is the only table where wildcard'd TCP sockets can
	 * exist.  listening_hash is only hashed by local port number.
	 * If lhash2 is initialized, the same socket will also be hashed
	 * to lhash2 by port and address.
	 */
	struct inet_listen_hashbucket	listening_hash[INET_LHTABLE_SIZE] ____cacheline_aligned_in_smp;
};

// /include/net/request_sock.h
/** struct listen_sock - listen state
 *
 * @max_qlen_log - log_2 of maximal queued SYNs/REQUESTs
 */
struct listen_sock {
	u8			max_qlen_log;
	u8			synflood_warned;
	/* 2 bytes hole, try to use */
	int			qlen;
	int			qlen_young;
	int			clock_hand;
	u32			hash_rnd;
	u32			nr_table_entries;
	struct request_sock	*syn_table[0];
};

// /include/net/tcp_states.h
enum {
	TCP_ESTABLISHED = 1,
	TCP_SYN_SENT,
	TCP_SYN_RECV,
	TCP_FIN_WAIT1,
	TCP_FIN_WAIT2,
	TCP_TIME_WAIT,
	TCP_CLOSE,
	TCP_CLOSE_WAIT,
	TCP_LAST_ACK,
	TCP_LISTEN,
	TCP_CLOSING,	/* Now a valid state */
	TCP_NEW_SYN_RECV,
	TCP_MAX_STATES	/* Leave at the end! */
};

// linux/net/ipv4/udp.c
407 if (sk->sk_incoming_cpu == raw_smp_processor_id())
408			score++;

// linux/net/ipv4/udp.c
425	static struct sock *udp4_lib_lookup2(struct net *net,
426					     __be32 saddr, __be16 sport,
427					     __be32 daddr, unsigned int hnum,
428					     int dif, int sdif, bool exact_dif,
429					     struct udp_hslot *hslot2,
430					     struct sk_buff *skb)
431	{
432		struct sock *sk, *result;
433		int score, badness;
434		u32 hash = 0;
435	
436		result = NULL;
437		badness = 0;
438		udp_portaddr_for_each_entry_rcu(sk, &hslot2->head) {
439			score = compute_score(sk, net, saddr, sport,
440					      daddr, hnum, dif, sdif, exact_dif);
441			if (score > badness) {
442				if (sk->sk_reuseport) {
443					hash = udp_ehashfn(net, daddr, hnum,
444							   saddr, sport);
445					result = reuseport_select_sock(sk, hash, skb,
446								sizeof(struct udphdr));
447					if (result)
448						return result;
449				}
450				badness = score;
451				result = sk;
452			}
453		}
454		return result;
455	}

// 设置flag
if (setsockopt(fd, SOL_SOCKET, SO_ZEROCOPY, &one, sizeof(one)))
        error(1, errno, "setsockopt zerocopy");
// 发送数据
ret = send(fd, buf, sizeof(buf), MSG_ZEROCOPY);
// 查询完成情况
pfd.fd = fd;
pfd.events = 0;
if (poll(&pfd, 1, -1) != 1 || pfd.revents & POLLERR == 0)
        error(1, errno, "poll");
ret = recvmsg(fd, &msg, MSG_ERRQUEUE);
if (ret == -1)
        error(1, errno, "recvmsg");
read_notification(msg);

/**
 * get_user_pages_fast() - pin user pages in memory
 * @start:	starting user address
 * @nr_pages:	number of pages from start to pin
 * @gup_flags:	flags modifying pin behaviour
 * @pages:	array that receives pointers to the pages pinned.
 *		Should be at least nr_pages long.
 *
 * Attempt to pin user pages in memory without taking mm->mmap_sem.
 * If not successful, it will fall back to taking the lock and
 * calling get_user_pages().
 *
 * Returns number of pages pinned. This may be fewer than the number
 * requested. If nr_pages is 0 or negative, returns 0. If no pages
 * were pinned, returns -errno.
 */
int get_user_pages_fast(unsigned long start, int nr_pages,
			unsigned int gup_flags, struct page **pages)
{
	//...
}

// echo-naive
while (1) {
		int n = recv(cd, buf, sizeof(buf), 0);
		if (n < 0) {
			//...
		}
		if (n == 0) {
			/...
		}

		int m = send(cd, buf, n, MSG_NOSIGNAL);
		if (m < 0) {
			//...
		}
		if (m == 0) {
			 break;
		}
}

// ehco-sockmap
SEC("prog_parser")
int _prog_parser(struct __sk_buff *skb) {
	return skb->len;
}

SEC("prog_verdict")
int _prog_verdict(struct __sk_buff *skb) {
	uint32_t idx = 0;
	return bpf_sk_redirect_map(skb, &sock_map, idx, 0);
}

struct ring_queue {
	int flags;		/* Flags supplied at creation */

	/* Ring producer status */
	struct prod {
		u32 watermark;	/* Maximum items before EDQUOT */
		u32 sp_enqueue;	/* True, if single producer */
		u32 size;	/* Size of ring */
		u32 mask;	/* Mask (size-1) of ring */
		u32 head;	/* Producer head */
		u32 tail;	/* Producer tail */
	} prod ____cacheline_aligned_in_smp;

	/* Ring consumer status */
	struct cons {
		u32 sc_dequeue;	/* True, if single consumer */
		u32 size;	/* Size of the ring */
		u32 mask;	/* Mask (size-1) of ring */
		u32 head;	/* Consumer head */
		u32 tail;	/* Consumer tail */
#ifdef CONFIG_LIB_RING_QUEUE_SPLIT_PROD_CONS
	} cons ____cacheline_aligned_in_smp;
#else
	} cons;
#endif

	/* Memory space of ring starts here.
	 * not volatile so need to be careful
	 * about compiler re-ordering */
	void *ring[0] ____cacheline_aligned_in_smp;
};

static inline int
__ring_queue_mp_do_enqueue(struct ring_queue *r, void * const *obj_table,
			 unsigned n, enum ring_queue_queue_behavior behavior)
{
	u32 prod_head, prod_next;
	u32 cons_tail, free_entries;
	const unsigned max = n;
	int success;
	unsigned i;
	u32 mask = r->prod.mask;
	int ret;

	/* move prod.head atomically */
	do {
		/* Reset n to the initial burst count */
		n = max;

		prod_head = READ_ONCE(r->prod.head);
		cons_tail = READ_ONCE(r->cons.tail);
		/* The subtraction is done between two unsigned 32bits value
		 * (the result is always modulo 32 bits even if we have
		 * prod_head > cons_tail). So 'free_entries' is always between 0
		 * and size(ring)-1. */
		free_entries = (mask + cons_tail - prod_head);

		/* check that we have enough room in ring */
		if (unlikely(n > free_entries)) {
			if (behavior == RING_QUEUE_FIXED) {
				return -ENOBUFS;
			} else {
				/* No free entry available */
				if (unlikely(free_entries == 0)) {
					return 0;
				}

				n = free_entries;
			}
		}

		prod_next = prod_head + n;
		success = (cmpxchg(&r->prod.head, prod_head,
				   prod_next) == prod_head);
	} while (unlikely(success == 0));
	/* smp_rmb() for cons.tail is implicit by cmpxchg */

	ENQUEUE_PTRS(); /* write entries in ring */
	smp_wmb(); /* matching dequeue LOADs */

	/* if we exceed the watermark */
	if (unlikely(((mask + 1) - free_entries + n) > r->prod.watermark)) {
		ret = (behavior == RING_QUEUE_FIXED) ? -EDQUOT :
				(int)(n | RING_QUEUE_QUOT_EXCEED);
	} else {
		ret = (behavior == RING_QUEUE_FIXED) ? 0 : n;
	}

	/* If there are other enqueues in progress that preceeded us,
	 * we need to wait for them to complete
	 */
	while (unlikely(READ_ONCE(r->prod.tail) != prod_head))
		cpu_relax();

	WRITE_ONCE(r->prod.tail, prod_next);
	return ret;
}

struct alf_actor {
	u32 head;
	u32 tail;
};

struct alf_queue {
	u32 size;
	u32 mask;
	u32 flags;
	struct alf_actor producer ____cacheline_aligned_in_smp;
	struct alf_actor consumer ____cacheline_aligned_in_smp;
	void *ring[0] ____cacheline_aligned_in_smp;
};

static inline int
alf_mp_enqueue(const u32 n;
	       struct alf_queue *q, void *ptr[n], const u32 n)
{
	u32 p_head, p_next, c_tail, space;

	/* Reserve part of the array for enqueue STORE/WRITE */
	do {
		p_head = READ_ONCE(q->producer.head);
		c_tail = READ_ONCE(q->consumer.tail);/* as smp_load_aquire */

		space = q->size + c_tail - p_head;
		if (n > space)
			return 0;

		p_next = p_head + n;
	}
	while (unlikely(cmpxchg(&q->producer.head, p_head, p_next) != p_head));
	/* The memory barrier of smp_load_acquire(&q->consumer.tail)
	 * is satisfied by cmpxchg implicit full memory barrier
	 */

	/* STORE the elems into the queue array */
	__helper_alf_enqueue_store(p_head, q, ptr, n);
	smp_wmb(); /* Write-Memory-Barrier matching dequeue LOADs */

	/* Wait for other concurrent preceding enqueues not yet done,
	 * this part make us none-wait-free and could be problematic
	 * in case of congestion with many CPUs
	 */
	while (unlikely(READ_ONCE(q->producer.tail) != p_head))
		cpu_relax();
	/* Mark this enq done and avail for consumption */
	WRITE_ONCE(q->producer.tail, p_next);

	return n;
}