Posted on by Pavlo Golub

Intro – asynchronous chain execution in pg_timetable

I wrote about the new pg_timetable 3 major release not so long ago. Three essential features were highlighted:

Meanwhile, two minor releases have come out, and the current version is v3.2. It focuses on the completely new and fantastic feature asynchronous control over job execution:

  • you can start jobs manually,
  • you can stop jobs manually.

This machinery opens many possibilities, e.g.,

  • starting multiple chains simultaneously,
  • emulating conditional logic inside tasks,
  • jobs debugging,
  • cancelling long-running and/or frozen tasks etc.

Experimental environment: tutorial in asynchronous chain execution using pg_timetable

🔔 Notice!: In this tutorial, I will use two consoles. First for pg_timetable and the second for psql. You may use whatever SQL client you prefer.

Let’s try to set up with a clean database named timetable, running on a local host. Run pg_timetable in the first console:

$ pg_timetable --clientname=loader postgresql://scheduler@localhost/timetable
[ 2021-01-25 11:38:29.780 | LOG   ]: Connection established...
[ 2021-01-25 11:38:29.826 | LOG   ]: Proceeding as 'loader' with client PID 11560
[ 2021-01-25 11:38:29.827 | LOG   ]: Executing script: DDL
[ 2021-01-25 11:38:29.956 | LOG   ]: Schema file executed: DDL
[ 2021-01-25 11:38:29.956 | LOG   ]: Executing script: JSON Schema
[ 2021-01-25 11:38:29.962 | LOG   ]: Schema file executed: JSON Schema
[ 2021-01-25 11:38:29.962 | LOG   ]: Executing script: Built-in Tasks
[ 2021-01-25 11:38:29.964 | LOG   ]: Schema file executed: Built-in Tasks
[ 2021-01-25 11:38:29.964 | LOG   ]: Executing script: Job Functions
[ 2021-01-25 11:38:29.969 | LOG   ]: Schema file executed: Job Functions
[ 2021-01-25 11:38:29.969 | LOG   ]: Configuration schema created...
[ 2021-01-25 11:38:29.971 | LOG   ]: Accepting asynchronous chains execution requests...
[ 2021-01-25 11:38:29.972 | LOG   ]: Checking for @reboot task chains...
[ 2021-01-25 11:38:29.973 | LOG   ]: Number of chains to be executed: 0
[ 2021-01-25 11:38:29.974 | LOG   ]: Checking for task chains...
[ 2021-01-25 11:38:29.974 | LOG   ]: Checking for interval task chains...
[ 2021-01-25 11:38:29.976 | LOG   ]: Number of chains to be executed: 0
[ 2021-01-25 11:38:30.025 | LOG   ]: Number of active interval chains: 0
...

Now with any sql client (I prefer psql), execute the content of the samples/basic.sql script:

$ psql postgresql://scheduler@localhost/timetable
psql (13.0)
Type "help" for help.

timetable=> \i samples/basic.sql
INSERT 0 1

In a minute or less, you will see to following output in the pg_timetable console:

...
[ 2021-01-25 11:41:30.068 | LOG   ]: Number of chains to be executed: 1
[ 2021-01-25 11:41:30.075 | LOG   ]: Starting chain ID: 1; configuration ID: 1
[ 2021-01-25 11:41:30.093 | LOG   ]: Executed successfully chain ID: 1; configuration ID: 1
...

That means our chain is present in the system and the configuration is active. Let’s check what it is. From the psql console (or any other sql client):

timetable=> SELECT * FROM timetable.chain_execution_config;
-[ RECORD 1 ]--------------+--------------------
chain_execution_config     | 1
chain_id                   | 1
chain_name                 | notify every minute
run_at                     | * * * * *
max_instances              | 1
live                       | t
self_destruct              | f
exclusive_execution        | f
excluded_execution_configs |
client_name                |

We see our chain with chain_id = 1 has a single execution configuration. According to this configuration:

  • The chain should be executed each minute (run_at = '* * * * *');
  • There is only one instance of this chain allowed to be running (max_instances = 1);
  • This configuration is enabled and active (live = TRUE);
  • It’s not supposed to delete itself after successful execution (self_destruct = FALSE);
  • The chain can be executed in parallel with other jobs (exclusive_execution = FALSE);
  • The chain can be executed by any client (client_name IS NULL).

Let’s now disable this chain for the purpose of the experiment. From the psql console (or any other sql client):

timetable=> UPDATE timetable.chain_execution_config 
  SET live = FALSE 
  WHERE chain_execution_config = 1;

UPDATE 1

Right after this, you will see the following lines in the pg_timetable console:

...
[ 2021-01-25 11:53:30.183 | LOG   ]: Checking for task chains...
[ 2021-01-25 11:53:30.189 | LOG   ]: Checking for interval task chains...
[ 2021-01-25 11:53:30.190 | LOG   ]: Number of chains to be executed: 0
[ 2021-01-25 11:53:30.191 | LOG   ]: Number of active interval chains: 0
...

This means we have neither regularly scheduled chains nor interval enabled chains.

Starting a chain manually

To start a chain, one should use the newly introduced function

timetable.notify_chain_start(chain_config_id BIGINT, worker_name TEXT)

There are two crucial moments here: Despite the fact that we are starting a chain, we need to specify a chain configuration entry. This is because each chain may have several different configurations and different input arguments for each task in it.

From the psql console (or any other sql client):

timetable=> SELECT timetable.notify_chain_start(1, 'loader');

In less than a minute (because the main loop is 1 minute) you will see the following output in the pg_timetable console:

...
[ 2021-01-25 11:57:30.228 | LOG   ]: Adding asynchronous chain to working queue: {1 START 1611572244}
[ 2021-01-25 11:57:30.230 | LOG   ]: Starting chain ID: 1; configuration ID: 1
[ 2021-01-25 11:57:30.305 | LOG   ]: Executed successfully chain ID: 1; configuration ID: 1
...

Hooray! 🍾🎉 It’s working!!!!

🔔 Attention!

  1. We can start any chain configuration, even a disabled one!
  2. But it runs only once, even if the run_at field states it should be invoked each minute!
  3. You must specify the client name responsible for the execution!

For simplicity and performance reasons, the function doesn’t check if there is an active session with such a client name active. You, however, can always easily check this:

timetable=> SELECT * FROM timetable.active_session;
 client_pid | client_name | server_pid
------------+-------------+------------
      11560 | loader      |      10448
      11560 | loader      |      15920
      11560 | loader      |      13816
(3 rows)

So, for example, to start chain execution on each connected client, one may use this query:

timetable=> SELECT DISTINCT ON (client_name) 
  timetable.notify_chain_start(1, client_name) 
  FROM timetable.active_session;

Pay attention, we used DISTINCT to avoid duplicated execution for the same client session. As I described in the previous post, one pg_timetable session usually has several opened database connections.

Stopping a chain manually

To stop a chain, one should use the newly introduced function

timetable.notify_chain_stop(chain_config_id BIGINT, worker_name TEXT)

We need to specify a chain configuration entry and pg_timetable client name affected.

For the purpose of the experiment, let’s change our chain a little. From the psql console (or any other sql client):

timetable=> SELECT * FROM timetable.task_chain WHERE chain_id = 1;
 chain_id | parent_id | task_id | run_uid | database_connection | ignore_error | autonomous
----------+-----------+---------+---------+---------------------+--------------+------------
        1 |           |       6 |         |                     | t            | f
(1 row)


timetable=> SELECT * FROM timetable.base_task WHERE task_id = 6;
 task_id |            name             | kind |          script
---------+-----------------------------+------+--------------------------
       6 | notify channel with payload | SQL  | SELECT pg_notify($1, $2)
(1 row)


timetable=> UPDATE timetable.base_task 
  SET script = $$SELECT pg_sleep_for('5 minutes'), pg_notify($1, $2)$$ 
  WHERE task_id = 6;

UPDATE 1

timetable=> SELECT * FROM timetable.base_task WHERE task_id = 6;
 task_id |            name             | kind |              script
---------+-----------------------------+------+----------------------------------
       6 | notify channel with payload | SQL  | SELECT pg_sleep_for('5 minutes'), pg_notify($1, $2)
(1 row)

First, we list all tasks in our chain with chain_id = 1.

Our chain consists only of one task with task_id = 6:

  • it’s a simple SQL task
  • it calls the pg_notify function with arguments provided by the timetable.chain_execution_parameters table.

🔔 Notice! You can see the ER-schema of the timetable schema here.

So I added a pg_sleep_for() function call to the script to emulate the frozen or long-running process.

Now let’s try to start the chain. From psql:

timetable=> SELECT timetable.notify_chain_start(1, 'loader');
 notify_chain_start
--------------------

(1 row)

After some time we will see the chain started in the pg_timetable console:

...
[ 2021-01-25 12:32:30.885 | LOG   ]: Adding asynchronous chain to working queue: {1 START 1611574297}
[ 2021-01-25 12:32:30.887 | LOG   ]: Starting chain ID: 1; configuration ID: 1
...

As you can see, there is no information about successful execution. That means our chain is still in progress. Let’s stop it:

timetable=> SELECT timetable.notify_chain_stop(1, 'loader');
 notify_chain_start
--------------------

(1 row)

And in the pg_timetable log we can see:

...
[ 2021-01-25 12:36:31.108 | LOG   ]: Adding asynchronous chain to working queue: {1 STOP 1611574582}
[ 2021-01-25 12:36:31.108 | ERROR ]: Exec {"args":["TT_CHANNEL","Ahoj from SQL base task"],"err":{"Op":"read","Net":"tcp","Source":{"IP":"::1","Port":61437,"Zone":""},"Addr":{"IP":"::1","Port":5432,"Zone":""},"Err":{}},"pid":19824,"sql":"SELECT pg_sleep_for('5 minutes'), pg_notify($1, $2)"}
[ 2021-01-25 12:36:31.109 | ERROR ]: Exec {"args":[],"err":{},"pid":19824,"sql":"rollback"}
[ 2021-01-25 12:36:31.114 | ERROR ]: Task execution failed: {"ChainConfig":1,"ChainID":1,"TaskID":6,"TaskName":"notify channel with payload","Script":"SELECT pg_sleep_for('5 minutes'), pg_notify($1, $2)","Kind":"SQL","RunUID":{"String":"","Valid":false},"IgnoreError":true,"Autonomous":false,"DatabaseConnection":{"String":"","Valid":false},"ConnectString":{"String":"","Valid":false},"StartedAt":"2021-01-25T12:32:30.9494341+01:00","Duration":240160068}; Error: read tcp [::1]:61437->[::1]:5432: i/o timeout
[ 2021-01-25 12:36:31.115 | LOG   ]: Executed successfully chain ID: 1; configuration ID: 1
...

What a weird message, you might be thinking. One of the lines says: “ERROR: Task execution failed…”. And the final line says: “LOG: Executed successfully…”. How is that even possible?!

The explanation is simple. Chains can ignore errors. There is a chain property for that. And that’s precisely the case! Even if some steps of the chain failed, it’s considered successful.

Finally

This was the third in a series of posts dedicated to the new pg_timetable v3 features. Stay tuned for the components to be highlighted:

  • Debug mode for developing!

The previous posts can be found here:

Love! Peace! Stay safe! ❤

Pavlo Golub

Pavlo is a PostgreSQL expert and developer at CYBERTEC. He's been working with PostgreSQL since 2002. He has an M.Ed. with an emphasis in Mathematics and Computer Science from the Central Ukrainian State Pedagogical University, and previously worked as a professor at the Kirovograd Medical Professional College. He is the co-founder of PostgreSQL Ukraine https://www.facebook.com/groups/postgresql.ua . He develops and maintains the PostgreSQL tools PGWatch2 and pg_timetable.