From e47dcf26618f289e70007b8543ddf1be7ea105af Mon Sep 17 00:00:00 2001 From: zaihuaji Date: Thu, 24 Sep 2026 13:07:43 -0500 Subject: [PATCH] report exactly one email per run and none for an empty run; bump version to 3.0.13 Park the progress report in dscheck.einfo instead of sending it, so it goes out exactly when PBS kills the job off its walltime, and have the final report send directly and drop the parked one - left behind it would be mailed a second time as a stale report and would hold the finished check back from being purged. A forked child could also reach quit_dsquasar() and report for the whole run, emailing a duplicate, clearing the parent's parked report and failing the shared check record while the parent was still working. Drop the report altogether when nothing was processed and nothing failed, for every backup action rather than the transfer one alone. Co-Authored-By: Claude Opus 4.6 --- pyproject.toml | 2 +- src/rda_python_dsquasar/dsquasar.py | 66 ++++++++++++++++++++--------- 2 files changed, 46 insertions(+), 22 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index c9d0784..8f693f6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "rda_python_dsquasar" -version = "3.0.12" +version = "3.0.13" authors = [ { name="Zaihua Ji", email="zji@ucar.edu" }, ] diff --git a/src/rda_python_dsquasar/dsquasar.py b/src/rda_python_dsquasar/dsquasar.py index b8a7840..887d06e 100644 --- a/src/rda_python_dsquasar/dsquasar.py +++ b/src/rda_python_dsquasar/dsquasar.py @@ -228,6 +228,12 @@ def start_actions(self): if acts&self.TARACT: self.build_tarfile_action() if acts&self.CINACT: self.create_infile_action() if self.dstart and acts&self.TBACTS == self.TBACTS: self.globus_transfer_action() + # the queues were empty, so nothing was processed and nothing failed. no dscheck + # record is registered for such a run either, so drop the report rather than mail + # one about 0 files. a parked progress report still forces one, to clean it up + if not (self.PGBACK['bckcnt'] or self.PGBACK['errcnt'] or + self.PGLOG['ERRCNT'] or self.PGBACK['einfo']): + self.PGBACK['doemail'] = 0 if self.PGBACK['doemail']: amsg = self.ACTMSG[self.PGBACK['action']] bmsg = self.BACKMSG[self.PGBACK['backflag']] if self.PGBACK['backflag'] else 'backup' @@ -343,7 +349,7 @@ def build_tarfile_action(self): self.backup_dataset_infiles(dsfiles, 'D') # gather and transfer tar files to Globus Quasar Servers - def globus_transfer_action(self): + def globus_transfer_action(self): dsfiles = {'B' : {}, 'D' : {}} fcnt = self.gather_dataset_tarfiles(dsfiles) if fcnt and self.dstart: @@ -752,6 +758,11 @@ def backup_dataset_tarfiles(self, dsfiles, backflag): # wait all child processes finish and then quit the main program def quit_dsquasar(self, qinfo, msg = None): + # a forked child reaches here through its own errcnt, which is a copy of the parent's + # made at the fork. only the parent reports for the run: a child reporting would email + # a second report, drop the progress report parked in dscheck.einfo by the parent, and + # mark the shared check record failed while the parent is still working + if self.PGSIG['PPID'] > 1: sys.exit(1) if self.PGBACK['mproc'] > 1: self.check_child(None, 0, self.LOGWRN, 1) if qinfo: if 'dslocks' in qinfo and qinfo['dslocks']: @@ -798,30 +809,43 @@ def set_walltime_deadline(self): self.PGLOG['CURBID'], self.seconds_to_string_time(self.WALLTIME), self.seconds_to_string_time(self.MAXRUNTIME)), self.LOGWRN) - # email a report for a run under dscheck control. the report is normally sent right - # away, but is cached into dscheck.einfo instead if a progress report is cached there - # already, so that the final report replaces it; the dscheck daemon sends and clears - # einfo once the check record is unlocked. + # email a report for a run under dscheck control. the progress report is parked in + # dscheck.einfo and the final report is sent right away. the run holds the check lock + # while it works, and the dscheck daemon only mails unlocked records, so a parked + # report goes out exactly when the check is unlocked - right after PBS kills the job + # off its walltime, in the same pass that resubmits it. + # a record with a non-empty einfo is skipped by both the start pass and the purge pass + # of the daemon, so the final report must drop the parked one: left behind it would be + # mailed a second time as a stale progress report, and would hold the finished check + # back from being purged. def report_dscheck_email(self, title, cache = 0): cnd = "cindex = {}".format(self.PGLOG['DSCHECK']['cindex']) - if not (cache or self.PGBACK['einfo']): - return self.build_customized_email("dscheck", "einfo", cnd, title, self.LOGWRN) - msg = self.get_email() - if not msg: return self.FAILURE - sender = self.PGLOG['CURUID'] + "@ucar.edu" - receiver = self.PGLOG['EMLADDR'] if self.PGLOG['EMLADDR'] else sender - if receiver.find(sender) < 0: self.add_carbon_copy(sender, 1) - ebuf = "From: {}\nTo: {}\n".format(sender, receiver) - if self.PGLOG['CCDADDR']: ebuf += "Cc: {}\n".format(self.PGLOG['CCDADDR']) - ebuf += "Subject: {}!\n\n{}\n".format(title, msg) - self.PGBACK['einfo'] = 1 - return self.cache_customized_email("dscheck", "einfo", cnd, ebuf, self.LOGWRN) + if cache: + msg = self.get_email() + if not msg: return self.FAILURE + sender = self.PGLOG['CURUID'] + "@ucar.edu" + receiver = self.PGLOG['EMLADDR'] if self.PGLOG['EMLADDR'] else sender + if receiver.find(sender) < 0: self.add_carbon_copy(sender, 1) + ebuf = "From: {}\nTo: {}\n".format(sender, receiver) + if self.PGLOG['CCDADDR']: ebuf += "Cc: {}\n".format(self.PGLOG['CCDADDR']) + ebuf += "Subject: {}!\n\n{}\n".format(title, msg) + self.PGBACK['einfo'] = 1 # report the progress only once per run + return self.cache_customized_email("dscheck", "einfo", cnd, ebuf, self.LOGWRN) + estat = self.build_customized_email("dscheck", "einfo", cnd, title, self.LOGWRN) + # a failed send has already replaced the parked report via cache_customized_email() + if estat == self.SUCCESS and self.PGBACK['einfo']: + if self.pgexec("UPDATE dscheck set einfo = NULL WHERE " + cnd, self.LOGWRN): + self.PGBACK['einfo'] = 0 + else: + self.pglog("Cannot clean the progress report of {}; it is emailed again".format(cnd), self.LOGWRN) + return estat # guard the long PBS batch jobs (-A 2/3/4/6) against the walltime: once past MAXRUNTIME - # the run keeps going, but a progress report is cached into dscheck.einfo so an email is - # still sent, by the dscheck daemon, if PBS kills the job at the walltime. the final - # report replaces it if the run does finish in time. the live email buffers are saved - # and put back, so the final report still carries everything logged before the cutoff. + # the run keeps going, but a progress report is parked in dscheck.einfo so that + # something is reported even if PBS kills the job at the walltime. the final report + # follows and drops the parked one if the run does finish in time. the live email + # buffers are saved and put back, so the final report still carries everything logged + # before the cutoff. # both queue depths are reported rather than the current phase's: -A 3 and -A 6 run the # build and the transfer phase in one run, and -A 3 tars each input file right after # creating it, so its status 'N' count stays near zero while that is the busy phase.