our $VERSION = 6.57;
our %log;
our $SESLOGLEVEL = exists $ENV{PERL_CORO_DEFAULT_LOGLEVEL} ? $ENV{PERL_CORO_DEFAULT_LOGLEVEL} : -1;
our $ERRLOGLEVEL = exists $ENV{PERL_CORO_STDERR_LOGLEVEL} ? $ENV{PERL_CORO_STDERR_LOGLEVEL} : -1;
sub find_coro {
my ($pid) = @_;
if (my ($coro) = grep $_ == $pid, Coro::State::list) {
$coro
} else {
print "$pid: no such coroutine\n";
undef
}
}
sub format_msg($$) {
my ($time, $micro) = Coro::Util::gettimeofday;
my ($sec, $min, $hour, $day, $mon, $year) = gmtime $time;
my $date = sprintf "%04d-%02d-%02dZ%02d:%02d:%02d.%04d",
$year + 1900, $mon + 1, $day, $hour, $min, $sec, $micro / 100;
sprintf "%s (%d) %s", $date, $_[0], $_[1]
}
sub format_num4($) {
my ($v) = @_;
return sprintf "%4d" , $v if $v < 1e4;
return sprintf "%3.0fk", $v / 1_000 if $v < 1e6;
return sprintf "%1.1fM", $v / 1_000_000 if $v < 1e7 * .995;
return sprintf "%3.0fM", $v / 1_000_000 if $v < 1e9;
return sprintf "%1.1fG", $v / 1_000_000_000 if $v < 1e10 * .995;
return sprintf "%3.0fG", $v / 1_000_000_000 if $v < 1e12;
return sprintf "%1.1fT", $v / 1_000_000_000_000 if $v < 1e13 * .995;
return sprintf "%3.0fT", $v / 1_000_000_000_000 if $v < 1e15;
"++++"
}
sub log($$) {
my ($level, $msg) = @_;
$msg =~ s/\s*$/\n/;
$_->($level, $msg) for values %log;
printf STDERR format_msg $level, $msg if $level <= $ERRLOGLEVEL;
}
sub session_loglevel($) {
$SESLOGLEVEL = shift;
}
sub stderr_loglevel($) {
$ERRLOGLEVEL = shift;
}
sub trace {
my ($coro, $loglevel) = @_;
$coro ||= $Coro::current;
$loglevel = 5 unless defined $loglevel;
(Coro::async {
if (eval { Coro::State::trace $coro, Coro::State::CC_TRACE | Coro::State::CC_TRACE_SUB; 1 }) {
Coro::Debug::log $loglevel, sprintf "[%d] tracing enabled", $coro + 0;
$coro->{_trace_line_cb} = sub {
Coro::Debug::log $loglevel, sprintf "[%d] at %s:%d\n", $Coro::current+0, @_;
};
$coro->{_trace_sub_cb} = sub {
Coro::Debug::log $loglevel, sprintf "[%d] %s %s %s\n",
$Coro::current+0,
$_[0] ? "enter" : "leave",
$_[1],
$_[2] ? ($_[0] ? "with (" : "returning (") . (
join ",",
map {
my $x = ref $_ ? overload::StrVal $_ : $_;
(substr $x, 40) = "..." if 40 + 3 < length $x;
$x =~ s/([^\x20-\x5b\x5d-\x7e])/sprintf "\\x{%02x}", ord $1/ge;
$x
} @{$_[2]}
) . ")" : "";
};
undef $coro;
} else {
Coro::Debug::log $loglevel, sprintf "[%d] unable to enable tracing: %s", $Coro::current + 0, $@;
}
})->prio (Coro::PRIO_MAX);
Coro::cede;
}
sub untrace {
my ($coro) = @_;
$coro ||= $Coro::current;
(Coro::async {
Coro::State::trace $coro, 0;
delete $coro->{_trace_sub_cb};
delete $coro->{_trace_line_cb};
})->prio (Coro::PRIO_MAX);
Coro::cede;
}
sub ps_listing {
my $times = Coro::State::enable_times;
my $flags = $1;
my $verbose = $flags =~ /v/;
my $desc_format = $flags =~ /w/ ? "%-24s" : "%-24.24s";
my $tim0_format = $times ? " %9s %8s " : " ";
my $tim1_format = $times ? " %9.3f %8.3f " : " ";
my $buf = sprintf "%20s %s%s %4s %4s$tim0_format$desc_format %s\n",
"PID", "S", "C", "RSS", "USES",
$times ? ("t_real", "t_cpu") : (),
"Description", "Where";
for my $coro (reverse Coro::State::list) {
my @bt;
Coro::State::call ($coro, sub {
for my $frame (1..10) {
my @frame = caller $frame;
@bt = @frame if $frame[2];
last unless $bt[0] =~ /^Coro/;
}
});
$bt[1] =~ s/^.*[\/\\]// if @bt && !$verbose;
$buf .= sprintf "%20s %s%s %4s %4s$tim1_format$desc_format %s\n",
$coro+0,
$coro->is_new ? "N" : $coro->is_running ? "U" : $coro->is_ready ? "R" : "-",
$coro->is_traced ? "T" : $coro->has_cctx ? "C" : "-",
format_num4 $coro->rss,
format_num4 $coro->usecount,
$times ? $coro->times : (),
$coro->debug_desc,
(@bt ? sprintf "[%s:%d]", $bt[1], $bt[2] : "-");
}
$buf
}
sub command($) {
my ($cmd) = @_;
$cmd =~ s/\s+$//;
if ($cmd =~ /^ps (?:\s* (\S+))? $/x) {
print ps_listing;
} elsif ($cmd =~ /^bt\s+(\d+)$/) {
if (my $coro = find_coro $1) {
my $bt;
Coro::State::call ($coro, sub {
local $Carp::CarpLevel = 2;
$bt = eval { Carp::longmess "coroutine is" } || "$@";
});
if ($bt) {
print $bt;
} else {
print "$1: unable to get backtrace\n";
}
}
} elsif ($cmd =~ /^(?:e|eval)\s+(\d+)\s+(.*)$/) {
if (my $coro = find_coro $1) {
my $cmd = eval "sub { $2 }";
my @res;
Coro::State::call ($coro, sub { @res = eval { &$cmd } });
print $@ ? $@ : (join " ", @res, "\n");
}
} elsif ($cmd =~ /^(?:tr|trace)\s+(\d+)$/) {
if (my $coro = find_coro $1) {
trace $coro;
}
} elsif ($cmd =~ /^(?:ut|untrace)\s+(\d+)$/) {
if (my $coro = find_coro $1) {
untrace $coro;
}
} elsif ($cmd =~ /^cancel\s+(\d+)$/) {
if (my $coro = find_coro $1) {
$coro->cancel;
}
} elsif ($cmd =~ /^ready\s+(\d+)$/) {
if (my $coro = find_coro $1) {
$coro->ready;
}
} elsif ($cmd =~ /^kill\s+(\d+)(?:\s+(.*))?$/) {
my $reason = defined $2 ? $2 : "killed";
if (my $coro = find_coro $1) {
$coro->throw ($reason);
}
} elsif ($cmd =~ /^enable_times(\s+\S.*)?\s*$/) {
my $enable = defined $1 ? 1*eval $1 : !Coro::State::enable_times;
Coro::State::enable_times $enable;
print "per-thread real and process time gathering ", $enable ? "enabled" : "disabled", ".\n";
} elsif ($cmd =~ /^help$/) {
print <<EOF;
ps [w|v] show the list of all coroutines (wide, verbose)
bt <pid> show a full backtrace of coroutine <pid>
eval <pid> <perl> evaluate <perl> expression in context of <pid>
trace <pid> enable tracing for this coroutine
untrace <pid> disable tracing for this coroutine
kill <pid> <reason> throws the given <reason> string in <pid>
cancel <pid> cancels this coroutine
ready <pid> force <pid> into the ready queue
enable_times <enable> enable or disable time profiling in ps
<anything else> evaluate as perl and print results
<anything else> & same as above, but evaluate asynchronously
you can use (find_coro <pid>) in perl expressions
to find the coro with the given pid, e.g.
(find_coro 9768720)->ready
EOF
} elsif ($cmd =~ /^(.*)&$/) {
my $cmd = $1;
my $sub = eval "sub { $cmd }";
my $fh = select;
Coro::async_pool {
$Coro::current->{desc} = $cmd;
my $t = Coro::Util::time;
my @res = eval { &$sub };
$t = Coro::Util::time - $t;
print {$fh}
"\rcommand: $cmd\n",
"execution time: $t\n",
"result: ", $@ ? $@ : (join " ", @res) . "\n",
"> ";
};
} else {
my @res = eval $cmd;
print $@ ? $@ : (join " ", @res) . "\n";
}
local $| = 1;
}
sub session($) {
my ($fh) = @_;
$fh = Coro::Handle::unblock $fh;
my $old_fh = select $fh;
my $guard = guard { select $old_fh };
my $loglevel = $SESLOGLEVEL;
local $log{$Coro::current} = sub {
return unless $_[0] <= $loglevel;
print $fh "\015", (format_msg $_[0], $_[1]), "> ";
};
print "coro debug session. use help for more info\n\n";
while ((print "> "), defined (my $cmd = $fh->readline ("\012"))) {
if ($cmd =~ /^exit\s*$/) {
print "bye.\n";
last;
} elsif ($cmd =~ /^(?:ll|loglevel)\s*(\d+)?\s*/) {
$loglevel = defined $1 ? $1 : -1;
} elsif ($cmd =~ /^(?:w|watch)\s*([0-9.]*)\s+(.*)/) {
my ($time, $cmd) = ($1*1 || 1, $2);
my $cancel;
Coro::async {
$Coro::current->{desc} = "watch $cmd";
select $fh;
until ($cancel) {
command $cmd;
Coro::Timer::sleep $time;
}
};
$fh->readable;
$cancel = 1;
} elsif ($cmd =~ /^help\s*/) {
command $cmd;
print <<EOF;
loglevel <int> enable logging for messages of level <int> and lower
watch <time> <command> repeat the given command until STDIN becomes readable
exit end this session
EOF
} else {
command $cmd;
}
Coro::cede;
}
}
sub new_unix_server {
my ($class, $path) = @_;
unlink $path;
my $unlink_guard = guard { unlink $path };
AnyEvent::Socket::tcp_server "unix/", $path, sub {
my ($fh) = @_;
$unlink_guard;
Coro::async_pool {
$Coro::current->desc ("[Coro::Debug session]");
session $fh;
};
} or Carp::croak "Coro::Debug::new_unix_server($path): $!";
}
sub new_tcp_server {
my ($class, $port) = @_;
AnyEvent::Socket::tcp_server undef, $port, sub {
my ($fh) = @_;
Coro::async_pool {
$Coro::current->desc ("[Coro::Debug session]");
session $fh;
};
} or Carp::croak "Coro::Debug::new_tcp_server($port): $!";
}
sub DESTROY {
my ($self) = @_;
unlink $self->{path} if exists $self->{path};
%$self = ();
}
1;