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

matomo / app / core / CronArchive / SharedSiteIds.php

matomo / app / core / CronArchive Last commit date

Performance 6 years ago FixedSiteIds.php 6 years ago SegmentArchivingRequestUrlProvider.php 6 years ago SharedSiteIds.php 6 years ago SitesToReprocessDistributedList.php 6 years ago StopArchiverException.php 6 years ago

SharedSiteIds.php

203 lines

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