From 7c9be7e08ea416a3fbaa0d7a288c263e695b67d6 Mon Sep 17 00:00:00 2001 From: Kyle M Hall Date: Tue, 3 Jan 2023 15:35:42 +0000 Subject: [PATCH] =?UTF-8?q?Bug=C2=A032558:=20Add=20ability=20for=20backgro?= =?UTF-8?q?und=5Fjobs=5Fworker.pl=20to=20process=20multiple=20jobs=20simul?= =?UTF-8?q?taneously=20up=20to=20a=20limit?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Right now background_jobs_worker.pl only processes jobs in serial. It would make sense to handle jobs in parallel up to a user definable limit. Test Plan: 1) Apply this patch 2) Stop background_jobs_worker.pl 3) Generate some background jobs by editing records, placing holds, etc 4) Watch processes in a new terminal: watch -n 0.1 'ps aux | grep background_jobs_worker.pl' 5) Run background_jobs_worker.pl 6) Note the multiple forked processes in the ps output --- C4/SIP/ILS.pm | 4 ++-- C4/SIP/Sip/MsgType.pm | 9 ++------- misc/background_jobs_worker.pl | 29 +++++++++++------------------ 3 files changed, 15 insertions(+), 27 deletions(-) diff --git a/C4/SIP/ILS.pm b/C4/SIP/ILS.pm index ff9ee74067..253a3fc2d0 100644 --- a/C4/SIP/ILS.pm +++ b/C4/SIP/ILS.pm @@ -196,7 +196,7 @@ sub test_cardnumber_compare { } sub checkin { - my ( $self, $item_id, $trans_date, $return_date, $current_loc, $item_props, $cancel, $account ) = @_; + my ( $self, $item_id, $trans_date, $return_date, $current_loc, $item_props, $cancel, $account, $no_block ) = @_; my $checked_in_ok = $account->{checked_in_ok}; my $cv_triggers_alert = $account->{cv_triggers_alert}; @@ -211,7 +211,7 @@ sub checkin { my $data; if ($item) { - $data = $circ->do_checkin( $current_loc, $return_date, $account ); + $data = $circ->do_checkin( $current_loc, $return_date, $account, $no_block ); } else { $circ->alert(1); diff --git a/C4/SIP/Sip/MsgType.pm b/C4/SIP/Sip/MsgType.pm index 7ba61959a3..0515da1db3 100644 --- a/C4/SIP/Sip/MsgType.pm +++ b/C4/SIP/Sip/MsgType.pm @@ -670,13 +670,8 @@ sub handle_checkin { $ils->check_inst_id( $inst_id, "handle_checkin" ); if ( $no_block eq 'Y' ) { - - # Off-line transactions, ick. - siplog( "LOG_WARNING", "received no-block checkin from terminal '%s' - no-block checkin not supported", $account->{id} ); - #FIXME We need to write the routine called below - #$status = $ils->checkin_no_block( $item_id, $trans_date, $return_date, $item_props, $cancel ); - #Until we do, lets just checkin the item - $status = $ils->checkin( $item_id, $trans_date, $return_date, $my_branch, $item_props, $cancel, $account ); + siplog( "LOG_WARNING", "received no-block checkin from terminal '%s'", $account->{id} ); + $status = $ils->checkin( $item_id, $trans_date, $return_date, $my_branch, $item_props, $cancel, $account, $no_block ); } else { $status = $ils->checkin( $item_id, $trans_date, $return_date, $my_branch, $item_props, $cancel, $account ); } diff --git a/misc/background_jobs_worker.pl b/misc/background_jobs_worker.pl index eb36f34f96..37dd328655 100755 --- a/misc/background_jobs_worker.pl +++ b/misc/background_jobs_worker.pl @@ -52,11 +52,14 @@ use JSON qw( decode_json ); use Try::Tiny qw( catch try ); use Pod::Usage; use Getopt::Long; +use Parallel::ForkManager; use Koha::BackgroundJobs; my ( $help, @queues ); +my $max_processes = 10; GetOptions( + 'm|max-processes=i' => \$max_processes, 'h|help' => \$help, 'queue=s' => \@queues, ) || pod2usage(1); @@ -74,6 +77,8 @@ try { warn sprintf "Cannot connect to the message broker, the jobs will be processed anyway (%s)", $_; }; +my $pm = Parallel::ForkManager->new($max_processes); + if ( $conn ) { # FIXME cf note in Koha::BackgroundJob about $namespace my $namespace = C4::Context->config('memcached_namespace'); @@ -96,32 +101,20 @@ while (1) { # It could work in a first step, but then we will want to handle job that will be created from the message received my $job = Koha::BackgroundJobs->find($args->{job_id}); - process_job( $job, $args ); + $pm->start and next; + $job->process( { job_id => $job->id, %$args } ); $conn->ack( { frame => $frame } ); # FIXME depending on success? + $pm->finish; } else { my $jobs = Koha::BackgroundJobs->search({ status => 'new', queue => \@queues }); while ( my $job = $jobs->next ) { + $pm->start and next; my $args = $job->json->decode($job->data); - process_job( $job, { job_id => $job->id, %$args } ); + $job->process( { job_id => $job->id, %$args } ); + $pm->finish; } sleep 10; } } $conn->disconnect; - -sub process_job { - my ( $job, $args ) = @_; - - my $pid; - if ( $pid = fork ) { - wait; - return; - } - - die "fork failed!" unless defined $pid; - - $job->process( $args ); - - exit; -} -- 2.30.2