- if($contact['subhub']) {
- $poll_interval = get_config('system','pushpoll_frequency');
- $contact['priority'] = (($poll_interval !== false) ? intval($poll_interval) : 3);
- $hub_update = false;
-
- if((datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 day")) || $force)
- $hub_update = true;
- }
- else
- $hub_update = false;
-
- /**
- * Based on $contact['priority'], should we poll this site now? Or later?
- */
-
- switch ($contact['priority']) {
- case 5:
- if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 month"))
- $update = true;
- break;
- case 4:
- if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 week"))
- $update = true;
- break;
- case 3:
- if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 day"))
- $update = true;
- break;
- case 2:
- if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 12 hour"))
- $update = true;
- break;
- case 1:
- default:
- if(datetime_convert('UTC','UTC', 'now') > datetime_convert('UTC','UTC', $t . " + 1 hour"))
- $update = true;
- break;
- }
- if((! $update) && (! $force))
- continue;
- }
+ return($active >= $queues);
+}
+
+/**
+ * @brief Returns the number of active poller processes
+ *
+ * @return integer Number of active poller processes
+ */
+function poller_active_workers() {
+ $workers = q("SELECT COUNT(*) AS `processes` FROM `process` WHERE `command` = 'poller.php'");
+
+ return($workers[0]["processes"]);
+}
+
+/**
+ * @brief Check if we should pass some slow processes
+ *
+ * When the active processes of the highest priority are using more than 2/3
+ * of all processes, we let pass slower processes.
+ *
+ * @param string $highest_priority Returns the currently highest priority
+ * @return bool We let pass a slower process than $highest_priority
+ */
+function poller_passing_slow(&$highest_priority) {
+
+ $highest_priority = 0;
+
+ $r = q("SELECT `priority`
+ FROM `process`
+ INNER JOIN `workerqueue` ON `workerqueue`.`pid` = `process`.`pid`");
+
+ // No active processes at all? Fine
+ if (!dbm::is_result($r))
+ return(false);
+
+ $priorities = array();
+ foreach ($r AS $line)
+ $priorities[] = $line["priority"];
+
+ // Should not happen
+ if (count($priorities) == 0)
+ return(false);
+
+ $highest_priority = min($priorities);
+
+ // The highest process is already the slowest one?
+ // Then we quit
+ if ($highest_priority == PRIORITY_NEGLIGIBLE)
+ return(false);
+
+ $high = 0;
+ foreach ($priorities AS $priority)
+ if ($priority == $highest_priority)
+ ++$high;
+
+ logger("Highest priority: ".$highest_priority." Total processes: ".count($priorities)." Count high priority processes: ".$high, LOGGER_DEBUG);
+ $passing_slow = (($high/count($priorities)) > (2/3));
+
+ if ($passing_slow)
+ logger("Passing slower processes than priority ".$highest_priority, LOGGER_DEBUG);
+
+ return($passing_slow);
+}
+
+/**
+ * @brief Returns the next worker process
+ *
+ * @return string SQL statement
+ */
+function poller_worker_process() {
+
+ q("START TRANSACTION;");
+
+ // Check if we should pass some low priority process
+ $highest_priority = 0;
+
+ if (poller_passing_slow($highest_priority)) {
+ // Are there waiting processes with a higher priority than the currently highest?
+ $r = q("SELECT * FROM `workerqueue`
+ WHERE `executed` <= '%s' AND `priority` < %d
+ ORDER BY `priority`, `created` LIMIT 1",
+ dbesc(NULL_DATE),
+ intval($highest_priority));
+ if (dbm::is_result($r)) {
+ return $r;
+ }
+ // Give slower processes some processing time
+ $r = q("SELECT * FROM `workerqueue`
+ WHERE `executed` <= '%s' AND `priority` > %d
+ ORDER BY `priority`, `created` LIMIT 1",
+ dbesc(NULL_DATE),
+ intval($highest_priority));
+ }
+
+ // If there is no result (or we shouldn't pass lower processes) we check without priority limit
+ if (($highest_priority == 0) OR !dbm::is_result($r)) {
+ $r = q("SELECT * FROM `workerqueue` WHERE `executed` <= '%s' ORDER BY `priority`, `created` LIMIT 1", dbesc(NULL_DATE));
+ }
+ return $r;
+}