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