code:
#!/usr/bin/env perl
# This is a POC for reading the text representation of trace output related to
# page reclaim. It makes an attempt to extract some high-level information on
# what is going on. The accuracy of the parser may vary
#
# Example usage: trace-vmscan-postprocess.pl < /sys/kernel/debug/tracing/trace_pipe
# other options
#   --read-procstat	If the trace lacks process info, get it from /proc
#   --ignore-pid	Aggregate processes of the same name together
#
# Copyright (c) IBM Corporation 2009
# Author: Mel Gorman <mel@csn.ul.ie>
use strict;
use Getopt::Long;

# Tracepoint events
use constant MM_VMSCAN_DIRECT_RECLAIM_BEGIN	=> 1;
use constant MM_VMSCAN_DIRECT_RECLAIM_END	=> 2;
use constant MM_VMSCAN_KSWAPD_WAKE		=> 3;
use constant MM_VMSCAN_KSWAPD_SLEEP		=> 4;
use constant MM_VMSCAN_LRU_SHRINK_ACTIVE	=> 5;
use constant MM_VMSCAN_LRU_SHRINK_INACTIVE	=> 6;
use constant MM_VMSCAN_LRU_ISOLATE		=> 7;
use constant MM_VMSCAN_WRITEPAGE_FILE_SYNC	=> 8;
use constant MM_VMSCAN_WRITEPAGE_ANON_SYNC	=> 9;
use constant MM_VMSCAN_WRITEPAGE_FILE_ASYNC	=> 10;
use constant MM_VMSCAN_WRITEPAGE_ANON_ASYNC	=> 11;
use constant MM_VMSCAN_WRITEPAGE_ASYNC		=> 12;
use constant EVENT_UNKNOWN			=> 13;

# Per-order events
use constant MM_VMSCAN_DIRECT_RECLAIM_BEGIN_PERORDER => 11;
use constant MM_VMSCAN_WAKEUP_KSWAPD_PERORDER 	=> 12;
use constant MM_VMSCAN_KSWAPD_WAKE_PERORDER	=> 13;
use constant HIGH_KSWAPD_REWAKEUP_PERORDER	=> 14;

# Constants used to track state
use constant STATE_DIRECT_BEGIN 		=> 15;
use constant STATE_DIRECT_ORDER 		=> 16;
use constant STATE_KSWAPD_BEGIN			=> 17;
use constant STATE_KSWAPD_ORDER			=> 18;

# High-level events extrapolated from tracepoints
use constant HIGH_DIRECT_RECLAIM_LATENCY	=> 19;
use constant HIGH_KSWAPD_LATENCY		=> 20;
use constant HIGH_KSWAPD_REWAKEUP		=> 21;
use constant HIGH_NR_SCANNED			=> 22;
use constant HIGH_NR_TAKEN			=> 23;
use constant HIGH_NR_RECLAIMED			=> 24;
use constant HIGH_NR_FILE_SCANNED		=> 25;
use constant HIGH_NR_ANON_SCANNED		=> 26;
use constant HIGH_NR_FILE_RECLAIMED		=> 27;
use constant HIGH_NR_ANON_RECLAIMED		=> 28;

my %perprocesspid;
my %perprocess;
my %last_procmap;
my $opt_ignorepid;
my $opt_read_procstat;

my $total_wakeup_kswapd;
my ($total_direct_reclaim, $total_direct_nr_scanned);
my ($total_direct_nr_file_scanned, $total_direct_nr_anon_scanned);
my ($total_direct_latency, $total_kswapd_latency);
my ($total_direct_nr_reclaimed);
my ($total_direct_nr_file_reclaimed, $total_direct_nr_anon_reclaimed);
my ($total_direct_writepage_file_sync, $total_direct_writepage_file_async);
my ($total_direct_writepage_anon_sync, $total_direct_writepage_anon_async);
my ($total_kswapd_nr_scanned, $total_kswapd_wake);
my ($total_kswapd_nr_file_scanned, $total_kswapd_nr_anon_scanned);
my ($total_kswapd_writepage_file_sync, $total_kswapd_writepage_file_async);
my ($total_kswapd_writepage_anon_sync, $total_kswapd_writepage_anon_async);
my ($total_kswapd_nr_reclaimed);
my ($total_kswapd_nr_file_reclaimed, $total_kswapd_nr_anon_reclaimed);

# Catch sigint and exit on request
my $sigint_report = 0;
my $sigint_exit = 0;
my $sigint_pending = 0;
my $sigint_received = 0;
sub sigint_handler {
	my $current_time = time;
	if ($current_time - 2 > $sigint_received) {
		print "SIGINT received, report pending. Hit ctrl-c again to exit\n";
		$sigint_report = 1;
	} else {
		if (!$sigint_exit) {
			print "Second SIGINT received quickly, exiting\n";
		}
		$sigint_exit++;
	}

	if ($sigint_exit > 3) {
		print "Many SIGINTs received, exiting now without report\n";
		exit;
	}

	$sigint_received = $current_time;
	$sigint_pending = 1;
}
$SIG{INT} = "sigint_handler";

# Parse command line options
GetOptions(
	'ignore-pid'	 =>	\$opt_ignorepid,
	'read-procstat'	 =>	\$opt_read_procstat,
);

# Defaults for dynamically discovered regex's
my $regex_direct_begin_default = 'order=([0-9]*) may_writepage=([0-9]*) gfp_flags=([A-Z_|]*)';
my $regex_direct_end_default = 'nr_reclaimed=([0-9]*)';
my $regex_kswapd_wake_default = 'nid=([0-9]*) order=([0-9]*)';
my $regex_kswapd_sleep_default = 'nid=([0-9]*)';
my $regex_wakeup_kswapd_default = 'nid=([0-9]*) zid=([0-9]*) order=([0-9]*) gfp_flags=([A-Z_|]*)';
my $regex_lru_isolate_default = 'isolate_mode=([0-9]*) classzone_idx=([0-9]*) order=([0-9]*) nr_requested=([0-9]*) nr_scanned=([0-9]*) nr_skipped=([0-9]*) nr_taken=([0-9]*) lru=([a-z_]*)';
my $regex_lru_shrink_inactive_default = 'nid=([0-9]*) nr_scanned=([0-9]*) nr_reclaimed=([0-9]*) nr_dirty=([0-9]*) nr_writeback=([0-9]*) nr_congested=([0-9]*) nr_immediate=([0-9]*) nr_activate_anon=([0-9]*) nr_activate_file=([0-9]*) nr_ref_keep=([0-9]*) nr_unmap_fail=([0-9]*) priority=([0-9]*) flags=([A-Z_|]*)';
my $regex_lru_shrink_active_default = 'lru=([A-Z_]*) nr_scanned=([0-9]*) nr_rotated=([0-9]*) priority=([0-9]*)';
my $regex_writepage_default = 'page=([0-9a-f]*) pfn=([0-9]*) flags=([A-Z_|]*)';

# Dyanically discovered regex
my $regex_direct_begin;
my $regex_direct_end;
my $regex_kswapd_wake;
my $regex_kswapd_sleep;
my $regex_wakeup_kswapd;
my $regex_lru_isolate;
my $regex_lru_shrink_inactive;
my $regex_lru_shrink_active;
my $regex_writepage;

# Static regex used. Specified like this for readability and for use with /o
#                      (process_pid)     (cpus      )   ( time  )   (tpoint    ) (details)
my $regex_traceevent = '\s*([a-zA-Z0-9-]*)\s*(\[[0-9]*\])(\s*[dX.][Nnp.][Hhs.][0-9a-fA-F.]*|)\s*([0-9.]*):\s*([a-zA-Z_]*):\s*(.*)';
my $regex_statname = '[-0-9]*\s\((.*)\).*';
my $regex_statppid = '[-0-9]*\s\(.*\)\s[A-Za-z]\s([0-9]*).*';

sub generate_traceevent_regex {
	my $event = shift;
	my $default = shift;
	my $regex;

	# Read the event format or use the default
	if (!open (FORMAT, "/sys/kernel/debug/tracing/events/$event/format")) {
		print("WARNING: Event $event format string not found\n");
		return $default;
	} else {
		my $line;
		while (!eof(FORMAT)) {
			$line = <FORMAT>;
			$line =~ s/, REC->.*//;
			if ($line =~ /^print fmt:\s"(.*)".*/) {
				$regex = $1;
				$regex =~ s/%s/\([0-9a-zA-Z|_]*\)/g;
				$regex =~ s/%p/\([0-9a-f]*\)/g;
				$regex =~ s/%d/\([-0-9]*\)/g;
				$regex =~ s/%ld/\([-0-9]*\)/g;
				$regex =~ s/%lu/\([0-9]*\)/g;
			}
		}
	}

	# Can't handle the print_flags stuff but in the context of this
	# script, it really doesn't matter
	$regex =~ s/\(REC.*\) \? __print_flags.*//;

	# Verify fields are in the right order
	my $tuple;
	foreach $tuple (split /\s/, $regex) {
		my ($key, $value) = split(/=/, $tuple);
		my $expected = shift;
		if ($key ne $expected) {
			print("WARNING: Format not as expected for event $event '$key' != '$expected'\n");
			$regex =~ s/$key=\((.*)\)/$key=$1/;
		}
	}

	if (defined shift) {
		die("Fewer fields than expected in format");
	}

	return $regex;
}

$regex_direct_begin = generate_traceevent_regex(
			"vmscan/mm_vmscan_direct_reclaim_begin",
			$regex_direct_begin_default,
			"order", "may_writepage",
			"gfp_flags");
$regex_direct_end = generate_traceevent_regex(
			"vmscan/mm_vmscan_direct_reclaim_end",
			$regex_direct_end_default,
			"nr_reclaimed");
$regex_kswapd_wake = generate_traceevent_regex(
			"vmscan/mm_vmscan_kswapd_wake",
			$regex_kswapd_wake_default,
			"nid", "order");
$regex_kswapd_sleep = generate_traceevent_regex(
			"vmscan/mm_vmscan_kswapd_sleep",
			$regex_kswapd_sleep_default,
			"nid");
$regex_wakeup_kswapd = generate_traceevent_regex(
			"vmscan/mm_vmscan_wakeup_kswapd",
			$regex_wakeup_kswapd_default,
			"nid", "zid", "order", "gfp_flags");
$regex_lru_isolate = generate_traceevent_regex(
			"vmscan/mm_vmscan_lru_isolate",
			$regex_lru_isolate_default,
			"isolate_mode", "classzone_idx", "order",
			"nr_requested", "nr_scanned", "nr_skipped", "nr_taken",
			"lru");
$regex_lru_shrink_inactive = generate_traceevent_regex(
			"vmscan/mm_vmscan_lru_shrink_inactive",
			$regex_lru_shrink_inactive_default,
			"nid", "nr_scanned", "nr_reclaimed", "nr_dirty", "nr_writeback",
			"nr_congested", "nr_immediate", "nr_activate_anon",
			"nr_activate_file", "nr_ref_keep",
			"nr_unmap_fail", "priority", "flags");
$regex_lru_shrink_active = generate_traceevent_regex(
			"vmscan/mm_vmscan_lru_shrink_active",
			$regex_lru_shrink_active_default,
			"nid", "zid",
			"lru",
			"nr_scanned", "nr_rotated", "priority");
$regex_writepage = generate_traceevent_regex(
			"vmscan/mm_vmscan_writepage",
			$regex_writepage_default,
			"page", "pfn", "flags");

sub read_statline($) {
	my $pid = $_[0];
	my $statline;

	if (open(STAT, "/proc/$pid/stat")) {
		$statline = <STAT>;
		close(STAT);
	}

	if ($statline eq '') {
		$statline = "-1 (UNKNOWN_PROCESS_NAME) R 0";
	}

	return $statline;
}

sub guess_process_pid($$) {
	my $pid = $_[0];
	my $statline = $_[1];

	if ($pid == 0) {
		return "swapper-0";
	}

	if ($statline !~ /$regex_statname/o) {
		die("Failed to math stat line for process name :: $statline");
	}
	return "$1-$pid";
}

# Convert sec.usec timestamp format
sub timestamp_to_ms($) {
	my $timestamp = $_[0];

	my ($sec, $usec) = split (/\./, $timestamp);
	return ($sec * 1000) + ($usec / 1000);
}

sub process_events {
	my $traceevent;
	my $process_pid;
	my $cpus;
	my $timestamp;
	my $tracepoint;
	my $details;
	my $statline;

	# Read each line of the event log
EVENT_PROCESS:
	while ($traceevent = <STDIN>) {
		if ($traceevent =~ /$regex_traceevent/o) {
			$process_pid = $1;
			$timestamp = $4;
			$tracepoint = $5;

			$process_pid =~ /(.*)-([0-9]*)$/;
			my $process = $1;
			my $pid = $2;

			if ($process eq "") {
				$process = $last_procmap{$pid};
				$process_pid = "$process-$pid";
			}
			$last_procmap{$pid} = $process;

			if ($opt_read_procstat) {
				$statline = read_statline($pid);
				if ($opt_read_procstat && $process eq '') {
					$process_pid = guess_process_pid($pid, $statline);
				}
			}
		} else {
			next;
		}

		# Perl Switch() sucks majorly
		if ($tracepoint eq "mm_vmscan_direct_reclaim_begin") {
			$timestamp = timestamp_to_ms($timestamp);
			$perprocesspid{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN}++;
			$perprocesspid{$process_pid}->{STATE_DIRECT_BEGIN} = $timestamp;

			$details = $6;
			if ($details !~ /$regex_direct_begin/o) {
				print "WARNING: Failed to parse mm_vmscan_direct_reclaim_begin as expected\n";
				print "         $details\n";
				print "         $regex_direct_begin\n";
				next;
			}
			my $order = $1;
			$perprocesspid{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN_PERORDER}[$order]++;
			$perprocesspid{$process_pid}->{STATE_DIRECT_ORDER} = $order;
		} elsif ($tracepoint eq "mm_vmscan_direct_reclaim_end") {
			# Count the event itself
			my $index = $perprocesspid{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_END};
			$perprocesspid{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_END}++;

			# Record how long direct reclaim took this time
			if (defined $perprocesspid{$process_pid}->{STATE_DIRECT_BEGIN}) {
				$timestamp = timestamp_to_ms($timestamp);
				my $order = $perprocesspid{$process_pid}->{STATE_DIRECT_ORDER};
				my $latency = ($timestamp - $perprocesspid{$process_pid}->{STATE_DIRECT_BEGIN});
				$perprocesspid{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$index] = "$order-$latency";
			}
		} elsif ($tracepoint eq "mm_vmscan_kswapd_wake") {
			$details = $6;
			if ($details !~ /$regex_kswapd_wake/o) {
				print "WARNING: Failed to parse mm_vmscan_kswapd_wake as expected\n";
				print "         $details\n";
				print "         $regex_kswapd_wake\n";
				next;
			}

			my $order = $2;
			$perprocesspid{$process_pid}->{STATE_KSWAPD_ORDER} = $order;
			if (!$perprocesspid{$process_pid}->{STATE_KSWAPD_BEGIN}) {
				$timestamp = timestamp_to_ms($timestamp);
				$perprocesspid{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE}++;
				$perprocesspid{$process_pid}->{STATE_KSWAPD_BEGIN} = $timestamp;
				$perprocesspid{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE_PERORDER}[$order]++;
			} else {
				$perprocesspid{$process_pid}->{HIGH_KSWAPD_REWAKEUP}++;
				$perprocesspid{$process_pid}->{HIGH_KSWAPD_REWAKEUP_PERORDER}[$order]++;
			}
		} elsif ($tracepoint eq "mm_vmscan_kswapd_sleep") {

			# Count the event itself
			my $index = $perprocesspid{$process_pid}->{MM_VMSCAN_KSWAPD_SLEEP};
			$perprocesspid{$process_pid}->{MM_VMSCAN_KSWAPD_SLEEP}++;

			# Record how long kswapd was awake
			$timestamp = timestamp_to_ms($timestamp);
			my $order = $perprocesspid{$process_pid}->{STATE_KSWAPD_ORDER};
			my $latency = ($timestamp - $perprocesspid{$process_pid}->{STATE_KSWAPD_BEGIN});
			$perprocesspid{$process_pid}->{HIGH_KSWAPD_LATENCY}[$index] = "$order-$latency";
			$perprocesspid{$process_pid}->{STATE_KSWAPD_BEGIN} = 0;
		} elsif ($tracepoint eq "mm_vmscan_wakeup_kswapd") {
			$perprocesspid{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD}++;

			$details = $6;
			if ($details !~ /$regex_wakeup_kswapd/o) {
				print "WARNING: Failed to parse mm_vmscan_wakeup_kswapd as expected\n";
				print "         $details\n";
				print "         $regex_wakeup_kswapd\n";
				next;
			}
			my $order = $3;
			$perprocesspid{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD_PERORDER}[$order]++;
		} elsif ($tracepoint eq "mm_vmscan_lru_isolate") {
			$details = $6;
			if ($details !~ /$regex_lru_isolate/o) {
				print "WARNING: Failed to parse mm_vmscan_lru_isolate as expected\n";
				print "         $details\n";
				print "         $regex_lru_isolate/o\n";
				next;
			}
			my $isolate_mode = $1;
			my $nr_scanned = $5;
			my $file = $8;

			# To closer match vmstat scanning statistics, only count isolate_both
			# and isolate_inactive as scanning. isolate_active is rotation
			# isolate_inactive == 1
			# isolate_active   == 2
			# isolate_both     == 3
			if ($isolate_mode != 2) {
				$perprocesspid{$process_pid}->{HIGH_NR_SCANNED} += $nr_scanned;
				if ($file =~ /_file/) {
					$perprocesspid{$process_pid}->{HIGH_NR_FILE_SCANNED} += $nr_scanned;
				} else {
					$perprocesspid{$process_pid}->{HIGH_NR_ANON_SCANNED} += $nr_scanned;
				}
			}
		} elsif ($tracepoint eq "mm_vmscan_lru_shrink_inactive") {
			$details = $6;
			if ($details !~ /$regex_lru_shrink_inactive/o) {
				print "WARNING: Failed to parse mm_vmscan_lru_shrink_inactive as expected\n";
				print "         $details\n";
				print "         $regex_lru_shrink_inactive/o\n";
				next;
			}

			my $nr_reclaimed = $3;
			my $flags = $13;
			my $file = 0;
			if ($flags =~ /RECLAIM_WB_FILE/) {
				$file = 1;
			}
			$perprocesspid{$process_pid}->{HIGH_NR_RECLAIMED} += $nr_reclaimed;
			if ($file) {
				$perprocesspid{$process_pid}->{HIGH_NR_FILE_RECLAIMED} += $nr_reclaimed;
			} else {
				$perprocesspid{$process_pid}->{HIGH_NR_ANON_RECLAIMED} += $nr_reclaimed;
			}
		} elsif ($tracepoint eq "mm_vmscan_writepage") {
			$details = $6;
			if ($details !~ /$regex_writepage/o) {
				print "WARNING: Failed to parse mm_vmscan_writepage as expected\n";
				print "         $details\n";
				print "         $regex_writepage\n";
				next;
			}

			my $flags = $3;
			my $file = 0;
			my $sync_io = 0;
			if ($flags =~ /RECLAIM_WB_FILE/) {
				$file = 1;
			}
			if ($flags =~ /RECLAIM_WB_SYNC/) {
				$sync_io = 1;
			}
			if ($sync_io) {
				if ($file) {
					$perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_SYNC}++;
				} else {
					$perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_SYNC}++;
				}
			} else {
				if ($file) {
					$perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_ASYNC}++;
				} else {
					$perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_ASYNC}++;
				}
			}
		} else {
			$perprocesspid{$process_pid}->{EVENT_UNKNOWN}++;
		}

		if ($sigint_pending) {
			last EVENT_PROCESS;
		}
	}
}

sub dump_stats {
	my $hashref = shift;
	my %stats = %$hashref;

	# Dump per-process stats
	my $process_pid;
	my $max_strlen = 0;

	# Get the maximum process name
	foreach $process_pid (keys %perprocesspid) {
		my $len = length($process_pid);
		if ($len > $max_strlen) {
			$max_strlen = $len;
		}
	}
	$max_strlen += 2;

	# Work out latencies
	printf("\n") if !$opt_ignorepid;
	printf("Reclaim latencies expressed as order-latency_in_ms\n") if !$opt_ignorepid;
	foreach $process_pid (keys %stats) {

		if (!$stats{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[0] &&
				!$stats{$process_pid}->{HIGH_KSWAPD_LATENCY}[0]) {
			next;
		}

		printf "%-" . $max_strlen . "s ", $process_pid if !$opt_ignorepid;
		my $index = 0;
		while (defined $stats{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$index] ||
			defined $stats{$process_pid}->{HIGH_KSWAPD_LATENCY}[$index]) {

			if ($stats{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$index]) {
				printf("%s ", $stats{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$index]) if !$opt_ignorepid;
				my ($dummy, $latency) = split(/-/, $stats{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$index]);
				$total_direct_latency += $latency;
			} else {
				printf("%s ", $stats{$process_pid}->{HIGH_KSWAPD_LATENCY}[$index]) if !$opt_ignorepid;
				my ($dummy, $latency) = split(/-/, $stats{$process_pid}->{HIGH_KSWAPD_LATENCY}[$index]);
				$total_kswapd_latency += $latency;
			}
			$index++;
		}
		print "\n" if !$opt_ignorepid;
	}

	# Print out process activity
	printf("\n");
	printf("%-" . $max_strlen . "s %8s %10s   %8s %8s  %8s %8s %8s %8s\n", "Process", "Direct",  "Wokeup", "Pages",   "Pages",   "Pages",   "Pages",     "Time");
	printf("%-" . $max_strlen . "s %8s %10s   %8s %8s  %8s %8s %8s %8s\n", "details", "Rclms",   "Kswapd", "Scanned", "Rclmed",  "Sync-IO", "ASync-IO",  "Stalled");
	foreach $process_pid (keys %stats) {

		if (!$stats{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN}) {
			next;
		}

		$total_direct_reclaim += $stats{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN};
		$total_wakeup_kswapd += $stats{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD};
		$total_direct_nr_scanned += $stats{$process_pid}->{HIGH_NR_SCANNED};
		$total_direct_nr_file_scanned += $stats{$process_pid}->{HIGH_NR_FILE_SCANNED};
		$total_direct_nr_anon_scanned += $stats{$process_pid}->{HIGH_NR_ANON_SCANNED};
		$total_direct_nr_reclaimed += $stats{$process_pid}->{HIGH_NR_RECLAIMED};
		$total_direct_nr_file_reclaimed += $stats{$process_pid}->{HIGH_NR_FILE_RECLAIMED};
		$total_direct_nr_anon_reclaimed += $stats{$process_pid}->{HIGH_NR_ANON_RECLAIMED};
		$total_direct_writepage_file_sync += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_SYNC};
		$total_direct_writepage_anon_sync += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_SYNC};
		$total_direct_writepage_file_async += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_ASYNC};

		$total_direct_writepage_anon_async += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_ASYNC};

		my $index = 0;
		my $this_reclaim_delay = 0;
		while (defined $stats{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$index]) {
			 my ($dummy, $latency) = split(/-/, $stats{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$index]);
			$this_reclaim_delay += $latency;
			$index++;
		}

		printf("%-" . $max_strlen . "s %8d %10d   %8u %8u  %8u %8u %8.3f",
			$process_pid,
			$stats{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN},
			$stats{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD},
			$stats{$process_pid}->{HIGH_NR_SCANNED},
			$stats{$process_pid}->{HIGH_NR_FILE_SCANNED},
			$stats{$process_pid}->{HIGH_NR_ANON_SCANNED},
			$stats{$process_pid}->{HIGH_NR_RECLAIMED},
			$stats{$process_pid}->{HIGH_NR_FILE_RECLAIMED},
			$stats{$process_pid}->{HIGH_NR_ANON_RECLAIMED},
			$stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_SYNC} + $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_SYNC},
			$stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_ASYNC} + $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_ASYNC},
			$this_reclaim_delay / 1000);

		if ($stats{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN}) {
			print "      ";
			for (my $order = 0; $order < 20; $order++) {
				my $count = $stats{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN_PERORDER}[$order];
				if ($count != 0) {
					print "direct-$order=$count ";
				}
			}
		}
		if ($stats{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD}) {
			print "      ";
			for (my $order = 0; $order < 20; $order++) {
				my $count = $stats{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD_PERORDER}[$order];
				if ($count != 0) {
					print "wakeup-$order=$count ";
				}
			}
		}

		print "\n";
	}

	# Print out kswapd activity
	printf("\n");
	printf("%-" . $max_strlen . "s %8s %10s   %8s   %8s %8s %8s\n", "Kswapd",   "Kswapd",  "Order",     "Pages",   "Pages",   "Pages",  "Pages");
	printf("%-" . $max_strlen . "s %8s %10s   %8s   %8s %8s %8s\n", "Instance", "Wakeups", "Re-wakeup", "Scanned", "Rclmed",  "Sync-IO", "ASync-IO");
	foreach $process_pid (keys %stats) {

		if (!$stats{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE}) {
			next;
		}

		$total_kswapd_wake += $stats{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE};
		$total_kswapd_nr_scanned += $stats{$process_pid}->{HIGH_NR_SCANNED};
		$total_kswapd_nr_file_scanned += $stats{$process_pid}->{HIGH_NR_FILE_SCANNED};
		$total_kswapd_nr_anon_scanned += $stats{$process_pid}->{HIGH_NR_ANON_SCANNED};
		$total_kswapd_nr_reclaimed += $stats{$process_pid}->{HIGH_NR_RECLAIMED};
		$total_kswapd_nr_file_reclaimed += $stats{$process_pid}->{HIGH_NR_FILE_RECLAIMED};
		$total_kswapd_nr_anon_reclaimed += $stats{$process_pid}->{HIGH_NR_ANON_RECLAIMED};
		$total_kswapd_writepage_file_sync += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_SYNC};
		$total_kswapd_writepage_anon_sync += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_SYNC};
		$total_kswapd_writepage_file_async += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_ASYNC};
		$total_kswapd_writepage_anon_async += $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_ASYNC};

		printf("%-" . $max_strlen . "s %8d %10d   %8u %8u  %8i %8u",
			$process_pid,
			$stats{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE},
			$stats{$process_pid}->{HIGH_KSWAPD_REWAKEUP},
			$stats{$process_pid}->{HIGH_NR_SCANNED},
			$stats{$process_pid}->{HIGH_NR_FILE_SCANNED},
			$stats{$process_pid}->{HIGH_NR_ANON_SCANNED},
			$stats{$process_pid}->{HIGH_NR_RECLAIMED},
			$stats{$process_pid}->{HIGH_NR_FILE_RECLAIMED},
			$stats{$process_pid}->{HIGH_NR_ANON_RECLAIMED},
			$stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_SYNC} + $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_SYNC},
			$stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_ASYNC} + $stats{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_ASYNC});

		if ($stats{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE}) {
			print "      ";
			for (my $order = 0; $order < 20; $order++) {
				my $count = $stats{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE_PERORDER}[$order];
				if ($count != 0) {
					print "wake-$order=$count ";
				}
			}
		}
		if ($stats{$process_pid}->{HIGH_KSWAPD_REWAKEUP}) {
			print "      ";
			for (my $order = 0; $order < 20; $order++) {
				my $count = $stats{$process_pid}->{HIGH_KSWAPD_REWAKEUP_PERORDER}[$order];
				if ($count != 0) {
					print "rewake-$order=$count ";
				}
			}
		}
		printf("\n");
	}

	# Print out summaries
	$total_direct_latency /= 1000;
	$total_kswapd_latency /= 1000;
	print "\nSummary\n";
	print "Direct reclaims:     			$total_direct_reclaim\n";
	print "Direct reclaim pages scanned:		$total_direct_nr_scanned\n";
	print "Direct reclaim file pages scanned:	$total_direct_nr_file_scanned\n";
	print "Direct reclaim anon pages scanned:	$total_direct_nr_anon_scanned\n";
	print "Direct reclaim pages reclaimed:		$total_direct_nr_reclaimed\n";
	print "Direct reclaim file pages reclaimed:	$total_direct_nr_file_reclaimed\n";
	print "Direct reclaim anon pages reclaimed:	$total_direct_nr_anon_reclaimed\n";
	print "Direct reclaim write file sync I/O:	$total_direct_writepage_file_sync\n";
	print "Direct reclaim write anon sync I/O:	$total_direct_writepage_anon_sync\n";
	print "Direct reclaim write file async I/O:	$total_direct_writepage_file_async\n";
	print "Direct reclaim write anon async I/O:	$total_direct_writepage_anon_async\n";
	print "Wake kswapd requests:			$total_wakeup_kswapd\n";
	printf "Time stalled direct reclaim: 		%-1.2f seconds\n", $total_direct_latency;
	print "\n";
	print "Kswapd wakeups:				$total_kswapd_wake\n";
	print "Kswapd pages scanned:			$total_kswapd_nr_scanned\n";
	print "Kswapd file pages scanned:		$total_kswapd_nr_file_scanned\n";
	print "Kswapd anon pages scanned:		$total_kswapd_nr_anon_scanned\n";
	print "Kswapd pages reclaimed:			$total_kswapd_nr_reclaimed\n";
	print "Kswapd file pages reclaimed:		$total_kswapd_nr_file_reclaimed\n";
	print "Kswapd anon pages reclaimed:		$total_kswapd_nr_anon_reclaimed\n";
	print "Kswapd reclaim write file sync I/O:	$total_kswapd_writepage_file_sync\n";
	print "Kswapd reclaim write anon sync I/O:	$total_kswapd_writepage_anon_sync\n";
	print "Kswapd reclaim write file async I/O:	$total_kswapd_writepage_file_async\n";
	print "Kswapd reclaim write anon async I/O:	$total_kswapd_writepage_anon_async\n";
	printf "Time kswapd awake:			%-1.2f seconds\n", $total_kswapd_latency;
}

sub aggregate_perprocesspid() {
	my $process_pid;
	my $process;
	undef %perprocess;

	foreach $process_pid (keys %perprocesspid) {
		$process = $process_pid;
		$process =~ s/-([0-9])*$//;
		if ($process eq '') {
			$process = "NO_PROCESS_NAME";
		}

		$perprocess{$process}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN} += $perprocesspid{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN};
		$perprocess{$process}->{MM_VMSCAN_KSWAPD_WAKE} += $perprocesspid{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE};
		$perprocess{$process}->{MM_VMSCAN_WAKEUP_KSWAPD} += $perprocesspid{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD};
		$perprocess{$process}->{HIGH_KSWAPD_REWAKEUP} += $perprocesspid{$process_pid}->{HIGH_KSWAPD_REWAKEUP};
		$perprocess{$process}->{HIGH_NR_SCANNED} += $perprocesspid{$process_pid}->{HIGH_NR_SCANNED};
		$perprocess{$process}->{HIGH_NR_FILE_SCANNED} += $perprocesspid{$process_pid}->{HIGH_NR_FILE_SCANNED};
		$perprocess{$process}->{HIGH_NR_ANON_SCANNED} += $perprocesspid{$process_pid}->{HIGH_NR_ANON_SCANNED};
		$perprocess{$process}->{HIGH_NR_RECLAIMED} += $perprocesspid{$process_pid}->{HIGH_NR_RECLAIMED};
		$perprocess{$process}->{HIGH_NR_FILE_RECLAIMED} += $perprocesspid{$process_pid}->{HIGH_NR_FILE_RECLAIMED};
		$perprocess{$process}->{HIGH_NR_ANON_RECLAIMED} += $perprocesspid{$process_pid}->{HIGH_NR_ANON_RECLAIMED};
		$perprocess{$process}->{MM_VMSCAN_WRITEPAGE_FILE_SYNC} += $perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_SYNC};
		$perprocess{$process}->{MM_VMSCAN_WRITEPAGE_ANON_SYNC} += $perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_SYNC};
		$perprocess{$process}->{MM_VMSCAN_WRITEPAGE_FILE_ASYNC} += $perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_FILE_ASYNC};
		$perprocess{$process}->{MM_VMSCAN_WRITEPAGE_ANON_ASYNC} += $perprocesspid{$process_pid}->{MM_VMSCAN_WRITEPAGE_ANON_ASYNC};

		for (my $order = 0; $order < 20; $order++) {
			$perprocess{$process}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN_PERORDER}[$order] += $perprocesspid{$process_pid}->{MM_VMSCAN_DIRECT_RECLAIM_BEGIN_PERORDER}[$order];
			$perprocess{$process}->{MM_VMSCAN_WAKEUP_KSWAPD_PERORDER}[$order] += $perprocesspid{$process_pid}->{MM_VMSCAN_WAKEUP_KSWAPD_PERORDER}[$order];
			$perprocess{$process}->{MM_VMSCAN_KSWAPD_WAKE_PERORDER}[$order] += $perprocesspid{$process_pid}->{MM_VMSCAN_KSWAPD_WAKE_PERORDER}[$order];

		}

		# Aggregate direct reclaim latencies
		my $wr_index = $perprocess{$process}->{MM_VMSCAN_DIRECT_RECLAIM_END};
		my $rd_index = 0;
		while (defined $perprocesspid{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$rd_index]) {
			$perprocess{$process}->{HIGH_DIRECT_RECLAIM_LATENCY}[$wr_index] = $perprocesspid{$process_pid}->{HIGH_DIRECT_RECLAIM_LATENCY}[$rd_index];
			$rd_index++;
			$wr_index++;
		}
		$perprocess{$process}->{MM_VMSCAN_DIRECT_RECLAIM_END} = $wr_index;

		# Aggregate kswapd latencies
		my $wr_index = $perprocess{$process}->{MM_VMSCAN_KSWAPD_SLEEP};
		my $rd_index = 0;
		while (defined $perprocesspid{$process_pid}->{HIGH_KSWAPD_LATENCY}[$rd_index]) {
			$perprocess{$process}->{HIGH_KSWAPD_LATENCY}[$wr_index] = $perprocesspid{$process_pid}->{HIGH_KSWAPD_LATENCY}[$rd_index];
			$rd_index++;
			$wr_index++;
		}
		$perprocess{$process}->{MM_VMSCAN_DIRECT_RECLAIM_END} = $wr_index;
	}
}

sub report() {
	if (!$opt_ignorepid) {
		dump_stats(\%perprocesspid);
	} else {
		aggregate_perprocesspid();
		dump_stats(\%perprocess);
	}
}

# Process events or signals until neither is available
sub signal_loop() {
	my $sigint_processed;
	do {
		$sigint_processed = 0;
		process_events();

		# Handle pending signals if any
		if ($sigint_pending) {
			my $current_time = time;

			if ($sigint_exit) {
				print "Received exit signal\n";
				$sigint_pending = 0;
			}
			if ($sigint_report) {
				if ($current_time >= $sigint_received + 2) {
					report();
					$sigint_report = 0;
					$sigint_pending = 0;
					$sigint_processed = 1;
				}
			}
		}
	} while ($sigint_pending || $sigint_processed);
}

signal_loop();
report();


code:
.. SPDX-License-Identifier: GPL-2.0 OR GFDL-1.2-no-invariants-only

===========================
Lockless Ring Buffer Design
===========================

Copyright 2009 Red Hat Inc.

:Author:   Steven Rostedt <srostedt@redhat.com>
:License:  The GNU Free Documentation License, Version 1.2
           (dual licensed under the GPL v2)
:Reviewers:  Mathieu Desnoyers, Huang Ying, Hidetoshi Seto,
	     and Frederic Weisbecker.

Written for: 2.6.31

Terminology used in this Document
---------------------------------

tail
	- where new writes happen in the ring buffer.

head
	- where new reads happen in the ring buffer.

producer
	- the task that writes into the ring buffer (same as writer)

writer
	- same as producer

consumer
	- the task that reads from the buffer (same as reader)

reader
	- same as consumer.

reader_page
	- A page outside the ring buffer used solely (for the most part)
	  by the reader.

head_page
	- a pointer to the page that the reader will use next

tail_page
	- a pointer to the page that will be written to next

commit_page
	- a pointer to the page with the last finished non-nested write.

cmpxchg
	- hardware-assisted atomic transaction that performs the following::

	    A = B if previous A == C

	    R = cmpxchg(A, C, B) is saying that we replace A with B if and only
		if current A is equal to C, and we put the old (current)
		A into R

	    R gets the previous A regardless if A is updated with B or not.

	  To see if the update was successful a compare of ``R == C``
	  may be used.

The Generic Ring Buffer
-----------------------

The ring buffer can be used in either an overwrite mode or in
producer/consumer mode.

Producer/consumer mode is where if the producer were to fill up the
buffer before the consumer could free up anything, the producer
will stop writing to the buffer. This will lose most recent events.

Overwrite mode is where if the producer were to fill up the buffer
before the consumer could free up anything, the producer will
overwrite the older data. This will lose the oldest events.

No two writers can write at the same time (on the same per-cpu buffer),
but a writer may interrupt another writer, but it must finish writing
before the previous writer may continue. This is very important to the
algorithm. The writers act like a "stack". The way interrupts works
enforces this behavior::

  writer1 start
     <preempted> writer2 start
         <preempted> writer3 start
                     writer3 finishes
                 writer2 finishes
  writer1 finishes

This is very much like a writer being preempted by an interrupt and
the interrupt doing a write as well.

Readers can happen at any time. But no two readers may run at the
same time, nor can a reader preempt/interrupt another reader. A reader
cannot preempt/interrupt a writer, but it may read/consume from the
buffer at the same time as a writer is writing, but the reader must be
on another processor to do so. A reader may read on its own processor
and can be preempted by a writer.

A writer can preempt a reader, but a reader cannot preempt a writer.
But a reader can read the buffer at the same time (on another processor)
as a writer.

The ring buffer is made up of a list of pages held together by a linked list.

At initialization a reader page is allocated for the reader that is not
part of the ring buffer.

The head_page, tail_page and commit_page are all initialized to point
to the same page.

The reader page is initialized to have its next pointer pointing to
the head page, and its previous pointer pointing to a page before
the head page.

The reader has its own page to use. At start up time, this page is
allocated but is not attached to the list. When the reader wants
to read from the buffer, if its page is empty (like it is on start-up),
it will swap its page with the head_page. The old reader page will
become part of the ring buffer and the head_page will be removed.
The page after the inserted page (old reader_page) will become the
new head page.

Once the new page is given to the reader, the reader could do what
it wants with it, as long as a writer has left that page.

A sample of how the reader page is swapped: Note this does not
show the head page in the buffer, it is for demonstrating a swap
only.

::

  +------+
  |reader|          RING BUFFER
  |page  |
  +------+
                  +---+   +---+   +---+
                  |   |-->|   |-->|   |
                  |   |<--|   |<--|   |
                  +---+   +---+   +---+
                   ^ |             ^ |
                   | +-------------+ |
                   +-----------------+

  +------+
  |reader|          RING BUFFER
  |page  |-------------------+
  +------+                   v
    |             +---+   +---+   +---+
    |             |   |-->|   |-->|   |
    |             |   |<--|   |<--|   |<-+
    |             +---+   +---+   +---+  |
    |              ^ |             ^ |   |
    |              | +-------------+ |   |
    |              +-----------------+   |
    +------------------------------------+

  +------+
  |reader|          RING BUFFER
  |page  |-------------------+
  +------+ <---------------+ v
    |  ^          +---+   +---+   +---+
    |  |          |   |-->|   |-->|   |
    |  |          |   |   |   |<--|   |<-+
    |  |          +---+   +---+   +---+  |
    |  |             |             ^ |   |
    |  |             +-------------+ |   |
    |  +-----------------------------+   |
    +------------------------------------+

  +------+
  |buffer|          RING BUFFER
  |page  |-------------------+
  +------+ <---------------+ v
    |  ^          +---+   +---+   +---+
    |  |          |   |   |   |-->|   |
    |  |  New     |   |   |   |<--|   |<-+
    |  | Reader   +---+   +---+   +---+  |
    |  |  page ----^                 |   |
    |  |                             |   |
    |  +-----------------------------+   |
    +------------------------------------+

It is possible that the page swapped is the commit page and the tail page,
if what is in the ring buffer is less than what is held in a buffer page.

::

            reader page    commit page   tail page
                |              |             |
                v              |             |
               +---+           |             |
               |   |<----------+             |
               |   |<------------------------+
               |   |------+
               +---+      |
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

This case is still valid for this algorithm.
When the writer leaves the page, it simply goes into the ring buffer
since the reader page still points to the next location in the ring
buffer.

The main pointers:

  reader page
	    - The page used solely by the reader and is not part
              of the ring buffer (may be swapped in)

  head page
	    - the next page in the ring buffer that will be swapped
              with the reader page.

  tail page
	    - the page where the next write will take place.

  commit page
	    - the page that last finished a write.

The commit page only is updated by the outermost writer in the
writer stack. A writer that preempts another writer will not move the
commit page.

When data is written into the ring buffer, a position is reserved
in the ring buffer and passed back to the writer. When the writer
is finished writing data into that position, it commits the write.

Another write (or a read) may take place at anytime during this
transaction. If another write happens it must finish before continuing
with the previous write.

   Write reserve::

       Buffer page
      +---------+
      |written  |
      +---------+  <--- given back to writer (current commit)
      |reserved |
      +---------+ <--- tail pointer
      | empty   |
      +---------+

   Write commit::

       Buffer page
      +---------+
      |written  |
      +---------+
      |written  |
      +---------+  <--- next position for write (current commit)
      | empty   |
      +---------+

 If a write happens after the first reserve::

       Buffer page
      +---------+
      |written  |
      +---------+  <-- current commit
      |reserved |
      +---------+  <--- given back to second writer
      |reserved |
      +---------+ <--- tail pointer

  After second writer commits::

       Buffer page
      +---------+
      |written  |
      +---------+  <--(last full commit)
      |reserved |
      +---------+
      |pending  |
      |commit   |
      +---------+ <--- tail pointer

  When the first writer commits::

       Buffer page
      +---------+
      |written  |
      +---------+
      |written  |
      +---------+
      |written  |
      +---------+  <--(last full commit and tail pointer)

The commit pointer points to the last write location that was
committed without preempting another write. When a write that
preempted another write is committed, it only becomes a pending commit
and will not be a full commit until all writes have been committed.

The commit page points to the page that has the last full commit.
The tail page points to the page with the last write (before
committing).

The tail page is always equal to or after the commit page. It may
be several pages ahead. If the tail page catches up to the commit
page then no more writes may take place (regardless of the mode
of the ring buffer: overwrite and produce/consumer).

The order of pages is::

 head page
 commit page
 tail page

Possible scenario::

                               tail page
    head page         commit page  |
        |                 |        |
        v                 v        v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

There is a special case that the head page is after either the commit page
and possibly the tail page. That is when the commit (and tail) page has been
swapped with the reader page. This is because the head page is always
part of the ring buffer, but the reader page is not. Whenever there
has been less than a full page that has been committed inside the ring buffer,
and a reader swaps out a page, it will be swapping out the commit page.

::

            reader page    commit page   tail page
                |              |             |
                v              |             |
               +---+           |             |
               |   |<----------+             |
               |   |<------------------------+
               |   |------+
               +---+      |
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+
                          ^
                          |
                      head page

In this case, the head page will not move when the tail and commit
move back into the ring buffer.

The reader cannot swap a page into the ring buffer if the commit page
is still on that page. If the read meets the last commit (real commit
not pending or reserved), then there is nothing more to read.
The buffer is considered empty until another full commit finishes.

When the tail meets the head page, if the buffer is in overwrite mode,
the head page will be pushed ahead one. If the buffer is in producer/consumer
mode, the write will fail.

Overwrite mode::

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+
                          ^
                          |
                      head page

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+
                                   ^
                                   |
                               head page

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+
                                   ^
                                   |
                               head page

Note, the reader page will still point to the previous head page.
But when a swap takes place, it will use the most recent head page.

Making the Ring Buffer Lockless:
--------------------------------

The main idea behind the lockless algorithm is to combine the moving
of the head_page pointer with the swapping of pages with the reader.
State flags are placed inside the pointer to the page. To do this,
each page must be aligned in memory by 4 bytes. This will allow the 2
least significant bits of the address to be used as flags, since
they will always be zero for the address. To get the address,
simply mask out the flags::

  MASK = ~3

  address & MASK

Two flags will be kept by these two bits:

   HEADER
	- the page being pointed to is a head page

   UPDATE
	- the page being pointed to is being updated by a writer
          and was or is about to be a head page.

::

	      reader page
		  |
		  v
		+---+
		|   |------+
		+---+      |
			    |
			    v
	+---+    +---+    +---+    +---+
    <---|   |--->|   |-H->|   |--->|   |--->
    --->|   |<---|   |<---|   |<---|   |<---
	+---+    +---+    +---+    +---+

The above pointer "-H->" would have the HEADER flag set. That is
the next page is the next page to be swapped out by the reader.
This pointer means the next page is the head page.

When the tail page meets the head pointer, it will use cmpxchg to
change the pointer to the UPDATE state::

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-H->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

"-U->" represents a pointer in the UPDATE state.

Any access to the reader will need to take some sort of lock to serialize
the readers. But the writers will never take a lock to write to the
ring buffer. This means we only need to worry about a single reader,
and writes only preempt in "stack" formation.

When the reader tries to swap the page with the ring buffer, it
will also use cmpxchg. If the flag bit in the pointer to the
head page does not have the HEADER flag set, the compare will fail
and the reader will need to look for the new head page and try again.
Note, the flags UPDATE and HEADER are never set at the same time.

The reader swaps the reader page as follows::

  +------+
  |reader|          RING BUFFER
  |page  |
  +------+
                  +---+    +---+    +---+
                  |   |--->|   |--->|   |
                  |   |<---|   |<---|   |
                  +---+    +---+    +---+
                   ^ |               ^ |
                   | +---------------+ |
                   +-----H-------------+

The reader sets the reader page next pointer as HEADER to the page after
the head page::

  +------+
  |reader|          RING BUFFER
  |page  |-------H-----------+
  +------+                   v
    |             +---+    +---+    +---+
    |             |   |--->|   |--->|   |
    |             |   |<---|   |<---|   |<-+
    |             +---+    +---+    +---+  |
    |              ^ |               ^ |   |
    |              | +---------------+ |   |
    |              +-----H-------------+   |
    +--------------------------------------+

It does a cmpxchg with the pointer to the previous head page to make it
point to the reader page. Note that the new pointer does not have the HEADER
flag set.  This action atomically moves the head page forward::

  +------+
  |reader|          RING BUFFER
  |page  |-------H-----------+
  +------+                   v
    |  ^          +---+   +---+   +---+
    |  |          |   |-->|   |-->|   |
    |  |          |   |<--|   |<--|   |<-+
    |  |          +---+   +---+   +---+  |
    |  |             |             ^ |   |
    |  |             +-------------+ |   |
    |  +-----------------------------+   |
    +------------------------------------+

After the new head page is set, the previous pointer of the head page is
updated to the reader page::

  +------+
  |reader|          RING BUFFER
  |page  |-------H-----------+
  +------+ <---------------+ v
    |  ^          +---+   +---+   +---+
    |  |          |   |-->|   |-->|   |
    |  |          |   |   |   |<--|   |<-+
    |  |          +---+   +---+   +---+  |
    |  |             |             ^ |   |
    |  |             +-------------+ |   |
    |  +-----------------------------+   |
    +------------------------------------+

  +------+
  |buffer|          RING BUFFER
  |page  |-------H-----------+  <--- New head page
  +------+ <---------------+ v
    |  ^          +---+   +---+   +---+
    |  |          |   |   |   |-->|   |
    |  |  New     |   |   |   |<--|   |<-+
    |  | Reader   +---+   +---+   +---+  |
    |  |  page ----^                 |   |
    |  |                             |   |
    |  +-----------------------------+   |
    +------------------------------------+

Another important point: The page that the reader page points back to
by its previous pointer (the one that now points to the new head page)
never points back to the reader page. That is because the reader page is
not part of the ring buffer. Traversing the ring buffer via the next pointers
will always stay in the ring buffer. Traversing the ring buffer via the
prev pointers may not.

Note, the way to determine a reader page is simply by examining the previous
pointer of the page. If the next pointer of the previous page does not
point back to the original page, then the original page is a reader page::

             +--------+
             | reader |  next   +----+
             |  page  |-------->|    |<====== (buffer page)
             +--------+         +----+
                 |                | ^
                 |                v | next
            prev |              +----+
                 +------------->|    |
                                +----+

The way the head page moves forward:

When the tail page meets the head page and the buffer is in overwrite mode
and more writes take place, the head page must be moved forward before the
writer may move the tail page. The way this is done is that the writer
performs a cmpxchg to convert the pointer to the head page from the HEADER
flag to have the UPDATE flag set. Once this is done, the reader will
not be able to swap the head page from the buffer, nor will it be able to
move the head page, until the writer is finished with the move.

This eliminates any races that the reader can have on the writer. The reader
must spin, and this is why the reader cannot preempt the writer::

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-H->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The following page will be made into the new head page::

             tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

After the new head page has been set, we can set the old head page
pointer back to NORMAL::

             tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

After the head page has been moved, the tail page may now move forward::

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The above are the trivial updates. Now for the more complex scenarios.

As stated before, if enough writes preempt the first write, the
tail page may make it all the way around the buffer and meet the commit
page. At this time, we must start dropping writes (usually with some kind
of warning to the user). But what happens if the commit was still on the
reader page? The commit page is not part of the ring buffer. The tail page
must account for this::

            reader page    commit page
                |              |
                v              |
               +---+           |
               |   |<----------+
               |   |
               |   |------+
               +---+      |
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-H->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+
                 ^
                 |
             tail page

If the tail page were to simply push the head page forward, the commit when
leaving the reader page would not be pointing to the correct page.

The solution to this is to test if the commit page is on the reader page
before pushing the head page. If it is, then it can be assumed that the
tail page wrapped the buffer, and we must drop new writes.

This is not a race condition, because the commit page can only be moved
by the outermost writer (the writer that was preempted).
This means that the commit will not move while a writer is moving the
tail page. The reader cannot swap the reader page if it is also being
used as the commit page. The reader can simply check that the commit
is off the reader page. Once the commit page leaves the reader page
it will never go back on it unless a reader does another swap with the
buffer page that is also the commit page.

Nested writes
-------------

In the pushing forward of the tail page we must first push forward
the head page if the head page is the next page. If the head page
is not the next page, the tail page is simply updated with a cmpxchg.

Only writers move the tail page. This must be done atomically to protect
against nested writers::

  temp_page = tail_page
  next_page = temp_page->next
  cmpxchg(tail_page, temp_page, next_page)

The above will update the tail page if it is still pointing to the expected
page. If this fails, a nested write pushed it forward, the current write
does not need to push it::

             temp page
                 |
                 v
              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

Nested write comes in and moves the tail page forward::

                      tail page (moved by nested writer)
              temp page   |
                 |        |
                 v        v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The above would fail the cmpxchg, but since the tail page has already
been moved forward, the writer will just try again to reserve storage
on the new tail page.

But the moving of the head page is a bit more complex::

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-H->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The write converts the head page pointer to UPDATE::

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

But if a nested writer preempts here, it will see that the next
page is a head page, but it is also nested. It will detect that
it is nested and will save that information. The detection is the
fact that it sees the UPDATE flag instead of a HEADER or NORMAL
pointer.

The nested writer will set the new head page pointer::

             tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

But it will not reset the update back to normal. Only the writer
that converted a pointer from HEAD to UPDATE will convert it back
to NORMAL::

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

After the nested writer finishes, the outermost writer will convert
the UPDATE pointer to NORMAL::

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

It can be even more complex if several nested writes came in and moved
the tail page ahead several pages::

  (first writer)

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-H->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The write converts the head page pointer to UPDATE::

              tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

Next writer comes in, and sees the update and sets up the new
head page::

  (second writer)

             tail page
                 |
                 v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The nested writer moves the tail page forward. But does not set the old
update page to NORMAL because it is not the outermost writer::

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-H->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

Another writer preempts and sees the page after the tail page is a head page.
It changes it from HEAD to UPDATE::

  (third writer)

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-U->|   |--->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The writer will move the head page forward::

  (third writer)

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-U->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

But now that the third writer did change the HEAD flag to UPDATE it
will convert it to normal::

  (third writer)

                      tail page
                          |
                          v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

Then it will move the tail page, and return back to the second writer::

  (second writer)

                               tail page
                                   |
                                   v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The second writer will fail to move the tail page because it was already
moved, so it will try again and add its data to the new tail page.
It will return to the first writer::

  (first writer)

                               tail page
                                   |
                                   v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

The first writer cannot know atomically if the tail page moved
while it updates the HEAD page. It will then update the head page to
what it thinks is the new head page::

  (first writer)

                               tail page
                                   |
                                   v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-H->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

Since the cmpxchg returns the old value of the pointer the first writer
will see it succeeded in updating the pointer from NORMAL to HEAD.
But as we can see, this is not good enough. It must also check to see
if the tail page is either where it use to be or on the next page::

  (first writer)

                 A        B    tail page
                 |        |        |
                 v        v        v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |-H->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

If tail page != A and tail page != B, then it must reset the pointer
back to NORMAL. The fact that it only needs to worry about nested
writers means that it only needs to check this after setting the HEAD page::

  (first writer)

                 A        B    tail page
                 |        |        |
                 v        v        v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |-U->|   |--->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

Now the writer can update the head page. This is also why the head page must
remain in UPDATE and only reset by the outermost writer. This prevents
the reader from seeing the incorrect head page::

  (first writer)

                 A        B    tail page
                 |        |        |
                 v        v        v
      +---+    +---+    +---+    +---+
  <---|   |--->|   |--->|   |--->|   |-H->
  --->|   |<---|   |<---|   |<---|   |<---
      +---+    +---+    +---+    +---+

repo_name: angery/linux


code:
Deterministic Automata Instrumentation
======================================

The RV monitor file created by dot2k, with the name "$MODEL_NAME.c"
includes a section dedicated to instrumentation.

In the example of the wip.dot monitor created on [1], it will look like::

  /*
   * This is the instrumentation part of the monitor.
   *
   * This is the section where manual work is required. Here the kernel events
   * are translated into model's event.
   *
   */
  static void handle_preempt_disable(void *data, /* XXX: fill header */)
  {
	da_handle_event_wip(preempt_disable_wip);
  }

  static void handle_preempt_enable(void *data, /* XXX: fill header */)
  {
	da_handle_event_wip(preempt_enable_wip);
  }

  static void handle_sched_waking(void *data, /* XXX: fill header */)
  {
	da_handle_event_wip(sched_waking_wip);
  }

  static int enable_wip(void)
  {
	int retval;

	retval = da_monitor_init_wip();
	if (retval)
		return retval;

	rv_attach_trace_probe("wip", /* XXX: tracepoint */, handle_preempt_disable);
	rv_attach_trace_probe("wip", /* XXX: tracepoint */, handle_preempt_enable);
	rv_attach_trace_probe("wip", /* XXX: tracepoint */, handle_sched_waking);

	return 0;
  }

The comment at the top of the section explains the general idea: the
instrumentation section translates *kernel events* into the *model's
event*.

Tracing callback functions
--------------------------

The first three functions are the starting point of the callback *handler
functions* for each of the three events from the wip model. The developer
does not necessarily need to use them: they are just starting points.

Using the example of::

 void handle_preempt_disable(void *data, /* XXX: fill header */)
 {
        da_handle_event_wip(preempt_disable_wip);
 }

The preempt_disable event from the model connects directly to the
preemptirq:preempt_disable. The preemptirq:preempt_disable event
has the following signature, from include/trace/events/preemptirq.h::

  TP_PROTO(unsigned long ip, unsigned long parent_ip)

Hence, the handle_preempt_disable() function will look like::

  void handle_preempt_disable(void *data, unsigned long ip, unsigned long parent_ip)

In this case, the kernel event translates one to one with the automata
event, and indeed, no other change is required for this function.

The next handler function, handle_preempt_enable() has the same argument
list from the handle_preempt_disable(). The difference is that the
preempt_enable event will be used to synchronize the system to the model.

Initially, the *model* is placed in the initial state. However, the *system*
might or might not be in the initial state. The monitor cannot start
processing events until it knows that the system has reached the initial state.
Otherwise, the monitor and the system could be out-of-sync.

Looking at the automata definition, it is possible to see that the system
and the model are expected to return to the initial state after the
preempt_enable execution. Hence, it can be used to synchronize the
system and the model at the initialization of the monitoring section.

The start is informed via a special handle function, the
"da_handle_start_event_$(MONITOR_NAME)(event)", in this case::

  da_handle_start_event_wip(preempt_enable_wip);

So, the callback function will look like::

  void handle_preempt_enable(void *data, unsigned long ip, unsigned long parent_ip)
  {
        da_handle_start_event_wip(preempt_enable_wip);
  }

Finally, the "handle_sched_waking()" will look like::

  void handle_sched_waking(void *data, struct task_struct *task)
  {
        da_handle_event_wip(sched_waking_wip);
  }

And the explanation is left for the reader as an exercise.

enable and disable functions
----------------------------

dot2k automatically creates two special functions::

  enable_$(MONITOR_NAME)()
  disable_$(MONITOR_NAME)()

These functions are called when the monitor is enabled and disabled,
respectively.

They should be used to *attach* and *detach* the instrumentation to the running
system. The developer must add to the relative function all that is needed to
*attach* and *detach* its monitor to the system.

For the wip case, these functions were named::

 enable_wip()
 disable_wip()

But no change was required because: by default, these functions *attach* and
*detach* the tracepoints_to_attach, which was enough for this case.

Instrumentation helpers
-----------------------

To complete the instrumentation, the *handler functions* need to be attached to a
kernel event, at the monitoring enable phase.

The RV interface also facilitates this step. For example, the macro "rv_attach_trace_probe()"
is used to connect the wip model events to the relative kernel event. dot2k automatically
adds "rv_attach_trace_probe()" function call for each model event in the enable phase, as
a suggestion.

For example, from the wip sample model::

  static int enable_wip(void)
  {
        int retval;

        retval = da_monitor_init_wip();
        if (retval)
                return retval;

        rv_attach_trace_probe("wip", /* XXX: tracepoint */, handle_preempt_enable);
        rv_attach_trace_probe("wip", /* XXX: tracepoint */, handle_sched_waking);
        rv_attach_trace_probe("wip", /* XXX: tracepoint */, handle_preempt_disable);

        return 0;
  }

The probes then need to be detached at the disable phase.

[1] The wip model is presented in::

  Documentation/trace/rv/deterministic_automata.rst

The wip monitor is presented in::

  Documentation/trace/rv/da_monitor_synthesis.rst


code:
Deterministic Automata Monitor Synthesis
========================================

The starting point for the application of runtime verification (RV) technics
is the *specification* or *modeling* of the desired (or undesired) behavior
of the system under scrutiny.

The formal representation needs to be then *synthesized* into a *monitor*
that can then be used in the analysis of the trace of the system. The
*monitor* connects to the system via an *instrumentation* that converts
the events from the *system* to the events of the *specification*.

In Linux terms, the runtime verification monitors are encapsulated inside
the *RV monitor* abstraction. The RV monitor includes a set of instances
of the monitor (per-cpu monitor, per-task monitor, and so on), the helper
functions that glue the monitor to the system reference model, and the
trace output as a reaction to event parsing and exceptions, as depicted
below::

 Linux  +----- RV Monitor ----------------------------------+ Formal
  Realm |                                                   |  Realm
  +-------------------+     +----------------+     +-----------------+
  |   Linux kernel    |     |     Monitor    |     |     Reference   |
  |     Tracing       |  -> |   Instance(s)  | <-  |       Model     |
  | (instrumentation) |     | (verification) |     | (specification) |
  +-------------------+     +----------------+     +-----------------+
         |                          |                       |
         |                          V                       |
         |                     +----------+                 |
         |                     | Reaction |                 |
         |                     +--+--+--+-+                 |
         |                        |  |  |                   |
         |                        |  |  +-> trace output ?  |
         +------------------------|--|----------------------+
                                  |  +----> panic ?
                                  +-------> <user-specified>

DA monitor synthesis
--------------------

The synthesis of automata-based models into the Linux *RV monitor* abstraction
is automated by the dot2k tool and the rv/da_monitor.h header file that
contains a set of macros that automatically generate the monitor's code.

dot2k
-----

The dot2k utility leverages dot2c by converting an automaton model in
the DOT format into the C representation [1] and creating the skeleton of
a kernel monitor in C.

For example, it is possible to transform the wip.dot model present in
[1] into a per-cpu monitor with the following command::

  $ dot2k -d wip.dot -t per_cpu

This will create a directory named wip/ with the following files:

- wip.h: the wip model in C
- wip.c: the RV monitor

The wip.c file contains the monitor declaration and the starting point for
the system instrumentation.

Monitor macros
--------------

The rv/da_monitor.h enables automatic code generation for the *Monitor
Instance(s)* using C macros.

The benefits of the usage of macro for monitor synthesis are 3-fold as it:

- Reduces the code duplication;
- Facilitates the bug fix/improvement;
- Avoids the case of developers changing the core of the monitor code
  to manipulate the model in a (let's say) non-standard way.

This initial implementation presents three different types of monitor instances:

- ``#define DECLARE_DA_MON_GLOBAL(name, type)``
- ``#define DECLARE_DA_MON_PER_CPU(name, type)``
- ``#define DECLARE_DA_MON_PER_TASK(name, type)``

The first declares the functions for a global deterministic automata monitor,
the second for monitors with per-cpu instances, and the third with per-task
instances.

In all cases, the 'name' argument is a string that identifies the monitor, and
the 'type' argument is the data type used by dot2k on the representation of
the model in C.

For example, the wip model with two states and three events can be
stored in an 'unsigned char' type. Considering that the preemption control
is a per-cpu behavior, the monitor declaration in the 'wip.c' file is::

  DECLARE_DA_MON_PER_CPU(wip, unsigned char);

The monitor is executed by sending events to be processed via the functions
presented below::

  da_handle_event_$(MONITOR_NAME)($(event from event enum));
  da_handle_start_event_$(MONITOR_NAME)($(event from event enum));
  da_handle_start_run_event_$(MONITOR_NAME)($(event from event enum));

The function ``da_handle_event_$(MONITOR_NAME)()`` is the regular case where
the event will be processed if the monitor is processing events.

When a monitor is enabled, it is placed in the initial state of the automata.
However, the monitor does not know if the system is in the *initial state*.

The ``da_handle_start_event_$(MONITOR_NAME)()`` function is used to notify the
monitor that the system is returning to the initial state, so the monitor can
start monitoring the next event.

The ``da_handle_start_run_event_$(MONITOR_NAME)()`` function is used to notify
the monitor that the system is known to be in the initial state, so the
monitor can start monitoring and monitor the current event.

Using the wip model as example, the events "preempt_disable" and
"sched_waking" should be sent to monitor, respectively, via [2]::

  da_handle_event_wip(preempt_disable_wip);
  da_handle_event_wip(sched_waking_wip);

While the event "preempt_enabled" will use::

  da_handle_start_event_wip(preempt_enable_wip);

To notify the monitor that the system will be returning to the initial state,
so the system and the monitor should be in sync.

Final remarks
-------------

With the monitor synthesis in place using the rv/da_monitor.h and
dot2k, the developer's work should be limited to the instrumentation
of the system, increasing the confidence in the overall approach.

[1] For details about deterministic automata format and the translation
from one representation to another, see::

  Documentation/trace/rv/deterministic_automata.rst

[2] dot2k appends the monitor's name suffix to the events enums to
avoid conflicting variables when exporting the global vmlinux.h
use by BPF programs.


code:
Deterministic Automata
======================

Formally, a deterministic automaton, denoted by G, is defined as a quintuple:

        *G* = { *X*, *E*, *f*, x\ :subscript:`0`, X\ :subscript:`m` }

where:

- *X* is the set of states;
- *E* is the finite set of events;
- x\ :subscript:`0` is the initial state;
- X\ :subscript:`m` (subset of *X*) is the set of marked (or final) states.
- *f* : *X* x *E* -> *X* $ is the transition function. It defines the state
  transition in the occurrence of an event from *E* in the state *X*. In the
  special case of deterministic automata, the occurrence of the event in *E*
  in a state in *X* has a deterministic next state from *X*.

For example, a given automaton named 'wip' (wakeup in preemptive) can
be defined as:

- *X* = { ``preemptive``, ``non_preemptive``}
- *E* = { ``preempt_enable``, ``preempt_disable``, ``sched_waking``}
- x\ :subscript:`0` = ``preemptive``
- X\ :subscript:`m` = {``preemptive``}
- *f* =
   - *f*\ (``preemptive``, ``preempt_disable``) = ``non_preemptive``
   - *f*\ (``non_preemptive``, ``sched_waking``) = ``non_preemptive``
   - *f*\ (``non_preemptive``, ``preempt_enable``) = ``preemptive``

One of the benefits of this formal definition is that it can be presented
in multiple formats. For example, using a *graphical representation*, using
vertices (nodes) and edges, which is very intuitive for *operating system*
practitioners, without any loss.

The previous 'wip' automaton can also be represented as::

                       preempt_enable
          +---------------------------------+
          v                                 |
        #============#  preempt_disable   +------------------+
    --> H preemptive H -----------------> |  non_preemptive  |
        #============#                    +------------------+
                                            ^              |
                                            | sched_waking |
                                            +--------------+

Deterministic Automaton in C
----------------------------

In the paper "Efficient formal verification for the Linux kernel",
the authors present a simple way to represent an automaton in C that can
be used as regular code in the Linux kernel.

For example, the 'wip' automata can be presented as (augmented with comments)::

  /* enum representation of X (set of states) to be used as index */
  enum states {
	preemptive = 0,
	non_preemptive,
	state_max
  };

  #define INVALID_STATE state_max

  /* enum representation of E (set of events) to be used as index */
  enum events {
	preempt_disable = 0,
	preempt_enable,
	sched_waking,
	event_max
  };

  struct automaton {
	char *state_names[state_max];                   // X: the set of states
	char *event_names[event_max];                   // E: the finite set of events
	unsigned char function[state_max][event_max];   // f: transition function
	unsigned char initial_state;                    // x_0: the initial state
	bool final_states[state_max];                   // X_m: the set of marked states
  };

  struct automaton aut = {
	.state_names = {
		"preemptive",
		"non_preemptive"
	},
	.event_names = {
		"preempt_disable",
		"preempt_enable",
		"sched_waking"
	},
	.function = {
		{ non_preemptive,  INVALID_STATE,  INVALID_STATE },
		{  INVALID_STATE,     preemptive, non_preemptive },
	},
	.initial_state = preemptive,
	.final_states = { 1, 0 },
  };

The *transition function* is represented as a matrix of states (lines) and
events (columns), and so the function *f* : *X* x *E* -> *X* can be solved
in O(1). For example::

  next_state = automaton_wip.function[curr_state][event];

Graphviz .dot format
--------------------

The Graphviz open-source tool can produce the graphical representation
of an automaton using the (textual) DOT language as the source code.
The DOT format is widely used and can be converted to many other formats.

For example, this is the 'wip' model in DOT::

  digraph state_automaton {
        {node [shape = circle] "non_preemptive"};
        {node [shape = plaintext, style=invis, label=""] "__init_preemptive"};
        {node [shape = doublecircle] "preemptive"};
        {node [shape = circle] "preemptive"};
        "__init_preemptive" -> "preemptive";
        "non_preemptive" [label = "non_preemptive"];
        "non_preemptive" -> "non_preemptive" [ label = "sched_waking" ];
        "non_preemptive" -> "preemptive" [ label = "preempt_enable" ];
        "preemptive" [label = "preemptive"];
        "preemptive" -> "non_preemptive" [ label = "preempt_disable" ];
        { rank = min ;
                "__init_preemptive";
                "preemptive";
        }
  }

This DOT format can be transformed into a bitmap or vectorial image
using the dot utility, or into an ASCII art using graph-easy. For
instance::

  $ dot -Tsvg -o wip.svg wip.dot
  $ graph-easy wip.dot > wip.txt

dot2c
-----

dot2c is a utility that can parse a .dot file containing an automaton as
in the example above and automatically convert it to the C representation
presented in [3].

For example, having the previous 'wip' model into a file named 'wip.dot',
the following command will transform the .dot file into the C
representation (previously shown) in the 'wip.h' file::

  $ dot2c wip.dot > wip.h

The 'wip.h' content is the code sample in section 'Deterministic Automaton
in C'.

Remarks
-------

The automata formalism allows modeling discrete event systems (DES) in
multiple formats, suitable for different applications/users.

For example, the formal description using set theory is better suitable
for automata operations, while the graphical format for human interpretation;
and computer languages for machine execution.

References
----------

Many textbooks cover automata formalism. For a brief introduction see::

  O'Regan, Gerard. Concise guide to software engineering. Springer,
  Cham, 2017.

For a detailed description, including operations, and application on Discrete
Event Systems (DES), see::

  Cassandras, Christos G., and Stephane Lafortune, eds. Introduction to discrete
  event systems. Boston, MA: Springer US, 2008.

For the C representation in kernel, see::

  De Oliveira, Daniel Bristot; Cucinotta, Tommaso; De Oliveira, Romulo
  Silva. Efficient formal verification for the Linux kernel. In:
  International Conference on Software Engineering and Formal Methods.
  Springer, Cham, 2019. p. 315-332.


code:
====================
Runtime Verification
====================

.. toctree::
   :maxdepth: 2
   :glob:

   runtime-verification.rst
   deterministic_automata.rst
   da_monitor_synthesis.rst
   da_monitor_instrumentation.rst
   monitor_wip.rst
   monitor_wwnr.rst


code:
Monitor wip
===========

- Name: wip - wakeup in preemptive
- Type: per-cpu deterministic automaton
- Author: Daniel Bristot de Oliveira <bristot@kernel.org>

Description
-----------

The wakeup in preemptive (wip) monitor is a sample per-cpu monitor
that verifies if the wakeup events always take place with
preemption disabled::

                     |
                     |
                     v
                   #==================#
                   H    preemptive    H <+
                   #==================#  |
                     |                   |
                     | preempt_disable   | preempt_enable
                     v                   |
    sched_waking   +------------------+  |
  +--------------- |                  |  |
  |                |  non_preemptive  |  |
  +--------------> |                  | -+
                   +------------------+

The wakeup event always takes place with preemption disabled because
of the scheduler synchronization. However, because the preempt_count
and its trace event are not atomic with regard to interrupts, some
inconsistencies might happen. For example::

  preempt_disable() {
	__preempt_count_add(1)
	------->	smp_apic_timer_interrupt() {
				preempt_disable()
					do not trace (preempt count >= 1)

				wake up a thread

				preempt_enable()
					 do not trace (preempt count >= 1)
			}
	<------
	trace_preempt_disable();
  }

This problem was reported and discussed here:
  https://lore.kernel.org/r/cover.1559051152.git.bristot@redhat.com/

Specification
-------------
Grapviz Dot file in tools/verification/models/wip.dot


code:
Monitor wwnr
============

- Name: wwrn - wakeup while not running
- Type: per-task deterministic automaton
- Author: Daniel Bristot de Oliveira <bristot@kernel.org>

Description
-----------

This is a per-task sample monitor, with the following
definition::

               |
               |
               v
    wakeup   +-------------+
  +--------- |             |
  |          | not_running |
  +--------> |             | <+
             +-------------+  |
               |              |
               | switch_in    | switch_out
               v              |
             +-------------+  |
             |   running   | -+
             +-------------+

This model is borken, the reason is that a task can be running
in the processor without being set as RUNNABLE. Think about a
task about to sleep::

  1:      set_current_state(TASK_UNINTERRUPTIBLE);
  2:      schedule();

And then imagine an IRQ happening in between the lines one and two,
waking the task up. BOOM, the wakeup will happen while the task is
running.

- Why do we need this model, so?
- To test the reactors.

Specification
-------------
Grapviz Dot file in tools/verification/models/wwnr.dot


code:
====================
Runtime Verification
====================

Runtime Verification (RV) is a lightweight (yet rigorous) method that
complements classical exhaustive verification techniques (such as *model
checking* and *theorem proving*) with a more practical approach for complex
systems.

Instead of relying on a fine-grained model of a system (e.g., a
re-implementation a instruction level), RV works by analyzing the trace of the
system's actual execution, comparing it against a formal specification of
the system behavior.

The main advantage is that RV can give precise information on the runtime
behavior of the monitored system, without the pitfalls of developing models
that require a re-implementation of the entire system in a modeling language.
Moreover, given an efficient monitoring method, it is possible execute an
*online* verification of a system, enabling the *reaction* for unexpected
events, avoiding, for example, the propagation of a failure on safety-critical
systems.

Runtime Monitors and Reactors
=============================

A monitor is the central part of the runtime verification of a system. The
monitor stands in between the formal specification of the desired (or
undesired) behavior, and the trace of the actual system.

In Linux terms, the runtime verification monitors are encapsulated inside the
*RV monitor* abstraction. A *RV monitor* includes a reference model of the
system, a set of instances of the monitor (per-cpu monitor, per-task monitor,
and so on), and the helper functions that glue the monitor to the system via
trace, as depicted bellow::

 Linux   +---- RV Monitor ----------------------------------+ Formal
  Realm  |                                                  |  Realm
  +-------------------+     +----------------+     +-----------------+
  |   Linux kernel    |     |     Monitor    |     |     Reference   |
  |     Tracing       |  -> |   Instance(s)  | <-  |       Model     |
  | (instrumentation) |     | (verification) |     | (specification) |
  +-------------------+     +----------------+     +-----------------+
         |                          |                       |
         |                          V                       |
         |                     +----------+                 |
         |                     | Reaction |                 |
         |                     +--+--+--+-+                 |
         |                        |  |  |                   |
         |                        |  |  +-> trace output ?  |
         +------------------------|--|----------------------+
                                  |  +----> panic ?
                                  +-------> <user-specified>

In addition to the verification and monitoring of the system, a monitor can
react to an unexpected event. The forms of reaction can vary from logging the
event occurrence to the enforcement of the correct behavior to the extreme
action of taking a system down to avoid the propagation of a failure.

In Linux terms, a *reactor* is an reaction method available for *RV monitors*.
By default, all monitors should provide a trace output of their actions,
which is already a reaction. In addition, other reactions will be available
so the user can enable them as needed.

For further information about the principles of runtime verification and
RV applied to Linux:

  Bartocci, Ezio, et al. *Introduction to runtime verification.* In: Lectures on
  Runtime Verification. Springer, Cham, 2018. p. 1-33.

  Falcone, Ylies, et al. *A taxonomy for classifying runtime verification tools.*
  In: International Conference on Runtime Verification. Springer, Cham, 2018. p.
  241-262.

  De Oliveira, Daniel Bristot. *Automata-based formal analysis and
  verification of the real-time Linux kernel.* Ph.D. Thesis, 2020.

Online RV monitors
==================

Monitors can be classified as *offline* and *online* monitors. *Offline*
monitor process the traces generated by a system after the events, generally by
reading the trace execution from a permanent storage system. *Online* monitors
process the trace during the execution of the system. Online monitors are said
to be *synchronous* if the processing of an event is attached to the system
execution, blocking the system during the event monitoring. On the other hand,
an *asynchronous* monitor has its execution detached from the system. Each type
of monitor has a set of advantages. For example, *offline* monitors can be
executed on different machines but require operations to save the log to a
file. In contrast, *synchronous online* method can react at the exact moment
a violation occurs.

Another important aspect regarding monitors is the overhead associated with the
event analysis. If the system generates events at a frequency higher than the
monitor's ability to process them in the same system, only the *offline*
methods are viable. On the other hand, if the tracing of the events incurs
on higher overhead than the simple handling of an event by a monitor, then a
*synchronous online* monitors will incur on lower overhead.

Indeed, the research presented in:

  De Oliveira, Daniel Bristot; Cucinotta, Tommaso; De Oliveira, Romulo Silva.
  *Efficient formal verification for the Linux kernel.* In: International
  Conference on Software Engineering and Formal Methods. Springer, Cham, 2019.
  p. 315-332.

Shows that for Deterministic Automata models, the synchronous processing of
events in-kernel causes lower overhead than saving the same events to the trace
buffer, not even considering collecting the trace for user-space analysis.
This motivated the development of an in-kernel interface for online monitors.

For further information about modeling of Linux kernel behavior using automata,
see:

  De Oliveira, Daniel B.; De Oliveira, Romulo S.; Cucinotta, Tommaso. *A thread
  synchronization model for the PREEMPT_RT Linux kernel.* Journal of Systems
  Architecture, 2020, 107: 101729.

The user interface
==================

The user interface resembles the tracing interface (on purpose). It is
currently at "/sys/kernel/tracing/rv/".

The following files/folders are currently available:

**available_monitors**

- Reading list the available monitors, one per line

For example::

   # cat available_monitors
   wip
   wwnr

**available_reactors**

- Reading shows the available reactors, one per line.

For example::

   # cat available_reactors
   nop
   panic
   printk

**enabled_monitors**:

- Reading lists the enabled monitors, one per line
- Writing to it enables a given monitor
- Writing a monitor name with a '!' prefix disables it
- Truncating the file disables all enabled monitors

For example::

   # cat enabled_monitors
   # echo wip > enabled_monitors
   # echo wwnr >> enabled_monitors
   # cat enabled_monitors
   wip
   wwnr
   # echo '!wip' >> enabled_monitors
   # cat enabled_monitors
   wwnr
   # echo > enabled_monitors
   # cat enabled_monitors
   #

Note that it is possible to enable more than one monitor concurrently.

**monitoring_on**

This is an on/off general switcher for monitoring. It resembles the
"tracing_on" switcher in the trace interface.

- Writing "0" stops the monitoring
- Writing "1" continues the monitoring
- Reading returns the current status of the monitoring

Note that it does not disable enabled monitors but stop the per-entity
monitors monitoring the events received from the system.

**reacting_on**

- Writing "0" prevents reactions for happening
- Writing "1" enable reactions
- Reading returns the current status of the reaction

**monitors/**

Each monitor will have its own directory inside "monitors/". There the
monitor-specific files will be presented. The "monitors/" directory resembles
the "events" directory on tracefs.

For example::

   # cd monitors/wip/
   # ls
   desc  enable
   # cat desc
   wakeup in preemptive per-cpu testing monitor.
   # cat enable
   0

**monitors/MONITOR/desc**

- Reading shows a description of the monitor *MONITOR*

**monitors/MONITOR/enable**

- Writing "0" disables the *MONITOR*
- Writing "1" enables the *MONITOR*
- Reading return the current status of the *MONITOR*

**monitors/MONITOR/reactors**

- List available reactors, with the select reaction for the given *MONITOR*
  inside "[]". The default one is the nop (no operation) reactor.
- Writing the name of a reactor enables it to the given MONITOR.

For example::

   # cat monitors/wip/reactors
   [nop]
   panic
   printk
   # echo panic > monitors/wip/reactors
   # cat monitors/wip/reactors
   nop
   [panic]
   printk


code:
.. SPDX-License-Identifier: GPL-2.0

===================
System Trace Module
===================

System Trace Module (STM) is a device described in MIPI STP specs as
STP trace stream generator. STP (System Trace Protocol) is a trace
protocol multiplexing data from multiple trace sources, each one of
which is assigned a unique pair of master and channel. While some of
these masters and channels are statically allocated to certain
hardware trace sources, others are available to software. Software
trace sources are usually free to pick for themselves any
master/channel combination from this pool.

On the receiving end of this STP stream (the decoder side), trace
sources can only be identified by master/channel combination, so in
order for the decoder to be able to make sense of the trace that
involves multiple trace sources, it needs to be able to map those
master/channel pairs to the trace sources that it understands.

For instance, it is helpful to know that syslog messages come on
master 7 channel 15, while arbitrary user applications can use masters
48 to 63 and channels 0 to 127.

To solve this mapping problem, stm class provides a policy management
mechanism via configfs, that allows defining rules that map string
identifiers to ranges of masters and channels. If these rules (policy)
are consistent with what decoder expects, it will be able to properly
process the trace data.

This policy is a tree structure containing rules (policy_node) that
have a name (string identifier) and a range of masters and channels
associated with it, located in "stp-policy" subsystem directory in
configfs. The topmost directory's name (the policy) is formatted as
the STM device name to which this policy applies and an arbitrary
string identifier separated by a stop. From the example above, a rule
may look like this::

	$ ls /config/stp-policy/dummy_stm.my-policy/user
	channels masters
	$ cat /config/stp-policy/dummy_stm.my-policy/user/masters
	48 63
	$ cat /config/stp-policy/dummy_stm.my-policy/user/channels
	0 127

which means that the master allocation pool for this rule consists of
masters 48 through 63 and channel allocation pool has channels 0
through 127 in it. Now, any producer (trace source) identifying itself
with "user" identification string will be allocated a master and
channel from within these ranges.

These rules can be nested, for example, one can define a rule "dummy"
under "user" directory from the example above and this new rule will
be used for trace sources with the id string of "user/dummy".

Trace sources have to open the stm class device's node and write their
trace data into its file descriptor.

In order to find an appropriate policy node for a given trace source,
several mechanisms can be used. First, a trace source can explicitly
identify itself by calling an STP_POLICY_ID_SET ioctl on the character
device's file descriptor, providing their id string, before they write
any data there. Secondly, if they chose not to perform the explicit
identification (because you may not want to patch existing software
to do this), they can just start writing the data, at which point the
stm core will try to find a policy node with the name matching the
task's name (e.g., "syslogd") and if one exists, it will be used.
Thirdly, if the task name can't be found among the policy nodes, the
catch-all entry "default" will be used, if it exists. This entry also
needs to be created and configured by the system administrator or
whatever tools are taking care of the policy configuration. Finally,
if all the above steps failed, the write() to an stm file descriptor
will return a error (EINVAL).

Previously, if no policy nodes were found for a trace source, the stm
class would silently fall back to allocating the first available
contiguous range of master/channels from the beginning of the device's
master/channel range. The new requirement for a policy node to exist
will help programmers and sysadmins identify gaps in configuration
and have better control over the un-identified sources.

Some STM devices may allow direct mapping of the channel mmio regions
to userspace for zero-copy writing. One mappable page (in terms of
mmu) will usually contain multiple channels' mmios, so the user will
need to allocate that many channels to themselves (via the
aforementioned ioctl() call) to be able to do this. That is, if your
stm device's channel mmio region is 64 bytes and hardware page size is
4096 bytes, after a successful STP_POLICY_ID_SET ioctl() call with
width==64, you should be able to mmap() one page on this file
descriptor and obtain direct access to an mmio region for 64 channels.

Examples of STM devices are Intel(R) Trace Hub [1] and Coresight STM
[2].

stm_source
==========

For kernel-based trace sources, there is "stm_source" device
class. Devices of this class can be connected and disconnected to/from
stm devices at runtime via a sysfs attribute called "stm_source_link"
by writing the name of the desired stm device there, for example::

	$ echo dummy_stm.0 > /sys/class/stm_source/console/stm_source_link

For examples on how to use stm_source interface in the kernel, refer
to stm_console, stm_heartbeat or stm_ftrace drivers.

Each stm_source device will need to assume a master and a range of
channels, depending on how many channels it requires. These are
allocated for the device according to the policy configuration. If
there's a node in the root of the policy directory that matches the
stm_source device's name (for example, "console"), this node will be
used to allocate master and channel numbers. If there's no such policy
node, the stm core will use the catch-all entry "default", if one
exists. If neither policy nodes exist, the write() to stm_source_link
will return an error.

stm_console
===========

One implementation of this interface also used in the example above is
the "stm_console" driver, which basically provides a one-way console
for kernel messages over an stm device.

To configure the master/channel pair that will be assigned to this
console in the STP stream, create a "console" policy entry (see the
beginning of this text on how to do that). When initialized, it will
consume one channel.

stm_ftrace
==========

This is another "stm_source" device, once the stm_ftrace has been
linked with an stm device, and if "function" tracer is enabled,
function address and parent function address which Ftrace subsystem
would store into ring buffer will be exported via the stm device at
the same time.

Currently only Ftrace "function" tracer is supported.

* [1] https://software.intel.com/sites/default/files/managed/d3/3c/intel-th-developer-manual.pdf
* [2] http://infocenter.arm.com/help/index.jsp?topic=/com.arm.doc.ddi0444b/index.html


code:
.. SPDX-License-Identifier: GPL-2.0

===================
MIPI SyS-T over STP
===================

The MIPI SyS-T protocol driver can be used with STM class devices to
generate standardized trace stream. Aside from being a standard, it
provides better trace source identification and timestamp correlation.

In order to use the MIPI SyS-T protocol driver with your STM device,
first, you'll need CONFIG_STM_PROTO_SYS_T.

Now, you can select which protocol driver you want to use when you create
a policy for your STM device, by specifying it in the policy name:

# mkdir /config/stp-policy/dummy_stm.0:p_sys-t.my-policy/

In other words, the policy name format is extended like this:

  <device_name>:<protocol_name>.<policy_name>

With Intel TH, therefore it can look like "0-sth:p_sys-t.my-policy".

If the protocol name is omitted, the STM class will chose whichever
protocol driver was loaded first.

You can also double check that everything is working as expected by

# cat /config/stp-policy/dummy_stm.0:p_sys-t.my-policy/protocol
p_sys-t

Now, with the MIPI SyS-T protocol driver, each policy node in the
configfs gets a few additional attributes, which determine per-source
parameters specific to the protocol:

# mkdir /config/stp-policy/dummy_stm.0:p_sys-t.my-policy/default
# ls /config/stp-policy/dummy_stm.0:p_sys-t.my-policy/default
channels
clocksync_interval
do_len
masters
ts_interval
uuid

The most important one here is the "uuid", which determines the UUID
that will be used to tag all data coming from this source. It is
automatically generated when a new node is created, but it is likely
that you would want to change it.

do_len switches on/off the additional "payload length" field in the
MIPI SyS-T message header. It is off by default as the STP already
marks message boundaries.

ts_interval and clocksync_interval determine how much time in milliseconds
can pass before we need to include a protocol (not transport, aka STP)
timestamp in a message header or send a CLOCKSYNC packet, respectively.

See Documentation/ABI/testing/configfs-stp-policy-p_sys-t for more
details.

* [1] https://www.mipi.org/specifications/sys-t