Skip to content

fixed getTasksToExecCommand_JEDI to avoid contention by locking tasks separately #629

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
tmaeno merged 1 commit into from
Nov 24, 2025
Merged

fixed getTasksToExecCommand_JEDI to avoid contention by locking tasks separately #629

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 20 additions & 8 deletions pandaserver/taskbuffer/db_proxy_mods/task_complex_module.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4411,8 +4411,6 @@ def getTasksToExecCommand_JEDI(self, vo, prodSourceLabel):
if varMap[":status"] in ["dummy", "paused"]:
continue
self.conn.begin()
# FIXME
# varMap[':timeLimit'] = naive_utcnow() - datetime.timedelta(hours=1)
varMap[":timeLimit"] = naive_utcnow() - datetime.timedelta(minutes=5)
sqlOrpS = "SELECT jediTaskID,errorDialog,oldStatus "
sqlOrpS += "FROM {0}.JEDI_Tasks tabT,{0}.JEDI_AUX_Status_MinTaskID tabA ".format(panda_config.schemaJEDI)
Expand All @@ -4424,24 +4422,38 @@ def getTasksToExecCommand_JEDI(self, vo, prodSourceLabel):
if prodSourceLabel not in [None, "any"]:
sqlOrpS += "AND prodSourceLabel=:prodSourceLabel "
varMap[":prodSourceLabel"] = prodSourceLabel
sqlOrpS += "FOR UPDATE "
tmpLog.debug(sqlOrpS + comment + str(varMap))
self.cur.execute(sqlOrpS + comment, varMap)
resList = self.cur.fetchall()
# commit
if not self._commit():
raise RuntimeError("Commit error")
# update modtime to avoid immediate reattempts
sql_orphaned_lock = f"SELECT modificationtime FROM {panda_config.schemaJEDI}.JEDI_Tasks WHERE jediTaskID=:jediTaskID FOR UPDATE NOWAIT "
sqlOrpU = f"UPDATE {panda_config.schemaJEDI}.JEDI_Tasks SET modificationtime=CURRENT_DATE "
sqlOrpU += "WHERE jediTaskID=:jediTaskID "
for jediTaskID, comComment, oldStatus in resList:
varMap = {}
varMap[":jediTaskID"] = jediTaskID
self.conn.begin()
# lock
try:
varMap = {}
varMap[":jediTaskID"] = jediTaskID
self.cur.execute(sql_orphaned_lock + comment, varMap)
except Exception as e:
tmpLog.debug(f"skip locked jediTaskID={jediTaskID} due to {str(e)}")
# commit
if not self._commit():
raise RuntimeError("Commit error")
continue
# update
tmpLog.debug(sqlOrpU + comment + str(varMap))
self.cur.execute(sqlOrpU + comment, varMap)
nRow = self.cur.rowcount
if nRow == 1 and jediTaskID not in retTaskIDs:
retTaskIDs[jediTaskID] = {"command": commandStr, "comment": comComment, "oldStatus": oldStatus}
# commit
if not self._commit():
raise RuntimeError("Commit error")
# commit
if not self._commit():
raise RuntimeError("Commit error")
# read clob
sqlCC = f"SELECT comm_parameters FROM {panda_config.schemaDEFT}.PRODSYS_COMM WHERE comm_task=:comm_task "
for jediTaskID in retTaskIDs.keys():
Expand Down