Codebase list libmce-perl / ced942c
New upstream version 1.838 gregor herrmann 5 years ago
33 changed file(s) with 351 addition(s) and 379 deletion(s). Raw diff Collapse all Expand all
+7
-0
Changes less more
00
11 Revision history for Perl module MCE.
2
3 1.838 Wed Jan 23 08:30:00 EST 2019
4
5 * IPC update, raising reliability across multiple platforms.
6 * Improved hack for the Windows platform for nested MCE sessions.
7 * Added _sysread, _sysseek, _syswrite, and _nonblocking to MCE::Util.
8 * Added barrier option to MCE::Queue: allows one to disable.
29
310 1.837 Sat Aug 25 13:00:00 EST 2018
411
+26
-25
META.json less more
4242 },
4343 "requires" : {
4444 "Carp" : "0",
45 "Errno" : "0",
4546 "Fcntl" : "0",
4647 "File::Path" : "0",
4748 "Getopt::Long" : "0",
6364 "provides" : {
6465 "MCE" : {
6566 "file" : "lib/MCE.pm",
66 "version" : "1.837"
67 "version" : "1.838"
6768 },
6869 "MCE::Candy" : {
6970 "file" : "lib/MCE/Candy.pm",
70 "version" : "1.837"
71 "version" : "1.838"
7172 },
7273 "MCE::Core::Input::Generator" : {
7374 "file" : "lib/MCE/Core/Input/Generator.pm",
74 "version" : "1.837"
75 "version" : "1.838"
7576 },
7677 "MCE::Core::Input::Handle" : {
7778 "file" : "lib/MCE/Core/Input/Handle.pm",
78 "version" : "1.837"
79 "version" : "1.838"
7980 },
8081 "MCE::Core::Input::Iterator" : {
8182 "file" : "lib/MCE/Core/Input/Iterator.pm",
82 "version" : "1.837"
83 "version" : "1.838"
8384 },
8485 "MCE::Core::Input::Request" : {
8586 "file" : "lib/MCE/Core/Input/Request.pm",
86 "version" : "1.837"
87 "version" : "1.838"
8788 },
8889 "MCE::Core::Input::Sequence" : {
8990 "file" : "lib/MCE/Core/Input/Sequence.pm",
90 "version" : "1.837"
91 "version" : "1.838"
9192 },
9293 "MCE::Core::Manager" : {
9394 "file" : "lib/MCE/Core/Manager.pm",
94 "version" : "1.837"
95 "version" : "1.838"
9596 },
9697 "MCE::Core::Validation" : {
9798 "file" : "lib/MCE/Core/Validation.pm",
98 "version" : "1.837"
99 "version" : "1.838"
99100 },
100101 "MCE::Core::Worker" : {
101102 "file" : "lib/MCE/Core/Worker.pm",
102 "version" : "1.837"
103 "version" : "1.838"
103104 },
104105 "MCE::Flow" : {
105106 "file" : "lib/MCE/Flow.pm",
106 "version" : "1.837"
107 "version" : "1.838"
107108 },
108109 "MCE::Grep" : {
109110 "file" : "lib/MCE/Grep.pm",
110 "version" : "1.837"
111 "version" : "1.838"
111112 },
112113 "MCE::Loop" : {
113114 "file" : "lib/MCE/Loop.pm",
114 "version" : "1.837"
115 "version" : "1.838"
115116 },
116117 "MCE::Map" : {
117118 "file" : "lib/MCE/Map.pm",
118 "version" : "1.837"
119 "version" : "1.838"
119120 },
120121 "MCE::Mutex" : {
121122 "file" : "lib/MCE/Mutex.pm",
122 "version" : "1.837"
123 "version" : "1.838"
123124 },
124125 "MCE::Mutex::Channel" : {
125126 "file" : "lib/MCE/Mutex/Channel.pm",
126 "version" : "1.837"
127 "version" : "1.838"
127128 },
128129 "MCE::Mutex::Flock" : {
129130 "file" : "lib/MCE/Mutex/Flock.pm",
130 "version" : "1.837"
131 "version" : "1.838"
131132 },
132133 "MCE::Queue" : {
133134 "file" : "lib/MCE/Queue.pm",
134 "version" : "1.837"
135 "version" : "1.838"
135136 },
136137 "MCE::Relay" : {
137138 "file" : "lib/MCE/Relay.pm",
138 "version" : "1.837"
139 "version" : "1.838"
139140 },
140141 "MCE::Signal" : {
141142 "file" : "lib/MCE/Signal.pm",
142 "version" : "1.837"
143 "version" : "1.838"
143144 },
144145 "MCE::Step" : {
145146 "file" : "lib/MCE/Step.pm",
146 "version" : "1.837"
147 "version" : "1.838"
147148 },
148149 "MCE::Stream" : {
149150 "file" : "lib/MCE/Stream.pm",
150 "version" : "1.837"
151 "version" : "1.838"
151152 },
152153 "MCE::Subs" : {
153154 "file" : "lib/MCE/Subs.pm",
154 "version" : "1.837"
155 "version" : "1.838"
155156 },
156157 "MCE::Util" : {
157158 "file" : "lib/MCE/Util.pm",
158 "version" : "1.837"
159 "version" : "1.838"
159160 }
160161 },
161162 "release_status" : "stable",
171172 "url" : "https://github.com/marioroy/mce-perl.git"
172173 }
173174 },
174 "version" : "1.837"
175 "version" : "1.838"
175176 }
+26
-25
META.yml less more
2424 provides:
2525 MCE:
2626 file: lib/MCE.pm
27 version: '1.837'
27 version: '1.838'
2828 MCE::Candy:
2929 file: lib/MCE/Candy.pm
30 version: '1.837'
30 version: '1.838'
3131 MCE::Core::Input::Generator:
3232 file: lib/MCE/Core/Input/Generator.pm
33 version: '1.837'
33 version: '1.838'
3434 MCE::Core::Input::Handle:
3535 file: lib/MCE/Core/Input/Handle.pm
36 version: '1.837'
36 version: '1.838'
3737 MCE::Core::Input::Iterator:
3838 file: lib/MCE/Core/Input/Iterator.pm
39 version: '1.837'
39 version: '1.838'
4040 MCE::Core::Input::Request:
4141 file: lib/MCE/Core/Input/Request.pm
42 version: '1.837'
42 version: '1.838'
4343 MCE::Core::Input::Sequence:
4444 file: lib/MCE/Core/Input/Sequence.pm
45 version: '1.837'
45 version: '1.838'
4646 MCE::Core::Manager:
4747 file: lib/MCE/Core/Manager.pm
48 version: '1.837'
48 version: '1.838'
4949 MCE::Core::Validation:
5050 file: lib/MCE/Core/Validation.pm
51 version: '1.837'
51 version: '1.838'
5252 MCE::Core::Worker:
5353 file: lib/MCE/Core/Worker.pm
54 version: '1.837'
54 version: '1.838'
5555 MCE::Flow:
5656 file: lib/MCE/Flow.pm
57 version: '1.837'
57 version: '1.838'
5858 MCE::Grep:
5959 file: lib/MCE/Grep.pm
60 version: '1.837'
60 version: '1.838'
6161 MCE::Loop:
6262 file: lib/MCE/Loop.pm
63 version: '1.837'
63 version: '1.838'
6464 MCE::Map:
6565 file: lib/MCE/Map.pm
66 version: '1.837'
66 version: '1.838'
6767 MCE::Mutex:
6868 file: lib/MCE/Mutex.pm
69 version: '1.837'
69 version: '1.838'
7070 MCE::Mutex::Channel:
7171 file: lib/MCE/Mutex/Channel.pm
72 version: '1.837'
72 version: '1.838'
7373 MCE::Mutex::Flock:
7474 file: lib/MCE/Mutex/Flock.pm
75 version: '1.837'
75 version: '1.838'
7676 MCE::Queue:
7777 file: lib/MCE/Queue.pm
78 version: '1.837'
78 version: '1.838'
7979 MCE::Relay:
8080 file: lib/MCE/Relay.pm
81 version: '1.837'
81 version: '1.838'
8282 MCE::Signal:
8383 file: lib/MCE/Signal.pm
84 version: '1.837'
84 version: '1.838'
8585 MCE::Step:
8686 file: lib/MCE/Step.pm
87 version: '1.837'
87 version: '1.838'
8888 MCE::Stream:
8989 file: lib/MCE/Stream.pm
90 version: '1.837'
90 version: '1.838'
9191 MCE::Subs:
9292 file: lib/MCE/Subs.pm
93 version: '1.837'
93 version: '1.838'
9494 MCE::Util:
9595 file: lib/MCE/Util.pm
96 version: '1.837'
96 version: '1.838'
9797 recommends:
9898 Sereal::Decoder: '3.015'
9999 Sereal::Encoder: '3.015'
100100 requires:
101101 Carp: '0'
102 Errno: '0'
102103 Fcntl: '0'
103104 File::Path: '0'
104105 Getopt::Long: '0'
119120 homepage: https://github.com/marioroy/mce-perl
120121 license: http://dev.perl.org/licenses/
121122 repository: https://github.com/marioroy/mce-perl.git
122 version: '1.837'
123 version: '1.838'
+26
-25
Makefile.PL less more
1616 ABSTRACT => 'Many-Core Engine for Perl providing parallel processing capabilities',
1717 AUTHOR => 'Mario E. Roy <marioeroy AT gmail DOT com>',
1818 NAME => 'MCE',
19 VERSION => '1.837',
19 VERSION => '1.838',
2020
2121 EXE_FILES => [ @exe_files ],
2222
2929 'strict' => 0,
3030 'warnings' => 0,
3131 'Carp' => 0,
32 'Errno' => 0,
3233 'Fcntl' => 0,
3334 'File::Path' => 0,
3435 'Getopt::Long' => 0,
6667 'provides' => {
6768 'MCE' => {
6869 'file' => 'lib/MCE.pm',
69 'version' => '1.837'
70 'version' => '1.838'
7071 },
7172 'MCE::Candy' => {
7273 'file' => 'lib/MCE/Candy.pm',
73 'version' => '1.837'
74 'version' => '1.838'
7475 },
7576 'MCE::Core::Input::Generator' => {
7677 'file' => 'lib/MCE/Core/Input/Generator.pm',
77 'version' => '1.837'
78 'version' => '1.838'
7879 },
7980 'MCE::Core::Input::Handle' => {
8081 'file' => 'lib/MCE/Core/Input/Handle.pm',
81 'version' => '1.837'
82 'version' => '1.838'
8283 },
8384 'MCE::Core::Input::Iterator' => {
8485 'file' => 'lib/MCE/Core/Input/Iterator.pm',
85 'version' => '1.837'
86 'version' => '1.838'
8687 },
8788 'MCE::Core::Input::Request' => {
8889 'file' => 'lib/MCE/Core/Input/Request.pm',
89 'version' => '1.837'
90 'version' => '1.838'
9091 },
9192 'MCE::Core::Input::Sequence' => {
9293 'file' => 'lib/MCE/Core/Input/Sequence.pm',
93 'version' => '1.837'
94 'version' => '1.838'
9495 },
9596 'MCE::Core::Manager' => {
9697 'file' => 'lib/MCE/Core/Manager.pm',
97 'version' => '1.837'
98 'version' => '1.838'
9899 },
99100 'MCE::Core::Validation' => {
100101 'file' => 'lib/MCE/Core/Validation.pm',
101 'version' => '1.837'
102 'version' => '1.838'
102103 },
103104 'MCE::Core::Worker' => {
104105 'file' => 'lib/MCE/Core/Worker.pm',
105 'version' => '1.837'
106 'version' => '1.838'
106107 },
107108 'MCE::Flow' => {
108109 'file' => 'lib/MCE/Flow.pm',
109 'version' => '1.837'
110 'version' => '1.838'
110111 },
111112 'MCE::Grep' => {
112113 'file' => 'lib/MCE/Grep.pm',
113 'version' => '1.837'
114 'version' => '1.838'
114115 },
115116 'MCE::Loop' => {
116117 'file' => 'lib/MCE/Loop.pm',
117 'version' => '1.837'
118 'version' => '1.838'
118119 },
119120 'MCE::Map' => {
120121 'file' => 'lib/MCE/Map.pm',
121 'version' => '1.837'
122 'version' => '1.838'
122123 },
123124 'MCE::Mutex' => {
124125 'file' => 'lib/MCE/Mutex.pm',
125 'version' => '1.837'
126 'version' => '1.838'
126127 },
127128 'MCE::Mutex::Channel' => {
128129 'file' => 'lib/MCE/Mutex/Channel.pm',
129 'version' => '1.837'
130 'version' => '1.838'
130131 },
131132 'MCE::Mutex::Flock' => {
132133 'file' => 'lib/MCE/Mutex/Flock.pm',
133 'version' => '1.837'
134 'version' => '1.838'
134135 },
135136 'MCE::Queue' => {
136137 'file' => 'lib/MCE/Queue.pm',
137 'version' => '1.837'
138 'version' => '1.838'
138139 },
139140 'MCE::Relay' => {
140141 'file' => 'lib/MCE/Relay.pm',
141 'version' => '1.837'
142 'version' => '1.838'
142143 },
143144 'MCE::Signal' => {
144145 'file' => 'lib/MCE/Signal.pm',
145 'version' => '1.837'
146 'version' => '1.838'
146147 },
147148 'MCE::Step' => {
148149 'file' => 'lib/MCE/Step.pm',
149 'version' => '1.837'
150 'version' => '1.838'
150151 },
151152 'MCE::Stream' => {
152153 'file' => 'lib/MCE/Stream.pm',
153 'version' => '1.837'
154 'version' => '1.838'
154155 },
155156 'MCE::Subs' => {
156157 'file' => 'lib/MCE/Subs.pm',
157 'version' => '1.837'
158 'version' => '1.838'
158159 },
159160 'MCE::Util' => {
160161 'file' => 'lib/MCE/Util.pm',
161 'version' => '1.837'
162 'version' => '1.838'
162163 }
163164 },
164165 'prereqs' => {
+3
-2
README.md less more
00 ## Many-Core Engine for Perl
11
2 This document describes MCE version 1.837.
2 This document describes MCE version 1.838.
33
44 Many-Core Engine (MCE) for Perl helps enable a new level of performance by
55 maximizing all available cores.
151151 bytes
152152 constant
153153 Carp
154 Errno
154155 Fcntl
155156 File::Path
156157 IO::Handle
176177
177178 ### Copyright and Licensing
178179
179 Copyright (C) 2012-2018 by Mario E. Roy <marioeroy AT gmail DOT com>
180 Copyright (C) 2012-2019 by Mario E. Roy <marioeroy AT gmail DOT com>
180181
181182 This program is free software; you can redistribute it and/or modify
182183 it under the same terms as Perl itself:
+3
-2
bin/mce_grep less more
9696 );
9797 use Fcntl qw( O_RDONLY );
9898 use Scalar::Util qw( looks_like_number );
99 use Errno ();
99100
100101 use MCE::Signal qw( -use_dev_shm );
101102 use MCE 1.5; # or later release
721722 WRITE: {
722723 $wrote += ( syswrite (
723724 $cmd_fh, $$chunk_ref, length($$chunk_ref) - $wrote, $wrote
724 )) || do {
725 redo WRITE if ($! && $!{'EINTR'});
725 )) or do {
726 redo WRITE if $! == Errno::EINTR();
726727 };
727728 }
728729
+2
-2
lib/MCE/Candy.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 our @CARP_NOT = qw( MCE );
1616
218218
219219 =head1 VERSION
220220
221 This document describes MCE::Candy version 1.837
221 This document describes MCE::Candy version 1.838
222222
223223 =head1 DESCRIPTION
224224
+1
-1
lib/MCE/Core/Input/Generator.pm less more
1414 use strict;
1515 use warnings;
1616
17 our $VERSION = '1.837';
17 our $VERSION = '1.838';
1818
1919 ## Items below are folded into MCE.
2020
+21
-28
lib/MCE/Core/Input/Handle.pm less more
1313 use strict;
1414 use warnings;
1515
16 our $VERSION = '1.837';
16 our $VERSION = '1.838';
1717
1818 ## Items below are folded into MCE.
1919
3737 # To minimize memory consumption, SEEK_CUR equals 1 on most platforms.
3838 # e.g. use Fcntl qw(SEEK_CUR);
3939
40 sysseek($_[0], 0, 1);
40 MCE::Util::_sysseek($_[0], 0, 1);
4141 }
4242
4343 sub _worker_read_handle {
7070 $_DAT_LOCK = $self->{'_mutex_'.( $self->{_wid} % 6 + 1 )};
7171 }
7272 $_dat_ex = sub {
73 sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
73 MCE::Util::_sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
7474 unless $_DAT_LOCK->{ $_pid };
7575 };
7676 $_dat_un = sub {
77 syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
77 MCE::Util::_syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
7878 if $_DAT_LOCK->{ $_pid };
7979 };
8080 }
111111
112112 ## Obtain the next chunk_id and offset position.
113113 $_dat_ex->() if $_lock_chn;
114
115 1 until sysread($_QUE_R_SOCK, $_next, $_que_read_size) || ($! && !$!{'EINTR'});
114 MCE::Util::_sysread($_QUE_R_SOCK, $_next, $_que_read_size);
116115
117116 ($_chunk_id, $_offset_pos) = unpack($_que_template, $_next);
118117
119118 if ($_offset_pos >= $_data_size) {
120 1 until syswrite ( $_QUE_W_SOCK,
121 pack($_que_template, 0, $_offset_pos)
122 ) || ($! && !$!{'EINTR'});
123
119 MCE::Util::_syswrite($_QUE_W_SOCK, pack($_que_template, 0, $_offset_pos));
124120 $_dat_un->() if $_lock_chn;
125
126121 close $_IN_FILE; undef $_IN_FILE;
127122 return;
128123 }
176171 }
177172 }
178173
179 1 until syswrite ( $_QUE_W_SOCK,
180 pack($_que_template, $_chunk_id, tell $_IN_FILE)
181 ) || ($! && !$!{'EINTR'});
182
174 MCE::Util::_syswrite(
175 $_QUE_W_SOCK, pack($_que_template, $_chunk_id, tell $_IN_FILE)
176 );
183177 $_dat_un->() if $_lock_chn;
184178 }
185179 else { # Large chunk.
186180 local $/ = $_RS if ($/ ne $_RS);
187181
188182 if ($_parallel_io && $_RS eq $LF) {
189 1 until syswrite ( $_QUE_W_SOCK,
183 MCE::Util::_syswrite(
184 $_QUE_W_SOCK,
190185 pack($_que_template, $_chunk_id, $_offset_pos + $_chunk_size)
191 ) || ($! && !$!{'EINTR'});
192
186 );
193187 $_dat_un->() if $_lock_chn;
194188
195189 $_tmp_cs = $_chunk_size;
200194 }
201195
202196 if ($_proc_type == READ_FILE) {
203 sysseek $_IN_FILE, tell( $_IN_FILE ), 0;
204 sysread $_IN_FILE, $_, $_tmp_cs, $_p;
205 seek $_IN_FILE, _systell( $_IN_FILE ), 0;
197 MCE::Util::_sysseek($_IN_FILE, tell( $_IN_FILE ), 0);
198 MCE::Util::_sysread($_IN_FILE, $_, $_tmp_cs, $_p);
199 seek $_IN_FILE, _systell($_IN_FILE), 0;
206200 }
207201 else {
208202 read $_IN_FILE, $_, $_tmp_cs, $_p;
212206 }
213207 else {
214208 if ($_proc_type == READ_FILE) {
215 sysseek $_IN_FILE, $_offset_pos, 0;
216 sysread $_IN_FILE, $_, $_chunk_size, $_p;
217 seek $_IN_FILE, _systell( $_IN_FILE ), 0;
209 MCE::Util::_sysseek($_IN_FILE, $_offset_pos, 0);
210 MCE::Util::_sysread($_IN_FILE, $_, $_chunk_size, $_p);
211 seek $_IN_FILE, _systell($_IN_FILE), 0;
218212 }
219213 else {
220214 seek $_IN_FILE, $_offset_pos, 0;
223217
224218 $_ .= <$_IN_FILE>;
225219
226 1 until syswrite ( $_QUE_W_SOCK,
227 pack($_que_template, $_chunk_id, tell $_IN_FILE)
228 ) || ($! && !$!{'EINTR'});
229
220 MCE::Util::_syswrite(
221 $_QUE_W_SOCK, pack($_que_template, $_chunk_id, tell $_IN_FILE)
222 );
230223 $_dat_un->() if $_lock_chn;
231224 }
232225 }
+3
-3
lib/MCE/Core/Input/Iterator.pm less more
1313 use strict;
1414 use warnings;
1515
16 our $VERSION = '1.837';
16 our $VERSION = '1.838';
1717
1818 ## Items below are folded into MCE.
1919
5454
5555 # inlined for performance
5656 $_dat_ex = sub {
57 sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
57 MCE::Util::_sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
5858 unless $_DAT_LOCK->{ $_pid };
5959 };
6060 $_dat_un = sub {
61 syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
61 MCE::Util::_syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
6262 if $_DAT_LOCK->{ $_pid };
6363 };
6464 }
+3
-3
lib/MCE/Core/Input/Request.pm less more
1313 use strict;
1414 use warnings;
1515
16 our $VERSION = '1.837';
16 our $VERSION = '1.838';
1717
1818 ## Items below are folded into MCE.
1919
5757
5858 # inlined for performance
5959 $_dat_ex = sub {
60 sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
60 MCE::Util::_sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
6161 unless $_DAT_LOCK->{ $_pid };
6262 };
6363 $_dat_un = sub {
64 syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
64 MCE::Util::_syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
6565 if $_DAT_LOCK->{ $_pid };
6666 };
6767 }
+7
-11
lib/MCE/Core/Input/Sequence.pm less more
1313 use strict;
1414 use warnings;
1515
16 our $VERSION = '1.837';
16 our $VERSION = '1.838';
1717
1818 ## Items below are folded into MCE.
1919
5959 $_DAT_LOCK = $self->{'_mutex_'.( $self->{_wid} % 6 + 1 )};
6060 }
6161 $_dat_ex = sub {
62 sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
62 MCE::Util::_sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
6363 unless $_DAT_LOCK->{ $_pid };
6464 };
6565 $_dat_un = sub {
66 syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
66 MCE::Util::_syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
6767 if $_DAT_LOCK->{ $_pid };
6868 };
6969 }
9696
9797 ## Obtain the next chunk_id and sequence number.
9898 $_dat_ex->() if $_lock_chn;
99
100 1 until sysread($_QUE_R_SOCK, $_next, $_que_read_size) || ($! && !$!{'EINTR'});
99 MCE::Util::_sysread($_QUE_R_SOCK, $_next, $_que_read_size);
101100
102101 ($_chunk_id, $_offset) = unpack($_que_template, $_next);
103102
104103 if ($_offset >= $_abort) {
105 1 until syswrite (
106 $_QUE_W_SOCK, pack($_que_template, 0, $_offset)
107 ) || ($! && !$!{'EINTR'});
108
104 MCE::Util::_syswrite($_QUE_W_SOCK, pack($_que_template, 0, $_offset));
109105 $_dat_un->() if $_lock_chn;
110106 return;
111107 }
112108
113 1 until syswrite (
109 MCE::Util::_syswrite(
114110 $_QUE_W_SOCK, pack($_que_template, $_chunk_id + 1, $_offset + 1)
115 ) || ($! && !$!{'EINTR'});
111 );
116112
117113 $_dat_un->() if $_lock_chn;
118114 $_chunk_id++;
+14
-65
lib/MCE/Core/Manager.pm less more
1313 use strict;
1414 use warnings;
1515
16 our $VERSION = '1.837';
16 our $VERSION = '1.838';
1717
1818 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1919 ## no critic (TestingAndDebugging::ProhibitNoStrict)
147147 if ($_task_id == 0 && defined $_syn_flag && $_sync_cnt) {
148148 if ($_sync_cnt == $_total_running) {
149149 for my $_i (1 .. $_total_running) {
150 1 until syswrite($_BSB_W_SOCK, $LF) || ($! && !$!{'EINTR'});
150 MCE::Util::_syswrite($_BSB_W_SOCK, $LF);
151151 }
152152 undef $_syn_flag;
153153 }
179179 if ($_task_id == 0 && defined $_syn_flag && $_sync_cnt) {
180180 if ($_sync_cnt == $_total_running) {
181181 for my $_i (1 .. $_total_running) {
182 1 until syswrite($_BSB_W_SOCK, $LF) || ($! && !$!{'EINTR'});
182 MCE::Util::_syswrite($_BSB_W_SOCK, $LF);
183183 }
184184 undef $_syn_flag;
185185 }
546546
547547 binmode $_sendto_fhs{$_file};
548548
549 ## Select new FH, turn on autoflush, restore the old FH.
550549 if ($_flush_file) {
551550 local $!;
552 # IO::Handle->autoflush not available in older Perl.
553 select(( select($_sendto_fhs{$_file}), $| = 1 )[0]);
551 $_sendto_fhs{$_file}->autoflush(1);
554552 }
555553 }
556554
575573
576574 binmode $_sendto_fhs{$_fd};
577575
578 ## Select new FH, turn on autoflush, restore the old FH.
579576 if ($_flush_file) {
580577 local $!;
581 # IO::Handle->autoflush not available in older Perl.
582 select(( select($_sendto_fhs{$_fd}), $| = 1 )[0]);
578 $_sendto_fhs{$_fd}->autoflush(1);
583579 }
584580 }
585581
602598
603599 if (++$_sync_cnt == $_total_running) {
604600 for my $_i (1 .. $_total_running) {
605 1 until syswrite($_BSB_W_SOCK, $LF) || ($! && !$!{'EINTR'});
601 MCE::Util::_syswrite($_BSB_W_SOCK, $LF);
606602 }
607603 undef $_syn_flag;
608604 }
617613 : $self->{_total_running};
618614
619615 for my $_i (1 .. $_total_running) {
620 1 until syswrite($_BSE_W_SOCK, $LF) || ($! && !$!{'EINTR'});
616 MCE::Util::_syswrite($_BSE_W_SOCK, $LF);
621617 }
622618 }
623619
625621 },
626622
627623 OUTPUT_S_IPC.$LF => sub { # Change to win32 IPC
628 1 until syswrite($_DAT_R_SOCK, $LF) || ($! && !$!{'EINTR'});
624 MCE::Util::_syswrite($_DAT_R_SOCK, $LF);
629625
630626 $_win32_ipc = 1, goto _LOOP unless $_win32_ipc;
631627
793789 }
794790
795791 ## Autoflush STDERR-STDOUT handles if requested.
796 ## Make MCE_STDOUT the default handle.
797
798 my $_old_hndl = select $_MCE_STDOUT;
799792
800793 {
801794 local $!;
802 # IO::Handle->autoflush not available in older Perl.
803 select($_MCE_STDERR), $| = 1 if ($self->{flush_stderr});
804 select($_MCE_STDOUT), $| = 1 if ($self->{flush_stdout});
805 select($_MCE_STDOUT);
795 $_MCE_STDERR->autoflush(1) if $self->{flush_stderr};
796 $_MCE_STDOUT->autoflush(1) if $self->{flush_stdout};
806797 }
807798
808799 ## -------------------------------------------------------------------------
873864
874865 ## Wait on requests *without* timeout capability.
875866
876 elsif ($^O eq 'MSWin32' && $_win32_ipc) {
877 # The normal loop hangs on Windows when processes/threads start/exit.
878 # Using ioctl() properly, http://www.perlmonks.org/?node_id=780083
879
880 my $_val_bytes = "\x00\x00\x00\x00";
881 my $_ptr_bytes = unpack( 'I', pack('P', $_val_bytes) );
882 my ($_done, $_count, $_nbytes, $_start) = (0);
883
884 while (!$_done) {
885 $_start = time, $_count = 1;
886
887 # MSWin32 FIONREAD
888 IOCTL: ioctl($_DAT_R_SOCK, 0x4004667f, $_ptr_bytes);
889
890 unless ($_nbytes = unpack('I', $_val_bytes)) {
891 if ($_count) {
892 # delay after a while to not consume a CPU core
893 $_count = 0 if ++$_count % 50 == 0 && time - $_start > 0.030;
894 } else {
895 sleep 0.030;
896 }
897 goto IOCTL;
898 }
899
900 do {
901 sysread($_DAT_R_SOCK, $_func, 8);
902 $_done = 1, last() unless length($_func) == 8;
903 $_DAU_R_SOCK = $_channels->[ substr($_func, -2, 2, '') ];
904
905 if (exists $_core_output_function{$_func}) {
906 $_core_output_function{$_func}();
907 } elsif (exists $_plugin_function->{$_func}) {
908 $_plugin_function->{$_func}();
909 }
910
911 } while (($_nbytes -= 8) >= 8);
912
913 last unless $self->{_total_running};
914 }
915 }
916
917867 elsif ($^O eq 'MSWin32') {
868 MCE::Util::_nonblocking($_DAT_R_SOCK, 1) if $_win32_ipc;
869
918870 while ($self->{_total_running}) {
919 sysread($_DAT_R_SOCK, $_func, 8);
871 MCE::Util::_sysread($_DAT_R_SOCK, $_func, 8);
920872 last() unless length($_func) == 8;
921873 $_DAU_R_SOCK = $_channels->[ substr($_func, -2, 2, '') ];
922874
927879 }
928880 }
929881 }
930
931882 else {
932883 while ($self->{_total_running}) {
933884 $_func = <$_DAT_R_SOCK>;
960911 delete $_sendto_fhs{$_p};
961912 }
962913
963 ## Restore the default handle. Close MCE STDOUT/STDERR handles.
964
965 select $_old_hndl;
914 ## Close MCE STDOUT/STDERR handles.
966915
967916 eval q{
968917 close $_MCE_STDOUT if (fileno $_MCE_STDOUT > 2);
+1
-1
lib/MCE/Core/Validation.pm less more
1313 use strict;
1414 use warnings;
1515
16 our $VERSION = '1.837';
16 our $VERSION = '1.838';
1717
1818 ## Items below are folded into MCE.
1919
+6
-7
lib/MCE/Core/Worker.pm less more
1313 use strict;
1414 use warnings;
1515
16 our $VERSION = '1.837';
16 our $VERSION = '1.838';
1717
1818 my $_has_threads = $INC{'threads.pm'} ? 1 : 0;
1919 my $_tid = $_has_threads ? threads->tid() : 0;
282282 # inlined for performance
283283 $_dat_ex = sub {
284284 my $_pid = $_has_threads ? $$ .'.'. $_tid : $$;
285 sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
285 MCE::Util::_sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
286286 unless $_DAT_LOCK->{ $_pid };
287287 };
288288 $_dat_un = sub {
289289 my $_pid = $_has_threads ? $$ .'.'. $_tid : $$;
290 syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
290 MCE::Util::_syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
291291 if $_DAT_LOCK->{ $_pid };
292292 };
293293 }
294294
295295 {
296296 local $!;
297 # IO::Handle->autoflush not available in older Perl.
298 select(( select(*STDERR), $| = 1 )[0]) if defined(fileno *STDERR);
299 select(( select(*STDOUT), $| = 1 )[0]) if defined(fileno *STDOUT);
297 (*STDERR)->autoflush(1) if defined( fileno *STDERR );
298 (*STDOUT)->autoflush(1) if defined( fileno *STDOUT );
300299 }
301300
302301 return;
585584 _worker_do($self, {}), next if ($_response eq "_data\n");
586585
587586 ## Wait here until MCE completes job submission to all workers.
588 1 until sysread($self->{_bse_r_sock}, my($_b), 1) || ($! && !$!{'EINTR'});
587 MCE::Util::_sysread($self->{_bse_r_sock}, my($_b), 1);
589588
590589 ## Normal request.
591590 if (defined $_job_delay && $_job_delay > 0.0) {
+1
-1
lib/MCE/Core.pod less more
44
55 =head1 VERSION
66
7 This document describes MCE::Core version 1.837
7 This document describes MCE::Core version 1.838
88
99 =head1 SYNOPSIS
1010
+1
-1
lib/MCE/Examples.pod less more
44
55 =head1 VERSION
66
7 This document describes MCE::Examples version 1.837
7 This document describes MCE::Examples version 1.838
88
99 =head1 INCLUDED WITH THE DISTRIBUTION
1010
+2
-2
lib/MCE/Flow.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
479479
480480 =head1 VERSION
481481
482 This document describes MCE::Flow version 1.837
482 This document describes MCE::Flow version 1.838
483483
484484 =head1 DESCRIPTION
485485
+2
-2
lib/MCE/Grep.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
434434
435435 =head1 VERSION
436436
437 This document describes MCE::Grep version 1.837
437 This document describes MCE::Grep version 1.838
438438
439439 =head1 SYNOPSIS
440440
+2
-2
lib/MCE/Loop.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
349349
350350 =head1 VERSION
351351
352 This document describes MCE::Loop version 1.837
352 This document describes MCE::Loop version 1.838
353353
354354 =head1 DESCRIPTION
355355
+2
-2
lib/MCE/Map.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
434434
435435 =head1 VERSION
436436
437 This document describes MCE::Map version 1.837
437 This document describes MCE::Map version 1.838
438438
439439 =head1 SYNOPSIS
440440
+9
-8
lib/MCE/Mutex/Channel.pm less more
1010
1111 no warnings qw( threads recursion uninitialized once );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 use base 'MCE::Mutex';
1616 use Scalar::Util qw(refaddr weaken);
2828 sub DESTROY {
2929 my ($pid, $obj) = ($has_threads ? $$ .'.'. $tid : $$, @_);
3030
31 syswrite($obj->{_w_sock}, '0'), $obj->{ $pid } = 0 if $obj->{ $pid };
31 MCE::Util::_syswrite($obj->{_w_sock}, '0'), $obj->{ $pid } = 0
32 if $obj->{ $pid };
3233
3334 if ($obj->{'_init_pid'} eq $pid) {
3435 my $addr = refaddr $obj;
6263 ? MCE::Util::_pipe_pair(\%obj, qw(_r_sock _w_sock))
6364 : MCE::Util::_sock_pair(\%obj, qw(_r_sock _w_sock));
6465
65 1 until syswrite($obj{_w_sock}, '0') || ($! && !$!{'EINTR'});
66 MCE::Util::_syswrite($obj{_w_sock}, '0');
6667
6768 if (caller !~ /^MCE:?/ || caller(1) !~ /^MCE:?/) {
6869 push(@MUTEX, \%obj); weaken($MUTEX[-1]);
7475 sub lock {
7576 my ($pid, $obj) = ($has_threads ? $$ .'.'. $tid : $$, @_);
7677
77 sysread($obj->{_r_sock}, my($b), 1), $obj->{ $pid } = 1
78 MCE::Util::_sysread($obj->{_r_sock}, my($b), 1), $obj->{ $pid } = 1
7879 unless $obj->{ $pid };
7980
8081 return;
8687 sub unlock {
8788 my ($pid, $obj) = ($has_threads ? $$ .'.'. $tid : $$, @_);
8889
89 syswrite($obj->{_w_sock}, '0'), $obj->{ $pid } = 0
90 MCE::Util::_syswrite($obj->{_w_sock}, '0'), $obj->{ $pid } = 0
9091 if $obj->{ $pid };
9192
9293 return;
99100 return unless ref($code) eq 'CODE';
100101
101102 # lock, run, unlock - inlined for performance
102 sysread($obj->{_r_sock}, my($b), 1), $obj->{ $pid } = 1
103 MCE::Util::_sysread($obj->{_r_sock}, my($b), 1), $obj->{ $pid } = 1
103104 unless $obj->{ $pid };
104105
105106 (defined wantarray)
106107 ? @ret = wantarray ? $code->(@_) : scalar $code->(@_)
107108 : $code->(@_);
108109
109 syswrite($obj->{_w_sock}, '0'), $obj->{ $pid } = 0;
110 MCE::Util::_syswrite($obj->{_w_sock}, '0'), $obj->{ $pid } = 0;
110111
111112 return wantarray ? @ret : $ret[-1];
112113 }
129130
130131 =head1 VERSION
131132
132 This document describes MCE::Mutex::Channel version 1.837
133 This document describes MCE::Mutex::Channel version 1.838
133134
134135 =head1 DESCRIPTION
135136
+2
-2
lib/MCE/Mutex/Flock.pm less more
1010
1111 no warnings qw( threads recursion uninitialized once );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 use base 'MCE::Mutex';
1616 use Fcntl ':flock';
185185
186186 =head1 VERSION
187187
188 This document describes MCE::Mutex::Flock version 1.837
188 This document describes MCE::Mutex::Flock version 1.838
189189
190190 =head1 DESCRIPTION
191191
+3
-6
lib/MCE/Mutex.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (TestingAndDebugging::ProhibitNoStrict)
1717
1818 use Carp ();
1919
20 use MCE::Mutex::Channel ();
21
2220 sub new {
2321 my ($class, %argv) = @_;
2422
2725
2826 $pkg = ucfirst( lc $pkg );
2927
30 if ($INC{"MCE/Mutex/$pkg.pm"} || eval "require MCE::Mutex::$pkg; 1") {
28 if (eval "require MCE::Mutex::$pkg; 1") {
3129 no strict 'refs'; $pkg = 'MCE::Mutex::'.$pkg;
32
3330 return $pkg->new(%argv);
3431 }
3532
7067
7168 =head1 VERSION
7269
73 This document describes MCE::Mutex version 1.837
70 This document describes MCE::Mutex version 1.838
7471
7572 =head1 SYNOPSIS
7673
+60
-75
lib/MCE/Queue.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (Subroutines::ProhibitExplicitReturnUndef)
1616 ## no critic (TestingAndDebugging::ProhibitNoStrict)
117117 my $_tid = $_has_threads ? threads->tid() : 0;
118118
119119 my %_valid_fields_new = map { $_ => 1 } qw(
120 await fast gather porder queue type
120 await barrier fast gather porder queue type
121121 );
122122
123123 my $_all = {};
171171 $_Q->{_heap} = []; # Priority heap [ pN, p2, p1 ] in heap order
172172 # fyi, _datp will always dequeue before _datq
173173
174 $_Q->{_await} = (exists $_argv{await} && defined $_argv{await})
174 $_Q->{_await} = (defined $_argv{await})
175175 ? $_argv{await} : $_def->{$_pkg}{AWAIT} || 0;
176 $_Q->{_fast} = (exists $_argv{fast} && defined $_argv{fast})
176 $_Q->{_fast} = (defined $_argv{fast})
177177 ? $_argv{fast} : $_def->{$_pkg}{FAST} || 0;
178178
179 $_Q->{_porder} = (exists $_argv{porder} && defined $_argv{porder})
179 $_Q->{_porder} = (defined $_argv{porder})
180180 ? $_argv{porder} : $_def->{$_pkg}{PORDER} || $HIGHEST;
181 $_Q->{_type} = (exists $_argv{type} && defined $_argv{type})
181 $_Q->{_type} = (defined $_argv{type})
182182 ? $_argv{type} : $_def->{$_pkg}{TYPE} || $FIFO;
183183
184184 ## -------------------------------------------------------------------------
206206 $_Q->{_id} = ++$_qid; $_all->{$_qid} = $_Q;
207207 $_Q->{_dsem} = 0 if $_Q->{_fast};
208208
209 if ($^O ne 'MSWin32' && $_tid == 0 && !$_Q->{_fast}) {
209 my $_barrier = defined $_argv{barrier} ? $_argv{barrier} : 1;
210
211 if ($^O ne 'MSWin32' && $_tid == 0 && !$_Q->{_fast} && $_barrier) {
210212 if (caller() !~ /^MCE::/) {
211213 for my $_i (0 .. MUTEX_LOCKS - 1) {
212214 $_Q->{'_mutex_'.$_i} = MCE::Mutex->new( impl => 'Channel' );
218220 MCE::Util::_sock_pair($_Q, qw(_ar_sock _aw_sock)) if $_Q->{_await};
219221
220222 if (exists $_argv{queue} && scalar @{ $_argv{queue} }) {
221 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
223 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
222224 }
223225
224226 return $_Q;
404406 $_Q->{_tsem} = $_t;
405407
406408 if ($_Q->pending() <= $_t) {
407 1 until syswrite($_Q->{_aw_sock}, $LF) || ($! && !$!{'EINTR'});
409 MCE::Util::_syswrite($_Q->{_aw_sock}, $LF);
408410 } else {
409411 $_Q->{_asem} += 1;
410412 }
457459 return;
458460 }
459461 if (!$_Q->{_nb_flag} && !$_Q->_has_data()) {
460 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
462 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
461463 }
462464 push @{ $_Q->{_datq} }, @{ $_MCE->{thaw}($_buf) };
463465 }
501503 return;
502504 }
503505 if (!$_Q->{_nb_flag} && !$_Q->_has_data()) {
504 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
506 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
505507 }
506508 push @{ $_Q->{_datq} }, $_;
507509 }
561563 if ($_pending) {
562564 $_pending = MAX_DQ_DEPTH if ($_pending > MAX_DQ_DEPTH);
563565 for my $_i (1 .. $_pending) {
564 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
566 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
565567 }
566568 }
567569 $_Q->{_dsem} = $_pending;
572574 }
573575 else {
574576 ## Otherwise, never to exceed one byte in the channel
575 if ($_Q->_has_data()) {
576 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
577 }
577 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF) if $_Q->_has_data();
578578 }
579579
580580 if (exists $_Q->{_ended} && !$_Q->_has_data()) {
581 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
581 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
582582 }
583583
584584 if ($_cnt) {
599599
600600 if ($_Q->{_await} && $_Q->{_asem} && $_Q->pending() <= $_Q->{_tsem}) {
601601 for my $_i (1 .. $_Q->{_asem}) {
602 1 until syswrite($_Q->{_aw_sock}, $LF) || ($! && !$!{'EINTR'});
602 MCE::Util::_syswrite($_Q->{_aw_sock}, $LF);
603603 }
604604 $_Q->{_asem} = 0;
605605 }
618618 $_Q = $_all->{$_id};
619619
620620 if (!$_Q->{_nb_flag} && $_Q->_has_data()) {
621 1 until sysread($_Q->{_qr_sock}, my($_b), 1) || ($! && !$!{'EINTR'});
621 MCE::Util::_sysread($_Q->{_qr_sock}, my($_b), 1);
622622 }
623623
624624 if ($_cnt == 1) {
659659
660660 if ($_Q->{_await} && $_Q->{_asem} && $_Q->pending() <= $_Q->{_tsem}) {
661661 for my $_i (1 .. $_Q->{_asem}) {
662 1 until syswrite($_Q->{_aw_sock}, $LF) || ($! && !$!{'EINTR'});
662 MCE::Util::_syswrite($_Q->{_aw_sock}, $LF);
663663 }
664664 $_Q->{_asem} = 0;
665665 }
844844 sub _mce_m_clear {
845845 my ($_Q) = @_;
846846
847 if ($_Q->{_fast}) {
848 warn "Queue: (clear) is not allowed for fast => 1\n";
849 }
850 else {
851 if ($_Q->_has_data()) {
852 1 until sysread($_Q->{_qr_sock}, my($_buf), 1) || ($! && !$!{'EINTR'});
853 }
854 %{ $_Q->{_datp} } = ();
855 @{ $_Q->{_datq} } = ();
856 @{ $_Q->{_heap} } = ();
857 }
847 %{ $_Q->{_datp} } = ();
848 @{ $_Q->{_datq} } = ();
849 @{ $_Q->{_heap} } = ();
858850
859851 return;
860852 }
865857 my ($_Q) = @_;
866858
867859 if (!exists $_Q->{_ended}) {
868 if (!$_Q->{_nb_flag}) {
869 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
870 }
860 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF) unless $_Q->{_nb_flag};
871861 $_Q->{_ended} = undef;
872862 }
873863
886876 return;
887877 }
888878 if (!$_Q->{_nb_flag} && !$_Q->_has_data()) {
889 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
879 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
890880 }
891881
892882 ## Append item(s) into the queue.
910900 return;
911901 }
912902 if (!$_Q->{_nb_flag} && !$_Q->_has_data()) {
913 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
903 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
914904 }
915905
916906 $_Q->_enqueuep($_p, @_);
925915 my ($_Q, $_cnt) = @_;
926916 my (@_items, $_buf, $_next, $_pending);
927917
928 1 until sysread($_Q->{_qr_sock}, $_next, 1) || ($! && !$!{'EINTR'});
918 MCE::Util::_sysread($_Q->{_qr_sock}, $_next, 1);
929919
930920 if (defined $_cnt && $_cnt ne '1') {
931921 _croak('Queue: (dequeue count argument) is not valid')
954944 if ($_pending) {
955945 $_pending = MAX_DQ_DEPTH if ($_pending > MAX_DQ_DEPTH);
956946 for my $_i (1 .. $_pending) {
957 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
947 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
958948 }
959949 }
960950 $_Q->{_dsem} = $_pending;
965955 }
966956 else {
967957 ## Otherwise, never to exceed one byte in the channel
968 if ($_Q->_has_data()) {
969 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
970 }
958 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF) if $_Q->_has_data();
971959 }
972960
973961 if (exists $_Q->{_ended} && !$_Q->_has_data()) {
974 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
962 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
975963 }
976964
977965 $_Q->{_nb_flag} = 0;
992980 }
993981
994982 if (!$_Q->{_nb_flag} && $_Q->_has_data()) {
995 1 until sysread($_Q->{_qr_sock}, my($_b), 1) || ($! && !$!{'EINTR'});
983 MCE::Util::_sysread($_Q->{_qr_sock}, my($_b), 1);
996984 }
997985
998986 if (defined $_cnt && $_cnt ne '1') {
10511039 return;
10521040 }
10531041 if (!$_Q->{_nb_flag} && !$_Q->_has_data()) {
1054 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
1042 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
10551043 }
10561044
10571045 if (abs($_i) > scalar @{ $_Q->{_datq} }) {
10991087 return;
11001088 }
11011089 if (!$_Q->{_nb_flag} && !$_Q->_has_data()) {
1102 1 until syswrite($_Q->{_qw_sock}, $LF) || ($! && !$!{'EINTR'});
1090 MCE::Util::_syswrite($_Q->{_qw_sock}, $LF);
11031091 }
11041092
11051093 if (exists $_Q->{_datp}->{$_p} && scalar @{ $_Q->{_datp}->{$_p} }) {
12211209 );
12221210
12231211 my $_is_MSWin32 = ($^O eq 'MSWin32') ? 1 : 0;
1224 my $_rdy = \&MCE::Util::_sock_ready;
12251212
12261213 my $_req1 = sub {
12271214 local $\ = undef if (defined $\);
12821269 # inlined for performance
12831270 $_dat_ex = sub {
12841271 my $_pid = $_has_threads ? $$ .'.'. $_tid : $$;
1285 sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
1272 MCE::Util::_sysread($_DAT_LOCK->{_r_sock}, my($b), 1), $_DAT_LOCK->{ $_pid } = 1
12861273 unless $_DAT_LOCK->{ $_pid };
12871274 };
12881275 $_dat_un = sub {
12891276 my $_pid = $_has_threads ? $$ .'.'. $_tid : $$;
1290 syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
1277 MCE::Util::_syswrite($_DAT_LOCK->{_w_sock}, '0'), $_DAT_LOCK->{ $_pid } = 0
12911278 if $_DAT_LOCK->{ $_pid };
12921279 };
12931280 }
13291316 $_t = 0 if ($_t < 0);
13301317 $_req2->(OUTPUT_W_QUE, $_Q->{_id}.$LF . $_t.$LF);
13311318
1332 $_rdy->($_Q->{_ar_sock}) if $_is_MSWin32;
1333 1 until sysread($_Q->{_ar_sock}, $_next, 1) || ($! && !$!{'EINTR'});
1319 MCE::Util::_sock_ready($_Q->{_ar_sock}) if $_is_MSWin32;
1320 MCE::Util::_sysread($_Q->{_ar_sock}, $_next, 1);
13341321
13351322 return;
13361323 }
13401327
13411328 return $_Q->_mce_m_clear() if (exists $_all->{ $_Q->{_id} });
13421329
1343 ($_Q->{_fast})
1344 ? warn "Queue: (clear) is not allowed for fast => 1\n"
1345 : $_req2->(OUTPUT_C_QUE, $_Q->{_id}.$LF);
1330 $_req2->(OUTPUT_C_QUE, $_Q->{_id}.$LF);
13461331
13471332 return;
13481333 }
14281413
14291414 if (exists $_Q->{'_mutex_0'}) {
14301415 $_Q->{'_mutex_'.$_mutexi}->lock();
1431
1432 $_rdy->($_Q->{_qr_sock}) if $_is_MSWin32;
1433 1 until sysread($_Q->{_qr_sock}, $_next, 1) || ($! && !$!{'EINTR'});
1416 MCE::Util::_sysread($_Q->{_qr_sock}, $_next, 1);
14341417
14351418 $_dat_ex->() if $_lock_chn;
1436 print({$_DAT_W_SOCK} OUTPUT_D_QUE.$LF . $_chn.$LF),
1437 print({$_DAU_W_SOCK} $_Q->{_id}.$LF . $_cnt.$LF);
1438
14391419 $_Q->{'_mutex_'.$_mutexi}->unlock();
14401420 }
14411421 else {
1442 $_rdy->($_Q->{_qr_sock}) if $_is_MSWin32;
1443 1 until sysread($_Q->{_qr_sock}, $_next, 1) || ($! && !$!{'EINTR'});
1422 MCE::Util::_sock_ready($_Q->{_qr_sock}) if $_is_MSWin32;
1423 MCE::Util::_sysread($_Q->{_qr_sock}, $_next, 1);
14441424
14451425 $_dat_ex->() if $_lock_chn;
1446 print({$_DAT_W_SOCK} OUTPUT_D_QUE.$LF . $_chn.$LF),
1447 print({$_DAU_W_SOCK} $_Q->{_id}.$LF . $_cnt.$LF);
1448 }
1426 }
1427
1428 print({$_DAT_W_SOCK} OUTPUT_D_QUE.$LF . $_chn.$LF),
1429 print({$_DAU_W_SOCK} $_Q->{_id}.$LF . $_cnt.$LF);
14491430
14501431 chomp($_len = <$_DAU_W_SOCK>);
14511432
16201601
16211602 =head1 VERSION
16221603
1623 This document describes MCE::Queue version 1.837
1604 This document describes MCE::Queue version 1.838
16241605
16251606 =head1 SYNOPSIS
16261607
17421723 =head2 MCE::Queue->new ( [ queue => \@array, await => 1, fast => 1 ] )
17431724
17441725 This creates a new queue. Available options are queue, porder, type, await,
1745 fast, and gather.
1726 barrier, fast, and gather.
17461727
17471728 use MCE;
17481729 use MCE::Queue;
17561737 my $q5 = MCE::Queue->new( type => $MCE::Queue::FIFO );
17571738 my $q6 = MCE::Queue->new( type => $MCE::Queue::LIFO );
17581739
1759 my $q7 = MCE::Queue->new( await => 1 );
1740 my $q7 = MCE::Queue->new( await => 1, barrier => 0 );
17601741 my $q8 = MCE::Queue->new( fast => 1 );
17611742
1762 The 'await' option, when enabled, allows workers to block (semaphore-like)
1763 until the number of items pending is equal or less than a threshold value.
1743 The C<await> option, when enabled, allows workers to block (semaphore-like)
1744 until the number of items pending is equal to or less than a threshold value.
17641745 The $q->await method is described below.
17651746
1766 The 'fast' option speeds up dequeues and is not enabled by default. It is
1767 beneficial for queues not calling (->clear or ->dequeue_nb) and not altering
1768 the optional count value while running; e.g. ->dequeue($count). Basically,
1769 do not enable 'fast' if varying the count dynamically.
1770
1771 The 'gather' option is mainly for running with MCE and wanting to pass item(s)
1747 On Unix platforms, C<barrier> mode (enabled by default) prevents many workers
1748 from dequeuing simultaneously to lessen overhead for the OS kernel. Specify 0
1749 to disable barrier mode and not allocate sockets. The barrier option has no
1750 effect if constructing the queue inside a thread or enabling C<fast>.
1751
1752 The C<fast> option speeds up dequeues and is not enabled by default. It is
1753 beneficial for queues not calling (->dequeue_nb) and not altering the count
1754 value while running; e.g. ->dequeue($count).
1755
1756 The C<gather> option is mainly for running with MCE and wanting to pass item(s)
17721757 to a callback function for appending to the queue. Multiple queues may point to
17731758 the same callback function. The callback receives the queue object as the first
17741759 argument and items after it.
17841769 ## Items are diverted to the callback function, not the queue.
17851770 $q7->enqueue( 'apple', 'orange' );
17861771
1787 Specifying the 'gather' option allows one to store items temporarily while
1772 Specifying the C<gather> option allows one to store items temporarily while
17881773 ensuring output order. Although a queue object is not required, this is
17891774 simply a demonstration of the gather option in the context of a queue.
17901775
+2
-2
lib/MCE/Relay.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
1616
342342
343343 =head1 VERSION
344344
345 This document describes MCE::Relay version 1.837
345 This document describes MCE::Relay version 1.838
346346
347347 =head1 SYNOPSIS
348348
+3
-2
lib/MCE/Signal.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616
195195 $SIG{__DIE__} = $SIG{__WARN__} = \&_NOOP;
196196
197197 if (exists $_sig_name_lkup{$_sig_name}) {
198 no warnings 'once';
198199 $SIG{INT} = $SIG{$_sig_name} = \&_NOOP,
199200 $_is_sig = $MCE::Signal::KILLED = 1;
200201 $_exit_status = 255 if ($_sig_name eq '__DIE__');
426427
427428 =head1 VERSION
428429
429 This document describes MCE::Signal version 1.837
430 This document describes MCE::Signal version 1.838
430431
431432 =head1 SYNOPSIS
432433
+2
-2
lib/MCE/Step.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
715715
716716 =head1 VERSION
717717
718 This document describes MCE::Step version 1.837
718 This document describes MCE::Step version 1.838
719719
720720 =head1 DESCRIPTION
721721
+2
-2
lib/MCE/Stream.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
671671
672672 =head1 VERSION
673673
674 This document describes MCE::Stream version 1.837
674 This document describes MCE::Stream version 1.838
675675
676676 =head1 SYNOPSIS
677677
+2
-2
lib/MCE/Subs.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
1616 ## no critic (TestingAndDebugging::ProhibitNoStrict)
203203
204204 =head1 VERSION
205205
206 This document describes MCE::Subs version 1.837
206 This document describes MCE::Subs version 1.838
207207
208208 =head1 SYNOPSIS
209209
+77
-40
lib/MCE/Util.pm less more
1010
1111 no warnings qw( threads recursion uninitialized numeric );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616
17 use IO::Handle ();
1718 use Socket qw( PF_UNIX PF_UNSPEC SOCK_STREAM SOL_SOCKET SO_SNDBUF SO_RCVBUF );
1819 use Time::HiRes qw( sleep time );
20 use Errno ();
1921 use base qw( Exporter );
2022 use bytes;
2123
4547 my $g_ncpu;
4648
4749 sub get_ncpu {
48
4950 return $g_ncpu if (defined $g_ncpu);
5051
5152 local $ENV{PATH} = "/usr/sbin:/sbin:/usr/bin:/bin:$ENV{PATH}";
148149 ###############################################################################
149150
150151 sub _destroy_pipes {
151
152152 my ($_obj, @_params) = @_;
153
154153 local ($!,$?); local $SIG{__DIE__};
155154
156155 for my $_p (@_params) {
173172 }
174173
175174 sub _destroy_socks {
176
177175 my ($_obj, @_params) = @_;
178
179176 local ($!,$?,$@); local $SIG{__DIE__};
180177
181178 for my $_p (@_params) {
185182 for my $_i (0 .. @{ $_obj->{$_p} } - 1) {
186183 next unless (defined $_obj->{$_p}[$_i]);
187184 if (fileno $_obj->{$_p}[$_i]) {
188 syswrite($_obj->{$_p}[$_i], '0') if $_is_winenv;
185 MCE::Util::_syswrite($_obj->{$_p}[$_i], '0') if $_is_winenv;
189186 eval q{ CORE::shutdown($_obj->{$_p}[$_i], 2) };
190187 close $_obj->{$_p}[$_i];
191188 }
194191 }
195192 else {
196193 if (fileno $_obj->{$_p}) {
197 syswrite($_obj->{$_p}, '0') if $_is_winenv;
194 MCE::Util::_syswrite($_obj->{$_p}, '0') if $_is_winenv;
198195 eval q{ CORE::shutdown($_obj->{$_p}, 2) };
199196 close $_obj->{$_p};
200197 }
206203 }
207204
208205 sub _pipe_pair {
209
210206 my ($_obj, $_r_sock, $_w_sock, $_i) = @_;
211
212207 local $!;
213208
214209 if (defined $_i) {
215210 # remove tainted'ness
216211 ($_i) = $_i =~ /(.*)/;
217
218 pipe($_obj->{$_r_sock}[$_i], $_obj->{$_w_sock}[$_i])
219 or die "pipe: $!\n";
220
221 # IO::Handle->autoflush not available in older Perl.
222 select(( select($_obj->{$_w_sock}[$_i]), $| = 1 )[0]);
212 pipe($_obj->{$_r_sock}[$_i], $_obj->{$_w_sock}[$_i]) or die "pipe: $!\n";
213 $_obj->{$_w_sock}[$_i]->autoflush(1);
223214 }
224215 else {
225 pipe($_obj->{$_r_sock}, $_obj->{$_w_sock})
226 or die "pipe: $!\n";
227
228 select(( select($_obj->{$_w_sock}), $| = 1 )[0]); # Ditto.
216 pipe($_obj->{$_r_sock}, $_obj->{$_w_sock}) or die "pipe: $!\n";
217 $_obj->{$_w_sock}->autoflush(1);
229218 }
230219
231220 return;
232221 }
233222
234223 sub _sock_pair {
235
236224 my ($_obj, $_r_sock, $_w_sock, $_i) = @_;
237
238225 my $_size = 16384; local $!;
239226
240227 if (defined $_i) {
251238 setsockopt($_obj->{$_w_sock}[$_i], SOL_SOCKET, SO_RCVBUF, int $_size);
252239 }
253240
254 # IO::Handle->autoflush not available in older Perl.
255 select(( select($_obj->{$_w_sock}[$_i]), $| = 1 )[0]);
256 select(( select($_obj->{$_r_sock}[$_i]), $| = 1 )[0]);
241 $_obj->{$_w_sock}[$_i]->autoflush(1);
242 $_obj->{$_r_sock}[$_i]->autoflush(1);
257243 }
258244 else {
259245 socketpair( $_obj->{$_r_sock}, $_obj->{$_w_sock},
266252 setsockopt($_obj->{$_w_sock}, SOL_SOCKET, SO_RCVBUF, int $_size);
267253 }
268254
269 select(( select($_obj->{$_w_sock}), $| = 1 )[0]); # Ditto.
270 select(( select($_obj->{$_r_sock}), $| = 1 )[0]);
255 $_obj->{$_w_sock}->autoflush(1);
256 $_obj->{$_r_sock}->autoflush(1);
271257 }
272258
273259 return;
274260 }
275261
276262 sub _sock_ready {
277
278263 my ($_socket, $_timeout) = @_;
279
280264 return '' if !defined $_timeout && exists $_sock_ready{"$_socket"};
281265
282266 my $_val_bytes = "\x00\x00\x00\x00";
283267 my $_ptr_bytes = unpack('I', pack('P', $_val_bytes));
284 my ($_count, $_start) = (1, time);
268 my ($_delay, $_start) = (0, time);
285269
286270 if (!defined $_timeout) {
287271 $_sock_ready{"$_socket"} = undef;
291275 }
292276
293277 while (1) {
294 # MSWin32 FIONREAD
278 # MSWin32 FIONREAD - from winsock2.h macro
295279 ioctl($_socket, 0x4004667f, $_ptr_bytes);
296280
297281 return '' if $_val_bytes ne $_zero_bytes;
298282 return 1 if $_timeout && time > $_timeout;
299283
300 if ($_count) {
301 # delay after a while to not consume a CPU core
302 $_count = 0 if ++$_count % 50 == 0 && time - $_start > 0.005;
303 next;
304 }
305
306 sleep 0.030;
307 }
284 # delay after a while to not consume a CPU core
285 sleep(0.030), next if $_delay;
286 $_delay = 1 if time - $_start > 0.005;
287 }
288 }
289
290 sub _sysread {
291 my ($_delay, $_start);
292
293 SYSREAD: ( @_ == 3
294 ? sysread($_[0], $_[1], $_[2])
295 : sysread($_[0], $_[1], $_[2], $_[3])
296 ) or do {
297 goto SYSREAD if $! == Errno::EINTR();
298
299 # non-blocking operation could not be completed
300 if ( $! == Errno::EWOULDBLOCK() || $! == Errno::EAGAIN() ) {
301 sleep(0.030), goto SYSREAD if $_delay;
302
303 # delay after a while to not consume a CPU core
304 $_start = time unless $_start;
305 $_delay = 1 if time - $_start > 0.005;
306
307 goto SYSREAD;
308 }
309 };
310
311 return;
312 }
313
314 sub _sysseek {
315 my $_pos;
316
317 SYSSEEK: $_pos = sysseek($_[0], $_[1], $_[2]) or do {
318 goto SYSSEEK if $! == Errno::EINTR();
319 };
320
321 return $_pos;
322 }
323
324 sub _syswrite {
325 syswrite($_[0], $_[1]) or do {
326 goto \&_syswrite if $! == Errno::EINTR();
327 };
328 return;
329 }
330
331 sub _nonblocking {
332 if ($^O eq 'MSWin32') {
333 my $nonblocking = ( $_[1] ) ? "\x00\x00\x00\x01" : "\x00\x00\x00\x00";
334
335 # MSWin32 FIONBIO - from winsock2.h macro
336 ioctl($_[0], 0x8004667e, unpack("I", pack('P', $nonblocking)));
337 }
338 else {
339 my $nonblocking = ( $_[1] ) ? 0 : 1;
340
341 $_[0]->blocking($nonblocking);
342 }
343
344 return;
308345 }
309346
310347 1;
323360
324361 =head1 VERSION
325362
326 This document describes MCE::Util version 1.837
363 This document describes MCE::Util version 1.838
327364
328365 =head1 SYNOPSIS
329366
+28
-26
lib/MCE.pm less more
1010
1111 no warnings qw( threads recursion uninitialized );
1212
13 our $VERSION = '1.837';
13 our $VERSION = '1.838';
1414
1515 ## no critic (BuiltinFunctions::ProhibitStringyEval)
1616 ## no critic (Subroutines::ProhibitSubroutinePrototypes)
578578 my $_DAT_W_SOCK = $TOP_HDLR->{_dat_w_sock}->[0];
579579 print {$_DAT_W_SOCK} OUTPUT_S_IPC.$LF . '0'.$LF;
580580
581 1 until sysread($_DAT_W_SOCK, my($_buf), 1) || ($! && !$!{'EINTR'});
581 MCE::Util::_sysread($_DAT_W_SOCK, my($_buf), 1);
582582 }
583583 }
584584
611611 : MCE::Util::_sock_pair($self, qw(_que_r_sock _que_w_sock));
612612
613613 if (defined $self->{init_relay}) { # relay
614 unless (exists $INC{'MCE/Relay.pm'}) {
614 unless ($INC{'MCE/Relay.pm'}) {
615615 require MCE::Relay; MCE::Relay->import();
616616 }
617617 MCE::Util::_sock_pair($self, qw(_rla_r_sock _rla_w_sock), $_)
625625 ## Spawn workers.
626626 $self->{_pids} = [], $self->{_thrs} = [], $self->{_tids} = [];
627627 $self->{_status} = [], $self->{_state} = [], $self->{_task} = [];
628
629 local $SIG{TTIN} unless $_is_MSWin32;
630 local $SIG{TTOU} unless $_is_MSWin32;
631 local $SIG{WINCH} unless $_is_MSWin32;
628632
629633 if (!defined $self->{user_tasks}) {
630634 $self->{_total_workers} = $_max_workers;
757761 unless (defined $MCE->{init_relay});
758762 }
759763
760 *relay_unlock = \&relay;
764 {
765 no warnings 'once';
766 *relay_unlock = \&relay;
767 }
761768
762769 sub AUTOLOAD {
763770 # $AUTOLOAD = MCE::<method_name>
11151122
11161123 ## Insert the first message into the queue if defined.
11171124 if (defined $_first_msg) {
1118 1 until syswrite (
1125 MCE::Util::_syswrite(
11191126 $self->{_que_w_sock}, pack($_que_template, 0, $_first_msg)
1120 ) || ($! && !$!{'EINTR'});
1127 );
11211128 }
11221129
11231130 ## Submit params data to workers.
11541161 ## Notify workers to commence processing.
11551162 if ($_is_MSWin32) {
11561163 my $_buf = _sprintf("%${_total_workers}s", "");
1157 syswrite $self->{_bse_w_sock}, $_buf;
1164 MCE::Util::_syswrite($self->{_bse_w_sock}, $_buf);
11581165 } else {
11591166 my $_BSE_W_SOCK = $self->{_bse_w_sock};
11601167 for my $_i (1 .. $_total_workers) {
1161 1 until syswrite($_BSE_W_SOCK, $LF) || ($! && !$!{'EINTR'});
1168 MCE::Util::_syswrite($_BSE_W_SOCK, $LF);
11621169 }
11631170 }
11641171 }
11721179
11731180 ## Remove the last message from the queue.
11741181 if (!$_send_cnt && $_run_mode ne 'nodata') {
1175 if (defined $self->{_que_r_sock}) {
1176 1 until sysread (
1177 $self->{_que_r_sock}, my($_buf), $_que_read_size
1178 ) || ($! && !$!{'EINTR'});
1179 }
1182 MCE::Util::_sysread($self->{_que_r_sock}, my($_buf), $_que_read_size)
1183 if ( defined $self->{_que_r_sock} );
11801184 }
11811185
11821186 $self->{_send_cnt} = 0;
13981402
13991403 ## Wait until all workers from (task_id 0) have synced.
14001404 MCE::Util::_sock_ready($_BSB_R_SOCK, -1) if $_is_MSWin32;
1401 1 until sysread($_BSB_R_SOCK, $_buf, 1) || ($! && !$!{'EINTR'});
1405 MCE::Util::_sysread($_BSB_R_SOCK, $_buf, 1);
14021406
14031407 ## Notify the manager process (barrier end).
14041408 print {$_DAT_W_SOCK} OUTPUT_E_SYN.$LF . $_chn.$LF;
14051409
14061410 ## Wait until all workers from (task_id 0) have un-synced.
1407 1 until sysread($_BSE_R_SOCK, $_buf, 1) || ($! && !$!{'EINTR'});
1411 MCE::Util::_sysread($_BSE_R_SOCK, $_buf, 1);
14081412
14091413 return;
14101414 }
14541458 local $\ = undef;
14551459
14561460 if ($_abort_msg > 0) {
1457 1 until sysread (
1458 $_QUE_R_SOCK, my($_next), $_que_read_size
1459 ) || ($! && !$!{'EINTR'});
1460 1 until syswrite (
1461 $_QUE_W_SOCK, pack($_que_template, 0, $_abort_msg)
1462 ) || ($! && !$!{'EINTR'});
1461 MCE::Util::_sysread($_QUE_R_SOCK, my($_next), $_que_read_size);
1462 MCE::Util::_syswrite($_QUE_W_SOCK, pack($_que_template, 0, $_abort_msg));
14631463 }
14641464
14651465 if ($self->{_wid} > 0) {
17921792
17931793 threads->exit(0) if $self->{use_threads};
17941794
1795 $SIG{HUP} = $SIG{INT} = $SIG{QUIT} = $SIG{TERM} = sub {
1796 $SIG{$_[0]} = $SIG{INT} = sub { };
1797 CORE::kill($_[0], getppid()) if ($_[0] eq 'INT' && !$_is_MSWin32);
1798 CORE::kill('KILL', $$);
1799 };
1795 if (! $_tid) {
1796 $SIG{HUP} = $SIG{INT} = $SIG{QUIT} = $SIG{TERM} = sub {
1797 $SIG{$_[0]} = $SIG{INT} = sub { };
1798 CORE::kill($_[0], getppid()) if ($_[0] eq 'INT' && !$_is_MSWin32);
1799 CORE::kill('KILL', $$);
1800 };
1801 }
18001802
18011803 if ($self->{posix_exit} && !$_is_MSWin32) {
18021804 eval { MCE::Mutex::Channel::_destroy() };
+2
-2
lib/MCE.pod less more
44
55 =head1 VERSION
66
7 This document describes MCE version 1.837
7 This document describes MCE version 1.838
88
99 Many-Core Engine (MCE) for Perl helps enable a new level of performance by
1010 maximizing all available cores.
281281
282282 =head1 COPYRIGHT AND LICENSE
283283
284 Copyright (C) 2012-2018 by Mario E. Roy
284 Copyright (C) 2012-2019 by Mario E. Roy
285285
286286 MCE is released under the same license as Perl.
287287