🚀 go-pugleaf

On Feb 24, 4:31 pm, Slawomir Piasecki <piase...@gmail.com> wrote:
> witam wszystkich
>
> mam prawie napisany program potrzebny mi do obliczen na uczelni
> w ktorym wykorzystuje rekurencji, program bedzie dzialac na maszynie
> z procesorem osmio rdzeniowym, i pomyslalem sobie ze mozna by
> w tym momencie wykorzystac do pracy wszystkie osiem rdzeni korzystajac
> z watkow,

No to źle pomyślałeś, bo akurat w Pythonie uniemożliwia to GIL.
Jak chcesz zrównoleglać, to multiprocessing (http://docs.python.org/
library/multiprocessing.html)

> ale juz na poczatku sie zawiesilem, bo nie wiem jak
> kontrolowac
> w takim wypadku ilosc uruchomionych watkow, mozecie mnie jakos
> oswiecic,
> naprowadzic
> chyba kod jaki mam nie jest szczegolnie potrzebny gdyz pytanie jest
> jak
> sensownie polaczyc watki z rekurencja

Jeśli rekurencja bez efektów ubocznych, to banalnie prosto:
zrobić 8 workerów, kolejkę (multiprocessing.Queue), wywołanie
rekurencyjne zastąpić wrzucaniem podzadań do kolejki.

ok wiec multiprocess, poczytalem troche dokumentacji i troche
tutoriali
i postanowilem wypisac wstepnie maly programik:

[code]
def f(q, n):
    for i in range(2,4):
      t = i * n

      if t < 100:
        print "dodaje ", t
        q.put(t) # dodaj do kolejki

def run(nrStart):
     q = Queue()

     for i in range(nrStart):
       q.put(i) # dodaj do kolejki

     nr = 2
     pula = []

     while True:
       for i in pula:
         if i.is_alive() is False:
           pula.remove(i) # usun z puli jesli juz wykonany

       while len(pula) < nr:
           if q.qsize() > 0:
             dat = q.get() # pobierz z kolejki
             print "startuje ", dat
             p = Process(target=f, args=(q, dat)).start() # wystartuj
z procesem
             pula.append(p) # dodaj do puli

           else:
             if len(pula) == 0:
               return # zakoncz
             else:
               break # wroc do petli

       sleep(5) # czekaj

if __name__ == '__main__':
 run(5)

[/code]

i dostaje po odpaleniu komunikat:
  File "tst.py", line 28, in run
    if i.is_alive() is False:
AttributeError: 'NoneType' object has no attribute 'is_alive'
Exception in thread QueueFeederThread (most likely raised during
interpreter shutdown):
Traceback (most recent call last):
  File "/usr/lib/python2.6/threading.py", line 522, in
__bootstrap_inner
  File "/usr/lib/python2.6/threading.py", line 477, in run
  File "/usr/lib/python2.6/multiprocessing/queues.py", line 233, in
_feed
<type 'exceptions.TypeError'>: 'NoneType' object is not callable

wiec tutaj sie blokuje i nie wiem jak sprawdzic czy dany process jesce
dziala czy nie,

pozdrawiam
p.s.

On Feb 25, 5:52 pm, Slawomir Piasecki <piase...@gmail.com> wrote:
> ok wiec multiprocess, poczytalem troche dokumentacji i troche
> tutoriali
> i postanowilem wypisac wstepnie maly programik:
>
> [code]
> def f(q, n):
>     for i in range(2,4):
>       t = i * n
>
>       if t < 100:
>         print "dodaje ", t
>         q.put(t) # dodaj do kolejki
>
> def run(nrStart):
>      q = Queue()
>
>      for i in range(nrStart):
>        q.put(i) # dodaj do kolejki
>
>      nr = 2
>      pula = []
>
>      while True:
>        for i in pula:
>          if i.is_alive() is False:
>            pula.remove(i) # usun z puli jesli juz wykonany
>
>        while len(pula) < nr:
>            if q.qsize() > 0:
>              dat = q.get() # pobierz z kolejki
>              print "startuje ", dat
>              p = Process(target=f, args=(q, dat)).start() # wystartuj
> z procesem
>              pula.append(p) # dodaj do puli
>
>            else:
>              if len(pula) == 0:
>                return # zakoncz
>              else:
>                break # wroc do petli
>
>        sleep(5) # czekaj
>
> if __name__ == '__main__':
>  run(5)
>
> [/code]
>
> i dostaje po odpaleniu komunikat:
>   File "tst.py", line 28, in run
>     if i.is_alive() is False:
> AttributeError: 'NoneType' object has no attribute 'is_alive'
> Exception in thread QueueFeederThread (most likely raised during
> interpreter shutdown):
> Traceback (most recent call last):
>   File "/usr/lib/python2.6/threading.py", line 522, in
> __bootstrap_inner
>   File "/usr/lib/python2.6/threading.py", line 477, in run
>   File "/usr/lib/python2.6/multiprocessing/queues.py", line 233, in
> _feed
> <type 'exceptions.TypeError'>: 'NoneType' object is not callable
>
> wiec tutaj sie blokuje i nie wiem jak sprawdzic czy dany process jesce
> dziala czy nie,
>
> pozdrawiam
> p.s.
Robisz:
p = Process(target=f, args=(q, dat)).start()

Metoda start() nie zwraca obiektu na którym została wywołana, tylko
None (bo to procedura). Powinno być:

p = Process(target=f, args=(q, dat))
p.start()

PS. "if i.is_alive() is False:" bym zamienił na  "if not
process.is_alive():" - jakoś tak bardziej czytelnie. Zastanowiłbym się
też nad faktem, że aktualnie właściwie robisz aktywne oczekiwanie. Z
opóźnieniem 5 sekund, ale zawsze.

ok zadzialalo zobaczymy jak pojdzie z moim programem
dzieki za pomoc

p.s.

W dniu 25.02.2010 17:52, Slawomir Piasecki pisze:

> def f(q, n):
>      for i in range(2,4):
>        t = i * n
>
>        if t<  100:
>          print "dodaje ", t
>          q.put(t) # dodaj do kolejki
>
> def run(nrStart):
>       q = Queue()
>
>       for i in range(nrStart):
>         q.put(i) # dodaj do kolejki
>
>       nr = 2
>       pula = []
>
>       while True:
>         for i in pula:
>           if i.is_alive() is False:
>             pula.remove(i) # usun z puli jesli juz wykonany
>
>         while len(pula)<  nr:
>             if q.qsize()>  0:
>               dat = q.get() # pobierz z kolejki
>               print "startuje ", dat
>               p = Process(target=f, args=(q, dat)).start() # wystartuj
> z procesem
>               pula.append(p) # dodaj do puli
>
>             else:
>               if len(pula) == 0:
>                 return # zakoncz
>               else:
>                 break # wroc do petli
>
>         sleep(5) # czekaj
>

Tak jak Łukasz powiedział czekasz za każdym razem 5 sekund. Więc jeśli 
proces trwa 0.5 sekundy (lub (x * 5s) + 0.5s) i zadań dla procesów masz 
100 to za każdym uruchomieniem oszczędzałbyś 4.5s * 100 = 450 sekund co 
daje około 7 minut 30 sekund oszczędności.

Do tego wystarczy użyć funkcji select.select() i jako input dać 
q._reader (oczwywiście tylko jeśli q to multiprocessing.Queue a nie 
Queue.Queue)

Zastanów się, czy nie lepiej jest sprawdzać czy masz coś już w kolejce i 
ewentualnie nie zwracać procesów, które zakończyły swoje zadanie do puli 
wolnych procesów i także użyć selecta albo użyć semafora o wartości 
początkowej równej maksymalnej liczbie procesów i wywoływać za każdym 
nowym zadaniem acquire na semaforze.

Nawet fajna logika programu. Ale o ile dobrze widzę, to z zewnętrznej 
pętli nie wyjdziesz nigdy.

No fajnie - sama zaczynam udzielać porad (mam nadzieję, że nie gadam 
głupot) :)

On Feb 25, 5:52 pm, Slawomir Piasecki <piase...@gmail.com> wrote:
> ok wiec multiprocess, poczytalem troche dokumentacji i troche
> tutoriali
> i postanowilem wypisac wstepnie maly programik:
[ciach]

MSZ, przesadnie skomplikowałeś. Masz rekurencyjne tworzenie wątków, ja
pisałem raczej o stworzeniu 8 workerów i ich recyklowaniu.

no super chyba za szybko sie ucieszylem

tylko nie mam pojecia co do mnie mowicie,
powyzszy kod dziala spoko, ale jak wprowadze zamiast wyswietlania
liczb
cos bardziej skomplikowanego to juz sie zaczyna znowu klopot

funkcje f zamienilem na taka:

"def br(self, q, tab):
    print "===================================
\n===================================\n\n"

    print "curretn multiplicity ", tab["mlt"]
    lastNr, currentMulti = int(tab["st"][0]), int(tab["mlt"])
    self.points(currentMulti) ## wyszukiwanie ptk z definicji

    point = self.continuation(tab) # glowne obliczenia

    if len(point) > 0:

      n = tab["st"][0] + 1
      tav = []

      for pt in range(len(point)):
        nxtM = point[pt]["m"] * currentMulti # next multiplicity

        if nxtM < self.multi and n < 3:
          tmpName = self.newName({n: "%s-%s" % (tab["mlt"], pt)}) #
new name
          orbit = self.getSelectData(point[pt]["lab"], tab["n"]) #
wybor jednej orbity
          self.multiPerVar(orbit, tmpName, nxtM) # zapis orbity z
odpowiednim okresem
          tav.append({"n" : tmpName, "mlt" : nxtM, "st" : [n, pt],
"c" : True, "prvN" : tab["n"]}) # dodanie do listy zadan

          if nxtM > 1: # wybor metody obliczeniowej w zaleznosci od
multiplicity
            tav[-1]["orb"] = False

          else:
            tav[-1]["orb"] = self.getSelectData(point[pt]["lab"],
tab["n"], True)

        else:
          print "local multiplicity is higher than global!"

      for step in range(len(tav)): # dodanie do kolejki
        print "add next BFpoint ", step
        q.put(tav[step])

      print "number of BFpoint ", len(tav)

    else:
      print "there is no bifurication point"

    print "end of computation"
"

i teraz po wywolaniue run'a
pojawia sie moment w ktorym jeden proces sie nie chce zakonczyc
ostatni komunikat jaki mi sie wyswietla to: "end of computation"

ale nie dostaje tego komunikatu:

for i in pula:

        if not i.is_alive():
          print "end of process"
          pula.remove(i)

a z tymi workerami to jak niby sie za to zabrac?
dzieki
p.s.