Fix missing data while reading from virtio_console

During some test, we figured out that data get droped while reading
from virtio_console unix socket. This happens, if a command produce a
huge amount of output. And the data can be dropped in the middle, which
has the effect, that the command is actually stated as "ok".

E.g. `cat large_file_140kb.txt` sporadically have some output gap somewhere
in the middle of the output.

Interesting is, that the logfile, which is created from qemu, contains
all data.

It seems to be a buffer size problem. But I wasn't able to set a proper
buffer size on the unix socket.

So switching to named pipes. With named pipes we are able to set the
buffer size. By default we set the size equal to
/proc/sys/fs/pipe-max-size and it can be overwritten by setting the
VIRTIO_CONSOLE_PIPE_SZ variable.

backend/qemu.pm

@@ -29,7 +29,7 @@ use Time::HiRes qw(sleep gettimeofday);
 use IO::Select;
 use IO::Socket::UNIX 'SOCK_STREAM';
 use IO::Handle;
-use POSIX qw(strftime :sys_wait_h);
+use POSIX qw(strftime :sys_wait_h mkfifo);
 use Mojo::JSON;
 use Carp;
 use Fcntl;
@@ -846,7 +846,9 @@ sub start_qemu {
             sp('device', 'virtio-serial');
             for (my $i = 0; $i < ($vars->{VIRTIO_CONSOLE_NUM} // 1); $i++) {
                 my $name = 'virtio_console' . ($i ? $i : '');
-                sp('chardev', [qv "socket path=$name server nowait id=$name logfile=$name.log logappend=on"]);
+                mkfifo($name . ".in",  0666);
+                mkfifo($name . ".out", 0666);
+                sp('chardev', [qv "pipe id=$name path=$name logfile=$name.log logappend=on"]);
                 sp('device',  [qv "virtconsole chardev=$name name=org.openqa.console.$name"]);
             }
         }

consoles/serial_screen.pm

@@ -26,10 +26,12 @@ use Carp 'croak';
 our $VERSION;
 
 sub new {
-    my ($class, $socket_fd) = @_;
+    my ($class, $fd_read, $fd_write) = @_;
     my $self = bless {class => $class}, $class;
-    $self->{socket_fd}    = $socket_fd;
+    $self->{fd_read}      = $fd_read;
+    $self->{fd_write}     = $fd_write // $fd_read;
     $self->{carry_buffer} = '';
+
     return $self;
 }
 
@@ -89,7 +91,7 @@ and ETX is the same as pressing Ctrl-C on a terminal.
 =cut
 sub type_string {
     my ($self, $nargs) = @_;
-    my $fd = $self->{socket_fd};
+    my $fd = $self->{fd_write};
 
     bmwqemu::log_call(%$nargs);
 
@@ -155,7 +157,7 @@ sub normalise_pattern {
 
   my $num_read = do_read($buffer [, max_size => 2048][,timeout => undef]);
 
-Attempts to read up to max_size bytes from the C<<$self->{socket_fd}>> into buffer.
+Attempts to read up to max_size bytes from the C<<$self->{fd_read}>> into buffer.
 The method returns as soon as some data is available, even if the given size has not been reached.
 Returns number of bytes read or undef on timeout. Note that 0 is a valid return code.
 If an failure occur the method croak.
@@ -168,7 +170,7 @@ sub do_read
     my $buffer = '';
     $args{timeout}  //= undef;    # wait till data is available
     $args{max_size} //= 2048;
-    my $fd = $self->{socket_fd};
+    my $fd = $self->{fd_read};
 
     my $rin = '';
     vec($rin, fileno($fd), 1) = 1;
@@ -223,7 +225,7 @@ and { matched => 0, string => 'text from the terminal' } on failure.
 =cut
 sub read_until {
     my ($self, $pattern, $timeout) = @_[0 .. 2];
-    my $fd       = $self->{socket_fd};
+    my $fd       = $self->{fd_read};
     my %nargs    = @_[3 .. $#_];
     my $buflen   = $nargs{buffer_size} || 4096;
     my $overflow = $nargs{record_output} ? '' : undef;
@@ -311,7 +313,7 @@ sub peak {
 
     bmwqemu::log_call(%nargs);
   LOOP: {
-        $read = sysread($self->{socket_fd}, $buf, $buflen);
+        $read = sysread($self->{fd_read}, $buf, $buflen);
         last LOOP unless defined $read;
 
         $self->{carry_buffer} .= $buf;

consoles/virtio_terminal.pm

@@ -21,14 +21,16 @@ use autodie;
 
 use base 'consoles::console';
 
+use Mojo::File 'path';
 use Socket qw(SOCK_NONBLOCK PF_UNIX SOCK_STREAM sockaddr_un);
 use Errno qw(EAGAIN EWOULDBLOCK);
 use English -no_match_vars;
 use Carp 'croak';
 use Scalar::Util 'blessed';
 use Cwd;
 use consoles::serial_screen ();
-use testapi 'check_var';
+use testapi qw(check_var get_var);
+use Fcntl;
 
 our $VERSION;
 
@@ -48,8 +50,8 @@ where it can start a tty on the virtual console. By default openSUSE and SLE
 automatically start agetty when the kernel finds the virtio console device, but
 another OS may require some additional configuration.
 
-It may also be possible to use a transport other than virtio. This code just
-requires a UNIX socket which inputs and outputs terminal ASCII/ANSI codes.
+It may also be possible to use a transport other than virtio. This code
+uses two pipes to communicate with virtio_consoles from qemu.
 
 =head1 SUBROUTINES/METHODS
 
@@ -58,8 +60,9 @@ requires a UNIX socket which inputs and outputs terminal ASCII/ANSI codes.
 sub new {
     my ($class, $testapi_console, $args) = @_;
     my $self = $class->SUPER::new($testapi_console, $args);
-    $self->{socket_fd}      = 0;
-    $self->{socket_path}    = $self->{args}->{socked_path} // cwd() . '/virtio_console';
+    $self->{fd_read}        = 0;
+    $self->{fd_write}       = 0;
+    $self->{pipe_prefix}    = $self->{args}->{socked_path} // cwd() . '/virtio_console';
     $self->{snapshots}      = {};
     $self->{preload_buffer} = '';
     return $self;
@@ -72,10 +75,12 @@ sub screen {
 
 sub disable {
     my ($self) = @_;
-    if ($self->{socket_fd} > 0) {
-        close $self->{socket_fd};
-        $self->{socket_fd} = 0;
-        $self->{screen}    = undef;
+    if ($self->{fd_read} > 0) {
+        close $self->{fd_read};
+        close $self->{fd_write};
+        $self->{fd_read}  = 0;
+        $self->{fd_write} = 0;
+        $self->{screen}   = undef;
     }
 }
 
@@ -99,47 +104,60 @@ sub load_snapshot {
     }
 }
 
-=head2 socket_path
-
-The file system path bound to a UNIX socket which will be used to transfer
-terminal data between the host and guest.
+=head2 F_GETPIPE_SZ
+This is a helper method for system which do not have F_GETPIPE_SZ in
+there Fcntl bindings. See https://perldoc.perl.org/Fcntl.html
+=cut
+sub F_GETPIPE_SZ
+{
+    return eval 'no warnings "all"; Fcntl::F_GETPIPE_SZ;' || 1032;
+}
 
+=head2 F_SETPIPE_SZ
+This is a helper method for system which do not have F_SETPIPE_SZ in
+there Fcntl bindings. See: https://perldoc.perl.org/Fcntl.html
 =cut
-sub socket_path {
-    my ($self) = @_;
-    return $self->{socket_path};
+sub F_SETPIPE_SZ
+{
+    return eval 'no warnings "all"; Fcntl::F_SETPIPE_SZ;' || 1031;
 }
 
-=head2 open_socket
+=head2 open_pipe
 
-  open_socket();
+  open_pipe();
 
-Opens a unix socket to the character device located at $socket_path.
+Opens a the read and write pipe based on C<$pipe_prefix>.
 
-Returns the file descriptor for the open socket, otherwise it dies.
+Returns the read and write file descriptors for the open sockets,
+otherwise it dies.
 
 =cut
-sub open_socket {
+sub open_pipe {
     my ($self) = @_;
-    my $fd;
-    bmwqemu::log_call(socket_path => $self->socket_path);
-
-    (-S $self->socket_path) || croak 'Could not find ' . $self->socket_path;
-    socket($fd, PF_UNIX, SOCK_STREAM | SOCK_NONBLOCK, 0)
-      || croak 'Could not create Unix socket: ' . $ERRNO;
-    connect($fd, sockaddr_un($self->socket_path))
-      || croak 'Could not connect to virtio-console chardev socket: ' . $ERRNO;
+    bmwqemu::log_call(pipe_prefix => $self->{pipe_prefix});
+
+    sysopen(my $fd_w, $self->{pipe_prefix} . '.in', O_WRONLY)
+      or die "Can't open in pipe for writing $!";
+    sysopen(my $fd_r, $self->{pipe_prefix} . '.out', O_NONBLOCK | O_RDONLY)
+      or die "Can't open out pipe for reading $!";
+
+    my $newsize = get_var('VIRTIO_CONSOLE_PIPE_SZ', path('/proc/sys/fs/pipe-max-size')->slurp());
+    for my $fd (($fd_w, $fd_r)) {
+        my $old = fcntl($fd, F_GETPIPE_SZ(), 0);
+        my $new = fcntl($fd, F_SETPIPE_SZ(), int($newsize));
+        bmwqemu::fctinfo("Set PIPE_SZ from $old to $new");
+    }
 
-    return $fd;
+    return ($fd_r, $fd_w);
 }
 
 sub activate {
     my ($self) = @_;
     if (!check_var('VIRTIO_CONSOLE', 0)) {
-        $self->{socket_fd}              = $self->open_socket unless $self->{socket_fd};
-        $self->{screen}                 = consoles::serial_screen::->new($self->{socket_fd});
+        ($self->{fd_read}, $self->{fd_write}) = $self->open_pipe() unless ($self->{fd_read});
+        $self->{screen} = consoles::serial_screen::->new($self->{fd_read}, $self->{fd_write});
         $self->{screen}->{carry_buffer} = $self->{preload_buffer};
-        $self->{preload_buffer}         = '';
+        $self->{preload_buffer} = '';
     }
     else {
         croak 'VIRTIO_CONSOLE is set 0, so no virtio-serial and virtconsole devices will be available to use with this console.';

t/10-terminal.t

@@ -15,7 +15,8 @@ use 5.018;
 use warnings;
 use Carp 'confess';
 use English -no_match_vars;
-use POSIX qw( :sys_wait_h sigprocmask sigsuspend );
+use POSIX qw( :sys_wait_h sigprocmask sigsuspend mkfifo);
+use Fcntl;
 use Socket qw( PF_UNIX SOCK_STREAM sockaddr_un );
 use Time::HiRes 'usleep';
 use File::Temp 'tempfile';
@@ -114,7 +115,7 @@ sub try_write_sequence {
 # Once we have read the data, echo it back like a real terminal, unless the
 # message is $next_test which we just use for synchronisation.
 sub try_read {
-    my ($fd, $expected) = @_;
+    my ($fd, $fd_w, $expected) = @_;
     my ($buf, $text);
 
     while (1) {
@@ -139,7 +140,7 @@ sub try_read {
     $text .= $buf;
 
     if ($expected ne $next_test) {
-        try_write($fd, $text);
+        try_write($fd_w, $text);
     }
     elsif ($text ne $next_test) {
         confess 'fake_terminal: Expecting special $next_test message, but got: ' . $text;
@@ -150,7 +151,7 @@ sub try_read {
 
 # A mock terminal which we can communicate with over a UNIX socket
 sub fake_terminal {
-    my ($sock_path) = @_;
+    my ($pipe_in, $pipe_out) = @_;
     my ($fd, $listen_fd);
 
     $SIG{ALRM} = sub {
@@ -161,25 +162,10 @@ sub fake_terminal {
 
     alarm $timeout;
 
-    socket($listen_fd, PF_UNIX, SOCK_STREAM, 0)
-      || confess "fake_terminal: Could not create socket: $ERRNO";
-    unlink($sock_path);
-    bind($listen_fd, sockaddr_un($sock_path))
-      || confess "fake_terminal: Could not bind socket to path $sock_path: $ERRNO";
-    listen($listen_fd, 1)
-      || confess "fake_terminal: Could not list on socket: $ERRNO";
-
-    #Signal to parent that the socket is listening
-    kill 'CONT', getppid;
-
-  ACCEPT: {
-        accept($fd, $listen_fd) || do {
-            if ($ERRNO{EINTR}) {
-                next ACCEPT;
-            }
-            confess "fake_terminal: Failed to accept connection: $ERRNO";
-        };
-    }
+    open(my $fd_r, "<", $pipe_in)
+      or die "Can't open in pipe for writing $!";
+    open(my $fd_w, ">", $pipe_out)
+      or die "Can't open out pipe for reading $!";
 
     $SIG{ALRM} = sub {
         fail('fake_terminal timed out while performing IO');
@@ -193,41 +179,41 @@ sub fake_terminal {
     my $tb = Test::More->builder;
     $tb->reset;
 
-    try_write($fd, $login_prompt_data);
-    ok(try_read($fd, $user_name_data), 'fake_terminal reads: Entered user name');
+    try_write($fd_w, $login_prompt_data);
+    ok(try_read($fd_r, $fd_w, $user_name_data), 'fake_terminal reads: Entered user name');
 
-    try_write($fd, $password_prompt_data);
-    ok(try_read($fd, $password_data), 'fake_terminal reads: Entered password');
+    try_write($fd_w, $password_prompt_data);
+    ok(try_read($fd_r, $fd_w, $password_data), 'fake_terminal reads: Entered password');
 
-    try_write($fd, $first_prompt_data);
-    ok(try_read($fd, $set_prompt_data), 'fake_terminal reads: Normalised bash prompt');
+    try_write($fd_w, $first_prompt_data);
+    ok(try_read($fd_r, $fd_w, $set_prompt_data), 'fake_terminal reads: Normalised bash prompt');
 
-    try_write($fd, $normalised_prompt_data);
+    try_write($fd_w, $normalised_prompt_data);
 
-    ok(try_read($fd, $C0_EOT), 'fake_terminal reads: C0 EOT control code');
-    ok(try_read($fd, $C0_ETX), 'fake_terminal reads: C0 ETX control code');
-    ok(try_read($fd, "\n"),    'fake_terminal reads: ret');
-    try_write($fd, $login_prompt_data);
+    ok(try_read($fd_r, $fd_w, $C0_EOT), 'fake_terminal reads: C0 EOT control code');
+    ok(try_read($fd_r, $fd_w, $C0_ETX), 'fake_terminal reads: C0 ETX control code');
+    ok(try_read($fd_r, $fd_w, "\n"),    'fake_terminal reads: ret');
+    try_write($fd_w, $login_prompt_data);
 
     alarm $timeout;
 
     # This for loop corresponds to the 'large amount of data tests'
     for (1 .. 2) {
-        try_read($fd, $next_test);
-        try_write_sequence($fd, $US_keyboard_data, $repeat_sequence_count, $stop_code_data);
+        try_read($fd_r, $fd_w, $next_test);
+        try_write_sequence($fd_w, $US_keyboard_data, $repeat_sequence_count, $stop_code_data);
     }
 
     # Trailing data/carry buffer tests
     for (1 .. 2) {
-        try_read($fd, $next_test);
-        try_write($fd, $US_keyboard_data . $stop_code_data . $stop_code_data);
+        try_read($fd_r, $fd_w, $next_test);
+        try_write($fd_w, $US_keyboard_data . $stop_code_data . $stop_code_data);
     }
 
     #alarm $timeout * 2;
     #try_write($fd, ($US_keyboard_data x 100_000) . $stop_code_data);
 
-    try_write($fd, $first_prompt_data);
-    try_read($fd, $next_test);
+    try_write($fd_w, $first_prompt_data);
+    try_read($fd_r, $fd_w, $next_test);
 
     alarm $timeout;
     $SIG{ALRM} = sub {
@@ -236,10 +222,10 @@ sub fake_terminal {
         exit(0);
     };
 
-    try_read($fd, $next_test);
-    try_write($fd, $US_keyboard_data);
+    try_read($fd_r, $fd_w, $next_test);
+    try_write($fd_w, $US_keyboard_data);
     # Keep the socket open while we test the timeout
-    try_read($fd, $next_test);
+    try_read($fd_r, $fd_w, $next_test);
 
     alarm 0;
     pass('fake_terminal managed to get all the way to the end without timing out!');
@@ -366,16 +352,19 @@ sub check_child {
 # so wait for fake terminal to create socket and emit SIGCONT. Sigsuspend only
 # returns if a signal is received which has a handler set. We must initially
 # block the signal incase SIGCONT is emitted before we reach sigsuspend.
-$SIG{CONT} = sub { };
-my $blockmask = POSIX::SigSet->new(POSIX::SIGCONT());
-my $oldmask   = POSIX::SigSet->new();
-sigprocmask(POSIX::SIG_BLOCK, $blockmask, $oldmask);
+my $pipe_in  = $socket_path . ".in";
+my $pipe_out = $socket_path . ".out";
+
+for (($pipe_in, $pipe_out)) {
+    unlink($_) if (-p $_);
+    mkfifo($_, 0666) or die("Cannot create fifo pipe $_");
+}
 
 my $fpid = fork || do {
-    fake_terminal($socket_path);
+    fake_terminal($pipe_in, $pipe_out);
     exit 0;
 };
-sigsuspend($oldmask);
+
 my $tpid = fork || do {
     test_terminal_directly;
     exit 0;
@@ -394,5 +383,6 @@ waitpid($tpid2, 0);
 check_child('Direct test with VIRTIO_CONSOLE=0', 255);
 
 done_testing;
-unlink $socket_path;
+unlink $socket_path . ".in";
+unlink $socket_path . ".out";
 say "The IO log file is at $log_path and the error log is $err_path.";