t/cmd_ipc.t

   1 #!perl -w
   2 # Copyright (C) 2021 all contributors <meta@public-inbox.org>
   3 # License: AGPL-3.0+ <https://www.gnu.org/licenses/agpl-3.0.txt>
   4 use strict;
   5 use v5.10.1;
   6 use Test::More;
   7 use PublicInbox::TestCommon;
   8 use Socket qw(AF_UNIX SOCK_STREAM MSG_EOR);
   9 pipe(my ($r, $w)) or BAIL_OUT;
  10 my ($send, $recv);
  11 require_ok 'PublicInbox::Spawn';
  12 my $SOCK_SEQPACKET = eval { Socket::SOCK_SEQPACKET() } // undef;
  13 use Time::HiRes qw(usleep);
  14
  15 my $do_test = sub { SKIP: {
  16         my ($type, $flag, $desc) = @_;
  17         defined $type or skip 'SOCK_SEQPACKET missing', 7;
  18         my ($s1, $s2);
  19         my $src = 'some payload' x 40;
  20         socketpair($s1, $s2, AF_UNIX, $type, 0) or BAIL_OUT $!;
  21         my $sfds = [ fileno($r), fileno($w), fileno($s1) ];
  22         $send->($s1, $sfds, $src, $flag);
  23         my (@fds) = $recv->($s2, my $buf, length($src) + 1);
  24         is($buf, $src, 'got buffer payload '.$desc);
  25         my ($r1, $w1, $s1a);
  26         my $opens = sub {
  27                 ok(open($r1, '<&=', $fds[0]), 'opened received $r');
  28                 ok(open($w1, '>&=', $fds[1]), 'opened received $w');
  29                 ok(open($s1a, '+>&=', $fds[2]), 'opened received $s1');
  30         };
  31         $opens->();
  32         my @exp = stat $r;
  33         my @cur = stat $r1;
  34         is("$exp[0]\0$exp[1]", "$cur[0]\0$cur[1]", '$r dev/ino matches');
  35         @exp = stat $w;
  36         @cur = stat $w1;
  37         is("$exp[0]\0$exp[1]", "$cur[0]\0$cur[1]", '$w dev/ino matches');
  38         @exp = stat $s1;
  39         @cur = stat $s1a;
  40         is("$exp[0]\0$exp[1]", "$cur[0]\0$cur[1]", '$s1 dev/ino matches');
  41         if (defined($SOCK_SEQPACKET) && $type == $SOCK_SEQPACKET) {
  42                 $r1 = $w1 = $s1a = undef;
  43                 $src = (',' x 1023) . '-' .('.' x 1024);
  44                 $send->($s1, $sfds, $src, $flag);
  45                 (@fds) = $recv->($s2, $buf, 1024);
  46                 is($buf, (',' x 1023) . '-', 'silently truncated buf');
  47                 $opens->();
  48                 $r1 = $w1 = $s1a = undef;
  49
  50                 $s2->blocking(0);
  51                 @fds = $recv->($s2, $buf, length($src) + 1);
  52                 ok($!{EAGAIN}, "EAGAIN set by ($desc)");
  53                 is_deeply(\@fds, [ undef ], "EAGAIN $desc");
  54                 $s2->blocking(1);
  55
  56                 if ($ENV{TEST_ALRM}) {
  57                         my $alrm = 0;
  58                         local $SIG{ALRM} = sub { $alrm++ };
  59                         my $tgt = $$;
  60                         my $pid = fork // xbail "fork: $!";
  61                         if ($pid == 0) {
  62                                 # need to loop since Perl signals are racy
  63                                 # (the interpreter doesn't self-pipe)
  64                                 while (usleep(1000)) {
  65                                         kill 'ALRM', $tgt;
  66                                 }
  67                         }
  68                         @fds = $recv->($s2, $buf, length($src) + 1);
  69                         ok($!{EINTR}, "EINTR set by ($desc)");
  70                         kill('KILL', $pid);
  71                         waitpid($pid, 0);
  72                         is_deeply(\@fds, [ undef ], "EINTR $desc");
  73                         ok($alrm, 'SIGALRM hit');
  74                 }
  75
  76                 close $s1;
  77                 @fds = $recv->($s2, $buf, length($src) + 1);
  78                 is_deeply(\@fds, [], "no FDs on EOF $desc");
  79                 is($buf, '', "buffer cleared on EOF ($desc)");
  80
  81                 socketpair($s1, $s2, AF_UNIX, $type, 0) or BAIL_OUT $!;
  82                 $s1->blocking(0);
  83                 my $nsent = 0;
  84                 while (defined(my $n = $send->($s1, $sfds, $src, $flag))) {
  85                         $nsent += $n;
  86                         fail "sent 0 bytes" if $n == 0;
  87                 }
  88                 ok($!{EAGAIN} || $!{ETOOMANYREFS},
  89                         "hit EAGAIN || ETOOMANYREFS on send $desc") or
  90                         diag "send failed with: $!";
  91                 ok($nsent > 0, 'sent some bytes');
  92
  93                 socketpair($s1, $s2, AF_UNIX, $type, 0) or BAIL_OUT $!;
  94                 is($send->($s1, [], $src, $flag), length($src), 'sent w/o FDs');
  95                 $buf = 'nope';
  96                 @fds = $recv->($s2, $buf, length($src));
  97                 is(scalar(@fds), 0, 'no FDs received');
  98                 is($buf, $src, 'recv w/o FDs');
  99
 100                 my $nr = 2 * 1024 * 1024;
 101                 while (1) {
 102                         vec(my $vec = '', $nr * 8 - 1, 1) = 1;
 103                         my $n = $send->($s1, [], $vec, $flag);
 104                         if (defined($n)) {
 105                                 $n == length($vec) or
 106                                         fail "short send: $n != ".length($vec);
 107                                 diag "sent $nr, retrying with more";
 108                                 $nr += 2 * 1024 * 1024;
 109                         } else {
 110                                 ok($!{EMSGSIZE} || $!{ENOBUFS},
 111                                         'got EMSGSIZE or ENOBUFS') or
 112                                         diag "$nr bytes fails with: $!";
 113                                 last;
 114                         }
 115                 }
 116         }
 117 } };
 118
 119 my $send_ic = PublicInbox::Spawn->can('send_cmd4');
 120 my $recv_ic = PublicInbox::Spawn->can('recv_cmd4');
 121 SKIP: {
 122         ($send_ic && $recv_ic) or skip 'Inline::C not installed/enabled', 12;
 123         $send = $send_ic;
 124         $recv = $recv_ic;
 125         $do_test->(SOCK_STREAM, 0, 'Inline::C stream');
 126         $do_test->($SOCK_SEQPACKET, MSG_EOR, 'Inline::C seqpacket');
 127 }
 128
 129 SKIP: {
 130         require_mods('Socket::MsgHdr', 13);
 131         require_ok 'PublicInbox::CmdIPC4';
 132         $send = PublicInbox::CmdIPC4->can('send_cmd4');
 133         $recv = PublicInbox::CmdIPC4->can('recv_cmd4');
 134         $do_test->(SOCK_STREAM, 0, 'MsgHdr stream');
 135         $do_test->($SOCK_SEQPACKET, MSG_EOR, 'MsgHdr seqpacket');
 136         SKIP: {
 137                 ($send_ic && $recv_ic) or
 138                         skip 'Inline::C not installed/enabled', 12;
 139                 $recv = $recv_ic;
 140                 $do_test->(SOCK_STREAM, 0, 'Inline::C -> MsgHdr stream');
 141                 $do_test->($SOCK_SEQPACKET, 0, 'Inline::C -> MsgHdr seqpacket');
 142         }
 143 }
 144
 145 done_testing;