PostgreSQLのSKIP LOCKEDを使ってテーブルをキューとして使用する

[programming] PostgreSQL

SKIP LOCKED

SKIP LOCKED は PostgreSQL 9.5 から入った新機能です。

これを使うと、FOR UPDATEの際に別トランザクションによって行ロックが取得されているレコードを除外することができます。すなわち、他のトランザクションによる行ロックが解除されるのを待つ必要がなくなります。

サンプル

idというカラムを持つidsというテーブルを作成し、3レコード作成しておきます。

CREATE TABLE ids AS SELECT generate_series(1, 3) AS id;

testdb=> CREATE TABLE ids AS SELECT generate_series(1, 3) AS id;
SELECT 3
testdb=> SELECT * FROM ids;
 id
----
  1
  2
  3
(3 rows)

1つめのトランザクションでは、FOR UPDATEでid=1の行ロックを取得します。

BEGIN;
SELECT * FROM ids WHERE id = 1 FOR UPDATE;

testdb=> BEGIN;
BEGIN
testdb=> SELECT * FROM ids WHERE id = 1 FOR UPDATE;
 id
----
  1
(1 row)

2つめのトランザクションでは、全行をFOR UPDATEで取得します。id=1のレコードがロックされているので、全行取ろうとしてもロックが競合して取得できません。(競合することがわかるようにNOWAITで待ち合わせないようにしています)

BEGIN;
SELECT * FROM ids FOR UPDATE NOWAIT;

testdb=> BEGIN;
BEGIN
testdb=> SELECT * FROM ids FOR UPDATE NOWAIT;
ERROR:  could not obtain lock on row in relation "ids"

このクエリにSKIP LOCKEDを付けると、ロックされているid=1のレコードを除外して取得できるようになります。

SELECT * FROM ids FOR UPDATE SKIP LOCKED;

testdb=> SELECT * FROM ids FOR UPDATE SKIP LOCKED;
 id
----
  2
  3
(2 rows)

SKIP LOCKEDを利用してキューを実装

これが便利なのは、テーブルをキューとして使うときだと思います。というか、それ以外で使い道を思いついていません。

SKIP LOCKEDを使うことによって、他のトランザクションの完了を待たずにロックしていない行を取得できるので、キューとしての性能を高めることが出来ます。

ということで、キューとしての使用方法を試してみます。queuesというテーブルを作り、idの昇順で取り出すことにします。

CREATE TABLE queues (
   id SERIAL,
   message TEXT,
   PRIMARY KEY(id)
);

INSERT INTO queues(message) SELECT 'message' || id FROM generate_series(1, 3) AS id;

testdb=> CREATE TABLE queues (
testdb(>   id SERIAL,
testdb(>   message TEXT,
testdb(>   PRIMARY KEY(id)
testdb(> );
CREATE TABLE
testdb=> INSERT INTO queues(message) SELECT 'message' || id FROM generate_series(1, 3) AS id;
INSERT 0 3
testdb=> SELECT * FROM queues;
 id | message
----+----------
  1 | message1
  2 | message2
  3 | message3
(3 rows)

ORDER BYとLIMIT 1で優先度順で先頭の1件取り出し、FOR UPDATE SKIP LOCKEDを付けてあげるだけです。

1つめのトランザクションで実行します。

BEGIN;
SELECT * FROM queues ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED;

testdb=> BEGIN;
BEGIN
testdb=> SELECT * FROM queues ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED;
 id | message
----+----------
  1 | message1
(1 row)

1つめのトランザクションが完了する前に、2つめのトランザクションで同じSQLを実行します。1つめのトランザクションの完了を待たされること無く、LOCKされていない行の中から優先度が高い行を取得できます。

BEGIN;
SELECT * FROM queues ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED;

testdb=> BEGIN;
BEGIN
testdb=> SELECT * FROM queues ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED;
 id | message
----+----------
  2 | message2
(1 row)

ただ、これだとqueuesテーブルから取り出したレコードを消せていないので、取り出した行を別途DELETEしてあげる必要があります。

そのまま同一のトランザクション内で取り出したレコードに対してDELETEするといった方法もありますが、RETURNINGを使うことによって、該当行の内容取得とDELETEをひとつのクエリで実行できます。

DELETE FROM queues 
  WHERE id = (SELECT id FROM queues ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED)
RETURNING *;

testdb=> DELETE FROM queues 
testdb->   WHERE id = (SELECT id FROM queues ORDER BY id LIMIT 1 FOR UPDATE SKIP LOCKED)
testdb-> RETURNING *;
 id | message
----+----------
  2 | message2
(1 row)

DELETE 1

RETURNINGは、INSERTやUPDATE、DELETEで処理したレコードの情報を取得する構文です。シーケンスによって払い出された値をINSERTの戻りで取得したいときなど、とても便利です。

ということで、とても手軽にキューが出来ました。

終わりに

似たような方法で、Advisory Locks(pg_try_advisory_lock)を使う方法がありますが、ORDER BYと組み合わせるとpg_try_advisory_lockを全行評価、すなわち全行ロックしてしまうので、優先度と組み合わせた取り出しが難しいです。

SKIP LOCKEDだと、そういった制約も回避できるので、9.5以降ならばSKIP LOCKEDを使った形のほうが簡単ではと思っています。