File: Scheduler.php

package info (click to toggle)
matomo 5.8.0-1
  • links: PTS, VCS
  • area: main
  • in suites: sid
  • size: 95,068 kB
  • sloc: php: 289,425; xml: 127,249; javascript: 112,130; python: 202; sh: 178; makefile: 20; sql: 10
file content (399 lines) | stat: -rw-r--r-- 13,489 bytes parent folder | download | duplicates (2)
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
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
<?php

/**
 * Matomo - free/libre analytics platform
 *
 * @link    https://matomo.org
 * @license https://www.gnu.org/licenses/gpl-3.0.html GPL v3 or later
 */

namespace Piwik\Scheduler;

use Piwik\Concurrency\Lock;
use Piwik\Piwik;
use Piwik\Timer;
use Piwik\Log\LoggerInterface;

/**
 * Schedules task execution.
 *
 * A scheduled task is a callback that should be executed every so often (such as daily,
 * weekly, monthly, etc.). They are registered by extending {@link \Piwik\Plugin\Tasks}.
 *
 * Tasks are executed when the `core:archive` command is executed.
 *
 * ### Examples
 *
 * **Scheduling a task**
 *
 *     class Tasks extends \Piwik\Plugin\Tasks
 *     {
 *         public function schedule()
 *         {
 *             $this->hourly('myTask');  // myTask() will be executed once every hour
 *         }
 *         public function myTask()
 *         {
 *             // do something
 *         }
 *     }
 *
 * **Executing all pending tasks**
 *
 *     $results = $scheduler->run();
 *     $task1Result = $results[0];
 *     $task1Name = $task1Result['task'];
 *     $task1Output = $task1Result['output'];
 *
 *     echo "Executed task '$task1Name'. Task output:\n$task1Output";
 */
class Scheduler
{
    /**
     * Is the scheduler running any task.
     * @var bool
     */
    private $isRunningTask = false;

    /**
     * Should the last run task be scheduled for a retry
     * @var bool
     */
    private $scheduleRetry = false;

    /**
     * @var Timetable
     */
    private $timetable;

    /**
     * @var TaskLoader
     */
    private $loader;

    /**
     * @var LoggerInterface
     */
    private $logger;

    /**
     * @var Lock
     */
    private $lock;

    /**
     * @var bool
     */
    private $hasReceivedAbortSignal = false;

    public function __construct(TaskLoader $loader, LoggerInterface $logger, ScheduledTaskLock $lock)
    {
        $this->timetable = new Timetable();
        $this->loader = $loader;
        $this->logger = $logger;
        $this->lock = $lock;
    }

    public function handleSignal(int $signal): void
    {
        $this->hasReceivedAbortSignal = in_array($signal, [\SIGINT, \SIGTERM], true);
    }

    /**
     * Executes tasks that are scheduled to run, then reschedules them.
     *
     * @return array An array describing the results of scheduled task execution. Each element
     *               in the array will have the following format:
     *
     *               ```
     *               array(
     *                   'task' => 'task name',
     *                   'output' => '... task output ...'
     *               )
     *               ```
     */
    public function run()
    {
        $tasks = $this->loader->loadTasks();

        $this->logger->debug('{count} scheduled tasks loaded', array('count' => count($tasks)));

        // remove from timetable tasks that are not active anymore
        $this->timetable->removeInactiveTasks($tasks);

        $this->logger->info("Starting Scheduled tasks... ");

        // for every priority level, starting with the highest and concluding with the lowest
        $executionResults = array();
        $readFromOption = true;
        for ($priority = Task::HIGHEST_PRIORITY; $priority <= Task::LOWEST_PRIORITY; ++$priority) {
            $this->logger->debug("Executing tasks with priority {priority}:", array('priority' => $priority));

            // loop through each task
            foreach ($tasks as $task) {
                if ($this->hasReceivedAbortSignal) {
                    $this->logger->info("Scheduler: Aborting due to received signal");
                    return $executionResults;
                }

                // if the task does not have the current priority level, don't execute it yet
                if ($task->getPriority() != $priority) {
                    continue;
                }

                $taskName = $task->getName();

                if (!$this->acquireLockForTask($taskName, $task->getTTL())) {
                    $this->logger->debug(
                        "Scheduler: '{task}' is currently executed by another process",
                        ['task' => $task->getName()]
                    );
                    continue;
                }

                if ($readFromOption) {
                    // because other jobs might execute the scheduled tasks as well we have to read the up to date time table to not handle the same task twice
                    // ideally we would read from option every time but using $readFromOption as a minor performance tweak. There can be easily 100 tasks
                    // of which we only execute very few and it's unlikely that the timetable changes too much in between while iterating over the loop and triggering the event.
                    // this way we only read from option when we actually execute or reschedule a task as this can take a few seconds.
                    $this->timetable->readFromOption();
                    $readFromOption = false;
                }

                $shouldExecuteTask = $this->timetable->shouldExecuteTask($taskName);

                if ($this->timetable->taskShouldBeRescheduled($taskName)) {
                    $readFromOption = true;
                    $rescheduledDate = $this->timetable->rescheduleTask($task);

                    $this->logger->debug("Task {task} is scheduled to run again for {date}.", array('task' => $taskName, 'date' => $rescheduledDate));
                }

                /**
                 * Triggered before a task is executed.
                 *
                 * A plugin can listen to it and modify whether a specific task should be executed or not. This way
                 * you can force certain tasks to be executed more often or for example to be never executed.
                 *
                 * @param bool &$shouldExecuteTask Decides whether the task will be executed.
                 * @param Task $task The task that is about to be executed.
                 */
                Piwik::postEvent('ScheduledTasks.shouldExecuteTask', array(&$shouldExecuteTask, $task));

                if ($shouldExecuteTask) {
                    $readFromOption = true;
                    $this->scheduleRetry = false;
                    $message = $this->executeTask($task);

                    // Task has thrown an exception and should be scheduled for a retry
                    if ($this->scheduleRetry) {
                        if ($this->timetable->getRetryCount($taskName) == 3) {
                            // Task has already been retried three times, give up
                            $this->timetable->clearRetryCount($taskName);

                            $this->logger->warning(
                                "Scheduler: '{task}' has already been retried three times, giving up",
                                ['task' => $taskName]
                            );
                        } else {
                            $readFromOption = true;
                            $rescheduledDate = $this->timetable->rescheduleTaskAndRunInOneHour($task);
                            $this->timetable->incrementRetryCount($taskName);

                            $this->logger->info(
                                "Scheduler: '{task}' retry scheduled for {date}",
                                ['task' => $taskName, 'date' => $rescheduledDate]
                            );
                        }
                        $this->scheduleRetry = false;
                    } else {
                        if ($this->timetable->getRetryCount($taskName) > 0) {
                            $this->timetable->clearRetryCount($taskName);
                        }
                    }

                    $executionResults[] = array('task' => $taskName, 'output' => $message);
                }

                $this->releaseLock();
            }
        }

        $this->logger->info("done");

        return $executionResults;
    }

    /**
     * Run a specific task now. Will ignore the schedule completely.
     *
     * @param string $taskName
     * @return string Task output.
     */
    public function runTaskNow($taskName)
    {
        $tasks = $this->loader->loadTasks();

        foreach ($tasks as $task) {
            if ($task->getName() === $taskName) {
                if (!$this->acquireLockForTask($taskName, $task->getTTL())) {
                    return 'Execution skipped. Another process is currently executing this task.';
                }

                $result = $this->executeTask($task);

                $this->releaseLock();

                return $result;
            }
        }

        throw new \InvalidArgumentException('Task ' . $taskName . ' not found');
    }

    /**
     * Determines a task's scheduled time and persists it, overwriting the previous scheduled time.
     *
     * Call this method if your task's scheduled time has changed due to, for example, an option that
     * was changed.
     *
     * @param Task $task Describes the scheduled task being rescheduled.
     * @api
     */
    public function rescheduleTask(Task $task)
    {
        $this->logger->debug('Rescheduling task {task}', array('task' => $task->getName()));

        $this->timetable->rescheduleTask($task);
    }

    /**
     * Determines a task's scheduled time and persists it, overwriting the previous scheduled time.
     *
     * Call this method if your task's scheduled time has changed due to, for example, an option that
     * was changed.
     *
     * The task will be run the first time tomorrow.
     *
     * @param Task $task Describes the scheduled task being rescheduled.
     * @api
     */
    public function rescheduleTaskAndRunTomorrow(Task $task)
    {
        $this->logger->debug('Rescheduling task and setting first run for tomorrow {task}', array('task' => $task->getName()));

        $this->timetable->rescheduleTaskAndRunTomorrow($task);
    }

    /**
     * Returns true if the scheduler is currently running a task.
     *
     * @return bool
     */
    public function isRunningTask()
    {
        return $this->isRunningTask;
    }

    /**
     * Return the next scheduled time given the class and method names of a scheduled task.
     *
     * @param string $className The name of the class that contains the scheduled task method.
     * @param string $methodName The name of the scheduled task method.
     * @param string|null $methodParameter Optional method parameter.
     * @return mixed int|bool The time in milliseconds when the scheduled task will be executed
     *                        next or false if it is not scheduled to run.
     */
    public function getScheduledTimeForMethod($className, $methodName, $methodParameter = null)
    {
        return $this->timetable->getScheduledTimeForMethod($className, $methodName, $methodParameter);
    }

    /**
     * Returns the list of the task names.
     *
     * @return string[]
     */
    public function getTaskList()
    {
        $tasks = $this->loader->loadTasks();

        return array_map(function (Task $task) {
            return $task->getName();
        }, $tasks);
    }

    private function acquireLockForTask(string $taskName, int $ttlInSeconds): bool
    {
        if (-1 === $ttlInSeconds) {
            // lock disabled, so don't try to acquire one
            return true;
        }

        return $this->lock->acquireLock($taskName, $ttlInSeconds);
    }

    private function releaseLock()
    {
        $this->lock->unlock();
    }

    /**
     * Executes the given task
     *
     * @param Task $task
     * @return string
     */
    private function executeTask($task)
    {
        $this->logger->info("Scheduler: executing task {taskName}...", array(
            'taskName' => $task->getName(),
        ));

        $this->isRunningTask = true;

        $timer = new Timer();

        /**
         * Triggered directly before a scheduled task is executed
         *
         * @param Task $task  The task that is about to be executed
         */
        Piwik::postEvent('ScheduledTasks.execute', array(&$task));

        try {
            $callable = array($task->getObjectInstance(), $task->getMethodName());
            call_user_func($callable, $task->getMethodParameter());
            $message = $timer->__toString();
        } catch (\Exception $e) {
            $this->logger->error(
                "Scheduler: Error {errorMessage} for task '{task}'",
                ['errorMessage' => $e->getMessage(), 'task' => $task->getName()]
            );
            $message = 'ERROR: ' . $e->getMessage();

            // If the task has indicated that retrying on exception is safe then flag for rescheduling
            if ($e instanceof RetryableException) {
                $this->scheduleRetry = true;
            }
        }

        $this->isRunningTask = false;

        /**
         * Triggered after a scheduled task is successfully executed.
         *
         * You can use the event to execute for example another task whenever a specific task is executed or to clean up
         * certain resources.
         *
         * @param Task $task The task that was just executed
         */
        Piwik::postEvent('ScheduledTasks.execute.end', array(&$task));

        $this->logger->info("Scheduler: finished. {timeElapsed}", array(
            'timeElapsed' => $timer,
        ));

        return $message;
    }
}