Concurrent Execution

[sclubricants]

I'm interested in this project but I'm not sure it would work for what I need.

I created a rough scheduler but the scheduling part is not very good. However I really like the execution part.

I do a lot of ETL operations on databases. In my case I need to updated several tables, then maybe call some database functions to compute some things and then load the data into another system. I created controller for each command that process one table at a time or other action. Then I assemble all the command strings and use brackets to group commands I want to run in parallel.

For example:

$cmdString = "remote/table4/run/download;
                        {database/table1/run;
                        database/table2/run;
                        database/table3/run};
                        remote/table4/run/upload;"

/*
first run is remote/table4/run/download;

then                 {database/table1/run;
                        database/table2/run;
                        database/table3/run};  all run at the same time

then last run is remote/table4/run/upload; runs
*/

    public function runParallelCmds($cmdString)
    {
        set_time_limit(900);

        try {
            $parentjob = [
                'result'  => true,
                'error'   => '',
                'start'   => date('Y-m-d H:i:s'),
                'finish'  => null,
                'subjobs' => [],
            ];

            $commands = array_map('trim', explode(';', $cmdString));

            $results = [];

            foreach ($commands as $command) {
                $pathSeperater = '/';
                if (PHP_OS_FAMILY === 'Windows') {
                    $pathSeperater = '\\';
                }

                // write pid file
                $file = str_replace('/', '-', $command);
                $path = WRITEPATH . 'proccess' . $pathSeperater . $file . '.txt';
                if (file_exists($path)) {
                    @unlink($path);
                }

                $results[$file] = [
                    'isfinished' => false,
                    'command'    => $command,
                    'start'      => date('Y-m-d H:i:s'),
                    'finish'     => null,
                    'path'       => $path,
                    'json'       => '',
                ];

                if (PHP_OS_FAMILY === 'Windows') {
                    pclose(popen("start /B C:\\xampp\\php\\php.exe C:\\xampp\\portal\\public\\index.php cron/{$command} 1> {$path} 2>&1 &", 'r'));
                } elseif (PHP_OS_FAMILY === 'Linux') {
                    exec("php8.0 /var/www/portal/public/index.php cron/{$command} > {$path} 2>&1 &");
                }
            }

            while (true) {
                sleep(1);

                foreach ($results as $key => $result) {
                    // look if pid file has results
                    if (file_exists($result['path'])) {
                        $file = new \CodeIgniter\Files\File($result['path']);
                        if ((int) $file->getSize() > 0 && $results[$key]['isfinished'] === false) {
                            $results[$key]['isfinished'] = true;
                            $results[$key]['finish']     = date('Y-m-d H:i:s');
                            $json                        = json_decode(file_get_contents($result['path']));
                            if ($json === false) {
                                $string              = substr(str_replace(['  '], ' ', str_replace(["\r", "\n"], ' ', file_get_contents($result['path']))), 0, 255);
                                $json                = new stdClass();
                                $json->result        = false;
                                $json->error         = trim($string);
                                $parentjob['result'] = false;
                                $parentjob['error'] .= 'Error ' . $results[$key]['command'] . ';' . '<br>';
                            } else {
                                if ($json->result === false) { // json returned but job was in error ?
                                    $parentjob['error'] .= 'Error ' . $results[$key]['command'] . '; ' . $json->error . '<br>';
                                    $parentjob['result'] = false; // added?
                                }
                            }
                            $results[$key]['json'] = $json;
                        }
                    }
                }

                $alljobsfinished = true;

                foreach ($results as $result) {
                    if ($result['isfinished'] === false) {
                        $alljobsfinished = false;
                    }
                }

                if ($alljobsfinished === true) {
                    $parentjob['finish']  = date('Y-m-d H:i:s');
                    $parentjob['subjobs'] = (array) $results;

                    // if job finished then delete pid file
                    foreach ($results as $result) {
                        @unlink($result['path']);
                    }
                    // if all jobs finished then break loop
                    break;
                }
            }
            if ($parentjob['result'] === true) {
                foreach ($results as $result) {
                    $parentjob['error'] .= 'Cmd: ' . $result['command'] . ' | ' . $result['json']->error . '<br>';
                }
            }
        } catch (Throwable $e) {
            $result = false;
            $error  = $e->getMessage() . ' ' . basename($e->getFile()) . ' Line ' . $e->getLine();

            return ['result' => false, 'error' => $error]; // ?
        }

            return ['result' => $parentjob['result'], 'error' => $parentjob['error']];
    }

runParallelCmds() is called in a loop of a list of jobs. Once all subjobs complete then next job will execute.

A process file is created for each parallel command. each command outputs json response with status. when pid is filled with response then status is set. Once all jobs have status or timeout the infinite loop breaks. The code isn't very elegant but it does the job. One day php will have multi threading :)

The speed at which jobs process is immensely improved!

Any chance on creating similar functionality with this project?

[MGatner]

If you're on PHP >= 8.1 you should definitely look into Fibers!

Tasks is really about scheduling commands, the framework-equivalent to a "cron job". Your needs seem to align better with a Queue. We have a mostly-working implementation at the feature/queue branch, though it needs some sprucing and the final verdict was to release it as a supplemental package. There are lots of great Queue services with existing PHP implementations too: you don't need to do this "in CodeIgniter".