Pg executor #84

gabriel-v · 2023-09-07T18:27:36Z

closes #78

If you want to accept this, I guess this would be the minimum left to do

add CLI subcommand to start executor worker process
add&fix type annotations
add instructions & docs sections for running
test internals of executor (send/receive to/from queue)
test a bunch of different / missing settings in the ini file
test executor with workflows

gregjohnso · 2024-02-06T23:20:45Z

@mattrasmus this would be very helpful for us. What is the approval mechanism here?

ernestum

Thanks for the great work on the alternative to an AWS executor @gabriel-v!

I am evaluating a number of pipeline/workflow solutions right now and redun is at the top of the list with the only catch being that there is no self-hosted executor solution.

In hope to get this feature upstream, I provide some code review. I also took the liberty to rebase it onto master and resolved the conflicts here (@gabriel-v feel free to pull my changes into this PR so we have one place to work on this together).

@mattrasmus what else do we need to get this upstream?

ernestum · 2024-04-16T09:27:49Z

redun/executors/postgres.py

+    conn = psycopg2.connect(**opt)
+    cur = conn.cursor()
+    try:
+        yield cur
+    finally:
+        cur.close()
+        conn.close()


Could this be:

with psycopg2.connect(**opt) as conn: with conn.cursor() as cursor: yield cursor

?

ernestum · 2024-04-16T09:29:23Z

redun/executors/postgres.py

+        conn.close()
+
+
+def run_worker_single(cur: psycopg2.cursor, queue: str, result_queue: Optional[str] = None) -> int:


The docstring says that this function executes a batch, however looking at the code it seems more like it executes a single job (or no job). So maybe fix the docstring?
Also maybe change the return value to a boolean?

ernestum · 2024-04-16T09:39:26Z

redun/executors/postgres.py

+    max_tasks: int = 1000,
+    max_time: int = 3600,


Docstring for the parameters is missing.

ernestum · 2024-04-16T09:45:07Z

redun/executors/postgres.py

+    cur.execute("COMMIT;")
+
+
+def run_worker_until_empty(


I would propose a name change here since it could be easily misunderstood as "run the worker until the worker is empty". For sure we mean the queue that goes empty here. But even that is not true when the queue is larger than 1k items or it takes more than 3600 seconds to run the job. What about just run_worker?

ernestum · 2024-04-16T09:45:38Z

redun/executors/postgres.py

+    queue: str,
+    result_queue: Optional[str] = None,
+    max_tasks: int = 1000,
+    max_time: int = 3600,


what about naming this max_seconds instead? Won't hurt to clarify the unions.

Also: max_... is misleading. We can easily exceed this time limit when we have a task that takes very long. The more verbose but correct name would be start_tasks_until_this_many_seconds_elapsed. Otherwise we would have to actively abort tasks right?

ernestum · 2024-04-16T10:33:34Z

redun/executors/postgres.py

+    obj: Dict[str, object] = {"func": func}
+    obj["args"] = args
+    obj["kw"] = kw
+    return encode_obj(obj)


Why not just:

return encode_obj({"func": func, "args": args, "kw": kw})

ernestum · 2024-04-16T10:38:22Z

redun/executors/postgres.py

+    Args:
+        cur (Cursor): Database cursor we use to run LISTEN/UNLISTEN
+        queue: table queue name
+        timeout (int, optional): Max seconds to wait. Defaults to 60.


Maybe document the randomization of the timeout here? also the code below waits up to 1.5 * timeout in seconds.

ernestum · 2024-04-16T10:48:10Z

redun/executors/postgres.py

+    """Inserts some payloads into the queue, then notifies any listeners of
+    that queue.
+
+    **WARNING**: This function requires the caller to run


I think we can change this function so it calls "commit" on it's own. It is used only in two places:

In line 431 in submit_task, where "commit" is called anyways right after.

In line 179 in run_worker_single. There it might look like we could accumulate multiple task submissions since we call it in a loop but in practice (due to LIMIT 1 in line 164) we only execute the loop body once.

ernestum · 2024-04-16T10:51:15Z

redun/executors/postgres.py

+def exec_task(
+    job_id: int,
+    module_name: str,
+    task_fullname: str,
+    args: Tuple,
+    kwargs: dict,
+    **extra,
+) -> Any:
+    """
+    Execute a task in the worker process.
+    """
+    # stolen from local_executor.py
+    load_task_module(module_name, task_fullname)
+    task = get_task_registry().get(task_fullname)
+    return task.func(*args, **kwargs)
+
+
+def exec_script_task(
+    job_id: int,
+    module_name: str,
+    task_fullname: str,
+    args: Tuple,
+    kwargs: dict,
+    **extra,
+) -> bytes:
+    """
+    Execute a script task from the task registry.
+    """
+    # stolen from local_executor.py
+    load_task_module(module_name, task_fullname)
+    task = get_task_registry().get(task_fullname)
+    command = get_task_command(task, args, kwargs)
+    return exec_script(command)


Instead of "stealing" from local_executor.py (which is just "redun/executors/local.py" I guess?) why not import it?

ernestum · 2024-04-16T10:58:24Z

redun/executors/postgres.py

+        return dict(
+            (
+                (k, v)
+                for (k, v) in config.items()
+                if k in ["dbname", "user", "password", "host", "port", "dsn"]
+            )
+        )


Maybe a tiny bit more concise and easier to understand?

return {k: config[k] for k in config.keys() & set(["dbname", "user", "password", "host", "port", "dsn"])}

gabriel-v · 2024-04-17T09:57:37Z

hi @ernestum

Thanks for the review. I'm no longer working on this experiment, it was a nice starting point into understanding how this works; I ended up working on a new framework using different opinions, requirements and backends

Feel free to take over the code if you want to further pursue this feature

ernestum · 2024-04-17T11:10:04Z

All right. Will do. Thanks for the huge initial chunk of work!

May I ask what other framework you ended up using @gabriel-v?

gabriel-v · 2024-04-23T15:49:38Z

Did a custom reinterpretation of this repo's DAG definition API using async python and scylladb, seeing some great throughput out of it - will leave a comment here when we publish it if you're interested

ernestum · 2024-04-24T12:52:40Z

Did a custom reinterpretation of this repo's DAG definition API using async python and scylladb, seeing some great throughput out of it - will leave a comment here when we publish it if you're interested

Yes that would be great!

gabriel-v · 2024-05-16T13:41:32Z

meanwhile you can look at https://github.com/temporalio/samples-python

we're evaluating it and it's a very strong contender

gabriel-v force-pushed the pg-executor branch from 2615083 to bff0346 Compare September 7, 2023 18:50

gabriel-v mentioned this pull request Sep 7, 2023

PostgreSQL Queue Executor (Distributed executor without Amazon/k8s) #78

Open

9 tasks

gabriel-v force-pushed the pg-executor branch from 477cef2 to 60fa23e Compare September 8, 2023 13:16

add postgres executor

056484d

gabriel-v force-pushed the pg-executor branch from b384e9c to 056484d Compare September 8, 2023 15:08

add tests for config check, simple workflow

f8a5889

gabriel-v force-pushed the pg-executor branch from 87c9c88 to f8a5889 Compare September 8, 2023 17:44

ernestum reviewed Apr 16, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Pg executor #84

Pg executor #84

gabriel-v commented Sep 7, 2023 •

edited

Loading

gregjohnso commented Feb 6, 2024

ernestum left a comment

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

ernestum Apr 16, 2024

gabriel-v commented Apr 17, 2024

ernestum commented Apr 17, 2024

gabriel-v commented Apr 23, 2024

ernestum commented Apr 24, 2024

gabriel-v commented May 16, 2024

		conn.close()


		def run_worker_single(cur: psycopg2.cursor, queue: str, result_queue: Optional[str] = None) -> int:

Pg executor #84

Are you sure you want to change the base?

Pg executor #84

Conversation

gabriel-v commented Sep 7, 2023 • edited Loading

gregjohnso commented Feb 6, 2024

ernestum left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

gabriel-v commented Apr 17, 2024

ernestum commented Apr 17, 2024

gabriel-v commented Apr 23, 2024

ernestum commented Apr 24, 2024

gabriel-v commented May 16, 2024

gabriel-v commented Sep 7, 2023 •

edited

Loading