Only return a subset of jobs ready to run per user when querying for jobs ready to run

[natefoo]

Observing a problem we currently have on usegalaxy.org, where there are 21,000 jobs in the new state, but very few of them are dispatched on each iteration of the handler monitor thread because those jobs are mostly owned by a small number of users and the concurrency limits are preventing their new jobs from dispatching. Every iteration of the loop the monitor thread will resolve destinations (i.e. run them through your dynamic rules...), check limits, etc. for all 21k of these jobs and then defer the overwhelming majority of them until the next loop. All of that takes a lot of time. In the meantime, any new jobs submitted by other users will not even be dispatched to a runner plugin until the end of the next iteration of the loop.

So this change uses windowing/partitioning to only return the top ready_window_size (default: 100) jobs per user on each iteration of the monitor thread.

A few things to be aware of:

• Anonymous users are treated as a single user by this algorithm. This is probably fixable by having a separate query for anonymous jobs if this becomes a problem (we could even do the rank/window by session) since one extra query is unlikely to be problematic from a performance standpoint.

• This applies to jobs whose inputs are terminal - jobs whose inputs are not terminal are already filtered out of the query.

• If you are using per-destination (or per-destination-tag) limits, small window sizes could prevent a user's jobs on destinations where they are under the limit from running if their oldest jobs are on destinations where they are over the limit. For example let's say you have set the window size to 20, you have configured 2 job destinations, multicore and singlecore each with a limit of 10 concurrent jobs per user, and you have 10 jobs running on the multicore destination but 0 jobs running on singlecore. You also have 100 independent jobs in the new state whose inputs are all ready, the first 40 of which are destined for multicore and the last 60 of which are destined for singlecore.

  Your singlecore jobs will have to wait until there are only 19 multicore jobs waiting to be dispatched before any of the singlecore jobs will start being dispatched, even though you are under your limits there.

  You can mitigate this by not setting ready_window_size too low, or avoid it entirely by having separate job handlers (see the handler attrib for <tool> tags in the job conf) for these destinations.

• Although you can use this as a form of concurrency limits, I think it's better to use this as a safety measure so one person having 10k pending jobs doesn't destroy that handler for everyone else. If you need concurrency limits, Galaxy already does that in a much more fine-grained manner.

I thought about making this configurable such that the old non-windowed query would be the default but this seems to work fine even with SQLite.

As for performance: the regular job readiness query for one handler on usegalaxy.org currently returns 6902 rows, with this query and ready_window_size set to 20, it returns 374 rows, with no hit on query performance (both run in ~320ms).

[hexylena]

Sounds very related to my issue from the other day, #11130

[jmchilton]

xref #11228 (comment)

We should stop updating this file and update the YAML version instead.

[natefoo]

While I agree, we need to update all of the documentation and move the YAML over to the sample dir, update the Ansible role, etc. AFAIK I am the only person using it so it seems a bit premature to dump the XML without at least having the docs to help people transition.

[jmchilton]

Let me rephrase then - please update both sources - I put a lot of effort into migrating this documentation to YAML. If you are only going to document it in one place - I would prefer it to be the YAML because that will be the source of truth at some point (you promised to do the cutover for me like two years ago 😉).

I approved the PR though, feel free to ignore me.

[natefoo]

Oh ok - I'll do that.

[natefoo]

The updating both thing, not the ignoring. ;D

[mvdbeek]

Do you want to target 21.01 for the smorgasboard event ? It's not like we can really test this against test ...

[natefoo]

No, I don't feel comfortable getting this into 21.01, I'll cherry pick to the usegalaxy branch instead.

[natefoo]

@hexylena I'll also say that because handler assignment is random and usegalaxy.eu runs like what, 20 handlers? That this is not going to work exactly as you might be hoping as a form of concurrency limiting. Each handler will still be able to dispatch ready_window_size of its assigned jobs per user per iteration. That said, it will do more than nothing.

[hexylena]

8. But, ok, yeah, makes sense.

[natefoo]

The output of the timed-out test won't load for me, did I break anything here?

[mvdbeek]

If you click on the extended menu you can download the raw logs:

2021-02-13T04:30:55.1213145Z 2021-02-13 04:30:55,120 ERROR [galaxy.jobs.handler] Exception in monitor_step
2021-02-13T04:30:55.1213822Z Traceback (most recent call last):
2021-02-13T04:30:55.1215098Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
2021-02-13T04:30:55.1216001Z     cursor, statement, parameters, context
2021-02-13T04:30:55.1217122Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 609, in do_execute
2021-02-13T04:30:55.1218006Z     cursor.execute(statement, parameters)
2021-02-13T04:30:55.1219036Z sqlite3.OperationalError: near "(": syntax error
2021-02-13T04:30:55.1219467Z
2021-02-13T04:30:55.1219964Z The above exception was the direct cause of the following exception:
2021-02-13T04:30:55.1219964Z The above exception was the direct cause of the following exception:
2021-02-13T04:30:55.1220378Z
2021-02-13T04:30:55.1220882Z Traceback (most recent call last):
2021-02-13T04:30:55.1221809Z   File "/home/runner/work/galaxy/galaxy/galaxy root/lib/galaxy/jobs/handler.py", line 281, in __monitor
2021-02-13T04:30:55.1222463Z     self.__monitor_step()
2021-02-13T04:30:55.1223374Z   File "/home/runner/work/galaxy/galaxy/galaxy root/lib/galaxy/jobs/handler.py", line 299, in __monitor_step
2021-02-13T04:30:55.1224106Z     self.__handle_waiting_jobs()
2021-02-13T04:30:55.1224887Z   File "/home/runner/work/galaxy/galaxy/galaxy root/lib/galaxy/jobs/handler.py", line 344, in __handle_waiting_jobs
2021-02-13T04:30:55.1226099Z     .filter(ranked.c.rank <= self.app.job_config.handler_ready_window_size).all()
2021-02-13T04:30:55.1227552Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 3373, in all
2021-02-13T04:30:55.1228219Z     return list(self)
2021-02-13T04:30:55.1229131Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 3535, in __iter__
2021-02-13T04:30:55.1229876Z     return self._execute_and_instances(context)
2021-02-13T04:30:55.1230942Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/orm/query.py", line 3560, in _execute_and_instances
2021-02-13T04:30:55.1231878Z     result = conn.execute(querycontext.statement, self._params)
2021-02-13T04:30:55.1233031Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1011, in execute
2021-02-13T04:30:55.1233856Z     return meth(self, multiparams, params)
2021-02-13T04:30:55.1234958Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/sql/elements.py", line 298, in _execute_on_connection
2021-02-13T04:30:55.1235921Z     return connection._execute_clauseelement(self, multiparams, params)
2021-02-13T04:30:55.1237145Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1130, in _execute_clauseelement
2021-02-13T04:30:55.1237894Z     distilled_params,
2021-02-13T04:30:55.1238874Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1317, in _execute_context
2021-02-13T04:30:55.1239673Z     e, statement, parameters, cursor, context
2021-02-13T04:30:55.1240905Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1511, in _handle_dbapi_exception
2021-02-13T04:30:55.1241835Z     sqlalchemy_exception, with_traceback=exc_info[2], from_=e
2021-02-13T04:30:55.1242940Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/util/compat.py", line 182, in raise_
2021-02-13T04:30:55.1243677Z     raise exception
2021-02-13T04:30:55.1244755Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/base.py", line 1277, in _execute_context
2021-02-13T04:30:55.1245548Z     cursor, statement, parameters, context
2021-02-13T04:30:55.1246588Z   File "/home/runner/.planemo/gx_venv_3.7/lib/python3.7/site-packages/sqlalchemy/engine/default.py", line 609, in do_execute
2021-02-13T04:30:55.1247396Z     cursor.execute(statement, parameters)
2021-02-13T04:30:55.1248302Z sqlalchemy.exc.OperationalError: (sqlite3.OperationalError) near "(": syntax error
2021-02-13T04:30:55.1253388Z [SQL: SELECT job.id AS job_id, job.create_time AS job_create_time, job.update_time AS job_update_time, job.history_id AS job_history_id, job.library_folder_id AS job_library_folder_id, job.tool_id AS job_tool_id, job.tool_version AS job_tool_version, job.galaxy_version AS job_galaxy_version, job.dynamic_tool_id AS job_dynamic_tool_id, job.state AS job_state, job.info AS job_info, job.copied_from_job_id AS job_copied_from_job_id, job.command_line AS job_command_line, job.dependencies AS job_dependencies, job.job_messages AS job_job_messages, job.param_filename AS job_param_filename, job.runner_name AS job_runner_name_1, job.job_stdout AS job_job_stdout, job.job_stderr AS job_job_stderr, job.tool_stdout AS job_tool_stdout, job.tool_stderr AS job_tool_stderr, job.exit_code AS job_exit_code, job.traceback AS job_traceback, job.session_id AS job_session_id, job.user_id AS job_user_id, job.job_runner_name AS job_job_runner_name, job.job_runner_external_id AS job_job_runner_external_id, job.destination_id AS job_destination_id, job.destination_params AS job_destination_params, job.object_store_id AS job_object_store_id, job.imported AS job_imported, job.params AS job_params, job.handler AS job_handler, post_job_action_association_1.id AS post_job_action_association_1_id, post_job_action_association_1.job_id AS post_job_action_association_1_job_id, post_job_action_association_1.post_job_action_id AS post_job_action_association_1_post_job_action_id
2021-02-13T04:30:55.1261193Z FROM job JOIN (SELECT job.id AS id, job.create_time AS create_time, job.update_time AS update_time, job.history_id AS history_id, job.library_folder_id AS library_folder_id, job.tool_id AS tool_id, job.tool_version AS tool_version, job.galaxy_version AS galaxy_version, job.dynamic_tool_id AS dynamic_tool_id, job.state AS state, job.info AS info, job.copied_from_job_id AS copied_from_job_id, job.command_line AS command_line, job.dependencies AS dependencies, job.job_messages AS job_messages, job.param_filename AS param_filename, job.runner_name AS runner_name, job.job_stdout AS job_stdout, job.job_stderr AS job_stderr, job.tool_stdout AS tool_stdout, job.tool_stderr AS tool_stderr, job.exit_code AS exit_code, job.traceback AS traceback, job.session_id AS session_id, job.user_id AS user_id, job.job_runner_name AS job_runner_name, job.job_runner_external_id AS job_runner_external_id, job.destination_id AS destination_id, job.destination_params AS destination_params, job.object_store_id AS object_store_id, job.imported AS imported, job.params AS params, job.handler AS handler, rank() OVER (PARTITION BY job.user_id ORDER BY job.id) AS rank
2021-02-13T04:30:55.1264982Z FROM job LEFT OUTER JOIN galaxy_user ON galaxy_user.id = job.user_id
2021-02-13T04:30:55.1265636Z WHERE job.state = ? AND job.handler = ? AND job.id NOT IN (SELECT job.id
2021-02-13T04:30:55.1266879Z FROM job JOIN job_to_input_dataset ON job.id = job_to_input_dataset.job_id JOIN history_dataset_association ON history_dataset_association.id = job_to_input_dataset.dataset_id JOIN dataset ON dataset.id = history_dataset_association.dataset_id
2021-02-13T04:30:55.1268153Z WHERE job.state = ? AND dataset.state IN (?, ?, ?, ?, ?)) AND job.id NOT IN (SELECT job.id
2021-02-13T04:30:55.1269553Z FROM job JOIN job_to_input_library_dataset ON job.id = job_to_input_library_dataset.job_id JOIN library_dataset_dataset_association ON library_dataset_dataset_association.id = job_to_input_library_dataset.ldda_id JOIN dataset ON dataset.id = library_dataset_dataset_association.dataset_id
2021-02-13T04:30:55.1272658Z WHERE job.state = ? AND dataset.state IN (?, ?, ?, ?, ?)) ORDER BY job.id) AS anon_1 ON job.id = anon_1.id LEFT OUTER JOIN post_job_action_association AS post_job_action_association_1 ON job.id = post_job_action_association_1.job_id
2021-02-13T04:30:55.1273689Z WHERE anon_1.rank <= ?]
2021-02-13T04:30:55.1276092Z [parameters: (<states.NEW: 'new'>, 'main', <states.NEW: 'new'>, 'new', 'upload', 'queued', 'running', 'setting_metadata', <states.NEW: 'new'>, 'new', 'upload', 'queued', 'running', 'setting_metadata', 100)]
2021-02-13T04:30:55.1277500Z (Background on this error at: http://sqlalche.me/e/13/e3q8)

[natefoo]

Thanks @jmchilton @mvdbeek !

Templating that query and manually running it on a Galaxy sqlite database using sqlite3 on the command line I do indeed get the same error message. So then I tried to determine what is different about that query than the one that is generated normally and works fine with SQLite for the other tests, so I grabbed the query using database_engine_option_echo and the queries are identical. The only thing I can figure is the sqlite3 lib version used by that test is older and doesn't support the window syntax - CPython's sqlite3 lib is a C extension that is dynamically linked to libsqlite3 at runtime.

I am thinking to work around this I will just stick a branch in there to use the un-windowed query when the engine is SQLite, even if it may work with newer versions of SQLite, because we'd never need this functionality on SQLite anyway.