silverstripe-fulltextsearch/code/search/processors/SearchUpdateBatchedProcessor.php
Damian Mooyman 60b05db1ce API Separate searchupdate / commit into separate queued-jobs
API Enable indexes to deprecate other classes
API Enable indexes to be globally specified in config
2015-05-08 15:22:46 +12:00

158 lines
3.5 KiB
PHP

<?php
/**
* Provides batching of search updates
*/
abstract class SearchUpdateBatchedProcessor extends SearchUpdateProcessor {
/**
* List of batches to be processed
*
* @var array
*/
protected $batches;
/**
* Pointer to index of $batches assigned to $current.
* Set to 0 (first index) if not started, or count + 1 if completed.
*
* @var int
*/
protected $currentBatch;
/**
* List of indexes successfully comitted in the current batch
*
* @var array
*/
protected $completedIndexes;
/**
* Maximum number of record-states to process in one batch.
* Set to zero to process all records in a single batch
*
* @config
* @var int
*/
private static $batch_size = 100;
/**
* Up to this number of additional ids can be added to any batch in order to reduce the number
* of batches
*
* @config
* @var int
*/
private static $batch_soft_cap = 10;
public function __construct() {
parent::__construct();
$this->batches = array();
$this->setBatch(0);
}
/**
* Set the current batch index
*
* @param int $batch Index of the batch
*/
protected function setBatch($batch) {
$this->currentBatch = $batch;
}
protected function getSource() {
if(isset($this->batches[$this->currentBatch])) {
return $this->batches[$this->currentBatch];
}
}
/**
* Process the current queue
*
* @return boolean
*/
public function process() {
// Skip blank queues
if(empty($this->batches)) return true;
// Don't re-process completed queue
if($this->currentBatch >= count($this->batches)) return true;
// Send current patch to indexes
$this->prepareIndexes();
// Advance to next batch if successful
$this->setBatch($this->currentBatch + 1);
return true;
}
/**
* Segments batches acording to the specified rules
*
* @param array $source Source input
* @return array Batches
*/
protected function segmentBatches($source) {
// Measure batch_size
$batchSize = Config::inst()->get(get_class(), 'batch_size');
if($batchSize === 0) return array($source);
$softCap = Config::inst()->get(get_class(), 'batch_soft_cap');
// Clear batches
$batches = array();
$current = array();
$currentSize = 0;
// Build batches from data
foreach ($source as $base => $statefulids) {
if (!$statefulids) continue;
foreach ($statefulids as $stateKey => $statefulid) {
$state = $statefulid['state'];
$ids = $statefulid['ids'];
if(!$ids) continue;
// Extract items from $ids until empty
while($ids) {
// Estimate maximum number of items to take for this iteration, allowing for the soft cap
$take = $batchSize - $currentSize;
if(count($ids) <= $take + $softCap) $take += $softCap;
$items = array_slice($ids, 0, $take, true);
$ids = array_slice($ids, count($items), null, true);
// Update batch
$currentSize += count($items);
$merge = array(
$base => array(
$stateKey => array(
'state' => $state,
'ids' => $items
)
)
);
$current = $current ? array_merge_recursive($current, $merge) : $merge;
if($currentSize >= $batchSize) {
$batches[] = $current;
$current = array();
$currentSize = 0;
}
}
}
}
// Add incomplete batch
if($currentSize) $batches[] = $current;
return $batches;
}
public function batchData() {
$this->batches = $this->segmentBatches($this->dirty);
$this->setBatch(0);
}
public function triggerProcessing() {
$this->batchData();
}
}