Python Parallel Processing Sorunu

[Trans]

Selamlar,

Python noobuyum, yeni başladım sayılır o yüzden yardım rica ediyorum. Yapmak istediğim işlem basit, N adet process oluşacak ve bunlar infinite loop içerisinde olan işlemi çağıracaklar ve içeride belirli işlemler yapılacak ( ilgili işlemin kaç saat süreceği belirli değil o nedenle while true olarak belirttim ) . Buradaki en büyük problemim global olarak tanımladığım değişkenlerde değişiklikler olacak olması, beceremedim :) Aşağıda genel yapmak istediğimin örneğini atıyorum. 

şimdiye kadar denediklerim;

• multiprocessing.Value yerine daha önceden düz değişken tanımlayıp metodlara parametre olarak paslamayıp, içinde global olarak belirtmiştim yemedi.
• Lock eklemeyi denedim, beceremedim

Basit tipler(int, double vs) yerine Array, List, Dictionary vs kullanmak istesem nasıl yapmak gerekecek

Program içerisinde sabit ve değişecek olan değişkenleri farklı bir class a almak istesem nasıl yapabilirim, bu konuda da yardım edebilirseniz sevinirim.

from datetime import time
from multiprocessing import Process, Value
import ctypes

count = Value(ctypes.c_int, 0)
value = Value(ctypes.c_double, 100.0)
increase_array = []
decrease_array = []


def increase_append(c):
    val = {"c": c.value, "count": count.value}
    increase_array.append(val)
    print(increase_array)


def decrease_append(c):
    val = {"c": c.value, "count": count.value}
    decrease_array.append(val)
    print(decrease_array)


def increase_count(c):
    c.value += 1
    print(count.value)
    increase_append(c)


def decrease_count(c):
    c.value -= 1
    print(count.value)
    decrease_append(c)


def make_operations(c):
    while True:
        time.sleep(5)
        if c.value < 20:
            increase_count(c)
        else:
            decrease_count(c)


if __name__ == '__main__':
    proc = []
    for i in range(50):
        process = Process(target=make_operations, args=[count])
        process.start()
        proc.append(process)
    while True:
        for p in proc:
            p.join()

 

[Lancelion]

multiprocessing.value veya manager -biraz overkill- kullanman lazim, lock kullanarak bunlarsiz da yapabilirsin.

https://docs.python.org/3/library/multiprocessing.html#shared-ctypes-objects

https://stackoverflow.com/questions/9436757/how-to-use-a-multiprocessing-manager

[sardalya]

mutex

[Trans]

Selamlar,

Teşekkürler, Manager ile hallettim.

Spoiler

from multiprocessing import Process, Manager
import random


def increase_append(attr):
    val = {"c": attr.value, "count": attr.count}
    tmp = attr.increase_array
    tmp.append(val)
    attr.increase_array = tmp
    print(attr.increase_array)


def decrease_append(attr):
    val = {"c": attr.value, "count": attr.count}
    tmp = attr.decrease_array
    tmp.append(val)
    attr.decrease_array = tmp
    print(attr.decrease_array)


def increase_count(attr):
    attr.count += 1
    print(attr.count)
    increase_append(attr)


def decrease_count(attr):
    attr.count -= 1
    print(attr.count)
    decrease_append(attr)


def make_operations(attr):
    num = random.randrange(10)
    if attr.count < num:
        increase_count(attr)
    else:
        decrease_count(attr)


if __name__ == '__main__':
    proc = []

    manager = Manager()
    attrib = manager.Namespace()
    attrib.count = 0
    attrib.value = 0.0
    attrib.increase_array = []
    attrib.decrease_array = []

    while True:
        for i in range(6):
            process = Process(target=make_operations, args=[attrib])
            process.start()
            proc.append(process)

        for p in proc:
            p.join()

 

 

[reyou]

bu bir coin mining algoritmasi mi yoksa?!

[Oce4n]

ileri matematik de olabilir.

[Trans]

Coin mining algoritması değil aslında, market price analysis gibi bir proje yapmaya çalışıyorum. Amacım biraz da pythona alışmak. Hobi amaçlı takılıyorum. Üstteki parallel processing sorununu çözdükten sonra, neden db kullanmayayım diyip postgresql kullanayım dedim.

Bunun için de pyscopg2 librarysini kullanmaya başladım. Burada da şöyle bir sıkıntım oluştu;

paralel olarak herhangi bir tabloda update işlemi yapmak istediğim zaman istediğim limitten fazla işlem yapıyor.

Örnek olarak paralel olarak X tablosundan bir değer çekiyorum ( A kolonu diyelim ) , bu değer 100 den büyük ise, Y tablosundaki B kolonunu, A+100 değeri kadar update ediyorum. B kolonunun da maksimum değerini yine kod içerisinde 1000 olarak kontrol ediyorum. Eğer güncellenecek değer 1000 den büyük ise işlem yapmamasını istiyorum.

Burada ilk process A kolonunu 100 değerini çektiği zaman, B kolonunu 200 yapmasını bekliyorum, ikinci process de 200 değerini gördüğü için 300 yapmasını bekliyorum, böyle böyle 9. process son işlemi yaptığı zaman durmasını istiyorum.

Fakat olan ve çoğunlukla denk geldiğim, aynı anda birden çok işlem başladığı için, örnek olarak 5 process başlayıp son değer olarak 800 ü gördüğü zaman, aynı anda hepsi 100 arttırıp son değeri 1300 yapıyorlar. 

Bu sorunu connection ı processler arasında taşımak harici nasıl aşabilirim. 

Mart 9, 2021 Trans tarafından düzenlendi

[riglous]

Bunu garanti altina alabilecegin en mantikli cozum DB'ye trigger koyup, update sirasinda deger istediginden buyuk olacaksa hata dondurup islemi commit etmemek olacaktir.