对 QuerySet 中的第一个实例进行原子更新

Question

我正在开发一个系统，该系统在为许多工作机器提供工作时必须处理许多竞争条件。

客户端将在系统中查询状态=“0”（待办事项）的作业，然后以原子方式更新状态=“1”（锁定）的“最旧”行，并检索该行的 ID（用于更新带有工人信息的作业，例如哪台机器正在处理它等）。

这里的主要问题是可能有任意数量的客户端同时更新。一种解决方案是锁定大约 20 个 status='0' 的行，更新最旧的行，然后再次释放所有锁定。我一直在研究 TransactionMiddleware，但我不知道这将如何防止在我查询它之后从我下面更新最旧的情况。

我研究了 QuerySet.update() 的事情，它看起来很有希望，但是在两个客户端获得相同记录的情况下，状态只会更新，并且我们将有两个工作人员从事同一项工作..我真的很茫然。

我还找到了票证 #2705 ，它似乎可以很好地处理这种情况，但我不知道如何由于我有限的 SVN 经验，从那里获取代码（最后的更新只是差异，但我不知道如何将其与代码主干合并）。

代码：结果=作业

class Result(models.Model):
"""
Result: completed- and pending runs

'ToDo': job hasn't been acquired by a client
'Locked': job has been acquired
'Paused'
"""
# relations
run = models.ForeignKey(Run)
input = models.ForeignKey(Input)

PROOF_CHOICES = (
    (1, 'Maybe'),
    (2, 'No'),
    (3, 'Yes'),
    (4, 'Killed'),
    (5, 'Error'),
    (6, 'NA'),
)
proof_status = models.IntegerField(
    choices=PROOF_CHOICES,
    default=6,
    editable=False)

STATUS_CHOICES = (
    (0, 'ToDo'),
    (1, 'Locked'),
    (2, 'Done'),
)
result_status = models.IntegerField(choices=STATUS_CHOICES, editable=False, default=0)

# != 'None' => status = 'Done'
proof_data = models.FileField(upload_to='results/',
    null=True, blank=True)
# part of the proof_data
stderr = models.TextField(editable=False,
    null=True, blank=True)

realtime = models.TimeField(editable=False,
    null=True, blank=True)
usertime = models.TimeField(editable=False,
    null=True, blank=True)
systemtime = models.TimeField(editable=False,
    null=True, blank=True)

# updated when client sets status to locked
start_time = models.DateTimeField(editable=False)

worker = models.ForeignKey('Worker', related_name='solved',
    null=True, blank=True)

Answer 1

要将 #2705 合并到您的 django 中，您需要先下载它：

cd <django-dir>
wget http://code.djangoproject.com/attachment/ticket/2705/for_update_11366_cdestigter.diff?format=raw

然后将 svn 倒带到必要的 django 版本：

svn update -r11366

然后应用它：

patch -p1 for_update_11366_cdestigter.diff

它将告诉您哪些文件已成功修补，哪些文件未成功修补。在不太可能发生冲突的情况下，您可以手动修复它们，查看 http://code.djangoproject.com/attachment/ticket/2705/for_update_11366_cdestigter.diff

要取消应用补丁，只需编写

svn revert --recursive . 

Answer 2

如果您的 django 在一台机器上运行，则有一种更简单的方法可以做到这一点...请原谅伪代码，因为您的实现细节尚不清楚。

from threading import Lock

workers_lock = Lock()

def get_work(request):
    workers_lock.acquire()
    try:
        # Imagine this method exists for brevity
        work_item = WorkItem.get_oldest()
        work_item.result_status = 1
        work_item.save()
    finally:
        workers_lock.release()

    return work_item

Answer 3

我脑子里有两个选择。一种是在检索行时立即锁定行，并且只有在相应的行被标记为正在使用时才释放锁。这里的问题是，没有其他客户端进程可以查看未选择的作业。如果您总是自动选择最后一个，那么它可能是一个足够简短的窗口，适合您。

另一种选择是带回查询时打开的行，但每当客户端尝试获取要处理的作业时再次检查。当客户端尝试更新作业以对其进行操作时，首先会检查该作业是否仍然可用。如果其他人已经抓住了它，则会向客户端发回通知。这允许所有客户端将所有作业视为快照，但如果他们不断获取最新作业，那么您可能会让客户端不断收到作业已在使用中的通知。也许这就是您所指的竞争条件？

解决这个问题的一种方法是将特定组中的工作返回给客户，这样他们就不会总是得到相同的列表。例如，按地理区域甚至随机地分解它们。例如，每个客户端的 ID 可以为 0 到 9。对作业中的 ID 取模，并将具有相同结尾数字的作业发送回客户端。不过，不要将其仅限于这些工作，因为您不希望出现您无法完成的工作。例如，如果您有 1、2 和 3 的客户以及 104 的工作，那么没有人能够获得它。因此，一旦没有足够的具有正确结尾数字的职位，职位就会开始返回其他数字以填充列表。您可能需要在这里尝试一下确切的算法，但希望这能给您一个想法。

如何锁定数据库中的行以更新它们和/或发回通知将在很大程度上取决于您的 RDBMS。在 MS SQL Server 中，只要中间不需要用户干预，您就可以将所有这些工作很好地包装在存储过程中。

我希望这有帮助。