—— coding: utf-8 ——

"““用Python模拟eBPF三种映射类型的核心语义.””" import random class Hash: "““模拟BPF Hash映射：键值对存储，O(1)查询.””" def init(self): self.store = {} def update(self, key, inc=1): self.store[key] = self.store.get(key, 0) + inc def lookup(self, key): return self.store.get(key, 0) class Array: "““模拟BPF Array映射：按索引访问，预分配固定大小.””" def init(self, size): self.size = size self.store = [0] * size def update(self, idx, inc=1): if 0 <= idx < self.size: self.store[idx] += inc def lookup(self, idx): return self.store[idx] if 0 <= idx < self.size else 0 class Ringbuf: "““模拟BPF Ringbuf: 环形缓冲，流式读写.””" def init(self, capacity): self.capacity = capacity self.buffer = [None] * capacity self.head = 0 self.tail = 0 def push(self, item): self.buffer[self.tail] = item self.tail = (self.tail + 1) % self.capacity if self.head == self.tail: self.head = (self.head + 1) % self.capacity def pop(self): if self.head == self.tail: return None item = self.buffer[self.head] self.head = (self.head + 1) % self.capacity return item def simulate_packets(n=10000): "““模拟n个数据包到达，三种映射同时记录.””" h = Hash() a = Array(16) r = Ringbuf(128) for _ in range(n): ip = "10.0.0.{}“.format(random.randint(0, 15)) h.update(ip, 1) # 键：IP, 值：计数 a.update(int(ip.rsplit(”.“, 1)[1]), 1) # 索引：IP最后一段 r.push(ip) # 环形缓冲存事件流 top_ip = max(h.store, key=h.store.get) print(“Hash统计 最频繁IP:”, top_ip, h.lookup(top_ip)) print(“Array统计 IP段分布：”, a.store) print(“Ringbuf 最近5条IP:”, [r.pop() for _ in range(5)]) if name == ”main": simulate_packets(10000) ```