From f8d6e6a04f9bfa22b622227e1c267f45ad8e8bda Mon Sep 17 00:00:00 2001 From: Darko Date: Fri, 26 Sep 2014 13:39:55 +0200 Subject: [PATCH] First push for multiprocessing. --- bin/monitor.php | 14 +- bin/update_groups.php | 24 +- lib/DB/db.sql | 1 - lib/IRCScraper/scrape.php | 4 +- .../misc/sphinxsearch/create_se_tables.php | 4 +- lib/copy_this/misc/sphinxsearch/optimize.php | 2 +- .../misc/sphinxsearch/populate_rt_indexes.php | 6 +- .../misc/sphinxsearch/toggle_search_type.php | 4 +- .../multiprocessing/.do_not_run/require.php | 6 + .../multiprocessing/.do_not_run/switch.php | 337 +++ .../nix_scripts/multiprocessing/.gitignore | 1 + .../nix_scripts/multiprocessing/Forking.php | 1098 ++++++++++ .../multiprocessing/ForkingImportNZB.php | 90 + .../nix_scripts/multiprocessing/README.md | 19 + .../nix_scripts/multiprocessing/backfill.php | 7 + .../nix_scripts/multiprocessing/binaries.php | 9 + .../multiprocessing/fixrelnames.php | 16 + .../multiprocessing/forkdaemon-php/LICENSE | 19 + .../multiprocessing/forkdaemon-php/README.md | 20 + .../forkdaemon-php/composer.json | 14 + .../forkdaemon-php/examples/blocking.php | 63 + .../forkdaemon-php/examples/bucket.php | 78 + .../forkdaemon-php/examples/identifier.php | 81 + .../forkdaemon-php/examples/nonblocking.php | 71 + .../examples/result_callback.php | 86 + .../examples/result_deferred.php | 92 + .../forkdaemon-php/fork_daemon.php | 1855 +++++++++++++++++ .../nix_scripts/multiprocessing/import.php | 32 + .../multiprocessing/postprocess.php | 18 + .../nix_scripts/multiprocessing/releases.php | 4 + .../nix_scripts/multiprocessing/requestid.php | 4 + .../nix_scripts/multiprocessing/safe.php | 12 + .../multiprocessing/settings.php.example | 49 + .../multiprocessing/update_per_group.php | 5 + lib/copy_this/www/automated.config.php | 14 +- lib/copy_this/www/lib/RequestID.php | 10 +- lib/copy_this/www/lib/RequestIDLocal.php | 2 +- lib/copy_this/www/lib/RequestIDWeb.php | 4 +- lib/copy_this/www/lib/SphinxSearch.php | 4 +- lib/copy_this/www/lib/binaries.php | 2 +- lib/copy_this/www/lib/framework/basepage.php | 2 +- lib/copy_this/www/lib/releases.php | 2 +- lib/copy_this/www/pages/upcoming.php | 2 +- lib/namefixer.php | 2 +- lib/requestid.php | 6 +- lib/testing/getMovieCovers.php | 2 +- python/requestid_threaded.py | 2 +- 47 files changed, 4148 insertions(+), 51 deletions(-) create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/.do_not_run/require.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/.do_not_run/switch.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/.gitignore create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/Forking.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/ForkingImportNZB.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/README.md create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/backfill.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/binaries.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/fixrelnames.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/LICENSE create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/README.md create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/composer.json create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/blocking.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/bucket.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/identifier.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/nonblocking.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_callback.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_deferred.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/fork_daemon.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/import.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/postprocess.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/releases.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/requestid.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/safe.php create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/settings.php.example create mode 100644 lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/update_per_group.php diff --git a/bin/monitor.php b/bin/monitor.php index d471c1e61..e1b31dd31 100644 --- a/bin/monitor.php +++ b/bin/monitor.php @@ -169,6 +169,7 @@ $_cj = dirname(__FILE__) . "/../nnscripts"; $_lib = dirname(__FILE__) . "/../lib"; $_py = dirname(__FILE__) . "/../python"; $_user = dirname(__FILE__) . "/../user_scripts"; +$_multi = dirname(_FILE_) . "/../../multiprocessing"; $NNPATH = "/var/www/newznab/misc/update_scripts/"; @@ -1677,13 +1678,12 @@ while ($i > 0) { $color = get_color($colors_start, $colors_end, $colors_exc); $log = writelog($panes3[0]); shell_exec("tmux respawnp -t${tmux_session}:3.0 ' \ - cd $_lib && $_php fixReleaseNames.php 1 true other yes show 2>&1 $log; \ - cd $_py && $_python ${DIR}/../python/groupfixrelnames_threaded.py nfo 2>&1 $log; \ - $_python ${DIR}/../python/groupfixrelnames_threaded.py md5 2>&1 $log; \ - $_python ${DIR}/../python/groupfixrelnames_threaded.py filename 2>&1 $log; \ - $_python ${DIR}/../python/groupfixrelnames_threaded.py par2 2>&1 $log; \ - $_python ${DIR}/../python/groupfixrelnames_threaded.py miscsorter 2>&1 $log; \ - $_python ${DIR}/../python/groupfixrelnames_threaded.py predbft 2>&1 $log; $_sleep $fix_timer' 2>&1 1> /dev/null" + cd $_multi && $_php fixrelnames.php md5 $log; \ + $_php fixrelnames.php filename $log; \ + $_php fixrelnames.php nfo $log; \ + $_php fixrelnames.php par2 $log; \ + $_php fixrelnames.php miscsorter $log; \ + $_php fixrelnames.php predbft $log; $_sleep $fix_timer' 2>&1 1> /dev/null" ); $time27 = TIME(); } elseif (($maxload >= get_load()) && ($fix_names == 1)) { diff --git a/bin/update_groups.php b/bin/update_groups.php index a6d2c98ae..e1d03e441 100644 --- a/bin/update_groups.php +++ b/bin/update_groups.php @@ -1,7 +1,7 @@ $pdo]); if ($nntp->doConnect() !== true) { - exit($c->error("Unable to connect to usenet.")); + exit($pdo->log->error("Unable to connect to usenet.")); } -echo $c->header("Getting first/last for all your active groups."); + +echo $pdo->log->header("Getting first/last for all your active groups."); $data = $nntp->getGroups(); if ($nntp->isError($data)) { - exit($c->error("Failed to getGroups() from nntp server.")); + exit($pdo->log->error("Failed to getGroups() from nntp server.")); } -echo $c->header("Inserting new values into shortgroups table."); +echo $pdo->log->header("Inserting new values into shortgroups table."); -$db = new DB(); -$db->queryExec('TRUNCATE TABLE shortgroups'); +$pdo->queryExec('TRUNCATE TABLE shortgroups'); // Put into an array all active groups -$res = $db->query('SELECT name FROM groups WHERE active = 1'); +$res = $pdo->query('SELECT name FROM groups WHERE active = 1 OR backfill = 1'); foreach ($data as $newgroup) { if (myInArray($res, $newgroup['group'], 'name')) { - $db->queryInsert(sprintf('INSERT INTO shortgroups (name, first_record, last_record, updated) VALUES (%s, %s, %s, NOW())', $db->escapeString($newgroup['group']), $db->escapeString($newgroup['first']), $db->escapeString($newgroup['last']))); - echo $c->primary('Updated ' . $newgroup['group']); + $pdo->queryInsert(sprintf('INSERT INTO shortgroups (name, first_record, last_record, updated) VALUES (%s, %s, %s, NOW())', $pdo->escapeString($newgroup['group']), $pdo->escapeString($newgroup['first']), $pdo->escapeString($newgroup['last']))); + echo $pdo->log->primary('Updated ' . $newgroup['group']); } } -echo $c->header('Running time: ' . $consoleTools->convertTimer(TIME() - $start)); +echo $pdo->log->header('Running time: ' . $consoleTools->convertTimer(TIME() - $start)); function myInArray($array, $value, $key) { diff --git a/lib/DB/db.sql b/lib/DB/db.sql index 8251c9248..85ffd75da 100644 --- a/lib/DB/db.sql +++ b/lib/DB/db.sql @@ -191,7 +191,6 @@ INSERT INTO tmux (setting, value) VALUES ('defrag_cache', '900'), ('predb', '0'), ('predb_timer', '600'), ('safebackfilldate', '2012-06-24'), -('safepartrepair', '0'), ('request_hours', '1'), ('trakttvkey', ''), ('fanarttvkey', ''), diff --git a/lib/IRCScraper/scrape.php b/lib/IRCScraper/scrape.php index 75f40fa9b..33af0b2ad 100644 --- a/lib/IRCScraper/scrape.php +++ b/lib/IRCScraper/scrape.php @@ -2,7 +2,7 @@ require_once(dirname(__FILE__)."/../../bin/config.php"); require_once(dirname(__FILE__) . "/../IRCScraper.php"); -if (!is_file(NN_ROOT . '../misc/update_scripts/nix_scripts/tmux/lib/IRCScraper/settings.php')) { +if (!is_file(NN_TMUX . 'lib' . DS . 'IRCScraper' .DS . 'settings.php')) { exit('Copy settings_example.php to settings.php and change the settings.' . PHP_EOL); } @@ -19,7 +19,7 @@ if (!isset($argv[1]) || $argv[1] !== 'true') { ); } -require_once NN_ROOT . '../misc/update_scripts/nix_scripts/tmux/lib/IRCScraper/settings.php'; +require_once (NN_TMUX . 'lib' . DS . 'IRCScraper' .DS . 'settings.php'); if (!defined('SCRAPE_IRC_NICKNAME')) { exit('ERROR! You must update settings.php using settings_example.php.'); diff --git a/lib/copy_this/misc/sphinxsearch/create_se_tables.php b/lib/copy_this/misc/sphinxsearch/create_se_tables.php index dbd7514db..d3871963a 100644 --- a/lib/copy_this/misc/sphinxsearch/create_se_tables.php +++ b/lib/copy_this/misc/sphinxsearch/create_se_tables.php @@ -1,7 +1,7 @@ (string): python|php +// The type of process we want to do: $options[1] => (string): releases +$options = explode(' ', $argv[1]); + +switch ($options[1]) { + + // Runs backFill interval or all. + // $options[2] => (string)group name, Name of group to work on. + // $options[3] => (int) backfill type from tmux settings. 1 = Backfill interval , 2 = Bakfill all + case 'backfill': + if (in_array((int)$options[3], [1, 2])) { + $pdo = new \DB(); + $value = $pdo->queryOneRow("SELECT value FROM tmux WHERE setting = 'backfill_qty'"); + if ($value !== false) { + $nntp = nntp($pdo); + (new \Backfill())->backfillAllGroups($options[2], ($options[3] == 1 ? '' : $value['value'])); + } + } + break; + + /* BackFill up to x number of articles for all groups. + * + * $options[2] => (string) Group name. + * $options[3] => (int) Quantity of articles to download. + */ + case 'backfill_all_quantity': + $pdo = new \DB(); + $nntp = nntp($pdo); + (new \Backfill())->backfillAllGroups($options[2], $options[3]); + break; + + // BackFill a single group, 10000 parts. + // $options[2] => (string)group name, Name of group to work on. + case 'backfill_all_quick': + $pdo = new \DB(); + $nntp = nntp($pdo); + (new \Backfill())->backfillAllGroups($options[2], 10000, 'normal'); + break; + + /* Get a range of article headers for a group. + * + * $options[2] => (string) backfill/binaries + * $options[3] => (string) Group name. + * $options[4] => (int) First article number in range. + * $options[5] => (int) Last article number in range. + * $options[6] => (int) Number of threads. + */ + case 'get_range': + $pdo = new \DB(); + $nntp = nntp($pdo); + $groups = new \Groups(); + $groupMySQL = $groups->getByName($options[3]); + if ($nntp->isError($nntp->selectGroup($groupMySQL['name']))) { + if ($nntp->isError($nntp->dataError($nntp, $groupMySQL['name']))) { + return; + } + } + $binaries = new \Binaries(); + $return = $binaries->scan($groupMySQL, $options[4], $options[5], ($site->safepartrepair == 1 ? 'update' : 'backfill')); + if (empty($return)) { + exit(); + } + $columns = []; + switch ($options[2]) { + case 'binaries': + if ($return['lastArticleNumber'] <= $groupMySQL['last_record']){ + exit(); + } + $columns[1] = sprintf( + 'last_record_postdate = %s', + $pdo->from_unixtime( + (is_numeric($return['lastArticleDate']) ? $return['lastArticleDate'] : strtotime($return['lastArticleDate'])) + ) + ); + $columns[2] = sprintf('last_record = %s', $return['lastArticleNumber']); + $query = sprintf( + 'UPDATE groups SET %s, %s, last_updated = NOW() WHERE ID = %d AND last_record < %s', + $columns[1], + $columns[2], + $groupMySQL['ID'], + $return['lastArticleNumber'] + ); + break; + case 'backfill': + if ($return['firstArticleNumber'] >= $groupMySQL['first_record']){ + exit(); + } + $columns[1] = sprintf( + 'first_record_postdate = %s', + $pdo->from_unixtime( + (is_numeric($return['firstArticleDate']) ? $return['firstArticleDate'] : strtotime($return['firstArticleDate'])) + ) + ); + $columns[2] = sprintf('first_record = %s', $return['firstArticleNumber']); + $query = sprintf( + 'UPDATE groups SET %s, %s, last_updated = NOW() WHERE ID = %d AND first_record > %s', + $columns[1], + $columns[2], + $groupMySQL['ID'], + $return['firstArticleNumber'] + ); + break; + default: + exit(); + } + $pdo->queryExec($query); + break; + + /* Do part repair for a group. + * + * $options[2] => (string) Group name. + */ + case 'part_repair': + $pdo = new \DB(); + $groups = new \Groups(['Settings' => $pdo]); + $groupMySQL = $groups->getByName($options[2]); + $nntp = nntp($pdo); + // Select group, here, only once + $data = $nntp->selectGroup($groupMySQL['name']); + if ($nntp->isError($data)) { + if ($nntp->dataError($nntp, $groupMySQL['name']) === false) { + exit(); + } + } + (new \Binaries())->partRepair($nntp, $groupMySQL); + break; + + // Process releases. + // $options[2] => (string)groupCount, number of groups terminated by _ | (int)groupID, group to work on + case 'releases': + $pdo = new \DB(); + $releases = new \Releases(); + + //Runs function that are per group + if (is_numeric($options[2])) { + + if ($options[0] === 'python') { + collectionCheck($pdo, $options[2]); + } + + processReleases($releases, $options[2]); + + } else { + + // Run functions that run on releases table after all others completed. + $groupCount = rtrim($options[2], '_'); + if (!is_numeric($groupCount)) { + $groupCount = 1; + } + $releases->deletedReleasesByGroup(); + $releases->deleteReleases(); + $releases->processRequestIDs('', (5000 * $groupCount), true); + $releases->processRequestIDs('', (1000 * $groupCount), false); + $releases->categorizeReleases(2); + } + break; + + // Process all local requestID for a single group. + // $options[2] => (int)groupID, group to work on + case 'requestid': + if (is_numeric($options[2])) { + (new \RequestIDLocal(['Echo' => true]))->lookupRequestIDs(['GroupID' => $options[2], 'limit' => 5000]); + } + break; + + /* Update a single group's article headers. + * + * $options[2] => (string) Group name. + */ + case 'update_group_headers': + $pdo = new \DB(); + $nntp = nntp($pdo); + $groups = new \Groups(); + $groupMySQL = $groups->getByName($options[2]); + (new \Binaries())->updateGroup($groupMySQL); + break; + + + // Do a single group (update_binaries/backFill/update_releases/postprocess). + // $options[2] => (int)groupID, group to work on + case 'update_per_group': + if (is_numeric($options[2])) { + + $pdo = new \DB(); + + // Get the group info from MySQL. + $groupMySQL = $pdo->queryOneRow(sprintf('SELECT * FROM groups WHERE ID = %d', $options[2])); + + if ($groupMySQL === false) { + exit('ERROR: Group not found with ID ' . $options[2] . PHP_EOL); + } + + // Connect to NNTP. + $nntp = nntp($pdo); + $backFill = new \Backfill(); + + // Update the group for new binaries. + (new \Binaries())->updateGroup($groupMySQL); + + // BackFill the group with 20k articles. + $backFill->backfillAllGroups($groupMySQL['name'], 20000, 'normal'); + + // Check if we got anything from binaries/backFill, exit if not. + collectionCheck($pdo, $options[2]); + + // Create releases. + processReleases(new \Releases(), $options[2]); + + // Post process the releases. + (new \ProcessAdditional(['Echo' => true, 'NNTP' => $nntp, 'Settings' => $pdo]))->start($options[2]); + (new \Info(['Echo' => true, 'Settings' => $pdo]))->processNfoFiles($nntp, $options[2]); + + } + break; + + // Post process additional and NFO. + // $options[2] => (char)Letter or number a-f 0-9, first character of release guid. + case 'pp_additional': + case 'pp_nfo': + if (charCheck($options[2])) { + $pdo = new \DB(); + + // Create the connection here and pass, this is for post processing, so check for alternate. + $nntp = nntp($pdo, true); + + if ($options[1] === 'pp_nfo') { + (new \Info(['Echo' => true, 'Settings' => $pdo]))->processInfoFiles($nntp, '', $options[2]); + } else { + (new \ProcessAdditional(['Echo' => true, 'NNTP' => $nntp, 'Settings' => $pdo]))->start('', $options[2]); + } + } + break; + + /* Post process movies. + * + * $options[2] (char) Single character, first letter of release guid. + * $options[3] (int) Process all releases or renamed releases only. + */ + case 'pp_movie': + if (charCheck($options[2])) { + $pdo = new \DB(); + (new \PProcess(['Settings' => $pdo]))->processMovies('', $options[2], (isset($options[3]) ? $options[3] : '')); + } + break; + + /* Post process TV. + * + * $options[2] (char) Single character, first letter of release guid. + * $options[3] (int) Process all releases or renamed releases only. + */ + case 'pp_tv': + if (charCheck($options[2])) { + $pdo = new \DB(); + (new \PProcess(['Settings' => $pdo]))->processTv('', $options[2], (isset($options[3]) ? $options[3] : '')); + } + break; +} + +/** + * Create / process releases for a groupID. + * + * @param \Releases $releases + * @param int $groupID + */ +function processReleases($releases, $groupID) +{ + $releases->processIncompleteCollections($groupID); + $releases->processCollectionSizes($groupID); + $releases->deleteUnwantedCollections($groupID); + $releases->createReleases($groupID); + $releases->createNZBs($groupID); + $releases->deleteCollections($groupID); +} + +/** + * Check if the character contains a-f or 0-9. + * + * @param string $char + * + * @return bool + */ +function charCheck($char) +{ + if (in_array($char, ['a','b','c','d','e','f','0','1','2','3','4','5','6','7','8','9'])) { + return true; + } + return false; +} + +/** + * Check if the group should be processed. + * + * @param \DB $pdo + * @param int $groupID + */ +function collectionCheck(&$pdo, $groupID) +{ + if ($pdo->queryOneRow(sprintf('SELECT ID FROM collections_%d LIMIT 1', $groupID)) === false) { + exit(); + } +} + +/** + * Connect to usenet, return NNTP object. + * + * @param \DB $pdo + * @param bool $alternate Use alternate NNTP provider. + * + * @return NNTP + */ +function &nntp(&$pdo, $alternate = false) +{ + $nntp = new \NNTP(['Settings' => $pdo]); + $s = new Sites(); + $site = $s->get(); + if (($alternate && $site->alternate_nntp == 1 ? $nntp->doConnect(true, true) : $nntp->doConnect()) !== true) { + exit("ERROR: Unable to connect to usenet." . PHP_EOL); + } + + return $nntp; +} \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/.gitignore b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/.gitignore new file mode 100644 index 000000000..9f37e3549 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/.gitignore @@ -0,0 +1 @@ +/settings.php diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/Forking.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/Forking.php new file mode 100644 index 000000000..e9c4891fc --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/Forking.php @@ -0,0 +1,1098 @@ +_colorCLI = new \ColorCLI(); + $this->site = new \Sites(); + + $this->register_logging( + [0 => $this, 1 => 'logger'], + (defined('NN_MULTIPROCESSING_LOG_TYPE') ? NN_MULTIPROCESSING_LOG_TYPE : \fork_daemon::LOG_LEVEL_INFO) + ); + + $this->max_work_per_child_set(1); + if (defined('NN_MULTIPROCESSING_MAX_CHILD_WORK')) { + $this->max_work_per_child_set(NN_MULTIPROCESSING_MAX_CHILD_WORK); + } + + $this->child_max_run_time_set(1800); + if (defined('NN_MULTIPROCESSING_MAX_CHILD_TIME')) { + $this->child_max_run_time_set(NN_MULTIPROCESSING_MAX_CHILD_TIME); + } + + // Use a single exit method for all children, makes things easier. + $this->register_parent_child_exit([0 => $this, 1 => 'childExit']); + + $this->outputType = self::OUTPUT_REALTIME; + if (defined('NN_MULTIPROCESSING_CHILD_OUTPUT_TYPE')) { + switch (NN_MULTIPROCESSING_CHILD_OUTPUT_TYPE) { + case 0: + $this->outputType = self::OUTPUT_NONE; + break; + case 1: + $this->outputType = self::OUTPUT_REALTIME; + break; + case 2: + $this->outputType = self::OUTPUT_SERIALLY; + break; + default: + $this->outputType = self::OUTPUT_REALTIME; + } + } + + $this->dnr_path = PHP_BINARY . ' ' . NN_MULTI . '.do_not_run' . DS . 'switch.php "php '; + } + + /** + * Setup the class to work on a type of work, then process the work. + * Valid work types: + * + * @param string $type The type of multiProcessing to do : backfill, binaries, releases, postprocess + * @param array $options Array containing arguments for the type of work. + * + * @throws ForkingException + */ + public function processWorkType($type, array $options = []) + { + // Set/reset some variables. + $startTime = microtime(true); + $this->workType = $type; + $this->workTypeOptions = $options; + $this->processAdditional = $this->processNFO = $this->processTV = $this->processMovies = $this->tablePerGroup = $this->ppRenamedOnly = false; + $this->work = []; + + // Init Settings here, as forking causes errors when it's destroyed. + $this->pdo = new \DB(); + + // Process extra work that should not be forked and done before forking. + $this->processStartWork(); + + // Get work to fork. + $this->getWork(); + + // Now we destroy settings, to prevent errors from forking. + unset($this->pdo); + + // Process the work we got. + $this->processWork(); + + // Process extra work that should not be forked and done after. + $this->processEndWork(); + + if (NN_ECHOCLI) { + $this->_colorCLI->doEcho( + $this->_colorCLI->header( + 'Multi-processing for ' . $this->workType . ' finished in ' . (microtime(true) - $startTime) . + ' seconds at ' . date(DATE_RFC2822) . '.' . PHP_EOL + ) + ); + } + } + + /** + * Only post process renamed movie / tv releases? + * @var bool + */ + private $ppRenamedOnly; + + /** + * Get work for our workers to work on, set the max child processes here. + */ + private function getWork() + { + $maxProcesses = 0; + + switch ($this->workType) { + + case 'backfill': + $maxProcesses = $this->backfillMainMethod(); + break; + + case 'binaries': + $maxProcesses = $this->binariesMainMethod(); + break; + + case 'fixRelNames_nfo': + case 'fixRelNames_filename': + case 'fixRelNames_md5': + case 'fixRelNames_par2': + case 'fixRelNames_miscsorter': + case 'fixRelNames_predbft': + $maxProcesses = $this->fixRelNamesMainMethod(); + break; + + case 'releases': + $maxProcesses = $this->releasesMainMethod(); + break; + + case 'postProcess_ama': + $this->processSingle(); + break; + + case 'postProcess_add': + $maxProcesses = $this->postProcessAddMainMethod(); + break; + + case 'postProcess_mov': + $this->ppRenamedOnly = (isset($this->workTypeOptions[0]) && $this->workTypeOptions[0] === true ? true : false); + $maxProcesses = $this->postProcessMovMainMethod(); + break; + + case 'postProcess_nfo': + $maxProcesses = $this->postProcessNfoMainMethod(); + break; + + case 'postProcess_sha': + $this->processSharing(); + break; + + case 'postProcess_tv': + $this->ppRenamedOnly = (isset($this->workTypeOptions[0]) && $this->workTypeOptions[0] === true ? true : false); + $maxProcesses = $this->postProcessTvMainMethod(); + break; + + case 'request_id': + $maxProcesses = $this->requestIDMainMethod(); + break; + + case 'safe_backfill': + $maxProcesses = $this->safeBackfillMainMethod(); + break; + + case 'safe_binaries': + $maxProcesses = $this->safeBinariesMainMethod(); + break; + + case 'update_per_group': + $maxProcesses = $this->updatePerGroupMainMethod(); + break; + } + + $this->setMaxProcesses($maxProcesses); + } + + /** + * Process work if we have any. + */ + private function processWork() + { + $this->_workCount = count($this->work); + if ($this->_workCount > 0) { + + if (NN_ECHOCLI) { + $this->_colorCLI->doEcho( + $this->_colorCLI->header( + 'Multi-processing started at ' . date(DATE_RFC2822) . ' for ' . $this->workType . ' with ' . $this->_workCount . + ' job(s) to do using a max of ' . $this->maxProcesses . ' child process(es).' + ) + ); + } + + $this->addwork($this->work); + $this->process_work(true); + } else { + if (NN_ECHOCLI) { + $this->_colorCLI->doEcho( + $this->_colorCLI->header('No work to do!') + ); + } + } + } + + /** + * Process any work that does not need to be forked, but needs to run at the end. + */ + private function processStartWork() + { + switch ($this->workType) { + //case 'safe_backfill': + case 'safe_binaries': + $this->_executeCommand( + PHP_BINARY . ' ' . NN_NIX . 'tmux/bin/update_groups.php' + ); + break; + } + } + + /** + * Process any work that does not need to be forked, but needs to run at the end. + */ + private function processEndWork() + { + switch ($this->workType) { + case 'releases': + if ($this->tablePerGroup === true) { + $this->_executeCommand( + $this->dnr_path . 'releases ' . count($this->work) . '_"' + ); + } + break; + case 'update_per_group': + $this->_executeCommand( + $this->dnr_path . 'releases ' . count($this->work) . '_"' + ); + break; + case 'safe_backfill': + $this->_executeCommand( + $this->dnr_path . 'backfill_all_quantity ' . $this->safeBackfillGroup . ' 1000' . '"' + ); + break; + } + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + //////////////////////////////////////// All backFill code here //////////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + /** + * @return int + */ + private function backfillMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'backFillChildWorker']); + // The option for backFill is for doing up to x articles. Else it's done by date. + $this->work = $this->pdo->query( + sprintf( + 'SELECT name %s FROM groups WHERE backfill = 1', + ($this->workTypeOptions[0] === false ? '' : (', ' . $this->workTypeOptions[0] . ' AS max')) + ) + ); + return $this->site->get('backfillthreads'); + } + + public function backFillChildWorker($groups, $identifier = '') + { + foreach ($groups as $group) { + $this->_executeCommand( + PHP_BINARY . ' ' . NN_UPDATE . 'backfill.php ' . + $group['name'] . (isset($group['max']) ? (' ' . $group['max']) : '') + ); + } + } + + /** + * @return int + */ + private function safeBackfillMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'safeBackfillChildWorker']); + + $run = $this->pdo->query("SELECT (SELECT value FROM tmux WHERE setting = 'backfill_qty') AS qty, (SELECT value FROM tmux WHERE setting = 'backfill') AS backfill, (SELECT value FROM tmux WHERE setting = 'backfill_order') AS orderby, (SELECT value FROM tmux WHERE setting = 'backfill_days') AS days, (SELECT value FROM site WHERE setting = 'maxmssgs') AS maxmsgs"); + $threads = $this->site->get('backfillthreads'); + + $orderby = "ORDER BY a.last_record ASC"; + switch ((int)$run[0]['orderby']) { + case 1: + $orderby = "ORDER BY first_record_postdate DESC"; + break; + + case 2: + $orderby = "ORDER BY first_record_postdate ASC"; + break; + + case 3: + $orderby = "ORDER BY name ASC"; + break; + + case 4: + $orderby = "ORDER BY name DESC"; + break; + + case 5: + $orderby = "ORDER BY a.last_record DESC"; + break; + } + + $backfilldays = ''; + if ($run[0]['days'] == 1) { + $backfilldays = "backfill_target"; + } elseif ($run[0]['days'] == 2) { + $backfilldays = round(abs(strtotime(date("Y-m-d")) - strtotime($this->site->get('safebackfilldate'))) / 86400);; + } + + $data = $this->pdo->queryOneRow( + sprintf( + "SELECT g.name, + g.first_record AS our_first, + MAX(a.first_record) AS their_first, + MAX(a.last_record) AS their_last + FROM groups g + INNER JOIN shortgroups a ON g.name = a.name + WHERE g.first_record IS NOT NULL + AND g.first_record_postdate IS NOT NULL + AND g.backfill = 1 + AND (NOW() - INTERVAL %s DAY) < g.first_record_postdate + GROUP BY a.name, a.last_record, g.name, g.first_record + %s", + $backfilldays, + $orderby + ) + ); + + $count = 0; + if ($data['name']) { + $this->safeBackfillGroup = $data['name']; + + $count = ($data['our_first'] - $data['their_first']); + } + + if ($count > 0) { + if ($count > ($run[0]['qty'] * $threads)) { + $geteach = ceil(($run[0]['qty'] * $threads) / $run[0]['maxmsgs']); + } else { + $geteach = $count / $run[0]['maxmsgs']; + } + + $queue = array(); + for ($i = 0; $i <= $geteach - 1; $i++) { + $queue[$i] = sprintf("get_range backfill %s %s %s %s", $data['name'], $data['our_first'] - $i * $run[0]['maxmsgs'] - $run[0]['maxmsgs'], $data['our_first'] - $i * $run[0]['maxmsgs'] - 1, $i + 1); + } + $this->work = $queue; + } + + return $threads; + } + + public function safeBackfillChildWorker($ranges, $identifier = '') + { + foreach ($ranges as $range) { + $this->_executeCommand( + $this->dnr_path . $range . '"' + ); + } + return; + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + //////////////////////////////////////// All binaries code here //////////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + private function binariesMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'binariesChildWorker']); + $this->work = $this->pdo->query( + sprintf( + 'SELECT name, %d AS max FROM groups WHERE active = 1', + $this->workTypeOptions[0] + ) + ); + return $this->site->get('binarythreads'); + } + + public function binariesChildWorker($groups, $identifier = '') + { + foreach ($groups as $group) { + $this->_executeCommand( + PHP_BINARY . ' ' . NN_UPDATE . 'update_binaries.php ' . $group['name'] . ' ' . $group['max'] + ); + } + } + + /** + * @return int + */ + private function safeBinariesMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'safeBinariesChildWorker']); + + $maxmssgs = $this->site->get('maxmssgs'); + $threads = $this->site->get('binarythreads'); + + $groups = $this->pdo->query("SELECT g.name AS groupname, g.last_record AS our_last, a.last_record AS their_last FROM groups g INNER JOIN shortgroups a ON g.active = 1 AND g.name = a.name ORDER BY a.last_record DESC"); + + if ($groups) { + $i = 1; + $queue = array(); + foreach ($groups as $group) { + if ($group['our_last'] == 0) { + $queue[$i] = sprintf("update_group_headers %s", $group['groupname']); + $i++; + } else { + //only process if more than 20k headers available and skip the first 20k + $count = $group['their_last'] - $group['our_last'] - 20000; + //echo "count: " . $count . "maxmsgs x2: " . ($maxmssgs * 2) . PHP_EOL; + if ($count <= $maxmssgs * 2) { + $queue[$i] = sprintf("update_group_headers %s", $group['groupname']); + $i++; + } else { + $queue[$i] = sprintf("part_repair %s", $group['groupname']); + $i++; + $geteach = floor($count / $maxmssgs); + $remaining = $count - $geteach * $maxmssgs; + //echo "maxmssgs: " . $maxmssgs . " geteach: " . $geteach . " remaining: " . $remaining . PHP_EOL; + for ($j = 0; $j < $geteach; $j++) { + $queue[$i] = sprintf("get_range binaries %s %s %s %s", $group['groupname'], $group['our_last'] + $j * $maxmssgs + 1, $group['our_last'] + $j * $maxmssgs + $maxmssgs, $i); + $i++; + } + //add remainder to queue + $queue[$i] = sprintf("get_range binaries %s %s %s %s", $group['groupname'], $group['our_last'] + ($j + 1) * $maxmssgs + 1, $group['our_last'] + ($j + 1) * $maxmssgs + $remaining + 1, $i); + $i++; + } + } + } + //var_dump($queue); + $this->work = $queue; + } + + return $threads; + } + + public function safeBinariesChildWorker($ranges, $identifier = '') + { + foreach ($ranges as $range) { + $this->_executeCommand( + $this->dnr_path . $range . '"' + ); + } + return; + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + //////////////////////////////////// All fix release names code here /////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + private function fixRelNamesMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'fixRelNamesChildWorker']); + + $join = ""; + $where = ""; + $groupby = "GROUP BY guidchar"; + $orderby = "ORDER BY guidchar ASC"; + $rowLimit = "LIMIT 16"; + $extrawhere = "AND r.prehashID = 0 AND r.nzbstatus = 1"; + $select = "DISTINCT LEFT(r.guid, 1) AS guidchar, COUNT(*) AS count"; + + + $threads = $this->site->get('fixnamethreads'); + $maxperrun = $this->site->get('fixnamesperrun'); + + if ($threads > 16) { + $threads = 16; + } + switch($this->workTypeOptions[0]) { + case "md5": + $join = "LEFT OUTER JOIN releasefiles rf ON r.ID = rf.releaseID AND rf.ishashed = 1"; + $where = "r.ishashed = 1 AND r.dehashstatus BETWEEN -6 AND 0"; + break; + + case "nfo": + $where = "r.proc_nfo = 0 AND r.nfostatus = 1"; + break; + + case "filename": + $join = "INNER JOIN releasefiles rf ON r.ID = rf.releaseID"; + $where = "r.proc_files = 0"; + break; + + case "par2": + $where = "r.proc_par2 = 0"; + break; + + case "miscsorter": + $where = "r.nfostatus = 1 AND r.proc_nfo = 1 AND r.proc_sorter = 0 AND r.isrenamed = 0"; + break; + + case "predbft": + $extrawhere = ""; + $where = "1=1"; + $rowLimit = sprintf("LIMIT %s", $threads); + break; + } + + $datas = $this->pdo->query(sprintf("SELECT %s FROM releases r %s WHERE %s %s %s %s %s", $select, $join, $where, $extrawhere, $groupby, $orderby, $rowLimit)); + + if ($datas) { + $count = 0; + $queue = array(); + foreach ($datas as $firstguid) { + if ($count >= $threads) { + $count = 0; + } + $count++; + if ($firstguid['count'] < $maxperrun) { + $limit = $firstguid['count']; + } else { + $limit = $maxperrun; + } + if ($limit > 0) { + $queue[$count] = sprintf("%s %s %s %s", $this->workTypeOptions[0], $firstguid['guidchar'], $limit, $count); + } + } + $this->work = $queue; + } + return $threads; + } + + public function fixRelNamesChildWorker($guids, $identifier = '') + { + foreach ($guids as $guid) { + $this->_executeCommand( + PHP_BINARY . ' ' . NN_NIX . 'tmux/bin/groupfixrelnames.php "' . $guid . '"' . ' true' + ); + } + return; + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + //////////////////////////////////////// All releases code here //////////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + private function releasesMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'releasesChildWorker']); + + $this->tablePerGroup = ($this->site->get('tablepergroup') == 1 ? true : false); + if ($this->tablePerGroup === true) { + + $groups = $this->pdo->queryDirect('SELECT ID FROM groups WHERE (active = 1 OR backfill = 1)'); + + if ($groups instanceof \Traversable) { + foreach($groups as $group) { + if ($this->pdo->queryOneRow(sprintf('SELECT ID FROM collections_%d LIMIT 1',$group['id'])) !== false) { + $this->work[] = ['id' => $group['id']]; + } + } + } + } else { + $this->work = $this->pdo->query('SELECT name FROM groups WHERE (active = 1 OR backfill = 1)'); + } + + return $this->site->get('releasesthreads'); + } + + public function releasesChildWorker($groups, $identifier = '') + { + foreach ($groups as $group) { + if ($this->tablePerGroup === true) { + $this->_executeCommand( + $this->dnr_path . 'releases ' . $group['id'] . '"' + ); + } else { + $this->_executeCommand( + PHP_BINARY . ' ' . NN_UPDATE . 'update_releases.php 1 false ' . $group['name'] + ); + } + } + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + /////////////////////////////////////// All post process code here ///////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + /** + * Only 1 exit method is used for post process, since they are all similar. + * + * @param $groups + * @param string $identifier + */ + public function postProcessChildWorker($groups, $identifier = '') + { + foreach ($groups as $group) { + $type = ''; + if ($this->processAdditional) { + $type = 'pp_additional '; + } else if ($this->processNFO) { + $type = 'pp_nfo '; + } else if ($this->processMovies) { + $type = 'pp_movie '; + } else if ($this->processTV) { + $type = 'pp_tv '; + } + + if ($type !== '') { + $this->_executeCommand( + $this->dnr_path . $type . $group['id'] . (isset($group['renamed']) ? (' ' . $group['renamed']) : '') . '"' + ); + } + } + } + + private $ppAddMinSize = ''; + private $ppAddMaxSize = ''; + + /** + * Check if we should process Additional's. + * @return bool + */ + private function checkProcessAdditional() + { + $this->ppAddMinSize = + (string)($this->site->get('minsizetopostprocess') != '') ? $this->site->get('minsizetopostprocess') : 1; + $this->ppAddMinSize = ($this->ppAddMinSize === 0 ? '' : 'AND r.size > ' . ($this->ppAddMinSize * 1048576)); + $this->ppAddMaxSize = + (string)($this->site->get('maxsizetopostprocess') != '') ? $this->site->get('maxsizetopostprocess') : 100; + $this->ppAddMaxSize = ($this->ppAddMaxSize === 0 ? '' : 'AND r.size < ' . ($this->ppAddMaxSize * 1073741824)); + return ( + $this->pdo->queryOneRow( + sprintf(' + SELECT r.ID + FROM releases r + LEFT JOIN category c ON c.ID = r.categoryID + WHERE r.nzbstatus = %d + AND r.passwordstatus BETWEEN -6 AND -1 + AND r.haspreview = -1 + AND c.disablepreview = 0 + %s %s + LIMIT 1', + \Enzebe::NZB_ADDED, + $this->ppAddMaxSize, + $this->ppAddMinSize + ) + ) === false ? false : true + ); + } + + private function postProcessAddMainMethod() + { + $maxProcesses = 1; + if ($this->checkProcessAdditional() === true) { + $this->processAdditional = true; + $this->register_child_run([0 => $this, 1 => 'postProcessChildWorker']); + $this->work = $this->pdo->query( + sprintf(' + SELECT LEFT(r.guid, 1) AS id + FROM releases r + LEFT JOIN category c ON c.ID = r.categoryID + WHERE r.nzbstatus = %d + AND r.passwordstatus BETWEEN -6 AND -1 + AND r.haspreview = -1 + AND c.disablepreview = 0 + %s %s + GROUP BY LEFT(r.guid, 1) + LIMIT 16', + \Enzebe::NZB_ADDED, + $this->ppAddMaxSize, + $this->ppAddMinSize + ) + ); + $maxProcesses = $this->site->get('postthreads'); + } + return $maxProcesses; + } + + private $nfoQueryString = ''; + + /** + * Check if we should process NFO's. + * @return bool + */ + private function checkProcessNfo() + { + if ($this->site->get('lookupnfo') == 1) { + $this->nfoQueryString = \Info::NfoQueryString($this->pdo); + return ( + $this->pdo->queryOneRow( + sprintf( + 'SELECT r.ID FROM releases r WHERE 1=1 %s LIMIT 1', + $this->nfoQueryString + ) + ) === false ? false : true + ); + } + return false; + } + + private function postProcessNfoMainMethod() + { + $maxProcesses = 1; + if ($this->checkProcessNfo() === true) { + $this->processNFO = true; + $this->register_child_run([0 => $this, 1 => 'postProcessChildWorker']); + $this->work = $this->pdo->query( + sprintf(' + SELECT LEFT(r.guid, 1) AS id + FROM releases r + WHERE 1=1 %s + GROUP BY LEFT(r.guid, 1) + LIMIT 16', + $this->nfoQueryString + ) + ); + $maxProcesses = $this->site->get('nfothreads'); + } + return $maxProcesses; + } + + /** + * Check if we should process Movies. + * @return bool + */ + private function checkProcessMovies() + { + if ($this->site->get('lookupimdb') > 0) { + return ( + $this->pdo->queryOneRow( + sprintf(' + SELECT ID + FROM releases + WHERE nzbstatus = %d + AND imdbiID IS NULL + AND categoryID BETWEEN 2000 AND 2999 + %s %s + LIMIT 1', + \NZB::NZB_ADDED, + ($this->site->get('lookupimdb') == 2 ? 'AND isrenamed = 1' : ''), + ($this->ppRenamedOnly ? 'AND isrenamed = 1' : '') + ) + ) === false ? false : true + ); + } + return false; + } + + private function postProcessMovMainMethod() + { + $maxProcesses = 1; + if ($this->checkProcessMovies() === true) { + $this->processMovies = true; + $this->register_child_run([0 => $this, 1 => 'postProcessChildWorker']); + $this->work = $this->pdo->query( + sprintf(' + SELECT LEFT(guid, 1) AS id, %d AS renamed + FROM releases + WHERE nzbstatus = %d + AND imdbID IS NULL + AND categoryID BETWEEN 2000 AND 2999 + %s %s + GROUP BY LEFT(guid, 1) + LIMIT 16', + ($this->ppRenamedOnly ? 2 : 1), + \Enzebe::NZB_ADDED, + ($this->site->get('lookupimdb') == 2 ? 'AND isrenamed = 1' : ''), + ($this->ppRenamedOnly ? 'AND isrenamed = 1' : '') + ) + ); + $maxProcesses = $this->site->get('postthreadsnon'); + } + return $maxProcesses; + } + + /** + * Check if we should process TV's. + * @return bool + */ + private function checkProcessTV() + { + if ($this->site->get('lookuptvrage') > 0) { + return ( + $this->pdo->queryOneRow( + sprintf(' + SELECT ID + FROM releases + WHERE nzbstatus = %d + AND size > 1048576 + AND rageID = -1 + AND categoryID BETWEEN 5000 AND 5999 + %s %s + LIMIT 1', + \Enzebe::NZB_ADDED, + ($this->site->get('lookuptvrage') == 2 ? 'AND isrenamed = 1' : ''), + ($this->ppRenamedOnly ? 'AND isrenamed = 1' : '') + ) + ) === false ? false : true + ); + } + return false; + } + + private function postProcessTvMainMethod() + { + $maxProcesses = 1; + if ($this->checkProcessTV() === true) { + $this->processTV = true; + $this->register_child_run([0 => $this, 1 => 'postProcessChildWorker']); + $this->work = $this->pdo->query( + sprintf(' + SELECT LEFT(guid, 1) AS id, %d AS renamed + FROM releases + WHERE nzbstatus = %d + AND rageID = -1 + AND size > 1048576 + AND categoryID BETWEEN 5000 AND 5999 + %s %s + GROUP BY LEFT(guid, 1) + LIMIT 16', + ($this->ppRenamedOnly ? 2 : 1), + \Enzebe::NZB_ADDED, + ($this->site->get('lookuptvrage') == 2 ? 'AND isrenamed = 1' : ''), + ($this->ppRenamedOnly ? 'AND isrenamed = 1' : '') + ) + ); + $maxProcesses = $this->site->get('postthreadsnon'); + } + return $maxProcesses; + } + + /** + * Process sharing. + * + * @return bool + */ + private function processSharing() + { + $sharing = $this->pdo->queryOneRow('SELECT enabled FROM sharing'); + if ($sharing !== false && $sharing['enabled'] == 1) { + $nntp = new \NNTP(['Settings' => $this->pdo]); + if (($this->site->get('alternate_nntp') == 1 ? $nntp->doConnect(true, true) : $nntp->doConnect()) === true) { + (new PProcess(['Settings' => $this->pdo, 'ColorCLI' => $this->_colorCLI]))->processSharing($nntp); + } + return true; + } + return false; + } + + /** + * Process all that require a single thread. + */ + private function processSingle() + { + $postProcess = new PProcess(['Settings' => $this->pdo, 'ColorCLI' => $this->_colorCLI]); + //$postProcess->processAnime(); + $postProcess->processBooks(); + $postProcess->processConsoles(); + $postProcess->processGames(); + $postProcess->processMusic(); + $postProcess->processXXX(); + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + ////////////////////////////////////// All requestID code goes here //////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + private function requestIDMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'requestIDChildWorker']); + $this->work = $this->pdo->query( + sprintf(' + SELECT DISTINCT(g.id) + FROM groups g + INNER JOIN releases r ON r.groupID = g.ID + WHERE (g.active = 1 OR g.backfill = 1) + AND r.nzbstatus = %d + AND r.prehashID = 0 + AND r.isrequestid = 1 + AND r.reqidstatus = %d', + \Enzebe::NZB_ADDED, + \RequestID::REQID_UPROC + ) + ); + return $this->site->get('reqidthreads'); + } + + public function requestIDChildWorker($groups, $identifier = '') + { + foreach ($groups as $group) { + $this->_executeCommand( + $this->dnr_path . 'requestid ' . $group['ID'] . '"' + ); + } + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + ///////////////////////////////// All "update_per_Group" code goes here //////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + private function updatePerGroupMainMethod() + { + $this->register_child_run([0 => $this, 1 => 'updatePerGroupChildWorker']); + $this->work = $this->pdo->query('SELECT ID FROM groups WHERE (active = 1 OR backfill = 1)'); + return $this->site->get('releasesthreads'); + } + + public function updatePerGroupChildWorker($groups, $identifier = '') + { + foreach ($groups as $group) { + $this->_executeCommand( + $this->dnr_path . 'update_per_group ' . $group['id'] . '"' + ); + } + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + //////////////////////////////////////////// Various methods /////////////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + /** + * Execute a shell command, use the appropriate PHP function based on user setting. + * + * @param string $command + */ + protected function _executeCommand($command) + { + switch($this->outputType) { + case self::OUTPUT_NONE: + exec($command); + break; + case self::OUTPUT_REALTIME: + passthru($command); + break; + case self::OUTPUT_SERIALLY: + echo shell_exec($command); + break; + } + } + + /** + * Set the amount of max child processes. + * @param int $maxProcesses + */ + private function setMaxProcesses($maxProcesses) + { + // Check if override setting is on. + if (defined('NN_MULTIPROCESSING_MAX_CHILDREN_OVERRIDE') && NN_MULTIPROCESSING_MAX_CHILDREN_OVERRIDE > 0) { + $maxProcesses = NN_MULTIPROCESSING_MAX_CHILDREN_OVERRIDE; + } + + if (is_numeric($maxProcesses) && $maxProcesses > 0) { + switch ($this->workType) { + case 'postProcess_tv': + case 'postProcess_mov': + case 'postProcess_nfo': + case 'postProcess_add': + if ($maxProcesses > 16) { + $maxProcesses = 16; + } + } + $this->maxProcesses = (int)$maxProcesses; + $this->max_children_set($this->maxProcesses); + } else { + $this->max_children_set(1); + } + } + + /** + * Echo a message to CLI. + * + * @param string $message + */ + public function logger($message) + { + if (NN_ECHOCLI) { + echo $message . PHP_EOL; + } + } + + /** + * This method is executed whenever a child is finished doing work. + * + * @param string $pid The PID numbers. + * @param string $identifier Optional identifier to give a PID a name. + */ + public function childExit($pid, $identifier = '') + { + if (NN_ECHOCLI) { + $this->_colorCLI->doEcho( + $this->_colorCLI->header( + 'Process ID #' . $pid . ' has completed.' . PHP_EOL . + 'There are ' . ($this->forked_children_count - 1) . ' process(es) still active with ' . + (--$this->_workCount) . ' job(s) left in the queue.' . PHP_EOL + ) + ); + } + } + + /** + * + */ + public function __destruct() + { + parent::__destruct(); + } + + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + ////////////////////////////////////////// All class vars here ///////////////////////////////////////////////////// + //////////////////////////////////////////////////////////////////////////////////////////////////////////////////// + + /** + * Path to do not run folder. + * @var string + */ + private $dnr_path = ''; + + /** + * Work to work on. + * @var array + */ + private $work = []; + + /** + * How much work do we have to do? + * @var int + */ + public $_workCount = 0; + + /** + * The type of work we want to work on. + * @var string + */ + private $workType = ''; + + /** + * List of passed in options for the current work type. + * @var array + */ + private $workTypeOptions = []; + + /** + * Max amount of child processes to do work at a time. + * @var int + */ + private $maxProcesses = 1; + + /** + * Are we using tablePerGroup? + * @var bool + */ + private $tablePerGroup = false; + + /** + * Group used for safe backfill. + * @var string + */ + private $safeBackfillGroup = ''; + + /** + * @var \DB + */ + public $pdo; + + /** + * @var bool + */ + private $processAdditional = false; // Should we process additional? + private $processNFO = false; // Should we process NFOs? + private $processMovies = false; // Should we process Movies? + private $processTV = false; // Should we process TV? +} + +class ForkingException extends \Exception {} diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/ForkingImportNZB.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/ForkingImportNZB.php new file mode 100644 index 000000000..11e781717 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/ForkingImportNZB.php @@ -0,0 +1,90 @@ + new DB(), + ]; + $options += $defaults; + + parent::__construct($options); + $this->importPath = (PHP_BINARY . ' ' . NN_MISC . 'testing' . DS . 'nzb-import.php '); + $this->pdo = $options['settings']; + } + + public function __destruct() + { + parent::__destruct(); + } + + private $deleteComplete; + private $deleteFailed; + private $useFileName; + private $maxPerProcess; + + public function start($folder, $maxProcesses, $deleteComplete, $deleteFailed, $useFileName, $maxPerProcess) + { + $startTime = microtime(true); + $directories = glob($folder . '/*' , GLOB_ONLYDIR); + + $this->_workCount = count($directories); + + if ($this->_workCount == 0) { + echo $this->pdo->log->error('No sub-folders were found in your specified folder (' . $folder . ').'); + exit(); + } + + if (NN_ECHOCLI) { + echo $this->pdo->log->header( + 'Multi-processing started at ' . date(DATE_RFC2822) . ' with ' . $this->_workCount . + ' job(s) to do using a max of ' . $maxProcesses . ' child process(es).' + ); + } + + $this->deleteComplete = $deleteComplete; + $this->deleteFailed = $deleteFailed; + $this->useFileName = $useFileName; + $this->maxPerProcess = $maxPerProcess; + + $this->max_children_set($maxProcesses); + $this->register_child_run([0 => $this, 1 => 'importChildWorker']); + $this->child_max_run_time_set(86400); + $this->addwork($directories); + $this->process_work(true); + + if (NN_ECHOCLI) { + $this->pdo->log->doEcho( + $this->pdo->log->header( + 'Multi-processing for import finished in ' . (microtime(true) - $startTime) . + ' seconds at ' . date(DATE_RFC2822) . '.' . PHP_EOL + ) + ); + } + } + + public function importChildWorker($directories, $identifier = '') + { + foreach ($directories as $directory) { + $this->_executeCommand( + $this->importPath . '"' . + $directory . '" ' . + $this->deleteComplete . ' ' . + $this->deleteFailed . ' ' . + $this->useFileName . ' ' . + $this->maxPerProcess + ); + } + } +} \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/README.md b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/README.md new file mode 100644 index 000000000..0e4c98e30 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/README.md @@ -0,0 +1,19 @@ +####These multi-processing scripts require a POSIX compliant operating system and the PHP pcntl extension. + + +####binaries.php +This will download new headers for all active groups using your binaries threads site setting. +You can pass a argument, a number to limit the max amount of new headers to download. + + +####releases.php +This is identical to the python releases_threaded.py +This will create new releases/delete unwanted releases, process requestID's, categorize releases by group +using your release threads site setting. + + +####update_per_group.php: +This is identical to the python update_threaded.py +This will download new headers for all active groups, backfill 20k headers from all backfill enabled groups, +create new releases/delete unwanted releases, process requestID's, categorize releases, process additional and NFO +by group using your release threads site setting. \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/backfill.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/backfill.php new file mode 100644 index 000000000..46f1d9ed0 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/backfill.php @@ -0,0 +1,7 @@ +processWorkType( + 'backfill', (isset($argv[1]) && is_numeric($argv[1]) && $argv[1] > 0 ? array(0 => $argv[1]) : array(0 => false)) +); \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/binaries.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/binaries.php new file mode 100644 index 000000000..75dbe418e --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/binaries.php @@ -0,0 +1,9 @@ + (Number) Set to 0 to ignore, else fetches up to x new headers for every active group.' . PHP_EOL + ); +} +declare(ticks=1); +require('.do_not_run/require.php'); +(new \Forking())->processWorkType('binaries', array(0 => $argv[1])); \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/fixrelnames.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/fixrelnames.php new file mode 100644 index 000000000..880626791 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/fixrelnames.php @@ -0,0 +1,16 @@ + Attempt to fix release name using the nfo.' . PHP_EOL . + 'filename => Attempt to fix release name using the filenames.' . PHP_EOL . + 'md5 => Attempt to fix release name using the MD5.' . PHP_EOL . + 'par2 => Attempt to fix release name using the par2.' . PHP_EOL . + 'miscsorter => Attempt to fix release name using magic.' . PHP_EOL . + 'predbft => Attempt to fix release name using Predb full text matching.' . PHP_EOL . PHP_EOL + ); +} + +declare(ticks=1); +require('.do_not_run/require.php'); +(new \Forking())->processWorkType('fixRelNames_' . $argv[1], [0 => $argv[1]]); diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/LICENSE b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/LICENSE new file mode 100644 index 000000000..0b50498ee --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/LICENSE @@ -0,0 +1,19 @@ +Copyright (c) 2013 Barracuda Networks, Inc. + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/README.md b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/README.md new file mode 100644 index 000000000..333b64221 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/README.md @@ -0,0 +1,20 @@ +# PHP Fork Daemon +A library to make setup and management of forking daemons in PHP easy. + +## Features +- Easy management of PHP forks +- Return result of children by callback or polling parent for results +- Splitting work units into buckets +- Preforking callbacks to manage resources before forking +- Dynamic setting of number of children / work per child + +## Usage +Check out the examples in the examples directory +``php example/blocking.php`` + +## Caveats +- You need to specify ``declare(ticks=1);`` before inclusion of the fork-daemon library, otherwise signals wont be handled. This *must* be done in the main PHP file, as ``declare(ticks=N);`` only works for the file in which it is declared and the files which that file includes. Reference: [PHP Documentation](http://php.net/manual/en/control-structures.declare.php#control-structures.declare.ticks) + +## License +Copyright 2013 Barracuda Networks, Inc. +Licensed under the MIT License diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/composer.json b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/composer.json new file mode 100644 index 000000000..b6efb800a --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/composer.json @@ -0,0 +1,14 @@ +{ + "name": "barracudanetworks/forkdaemon-php", + "type": "library", + "description": "A library to make setup and management of forking daemons in PHP easy.", + "keywords": ["forking", "daemons", "php"], + "homepage": "https://github.com/barracudanetworks/forkdaemon-php", + "license": "MIT", + "require": { + "php": ">=5.3.0" + }, + "autoload": { + "files": [ "fork_daemon.php" ] + } +} \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/blocking.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/blocking.php new file mode 100644 index 000000000..8053b8a53 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/blocking.php @@ -0,0 +1,63 @@ +max_children_set(5); +$server->max_work_per_child_set(3); +$server->register_child_run("process_child_run"); +$server->register_parent_child_exit("process_child_exit"); +$server->register_logging("logger", fork_daemon::LOG_LEVEL_ALL); + +test_blocking(); + +function test_blocking() +{ + global $server; + + echo "Adding 10 units of work\n"; + + $data_set = array(); + for($i=0; $i<10; $i++) $data_set[] = $i; + shuffle($data_set); + $server->addwork($data_set); + + /* process work blocking mode */ + $server->process_work(true); + + echo "Adding 15 more units of work\n"; + + $data_set = array(); + for($i=10; $i<25; $i++) $data_set[] = $i; + shuffle($data_set); + $server->addwork($data_set); + + /* process work blocking mode */ + $server->process_work(true); +} + +/* + * CALLBACK FUNCTIONS + */ + +/* registered call back function */ +function process_child_run($data_set, $identifier = "") +{ + echo "I'm child working on: " . implode(",", $data_set) . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; + sleep(rand(4,8)); +} + +/* registered call back function */ +function process_child_exit($pid, $identifier = "") +{ + echo "Child $pid just finished" . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; +} + +/* registered call back function */ +function logger($message) +{ + echo "logger: " . $message . PHP_EOL; +} diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/bucket.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/bucket.php new file mode 100644 index 000000000..69c770a81 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/bucket.php @@ -0,0 +1,78 @@ +max_children_set(5); +$server->max_work_per_child_set(3); +$server->register_child_run("process_child_run"); +$server->register_parent_child_exit("process_child_exit"); +$server->register_logging("logger", fork_daemon::LOG_LEVEL_ALL); + +test_bucket(); + +function test_bucket() +{ + global $server; + + define("BUCKET1", 1); + define("BUCKET2", 2); + + $server->add_bucket(BUCKET1); + $server->add_bucket(BUCKET2); + $server->max_children_set(2, BUCKET1); + $server->max_children_set(5, BUCKET2); + + $data_set = array(); + for($i=0; $i<100; $i++) $data_set[] = $i; + + /* add work to bucket 1 */ + shuffle($data_set); + $server->addwork($data_set, "", BUCKET1); + + /* add work to bucket 2 */ + shuffle($data_set); + $server->addwork($data_set, "", BUCKET2); + + /* wait until all work allocated */ + while ($server->work_sets_count(BUCKET1) > 0 || $server->work_sets_count(BUCKET2) > 0) + { + echo "work set count(1): " . $server->work_sets_count(BUCKET1) . ", count(2): " . $server->work_sets_count(BUCKET2) . "\n"; + if ($server->work_sets_count(BUCKET1) > 0) $server->process_work(false, BUCKET1); + if ($server->work_sets_count(BUCKET2) > 0) $server->process_work(false, BUCKET2); + sleep(1); + } + + /* wait until all children finish */ + while ($server->children_running() > 0) + { + echo "waiting for " . $server->children_running() . " children to finish\n"; + sleep(1); + } +} + +/* + * CALLBACK FUNCTIONS + */ + +/* registered call back function */ +function process_child_run($data_set, $identifier = "") +{ + echo "I'm child working on: " . implode(",", $data_set) . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; + sleep(rand(4,8)); +} + +/* registered call back function */ +function process_child_exit($pid, $identifier = "") +{ + echo "Child $pid just finished" . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; +} + +/* registered call back function */ +function logger($message) +{ + echo "logger: " . $message . PHP_EOL; +} diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/identifier.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/identifier.php new file mode 100644 index 000000000..0805f29a6 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/identifier.php @@ -0,0 +1,81 @@ +max_children_set(5); +$server->max_work_per_child_set(3); +$server->register_child_run("process_child_run"); +$server->register_parent_child_exit("process_child_exit"); +$server->register_logging("logger", fork_daemon::LOG_LEVEL_ALL); + +test_identifier(); + +function test_identifier() +{ + global $server; + + $server->child_single_work_item_set(true); + $server->max_work_per_child_set(1); + + echo "Adding 100 units of work\n"; + + /* add work */ + $data_set = array(); + for($i=0; $i<100; $i++) $data_set[] = $i; + shuffle($data_set); + $data_set = array_chunk($data_set, 3); + + $i = 0; + foreach ($data_set as $item) + { + $server->addwork($item, "IDn$i"); + $i++; + } + + echo "Processing work in non-blocking mode\n"; + + /* process work non blocking mode */ + $server->process_work(false); + + /* wait until all work allocated */ + while ($server->work_sets_count() > 0) + { + echo "work set count: " . $server->work_sets_count() . "\n"; + $server->process_work(false); + sleep(1); + } + + /* wait until all children finish */ + while ($server->children_running() > 0) + { + echo "waiting for " . $server->children_running() . " children to finish\n"; + sleep(1); + } +} + +/* + * CALLBACK FUNCTIONS + */ + +/* registered call back function */ +function process_child_run($data_set, $identifier = "") +{ + echo "I'm child working on: " . implode(",", $data_set) . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; + sleep(rand(4,8)); +} + +/* registered call back function */ +function process_child_exit($pid, $identifier = "") +{ + echo "Child $pid just finished" . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; +} + +/* registered call back function */ +function logger($message) +{ + echo "logger: " . $message . PHP_EOL; +} diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/nonblocking.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/nonblocking.php new file mode 100644 index 000000000..cfde30f3e --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/nonblocking.php @@ -0,0 +1,71 @@ +max_children_set(5); +$server->max_work_per_child_set(3); +$server->register_child_run("process_child_run"); +$server->register_parent_child_exit("process_child_exit"); +$server->register_logging("logger", fork_daemon::LOG_LEVEL_ALL); + +test_nonblocking(); + +function test_nonblocking() +{ + global $server; + + echo "Adding 100 units of work\n"; + + /* add work */ + $data_set = array(); + for($i=0; $i<100; $i++) $data_set[] = $i; + shuffle($data_set); + $server->addwork($data_set); + + echo "Processing work in non-blocking mode\n"; + + /* process work non blocking mode */ + $server->process_work(false); + + /* wait until all work allocated */ + while ($server->work_sets_count() > 0) + { + echo "work set count: " . $server->work_sets_count() . "\n"; + $server->process_work(false); + sleep(1); + } + + /* wait until all children finish */ + while ($server->children_running() > 0) + { + echo "waiting for " . $server->children_running() . " children to finish\n"; + sleep(1); + } +} + +/* + * CALLBACK FUNCTIONS + */ + +/* registered call back function */ +function process_child_run($data_set, $identifier = "") +{ + echo "I'm child working on: " . implode(",", $data_set) . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; + sleep(rand(4,8)); +} + +/* registered call back function */ +function process_child_exit($pid, $identifier = "") +{ + echo "Child $pid just finished" . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; +} + +/* registered call back function */ +function logger($message) +{ + echo "logger: " . $message . PHP_EOL; +} diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_callback.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_callback.php new file mode 100644 index 000000000..609b25e47 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_callback.php @@ -0,0 +1,86 @@ +max_children_set(5); +$server->max_work_per_child_set(3); +$server->register_child_run("process_child_run"); +$server->register_parent_child_exit("process_child_exit"); +$server->register_logging("logger", fork_daemon::LOG_LEVEL_ALL); +$server->register_parent_results("process_results"); + +test_nonblocking(); + +function test_nonblocking() +{ + global $server; + + echo "Adding 100 units of work\n"; + + /* add work */ + $data_set = array(); + for($i=0; $i<100; $i++) $data_set[] = $i; + shuffle($data_set); + $server->addwork($data_set); + + echo "Processing work in non-blocking mode\n"; + + /* process work non blocking mode */ + $server->process_work(false); + + /* wait until all work allocated */ + while ($server->work_sets_count() > 0) + { + echo "work set count: " . $server->work_sets_count() . "\n"; + $server->process_work(false); + sleep(1); + } + + /* wait until all children finish */ + while ($server->children_running() > 0) + { + echo "waiting for " . $server->children_running() . " children to finish\n"; + sleep(1); + } +} + +/* + * CALLBACK FUNCTIONS + */ + +function process_results($results, $identifier = "") +{ + echo "Results came back: " . implode(",", $results) . ($identifier == "" ? "" : " (id:$identifier)") . PHP_EOL; +} + +/* registered call back function */ +function process_child_run($data_set, $identifier = "") +{ + echo "I'm child working on: " . implode(",", $data_set) . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; + + $result = array_sum($data_set); + sleep(rand(1,3)); + + // return results + return $result; +} + +/* registered call back function */ +function process_child_exit($pid, $identifier = "") +{ + echo "Child $pid just finished" . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; +} + +/* registered call back function */ +function logger($message) +{ + echo "logger: " . $message . PHP_EOL; +} diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_deferred.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_deferred.php new file mode 100644 index 000000000..cfc66dd67 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/examples/result_deferred.php @@ -0,0 +1,92 @@ +max_children_set(100); +$server->max_work_per_child_set(3); +$server->store_result_set(true); +$server->register_child_run("process_child_run"); +$server->register_parent_child_exit("process_child_exit"); +$server->register_logging("logger", fork_daemon::LOG_LEVEL_ALL); +// no callback with this method since we check results at the end + +test_nonblocking(); + +// since deferred results, check at the end +$results = $server->get_all_results(); +var_dump($results); +echo "Sum: " . array_sum($results) . PHP_EOL; +echo "Count: " . count($results) . PHP_EOL; + +function test_nonblocking() +{ + global $server; + + echo "Adding 100 units of work\n"; + + /* add work */ + $data_set = array(); + for($i=0; $i<100; $i++) $data_set[] = $i; + shuffle($data_set); + $server->addwork($data_set); + + echo "Processing work in non-blocking mode\n"; + echo "Sum: " . array_sum($data_set) . PHP_EOL; + + /* process work non blocking mode */ + $server->process_work(false); + + /* wait until all work allocated */ + while ($server->work_sets_count() > 0) + { + echo "work set count: " . $server->work_sets_count() . "\n"; + $server->process_work(false); + sleep(1); + } + + /* wait until all children finish */ + while ($server->children_running() > 0) + { + echo "waiting for " . $server->children_running() . " children to finish\n"; + sleep(1); + } +} + +/* + * CALLBACK FUNCTIONS + */ + +/* registered call back function */ +function process_child_run($data_set, $identifier = "") +{ + echo "I'm child working on: " . implode(",", $data_set) . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; + + sleep(rand(1,3)); + + // just do a sum and return it as the result + $result = array_sum($data_set); + + // return results + return $result; +} + +/* registered call back function */ +function process_child_exit($pid, $identifier = "") +{ + echo "Child $pid just finished" . ($identifier == "" ? "" : " (id:$identifier)") . "\n"; +} + +/* registered call back function */ +function logger($message) +{ + echo "logger: " . $message . PHP_EOL; +} diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/fork_daemon.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/fork_daemon.php new file mode 100644 index 000000000..52f50cac6 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/forkdaemon-php/fork_daemon.php @@ -0,0 +1,1855 @@ + 86400); + + /** + * Function the child invokes with a set of worker units + * @access private + * @var integer $child_function_run + */ + private $child_function_run = array(self::DEFAULT_BUCKET => ''); + + /** + * Function the parent invokes when a child finishes + * @access private + * @var integer $parent_function_child_exited + */ + private $parent_function_child_exited = array(self::DEFAULT_BUCKET => ''); + + /** + * Function the child invokes when sigint/term is received + * @access private + * @var integer $child_function_exit + */ + private $child_function_exit = array(self::DEFAULT_BUCKET => ''); + + /** + * Function the parent invokes when a child is killed due to exceeding the max runtime + * @access private + * @var integer $child_function_timeout + */ + private $child_function_timeout = array(self::DEFAULT_BUCKET => ''); + + /** + * Function the parent invokes before forking a child + * @access private + * @var integer $parent_function_prefork + */ + private $parent_function_prefork = ''; + + /** + * Function the parent invokes when a child is spawned + * @access private + * @var integer $parent_function_fork + */ + private $parent_function_fork = array(self::DEFAULT_BUCKET => ''); + + /** + * Function the parent invokes when the parent receives a SIGHUP + * @access private + * @var integer $parent_function_sighup + */ + private $parent_function_sighup = ''; + + /** + * Property of the parent sighup function. If true, the parent + * will send sighup to all children when the parent receives a + * sighup. + * @access private + * @var integer $parent_function_sighup_cascade + */ + private $parent_function_sighup_cascade = true; + + /** + * Function the child invokes when the child receives a SIGHUP + * @access private + * @var integer $child_function_sighup + */ + private $child_function_sighup = array(self::DEFAULT_BUCKET => ''); + + /** + * Function the parent invokes when a child has results to post + * @access private + * @var integer $parent_function_results + */ + private $parent_function_results = array(self::DEFAULT_BUCKET => ''); + + /** + * Stores whether results are stored for retrieval by the parent + * @access private + * @var boolean $store_result + */ + private $store_result = false; + + /** + * Max number of seconds to wait for a child process + * exit once it has been requested to exit + * @access private + * @var integer $children_kill_timeout + */ + private $children_max_timeout = 30; + + /** + * Function the parent runs when the daemon is getting shutdown + * @access private + * @var integer $parent_function_exit + */ + private $parent_function_exit = ''; + + /** + * Stores whether the daemon is in single item mode or not + * @access private + * @var bool $child_single_work_item + */ + private $child_single_work_item = array(self::DEFAULT_BUCKET => false); + + /** + * Function to call when there is a message to log + * @access private + * @var array $log_function array of callables index by severity + * called with call_user_func($log_function, $message) + */ + private $log_function = null; + + /** + * Stores whether or not we have received an exit request + * @access private + * @default false + * @var bool $exit_request_status + */ + private $exit_request_status = false; + + /**************** SERVER CONTROLS ****************/ + /** + * Upper limit on the number of children started. + * @access private + * @var integer $max_children + */ + private $max_children = array(self::DEFAULT_BUCKET => 25); + + /** + * Upper limit on the number of work units sent to each child. + * @access private + * @var integer $max_work_per_child + */ + private $max_work_per_child = array(self::DEFAULT_BUCKET => 100); + + /** + * Interval to do house keeping in seconds + * @access private + * @var integer $housekeeping_check_interval + */ + private $housekeeping_check_interval = 20; + + /**************** TRACKING CONTROLS ****************/ + + /** + * track children of parent including their status and create time + * @access private + * @var array $forked_children + */ + private $forked_children = array(); + + /** + * number of tracked children (not stopped) + * @access private + * @var array $forked_children_count + */ + protected $forked_children_count = 0; + + /** + * track the work units to process + * @access private + * @var array $work_units + */ + private $work_units = array(self::DEFAULT_BUCKET => array()); + + /** + * track the buckets + * @access private + * @var array $buckets + */ + private $buckets = array(0 => self::DEFAULT_BUCKET); + + /** + * for the parent the track the results received from chilren + * @access private + * @var array $work_units + */ + private $results = array(self::DEFAULT_BUCKET => array()); + + /** + * within a child, track the bucket the child exists in. note, + * this shouldn't be set or referenced in the parent process + * @access private + * @var int $child_bucket + */ + private $child_bucket = null; + + /**************** MOST IMPORTANT CONTROLS ****************/ + + /** + * parent pid + * @access private + * @var array $parent_pid + */ + static private $parent_pid; + + /** + * last housekeeping check time + * @access private + * @var array $housekeeping_last_check + */ + private $housekeeping_last_check = 0; + + /**************** FUNCTION DEFINITIONS ****************/ + + /** + * Set and Get functions + */ + + /** + * Allows the app to set the max_children value + * @access public + * @param int $value the new max_children value. + * @param int $bucket the bucket to use + */ + public function max_children_set($value, $bucket = self::DEFAULT_BUCKET) + { + if ($value < 1) + { + $value = 0; + $this->log(($bucket === self::DEFAULT_BUCKET ? 'default' : $bucket) . ' bucket max_children set to 0, bucket will be disabled', self::LOG_LEVEL_WARN); + } + + $this->max_children[$bucket] = $value; + } + + /** + * Allows the app to retrieve the current max_children value. + * @access public + * @param int $bucket the bucket to use + * @return int the max_children value + */ + public function max_children_get($bucket = self::DEFAULT_BUCKET) + { + return($this->max_children[$bucket]); + } + + /** + * Allows the app to set the max_work_per_child value + * @access public + * @param int $value new max_work_per_child value. + * @param int $bucket the bucket to use + */ + public function max_work_per_child_set($value, $bucket = self::DEFAULT_BUCKET) + { + if ($this->child_single_work_item[$bucket]) + { + $value = 1; + } + + if ($value < 1) + { + $value = 0; + $this->log(($bucket === self::DEFAULT_BUCKET ? 'default' : $bucket) . ' bucket max_work_per_child set to 0, bucket will be disabled', self::LOG_LEVEL_WARN); + } + + $this->max_work_per_child[$bucket] = $value; + } + + /** + * Allows the app to retrieve the current max_work_per_child value. + * @access public + * @param int $bucket the bucket to use + * @return int the max_work_per_child value + */ + public function max_work_per_child_get($bucket = self::DEFAULT_BUCKET) + { + return($this->max_work_per_child[$bucket]); + } + + /** + * Allows the app to set the child_max_run_time value + * @access public + * @param int $value new child_max_run_time value. + * @param int $bucket the bucket to use + */ + public function child_max_run_time_set($value, $bucket = self::DEFAULT_BUCKET) + { + if ($value < 1) + { + $value = 0; + $this->log(($bucket === self::DEFAULT_BUCKET ? 'default' : $bucket) . ' bucket child_max_run_time set to 0', self::LOG_LEVEL_WARN); + } + + $this->child_max_run_time[$bucket] = $value; + } + + /** + * Allows the app to retrieve the current child_max_run_time value. + * @access public + * @param int $bucket the bucket to use + * @return int the child_max_run_time value + */ + public function child_max_run_time_get($bucket = self::DEFAULT_BUCKET) + { + return($this->child_max_run_time[$bucket]); + } + + /** + * Allows the app to set the child_single_work_item value + * @access public + * @param int $value new child_single_work_item value. + * @param int $bucket the bucket to use + */ + public function child_single_work_item_set($value, $bucket = self::DEFAULT_BUCKET) + { + if ($value < 1) + { + $value = 0; + $this->log(($bucket === self::DEFAULT_BUCKET ? 'default' : $bucket) . ' bucket child_single_work_item set to 0', self::LOG_LEVEL_WARN); + } + + $this->child_single_work_item[$bucket] = $value; + } + + /** + * Allows the app to retrieve the current child_single_work_item value. + * @access public + * @param int $bucket the bucket to use + * @return int the child_single_work_item value + */ + public function child_single_work_item_get($bucket = self::DEFAULT_BUCKET) + { + return($this->child_single_work_item[$bucket]); + } + + /** + * Allows the app to set the store_result value + * @access public + * @param int $value new store_result value. + */ + public function store_result_set($value) + { + $this->store_result = $value; + } + + /** + * Allows the app to retrieve the current store_result value. + * @access public + * @return boolean the store_result value + */ + public function store_result_get() + { + return $this->store_result; + } + + /** + * Allows the app to retrieve the current child_bucket value. + * @access public + * @return int the child_bucket value representing the bucket number of the child + */ + public function child_bucket_get() + { + // this function does not apply to the parent + if (self::$parent_pid == getmypid()) return false; + + return($this->child_bucket); + } + + + /** + * Creates a new bucket to house forking operations + * @access public + * @param int $bucket the bucket to create + */ + public function add_bucket($bucket) + { + /* create the bucket by copying values from the default bucket */ + $this->max_children[$bucket] = $this->max_children[self::DEFAULT_BUCKET]; + $this->child_single_work_item[$bucket] = $this->child_single_work_item[self::DEFAULT_BUCKET]; + $this->max_work_per_child[$bucket] = $this->max_work_per_child[self::DEFAULT_BUCKET]; + $this->child_max_run_time[$bucket] = $this->child_max_run_time[self::DEFAULT_BUCKET]; + $this->child_single_work_item[$bucket] = $this->child_single_work_item[self::DEFAULT_BUCKET]; + $this->child_function_run[$bucket] = $this->child_function_run[self::DEFAULT_BUCKET]; + $this->parent_function_fork[$bucket] = $this->parent_function_fork[self::DEFAULT_BUCKET]; + $this->child_function_sighup[$bucket] = $this->child_function_sighup[self::DEFAULT_BUCKET]; + $this->child_function_exit[$bucket] = $this->child_function_exit[self::DEFAULT_BUCKET]; + $this->child_function_timeout[$bucket] = $this->child_function_timeout[self::DEFAULT_BUCKET]; + $this->parent_function_child_exited[$bucket] = $this->parent_function_child_exited[self::DEFAULT_BUCKET]; + $this->work_units[$bucket] = array(); + $this->buckets[$bucket] = $bucket; + $this->results[$bucket] = array(); + } + + /** + * Allows the app to set the call back function for child processes + * @access public + * @param string name of function to be called. + * @param int $bucket the bucket to use + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_child_run($function_name, $bucket = self::DEFAULT_BUCKET) + { + /* call child function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->child_function_run[$bucket] = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set call back functions to cleanup resources before forking + * @access public + * @param array names of functions to be called. + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_parent_prefork($function_names) + { + $this->parent_function_prefork = $function_names; + return true; + } + + /** + * Allows the app to set the call back function for when a child process is spawned + * @access public + * @param string name of function to be called. + * @param int $bucket the bucket to use + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_parent_fork($function_name, $bucket = self::DEFAULT_BUCKET) + { + /* call child function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->parent_function_fork[$bucket] = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for when a parent process receives a SIGHUP + * @access public + * @param string name of function to be called. + * @param bool $cascade_signal if true, the parent will send a sighup to all of it's children + * @param int $bucket the bucket to use + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_parent_sighup($function_name, $cascade_signal = true) + { + /* call child function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->parent_function_sighup = $function_name; + $this->parent_function_sighup_cascade = $cascade_signal; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for when a child process receives a SIGHUP + * @access public + * @param string name of function to be called. + * @param int $bucket the bucket to use + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_child_sighup($function_name, $bucket = self::DEFAULT_BUCKET) + { + /* call child function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->child_function_sighup[$bucket] = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for when a child process exits + * @access public + * @param string name of function to be called. + * @param int $bucket the bucket to use + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_child_exit($function_name, $bucket = self::DEFAULT_BUCKET) + { + /* call child function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->child_function_exit[$bucket] = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for when a child process is killed to exceeding its max runtime + * @access public + * @param string name of function to be called. + * @param int $bucket the bucket to use + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_child_timeout($function_name, $bucket = self::DEFAULT_BUCKET) + { + /* call child function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->child_function_timeout[$bucket] = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for when the parent process exits + * @access public + * @param string name of function to be called. + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_parent_exit($function_name) + { + // call parent function + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->parent_function_exit = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for when a child exits in the parent + * @access public + * @param string name of function to be called. + * @param int $bucket the bucket to use + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_parent_child_exit($function_name, $bucket = self::DEFAULT_BUCKET) + { + /* call parent function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->parent_function_child_exited[$bucket] = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for when the a child has results + * @access public + * @param string name of function to be called. + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_parent_results($function_name, $bucket = self::DEFAULT_BUCKET) + { + // call parent function + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->parent_function_results[$bucket] = $function_name; + return true; + } + + return false; + } + + /** + * Allows the app to set the call back function for logging + * @access public + * @param callable name of function to be called. + * @param int $severity the severity level + * @return bool true if the callback was successfully registered, false if it failed + */ + public function register_logging($function_name, $severity) + { + /* call parent function */ + if ( ( is_array($function_name) && method_exists($function_name[0], $function_name[1]) ) || method_exists($this, $function_name) || function_exists($function_name) ) + { + $this->log_function[$severity] = $function_name; + return true; + } + + return false; + } + + /************ NORMAL FUNCTION DEFS ************/ + + /** + * This is the class constructor, initializes the object. + * @access public + */ + public function __construct() + { + /* record pid of parent process */ + self::$parent_pid = getmypid(); + + /* install signal handlers */ + declare(ticks = 1); + pcntl_signal(SIGHUP, array(&$this, 'signal_handler_sighup')); + pcntl_signal(SIGCHLD, array(&$this, 'signal_handler_sigchild')); + pcntl_signal(SIGTERM, array(&$this, 'signal_handler_sigint')); + pcntl_signal(SIGINT, array(&$this, 'signal_handler_sigint')); + pcntl_signal(SIGALRM, SIG_IGN); + pcntl_signal(SIGUSR2, SIG_IGN); + pcntl_signal(SIGBUS, SIG_IGN); + pcntl_signal(SIGPIPE, SIG_IGN); + pcntl_signal(SIGABRT, SIG_IGN); + pcntl_signal(SIGFPE, SIG_IGN); + pcntl_signal(SIGILL, SIG_IGN); + pcntl_signal(SIGQUIT, SIG_IGN); + pcntl_signal(SIGTRAP, SIG_IGN); + pcntl_signal(SIGSYS, SIG_IGN); + + /* add barracuda specific prefork functions (doesn't hurt anything) */ + $this->parent_function_prefork = array('db_clear_connection_cache', 'memcache_clear_connection_cache'); + } + + /** + * Destructor does not do anything. + * @access public + */ + public function __destruct() + { + } + + /** + * Handle both parent and child registered sighup callbacks. + * + * @param int $signal_number is the signal that called this function. (should be '1' for SIGHUP) + * @access public + */ + public function signal_handler_sighup($signal_number) + { + if (self::$parent_pid == getmypid()) + { + // parent received sighup + $this->log('parent process [' . getmypid() . '] received sighup', self::LOG_LEVEL_DEBUG); + + // call parent's sighup registered callback + $this->invoke_callback($this->parent_function_sighup, $parameters = array(), true); + + // if cascading, send sighup to all child processes + if ($this->parent_function_sighup_cascade === true) + { + foreach ($this->forked_children as $pid => $pid_info) + { + if ($pid_info['status'] == self::STOPPED) + continue; + $this->log('parent process [' . getmypid() . '] sending sighup to child ' . $pid, self::LOG_LEVEL_DEBUG); + posix_kill($pid, SIGHUP); + } + } + } + else + { + // child received sighup. note a child is only in one bucket, do not loop through all buckets + $this->log('child process [' . getmypid() . '] received sighup with bucket type [' . $this->child_bucket . ']', self::LOG_LEVEL_DEBUG); + $this->invoke_callback( + $this->child_function_sighup[$this->child_bucket], + array($this->child_bucket), + true + ); + } + } + + /** + * Handle parent registered sigchild callbacks. + * + * @param int $signal_number is the signal that called this function. + * @access public + */ + public function signal_handler_sigchild($signal_number) + { + // do not allow signals to interrupt this + declare(ticks = 0) + { + // reap all child zombie processes + if (self::$parent_pid == getmypid()) + { + $status = ''; + + do + { + // get child pid that exited + $child_pid = pcntl_waitpid(0, $status, WNOHANG); + if ($child_pid > 0) + { + // child exited + $identifier = false; + if (!isset($this->forked_children[$child_pid])) + { + die("Cannot find $child_pid in array!\n"); + } + + $child = $this->forked_children[$child_pid]; + $identifier = $child['identifier']; + + // call exit function if and only if its declared */ + if ($child['status'] == self::WORKER) + $this->invoke_callback($this->parent_function_child_exited[ $this->forked_children[$child_pid]['bucket'] ], array($child_pid, $this->forked_children[$child_pid]['identifier']), true); + + // stop the child pid + $this->forked_children[$child_pid]['status'] = self::STOPPED; + $this->forked_children_count--; + + // respawn helper processes + if ($child['status'] == self::HELPER && $child['respawn'] === true) + { + $this->log('Helper process ' . $child_pid . ' died, respawning', self::LOG_LEVEL_INFO); + $this->helper_process_spawn($child['function'], $child['arguments'], $child['identifier'], true); + } + + // Poll for results from any children + $this->post_results($child['bucket']); + } + elseif ($child_pid < 0) + { + // ignore acceptable error 'No child processes' given we force this signal to run potentially when no children exist + if (pcntl_get_last_error() == 10) continue; + + // pcntl_wait got an error + $this->log('pcntl_waitpid failed with error ' . pcntl_get_last_error() . ':' . pcntl_strerror((pcntl_get_last_error())), self::LOG_LEVEL_DEBUG); + } + } + while ($child_pid > 0); + } + } + } + + /** + * Handle both parent and child registered sigint callbacks + * + * User terminated by CTRL-C (detected only by the parent) + * + * @param int $signal_number is the signal that called this function + * @access public + */ + public function signal_handler_sigint($signal_number) + { + // log that we received an exit request + $this->received_exit_request(true); + + // kill child processes + if (self::$parent_pid == getmypid()) + { + foreach ($this->forked_children as $pid => &$pid_info) + { + if ($pid_info['status'] == self::STOPPED) + continue; + + // tell helpers not to respawn + if ($pid_info['status'] == self::HELPER) + $pid_info['respawn'] = false; + + $this->log('requesting child exit for pid: ' . $pid, self::LOG_LEVEL_INFO); + posix_kill($pid, SIGINT); + } + + sleep(1); + + // checking for missed sigchild + $this->signal_handler_sigchild(SIGCHLD); + + $start_time = time(); + + // wait for child processes to go away + while ($this->forked_children_count > 0) + { + if (time() > ($start_time + $this->children_max_timeout)) + { + foreach ($this->forked_children as $pid => $child) + { + if ($child['status'] == self::STOPPED) + continue; + + $this->log('force killing child pid: ' . $pid, self::LOG_LEVEL_INFO); + posix_kill($pid, SIGKILL); + + // stop the child + $this->forked_children[$pid]['status'] = self::STOPPED; + $this->forked_children_count--; + } + } + else + { + $this->log('waiting ' . ($start_time + $this->children_max_timeout - time()) . ' seconds for ' . $this->forked_children_count . ' children to clean up', self::LOG_LEVEL_INFO); + sleep(1); + $this->housekeeping_check(); + } + } + + // make call back to parent exit function if it exists + $this->invoke_callback($this->parent_function_exit, $parameters = array(self::$parent_pid), true); + } + else + { + // invoke child cleanup callback + if (isset($this->child_bucket)) + $this->invoke_callback($this->child_function_exit[$this->child_bucket], $parameters = array($this->child_bucket), true); + } + + exit(-1); + } + + /** + * Check or set if we have recieved an exit request + * + * @param boolen $requested (optional) have we received the request + * @return current exit request status + */ + public function received_exit_request($requested = null) + { + // if we are retreiving the value of the exit request + if ($requested === null) + { + return $this->exit_request_status; + } + + // ensure we have good data, or set to false if not + if (! is_bool($requested)) + { + $requested = false; + } + + // set and return the ne value + return ($this->exit_request_status = $requested); + } + + /** + * Add work to the group of work to be processed + * + * @param mixed array of items to be handed back to child in chunks + * @param string a unique identifier for this work + * @param int $bucket the bucket to use + * @param bool $sort_queue true to sort the work unit queue + */ + public function addwork(array $new_work_units, $identifier = '', $bucket = self::DEFAULT_BUCKET, $sort_queue = false) + { + // ensure bucket is setup before we try to add data to it + if (! array_key_exists($bucket, $this->work_units)) + $this->add_bucket($bucket); + + // add to queue to send + if ($this->child_single_work_item[$bucket]) + { + // prepend identifier with 'id-' because array_splice() re-arranges numeric keys + $this->work_units[$bucket]['id-' . $identifier] = $new_work_units; + } + elseif ($new_work_units === null || sizeof($new_work_units) === 0) + { + // no work + } + else + { + // merge in the new work units + $this->work_units[$bucket] = array_merge($this->work_units[$bucket], $new_work_units); + } + + // sort the queue + if ($sort_queue) + ksort($this->work_units[$bucket]); + + return; + } + + /* + * Based on identifier and bucket is a child working on the work + * + * @param string unique identifier for the work + * @param int $bucket the bucket + * @return bool true if child has work, false if not + */ + public function is_work_running($identifier, $bucket = self::DEFAULT_BUCKET) + { + foreach ($this->forked_children as $info) + { + if (($info['status'] != self::STOPPED) && ($info['identifier'] == $identifier) && ($info['bucket'] == $bucket)) + { + return true; + } + } + + return false; + } + + /* + * Return array of currently running children + * + * @param int $bucket the bucket + * @return bool true if child has work, false if not + */ + public function work_running($bucket = self::DEFAULT_BUCKET) + { + $results = array(); + foreach ($this->forked_children as $pid => $child) + { + if ($child['status'] != self::STOPPED) + $results[$pid] = $child; + } + return $results; + } + + /** + * Return a list of the buckets which have been created + * + * @param bool $include_default_bucket optionally include self::DEFAULT_BUCKET in returned value (DEFAULT: true) + * @return array list of buckets + */ + public function bucket_list($include_default_bucket = true) + { + $bucket_list = array(); + + foreach($this->buckets as $bucket_id) + { + // skip the default bucket if ignored + if ( ($include_default_bucket === false) && ($bucket_id === self::DEFAULT_BUCKET) ) + continue; + + $bucket_list[] = $bucket_id; + } + + return $bucket_list; + } + + /** + * Check to see if a bucket exists + * + * @return bool true if the bucket exists, false if it does not + */ + public function bucket_exists($bucket_id) + { + return (array_key_exists($bucket_id, $this->buckets)); + } + + /** + * Return the number of work sets queued + * + * A work set is a chunk of items to be worked on. A whole work set + * is handed off to a child processes. This size of the work sets can + * be controlled by $this->max_work_per_child_set() + * + * @param int $bucket the bucket to use + * @param bool $process_all_buckets if set to true, return the count of all buckets + * @return int the number of work sets queued + */ + public function work_sets_count($bucket = self::DEFAULT_BUCKET, $process_all_buckets = false) + { + // if asked to process all buckets, count all of them and return the count + if ($process_all_buckets === true) + { + $count = 0; + foreach($this->buckets as $bucket_slot) + { + $count += count($this->work_units[$bucket_slot]); + } + return $count; + } + + return count($this->work_units[$bucket]); + } + + /** + * Return the contents of work sets queued + * + * A work set is a chunk of items to be worked on. A whole work set + * is handed off to a child processes. This size of the work sets can + * be controlled by $this->max_work_per_child_set() + * + * @param int $bucket the bucket to use + * @return array contents of the bucket + */ + public function work_sets($bucket = self::DEFAULT_BUCKET) + { + return $this->work_units[$bucket]; + } + + /** + * Return the number of children running + * + * @param int $bucket the bucket to use + * @param bool $show_pending True to show children that are done, + * but not yet had their results retrieved + * @return int the number of children running + */ + public function children_running($bucket = self::DEFAULT_BUCKET, $show_pending = false) + { + // force reaping of children + $this->signal_handler_sigchild(SIGCHLD); + + // return global count if bucket is default + if ($bucket == self::DEFAULT_BUCKET) + return ($show_pending ? count($this->forked_children) : $this->forked_children_count); + + // count within the specified bucket + $count = 0; + foreach ($this->forked_children as $child) + { + if ($show_pending) + { + if ($child['bucket'] == $bucket) + $count++; + } + else if (($child['bucket'] == $bucket) && ($child['status'] != self::STOPPED)) + { + $count++; + } + } + + return $count; + } + + /** + * Returns the number of pending child items, including running children and + * work sets that have not been allocated. Children running includes those + * that have not had their results retrieved yet. + * + * @param type $bucket The bucket to check for pending children items + * @return int Number of pending children items + */ + public function children_pending($bucket = self::DEFAULT_BUCKET) + { + return $this->children_running($bucket, true) + $this->work_sets_count($bucket); + } + + /** + * Check if the current processes is a child + * + * @return bool true if the current PID is a child PID, false otherwise + */ + static public function is_child() + { + return (isset(self::$parent_pid) ? (self::$parent_pid != getmypid()) : false); + } + + /** + * Spawns a helper process + * + * Spawns a new helper process to perform duties under the parent server + * process without accepting connections. Helper processes can optionally + * be respawned when they die. + * + * @access public + * @param string $function_name helper function to call + * @param array $arguments function arguments + * @param string $identifier helper process unique identifier + * @param bool $respawn whether to respawn the helper process when it dies + */ + public function helper_process_spawn($function_name, $arguments = array(), $idenfifier = '', $respawn = true) + { + if ((is_array($function_name) && method_exists($function_name[0], $function_name[1])) || function_exists($function_name)) + { + // init the IPC sockets + list($socket_child, $socket_parent) = $this->ipc_init(); + + // do not process signals while we are forking + declare(ticks = 0); + $pid = pcntl_fork(); + + if ($pid == -1) + { + die("Forking error!\n"); + } + elseif ($pid == 0) + { + /* + * Child process + */ + + declare(ticks = 1); + + // close our socket (we only need the one to the parent) + socket_close($socket_child); + + // execute the function + $this->log('Calling function ' . $function_name, self::LOG_LEVEL_DEBUG); + $result = call_user_func_array($function_name, $arguments); + + // send the response to the parent + self::socket_send($socket_parent, $result); + + exit(0); + } + else + { + /* + * Parent process + */ + + declare(ticks = 1); + $this->log('Spawned new helper process with pid ' . $pid, self::LOG_LEVEL_INFO); + + // close our socket (we only need the one to the child) + socket_close($socket_parent); + + // track the child + $this->forked_children[$pid] = array( + 'ctime' => time(), + 'identifier' => $idenfifier, + 'status' => self::HELPER, + 'bucket' => self::DEFAULT_BUCKET, + 'respawn' => true, + 'function' => $function_name, + 'arguments' => $arguments, + 'socket' => $socket_child, + 'last_active' => microtime(true), + ); + $this->forked_children_count++; + } + } + else + { + $this->log("Unable to spawn undefined helper function '" . $function_name . "'", self::LOG_LEVEL_CRIT); + } + } + + /** + * Forces a helper process to respawn + * + * @param string $identifier id of the helper process to respawn + */ + public function helper_process_respawn($identifier) + { + if ($identifier == '') return false; + + foreach ($this->forked_children as $pid => $child) + { + if ($child['status'] == self::HELPER && $child['identifier'] == $identifier) + { + $this->log('Forcing helper process \'' . $identifier . '\' with pid ' . $pid . ' to respawn', self::LOG_LEVEL_INFO); + posix_kill($pid, SIGKILL); + } + } + } + + /** + * Kill a specified child(ren) by pid + * + * Note: This method will block until all requested pids have exited + * + * @param int $pids the child pid to kill + * @param int $kill_delay how many seconds to wait before sending sig kill on stuck processes + * @access public + */ + public function kill_child_pid($pids, $kill_delay = 30) + { + if (! is_array($pids)) $pids = array($pids); + + // send int sigs to the children + foreach ($pids as $index => $pid) + { + // make sure we own this pid + if (! array_key_exists($pid, $this->forked_children) || $this->forked_children[$pid]['status'] == self::STOPPED) + { + $this->log('Skipping kill request on pid ' . $pid . ' because we dont own it', self::LOG_LEVEL_INFO); + unset($pids[$index]); + continue; + } + + $this->log('Asking pid ' . $pid . ' to exit via sigint', self::LOG_LEVEL_INFO); + posix_kill($pid, SIGINT); + } + + // store the requst time + $request_time = microtime(true); + $time = 0; + + // make sure the children exit + while ((count($pids) > 0) && ($time >= $kill_delay)) + { + foreach ($pids as $index => $pid) + { + // check if the pid exited gracefully + if ($this->forked_children[$pid]['status'] == self::STOPPED) + { + $this->log('Pid ' . $pid . ' has exited gracefully', self::LOG_LEVEL_INFO); + unset($pids[$index]); + continue; + } + + $time = microtime(true) - $request_time; + if ($time < $kill_delay) + { + $this->log('Waiting ' . round($time, 0) . ' seconds for ' . count($pids) . ' to exit gracefully', self::LOG_LEVEL_INFO); + sleep(1); + continue; + } + + $this->log('Force killing pid ' . $pid, self::LOG_LEVEL_INFO); + posix_kill($pid, SIGKILL); + } + } + } + + /** + * Process work on the work queue + * + * This function will take work sets and hand them off to children. + * Part of the process is calling invoking fork_work_unit to fork + * off the child. If $blocking is set to true, this function will + * process all work units and wait until the children are done until + * returning. If $blocking is set to false, this function will + * start as many work units as max_children allows and then return. + * + * Note, if $blocking is turned off, the caller has to handle when + * the children are done with their current load. + * + * @param bool true for blocking mode, false for immediate return + * @param int $bucket the bucket to use + */ + public function process_work($blocking = true, $bucket = self::DEFAULT_BUCKET, $process_all_buckets = false) + { + $this->housekeeping_check(); + + // process work on all buckets if desired + if ($process_all_buckets === true) + { + foreach($this->buckets as $bucket_slot) + { + $this->process_work($blocking, $bucket_slot, false); + } + return true; + } + + // if room fork children + if ($blocking === true) + { + // process work until completed + while ($this->work_sets_count($bucket) > 0) + { + // check to make sure we have not hit or exceded the max children (globally or within the bucket) + while ( $this->children_running($bucket) >= $this->max_children[$bucket] ) + { + $this->housekeeping_check(); + $this->signal_handler_sigchild(SIGCHLD); + sleep(1); + } + + $this->process_work_unit($bucket); + } + + // wait until work finishes + while ($this->children_running($bucket) > 0) + { + sleep(1); + $this->housekeeping_check(); + $this->signal_handler_sigchild(SIGCHLD); + } + + // make call back to parent exit function if it exists + $this->invoke_callback($this->parent_function_exit, $parameters = array(self::$parent_pid), true); + } + else + { + // fork children until max + while ( $this->children_running($bucket) < $this->max_children[$bucket] ) + { + if ($this->work_sets_count($bucket) == 0) + return true; + + $this->process_work_unit($bucket); + } + } + + return true; + } + + /** + * Returns the first result available from the bucket. This will run + * a non-blocking poll of the children for updated results. + * + * @param string $bucket The bucket to check + * @return mixed The data retrieved from a child process on the buckets + */ + public function get_result($bucket = self::DEFAULT_BUCKET) + { + // check for additional results + $this->post_results($bucket); + + if (! $this->has_result($bucket)) + return null; + + return array_shift($this->results[$bucket]); + } + + /** + * Returns all the results currently in the results queue. This will + * run a non-blocking poll of the children for updated results. + * + * @param string $bucket The bucket to retrieves results + * @return mixed Array of results from each child that has finished. + */ + public function get_all_results($bucket = self::DEFAULT_BUCKET) + { + // check for additional results + $this->post_results($bucket); + + if (! $this->has_result($bucket)) + return array(); + + $results = $this->results[$bucket]; + $this->results[$bucket] = array(); + + return $results; + } + + /** + * Checks if there is a result on the bucket. Before checking, + * runs a non-blocking poll of the children for updated results. + * + * @param string $bucket The bucket to check + * @return int Returns true if there is a result + */ + public function has_result($bucket = self::DEFAULT_BUCKET) + { + // check for additional results + $this->post_results($bucket); + + return (! empty($this->results[$bucket])); + } + + /** + * Checks if any changed child sockets are in the bucket. + * + * @param type $bucket The bucket to get results in + * @return type Returns the number of changed sockets for children workers in $bucket, + * or empty array if none. + */ + private function get_changed_sockets($bucket = self::DEFAULT_BUCKET, $timeout = 0) + { + $write_dummy = null; + $exception_dummy = null; + + // grab all the children sockets + $sockets = array(); + foreach ($this->forked_children as $pid => $child) + { + if ($child['bucket'] == $bucket) + $sockets[$pid] = $child['socket']; + } + + if (! empty($sockets)) + { + // find changed sockets and return the array of them + $result = @socket_select($sockets, $write_dummy, $exception_dummy, $timeout); + if ($result !== false && $result > 0) + return $sockets; + } + + return null; + } + + /** + * Returns any pending results from the child sockets. If a + * child has no results and it has status self::STOPPED, this will remove + * the child record from $this->forked_children. + * + * NOTE: This must be polled to check for changed sockets. + * + * @param type $blocking Set to true to block until a result comes in + * @param type $bucket The bucket to look in + * @return type The result of the child worker + */ + private function fetch_results($blocking = true, $timeout = 0, $bucket = self::DEFAULT_BUCKET) + { + $start = microtime(true); + $results = array(); + + // loop while there is pending children and pending sockets; this + // will break early on timeouts and when not blocking. + do + { + $ready_sockets = $this->get_changed_sockets($bucket, $timeout); + if (is_array($ready_sockets)) + { + foreach ($ready_sockets as $pid => $socket) + { + $result = $this->socket_receive($socket); + if ($result !== false && (! is_null($result))) + { + $this->forked_children[$pid]['last_active'] = $start; + $results[$pid] = $result; + } + } + } + + // clean up forked children that have stopped and did not have recently + // active sockets. + foreach ($this->forked_children as $pid => &$child) + { + if (isset($child['last_active']) && ($child['last_active'] < $start) && ($child['status'] == self::STOPPED)) + { + // close the socket from the parent + unset($this->forked_children[$pid]); + } + } + unset($child); + + // check if timed out + if ($timeout && (microtime(true) - $start > $timeout)) + return $results; + + // return null if not blocking and we haven't seen results + if (! $blocking) + { + return $results; + } + } + while (count($this->forked_children) > 0); + + return $results; + } + + /** + * Posts any new results to a callback function if one is available, or stores + * them to the internal results storage if not. This does not block and will + * post any results that are available, so call while children are running + * to check and post more results. + * + * NOTE: This should be polled to update results. + * + * @param type $bucket The bucket to post the results in + * @return type Returns true on successfully posting results, even if none + * to post. Returns false on error from this function or error from + * the $this->parent_function_results callback. + */ + private function post_results($bucket = self::DEFAULT_BUCKET) + { + // fetch all the results up to this point + $results = $this->fetch_results(false, 0, $bucket); + if (is_array($results) && empty($results)) + return true; + + if (! empty($this->parent_function_results[$bucket])) + { + if ($this->invoke_callback($this->parent_function_results[$bucket], array($results), true) === false) + return false; + } + elseif ($this->store_result === true) + { + $this->results[$bucket] += $results; + } + + return true; + } + + /** + * Pulls items off the work queue for processing + * + * Process the work queue by taking up to max_work_per_child items + * off the queue. A new child is then spawned off to process the + * work. + * + * @param int $bucket the bucket to use + */ + private function process_work_unit($bucket = self::DEFAULT_BUCKET) + { + $child_work_units = array_splice($this->work_units[$bucket], 0, $this->max_work_per_child[$bucket]); + + if (count($child_work_units) > 0) + { + if ($this->child_single_work_item[$bucket]) + { + // break out identifier and unit + list($child_identifier, $child_work_unit) = each($child_work_units); + + // strip preceeding 'id-' from the identifier + if (strpos($child_identifier, 'id-') === 0) + $child_identifier = substr($child_identifier, 3); + + // process work unit + $this->fork_work_unit(array($child_work_unit, $child_identifier), $child_identifier, $bucket); + } + else + { + $this->fork_work_unit(array($child_work_units), '', $bucket); + } + } + + // Poll for results from children + $this->post_results($bucket); + } + + /** + * Fork one child with one unit of work + * + * Given a work unit array, fork a child and hand + * off the work unit to the child. + * + * @param mixed $work_unit an array of work to process + * @param string a unique identifier for this work + * @param int $bucket the bucket to use + * @return mixed the child pid on success or boolean false on failure + */ + private function fork_work_unit($work_unit, $identifier = '', $bucket = self::DEFAULT_BUCKET) + { + // prefork callback + foreach ($this->parent_function_prefork as $function) + { + $this->invoke_callback($function, array(), true); + } + + // init the IPC sockets + list($socket_child, $socket_parent) = $this->ipc_init(); + + // turn off signals temporarily to prevent a SIGCHLD from interupting the parent before $this->forked_children is updated + declare(ticks = 0); + + // spoon! + $pid = pcntl_fork(); + + if ($pid == -1) + { + /** + * Fork Error + */ + + $this->log('failed to fork', self::LOG_LEVEL_CRIT); + return false; + } + elseif ($pid) + { + /** + * Parent Process + */ + + // keep track of this pid in the parent + $this->forked_children[$pid] = array( + 'ctime' => time(), + 'identifier' => $identifier, + 'bucket' => $bucket, + 'status' => self::WORKER, + 'socket' => $socket_child, + 'last_active' => microtime(true), + ); + $this->forked_children_count++; + + // turn back on signals now that $this->forked_children has been updated + declare(ticks = 1); + + // close our socket (we only need the one to the child) + socket_close($socket_parent); + + // debug logging + $this->log('forking child ' . $pid . ' for bucket ' . $bucket, self::LOG_LEVEL_DEBUG); + + // parent spawned child callback + $this->invoke_callback($this->parent_function_fork[$bucket], $parameters = array($pid, $identifier), true); + } + else + { + /** + * Child Process + */ + + // free up unneeded parent memory for child process + $this->work_units = null; + $this->forked_children = null; + $this->results = null; + + // set child properties + $this->child_bucket = $bucket; + + // turn signals on for the child + declare(ticks = 1); + + // close our socket (we only need the one to the parent) + socket_close($socket_child); + + // re-seed the random generator to prevent clone from parent + srand(); + + // child run callback + $result = $this->invoke_callback($this->child_function_run[$bucket], $work_unit, false); + + // send the result to the parent + self::socket_send($socket_parent, $result); + + // delay the child's exit slightly to avoid race conditions + usleep(500); + + // exit after we complete one unit of work + exit; + } + + return $pid; + } + + /** + * Performs house keeping every housekeeping_check_interval seconds + * @access private + */ + private function housekeeping_check() + { + if ((time() - $this->housekeeping_last_check) >= $this->housekeeping_check_interval) + { + // check to make sure no children are violating the max run time + $this->kill_maxtime_violators(); + + // look for zombie children just in case + $this->signal_handler_sigchild(SIGCHLD); + + // update the last check time to now + $this->housekeeping_last_check = time(); + } + } + + /** + * Kills any children that have been running for too long. + * @access private + */ + private function kill_maxtime_violators() + { + foreach ($this->forked_children as $pid => $pid_info) + { + if ($pid_info['status'] == self::STOPPED) + continue; + + if ((time() - $pid_info['ctime']) > $this->child_max_run_time[$pid_info['bucket']]) + { + $this->log('Force kill ' . $pid . ' has run too long', self::LOG_LEVEL_INFO); + + // notify app that child process timed out + $this->invoke_callback($this->child_function_timeout{$pid_info['bucket']}, array($pid, $pid_info['identifier']), true); + + posix_kill($pid, SIGKILL); // its probably stuck on something, kill it immediately. + sleep(3); // give the child time to die + + // force signal handling + $this->signal_handler_sigchild(SIGCHLD); + } + } + } + + /** + * Invoke a call back function with parameters + * + * Given the name of a function and parameters to send it, invoke the function. + * This function will try using the objects inherited function if it exists. If not, + * it'll look for a declared function of the given name. + * + * @access private + * @param string $function_name the name of the function to invoke + * @param array $parameters an array of parameters to pass to function + * @param bool $optional is set to true, don't error if function_name not available + * @return mixed false on error, otherwise return of callback function + */ + private function invoke_callback($function_name, $parameters, $optional = false) + { + // call child function + if (is_array($function_name) && method_exists($function_name[0], $function_name[1])) + { + if (!is_array($parameters)) $parameters = array($parameters); + return call_user_func_array($function_name, $parameters); + } + elseif (method_exists($this, $function_name) ) + { + if (!is_array($parameters)) $parameters = array($parameters); + return call_user_func_array($this->$function_name, $parameters); + } + else if (function_exists($function_name)) + { + if (!is_array($parameters)) $parameters = array($parameters); + return call_user_func_array($function_name, $parameters); + } + else + { + if ($optional === false) + $this->log("Error there are no functions declared in scope to handle callback for function '" . $function_name . "'", self::LOG_LEVEL_CRIT); + } + } + + /** + * Initialize interprocess communication by setting up a pair + * of sockets and returning them as an array. + * + * @return type + */ + private function ipc_init() + { + // windows needs AF_INET + $domain = strtoupper(substr(PHP_OS, 0, 3)) == 'WIN' ? AF_INET : AF_UNIX; + + // create a socket pair for IPC + $sockets = array(); + if (socket_create_pair($domain, SOCK_STREAM, 0, $sockets) === false) + { + $this->log('socket_create_pair failed: ' . socket_strerror(socket_last_error()), self::LOG_LEVEL_CRIT); + return false; + } + + // return the sockets + return $sockets; + } + + /** + * Sends a serializable message to the socket. + * + * @param type $socket The socket to send the message on + * @param type $message The serializable message to send + * @return type Returns true on success, false on failure + */ + private function socket_send($socket, $message) + { + $serialized_message = @serialize($message); + if ($serialized_message == false) + { + $this->log('socket_send failed to serialize message', self::LOG_LEVEL_CRIT); + return false; + } + + $header = pack('N', strlen($serialized_message)); + $data = $header . $serialized_message; + $bytes_left = strlen($data); + while ($bytes_left > 0) + { + $bytes_sent = @socket_write($socket, $data); + if ($bytes_sent === false) + { + $this->log('socket_send error: ' . socket_strerror(socket_last_error()), self::LOG_LEVEL_CRIT); + return false; + } + + $bytes_left -= $bytes_sent; + $data = substr($data, $bytes_sent); + } + + return true; + } + + /** + * Receives a serialized message from the socket. + * + * @param type $socket Thes socket to receive the message from + * @return type Returns true on success, false on failure + */ + private function socket_receive($socket) + { + // initially read to the length of the header size, then + // expand to read more + $bytes_total = self::SOCKET_HEADER_SIZE; + $bytes_read = 0; + $have_header = false; + $socket_message = ''; + while ($bytes_read < $bytes_total) + { + $read = @socket_read($socket, $bytes_total - $bytes_read); + if ($read === false) + { + $this->log('socket_receive error: ' . socket_strerror(socket_last_error()), self::LOG_LEVEL_CRIT); + return false; + } + + // blank socket_read means done + if ($read == '') + break; + + $bytes_read += strlen($read); + $socket_message .= $read; + + if (!$have_header && $bytes_read >= self::SOCKET_HEADER_SIZE) + { + $have_header = true; + list($bytes_total) = array_values(unpack('N', $socket_message)); + $bytes_read = 0; + $socket_message = ''; + } + } + + $message = @unserialize($socket_message); + + return $message; + } + + /** + * Log a message + * + * @access private + * @param string $message the text to log + * @param int $severity the severity of the message + * @param bool true on success, false on error + */ + private function log($message, $severity) + { + if (!empty($this->log_function)) + { + if (isset($this->log_function[$severity])) + { + return call_user_func($this->log_function[$severity], $message); + } + elseif (isset($this->log_function[self::LOG_LEVEL_ALL])) + { + return call_user_func($this->log_function[self::LOG_LEVEL_ALL], $message); + } + } + // Barracuda specific logging class, to keep internal code working + elseif (method_exists('Log', 'message')) + { + return Log::message($message, $severity); + } + + return true; + } +} \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/import.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/import.php new file mode 100644 index 000000000..1947a6d8c --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/import.php @@ -0,0 +1,32 @@ + Delete the NZB files after they are imported (recommended), if you stop and restart you will have to go over the imported files again.' . PHP_EOL . PHP_EOL . + 'Fourth argument (optional)' . PHP_EOL . + 'true|false => Delete the NZB if importing it fails (not recommended).' . PHP_EOL . + 'Fifth argument (optional):' . PHP_EOL . + 'true|false => Use the NZB file name as the release name (not recommended), the names in the NZB are better.' . PHP_EOL . PHP_EOL . + 'Sixth argument (optional):' . PHP_EOL . + 'How many NZB files to import per process, if this is not set, it will do 50,000 per process.' . PHP_EOL . PHP_EOL . + 'Note that successfully imported NZB files WILL be deleted.' . PHP_EOL + + ); +} +(new \ForkingImportNZB())->start( + $argv[1], + (isset($argv[2]) && is_numeric($argv[2]) && $argv[2] > 0 ? $argv[2] : 1), + (isset($argv[3]) && $argv[3] === 'true' ? 'true' : 'false'), + (isset($argv[4]) && $argv[4] === 'true' ? 'true' : 'false'), + (isset($argv[5]) && $argv[5] === 'true' ? 'true' : 'false'), + (isset($argv[6]) && is_numeric($argv[6]) && $argv[6] > 0 ? $argv[6] : 50000) +); diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/postprocess.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/postprocess.php new file mode 100644 index 000000000..b9fe1f74f --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/postprocess.php @@ -0,0 +1,18 @@ + Do amazon processing, this does not use multi-processing, because of amazon API restrictions.' . PHP_EOL . + 'add => Do additional (rar|zip) processing.' . PHP_EOL . + 'mov => Do movie processing.' . PHP_EOL . + 'nfo => Do NFO processing.' . PHP_EOL . + 'sha => Do sharing processing, this does not using multi-processing.' . PHP_EOL . + 'tv => Do TV processing.' . PHP_EOL . PHP_EOL . + 'Second argument (optional):' . PHP_EOL . + 'true|false => Only post-process renamed releases. This is for the mov|tv options.' . PHP_EOL + ); +} + +declare(ticks=1); +require('.do_not_run/require.php'); +(new \Forking())->processWorkType('postProcess_' . $argv[1], (isset($argv[2]) && $argv[2] === 'true' ? [0 => true] : [])); \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/releases.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/releases.php new file mode 100644 index 000000000..fde1f1826 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/releases.php @@ -0,0 +1,4 @@ +processWorkType('releases'); \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/requestid.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/requestid.php new file mode 100644 index 000000000..3f1fd24d2 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/requestid.php @@ -0,0 +1,4 @@ +processWorkType('request_id'); \ No newline at end of file diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/safe.php b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/safe.php new file mode 100644 index 000000000..858416f52 --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/safe.php @@ -0,0 +1,12 @@ + Do Safe Binaries update.' . PHP_EOL . + 'backfill => Do Safe Backfill update.' . PHP_EOL + ); +} + +declare(ticks=1); +require('.do_not_run/require.php'); +(new \Forking())->processWorkType('safe_' . $argv[1]); diff --git a/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/settings.php.example b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/settings.php.example new file mode 100644 index 000000000..f1c86771b --- /dev/null +++ b/lib/copy_this/misc/update_scripts/nix_scripts/multiprocessing/settings.php.example @@ -0,0 +1,49 @@ +processWorkType('update_per_group'); \ No newline at end of file diff --git a/lib/copy_this/www/automated.config.php b/lib/copy_this/www/automated.config.php index 9a0998a41..2e4987dc5 100644 --- a/lib/copy_this/www/automated.config.php +++ b/lib/copy_this/www/automated.config.php @@ -16,8 +16,20 @@ define('NN_CORE', NN_LIB); // Refers to the web root for the Smarty lib define('NN_WWW', NN_ROOT); +//Refers to the misc folder +define('NN_MISC', NN_WWW . '..' . DS . 'misc' . DS); + +//Refers to update_scripts folder +define('NN_UPDATE', NN_MISC . 'update_scripts' . DS); + +//Refers to nix_scripts folder +define('NN_NIX', NN_UPDATE . 'nix_scripts' . DS ); + +//Refers to multiprocessing folder +define('NN_MULTI', NN_NIX . 'multiprocessing' . DS); + //refers to tmux folder -define('NN_TMUX', NN_ROOT . '..' . DS . 'misc' . DS . 'update_scripts' . DS . 'nix_scripts' . DS . 'tmux' .DS); +define('NN_TMUX', NN_NIX . 'tmux' . DS); // Refers to the covers folder define('NN_COVERS', NN_WWW . 'covers' .DS); diff --git a/lib/copy_this/www/lib/RequestID.php b/lib/copy_this/www/lib/RequestID.php index 574668db7..fe9ad9bc9 100644 --- a/lib/copy_this/www/lib/RequestID.php +++ b/lib/copy_this/www/lib/RequestID.php @@ -1,9 +1,9 @@ isLoggedIn()) { $page->show403(); } diff --git a/lib/namefixer.php b/lib/namefixer.php index f54c73fd6..8b86fc64f 100644 --- a/lib/namefixer.php +++ b/lib/namefixer.php @@ -9,7 +9,7 @@ require_once(WWW_DIR . "lib/nntp.php"); require_once(WWW_DIR . "/lib/ReleaseSearch.php"); require_once(WWW_DIR . "/lib/ColorCLI.php"); require_once(WWW_DIR . "/lib/ConsoleTools.php"); -require_once NN_LIB . 'SphinxSearch.php'; +require_once (NN_LIB . 'SphinxSearch.php'); require_once("ReleaseCleaner.php"); require_once("functions.php"); require_once("nzbcontents.php"); diff --git a/lib/requestid.php b/lib/requestid.php index ab5c3a1ec..217f5dee0 100644 --- a/lib/requestid.php +++ b/lib/requestid.php @@ -1,8 +1,8 @@