PostgreSQLの１レコードへの連続アクセスのテスト

環境

テスト環境は以下のようになります。

CentOS 5.5 64bit
ruby　1.9.2p0 (2010-08-18 revision 29036) [x86_64-linux]
PostgreSQL: Ver 8.4.4
RubyのPostgreSQL用ドライバ　postgres-pr (0.6.3)

テストスクリプト

テストスクリプトは以下のようになります。

test_postgres.rb

require "./MyPostgres"
require "pp"

TABLE_NAME = 'test_postgres'
PROCESS_MAX = 5
TEST_COUNT_MAX = 200

def pg_countup(max, title)
  print("\n", title, "\n")
  tp = MyPostgres.new(TABLE_NAME)
  tp.db_init()

  1.upto(PROCESS_MAX) do |index|
    #Thread.new do
    fork do
      testdb = MyPostgres.new(TABLE_NAME, 'fork')
      testdb.query(index, max) { |num, i, conn|
         yield num, i, conn
      }
    end
  end
  #sleep 5
  ret = Process.waitall
end

def test_postgres()
  tblname = TABLE_NAME
  max = TEST_COUNT_MAX

  title = "UPDATE #{tblname} SET count=count+1 WHERE id=1"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("UPDATE #{tblname} SET count=count+1 WHERE id=1;")
  }

  title = "SELECT count AS current FROM #{tblname} WHERE id=1 FOR UPDATE;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;"
  pg_countup(max, title) { |num, i, conn|
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1 FOR UPDATE;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
  }

  print("\n----------------")
  title = "BEGIN;\n" +
          "SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1 FOR UPDATE;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1 FOR UPDATE;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  title = "BEGIN;\n" +
          "SET TRANSACTION ISOLATION LEVEL READ COMMITTED;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1 FOR UPDATE;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL READ COMMITTED;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1 FOR UPDATE;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  title = "BEGIN;\n" +
          "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1 FOR UPDATE;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1 FOR UPDATE;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  title = "BEGIN;\n" +
          "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1 FOR UPDATE;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1 FOR UPDATE;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  print("\n----------------")
  title = "BEGIN;\n" +
          "SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  title = "BEGIN;\n" +
          "SET TRANSACTION ISOLATION LEVEL READ COMMITTED;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL READ COMMITTED;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  title = "BEGIN;\n" +
          "SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  title = "BEGIN;\n"
          "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;\n" +
          "SELECT count AS current FROM #{tblname} WHERE id=1;\n" +
          "UPDATE #{tblname} SET count=1+current WHERE id=1;\n" +
          "COMMIT"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;")
    res = conn.exec("SELECT count FROM #{tblname} WHERE id=1;")
    conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + " WHERE id=1;")
    conn.exec("COMMIT;");
  }

  print("\n----------------")
  title = "UPDATE #{tblname} SET count=1+(SELECT count FROM #{tblname} WHERE id=1) WHERE id=1;"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("UPDATE #{tblname} SET count=1+(SELECT count FROM #{tblname} WHERE id=1) WHERE id=1;")
  }

  title = "BEGIN\n" +
          "SET TRANSACTION ISOLATION LEVEL READ COMMITTED;\n" +
          "UPDATE #{tblname} SET count=1+(SELECT count FROM #{tblname} WHERE id=1) WHERE id=1;\n" +
          "COMMIT;"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL READ COMMITTED;")
    conn.exec("UPDATE #{tblname} SET count=1+(SELECT count FROM #{tblname} WHERE id=1) WHERE id=1;")
    conn.exec("COMMIT;")
  }

  title = "BEGIN\n" +
          "SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;\n" +
          "UPDATE #{tblname} SET count=1+(SELECT count FROM #{tblname} WHERE id=1) WHERE id=1;\n" +
          "COMMIT;"
  pg_countup(max, title) { |num, i, conn|
    conn.exec("BEGIN;")
    conn.exec("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;")
    conn.exec("UPDATE #{tblname} SET count=1+(SELECT count FROM #{tblname} WHERE id=1) WHERE id=1;")
    conn.exec("COMMIT;")
  }
end

test_postgres()
exit

PostgreSQLにアクセスするためのクラス

MyPostgres.rb

require "postgres"
require "pp"

#
# create table test_postgres(id integer, name varchar(32),count integer default 0);
#
class MyPostgres
  attr_accessor :table_name
  attr_accessor :exec_name

  def initialize(tbl_name, exec_name="")
     @table_name = tbl_name
     @exec_name = exec_name
  end

  def opendb()
    begin
      return PGconn.new("localhost",5432,"","","testdb","postgres","password")
    rescue => ex
      print(ex.class, " ERROR ", ex.message,"\n")
    end
  end

  def db_init()
    tblname = @table_name
    conn = opendb()
    conn.exec("delete from #{tblname}")
    conn.exec("insert into #{tblname}(id,name,count) values(1,'PostgreSQL',0);")
    conn.close
  end


  def query(num, count_max)
    begin
      tblname = @table_name
      conn = opendb()
      i = 0
      while (i < count_max)
        i += 1
        yield num, i, conn
      end
      res = conn.exec("SELECT count FROM #{tblname};")
      print_result(num, res)
    rescue => ex
       print(ex.class, " ERROR ", ex.message,"\n")
    ensure
       conn.close if conn
    end
  end

  def test_countup(num, count_max)
    tblname = @table_name
    query(num, count_max) { |num, i, conn|
        res = conn.exec("SELECT count FROM #{tblname} FOR UPDATE;")
        conn.exec("UPDATE #{tblname} SET count=1+" + res.result[0][0] + ";")
    }
  end

  def print_result(num, res)
    if (res.status && res.result[0])
       print @exec_name, num, " end count = ", res.result[0][0], "\n"
    else
       print "ERROR not get current count\n"
    end
  end

  def get_status(res)
    if res.status == PGresult::COMMAND_OK
      return 'status: COMMAND_OK',"\n"
    elsif res.status == PGresult::EMPTY_QUERY
      return 'status: EMPTY_QUERY',"\n"
    elsif res.status == PGresult::TUPLES_OK
      return 'status: TUPLES_OK',"\n"
    end
  end

end

結果

テストでは以下のようにテストスクリプトのtest_postgres.rbを実行しました。

$ ruby test_postgres.rb

テストではforkでプロセスを5個作成しています。それぞれが200まで加算していきますので、最後のプロセスのcountが1000になれば正常にカウントされたことになります。 1000に満たない場合はカウント漏れが発生しています。

PostgreSQLのテスト結果

UPDATE test_postgres SET count=count+1 WHERE id=1
fork1 end count = 680
fork2 end count = 847
fork3 end count = 937
fork4 end count = 992
fork5 end count = 1000

SELECT count AS current FROM test_postgres WHERE id=1 FOR UPDATE;
UPDATE test_postgres SET count=1+current WHERE id=1;
fork2 end count = 214
fork1 end count = 215
fork3 end count = 228
fork4 end count = 264
fork5 end count = 266

----------------
BEGIN;
SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
SELECT count AS current FROM test_postgres WHERE id=1 FOR UPDATE;
UPDATE test_postgres SET count=1+current WHERE id=1;
COMMIT
fork1 end count = 883
fork2 end count = 959
fork3 end count = 986
fork5 end count = 999
fork4 end count = 1000

BEGIN;
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT count AS current FROM test_postgres WHERE id=1 FOR UPDATE;
UPDATE test_postgres SET count=1+current WHERE id=1;
COMMIT
fork1 end count = 922
fork5 end count = 987
fork3 end count = 995
fork4 end count = 997
fork2 end count = 1000

BEGIN;
SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT count AS current FROM test_postgres WHERE id=1 FOR UPDATE;
UPDATE test_postgres SET count=1+current WHERE id=1;
COMMIT
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
fork1 end count = 200

BEGIN;
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
SELECT count AS current FROM test_postgres WHERE id=1 FOR UPDATE;
UPDATE test_postgres SET count=1+current WHERE id=1;
COMMIT
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L1600   RExecutePlan
fork1 end count = 200

----------------
BEGIN;
SET TRANSACTION ISOLATION LEVEL READ UNCOMMITTED;
SELECT count AS current FROM test_postgres WHERE id=1;
UPDATE test_postgres SET count=1+current WHERE id=1;
COMMIT
fork1 end count = 199
fork2 end count = 233
fork4 end count = 255
fork3 end count = 256
fork5 end count = 268

BEGIN;
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
SELECT count AS current FROM test_postgres WHERE id=1;
UPDATE test_postgres SET count=1+current WHERE id=1;
COMMIT
fork1 end count = 203
fork2 end count = 220
fork3 end count = 230
fork5 end count = 244
fork4 end count = 244

BEGIN;
SET TRANSACTION ISOLATION LEVEL REPEATABLE READ;
SELECT count AS current FROM test_postgres WHERE id=1;
UPDATE test_postgres SET count=1+current WHERE id=1;
COMMIT
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
fork1 end count = 200

BEGIN;

RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
fork1 end count = 200

----------------
UPDATE test_postgres SET count=1+(SELECT count FROM test_postgres WHERE id=1) WHERE id=1;
fork1 end count = 200
fork2 end count = 264
fork4 end count = 332
fork3 end count = 359
fork5 end count = 363

BEGIN
SET TRANSACTION ISOLATION LEVEL READ COMMITTED;
UPDATE test_postgres SET count=1+(SELECT count FROM test_postgres WHERE id=1) WHERE id=1;
COMMIT;
fork1 end count = 168
fork3 end count = 203
fork2 end count = 205
fork4 end count = 208
fork5 end count = 212

BEGIN
SET TRANSACTION ISOLATION LEVEL SERIALIZABLE;
UPDATE test_postgres SET count=1+(SELECT count FROM test_postgres WHERE id=1) WHERE id=1;
COMMIT;
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
RuntimeError ERROR ERROR        C40001  M同時更新のため直列化アクセスができませんでした FexecMain.c       L2090   RExecUpdate
fork1 end count = 200

注意点

注意１

PostgreSQLでは、トランザクション隔離レベルは、以下の２つしかありません。

READ COMMITTED
SERIALIZABLE

そのため、READ UNCOMMITTEDを指定した場合はREAD COMMITTEDに、REPEATABLE READを指定した場合はSERIALIZABLE になっていることに気をつける必要があります。

READ UNCOMMITTED => READ COMMITTED
REPEATABLE READ => SERIALIZABLE

結果を見ても READ UNCOMMITTEDとREAD COMMITTED、　REPEATABLE READ と SERIALIZABLE の結果は同じになっています。

注意２

SERIALIZABLEでは「同時更新のため直列化アクセスができませんでした」というエラーが起こっています。これはPostgreSQLのマニュアルのシリアライザブル隔離レベルに詳しい説明が書かれています。

UPDATE、DELETE、SELECT FOR UPDATE、およびSELECT FOR SHAREコマンドでは、SELECTと同じように対象行を検索します。これらのコマンドでは、トランザクションが開始された時点で既にコミットされている対象行のみを検出します。しかし、その対象行は、検出されるまでに、同時実行中の他のトランザクションによって、既に更新（もしくは削除あるいはロック）されている可能性があります。このような場合、シリアライザブルトランザクションは、最初の更新トランザクションが（それらがまだ進行中の場合）コミットもしくはロールバックするのを待ちます。最初の更新処理がロールバックされると、その結果は無視され、シリアライザブルトランザクションでは元々検出した行の更新を続行することができます。しかし、最初の更新処理がコミット（かつ、単にロックされるだけでなく、実際に行が更新または削除）されると、シリアライザブルトランザクションでは、以下のようなメッセージを出力してロールバックを行います。

ERROR: could not serialize access due to concurrent update

これは、シリアライザブルトランザクションでは、トランザクションが開始された後に別のトランザクションによって更新されたデータは変更またはロックすることができないためです。