diff -u b/migrate_tools/migrate_tools.drush.inc b/migrate_tools/migrate_tools.drush.inc --- b/migrate_tools/migrate_tools.drush.inc +++ b/migrate_tools/migrate_tools.drush.inc @@ -39,6 +39,8 @@ 'options' => array( 'all' => 'Process all migrations.', 'group' => 'Name of the migration group to import', + 'limit' => 'Limit on the number of items to process in each migration', + 'feedback' => 'Frequency of progress messages, in items processed', ), 'arguments' => array( 'migration' => 'Name of migration(s) to import. Delimit multiple using commas.', @@ -47,6 +49,7 @@ 'migrate-import --all' => 'Perform all migrations', 'migrate-import --group=beer' => 'Import all migrations in the beer group', 'migrate-import BeerTerm,BeerNode' => 'Import new terms and nodes', + 'migrate-import BeerUser --limit=2' => 'Import no more than 2 users', ), 'drupal dependencies' => array('migrate_tools'), 'aliases' => array('mi'), @@ -151,11 +154,19 @@ function drush_migrate_tools_migrate_import($migration_names = '') { $group_name = drush_get_option('group'); $all = drush_get_option('all'); + $options = []; if (!$all && !$group_name && !$migration_names) { drush_set_error('MIGRATE_ERROR', dt('You must specify --all, --group, or one or more migration names separated by commas')); return; } + if (drush_get_option('limit')) { + $options['limit'] = drush_get_option('limit'); + } + if (drush_get_option('feedback')) { + $options['feedback'] = drush_get_option('feedback'); + } + $log = new DrushLogMigrateMessage(); $migrations = drush_migrate_tools_migration_list($group_name, $migration_names); @@ -163,18 +174,9 @@ // Take it one group at a time, importing the migrations within each group. foreach ($migrations as $group_id => $migration_list) { foreach ($migration_list as $migration_id => $migration) { - $executable = new MigrateExecutable($migration, $log); + $executable = new MigrateExecutable($migration, $log, $options); // drush_op() provides --simulate support. drush_op(array($executable, 'import')); - $processed = $executable->getProcessedCount(); - drush_print(\Drupal::translation()->formatPlural($processed, - "Processed 1 item (!successes successfully, !failures failed, !ignored ignored) - done with '!name'", - "Processed !numitems items (!successes successfully, !failures failed, !ignored ignored) - done with '!name'", - array('!numitems' => $processed, - '!successes' => $executable->getImportedCount(), - '!failures' => $executable->getFailedCount(), - '!ignored' => $executable->getIgnoredCount(), - '!name' => $migration_id))); } } } diff -u b/migrate_tools/src/MigrateExecutable.php b/migrate_tools/src/MigrateExecutable.php --- b/migrate_tools/src/MigrateExecutable.php +++ b/migrate_tools/src/MigrateExecutable.php @@ -15,10 +15,39 @@ use Drupal\migrate\MigrateMapSaveEvent; use Drupal\migrate\MigrateMapDeleteEvent; use Drupal\migrate\MigratePostImportEvent; +use Drupal\migrate\MigratePostSaveEvent; class MigrateExecutable extends MigrateExecutableBase { /** + * The PHP max_execution_time. + * + * @var int + */ + protected $maxExecTime; + + /** + * The ratio of the memory limit at which an operation will be interrupted. + * + * @var float + */ + protected $memoryThreshold = 0.85; + + /** + * The ratio of the time limit at which an operation will be interrupted. + * + * @var float + */ + public $timeThreshold = 0.90; + + /** + * The PHP memory_limit expressed in bytes. + * + * @var int + */ + protected $memoryLimit; + + /** * Counters of map statuses. * * @var array @@ -39,16 +68,71 @@ protected $deleteCounter = 0; /** + * Maximum number of items to process in this migration. 0 indicates no limit + * is to be applied. + * + * @var int + */ + protected $itemLimit = 0; + + /** + * Frequency (in items) at which progress messages should be emitted. + * + * @var int + */ + protected $feedback = 0; + + /** + * Count of number of items processed so far in this migration. + * @var int + */ + protected $counter = 0; + + /** * {@inheritdoc} */ - public function __construct(MigrationInterface $migration, MigrateMessageInterface $message) { + public function __construct(MigrationInterface $migration, MigrateMessageInterface $message, array $options = []) { parent::__construct($migration, $message); + if (isset($options['limit'])) { + $this->itemLimit = $options['limit']; + } + if (isset($options['feedback'])) { + $this->feedback = $options['feedback']; + } \Drupal::service('event_dispatcher')->addListener(MigrateEvents::MAP_SAVE, array($this, 'onMapSave')); \Drupal::service('event_dispatcher')->addListener(MigrateEvents::MAP_DELETE, array($this, 'onMapDelete')); \Drupal::service('event_dispatcher')->addListener(MigrateEvents::POST_IMPORT, array($this, 'onPostImport')); + \Drupal::service('event_dispatcher')->addListener(MigrateEvents::POST_SAVE, + array($this, 'onPostSave')); + + // Record the memory limit in bytes + $limit = trim(ini_get('memory_limit')); + if ($limit == '-1') { + $this->memoryLimit = PHP_INT_MAX; + } + else { + if (!is_numeric($limit)) { + $last = strtolower(substr($limit, -1)); + switch ($last) { + case 'g': + $limit *= 1024; + case 'm': + $limit *= 1024; + case 'k': + $limit *= 1024; + break; + default: + throw new MigrateException($this->t('Invalid PHP memory_limit !limit', + array('!limit' => $limit))); + } + } + $this->memoryLimit = $limit; + } + // Record the maximum execution time limit. + $this->maxExecTime = ini_get('max_execution_time'); } /** @@ -122,6 +206,16 @@ } /** + * Reset all the per-status counters to 0. + */ + protected function resetCounters() { + foreach ($this->saveCounters as $status => $count) { + $this->saveCounters[$status] = 0; + } + $this->deleteCounter = 0; + } + + /** * React to migration completion. * * @param \Drupal\migrate\MigratePostImportEvent $event @@ -130,6 +224,201 @@ public function onPostImport(MigratePostImportEvent $event) { $migrate_last_imported_store = \Drupal::keyValue('migrate_last_imported'); $migrate_last_imported_store->set($event->getMigration()->id(), round(microtime(TRUE) * 1000)); + $this->progressMessage(); + } + + /** + * Emit information on what we've done since the last feedback (or the + * beginning of this migration). + * + * @param bool $done + */ + protected function progressMessage($done = TRUE) { + $processed = $this->getProcessedCount(); + if ($done) { + $singular_message = "Processed 1 item (!successes successfully, !failures failed, !ignored ignored) - done with '!name'"; + $plural_message = "Processed !numitems items (!successes successfully, !failures failed, !ignored ignored) - done with '!name'"; + } + else { + $singular_message = "Processed 1 item (!successes successfully, !failures failed, !ignored ignored) - continuing with '!name'"; + $plural_message = "Processed !numitems items (!successes successfully, !failures failed, !ignored ignored) - continuing with '!name'"; + } + $this->message->display(\Drupal::translation()->formatPlural($processed, + $singular_message, $plural_message, + array('!numitems' => $processed, + '!successes' => $this->getImportedCount(), + '!failures' => $this->getFailedCount(), + '!ignored' => $this->getIgnoredCount(), + '!name' => $this->migration->id()))); + } + + /** + * React to item import. + * + * @param \Drupal\migrate\MigratePostSaveEvent $event + * The post-save event. + */ + public function onPostSave(MigratePostSaveEvent $event) { + $migration = $event->getMigration(); + if ($this->feedback && ($this->counter) && $this->counter % $this->feedback == 0) { + $this->progressMessage(FALSE); + $this->resetCounters(); + } + $this->counter++; + if ($this->itemLimit && $this->counter >= $this->itemLimit) { + $migration->setMigrationResult(MigrationInterface::STATUS_STOPPING); + } + if ($this->timeOptionExceeded()) { + $migration->setMigrationResult(MigrationInterface::STATUS_STOPPING); + } + + // @todo: Finish integrating time/memory checking. + if ($this->memoryExceeded()) { +// return MigrationInterface::RESULT_INCOMPLETE; + } + if ($this->maxExecTimeExceeded()) { +// return MigrationInterface::RESULT_INCOMPLETE; + } + + } + + /** + * Tests whether we've exceeded the designated time limit. + * + * @return bool + * TRUE if the threshold is exceeded, FALSE if not. + */ + protected function timeOptionExceeded() { + // If there is no time limit, then it is not exceeded. + if (!$time_limit = $this->getTimeLimit()) { + return FALSE; + } + // Calculate if the time limit is exceeded. + $time_elapsed = $this->getTimeElapsed(); + if ($time_elapsed >= $time_limit) { + return TRUE; + } + else { + return FALSE; + } + } + + /** + * {@inheritdoc} + */ + public function getTimeLimit() { + $limit = $this->limit; + if (isset($limit['unit']) && isset($limit['value']) && ($limit['unit'] == 'seconds' || $limit['unit'] == 'second')) { + return $limit['value']; + } + else { + return NULL; + } + } + + /** + * Tests whether we've exceeded the desired memory threshold. + * + * If so, output a message. + * + * @return bool + * TRUE if the threshold is exceeded, otherwise FALSE. + */ + protected function memoryExceeded() { + $usage = $this->getMemoryUsage(); + $pct_memory = $usage / $this->memoryLimit; + if (!$threshold = $this->memoryThreshold) { + return FALSE; + } + if ($pct_memory > $threshold) { + $this->message->display( + $this->t('Memory usage is !usage (!pct% of limit !limit), reclaiming memory.', + array('!pct' => round($pct_memory*100), + '!usage' => $this->formatSize($usage), + '!limit' => $this->formatSize($this->memoryLimit))), + 'warning'); + $usage = $this->attemptMemoryReclaim(); + $pct_memory = $usage / $this->memoryLimit; + // Use a lower threshold - we don't want to be in a situation where we keep + // coming back here and trimming a tiny amount + if ($pct_memory > (0.90 * $threshold)) { + $this->message->display( + $this->t('Memory usage is now !usage (!pct% of limit !limit), not enough reclaimed, starting new batch', + array('!pct' => round($pct_memory*100), + '!usage' => $this->formatSize($usage), + '!limit' => $this->formatSize($this->memoryLimit))), + 'warning'); + return TRUE; + } + else { + $this->message->display( + $this->t('Memory usage is now !usage (!pct% of limit !limit), reclaimed enough, continuing', + array('!pct' => round($pct_memory*100), + '!usage' => $this->formatSize($usage), + '!limit' => $this->formatSize($this->memoryLimit))), + 'warning'); + return FALSE; + } + } + else { + return FALSE; + } + } + + /** + * Returns the memory usage so far. + * + * @return int + * The memory usage. + */ + protected function getMemoryUsage() { + return memory_get_usage(); + } + + /** + * Tries to reclaim memory. + * + * @return int + * The memory usage after reclaim. + */ + protected function attemptMemoryReclaim() { + // First, try resetting Drupal's static storage - this frequently releases + // plenty of memory to continue. + drupal_static_reset(); + // @TODO: explore resetting the container. + return memory_get_usage(); + } + + /** + * Generates a string representation for the given byte count. + * + * @param int $size + * A size in bytes. + * + * @return string + * A translated string representation of the size. + */ + protected function formatSize($size) { + return format_size($size); + } + + /** + * Tests whether we're approaching the PHP maximum execution time limit. + * + * @return bool + * TRUE if the threshold is exceeded, FALSE if not. + */ + protected function maxExecTimeExceeded() { + return $this->maxExecTime && (($this->getTimeElapsed() / $this->maxExecTime) > $this->timeThreshold); + } + + /** + * Returns the time elapsed. + * + * This allows a test to set a fake elapsed time. + */ + protected function getTimeElapsed() { + return time() - REQUEST_TIME; } }