3 /***************************************************************************\
4 * SPIP, Systeme de publication pour l'internet *
6 * Copyright (c) 2001-2012 *
7 * Arnaud Martin, Antoine Pitrou, Philippe Riviere, Emmanuel Saint-James *
9 * Ce programme est un logiciel libre distribue sous licence GNU/GPL. *
10 * Pour plus de details voir le fichier COPYING.txt ou l'aide en ligne. *
11 \***************************************************************************/
14 * Gestion des queues de travaux
18 if (!defined("_ECRIRE_INC_VERSION")) return;
20 define('_JQ_SCHEDULED',1);
21 define('_JQ_PENDING',0);
22 #define('_JQ_MAX_JOBS_EXECUTE',200); // pour personaliser le nombre de jobs traitables a chaque hit
23 #define('_JQ_MAX_JOBS_TIME_TO_EXECUTE',15); // pour personaliser le temps d'excution dispo a chaque hit
24 #define('_JQ_NB_JOBS_OVERFLOW',10000); // nombre de jobs a partir duquel on force le traitement en fin de hit pour purger
27 * Ajouter une tâche à la file
29 * Les tâches sont ensuites exécutées par date programmée croissant/priorité décroissante
32 * The function name to call.
34 * A human-readable description of the queued job.
36 * Optional array of arguments to pass to the function.
38 * Optional file path which needs to be included for $fucntion.
39 * @param $no_duplicate
40 * If TRUE, do not add the job to the queue if one with the same function and
41 * arguments already exists.
42 * If 'function_only' test of existence is only on function name (for cron job)
44 * time for starting the job. If 0, job will start as soon as possible
46 * -10 (low priority) to +10 (high priority), 0 is the default
50 function queue_add_job($function, $description, $arguments = array(), $file = '', $no_duplicate = false, $time=0, $priority=0){
51 include_spip('base/abstract_sql');
53 // cas pourri de ecrire/action/editer_site avec l'option reload=oui
54 if (defined('_GENIE_SYNDIC_NOW'))
55 $arguments['id_syndic'] = _GENIE_SYNDIC_NOW
;
57 // serialiser les arguments
58 $arguments = serialize($arguments);
59 $md5args = md5($arguments);
61 // si pas de date programee, des que possible
62 $duplicate_where = 'status='.intval(_JQ_SCHEDULED
).' AND ';
65 $duplicate_where = ""; // ne pas dupliquer si deja le meme job en cours d'execution
67 $date = date('Y-m-d H:i:s',$time);
70 'fonction'=>$function,
71 'descriptif'=>$description,
75 'priorite'=>max(-10,min(10,intval($priority))),
77 'status'=>_JQ_SCHEDULED
,
79 // si option ne pas dupliquer, regarder si la fonction existe deja
80 // avec les memes args et file
84 $id_job = sql_getfetsel('id_job','spip_jobs',
86 $duplicate_where . 'fonction='.sql_quote($function)
87 .(($no_duplicate==='function_only')?
'':
88 ' AND md5args='.sql_quote($md5args).' AND inclure='.sql_quote($file)))
92 $id_job = sql_insertq('spip_jobs',$set_job);
93 // en cas de concurrence, deux process peuvent arriver jusqu'ici en parallele
94 // avec le meme job unique a inserer. Dans ce cas, celui qui a eu l'id le plus grand
99 $id_prev = sql_getfetsel('id_job','spip_jobs',"id_job<".intval($id_job)." AND $duplicate_where")){
100 sql_delete('spip_jobs','id_job='.intval($id_job));
104 // verifier la non duplication qui peut etre problematique en cas de concurence
105 // il faut dans ce cas que seul le dernier ajoute se supprime !
107 // une option de debug pour verifier que les arguments en base sont bons
108 // ie cas d'un char non acceptables sur certains type de champs
109 // qui coupe la valeur
110 if (defined('_JQ_INSERT_CHECK_ARGS') AND $id_job) {
111 $args = sql_getfetsel('args', 'spip_jobs', 'id_job='.intval($id_job));
112 if ($args!==$arguments) {
113 spip_log('arguments job errones / longueur '.strlen($args)." vs ".strlen($arguments).' / valeur : '.var_export($arguments,true),'queue');
118 queue_update_next_job_time($time);
120 // si la mise en file d'attente du job echoue,
121 // il ne faut pas perdre l'execution de la fonction
122 // on la lance immediatement, c'est un fallback
123 // sauf en cas d'upgrade necessaire (table spip_jobs inexistante)
124 elseif($GLOBALS['meta']['version_installee']==$GLOBALS['spip_version_base']) {
125 $set_job['id_job'] = 0;
126 queue_start_job($set_job);
133 * Purger la file de tâche et reprgrammer les tâches périodiques
137 function queue_purger(){
138 include_spip('base/abstract_sql');
139 sql_delete('spip_jobs');
140 sql_delete("spip_jobs_liens","id_job NOT IN (".sql_get_select("id_job","spip_jobs").")");
141 include_spip('inc/genie');
142 genie_queue_watch_dist();
146 * Retirer une tache de la file d'attente
148 * id de la tache a retirer
151 function queue_remove_job($id_job){
152 include_spip('base/abstract_sql');
154 if ($row = sql_fetsel('fonction,inclure,date','spip_jobs','id_job='.intval($id_job))
155 AND $res = sql_delete('spip_jobs','id_job='.intval($id_job))){
156 queue_unlink_job($id_job);
157 // est-ce une tache cron qu'il faut relancer ?
158 if ($periode = queue_is_cron_job($row['fonction'],$row['inclure'])){
159 // relancer avec les nouveaux arguments de temps
160 include_spip('inc/genie');
161 // relancer avec la periode prevue
162 queue_genie_replan_job($row['fonction'],$periode,strtotime($row['date']));
164 queue_update_next_job_time();
170 * Associer une tache avec un objet
174 * @param array $objets
175 * can be a simple array('objet'=>'article','id_objet'=>23)
176 * or an array of simple array to link multiples objet in one time
178 function queue_link_job($id_job,$objets){
179 include_spip('base/abstract_sql');
181 if (is_array($objets) AND count($objets)){
182 if (is_array(reset($objets))){
183 foreach($objets as $k=>$o){
184 $objets[$k]['id_job'] = $id_job;
186 sql_insertq_multi('spip_jobs_liens',$objets);
189 sql_insertq('spip_jobs_liens',array_merge(array('id_job'=>$id_job),$objets));
194 * Dissocier une tache d'un objet
197 * id of job to unlink ibject with
199 * result of sql_delete
201 function queue_unlink_job($id_job){
202 return sql_delete("spip_jobs_liens","id_job=".intval($id_job));
206 * Lancer une tache decrite par sa ligne SQL
208 * describe the job, with field of table spip_jobs
210 * return the result of job
212 function queue_start_job($row){
214 // deserialiser les arguments
215 $args = unserialize($row['args']);
217 spip_log('arguments job errones '.var_export($row,true),'queue');
221 $fonction = $row['fonction'];
222 if (strlen($inclure = trim($row['inclure']))){
223 if (substr($inclure,-1)=='/'){ // c'est un chemin pour charger_fonction
224 $f = charger_fonction($fonction,rtrim($inclure,'/'),false);
229 include_spip($inclure);
232 if (!function_exists($fonction)){
233 spip_log("fonction $fonction ($inclure) inexistante ".var_export($row,true),'queue');
237 spip_log("queue [".$row['id_job']."]: $fonction() start", 'queue');
238 switch (count($args)) {
239 case 0: $res = $fonction(); break;
240 case 1: $res = $fonction($args[0]); break;
241 case 2: $res = $fonction($args[0],$args[1]); break;
242 case 3: $res = $fonction($args[0],$args[1], $args[2]); break;
243 case 4: $res = $fonction($args[0],$args[1], $args[2], $args[3]); break;
244 case 5: $res = $fonction($args[0],$args[1], $args[2], $args[3], $args[4]); break;
245 case 6: $res = $fonction($args[0],$args[1], $args[2], $args[3], $args[4], $args[5]); break;
246 case 7: $res = $fonction($args[0],$args[1], $args[2], $args[3], $args[4], $args[5], $args[6]); break;
247 case 8: $res = $fonction($args[0],$args[1], $args[2], $args[3], $args[4], $args[5], $args[6], $args[7]); break;
248 case 9: $res = $fonction($args[0],$args[1], $args[2], $args[3], $args[4], $args[5], $args[6], $args[7], $args[8]); break;
249 case 10:$res = $fonction($args[0],$args[1], $args[2], $args[3], $args[4], $args[5], $args[6], $args[7], $args[8], $args[9]); break;
251 # plus lent mais completement generique
252 $res = call_user_func_array($fonction, $args);
254 spip_log("queue [".$row['id_job']."]: $fonction() end", 'queue');
261 * Prend une par une les taches en attente
262 * et les lance, dans la limite d'un temps disponible total
263 * et d'un nombre maxi de taches
265 * La date de la prochaine tache a executer est mise a jour
266 * apres chaque chaque tache finie
267 * afin de relancer le scheduler uniquement quand c'est necessaire
269 * @param array $force_jobs
270 * list of id_job to execute when provided
272 function queue_schedule($force_jobs = null){
274 if (defined('_DEBUG_BLOCK_QUEUE')) {
275 spip_log("_DEBUG_BLOCK_QUEUE : schedule stop",'jq'._LOG_DEBUG
);
279 // rien a faire si le prochain job est encore dans le futur
280 if (queue_sleep_time_to_next_job() AND (!$force_jobs OR !count($force_jobs))){
281 spip_log("queue_sleep_time_to_next_job",'jq'._LOG_DEBUG
);
285 include_spip('base/abstract_sql');
287 if (!defined('_JQ_MAX_JOBS_TIME_TO_EXECUTE')){
288 $max_time = ini_get('max_execution_time')/2;
289 // valeur conservatrice si on a pas reussi a lire le max_execution_time
290 if (!$max_time) $max_time=5;
291 define('_JQ_MAX_JOBS_TIME_TO_EXECUTE',min($max_time,15)); // une valeur maxi en temps.
293 $end_time = $time + _JQ_MAX_JOBS_TIME_TO_EXECUTE
;
295 spip_log("JQ schedule $time / $end_time",'jq'._LOG_DEBUG
);
297 if (!defined('_JQ_MAX_JOBS_EXECUTE'))
298 define('_JQ_MAX_JOBS_EXECUTE',200);
301 // dont la date est passee (echus en attente),
305 // lorsqu'un job cron n'a pas fini, sa priorite est descendue
306 // pour qu'il ne bloque pas les autres jobs en attente
307 if (is_array($force_jobs) AND count($force_jobs))
308 $cond = "status=".intval(_JQ_SCHEDULED
)." AND ".sql_in("id_job", $force_jobs);
310 $now = date('Y-m-d H:i:s',$time);
311 $cond = "status=".intval(_JQ_SCHEDULED
)." AND date<=".sql_quote($now);
314 register_shutdown_function('queue_error_handler'); // recuperer les erreurs auant que possible
315 $res = sql_allfetsel('*','spip_jobs',$cond,'','priorite DESC,date','0,'.(_JQ_MAX_JOBS_EXECUTE+
1));
317 if ($row = array_shift($res)){
319 // il faut un verrou, a base de sql_delete
320 if (sql_delete('spip_jobs',"id_job=".intval($row['id_job'])." AND status=".intval(_JQ_SCHEDULED
))){
321 #spip_log("JQ schedule job ".$nbj." OK",'jq');
322 // on reinsert dans la base aussitot avec un status=_JQ_PENDING
323 $row['status'] = _JQ_PENDING
;
324 $row['date'] = date('Y-m-d H:i:s',$time);
325 sql_insertq('spip_jobs', $row);
327 // on a la main sur le job :
329 $result = queue_start_job($row);
332 queue_close_job($row, $time, $result);
335 spip_log("JQ schedule job end time ".$time,'jq'._LOG_DEBUG
);
336 } while ($nbj<_JQ_MAX_JOBS_EXECUTE
AND $row AND $time<$end_time);
337 spip_log("JQ schedule end time ".time(),'jq'._LOG_DEBUG
);
339 if ($row = array_shift($res)){
340 queue_update_next_job_time(0); // on sait qu'il y a encore des jobs a lancer ASAP
341 spip_log("JQ encore !",'jq'._LOG_DEBUG
);
344 queue_update_next_job_time();
349 * Terminer un job au status _JQ_PENDING :
350 * - le reprogrammer si c'est un cron
351 * - supprimer ses liens
352 * - le detruire en dernier
358 function queue_close_job(&$row,$time,$result=0){
359 // est-ce une tache cron qu'il faut relancer ?
360 if ($periode = queue_is_cron_job($row['fonction'],$row['inclure'])){
361 // relancer avec les nouveaux arguments de temps
362 include_spip('inc/genie');
364 // relancer tout de suite, mais en baissant la priorite
365 queue_genie_replan_job($row['fonction'],$periode,0-$result,null,$row['priorite']-1);
367 // relancer avec la periode prevue
368 queue_genie_replan_job($row['fonction'],$periode,$time);
370 // purger ses liens eventuels avec des objets
371 sql_delete("spip_jobs_liens","id_job=".intval($row['id_job']));
372 // supprimer le job fini
373 sql_delete('spip_jobs','id_job='.intval($row['id_job']));
377 * Recuperer des erreurs auant que possible
378 * en terminant la gestion de la queue
380 function queue_error_handler(){
381 // se remettre dans le bon dossier, car Apache le change parfois (toujours?)
384 queue_update_next_job_time();
389 * Tester si une tache etait une tache periodique a reprogrammer
391 * @param <type> $function
392 * @param <type> $inclure
395 function queue_is_cron_job($function,$inclure){
396 static $taches = null;
397 if (strncmp($inclure,'genie/',6)==0){
398 if (is_null($taches)){
399 include_spip('inc/genie');
400 $taches = taches_generales();
402 if (isset($taches[$function]))
403 return $taches[$function];
409 * Mettre a jour la date du prochain job a lancer
410 * Si une date est fournie (au format time unix)
411 * on fait simplement un min entre la date deja connue et celle fournie
412 * (cas de l'ajout simple
413 * ou cas $next_time=0 car l'on sait qu'il faut revenir ASAP)
415 * @param int $next_time
416 * temps de la tache ajoutee ou 0 pour ASAP
418 function queue_update_next_job_time($next_time=null){
419 static $nb_jobs_scheduled = null;
420 static $deja_la = false;
421 // prendre le min des $next_time que l'on voit passer ici, en cas de reentrance
423 // queue_close_job peut etre reentrant ici
424 if ($deja_la) return;
427 include_spip('base/abstract_sql');
430 // traiter les jobs morts au combat (_JQ_PENDING depuis plus de 180s)
431 // pour cause de timeout ou autre erreur fatale
432 $res = sql_allfetsel("*","spip_jobs","status=".intval(_JQ_PENDING
)." AND date<".sql_quote(date('Y-m-d H:i:s',$time-180)));
433 if (is_array($res)) {
434 foreach ($res as $row)
435 queue_close_job($row,$time);
438 // chercher la date du prochain job si pas connu
439 if (is_null($next) OR is_null(queue_sleep_time_to_next_job())){
440 $date = sql_getfetsel('date','spip_jobs',"status=".intval(_JQ_SCHEDULED
),'','date','0,1');
441 $next = strtotime($date);
443 if (!is_null($next_time)){
444 if (is_null($next) OR $next>$next_time)
449 if (is_null($nb_jobs_scheduled))
450 $nb_jobs_scheduled = sql_countsel('spip_jobs',"status=".intval(_JQ_SCHEDULED
)." AND date<".sql_quote(date('Y-m-d H:i:s',$time)));
451 elseif ($next<=$time)
452 $nb_jobs_scheduled++
;
453 // si trop de jobs en attente, on force la purge en fin de hit
454 // pour assurer le coup
455 if ($nb_jobs_scheduled>defined('_JQ_NB_JOBS_OVERFLOW')?_JQ_NB_JOBS_OVERFLOW
:10000)
456 define('_DIRECT_CRON_FORCE',true);
459 queue_set_next_job_time($next);
465 * Mettre a jour la date de prochain job
468 function queue_set_next_job_time($next) {
470 // utiliser le temps courant reel plutot que temps de la requete ici
473 // toujours relire la valeur pour comparer, pour tenir compte des maj concourrantes
474 // et ne mettre a jour que si il y a un interet a le faire
475 // permet ausis d'initialiser le nom de fichier a coup sur
476 $curr_next = $_SERVER['REQUEST_TIME'] +
queue_sleep_time_to_next_job(true);
478 ($curr_next<$time AND $next>$time) // le prochain job est dans le futur mais pas la date planifiee actuelle
479 OR $curr_next>$next // le prochain job est plus tot que la date planifiee actuelle
481 if (include_spip('inc/memoization') AND defined('_MEMOIZE_MEMORY') AND _MEMOIZE_MEMORY
) {
482 cache_set(_JQ_NEXT_JOB_TIME_FILENAME
,intval($next));
485 ecrire_fichier(_JQ_NEXT_JOB_TIME_FILENAME
,intval($next));
487 queue_sleep_time_to_next_job($next);
490 return queue_sleep_time_to_next_job();
494 * Déclenche le cron en asynchrone ou retourne le code HTML pour le déclencher
496 * Retourne le HTML à ajouter à la page pour declencher le cron
497 * ou rien si on a réussi à le lancer en asynchrone.
501 function queue_affichage_cron(){
504 // rien a faire si le prochain job est encore dans le futur
505 if (queue_sleep_time_to_next_job() OR defined('_DEBUG_BLOCK_QUEUE'))
508 // ne pas relancer si on vient de lancer dans la meme seconde par un hit concurent
509 if (file_exists($lock=_DIR_TMP
."cron.lock") AND !(@filemtime
($lock)<$_SERVER['REQUEST_TIME']))
513 // il y a des taches en attentes
515 $url_cron = generer_url_action('cron','',false,true);
517 // Si fsockopen est possible, on lance le cron via un socket
519 if(function_exists('fsockopen')){
520 $parts=parse_url($url_cron);
522 $fp = @fsockopen
($parts['host'],
523 isset($parts['port'])?
$parts['port']:80,
524 $errno, $errstr, 30);
527 $query = $parts['path'].($parts['query']?
"?".$parts['query']:"");
528 $out = "GET ".$query." HTTP/1.1\r\n";
529 $out.= "Host: ".$parts['host']."\r\n";
530 $out.= "Connection: Close\r\n\r\n";
537 // ici lancer le cron par un CURL asynchrone si CURL est present
538 if (function_exists("curl_init")){
539 //setting the curl parameters.
540 $ch = curl_init($url_cron);
541 curl_setopt($ch, CURLOPT_RETURNTRANSFER
, true);
542 // cf bug : http://www.php.net/manual/en/function.curl-setopt.php#104597
543 curl_setopt($ch, CURLOPT_NOSIGNAL
, 1);
544 // valeur mini pour que la requete soit lancee
545 curl_setopt($ch, CURLOPT_TIMEOUT_MS
, 100);
553 // si deja force, on retourne sans rien
554 if (defined('_DIRECT_CRON_FORCE'))
558 // inutile de faire un appel par image background,
559 // on force un appel direct en fin de hit
560 if ((defined('_IS_BOT') AND _IS_BOT
)){
561 define('_DIRECT_CRON_FORCE',true);
565 // en derniere solution, on insere une image background dans la page
566 $texte = '<!-- SPIP-CRON --><div style="background-image: url(\'' .
567 generer_url_action('cron') .