blob: 50f95ea2dd294af9df17a7be1b913643b71756fc (
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
|
<?php
/**
* Piwik - free/libre analytics platform
*
* @link http://piwik.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
{
private $siteIds = array();
private $currentSiteId;
private $done = false;
public function __construct($websiteIds)
{
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;
});
}
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('SharedSiteIdsToArchive', implode(',', $siteIds));
} else {
Option::delete('SharedSiteIdsToArchive');
}
}
public function getAllSiteIdsToArchive()
{
Option::clearCachedOption('SharedSiteIdsToArchive');
$siteIdsToArchive = Option::get('SharedSiteIdsToArchive');
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()
{
$self = $this;
$this->currentSiteId = $this->runExclusive(function () use ($self) {
$siteIds = $self->getAllSiteIdsToArchive();
if (empty($siteIds)) {
return null;
}
$nextSiteId = array_shift($siteIds);
$self->setSiteIdsToArchive($siteIds);
return $nextSiteId;
});
if (is_null($this->currentSiteId)) {
$this->done = true;
}
return $this->currentSiteId;
}
public static function isSupported()
{
return Process::isSupported();
}
}
|
