Django Yarış Durum Açığı (Race Condition)

Yarış durumu açığı(Race Condition) özellikle işlemsel (transactional) hareketli sistemlerde büyük maliyete sebep olan veri bütünlüğü hatasıdır. Yarış durumu açığını ölçeklendirilmemiş sistemlerde fark etmek oldukça zordur. Çünkü yarış durum açığı başlı başına ölçeklendirme sonucu oluşan bir hatadır ve deterministik bir özellik göstermez.

Yarış Durumu(Race Condition) Senaryosu

Django ile bir uygulama yazdınız ve uygulama içerisinde bir uçbiriminiz var. Bu uçbirim hesap bakiyesini arttırmak üzerine tasarlanmış. Hesap bakiyelerini aşağıdaki gibi bir model ile temsil ettiğinizi düşünün.

class TimeStamped(models.Model):
    class Meta:
        abstract=True
    olusturma_tarihi = models.DateTimeField(default=timezone.now)
    son_guncelleme_tarihi = models.DateTimeField(default=timezone.now)

    def save(self, *args, **kwargs):
        if not self.olusturma_tarihi:
            self.olusturma_tarihi = timezone.now()

        self.son_guncelleme_tarihi = timezone.now()
        return super(TimeStamped, self).save(*args, **kwargs)

class Hesap(TimeStamped):
    sahip = models.IntegerField(unique=True)
    bakiye = models.DecimalField(max_digits=30, decimal_places=10)

Böylelikle hesap sahibini temsil eden bir kullanıcınız olacak ve bunu sahip alanında tutacaksınız. Bunun yanında bakiye alanıyla da kullanıcının bakiyesini temsil edeceksiniz.

Uçbirimi temsil etmek üzere aşağıdaki gibi bir de view’ınız olsun.

from django.shortcuts import render
from django.http.response import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from .models import Hesap
from decimal import Decimal
# Create your views here.

@csrf_exempt
def hesap(request):
    if request.method == 'POST':
        if 'sahip' not in request.POST:
            return JsonResponse({"hata": "sahip alanı bulunmalı"}, status=400)
        elif 'artim' not in request.POST:
            return JsonResponse({"hata": "artim alanı bulunmalı"}, status=400)
        try:
            sahip=int(request.POST['sahip'])
            artim=float(request.POST['artim'])
        except Exception as e:
            print(e)
            return JsonResponse({"hata": "sahip tamsayı, artim decimal olmalı"}, status=400)
        sonuc = Hesap.objects.filter(sahip=sahip)
        if len(sonuc) > 0:
            sonuc[0].bakiye += Decimal(artim)
            sonuc[0].save()
        else:
            yeni_hesap = Hesap()
            yeni_hesap.bakiye = artim
            yeni_hesap.sahip = sahip
            yeni_hesap.save()
        return JsonResponse({"mesaj": "Tamam"})
    else:
        return JsonResponse({"hata": "Sadece POST isteklerine izin verilir."}, status=405)

Yukarıdaki kod içerisinde de gördüğünüz üzere olmayan sahip için yeni hesap bakiye alanı oluşturuyor var olanlar için ise artım yapıyoruz. Şu ana kadar herhangi bir problem gözükmemekte.

from django.test import TestCase, Client
from .models import Hesap
# Create your tests here.

class EnpointTestCase(TestCase):
    def test_hesap_endpoint(self):
        c = Client()
        c.post("/hesap", {"sahip": "1", "artim": "20"})
        self.assertEquals(Hesap.objects.get(sahip=1).bakiye, 20, "Yanlış bakiye")
        c.post("/hesap", {"sahip": "1", "artim": "30"})
        self.assertEquals(Hesap.objects.get(sahip=1).bakiye, 50, "Yanlış bakiye")
        c.post("/hesap", {"sahip": "1", "artim": "35"})
        self.assertEquals(Hesap.objects.get(sahip=1).bakiye, 85, "Yanlış bakiye")

Yukarıdaki testi de python manage.py test komutuyla çalıştırdığımızda herhangi bir sorunla karşılaşmayacağız.

Bu uçbirimi gerçek bir uygulama üzerine taşıdığınızda eğer yüksek trafikli bir sistemde canlıya alım gerçekleştirdiyseniz muhtemelen “Parayı yatırdığım halde bakiyemde görünmüyor” gibi başınızı ağrıtacak bir çok şikayetle karşı karşıya kalacaksınız. Bunun sabebiyse yarış durumu açığı. Yani günlük dosyasındaki işlem toplamının kullanıcı bakiyesinden daha fazla olduğunu görme ihtimaliniz hiç de az değil.

Yarış Durum Açığı (Race Condition) Nedir?

Yarış durum açığı web uygulamasına has bir durum değildir. Bir kaynağa birden fazla aktörün aynı anda manipülasyon yapmaya çalışmasıyla oluşan bir durumdur. Buradaki aktörler yük dengeleyici arkasındaki fiziksel makineler olmakla beraber aynı fiziksel makine üzerindeki birden fazla iş parçacığı da olabilir. Bir veri kaynağına aynı anda birden fazla aktörün manipülasyon yapmasıyla birlikte verinin bozulması durumuna yarış durumu(race condition) diyoruz.

Genellikle veri kaynağına sorgu –> bağıl işlem –> güncelleme döngüsünde görülen bu hata, yükün fazla olduğu sistemlerde deterministik olmayan veri bozulmalarına sebep olur ve test etmesi ciddi anlamda zordur. Yukarıdaki birim testinde de gördüğümüz üzere testlerinizi başarıyla tamamlayacaktır.

Yukarıdaki durumda aktörler kritik kaynak tarafına bağlanıp işlemlerini yaptığında veri bozulmasına müsait bir ortam oluşur.

Basit bir python koduyla bu durumu oluşturalım.

from threading import Thread
import time
#Kritik kaynak aktörlerin bağlanıp manipülasyon yapacakları kaynak
kritik_kaynak = 0

def arttir(artim):
    """
    kritik kaynağı, verilen artım sayısı kadar arttırır.
    :param artim:
    :return:
    """
    global kritik_kaynak
    for i in range(artim):
        kritik_kaynak += 1


basla=time.time()
threads = []
for _ in range(8):
    t = Thread(target=arttir, args=(100000, ))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Kritik kaynak son hali: {kritik_kaynak}")
print(f"Süre: {time.time()-basla}")

Yukarıdaki kod anlayacağınız üzere 8 iş parçacığı açıp herbirinde 100000 artım yapmakta ve kritik_kaynak son hali 800000 olması gerekmekte. Fakat çıktı farklı:


Kritik kaynak son hali: 711152
Süre: 0.1109309196472168

Process finished with exit code 0

Her denemede farklı çıkacağına da emin olabilirsiniz çünkü deterministik olmayan bir hatayla karşı karşıyayız.

Django Üzerinde Test

Django üzerinde yaptığımız uygulamayı test etmek için aşağıdaki kodu kullanıyoruz:

from threading import Thread, Lock
import time
import requests

def erisim(iterasyon, artim, sahip):
    """
    Uçbirim üzerinde seri artım yapar.
    :param iterasyon:
    :param artim:
    :param sahip:
    :return:
    """
    for i in range(iterasyon):
        response = requests.post("http://127.0.0.1:8000/hesap", data={"artim": str(artim), "sahip": str(sahip)})


basla=time.time()
threads = []
for _ in range(10):
    t = Thread(target=erisim, args=(100, 20, 1))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Süre: {time.time()-basla}")

Bu kodla beraber uygulamaya 10 ayrı iş parçacığının her birinden 100’er iterasyonla bakiyeyi 20 arttırması için uçbirim çağrıldı. Yani son durumda 1 sahibinin bakiyesi 10*100*20=20000 olmalı.

Sonuç ise farklı görünüyor, bu da yarış durum hatasına işaret ediyor.

Nasıl Önlenebilir?

Yarış durumu açığı(Race condition) aktörler arasında bir kilit süreci oluşturularak önlenebilmektedir. Aktörler kaynağa erişecekleri zaman kaynağın kilidini alır işlem yapar ve kaynağın kilidini bırakır. Aşağıdaki kod üzerinde bunun örneğini görmektesiniz.

from threading import Thread, Lock
import time
#Kritik kaynak aktörlerin bağlanıp manipülasyon yapacakları kaynak
kritik_kaynak = 0

def arttir(artim, tl):
    """
    kritik kaynağı, verilen artım sayısı kadar arttırır.
    :param artim:
    :return:
    """
    global kritik_kaynak
    for i in range(artim):
        with tl:
            kritik_kaynak += 1


basla=time.time()
threads = []
#Kaynak üzerinde kilit oluşturak için threading.Lock kullanıyoruz
tl = Lock()
for _ in range(8):
    t = Thread(target=arttir, args=(100000, tl))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(f"Kritik kaynak son hali: {kritik_kaynak}")
print(f"Süre: {time.time()-basla}")

Çıktısı ise aşağıdaki gibi:

Kritik kaynak son hali: 800000
Süre: 4.247368335723877

Process finished with exit code 0

Burada gözünüze çarpacak bir detay olacak. Kaynak daha fazla kullanıldığı için süre oldukça arttı. Gerçek bir uygulamada sürecin tamamını senkronize etmeyeceğiniz için bu denli performans düşüşü yaşamayacağınızı söyleyebilirim fakat kaynak erişimini senkronize etmek maliyeti de beraberinde getirir.

Django Üzerinde Çözüm

Django için ise senkronizasyon için farklı yöntemler mevcut. Web uygulamasında kaynağımız veri tabanı olduğu için veri tabanı üzerindeki özelliklerden faydalanarak çözümler oluşturulabilir.

İlk yöntem pesimistik kilit. Bu yöntemde yukarıdaki python kodu örneğindeki yöntemi uygularız ve veri tabanındaki satırları işlem bitimine kadar kilitleriz. Fakat bu da senkronizasyon maliyetini beraberinde getirecek.

Aşağıdaki view ile pesimistik kilitleme yapıyoruz ve testimizi gerçekleştiriyoruz.

from django.shortcuts import render
from django.http.response import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.db import transaction
from .models import Hesap
from decimal import Decimal
# Create your views here.


@csrf_exempt
def hesap(request):
    if request.method == 'POST':
        if 'sahip' not in request.POST:
            return JsonResponse({"hata": "sahip alanı bulunmalı"}, status=400)
        elif 'artim' not in request.POST:
            return JsonResponse({"hata": "artim alanı bulunmalı"}, status=400)
        try:
            sahip=int(request.POST['sahip'])
            artim=float(request.POST['artim'])
        except Exception as e:
            print(e)
            return JsonResponse({"hata": "sahip tamsayı, artim decimal olmalı"}, status=400)
        with transaction.atomic():#Transaction içerisinde select for update sorgusu yapıyoruz.
            sonuc = Hesap.objects.select_for_update().filter(sahip=sahip)
            if len(sonuc) > 0:
                sonuc[0].bakiye += Decimal(artim)
                sonuc[0].save()
            else:
                yeni_hesap = Hesap()
                yeni_hesap.bakiye = artim
                yeni_hesap.sahip = sahip
                yeni_hesap.save()
            return JsonResponse({"mesaj": "Tamam"})
    else:
        return JsonResponse({"hata": "Sadece POST isteklerine izin verilir."}, status=405)

Sonuç ise beklediğimiz gibi yarış durumunu(race condition) düzeltmiş. Fakat önceki testimizin bitişi 45 saniye civarında sürerken bu testin 88 saniye civarında olduğunu görüyoruz. Yani maliyet senkronizasyonun beraberinde gelmekte.

Bir diğer çözüm ise F() ifadeleri. Bu ifadeler db sorgusu üzerinde bağıl işlemler yapmamızı sağlıyor. Aşağıdaki view üzerinde F() ifadesi kullanıp tekrar test edelim:

from django.shortcuts import render
from django.http.response import JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.db.models import F
from .models import Hesap
from decimal import Decimal
# Create your views here.

@csrf_exempt
def hesap(request):
    if request.method == 'POST':
        if 'sahip' not in request.POST:
            return JsonResponse({"hata": "sahip alanı bulunmalı"}, status=400)
        elif 'artim' not in request.POST:
            return JsonResponse({"hata": "artim alanı bulunmalı"}, status=400)
        try:
            sahip=int(request.POST['sahip'])
            artim=float(request.POST['artim'])
        except Exception as e:
            print(e)
            return JsonResponse({"hata": "sahip tamsayı, artim decimal olmalı"}, status=400)
        sonuc = Hesap.objects.filter(sahip=sahip)
        if len(sonuc) > 0:
            #Bu satırda F ifadesi kullanılıyor.
            sonuc[0].bakiye = F("bakiye") + Decimal(artim)
            sonuc[0].save()
        else:
            yeni_hesap = Hesap()
            yeni_hesap.bakiye = artim
            yeni_hesap.sahip = sahip
            yeni_hesap.save()
        return JsonResponse({"mesaj": "Tamam"})
    else:
        return JsonResponse({"hata": "Sadece POST isteklerine izin verilir."}, status=405)

Testini yaptığımızda testin 45 saniye sürdüğünü ve sonucun doğru geldiğini görüyoruz.

Sonuç

Yüksek trafikli web uygulamalarında ortaya çıkan yarış durumu hatasını gidermenin yolunu artık biliyoruz.

Sevgiyle kalın…