Skip to content

Commit 4e0285b

Browse files
skrtdevskrtdev
committed
v1
1 parent 4f5a45b commit 4e0285b

File tree

6 files changed

+70
-100
lines changed

6 files changed

+70
-100
lines changed

‎composer.json

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,9 @@
1515
"autoload": {
1616
"psr-4":{
1717
"skrtdev\\async\\": "src/"
18-
}
18+
},
19+
"files":[
20+
"src/range.php"
21+
]
1922
}
2023
}

‎examples/concurrent.php

Lines changed: 5 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,14 @@
1010
require '../vendor/autoload.php';
1111

1212
use skrtdev\async\Pool;
13+
use function skrtdev\async\range;
1314

1415
$pool = new Pool();
1516

16-
for ($i=0; $i < 100; $i++) {
17-
$pool->parallel(function (int $i) {
18-
sleep(1);
19-
print("by the child n. $i".PHP_EOL);
20-
}, "my nice process name", $i);
21-
}
17+
$pool->iterate(range(0, 100), function (int $i) {
18+
sleep(1);
19+
print("by the child n. $i".PHP_EOL);
20+
});
2221

2322
print("Out of for, doing some external work...".PHP_EOL);
2423

‎examples/stayon.php

Lines changed: 6 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -5,12 +5,10 @@
55

66
$pool = new Pool();
77

8-
for ($i=0; $i < 10; $i++) {
9-
$pool->parallel(function () use ($i) {
10-
print("Parallel n.$i".PHP_EOL);
11-
while(true){
12-
sleep(1);
13-
}
8+
$pool->iterate(range(0, 10), function ($i) {
9+
print("Parallel n.$i".PHP_EOL);
10+
while(true){
11+
sleep(1);
12+
}
1413

15-
});
16-
}
14+
});

‎examples/while.php

Lines changed: 8 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -3,17 +3,17 @@
33
require '../vendor/autoload.php';
44

55
use skrtdev\async\Pool;
6+
use function skrtdev\async\range;
67

78
$pool = new Pool();
89

9-
for ($i=0; $i < 10; $i++) {
10-
$pool->parallel(function () use ($i) {
11-
for ($ii=0; $ii < 10; $ii++) {
12-
sleep(1);
13-
print("by the child n. $i - $ii".PHP_EOL);
14-
}
15-
});
16-
}
10+
$pool->iterate(range(0, 10), function ($i) {
11+
for ($ii=0; $ii < 10; $ii++) {
12+
sleep(1);
13+
print("by the child n. $i - $ii".PHP_EOL);
14+
}
15+
16+
});
1717

1818
while(true){
1919
sleep(1);

‎src/Pool.php

Lines changed: 38 additions & 77 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,6 @@
22

33
namespace skrtdev\async;
44

5-
use Closure;
65

76
class Pool{
87

@@ -19,15 +18,14 @@ class Pool{
1918

2019
public function __construct(?int $max_childs = null, bool $kill_childs = true)
2120
{
22-
if(!extension_loaded("pcntl")){
23-
throw new MissingExtensionException("PCNTL Extension is missing in your PHP build");
21+
if(!extension_loaded('pcntl')){
22+
throw new MissingExtensionException('PCNTL Extension is missing in your PHP build');
2423
}
2524
$this->pid = getmypid();
2625
$this->max_childs = $max_childs ?? (self::getCoresCount() ?? 1) * 10;
2726
$this->kill_childs = $kill_childs;
2827

29-
register_tick_function([$this, "tick"]);
30-
#pcntl_signal(SIGCHLD, SIG_IGN); // ignores the SIGCHLD signal
28+
register_tick_function([$this, 'tick']);
3129

3230
pcntl_async_signals(true);
3331

@@ -41,52 +39,33 @@ public function __construct(?int $max_childs = null, bool $kill_childs = true)
4139

4240
foreach ($this->childs as $key => $child) {
4341
if($pid === $child){
44-
self::breakpoint("removed child from signal handler");
42+
self::breakpoint('removed child from signal handler');
4543
unset($this->childs[$key]);
44+
break;
4645
}
4746
}
4847
}
4948
});
5049
}
5150

52-
public function checkChilds(): bool
53-
{
54-
return false;
55-
self::breakpoint("checkChilds()");
56-
$removed = 0;
57-
foreach ($this->childs as $key => $child) {
58-
if(!self::isProcessRunning($child)){
59-
unset($this->childs[$key]);
60-
$removed++;
61-
}
62-
}
63-
if($removed === 0){
64-
self::breakpoint("CheckChilds didn't remove any child");
65-
return false;
66-
}
67-
else{
68-
self::breakpoint("CheckChilds removed $removed childs");
69-
return true;
70-
};
71-
}
7251

73-
public function enqueue(Closure $closure, string $process_title = null, array $args = []): void
52+
public function enqueue(callable $callable, array $args = []): void
7453
{
75-
$this->queue[] = [$closure, $process_title, $args];
54+
$this->queue[] = [$callable, $args];
7655
}
7756

78-
protected function _parallel(Closure $closure, string $process_title = null, array $args = [])
57+
protected function _parallel(callable $callable, array $args = [])
7958
{
80-
self::breakpoint("started a parallel");
81-
self::breakpoint("parallel can be done: current childs: ".count($this->childs)."/".$this->max_childs);
59+
self::breakpoint('started a parallel');
60+
self::breakpoint('parallel can be done: current childs: '.count($this->childs).'/'.$this->max_childs);
8261
$pid = pcntl_fork();
8362
if ($pid == -1) {
84-
throw new CouldNotForkException("Pool could not fork");
63+
throw new CouldNotForkException('Pool could not fork');
8564
}
8665
elseif($pid){
8766
// we are the parent
8867
$this->childs[] = $pid;
89-
self::breakpoint("child started");
68+
self::breakpoint('child started');
9069
pcntl_wait($status, WNOHANG);
9170
}
9271
else{
@@ -95,63 +74,59 @@ protected function _parallel(Closure $closure, string $process_title = null, arr
9574
if (!$this->kill_childs) {
9675
pcntl_signal(SIGINT, SIG_IGN);
9776
}
98-
if(isset($process_title)){
99-
@cli_set_process_title($process_title);
100-
}
101-
$closure(...$args);
77+
$callable(...$args);
10278
exit;
10379
}
10480
}
10581

106-
public function parallel(Closure $closure, string $process_title = null, ...$args)
82+
public function parallel(callable $callable, ...$args)
10783
{
108-
if(count($this->childs) > $this->max_childs/2){
109-
#$this->checkChilds();
110-
}
11184
if($this->hasQueue()){
112-
self::breakpoint("resolving queue before parallel()");
85+
self::breakpoint('resolving queue before parallel()');
11386
$this->resolveQueue();
11487
if($this->hasQueue()){
115-
self::breakpoint("enqueueing because there is a queue");
116-
return $this->enqueue($closure, $process_title, $args);
88+
self::breakpoint('enqueueing because there is a queue');
89+
return $this->enqueue($callable, $args);
11790
}
11891
}
11992
elseif(count($this->childs) > $this->max_childs){
120-
self::breakpoint("enqueueing because of max reached (tried checkChilds but no results)");
121-
return $this->enqueue($closure, $process_title, $args);
93+
self::breakpoint('enqueueing because of max reached');
94+
return $this->enqueue($callable, $args);
95+
}
96+
return $this->_parallel($callable, $args);
97+
}
98+
99+
public function iterate(iterable $iterable, callable $callable): void
100+
{
101+
foreach ($iterable as $value) {
102+
$this->parallel($callable, $value);
122103
}
123-
return $this->_parallel($closure, $process_title, $args);
124104
}
125105

126106
public function resolveQueue(): void
127107
{
128108
if($this->is_resolving_queue) return;
129109

130110
if(count($this->childs) >= $this->max_childs){
131-
self::breakpoint("resolveQueue() -> too many childs, trying to remove...".PHP_EOL."check childs from resolveQueue()");
132-
if(true || !$this->checkChilds()){
133-
self::breakpoint("resolveQueue() exited because of too many childs");
134-
return;
135-
}
111+
self::breakpoint('resolveQueue() exited because of too many childs');
112+
return;
136113
}
137114

138115
$this->is_resolving_queue = true;
139116

140-
foreach ($this->queue as $key => $closure) {
117+
foreach ($this->queue as $key => $callable) {
141118
if(count($this->childs) < $this->max_childs){
142119
unset($this->queue[$key]);
143120
self::breakpoint("resolveQueue() is resolving n. $key");
144-
$this->_parallel(...$closure);
121+
$this->_parallel(...$callable);
145122
}
146123
else{
147-
self::breakpoint("resolveQueue() can't resolve, too many childs");
124+
self::breakpoint('resolveQueue() can\'t resolve, too many childs');
148125
break;
149-
self::breakpoint("check childs from resolveQueue()");
150-
$this->checkChilds();
151126
}
152127
}
153128
if(empty($this->queue)){
154-
self::breakpoint("queue is empty");
129+
self::breakpoint('queue is empty');
155130
}
156131

157132
$this->is_resolving_queue = false;
@@ -162,22 +137,16 @@ public function __destruct()
162137
{
163138
if($this->is_parent){
164139
$this->need_tick = false;
165-
self::breakpoint("triggered destructor");
140+
self::breakpoint('triggered destructor');
166141
$this->wait();
167142
}
168143
}
169144

170145
public static function getCoresCount(): ?int
171146
{
172-
if(isset(self::$cores_count) && self::$cores_count === 0) return null;
147+
if(isset(self::$cores_count)) return self::$cores_count === 0 ? null : self::$cores_count;
148+
173149

174-
if (defined('PHP_WINDOWS_VERSION_MAJOR')){
175-
$str = trim(shell_exec('wmic cpu get NumberOfCores 2>&1'));
176-
if (!preg_match('/(\d+)/', $str, $matches)) {
177-
$cores_count = null;
178-
}
179-
$cores_count = (int) $matches[1];
180-
}
181150
$ret = @shell_exec('nproc 2> /dev/null');
182151
if (is_string($ret)) {
183152
$ret = trim($ret);
@@ -199,19 +168,12 @@ public static function getCoresCount(): ?int
199168

200169
public static function breakpoint($value){
201170
return;
202-
usleep(5000);
203171
print($value.PHP_EOL);
204172
}
205173

206-
public static function isProcessRunning($pid): bool
207-
{
208-
return posix_getpgid($pid) !== false;
209-
}
210-
211174
public function tick()
212175
{
213176
if($this->is_parent && $this->need_tick && self::$last_tick !== time()){
214-
#print("tick".PHP_EOL);
215177
self::$last_tick = time();
216178
if(!$this->is_resolving_queue) $this->resolveQueue();
217179
}
@@ -240,7 +202,7 @@ public function getChildsCount(): int
240202
public function waitQueue(): void
241203
{
242204
while($this->hasQueue()){
243-
self::breakpoint("queue is not empty");
205+
self::breakpoint('queue is not empty');
244206
$this->resolveQueue();
245207
usleep(10000);
246208
}
@@ -249,8 +211,7 @@ public function waitQueue(): void
249211
public function waitChilds(): void
250212
{
251213
while($this->hasChilds()){
252-
self::breakpoint("there are still childs");
253-
#$this->checkChilds();
214+
self::breakpoint('there are still childs');
254215
usleep(10000);
255216
}
256217
}

‎src/range.php

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,9 @@
1+
<?php declare(strict_types=1);
2+
3+
namespace skrtdev\async;
4+
5+
function range (int $start, int $end, int $step = 1): \Generator {
6+
for ($i=$start; $i < $end; $i++) {
7+
yield $i;
8+
}
9+
}

0 commit comments

Comments
 (0)