From 7c9be7e08ea416a3fbaa0d7a288c263e695b67d6 Mon Sep 17 00:00:00 2001
From: Kyle M Hall <kyle@bywatersolutions.com>
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