Bugzilla – Attachment 162952 Details for
Bug 35920
Centralize code from workers
Home
|
New
|
Browse
|
Search
|
[?]
|
Reports
|
Help
|
New Account
|
Log In
[x]
|
Forgot Password
Login:
[x]
[patch]
Bug 35920: Use BackgroundWorker in es_indexer_daemon
Bug-35920-Use-BackgroundWorker-in-esindexerdaemon.patch (text/plain), 7.94 KB, created by
Marcel de Rooy
on 2024-03-08 11:03:30 UTC
(
hide
)
Description:
Bug 35920: Use BackgroundWorker in es_indexer_daemon
Filename:
MIME Type:
Creator:
Marcel de Rooy
Created:
2024-03-08 11:03:30 UTC
Size:
7.94 KB
patch
obsolete
>From 1a48cbe6bb8bcfd95e5e4a66193f8b5d18d92803 Mon Sep 17 00:00:00 2001 >From: Marcel de Rooy <m.de.rooy@rijksmuseum.nl> >Date: Fri, 8 Mar 2024 10:55:36 +0000 >Subject: [PATCH] Bug 35920: Use BackgroundWorker in es_indexer_daemon >Content-Type: text/plain; charset=utf-8 > >This needs incorporating the batch_size in the main loop. > >Test plan: >Test both regular worker and es indexer daemon here. >--- > Koha/BackgroundWorker.pm | 17 +++- > misc/workers/background_jobs_worker.pl | 15 +-- > misc/workers/es_indexer_daemon.pl | 124 ++++--------------------- > 3 files changed, 40 insertions(+), 116 deletions(-) > >diff --git a/Koha/BackgroundWorker.pm b/Koha/BackgroundWorker.pm >index 760ab8fe2a..6e1c132c0e 100644 >--- a/Koha/BackgroundWorker.pm >+++ b/Koha/BackgroundWorker.pm >@@ -50,6 +50,7 @@ Koha::BackgroundWorker - Centralized code for background worker scripts > > Main method to start the worker. > Available parameters: >+ - batch_size > - callback > - queues > - max_processes >@@ -90,6 +91,8 @@ sub run { > } > > ### The main loop >+ my @jobs; >+ my $lastseen = time; > while (1) { > if ($conn) { > my $frame = $conn->receive_frame( { timeout => $self->{mq_timeout} } ); >@@ -147,10 +150,15 @@ sub run { > } > $conn->ack( { frame => $frame } ); > >- $pm->start and next; >- srand(); # ensure each child process begins with a new seed >- &$callback( $job, $args ); >- $pm->finish; >+ push @jobs, [ $job, $args ]; >+ if ( @jobs >= $self->{batch_size} || time > $lastseen ) { >+ my @commits = @jobs; @jobs = (); >+ $lastseen = time; >+ $pm->start and next; >+ srand(); # ensure each child process begins with a new seed >+ &$callback(@commits); >+ $pm->finish; >+ } > > } else { > my $jobs = Koha::BackgroundJobs->search( { status => 'new', queue => \@queues } ); >@@ -183,6 +191,7 @@ sub run { > > sub _init { > my ($self) = @_; >+ $self->{batch_size} //= 1; > $self->{max_processes} ||= C4::Context->config('background_jobs_worker')->{max_processes} > if C4::Context->config('background_jobs_worker'); > $self->{max_processes} ||= 1; >diff --git a/misc/workers/background_jobs_worker.pl b/misc/workers/background_jobs_worker.pl >index fd28dd2b1a..c731c9e199 100755 >--- a/misc/workers/background_jobs_worker.pl >+++ b/misc/workers/background_jobs_worker.pl >@@ -82,10 +82,13 @@ Koha::BackgroundWorker->run( > ); > > sub process_job { >- my ( $job, $args ) = @_; >- try { >- $job->process( $args ); >- } catch { >- $job->status('failed')->store; >- }; >+ my ( @jobs ) = @_; >+ foreach my $elt (@jobs) { >+ my ( $job, $args ) = @$elt; >+ try { >+ $job->process( $args ); >+ } catch { >+ $job->status('failed')->store; >+ }; >+ } > } >diff --git a/misc/workers/es_indexer_daemon.pl b/misc/workers/es_indexer_daemon.pl >index 9e900b7195..6fbef67433 100755 >--- a/misc/workers/es_indexer_daemon.pl >+++ b/misc/workers/es_indexer_daemon.pl >@@ -56,136 +56,48 @@ use Try::Tiny; > use Pod::Usage; > use Getopt::Long; > use List::MoreUtils qw( natatime ); >-use Time::HiRes; > > use C4::Context; >-use Koha::Logger; > use Koha::BackgroundJobs; >+use Koha::Logger; > use Koha::SearchEngine; > use Koha::SearchEngine::Indexer; > >- >-my ( $help, $batch_size ); >- >-my $not_found_retries = {}; >-my $max_retries = $ENV{MAX_RETRIES} || 10; >- >+my ( $max_processes, $help, $batch_size ); > 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; >- > $batch_size //= 10; > > warn "Not using Elasticsearch" unless C4::Context->preference('SearchEngine') eq 'Elasticsearch'; >- >-my $logger = Koha::Logger->get({ interface => 'worker' }); >- >-my $conn; >-try { >- $conn = Koha::BackgroundJob->connect; >-} catch { >- warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; >-}; >- >-if ( $conn ) { >- # FIXME cf note in Koha::BackgroundJob about $namespace >- my $namespace = C4::Context->config('memcached_namespace'); >- $conn->subscribe( >- { >- destination => sprintf( "/queue/%s-%s", $namespace, 'elastic_index' ), >- ack => 'client', >- 'prefetch-count' => 1, >- } >- ); >-} > my $biblio_indexer = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::BIBLIOS_INDEX } ); > my $auth_indexer = Koha::SearchEngine::Indexer->new( { index => $Koha::SearchEngine::AUTHORITIES_INDEX } ); > my $config = $biblio_indexer->get_elasticsearch_params; > my $at_a_time = $config->{chunk_size} // 5000; >- >-my @jobs = (); >- >-while (1) { >- >- if ( $conn ) { >- my $frame = $conn->receive_frame; >- if ( !defined $frame ) { >- # maybe log connection problems >- next; # will reconnect automatically >- } >- >- my $args = try { >- my $body = $frame->body; >- decode_json($body); # TODO Should this be from_json? Check utf8 flag. >- } catch { >- Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame not processed - %s", $_); >- return; >- }; >- >- unless ( $args ) { >- Koha::Logger->get({ interface => 'worker' })->warn(sprintf "Frame does not have correct args, ignoring it"); >- $conn->nack( { frame => $frame, requeue => 'false' } ); >- next; >- } >- >- my $job = Koha::BackgroundJobs->find( $args->{job_id} ); >- >- if ( $job && $job->status ne 'new' ) { >- Koha::Logger->get( { interface => 'worker' } ) >- ->warn( sprintf "Job %s has wrong status %s", $args->{job_id}, $job->status ); >- >- # nack without requeue, we do not want to process this frame again >- $conn->nack( { frame => $frame, requeue => 'false' } ); >- next; >- } >- >- unless ($job) { >- $not_found_retries->{ $args->{job_id} } //= 0; >- if ( ++$not_found_retries->{ $args->{job_id} } >= $max_retries ) { >- Koha::Logger->get( { interface => 'worker' } ) >- ->warn( sprintf "Job %s not found, no more retry", $args->{job_id} ); >- >- # nack without requeue, we do not want to process this frame again >- $conn->nack( { frame => $frame, requeue => 'false' } ); >- next; >- } >- >- Koha::Logger->get( { interface => 'worker' } ) >- ->debug( sprintf "Job %s not found, will retry later", $args->{job_id} ); >- >- # nack to force requeue >- $conn->nack( { frame => $frame, requeue => 'true' } ); >- Time::HiRes::sleep(0.5); >- next; >- } >- $conn->ack( { frame => $frame } ); >- >- push @jobs, $job; >- if ( @jobs >= $batch_size || !$conn->can_read( { timeout => '0.1' } ) ) { >- commit(@jobs); >- @jobs = (); >- } >- >- } else { >- @jobs = Koha::BackgroundJobs->search( >- { status => 'new', queue => 'elastic_index' } )->as_list; >- commit(@jobs); >- @jobs = (); >- sleep 10; >+my $lastseen = time; >+ >+Koha::BackgroundWorker->run( >+ { >+ max_processes => $max_processes // $ENV{MAX_PROCESSES}, >+ mq_timeout => $ENV{MQ_TIMEOUT}, >+ max_retries => $ENV{MAX_RETRIES}, >+ callback => \&commit, >+ queues => [ 'elastic_index' ], > } >- >-} >-$conn->disconnect; >+); > > sub commit { > my (@jobs) = @_; >+ # Extract jobs only from passed arrayrefs >+ @jobs = map { $_->[0] } @jobs; > > my @bib_records; > my @auth_records; > >+ my $logger = Koha::Logger->get({ interface => 'worker' }); > my $jobs = Koha::BackgroundJobs->search( { id => [ map { $_->id } @jobs ] }); > # Start > $jobs->update({ >-- >2.30.2
You cannot view the attachment while viewing its details because your browser does not support IFRAMEs.
View the attachment on a separate page
.
View Attachment As Diff
View Attachment As Raw
Actions:
View
|
Diff
|
Splinter Review
Attachments on
bug 35920
:
161544
|
162951
|
162952