Django’da Race Condition Problemine Bir Bakış

Herkese selamlar^^ Bu yazımızda Race Condition problemine ve Django üzerindeki çözümlerine göz atacağız. Öncelikle Race Condition probleminin ne olduğunu inceleyelim.

Race Condition Nedir?

Birden fazla threadin çalıştığı bir sistemde aynı anda, birden çok thread ile aynı verinin güncellenme/oluşturulma işlemleri sırasında karşımıza çıkan bir sistem tasarım sorunudur. Race condition problemi önlenmediği durumda çöp datalara ve maddi kayıplara yol açabilir.

2 thread örneği içeren Python örneğini çalıştırdığımızda sonucun 40+30 = 70 olması beklenir. Fakat programı çalıştırdığımızda sonucun farklı olduğunu ve her çalıştırdığımızda değiştiğini gözlemleriz.

Olayı biraz daha şekillendirelim ve problemin nasıl gerçekleştiğine bir bakalım:

Thread 1 veriyi okuduktan sonra arttırma işlemini gerçekleştirdi ve henüz veritabanındaki veriyi güncellemedi. Thread 2 güncelleme öncesinde veritabanındaki verinin eski değerini okudu ve üzerinde arttırma işlemi gerçekleştirdi. Thread 1 veritabanındaki veriyi güncellese bile thread 2 eski veriyi okuduğu için burada race condition problemi ile karşılaştık ve veri kaybı yaşadık.

Pythonda veriye ulaşılan kod blokları(bizim örneğimizde increase fonksiyonu) Critical Session içine alınarak Race Condition problemini önleyebiliriz. Critical Session içerisinde yer alan kod bloklarına aynı anda tek bir thread ulaşabilir ve çalıştırabilir.

Race Condition problemini anladığımıza göre Django üzerindeki çözümlerine bakabiliriz. Dökümanda Race Condition problemi için birden fazla çözüm bulunmaktadır.

class Product(models.Model):
    slug = models.SlugField(_("Slug"), unique=True)
    name = models.CharField(_("Name"), max_length=250)
    storage = models.PositiveIntegerField(_("Storage"))

Product isimli bir model oluşturduk. Bir ürün satılması durumunda ürünün storage değerini azaltan bir fonksiyon hazırlayalım.

def reduce_storage(slug):
    product = Product.objects.get(slug=slug)
    product.storage -= 1
    product.save(update_fields=["storage"])

1- F() Object

Django F() objelerini kullanarak doğrudan veritabanı üzerinde verinin değerine ulaşabilir ve işlemler yapabiliriz. Bu sayede olası bir Race Condition problemi engellenebilir fakat bunun kesin bir çözüm yolu olmadığını unutmayalım. Yukarıdaki fonksiyonu F() object kullanarak tekrar yazalım:

def reduce_storage(slug):
    product = Product.objects.get(slug=slug)
    product.storage = F("storage") - 1
    product.save(update_fields=["storage"])

Yukarıdaki F() object ile verinin veritabanında o anki değeri bir azaltılır. Bu sayede potansiyel bir Race Condition probleminden kaçınmış oluruz.

F() object ile ilgili daha ayrıntılı bilgi edinmek için Django F() Object Kullanımı yazımı okuyabilirsiniz.

2- select_for_update()

Django ORM tarafından sunulan select_for_update methodu sorgu kümesine ait tüm satırları kilitler ve aynı anda birden fazla güncelleme/silme işlemi yapılmasını engeller.

@transaction.atomic
def reduce_storage(slug):
    product = Product.objects.select_for_update().get(slug=slug)
    product.storage -= 1
    product.save(updated_fields=["storage"])

select_for_update Kullanılırken Önemli Noktalar

Default olarak select_for_update ile sorgu tarafından seçilen tüm satırlar kilitlenir. select_related içinde belirtilen nesneler de sorgu kümesine ait satırlar ile birlikte kilitlenir.

Örneğin select_related ile kategori bilgisini de çektiğimizi varsayalım. Product ile birlikte kategori bilgisi de kilitlenir. Ürün satın alma sırasında kategorinin adı güncellenmek istediğimizde bu satırlar kilitli olduğu için güncelleme başarısız olur.
Bunun önüne geçmek için kilitlemek istediğimiz satırları of(…) içerisinde belirtebiliriz. Sorgu kümesinin(Queryset) modeline erişmek için self kullanılmalıdır.

Product.objects.select_for_update(of=('self',)).filter(storage__gte=100)

Kilitlenmiş bir satırda farklı bir sorgu yapıldığında, kilit serbest bırakılana kadar 2. sorgu engellenir. nowait=True parametresini geçerek aramayı engellenmez hale getirebiliriz. Bu durumda çakışan bir kilit zaten başka bir işlem tarafından alınmışsa, sorgu kümesi değerlendirildiğinde DatabaseError fırlatılır.

Product.objects.select_for_update(nowait=True).get(slug=slug)

skip_locked=True parametresini geçerek kilitlenen satırları yok sayabilir ve işleme devam edebiliriz.

time = timezone.now() - timedelta(hours=6)
products = (Product.objects.select_for_update(skip_locked=True)
.filter(createad_at__gte=time))

skip_locked=True ve nowait=True parametreleri aynı anda kullanılmamalıdır. Her iki parametreyi de aynı anda kullanarak select_for_update öğesini çağırma girişimimiz ValueError ile sonuçlanır.

time = timezone.now() - timedelta(hours=6)
products = (Product.objects.select_for_update(nowait=True, skip_locked=True)
.filter(createad_at__gte=time))

3- get_or_create()

Race condition problemi genellikle güncelleme işlemlerinde karşımıza çıkabilir gibi görünse de bir veriyi oluşturma sırasında da karşılaşabiliriz. Aşağıdaki gibi bir kod parçacığımızın olduğunu düşünelim:

try:
    product = Product.objects.get(slug="computer")
except DoesNotExist:
    product = Product(slug='computer', name="computer", storage=50)
    product.save()

Ürün veritabanında aranır, ürün yoksa yeni bir kayıt oluşturulur. Birden çok thread çalıştığını düşündüğümüzde veritabanında duplicate kayıt olma riski ortaya çıkar. Başka bir durumda IntegrityError fırlatabilir ve programımızın akışını bozulabilir. Bu sorunun önüne geçmek için Django’nun bizlere sunduğu get_or_create() methodunu kullanabiliriz.

Öncelikle get_or_create() methodunun kaynak koduna bir göz atalım:

def get_or_create(self, defaults=None, **kwargs):
        """
        Look up an object with the given kwargs, creating one if necessary.
        Return a tuple of (object, created), where created is a boolean
        specifying whether an object was created.
        """
        # The get() needs to be targeted at the write database in order
        # to avoid potential transaction consistency problems.
        self._for_write = True
        try:
            return self.get(**kwargs), False
        except self.model.DoesNotExist:
            params = self._extract_model_params(defaults, **kwargs)
            # Try to create an object using passed params.
            try:
                with transaction.atomic(using=self.db):
                    params = dict(resolve_callables(params))
                    return self.create(**params), True
            except IntegrityError:
                try:
                    return self.get(**kwargs), False
                except self.model.DoesNotExist:
                    pass
                raise

get() methodu ile nesne veritabanında aratılır. Nesnenin bulunaması durumunda yeni bir kayıt oluşturulmaya çalışılır. Eğer oluşturma aşamasında IntegrityError fırlatılırsa nesne tekrar veritabanında aratılır ve bulunan değer döndürülür.

get_or_create() methodunun çalışmasını incelediğimizde ilk get() methodunu geçen birden çok threadin aynı anda create işlemi yapması mümkündür. Bu nedenle, işlemin doğru çalışması için sorgulanan alanlarda bir benzersizlik(unique) durumu olmalıdır. Aksi takdirde duplicate veriler oluşma riski mevcuttur.

Ürün modelimizde slug alanımız unique bir alan oduğu için bizim durumumuzda get_or_create() methodu race condition problemini engelleyecektir.

product, created = Product.objects.get_or_create(
          slug=slug, 
          defaults={'name': slug, 'storage': 50}
)

update_or_create() methodu da get_or_create() ile benzerdir. Bu sebeple get_or_create() için söylediğimiz durumlar update_or_create() içinde geçerlidir.

Recap

• Race Condition, birden çok thread aynı anda bir veriyi oluşturmaya/güncellemeye çalıştığında ortaya çıkabilir.
• Race Condition problemini çözmek için Django’da 3 yöntem kullanılır: F() object, select_for_update() methodu ve get_or_create() methodu
• Race Condition problemi veritabanında hatalı veya duplicate verilerin oluşmasına sebep olabilir.

Bir yazının daha sonuna geldik. Keyifli ve faydalı bir yazı olmuştur umarım. Bir sonraki yazıda görüşmek üzere ^^

Kaynaklar

[1] Django Queryset API Reference ~ select_for_update()
[2] Django Queryset API Reference ~ get_or_create()
[3] Managing concurrency in Django using select_for_update