]> Sergey Matveev's repositories - public-inbox.git/blob - t/cmd_ipc.t
ipc: wq supports arbitrarily large payloads
[public-inbox.git] / 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(alarm);
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                 my $alrm = 0;
57                 local $SIG{ALRM} = sub { $alrm++ };
58                 alarm(0.001);
59                 @fds = $recv->($s2, $buf, length($src) + 1);
60                 ok($!{EINTR}, "EINTR set by ($desc)");
61                 is_deeply(\@fds, [ undef ], "EINTR $desc");
62                 is($alrm, 1, 'SIGALRM hit');
63
64                 close $s1;
65                 @fds = $recv->($s2, $buf, length($src) + 1);
66                 is_deeply(\@fds, [], "no FDs on EOF $desc");
67                 is($buf, '', "buffer cleared on EOF ($desc)");
68
69                 socketpair($s1, $s2, AF_UNIX, $type, 0) or BAIL_OUT $!;
70                 $s1->blocking(0);
71                 my $nsent = 0;
72                 while (defined(my $n = $send->($s1, $sfds, $src, $flag))) {
73                         $nsent += $n;
74                         fail "sent 0 bytes" if $n == 0;
75                 }
76                 ok($!{EAGAIN}, "hit EAGAIN on send $desc");
77                 ok($nsent > 0, 'sent some bytes');
78
79                 socketpair($s1, $s2, AF_UNIX, $type, 0) or BAIL_OUT $!;
80                 is($send->($s1, [], $src, $flag), length($src), 'sent w/o FDs');
81                 $buf = 'nope';
82                 @fds = $recv->($s2, $buf, length($src));
83                 is(scalar(@fds), 0, 'no FDs received');
84                 is($buf, $src, 'recv w/o FDs');
85
86                 my $nr = 2 * 1024 * 1024;
87                 while (1) {
88                         vec(my $vec = '', $nr * 8 - 1, 1) = 1;
89                         my $n = $send->($s1, [], $vec, $flag);
90                         if (defined($n)) {
91                                 $n == length($vec) or
92                                         fail "short send: $n != ".length($vec);
93                                 diag "sent $nr, retrying with more";
94                                 $nr += 2 * 1024 * 1024;
95                         } else {
96                                 ok($!{EMSGSIZE}, 'got EMSGSIZE');
97                                 # diag "$nr bytes hits EMSGSIZE";
98                                 last;
99                         }
100                 }
101         }
102 } };
103
104 my $send_ic = PublicInbox::Spawn->can('send_cmd4');
105 my $recv_ic = PublicInbox::Spawn->can('recv_cmd4');
106 SKIP: {
107         ($send_ic && $recv_ic) or skip 'Inline::C not installed/enabled', 12;
108         $send = $send_ic;
109         $recv = $recv_ic;
110         $do_test->(SOCK_STREAM, 0, 'Inline::C stream');
111         $do_test->($SOCK_SEQPACKET, MSG_EOR, 'Inline::C seqpacket');
112 }
113
114 SKIP: {
115         require_mods('Socket::MsgHdr', 13);
116         require_ok 'PublicInbox::CmdIPC4';
117         $send = PublicInbox::CmdIPC4->can('send_cmd4');
118         $recv = PublicInbox::CmdIPC4->can('recv_cmd4');
119         $do_test->(SOCK_STREAM, 0, 'MsgHdr stream');
120         $do_test->($SOCK_SEQPACKET, MSG_EOR, 'MsgHdr seqpacket');
121         SKIP: {
122                 ($send_ic && $recv_ic) or
123                         skip 'Inline::C not installed/enabled', 12;
124                 $recv = $recv_ic;
125                 $do_test->(SOCK_STREAM, 0, 'Inline::C -> MsgHdr stream');
126                 $do_test->($SOCK_SEQPACKET, 0, 'Inline::C -> MsgHdr seqpacket');
127         }
128 }
129
130 done_testing;