1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
|
use strict; use warnings;
use Test::More;
use Parallel::Iterator qw( iterate_as_array );
use IO::Handle;
use POSIX qw(:errno_h);
my $buffer_size = get_pipe_buffer_size();
plan 'skip_all' => "Can't calculate buffer size"
unless defined $buffer_size;
plan tests => 1;
# diag "I/O buffer size: $buffer_size\n";
{
# Random data
my $data = join '', map chr rand 256, ( 1 .. $buffer_size * 2 );
# Just in case someone decides to generate data by some other
# means...
die "Not enough data!" unless length $data > $buffer_size;
my @input = (
{
type => 'hash',
value => $data,
},
[ 1, $data, 3 ],
$data,
);
my @want = (
{
type => 'hash',
value => "$data!",
},
[ $data, $data ],
$data . $data,
);
for ( 1 .. 4 ) {
@input = ( @input, @input );
@want = ( @want, @want );
}
my @got = iterate_as_array(
{ workers => 5, nowarn => 1 },
sub {
my ( $id, $job ) = @_;
# Just munge the data in a predictable, detectable way...
if ( ref $job ) {
if ( 'HASH' eq ref $job ) {
$job->{value} .= '!';
return $job;
}
elsif ( 'ARRAY' eq ref $job ) {
return [ $data, $data ];
}
}
else {
return $job . $job;
}
},
\@input
);
is_deeply \@got, \@want, "big data structure";
}
# Find out how much data we can write to a pipe...
sub get_pipe_buffer_size {
my ( $in, $out ) = map IO::Handle->new, 1 .. 2;
unless ( pipe $in, $out ) {
diag "Can't make pipe ($!)\n";
return;
}
unless ( defined $out->blocking( 0 ) ) {
diag "Can't turn off blocking ($!)\n";
return;
}
my $chunk = ' ' x ( 1024 * 4 );
my $wrote = 0;
CHUNK: while ( 1 ) {
my $rc = $out->syswrite( $chunk, length $chunk );
last CHUNK if !defined $rc && $! == EAGAIN;
$wrote += $rc;
last CHUNK if $rc != length $chunk;
}
close $_ for $in, $out;
return $wrote;
}
1;
|
