带有锁定的Postgresql的一致性，并选择进行更新

Question

我有一个可以支持一定数量的并发操作的应用程序。这由postgres中的“插槽”表表示。当节点联机时，它们会在表中插入许多行，每个插槽一个。当作业声明插槽时，它们会更新表中声明其中一个插槽的行，并在完成时再次释放它。

插槽表如下所示：

CREATE TABLE slots (
    id INT8 PRIMARY KEY DEFAULT nextval('slots_seq'),
    node_name TEXT NOT NULL,
    job_name TEXT
);

在任何时候，它都有一些固定的行数，每行可以或可以不填写job_name。

当新作业要启动时，它将运行以下查询以获取应在其上运行的节点的名称：

BEGIN;
LOCK TABLE slots IN ACCESS EXCLUSIVE MODE;
SELECT id, node_name
    FROM slots
    WHERE job_name IS NULL
    LIMIT 1
    FOR UPDATE;

（从游标中读取出node_name和id）

UPDATE slots
    SET job_name = %(job_name)s
    WHERE id = %(slot_id)s;
COMMIT;

这通常可以声明行而不会丢失任何更新，但是并发级别更高，在执行了许多SELECT … FOR
UPDATE和UPDATE查询的同时，仅声明了几行。最终结果是，我们最终运行的作业要多于其可用的插槽。

我在锁定错误吗？有更好的方法来解决这个问题吗？什么不使用表锁？

事务级别SERIALIZABLE不会削减它，仅填充了几行。

我正在使用PostgreSQL版本8.4。

Answer 1

好吧，我用perl编写了一个程序来模拟正在发生的事情，因为我认为您所说的是不可能的。确实，运行模拟后，即使关闭锁定，我也没有任何问题（因为SELECT FOR UPDATE并且UPDATE应该进行必要的锁定）。

我在PG 8.3和PG 9.0上运行了它，并且在两个位置上都能正常工作。

我敦促您尝试该程序和/或尝试使用python版本，以拥有一个不错的紧密测试用例，您可以与该类共享该用例。如果有效，您可以调查差异之间的区别，如果无效，则可以与其他人一起玩。

#!/usr/bin/perl
use DBI;
$numchild = 0;
$SIG{CHLD} = sub { if (wait) {$numchild--;} };

sub worker($)
{
  my ($i) = @_;
  my ($job);

  my $dbh = DBI->connect("dbi:Pg:host=localhost",undef,undef,{'RaiseError'=>0, 'AutoCommit'=>0});

  my ($x) = 0;
  while(++$x)
  {
#    $dbh->do("lock table slots in access exclusive mode;") || die "Cannot lock at $i\n";
    my @id = $dbh->selectrow_array("select id from slots where job_name is NULL LIMIT 1 FOR UPDATE;");

    if ($#id < 0)
    {
      $dbh->rollback;
      sleep(.5);
      next;
    }
    $job = "$$-$i-($x)";
    $dbh->do("update slots set job_name='$job' where id=$id[0];") || die "Cannot update at $i\n";
    $dbh->commit || die "Cannot commit\n";
    last;
  }
  if (!$job)
  {
    print STDERR "Could not find slots in 5 attempts for $i $$\n" if ($ENV{'verbose'});
    return;
  }
  else
  {
    print STDERR "Got $job\n" if ($ENV{'verbose'} > 1);
  }
  sleep(rand(5));

#  $dbh->do("lock table slots in access exclusive mode;") || die "Cannot lock at $i\n";
  $dbh->do("update slots set usage=usage+1, job_name = NULL where job_name='$job';") || die "Cannot unlock $job";
  print STDERR "Unlocked $job\n" if ($ENV{'verbose'} > 2);
  $dbh->commit || die "Cannot commit";
}

my $dbh = DBI->connect("dbi:Pg:host=localhost",undef,undef,{'RaiseError'=>0, 'AutoCommit'=>0});

$dbh->do("drop table slots;");
$dbh->commit;
$dbh->do("create table slots (id serial primary key, job_name text, usage int);") || die "Cannot create\n";
$dbh->do("insert into slots values (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0), (DEFAULT,NULL,0);") || die "Cannot insert";
$dbh->commit;

for(my $i=0;$i<200;$i++)
{
  if (!fork)
  {
    worker($i);
    exit(0);
  }

  if (++$numchild > 50)
  {
    sleep(1);
  }
}
while (wait > 0)
{
  $numchild--;
  print "Waiting numchild $numchild\n";
  sleep(1);
}
my $dbh = DBI->connect("dbi:Pg:host=localhost",undef,undef,{'RaiseError'=>0, 'AutoCommit'=>0});
my $slots = $dbh->selectall_arrayref("select * from slots;") || die "Cannot do final select";
my $sum=0;
foreach my $slot (@$slots)
{
  printf("%02d %3d %s\n",$slot->[0], $slot->[2], $slot->[1]);
  $sum += $slot->[2];
}
print "Successfully made $sum entries\n";