Skip to content
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

Jump to bottom

cylc.task_pool: speed up queued db ops #1812

Merged
merged 2 commits into from
Apr 27, 2016
Merged

cylc.task_pool: speed up queued db ops #1812

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
a0f48e8
Speed up queued db ops
benfitzpatrick Apr 26, 2016
6bb3227
More speed up from review suggestion
benfitzpatrick Apr 27, 2016
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
52 changes: 29 additions & 23 deletions lib/cylc/task_pool.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1224,29 +1224,35 @@ def process_queued_db_ops(self):
"""Handle queued db operations for each task proxy."""
for itask in self.get_all_tasks():
# (runahead pool tasks too, to get new state recorders).
for table_name, db_inserts in sorted(itask.db_inserts_map.items()):
while db_inserts:
db_insert = db_inserts.pop(0)
db_insert.update({
"name": itask.tdef.name,
"cycle": str(itask.point),
})
if "submit_num" not in db_insert:
db_insert["submit_num"] = itask.submit_num
self.pri_dao.add_insert_item(table_name, db_insert)
self.pub_dao.add_insert_item(table_name, db_insert)

for table_name, db_updates in sorted(itask.db_updates_map.items()):
while db_updates:
set_args = db_updates.pop(0)
where_args = {
"cycle": str(itask.point), "name": itask.tdef.name}
if "submit_num" not in set_args:
where_args["submit_num"] = itask.submit_num
self.pri_dao.add_update_item(
table_name, set_args, where_args)
self.pub_dao.add_update_item(
table_name, set_args, where_args)
if any(itask.db_inserts_map.values()):
for table_name, db_inserts in sorted(
itask.db_inserts_map.items()):
while db_inserts:
db_insert = db_inserts.pop(0)
db_insert.update({
"name": itask.tdef.name,
"cycle": str(itask.point),
})
if "submit_num" not in db_insert:
db_insert["submit_num"] = itask.submit_num
self.pri_dao.add_insert_item(table_name, db_insert)
self.pub_dao.add_insert_item(table_name, db_insert)

if any(itask.db_updates_map.values()):
for table_name, db_updates in sorted(
itask.db_updates_map.items()):
while db_updates:
set_args = db_updates.pop(0)
where_args = {
"cycle": str(itask.point),
"name": itask.tdef.name
}
if "submit_num" not in set_args:
where_args["submit_num"] = itask.submit_num
self.pri_dao.add_update_item(
table_name, set_args, where_args)
self.pub_dao.add_update_item(
table_name, set_args, where_args)

# record any broadcast settings to be dumped out
bcast = BroadcastServer.get_inst()
Expand Down