SharedSiteIds.php — Matomo Analytics – Powerful, Privacy-First Insights for WordPress 5.0.3

matomo / app / core / CronArchive / SharedSiteIds.php

matomo / app / core / CronArchive Last commit date

Performance 2 years ago ArchiveFilter.php 2 years ago FixedSiteIds.php 2 years ago QueueConsumer.php 2 years ago ReArchiveList.php 2 years ago SegmentArchiving.php 2 years ago SharedSiteIds.php 2 years ago StopArchiverException.php 2 years ago

SharedSiteIds.php

187 lines

1	<?php
2
3	/**
4	* Matomo - free/libre analytics platform
5	*
6	* @link https://matomo.org
7	* @license http://www.gnu.org/licenses/gpl-3.0.html GPL v3 or later
8	*
9	*/
10	namespace Piwik\CronArchive;
11
12	use Exception;
13	use Piwik\CliMulti\Process;
14	use Piwik\Log;
15	use Piwik\Option;
16	/**
17	* This class saves all to be processed siteIds in an Option named 'SharedSiteIdsToArchive' and processes all sites
18	* within that list. If a user starts multiple archiver those archiver will help to finish processing that list.
19	*/
20	class SharedSiteIds
21	{
22	const OPTION_DEFAULT = 'SharedSiteIdsToArchive';
23	const OPTION_ALL_WEBSITES = 'SharedSiteIdsToArchive_AllWebsites';
24	const KEY_TIMESTAMP = '_ResetQueueTime';
25	/**
26	* @var string
27	*/
28	private $optionName;
29	private $siteIds = array();
30	private $currentSiteId;
31	private $done = false;
32	private $initialResetQueueTime = null;
33	private $isContinuingPreviousRun = false;
34	public function __construct($websiteIds, $optionName = self::OPTION_DEFAULT)
35	{
36	$this->optionName = $optionName;
37	if (empty($websiteIds)) {
38	$websiteIds = array();
39	}
40	$self = $this;
41	$this->siteIds = $this->runExclusive(function () use($self, $websiteIds) {
42	// if there are already sites to be archived registered, prefer the list of existing archive, meaning help
43	// to finish this queue of sites instead of starting a new queue
44	$existingWebsiteIds = $self->getAllSiteIdsToArchive();
45	if (!empty($existingWebsiteIds)) {
46	$this->isContinuingPreviousRun = true;
47	return $existingWebsiteIds;
48	}
49	$self->setQueueWasReset();
50	$self->setSiteIdsToArchive($websiteIds);
51	return $websiteIds;
52	});
53	$this->initialResetQueueTime = $this->getResetQueueTime();
54	}
55	public function setQueueWasReset()
56	{
57	Option::set($this->optionName . self::KEY_TIMESTAMP, floor(microtime(true) * 1000));
58	}
59	private function getResetQueueTime()
60	{
61	Option::clearCachedOption($this->optionName . self::KEY_TIMESTAMP);
62	return (int) Option::get($this->optionName . self::KEY_TIMESTAMP);
63	}
64	public function getInitialSiteIds()
65	{
66	return $this->siteIds;
67	}
68	/**
69	* Get the number of total websites that needs to be processed.
70	*
71	* @return int
72	*/
73	public function getNumSites()
74	{
75	return count($this->siteIds);
76	}
77	/**
78	* Get the number of already processed websites (not necessarily all of those where processed by this archiver).
79	*
80	* @return int
81	*/
82	public function getNumProcessedWebsites()
83	{
84	if ($this->done) {
85	return $this->getNumSites();
86	}
87	if (empty($this->currentSiteId)) {
88	return 0;
89	}
90	$index = array_search($this->currentSiteId, $this->siteIds);
91	if (false === $index) {
92	return 0;
93	}
94	return $index + 1;
95	}
96	public function setSiteIdsToArchive($siteIds)
97	{
98	if (!empty($siteIds)) {
99	Option::set($this->optionName, implode(',', $siteIds));
100	} else {
101	Option::delete($this->optionName);
102	}
103	}
104	public function getAllSiteIdsToArchive()
105	{
106	Option::clearCachedOption($this->optionName);
107	$siteIdsToArchive = Option::get($this->optionName);
108	if (empty($siteIdsToArchive)) {
109	return array();
110	}
111	return explode(',', trim($siteIdsToArchive));
112	}
113	/**
114	* If there are multiple archiver running on the same node it makes sure only one of them performs an action and it
115	* will wait until another one has finished. Any closure you pass here should be very fast as other processes wait
116	* for this closure to finish otherwise. Currently only used for making multiple archivers at the same time work.
117	* If a closure takes more than 5 seconds we assume it is dead and simply continue.
118	*
119	* @param \Closure $closure
120	* @return mixed
121	* @throws \Exception
122	*/
123	private function runExclusive($closure)
124	{
125	$process = new Process('archive.sharedsiteids');
126	while ($process->isRunning() && $process->getSecondsSinceCreation() < 5) {
127	// wait max 5 seconds, such an operation should not take longer
128	usleep(25 * 1000);
129	}
130	$process->startProcess();
131	try {
132	$result = $closure();
133	} catch (Exception $e) {
134	$process->finishProcess();
135	throw $e;
136	}
137	$process->finishProcess();
138	return $result;
139	}
140	/**
141	* Get the next site id that needs to be processed or null if all site ids where processed.
142	*
143	* @return int\|null
144	*/
145	public function getNextSiteId()
146	{
147	if ($this->done) {
148	// we make sure we don't check again whether there are more sites to be archived as the list of
149	// sharedSiteIds may have been reset by now.
150	return null;
151	}
152	if ($this->initialResetQueueTime !== $this->getResetQueueTime()) {
153	// queue was reset/finished by some other process
154	$this->currentSiteId = null;
155	$this->done = true;
156	Log::debug('The shared site ID queue was reset, stopping.');
157	return null;
158	}
159	$self = $this;
160	$this->currentSiteId = $this->runExclusive(function () use($self) {
161	$siteIds = $self->getAllSiteIdsToArchive();
162	if (empty($siteIds)) {
163	// done... no sites left to be archived
164	return null;
165	}
166	$nextSiteId = array_shift($siteIds);
167	$self->setSiteIdsToArchive($siteIds);
168	return $nextSiteId;
169	});
170	if (is_null($this->currentSiteId)) {
171	$this->done = true;
172	}
173	return $this->currentSiteId;
174	}
175	public static function isSupported()
176	{
177	return Process::isSupported();
178	}
179	/**
180	* @return bool
181	*/
182	public function isContinuingPreviousRun() : bool
183	{
184	return $this->isContinuingPreviousRun;
185	}
186	}
187