thetool/scripts/invoice/invoice-job-broker.php

#!/usr/bin/php
<?php

if (PHP_SAPI !== 'cli') {
    die("This program can only be run on the command line.\n");
}

require("../../config/config.php");
/*
fclose(STDIN);
fclose(STDOUT);
fclose(STDERR);
$STDIN = fopen('/dev/null', 'r');
$STDOUT = fopen(BASEDIR.'/var/log/invoice-job-broker.log', 'w');
$STDERR = fopen(BASEDIR.'/var/log/invoice-job-broker.err.log', 'w');
*/
define('mfUI',"cli");
define('FRONKDB_SQLDEBUG',false);
define("MFBASE_BYPASS_LOGIN", true);
error_reporting(E_ALL & ~(E_NOTICE | E_STRICT | E_DEPRECATED));

if(defined('MFLOCALE_TIME')) {
    setlocale(LC_TIME, MFLOCALE_TIME);
}
if(defined('MFLOCALE_MONETARY')) {
    setlocale(LC_MONETARY, MFLOCALE_MONETARY);
}
if(defined('MFLOCALE_NUMERIC')) {
    setlocale(LC_NUMERIC, MFLOCALE_NUMERIC);
}

require_once(LIBDIR."/mvcfronk/mfRouter/mfRouter.php");
require_once(LIBDIR."/mvcfronk/mfBase/mfBaseModel.php");
require_once(LIBDIR."/mvcfronk/mfBase/mfBaseController.php");

$me = new User(1);

define("INTERNAL_USER_ID", $me->id);
define("INTERNAL_USER_USERNAME", $me->username);

$request = array();

// Put commandline arguments into $request
if(count($argv)) {
    $args=$argv;
    array_shift($args); // shift scriptname off of args array
    foreach($args as $i => $arg) {
        if(preg_match('/^--(.+)/',$arg,$m)) {
            if(isset($args[$i+1]) && !preg_match('/^-/',$args[$i+1])) {
                $request[$m[1]] = $args[$i+1];
            } else {
                $request[$m[1]] = true;
            }
        } elseif(preg_match('/^-([a-zA-Z])(.+)/', $arg, $m)) {
            $request[$m[1]] = $m[2];
        } elseif(preg_match('/^-([^-])/', $arg, $m)) {
            if(isset($args[$i+1]) && !preg_match('/^-/',$args[$i+1])) {
                $request[$m[1]] = $args[$i + 1];
            } else {
                $request[$m[1]] = true;
            }
        }
    }
}

require_once(LIBDIR."/mvcfronk/mfRouter/mfRouter.php");
$log = mfLoghandler::singleton();

cli_set_process_title("thetool-invoice-job-broker");
pcntl_async_signals(true);
pcntl_signal(SIGTERM, 'signalHandler');

$forkcount = 0;
$childpids = [];

if(pidislocked()) {
    echo "Invoice Job Broker läuft bereits (pidfile vorhanden)\n";
    exit;
}
if(!lockpid()) {
    $log->error(__FILE__.": Error creating lock file!");
    die("Error creating lock file!\n");
}

$all_pids = [];

$valid_tasks = [
    "make-invoice-pdf",
    "send-invoice-email",
    "send-preorder-invoice-email",
];

while(1) {
    $processes = [];

    //sleep(5);
    //echo "looking for new jobs\n";
    $now = new DateTime("now");
    $jobs = InvoiceJobModel::search(["from_date<=" => $now->format("Y-m-d"), "to_date>=" => $now->format("Y-m-d"), "finished" => null], false, false, true);

    /*if(!count($jobs)) {
        echo "no more jobs. Exiting.\n";
        break;
    }*/

    foreach($jobs as $job) {
        $taskname = $job->task;

        if(!array_key_exists($job->task, $childpids)) {
            $childpids[$taskname] = [];
        }

        if($job->started && $job->status != "defer") {
            //echo "Job ".$job->id." is running already\n";
            continue;
        }

        $proc = [
            "job_id" => $job->id,
            "task" => $taskname,
            "pid" => false,
            "processtitle" => "$taskname - running job ".$job->id
        ];

        $processes[] = $proc;
    }

    if(!count($processes) && !count($all_pids)) {
        break;
    }

    foreach($processes as $proc) {
        //echo "process task ".$proc["task"]." pid: ".$proc["pid"]."\n";

        if($proc["pid"]) {
            // maybe look for new tasks here
            continue;
        }

        $taskname = $proc["task"];

        if($childpids[$taskname]) {
            //echo "cannot start new $taskname job, because another one is running already\n";
            continue;
        }
        //echo "Starting new child\n";
        $pid = pcntl_fork();


        if($pid === -1) {
            $log->debug("error forking");
            exit;
        } elseif($pid > 0) {
            // in parent
            $forkcount++;
            $childpids[$taskname] = $pid;
            $proc["pid"] = $pid;
            $all_pids[$pid] = $proc;
            //sleep(1);
        } else {
            // in child
            $mypid = getmypid();
            $job_id = $proc["job_id"];
            $job = new InvoiceJob($job_id, true);
            $job->status = "inprogress";
            $job->save();

            try {
                //echo "in pid $mypid\n";
                //echo "looking for runner for job $taskname\n";
                cli_set_process_title($proc["processtitle"]);


                $include_name = __DIR__ . "/job-runners/" . $taskname . ".php";
                if(!file_exists($include_name)) {
                    echo "[$mypid] Runner $include_name not found\n";
                }

                require($include_name);

                //echo "[$mypid] Runner $include_name is finished\n";

            } catch(\Exception $e) {
                // exit child process on error
                echo "$mypid caught exception: " . $e->getMessage();
                echo $e->getTraceAsString()."\n";
                exit;
            }
            exit; // make sure child exits when done
        }
    }

    if(count($all_pids)) {
        $status = false;
        $return_pid = pcntl_wait($status, WNOHANG);
        if($return_pid) {
            echo "child $return_pid returned\n";
            $pid_proc = $all_pids[$return_pid];
            $pid_task = $pid_proc["task"];
            $childpids[$pid_task] = null;
            unset($all_pids[$return_pid]);
        }
    }
    /*echo "No more PIDs, exiting loop\n";
    break;*/
    sleep(5);
}

unlockpid();

function signalHandler($sig) {
    //global $continue;
    global $childpids;
    global $invoice_job_lock;

    //$continue = false;
    //echo "in signal handler\n";

    if(count($childpids)) {
        foreach($childpids as $taskname => $pids) {
            foreach($pids as $childpid) {
                posix_kill($childpid, SIGTERM);
            }

        }
    }
    unlockpid();
    exit;
}

function client_log($pid, $text, $severity = "notice") {
    global $log;
    global $is_daemon;

    if($is_daemon) {
        echo "[".date('Y-m-d H:i:s')."] [$pid] $text\n";
    }
    $log->$severity($text);

    return true;
}

function pidislocked() {
    $pid = getmypid();
    $pidfile = __DIR__."/.invoice-job-broker.lock";
    if(file_exists($pidfile)) {
        return true;
    }
    return false;
}
function lockpid() {
    $pid = getmypid();
    $pidfile = __DIR__."/.invoice-job-broker.lock";
    file_put_contents($pidfile, $pid);
    if(file_exists($pidfile)) {
        return true;
    }
    return false;
}

function unlockpid() {
    $pid = getmypid();
    $pidfile = __DIR__."/.invoice-job-broker.lock";
    if(file_exists($pidfile)) {
        unlink($pidfile);
        return true;
    }
    return false;
}