Page MenuHomeSoftware Heritage
Differential D5956 Diff 21746 swh/scheduler/backend.py

Changeset View

Standalone View

View Options

swh/scheduler/backend.py

Show First 20 Lines • Show All 340 Lines • ▼ Show 20 Lines ) -> List[ListedOrigin]:
timestamp = utcnow() timestamp = utcnow()
origin_select_cols = ", ".join(ListedOrigin.select_columns()) origin_select_cols = ", ".join(ListedOrigin.select_columns())
query_args: List[Any] = [] query_args: List[Any] = []
where_clauses = [] where_clauses = []
# list of (name, query) handled as CTEs before the main query
common_table_expressions: List[Tuple[str, str]] = []
# "NOT enabled" = the lister said the origin no longer exists # "NOT enabled" = the lister said the origin no longer exists
where_clauses.append("enabled") where_clauses.append("enabled")
# Only schedule visits of the given type # Only schedule visits of the given type
where_clauses.append("visit_type = %s") where_clauses.append("visit_type = %s")
query_args.append(visit_type) query_args.append(visit_type)
if scheduled_cooldown: if scheduled_cooldown:
▲ Show 20 Lines • Show All 62 Lines • ▼ Show 20 Lines ) -> List[ListedOrigin]:
# order by decreasing visit lag # order by decreasing visit lag
order_by = """\ order_by = """\
listed_origins.last_update listed_origins.last_update
- GREATEST( - GREATEST(
origin_visit_stats.last_eventful, origin_visit_stats.last_eventful,
origin_visit_stats.last_uneventful origin_visit_stats.last_uneventful
) )
DESC DESC
""" """
ardumontUnsubmitted
Done
Inline Actions

We discussed the possibility to make this an enum with olasd (but that's a refactoring for another time).

ardumont: We discussed the possibility to make this an enum with olasd (but that's a refactoring for…
else: else:
raise UnknownPolicy(f"Unknown scheduling policy {policy}") raise UnknownPolicy(f"Unknown scheduling policy {policy}")
select_query = f""" # fmt: off
ardumontUnsubmitted
Done
Inline Actions
SELECT
- {origin_select_cols}
+ {origin_select_cols}, next_visit_queue_position
FROM
ardumont:
common_table_expressions.insert(0, ("selected_origins", f"""
ardumontUnsubmitted
Done
Inline Actions

we probably want those joins to be computed as well now.

ardumont: we probably want those joins to be computed as well now.
ardumontUnsubmitted
Done
Inline Actions

Drop this join which is no longer needed with the previous comment.

ardumont: Drop this join which is no longer needed with the previous comment.
SELECT SELECT
{origin_select_cols} {origin_select_cols}, next_visit_queue_position
FROM FROM
listed_origins listed_origins
LEFT JOIN LEFT JOIN
origin_visit_stats USING (url, visit_type) origin_visit_stats USING (url, visit_type)
WHERE WHERE
({") AND (".join(where_clauses)}) ({") AND (".join(where_clauses)})
ardumontUnsubmitted
Done
Inline Actions
elif policy == SCHEDULING_POLICY_ORIGINS_NO_LAST_UPDATE:
where_clauses.append("lo.last_update IS NULL")
- where_clauses.append(
- "origin_visit_stats.next_visit_queue_position IS NOT NULL"
- )
- where_clauses.append(
- "vsqp.position <= origin_visit_stats.next_visit_queue_position"
- )
- # TODO: Determine if it's actually what we want... this chooses an interval
- # of one day to fetch visits to schedule...
- where_clauses.append(
- "origin_visit_stats.next_visit_queue_position < vsqp.position + '1 day'"
- )
- order_by = "lo.last_update"
+ order_by = "origin_visit_stats.next_visit_queue_position nulls first"
else:

Use the limit parameter in the call to the method.

ardumont: Use the limit parameter in the call to the method.
ORDER BY ORDER BY
{order_by} {order_by}
LIMIT %s LIMIT %s
""" """))
# fmt: on
query_args.append(count) query_args.append(count)
query = f""" # fmt: off
WITH selected_origins AS ( common_table_expressions.append(("update_stats", """
{select_query}
),
update_stats AS (
INSERT INTO INSERT INTO
origin_visit_stats ( origin_visit_stats (url, visit_type, last_scheduled)
url, visit_type, last_scheduled
)
SELECT SELECT
url, visit_type, %s url, visit_type, %s
FROM FROM
selected_origins selected_origins
ON CONFLICT (url, visit_type) DO UPDATE ON CONFLICT (url, visit_type) DO UPDATE
SET last_scheduled = GREATEST( SET last_scheduled = GREATEST(
origin_visit_stats.last_scheduled, origin_visit_stats.last_scheduled,
EXCLUDED.last_scheduled EXCLUDED.last_scheduled
) )
"""))
# fmt: on
query_args.append(timestamp)
formatted_ctes = ",\n".join(
f"{name} AS (\n{cte}\n)" for name, cte in common_table_expressions
) )
query = f"""
WITH
{formatted_ctes}
SELECT SELECT
ardumontUnsubmitted
Done
Inline Actions
EXCLUDED.last_scheduled
)
+ ),
+ update_queue_position AS (
+ insert into queue_position(visit_type, position)
+ select visit_type, coalesce(max(next_visit_queue_position), now())
+ from selected_origins group by visit_type
+ on conflict(visit_type) do update
+ set position=greatest(
+ position, excluded.position
+ )
)
SELECT
*
FROM

This need to be a template and conditionned upon the right policy (the new one from this diff)

ardumont: This need to be a template and conditionned upon the right policy (the new one from this diff)
* {origin_select_cols}
ardumontUnsubmitted
Done
Inline Actions
SELECT
- *
+ {origin_select_cols}
FROM
ardumont:
FROM FROM
selected_origins selected_origins
""" """
query_args.append(timestamp)
cur.execute(query, tuple(query_args)) cur.execute(query, tuple(query_args))
return [ListedOrigin(**d) for d in cur] return [ListedOrigin(**d) for d in cur]
task_create_keys = [ task_create_keys = [
"type", "type",
"arguments", "arguments",
"next_run", "next_run",
▲ Show 20 Lines • Show All 575 Lines • Show Last 20 Lines
About · Developer information · Legal