mempool_monitor.py raw
1 #!/usr/bin/env python3
2 # Copyright (c) 2022-present The Bitcoin Core developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5
6 """ Example logging Bitcoin Core mempool events using the mempool:added,
7 mempool:removed, mempool:replaced, and mempool:rejected tracepoints. """
8
9 import curses
10 import sys
11 from datetime import datetime, timezone
12
13 from bcc import BPF, USDT
14
15 # BCC: The C program to be compiled to an eBPF program (by BCC) and loaded into
16 # a sandboxed Linux kernel VM.
17 PROGRAM = """
18 # include <uapi/linux/ptrace.h>
19
20 // The longest rejection reason is 114 chars and is generated in case of SCRIPT_ERR_EVAL_FALSE by
21 // strprintf("block-script-verify-flag-failed (%s)", ScriptErrorString(check.GetScriptError()))
22 #define MAX_REJECT_REASON_LENGTH 114
23 // The longest string returned by RemovalReasonToString() is 'sizelimit'
24 #define MAX_REMOVAL_REASON_LENGTH 9
25 #define HASH_LENGTH 32
26
27 struct added_event
28 {
29 u8 hash[HASH_LENGTH];
30 s32 vsize;
31 s64 fee;
32 };
33
34 struct removed_event
35 {
36 u8 hash[HASH_LENGTH];
37 char reason[MAX_REMOVAL_REASON_LENGTH];
38 s32 vsize;
39 s64 fee;
40 u64 entry_time;
41 };
42
43 struct rejected_event
44 {
45 u8 hash[HASH_LENGTH];
46 char reason[MAX_REJECT_REASON_LENGTH];
47 };
48
49 struct replaced_event
50 {
51 u8 replaced_hash[HASH_LENGTH];
52 s32 replaced_vsize;
53 s64 replaced_fee;
54 u64 replaced_entry_time;
55 u8 replacement_hash[HASH_LENGTH];
56 s32 replacement_vsize;
57 s64 replacement_fee;
58 };
59
60 // BPF perf buffer to push the data to user space.
61 BPF_PERF_OUTPUT(added_events);
62 BPF_PERF_OUTPUT(removed_events);
63 BPF_PERF_OUTPUT(rejected_events);
64 BPF_PERF_OUTPUT(replaced_events);
65
66 int trace_added(struct pt_regs *ctx) {
67 struct added_event added = {};
68 void *phash = NULL;
69 bpf_usdt_readarg(1, ctx, &phash);
70 bpf_probe_read_user(&added.hash, sizeof(added.hash), phash);
71 bpf_usdt_readarg(2, ctx, &added.vsize);
72 bpf_usdt_readarg(3, ctx, &added.fee);
73
74 added_events.perf_submit(ctx, &added, sizeof(added));
75 return 0;
76 }
77
78 int trace_removed(struct pt_regs *ctx) {
79 struct removed_event removed = {};
80 void *phash = NULL, *preason = NULL;
81 bpf_usdt_readarg(1, ctx, &phash);
82 bpf_probe_read_user(&removed.hash, sizeof(removed.hash), phash);
83 bpf_usdt_readarg(2, ctx, &preason);
84 bpf_probe_read_user_str(&removed.reason, sizeof(removed.reason), preason);
85 bpf_usdt_readarg(3, ctx, &removed.vsize);
86 bpf_usdt_readarg(4, ctx, &removed.fee);
87 bpf_usdt_readarg(5, ctx, &removed.entry_time);
88
89 removed_events.perf_submit(ctx, &removed, sizeof(removed));
90 return 0;
91 }
92
93 int trace_rejected(struct pt_regs *ctx) {
94 struct rejected_event rejected = {};
95 void *phash = NULL, *preason = NULL;
96 bpf_usdt_readarg(1, ctx, &phash);
97 bpf_probe_read_user(&rejected.hash, sizeof(rejected.hash), phash);
98 bpf_usdt_readarg(2, ctx, &preason);
99 bpf_probe_read_user_str(&rejected.reason, sizeof(rejected.reason), preason);
100 rejected_events.perf_submit(ctx, &rejected, sizeof(rejected));
101 return 0;
102 }
103
104 int trace_replaced(struct pt_regs *ctx) {
105 struct replaced_event replaced = {};
106 void *phash_replaced = NULL, *phash_replacement = NULL;
107 bpf_usdt_readarg(1, ctx, &phash_replaced);
108 bpf_probe_read_user(&replaced.replaced_hash, sizeof(replaced.replaced_hash), phash_replaced);
109 bpf_usdt_readarg(2, ctx, &replaced.replaced_vsize);
110 bpf_usdt_readarg(3, ctx, &replaced.replaced_fee);
111 bpf_usdt_readarg(4, ctx, &replaced.replaced_entry_time);
112 bpf_usdt_readarg(5, ctx, &phash_replacement);
113 bpf_probe_read_user(&replaced.replacement_hash, sizeof(replaced.replacement_hash), phash_replacement);
114 bpf_usdt_readarg(6, ctx, &replaced.replacement_vsize);
115 bpf_usdt_readarg(7, ctx, &replaced.replacement_fee);
116
117 replaced_events.perf_submit(ctx, &replaced, sizeof(replaced));
118 return 0;
119 }
120 """
121
122
123 def main(pid):
124 print(f"Hooking into bitcoind with pid {pid}")
125 bitcoind_with_usdts = USDT(pid=int(pid))
126
127 # attaching the trace functions defined in the BPF program
128 # to the tracepoints
129 bitcoind_with_usdts.enable_probe(probe="mempool:added", fn_name="trace_added")
130 bitcoind_with_usdts.enable_probe(probe="mempool:removed", fn_name="trace_removed")
131 bitcoind_with_usdts.enable_probe(probe="mempool:replaced", fn_name="trace_replaced")
132 bitcoind_with_usdts.enable_probe(probe="mempool:rejected", fn_name="trace_rejected")
133 bpf = BPF(text=PROGRAM, usdt_contexts=[bitcoind_with_usdts])
134
135 events = []
136
137 def get_timestamp():
138 return datetime.now(timezone.utc)
139
140 def handle_added(_, data, size):
141 event = bpf["added_events"].event(data)
142 events.append((get_timestamp(), "added", event))
143
144 def handle_removed(_, data, size):
145 event = bpf["removed_events"].event(data)
146 events.append((get_timestamp(), "removed", event))
147
148 def handle_rejected(_, data, size):
149 event = bpf["rejected_events"].event(data)
150 events.append((get_timestamp(), "rejected", event))
151
152 def handle_replaced(_, data, size):
153 event = bpf["replaced_events"].event(data)
154 events.append((get_timestamp(), "replaced", event))
155
156 bpf["added_events"].open_perf_buffer(handle_added)
157 # By default, open_perf_buffer uses eight pages for a buffer, making for a total
158 # buffer size of 32k on most machines. In practice, this size is insufficient:
159 # Each `mempool:removed` event takes up 57 bytes in the buffer (32 bytes for txid,
160 # 9 bytes for removal reason, and 8 bytes each for vsize and fee). Full blocks
161 # contain around 2k transactions, requiring a buffer size of around 114kB. To cover
162 # this amount, 32 4k pages are required.
163 bpf["removed_events"].open_perf_buffer(handle_removed, page_cnt=32)
164 bpf["rejected_events"].open_perf_buffer(handle_rejected)
165 bpf["replaced_events"].open_perf_buffer(handle_replaced)
166
167 curses.wrapper(loop, bpf, events)
168
169
170 def loop(screen, bpf, events):
171 dashboard = Dashboard(screen)
172 while True:
173 try:
174 bpf.perf_buffer_poll(timeout=50)
175 dashboard.render(events)
176 except KeyboardInterrupt:
177 exit()
178
179
180 class Dashboard:
181 """Visualization of mempool state using ncurses."""
182
183 INFO_WIN_HEIGHT = 2
184 EVENT_WIN_HEIGHT = 7
185
186 def __init__(self, screen):
187 screen.nodelay(True)
188 curses.curs_set(False)
189 self._screen = screen
190 self._time_started = datetime.now(timezone.utc)
191 self._timestamps = {"added": [], "removed": [], "rejected": [], "replaced": []}
192 self._event_history = {"added": 0, "removed": 0, "rejected": 0, "replaced": 0}
193 self._init_windows()
194
195 def _init_windows(self):
196 """Initialize all windows."""
197 self._init_info_win()
198 self._init_event_count_win()
199 self._init_event_rate_win()
200 self._init_event_log_win()
201
202 @staticmethod
203 def create_win(x, y, height, width, title=None):
204 """Helper function to create generic windows and decorate them with box and title if requested."""
205 win = curses.newwin(height, width, x, y)
206 if title:
207 win.box()
208 win.addstr(0, 2, title, curses.A_BOLD)
209 return win
210
211 def _init_info_win(self):
212 """Create and populate the info window."""
213 self._info_win = Dashboard.create_win(
214 x=0, y=1, height=Dashboard.INFO_WIN_HEIGHT, width=22
215 )
216 self._info_win.addstr(0, 0, "Mempool Monitor", curses.A_REVERSE)
217 self._info_win.addstr(1, 0, "Press CTRL-C to stop.", curses.A_NORMAL)
218 self._info_win.refresh()
219
220 def _init_event_count_win(self):
221 """Create and populate the event count window."""
222 self._event_count_win = Dashboard.create_win(
223 x=3, y=1, height=Dashboard.EVENT_WIN_HEIGHT, width=37, title="Event count"
224 )
225 header = " {:<8} {:>8} {:>7} {:>7} "
226 self._event_count_win.addstr(
227 1, 1, header.format("Event", "total", "1 min", "10 min"), curses.A_UNDERLINE
228 )
229 self._event_count_win.refresh()
230
231 def _init_event_rate_win(self):
232 """Create and populate the event rate window."""
233 self._event_rate_win = Dashboard.create_win(
234 x=3, y=40, height=Dashboard.EVENT_WIN_HEIGHT, width=42, title="Event rate"
235 )
236 header = " {:<8} {:>9} {:>9} {:>9} "
237 self._event_rate_win.addstr(
238 1, 1, header.format("Event", "total", "1 min", "10 min"), curses.A_UNDERLINE
239 )
240 self._event_rate_win.refresh()
241
242 def _init_event_log_win(self):
243 """Create windows showing event log. This comprises a dummy boxed window and an
244 inset window so line breaks don't overwrite box."""
245 # dummy boxed window
246 num_rows, num_cols = self._screen.getmaxyx()
247 space_above = Dashboard.INFO_WIN_HEIGHT + 1 + Dashboard.EVENT_WIN_HEIGHT + 1
248 box_win_height = num_rows - space_above
249 box_win_width = num_cols - 2
250 win_box = Dashboard.create_win(
251 x=space_above,
252 y=1,
253 height=box_win_height,
254 width=box_win_width,
255 title="Event log",
256 )
257 # actual logging window
258 log_lines = box_win_height - 2 # top and bottom box lines
259 log_line_len = box_win_width - 2 - 1 # box lines and left padding
260 win = win_box.derwin(log_lines, log_line_len, 1, 2)
261 win.idlok(True)
262 win.scrollok(True)
263 win_box.refresh()
264 win.refresh()
265 self._event_log_win_box = win_box
266 self._event_log_win = win
267
268 def calculate_metrics(self, events):
269 """Calculate count and rate metrics."""
270 count, rate = {}, {}
271 for event_ts, event_type, event_data in events:
272 self._timestamps[event_type].append(event_ts)
273 for event_type, ts in self._timestamps.items():
274 # remove timestamps older than ten minutes but keep track of their
275 # count for the 'total' metric
276 #
277 self._event_history[event_type] += len(
278 [t for t in ts if Dashboard.timestamp_age(t) >= 600]
279 )
280 ts = [t for t in ts if Dashboard.timestamp_age(t) < 600]
281 self._timestamps[event_type] = ts
282 # count metric
283 count_1m = len([t for t in ts if Dashboard.timestamp_age(t) < 60])
284 count_10m = len(ts)
285 count_total = self._event_history[event_type] + len(ts)
286 count[event_type] = (count_total, count_1m, count_10m)
287 # rate metric
288 runtime = Dashboard.timestamp_age(self._time_started)
289 rate_1m = count_1m / min(60, runtime)
290 rate_10m = count_10m / min(600, runtime)
291 rate_total = count_total / runtime
292 rate[event_type] = (rate_total, rate_1m, rate_10m)
293 return count, rate
294
295 def _update_event_count(self, count):
296 """Update the event count window."""
297 w = self._event_count_win
298 row_format = " {:<8} {:>6}tx {:>5}tx {:>5}tx "
299 for line, metric in enumerate(["added", "removed", "replaced", "rejected"]):
300 w.addstr(2 + line, 1, row_format.format(metric, *count[metric]))
301 w.refresh()
302
303 def _update_event_rate(self, rate):
304 """Update the event rate window."""
305 w = self._event_rate_win
306 row_format = " {:<8} {:>5.1f}tx/s {:>5.1f}tx/s {:>5.1f}tx/s "
307 for line, metric in enumerate(["added", "removed", "replaced", "rejected"]):
308 w.addstr(2 + line, 1, row_format.format(metric, *rate[metric]))
309 w.refresh()
310
311 def _update_event_log(self, events):
312 """Update the event log window."""
313 w = self._event_log_win
314 for event in events:
315 w.addstr(Dashboard.parse_event(event) + "\n")
316 w.refresh()
317
318 def render(self, events):
319 """Render the dashboard."""
320 count, rate = self.calculate_metrics(events)
321 self._update_event_count(count)
322 self._update_event_rate(rate)
323 self._update_event_log(events)
324 events.clear()
325
326 @staticmethod
327 def parse_event(event):
328 """Converts events into human-readable messages"""
329
330 ts_dt, type_, data = event
331 ts = ts_dt.strftime("%H:%M:%SZ")
332 if type_ == "added":
333 return (
334 f"{ts} added {bytes(data.hash)[::-1].hex()}"
335 f" with feerate {data.fee/data.vsize:.2f} sat/vB"
336 f" ({data.fee} sat, {data.vsize} vbytes)"
337 )
338
339 if type_ == "removed":
340 return (
341 f"{ts} removed {bytes(data.hash)[::-1].hex()}"
342 f" with feerate {data.fee/data.vsize:.2f} sat/vB"
343 f" ({data.fee} sat, {data.vsize} vbytes)"
344 f" received {ts_dt.timestamp()-data.entry_time:.1f} seconds ago"
345 f": {data.reason.decode('UTF-8')}"
346 )
347
348 if type_ == "rejected":
349 return (
350 f"{ts} rejected {bytes(data.hash)[::-1].hex()}"
351 f": {data.reason.decode('UTF-8')}"
352 )
353
354 if type_ == "replaced":
355 return (
356 f"{ts} replaced {bytes(data.replaced_hash)[::-1].hex()}"
357 f" with feerate {data.replaced_fee/data.replaced_vsize:.2f} sat/vB"
358 f" received {ts_dt.timestamp()-data.replaced_entry_time:.1f} seconds ago"
359 f" ({data.replaced_fee} sat, {data.replaced_vsize} vbytes)"
360 f" with {bytes(data.replacement_hash)[::-1].hex()}"
361 f" with feerate {data.replacement_fee/data.replacement_vsize:.2f} sat/vB"
362 f" ({data.replacement_fee} sat, {data.replacement_vsize} vbytes)"
363 )
364
365 raise NotImplementedError("Unsupported event type: {type_}")
366
367 @staticmethod
368 def timestamp_age(timestamp):
369 """Return age of timestamp in seconds."""
370 return (datetime.now(timezone.utc) - timestamp).total_seconds()
371
372
373 if __name__ == "__main__":
374 if len(sys.argv) < 2:
375 print("USAGE: ", sys.argv[0], "<pid of bitcoind>")
376 exit(1)
377
378 pid = sys.argv[1]
379 main(pid)
380