Exponential Basic 2.4.0.1 Exponential Basic CMS — PHP Class Reference

ezpcontentpublishingqueueprocessor.php
Go to the documentation of this file.
1 <?php
13 // required for proper signal handling
14 declare( ticks=1 );
15
21 class ezpContentPublishingQueueProcessor
22 {
23 public $sleepInterval;
24 public $signalQueue;
25 public function __construct()
26 {
27 $this->contentINI = eZINI::instance( 'content.ini' );
28 $this->allowedPublishingSlots = $this->contentINI->variable( 'PublishingSettings', 'PublishingProcessSlots' );
29 $this->sleepInterval = $this->contentINI->variable( 'PublishingSettings', 'AsynchronousPollingInterval' );
30
31 // output to log by default
32 $this->setOutput( new ezpAsynchronousPublisherLogOutput );
33
34 // initiates timer for the DB cleanup process
35 $this->cleanupLastTime = time();
36 $this->cleanupInterval = $this->contentINI->variable( 'PublishingSettings', 'AsynchronousCleanupInterval' );
37 $this->cleanupAgeLimit = $this->contentINI->variable( 'PublishingSettings', 'AsynchronousCleanupAgeLimit' );
38
39 // initiate current time for cache refresh
40 $this->lastRefreshTimestamp = time();
41
42 // Queue reader handler
43 $this->queueReader = $this->contentINI->variable( 'PublishingSettings', 'AsynchronousPublishingQueueReader' );
44 $reflection = new ReflectionClass( $this->queueReader );
45 if ( !$reflection->implementsInterface( 'ezpContentPublishingQueueReaderInterface' ) )
46 throw new Exception( "Configured asynchronous publishing queue reader doesn't implement ezpContentPublishingQueueReaderInterface", __CLASS__ );
47
48 call_user_func( array( $this->queueReader, 'init' ) );
49 }
50
55 public static function instance()
56 {
57 if ( !self::$instance instanceof ezpContentPublishingQueueProcessor )
58 {
59 self::$instance = new ezpContentPublishingQueueProcessor();
60 // signal handler for children termination signal
61 self::$instance->registerSignalHandler();
62 }
63
64 return self::$instance;
65 }
66
71 private function isSlotAvailable()
72 {
73 return ( ezpContentPublishingProcess::currentWorkingProcessCount() < $this->allowedPublishingSlots );
74 }
75
82 public function run()
83 {
84 // use DB exceptions so that errors can be fully handled
85 eZDB::setErrorHandling( eZDB::ERROR_HANDLING_EXCEPTIONS );
86
87 $this->cleanupDeadProcesses();
88
89 while ( $this->canProcess )
90 {
91 try {
92 $this->cleanupFinishedProcesses();
94 $publishingItem = $this->getNextItem();
95 if ( $publishingItem === false )
96 {
97 sleep( $this->sleepInterval );
98 continue;
99 }
100
101 if ( !$this->isSlotAvailable() )
102 {
103 $this->out->write( "No slot is available", 'async.log' );
104 sleep( $this->sleepInterval );
105 continue;
106 }
107
108 $this->out->write(
109 sprintf(
110 "Processing item #%d (content %d.%d)",
111 $publishingItem->attribute( 'ezcontentobject_version_id' ),
112 $publishingItem->version()->attribute( 'contentobject_id' ),
113 $publishingItem->version()->attribute( 'version' )
114 ),
115 'async.log'
116 );
117
118 $this->refreshCacheStateInfo();
119
120 $pid = $publishingItem->publish();
121 pcntl_signal( SIGINT, $this->signalHandler );
122 $this->currentJobs[$pid] = $publishingItem;
123
124 // In the event that a signal for this pid was caught before we get here, it will be in our
125 // signalQueue array
126 // Process it now as if we'd just received the signal
127 if ( isset( $this->signalQueue[$pid] ) )
128 {
129 $this->out->write( "found $pid in the signal queue, processing it now" );
130 $this->childSignalHandler( SIGCHLD, $pid, $this->signalQueue[$pid] );
131 unset( $this->signalQueue[$pid] );
132 }
133 } catch ( eZDBException $e ) {
134 $this->out->write( "Database error #" . $e->getCode() . ": " . $e->getMessage() );
135 // force the DB connection closed so that it is recreated
136 try {
137 $db = eZDB::instance();
138 $db->close();
139 $db = null;
140 eZDB::setInstance( null );
141 } catch( eZDBException $e ) {
142 // Do nothing, this will be retried until the DB is back up
143 }
144 sleep( 1 );
145 }
146 }
147 }
148
152 private function refreshCacheStateInfo()
153 {
154 if ( time() < ( $this->lastRefreshTimestamp + $this->sleepInterval ) )
155 {
156 return;
157 }
158
159 // Reload expiry timestamps
160 eZExpiryHandler::instance()->restore();
161
162 $this->lastRefreshTimestamp = time();
163 }
164
170 private function cleanupDeadProcesses()
171 {
172 $processes = ezpContentPublishingProcess::fetchProcesses( ezpContentPublishingProcess::STATUS_WORKING );
173 foreach( $processes as $process )
174 {
175 if ( !$process->isAlive() )
176 {
177 $process->reset();
178 }
179 }
180 }
181
188 private function cleanupFinishedProcesses()
189 {
190 if ( time() < ( $this->cleanupLastTime + $this->cleanupInterval ) )
191 {
192 return;
193 }
194
195 $processes = count( ezpContentPublishingProcess::fetchProcesses( ezpContentPublishingProcess::STATUS_FINISHED ) );
196 if ( $processes > 0 )
197 {
198 //Remove all objects at once
199 // this is required as the MySQL connection might be closed anytime by a fork
200 try
201 {
202 eZDebug::writeNotice( "ASYNC:: removing processes entries marked as STATUS_FINISHED in database.");
203 $db = eZDB::instance();
204 eZDB::setInstance( null );
205 $deleteBefore = time() - $this->cleanupAgeLimit;
206 $definition = ezpContentPublishingProcess::definition();
207 $processTable = $definition['name'];
208 $db->query( "DELETE from ". $processTable. " WHERE status =". ezpContentPublishingProcess::STATUS_FINISHED. " AND finished < ". $deleteBefore );
209 $this->cleanupLastTime = time();
210 }
211 catch( eZDBException $e )
212 {
213 // Do nothing, this will be retried until the DB is back up
214 }
215 }
216 }
217
221 public function childSignalHandler( $signo, $pid = null, $status = null )
222 {
223 // If no pid is provided, that means we're getting the signal from the system. Let's figure out
224 // which child process ended
225 if( $pid === null )
226 {
227 $pid = pcntl_waitpid( -1, $status, WNOHANG );
228 }
229
230 //Make sure we get all of the exited children
231 while( $pid > 0 )
232 {
233 if( $pid && isset( $this->currentJobs[$pid] ) )
234 {
235 $exitCode = pcntl_wexitstatus( $status );
236 if ( $exitCode != 0 )
237 {
238 $this->out->write( "Process #{$pid} of object version #".$this->currentJobs[$pid]->attribute( 'ezcontentobject_version_id' ) . " exited with status {$exitCode}" );
239 // this is required as the MySQL connection might be closed anytime by a fork
240 // this method is asynchronous, and might be triggered by any signal
241 // the only way is to use a dedicated DB connection, and close it afterwards
242 eZDB::setInstance( eZDB::instance( false, false, true ) );
243
244 $this->currentJobs[$pid]->reset();
245
246 eZDB::instance()->close();
247 eZDB::setInstance( null );
248 }
249 unset( $this->currentJobs[$pid] );
250 }
251 // A job has finished before the parent process could even note that it had been launched
252 // Let's make note of it and handle it when the parent process is ready for it
253 // echo "..... Adding $pid to the signal queue ..... \n";
254 elseif( $pid )
255 {
256 $this->signalQueue[$pid] = $status;
257 }
258 $pid = pcntl_waitpid( -1, $status, WNOHANG );
259 }
260 return true;
261 }
262
266 private function registerSignalHandler()
267 {
268 pcntl_signal( SIGCHLD, array( $this, 'childSignalHandler' ) );
269 }
270
274 public static function terminate()
275 {
276 self::instance()->_terminate();
277 }
278
279 private function _terminate()
280 {
281
282 $this->canProcess = false;
283
284 $this->out->write( "\n\n" );
285 while( !empty( $this->currentJobs ) )
286 {
287 $this->out->write( count( $this->currentJobs ) . ' jobs remaining, waiting...' );
288 sleep( 1 );
289 }
290
291 if ( empty( $this->currentJobs ) )
292 {
293 $this->out->write( 'No waiting children, exiting' );
294 }
295 }
296
303 public function setOutput( ezpAsynchronousPublisherOutput $output )
304 {
305 if ( !$output instanceof ezpAsynchronousPublisherOutput )
306 throw new Exception( "Invalid output handler" );
307 $this->out = $output;
308 }
309
315 public function setSignalHandler( $signalHandler)
316 {
317 if ( !is_callable( $signalHandler) )
318 {
319 throw new InvalidArgumentException( "\$signalHandler is not callable" );
320 }
321 $this->signalHandler = $signalHandler;
322 }
323
327 private function getNextItem()
328 {
329 return call_user_func( array( $this->queueReader, 'next' ) );
330 }
331
335 private $contentINI;
336
341 private $allowedPublishingSlots = null;
342
347 private static $instance = null;
348
349 private $canProcess = true;
350
355 private $currentJobs = array();
356
361 private $out;
362
367 private $signalHandler;
368
372 private $queueReader;
373
378 private $cleanupLastTime;
379
384 private $lastRefreshTimestamp;
385
390 private $cleanupInterval = 43200;
391
397 private $cleanupAgeLimit = 604800;
398 }
399 ?>
$status
$status
Definition: bugedit.php:473
eZDBException
Definition: exception.php:20
eZDebug\writeNotice
static writeNotice( $string, $label="", $backgroundClass="")
Definition: ezdebug.php:460
ezpAsynchronousPublisherLogOutput
Definition: log.php:17
ezpContentPublishingProcess\STATUS_WORKING
const STATUS_WORKING
Definition: ezpcontentpublishingprocess.php:20
ezpContentPublishingProcess\STATUS_FINISHED
const STATUS_FINISHED
Definition: ezpcontentpublishingprocess.php:21
ezpContentPublishingProcess\fetchProcesses
static fetchProcesses( $status)
Definition: ezpcontentpublishingprocess.php:298
ezpContentPublishingProcess\definition
static definition()
Definition: ezpcontentpublishingprocess.php:38
ezpContentPublishingProcess\currentWorkingProcessCount
static currentWorkingProcessCount()
Definition: ezpcontentpublishingprocess.php:127
ezpContentPublishingQueueProcessor
Definition: ezpcontentpublishingqueueprocessor.php:22
ezpContentPublishingQueueProcessor\cleanupFinishedProcesses
cleanupFinishedProcesses()
Definition: ezpcontentpublishingqueueprocessor.php:188
ezpContentPublishingQueueProcessor\$cleanupLastTime
$cleanupLastTime
Definition: ezpcontentpublishingqueueprocessor.php:378
ezpContentPublishingQueueProcessor\setOutput
setOutput(ezpAsynchronousPublisherOutput $output)
Definition: ezpcontentpublishingqueueprocessor.php:303
ezpContentPublishingQueueProcessor\refreshCacheStateInfo
refreshCacheStateInfo()
Definition: ezpcontentpublishingqueueprocessor.php:152
ezpContentPublishingQueueProcessor\__construct
__construct()
Definition: ezpcontentpublishingqueueprocessor.php:25
ezpContentPublishingQueueProcessor\$currentJobs
$currentJobs
Definition: ezpcontentpublishingqueueprocessor.php:355
ezpContentPublishingQueueProcessor\$cleanupInterval
$cleanupInterval
Definition: ezpcontentpublishingqueueprocessor.php:390
ezpContentPublishingQueueProcessor\$signalQueue
$signalQueue
Definition: ezpcontentpublishingqueueprocessor.php:24
ezpContentPublishingQueueProcessor\registerSignalHandler
registerSignalHandler()
Definition: ezpcontentpublishingqueueprocessor.php:266
ezpContentPublishingQueueProcessor\terminate
static terminate()
Definition: ezpcontentpublishingqueueprocessor.php:274
ezpContentPublishingQueueProcessor\$canProcess
$canProcess
Definition: ezpcontentpublishingqueueprocessor.php:349
ezpContentPublishingQueueProcessor\$lastRefreshTimestamp
$lastRefreshTimestamp
Definition: ezpcontentpublishingqueueprocessor.php:384
ezpContentPublishingQueueProcessor\$out
$out
Definition: ezpcontentpublishingqueueprocessor.php:361
ezpContentPublishingQueueProcessor\getNextItem
getNextItem()
Definition: ezpcontentpublishingqueueprocessor.php:327
ezpContentPublishingQueueProcessor\instance
static instance()
Definition: ezpcontentpublishingqueueprocessor.php:55
ezpContentPublishingQueueProcessor\setSignalHandler
setSignalHandler( $signalHandler)
Definition: ezpcontentpublishingqueueprocessor.php:315
ezpContentPublishingQueueProcessor\$instance
static $instance
Definition: ezpcontentpublishingqueueprocessor.php:347
ezpContentPublishingQueueProcessor\cleanupDeadProcesses
cleanupDeadProcesses()
Definition: ezpcontentpublishingqueueprocessor.php:170
ezpContentPublishingQueueProcessor\$contentINI
$contentINI
Definition: ezpcontentpublishingqueueprocessor.php:335
ezpContentPublishingQueueProcessor\$allowedPublishingSlots
$allowedPublishingSlots
Definition: ezpcontentpublishingqueueprocessor.php:341
ezpContentPublishingQueueProcessor\childSignalHandler
childSignalHandler( $signo, $pid=null, $status=null)
Definition: ezpcontentpublishingqueueprocessor.php:221
ezpContentPublishingQueueProcessor\_terminate
_terminate()
Definition: ezpcontentpublishingqueueprocessor.php:279
ezpContentPublishingQueueProcessor\$sleepInterval
$sleepInterval
Definition: ezpcontentpublishingqueueprocessor.php:23
ezpContentPublishingQueueProcessor\$queueReader
$queueReader
Definition: ezpcontentpublishingqueueprocessor.php:372
ezpContentPublishingQueueProcessor\$signalHandler
$signalHandler
Definition: ezpcontentpublishingqueueprocessor.php:367
ezpContentPublishingQueueProcessor\$cleanupAgeLimit
$cleanupAgeLimit
Definition: ezpcontentpublishingqueueprocessor.php:397
ezpContentPublishingQueueProcessor\isSlotAvailable
isSlotAvailable()
Definition: ezpcontentpublishingqueueprocessor.php:71
$db
$db
Definition: ezarticle/user/authorlist.php:53
$output
$output
Definition: formpreview.php:81
ezpAsynchronousPublisherOutput
Definition: asynchronouspublisheroutput.php:17
$instance
$instance
Definition: orderedit.php:667
Copyright © 1998 – 2026 7x. All rights reserved. Licensed under the GNU GPLv2+.