Bug 36549: Use Parallel::ForkManager in es_indexer_daemon
This patch copies the implementation from the general background
worked into the es_indexer daemon
Note: There is a 'sleep(10)' in the general indexer and I copied it here,
but I don't know why we do it?
To test:
1 - Apply the patch
2 - sudo koha-es-indexer --restart kohadev
3 - perl misc/maintenance/touch_all_biblios.pl
4 - Wait a bit
5 - Check the background jobs in the admin interface
6 - Confirm the jobs finish and records are searchable
7 - Test with and without rabbitmq enabled:
. Stop rabbitmq service: sudo -s service rabbitmq-server stop
. Repeat steps 2 to 5
8 - Sign off
Signed-off-by: David Nind <david@davidnind.com>
Signed-off-by: Aleisha Amohia <aleisha@catalyst.net.nz>
Signed-off-by: Pedro Amorim <pedro.amorim@openfifth.co.uk>
This commit is contained in:
parent
7a5c6b6066
commit
6b7db3cfb4
|
|
@ -48,6 +48,11 @@ This script will connect to the Stomp server (RabbitMQ) and subscribe to the Ela
|
|||
If a Stomp server is not active it will poll the database every 10s for new jobs in the Elasticsearch queue
|
||||
and process them in batches every second.
|
||||
|
||||
--m --max-processes specifies how many jobs to process simultaneously
|
||||
|
||||
Max processes will be set from the command line option, the environment variable MAX_PROCESSES, or the koha-conf file, in that order of precedence.
|
||||
By default the script will only run one job at a time.
|
||||
|
||||
=cut
|
||||
|
||||
use Modern::Perl;
|
||||
|
|
@ -55,8 +60,10 @@ use JSON qw( decode_json );
|
|||
use Try::Tiny;
|
||||
use Pod::Usage;
|
||||
use Getopt::Long;
|
||||
use Parallel::ForkManager;
|
||||
use List::MoreUtils qw( natatime );
|
||||
use Time::HiRes;
|
||||
use Storable qw(dclone);
|
||||
|
||||
use C4::Context;
|
||||
use Koha::Logger;
|
||||
|
|
@ -68,12 +75,18 @@ use Koha::SearchEngine::Indexer;
|
|||
my $help;
|
||||
my $batch_size = 10;
|
||||
|
||||
my $max_processes = $ENV{MAX_PROCESSES};
|
||||
$max_processes ||= C4::Context->config('background_jobs_worker')->{max_processes}
|
||||
if C4::Context->config('background_jobs_worker');
|
||||
$max_processes ||= 1;
|
||||
|
||||
my $not_found_retries = {};
|
||||
my $max_retries = $ENV{MAX_RETRIES} || 10;
|
||||
|
||||
GetOptions(
|
||||
'h|help' => \$help,
|
||||
'b|batch_size=s' => \$batch_size
|
||||
'm|max-processes=i' => \$max_processes,
|
||||
'h|help' => \$help,
|
||||
'b|batch_size=s' => \$batch_size
|
||||
) || pod2usage(1);
|
||||
|
||||
pod2usage(0) if $help;
|
||||
|
|
@ -95,6 +108,8 @@ if ( $notification_method eq 'STOMP' ) {
|
|||
warn $error if $error;
|
||||
}
|
||||
|
||||
my $pm = Parallel::ForkManager->new($max_processes);
|
||||
|
||||
if ($conn) {
|
||||
|
||||
# FIXME cf note in Koha::BackgroundJob about $namespace
|
||||
|
|
@ -121,6 +136,7 @@ while (1) {
|
|||
if ( !defined $frame ) {
|
||||
|
||||
# maybe log connection problems
|
||||
$pm->reap_finished_children;
|
||||
next; # will reconnect automatically
|
||||
}
|
||||
|
||||
|
|
@ -195,8 +211,13 @@ while (1) {
|
|||
|
||||
push @jobs, $job;
|
||||
if ( @jobs >= $batch_size || !$conn->can_read( { timeout => '0.1' } ) ) {
|
||||
commit(@jobs);
|
||||
my $commit_jobs = dclone( \@jobs );
|
||||
@jobs = ();
|
||||
$pm->start and next;
|
||||
srand(); # ensure each child process begins with a new seed
|
||||
commit( @{$commit_jobs} );
|
||||
$pm->finish;
|
||||
sleep 10;
|
||||
}
|
||||
|
||||
} else {
|
||||
|
|
@ -208,10 +229,15 @@ while (1) {
|
|||
{ status => 'new', queue => 'elastic_index' },
|
||||
{ rows => $batch_size }
|
||||
)->as_list;
|
||||
commit(@jobs);
|
||||
my $commit_jobs = dclone( \@jobs );
|
||||
@jobs = ();
|
||||
$pm->start and next;
|
||||
srand(); # ensure each child process begins with a new seed
|
||||
commit( @{$commit_jobs} );
|
||||
$pm->finish;
|
||||
sleep 10;
|
||||
}
|
||||
$pm->reap_finished_children;
|
||||
|
||||
}
|
||||
$conn->disconnect;
|
||||
|
|
|
|||
Loading…
Reference in New Issue