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

matomo / app / core / CronArchive / SharedSiteIds.php

matomo / app / core / CronArchive Last commit date

Performance 1 year ago ArchiveFilter.php 3 months ago FixedSiteIds.php 3 months ago QueueConsumer.php 7 months ago ReArchiveList.php 1 year ago SegmentArchiving.php 7 months ago SharedSiteIds.php 3 months ago StopArchiverException.php 2 years ago

SharedSiteIds.php

183 lines

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