[Python] Multitreading blijft hangen

[Verwijderd]

Beste Tweakers,

Een aantal weken ben ik al aan het programmeren, maar heb ondertussen een probleem.
Wat ik geprogrammeerd is het volgende, ik maak een queue en die vul ik met de bestanden die ik inlees op het systeem. Vervolgens trek ik uit de bestanden een aantal data en maak ik van die bestanden een hash.
Om het op te lossen heb ik overal al prints en exceptions proberen op te vangen om wat wijzer te worden, maar mij lukt het niet om duidelijk te krijgen, waarom hij vast loopt.

Weet wel waar die vast loopt. Hij loopt vast bij regel die met commentaar is aangeven.
Hopelijk is er iemand die kan zien wat ik verkeerd doe of waar het aan kan liggen... Ben zelf niet extreem thuis in Python (2.7.3)... Maar heb wel enige programmeer ervaring.

Code:

code:

#!/usr/bin/python

import os, sys, Queue, threading, re, time
import magic

from hash import *
from macs import *

fileList = []
rootDir = sys.argv[0]
top = os.getcwd()
start = '/'
nameThread = threading.currentThread().getName()

class Scanner(threading.Thread):
    def __init__(self, queue, dir):
        threading.Thread.__init__(self)
        self.queue = queue
        self.dir = dir
        
    def run(self):
            print "Adding filepaths to queue"
            self.addToQueue(self.dir)
            print "Adding filepaths to queue done"
                                
    def addToQueue(self, topdown=True):
        for root, dirs, files in os.walk(start, topdown=True, followlinks=False):
            for name in files:
                self.queue.put(os.path.join(root,name)) 
                print "Successfully added filepath\t",os.path.join(root,name), "to queue"
                
class Hasher(threading.Thread):
    def __init__(self, queue):
            threading.Thread.__init__(self)
            self.queue = queue
        
    def run(self):
        nextFile = self.queue.get()
        while nextFile:
            try:    
                print "\n","\n","Trying " + nextFile #laatst geprint
                self.scanFile(nextFile) #hier loopt hij vast en blijft hij hangen...
                nextFile = self.queue.get()
                self.queue.task_done()
                print "\n","Hashing of this file successfull"
            except:
                print "Failed to scan file:", "/n",sys.exc_info()[0] #geeft geen exception
                nextFile = self.queue.get()
                self.queue.task_done()
                pass
            
    def scanFile(self, name):
            filepath = self.queue.get()
            ftype = magic.from_file(filepath)
            mtype = magic.from_file(filepath, mime=True)
            btype = magic.from_buffer(open(filepath).read(1024))
fileQueue = Queue.Queue()

print "\n","Starting in directory:", start
print "\n","Starting Scanner"
for scanner in xrange(1):
    scanner = Scanner(fileQueue, rootDir)
    scanner.start()
    print "\n","Scanner started successfully"
    scanner.join()
    print "\n","Stopping Scanner"
    
print "\n","Starting Hasher"
for hasher in xrange(1):
    hasher = Hasher(fileQueue)
    hasher.start()
    print "\n","Hasher started successfully"
    hasher.join()
    print "\n","Stopping Hasher"

[epic007]

Queue.get() haalt het item echt uit de queue, en volgens mij roep je dit ergens te veel aan. Wat gebeurt er bijvoorbeeld als de queue van de Hasher maar één element bevat?

Queue.get() is ook blocking.

[ Voor 39% gewijzigd door epic007 op 06-06-2012 16:42 ]

[Verwijderd]

epic007 schreef op woensdag 06 juni 2012 @ 16:40:
self.queue.get() haalt het item echt uit de queue, en volgens mij roep je dit ergens te veel aan.

Als dat zo is waarom doorloopt hij dan eerst 25000 bestanden en dan pas blijft hij hangen?
Volgens mij is dus dit niet de oplossing...
Maar kan het proberen om er een of een aantal uit te halen

[HuHu]

Is je queue leeg? Je roept namelijk overal maar get() aan en dat haalt elke keer een bestand uit je queue. Op het moment dat je queue leeg is, blijft hij wachten.

Zie ook de documentatie van get: http://docs.python.org/library/queue.html#Queue.Queue.get

[Verwijderd]

HuHu schreef op woensdag 06 juni 2012 @ 16:43:
Is je queue leeg? Je roept namelijk overal maar get() aan en dat haalt elke keer een bestand uit je queue. Op het moment dat je queue leeg is, blijft hij wachten.

Zie ook de documentatie van get: http://docs.python.org/library/queue.html#Queue.Queue.get

Volgens mij is dat het ook niet want dan zou hij toch een exception moeten geven van Empty? Omdat de queue leeg is..
Maar hoe kan ik het makkelijkste testen als de queue leeg is? Wil je me even op weg helpen?

[HuHu]

Queue.get([block[, timeout]])
Remove and return an item from the queue. If optional args block is true and timeout is None (the default), block if necessary until an item is available. If timeout is a positive number, it blocks at most timeout seconds and raises the Empty exception if no item was available within that time. Otherwise (block is false), return an item if one is immediately available, else raise the Empty exception (timeout is ignored in that case).

Is block False? Ik weet niet zo 1-2-3 wat de default-waarde van block is, maar het zou zomaar True kunnen wezen. Vervang je get() eens door get(False).

[Verwijderd]

Na enig tijd scripting ben er toch achter gekomen dat de Queue toch echt leeg is.

Heb nu de volgende regel toegevoegd in def scanFile:

code:

filepath = self.queue.get(timeout=5, block=False)

Vervolgens geeft hij een heleboel Empty Queue meldingen en uiteindelijk krijg ik deze:

code:

Trying **BESTANDS_NAAM**
Faild to scan file: /n <class 'Queue.Empty'>
Exception in thread Thread-2
Traceback (most recent call last):
    File "/usr/lib/python2.7/threading.py", line 551, in __bootstrap_iner self.run()
    File "mainthread.py", line 55, in run
        self.queue.task_done()
    File "/usr/lib/python2.7/Queue.py", line 64, in task_done
        raise ValueError('task_done() called too many times')
ValueError: task_done() called too many times

Stopping Hasher

Hij doorloopt nu wel me hele code dus Dat is al heel wat, maar ik doe het volgens mij niet op de manier zoals het hoort...

[epic007]

Wat doet de parameter 'name' eigenlijk in je functie scanFile(..) ?

En je gebruikt nu threads, maar volgens mij is je programma op dit moment nog zo sequentieel als wat. Ga je dat nog veranderen?

[ Voor 49% gewijzigd door epic007 op 06-06-2012 17:39 ]

[Verwijderd]

epic007 schreef op woensdag 06 juni 2012 @ 17:37:
Wat doet de parameter 'name' eigenlijk in je functie scanFile(..) ?

En je gebruikt nu threads, maar volgens mij is je programma op dit moment nog zo sequentieel als wat. Ga je dat nog veranderen?

Moet ik deze reactie serieus nemen of heb je gewoon me code niet doorgelezen?!

[Zeebonk]

Misschien is het handig als je nog wat meer tijd aan de documentatie van multiprocessing besteed. Hier staat naast informatie over het gooien van excepties bij een lege queue en het belang van een timeout en de effecten van blocking en non-blocking calls, ook nog eens het volgende voorbeeld:

Python:

def queue_func(queue):
    for i in range(30):
        time.sleep(0.5 * random.random())
        queue.put(i*i)
    queue.put('STOP')

def test_queue():
    q = multiprocessing.Queue()

    p = multiprocessing.Process(target=queue_func, args=(q,))
    p.start()

    o = None
    while o != 'STOP':
        try:
            o = q.get(timeout=0.3)
            print o,
            sys.stdout.flush()
        except Empty:
            print 'TIMEOUT'

In dit voorbeeld gebruiken ze een token 'STOP' om aan te geven dat dat het laatste element in de queue is. Als je in jouw situatie N kind processen hebt, is het dus noodzaak om N keer een einde queue token in de queue te plaatsen.

[ Voor 22% gewijzigd door Zeebonk op 07-06-2012 21:53 ]

[Wolf87]

De parameter 'name' wordt nu inderdaad niet gebruikt in ScanFile omdat je hier opnieuw een "self.queue.get()" doet in plaats van de name parameter te gebruiken.

Zeebonk schreef op woensdag 06 juni 2012 @ 22:01:
Is het mogelijk om een lege regel tussen een alinea en een code blok te krijgen? Of ben ik dan aan het mierenneuken?

Met de br-tag is dat mogelijk

[ Voor 7% gewijzigd door Wolf87 op 07-06-2012 18:30 ]

[Verwijderd]

Wolf87 schreef op donderdag 07 juni 2012 @ 18:29:
De parameter 'name' wordt nu inderdaad niet gebruikt in ScanFile omdat je hier opnieuw een "self.queue.get()" doet in plaats van de name parameter te gebruiken.


[...]


Met de br-tag is dat mogelijk

Dit wil dus zeggen dat ik de "self.queue.get()" moet weg halen in dit stuk code?:

code:

def scanFile(self, name):
            filepath = self.queue.get()
            ftype = magic.from_file(filepath)
            mtype = magic.from_file(filepath, mime=True)
            btype = magic.from_buffer(open(filepath).read(1024))

[Wolf87]

Verwijderd schreef op donderdag 07 juni 2012 @ 19:08:
[...]


Dit wil dus zeggen dat ik de "self.queue.get()" moet weg halen in dit stuk code?:

code:

def scanFile(self, name):
            filepath = self.queue.get()
            ftype = magic.from_file(filepath)
            mtype = magic.from_file(filepath, mime=True)
            btype = magic.from_buffer(open(filepath).read(1024))

Dat kan, je kan in plaats van filepath "name" gebruiken en de regel "filepath=self.queue.get()" weghalen.

[Verwijderd]

Wolf87 schreef op donderdag 07 juni 2012 @ 19:36:
[...]


Dat kan, je kan in plaats van filepath "name" gebruiken en de regel "filepath=self.queue.get()" weghalen.

Als ik me code zo doe:

code:

def scanFile(self, name):
            #filepath = self.queue.get() #in commentaar, dus niet gebruikt
            ftype = magic.from_file(filepath)
            mtype = magic.from_file(filepath, mime=True)
            btype = magic.from_buffer(open(filepath).read(1024))

Dan krijg ik alleen maar exceptions, "Failed to scan file: <type 'exceptions.NameError>"
Ook als ik het volgende doe "name = self.queue.get()"
Dan krijg ik dezelfde error

[Zeebonk]

Natuurlijk gooit hij nu een exception, je gebruikt nu filepath terwijl die niet is gedefineerd. Ga nou nog eens stap voor stap door je code heen en kijk nou eens goed naar het voorbeeld dat ik gaf, het lijkt nu alsof je gewoon maar wat aan het doen bent.

(het voorbeeld dat ik gaf is niet met multithreading maar met multiprocessing en hoewel jouw probleem niet opgelost word door te switchen tussen deze twee wil ik toch even aangeven dat multithreading in Python geen concurrent threads biedt door de aanwezige GIL lock. Wil je echt concurrent workers hebben en dus tegelijk gebruik maken van meerdere cores, zal je multiprocessing moeten gebruiken)

[ Voor 45% gewijzigd door Zeebonk op 07-06-2012 21:56 ]

[t_captain]

Kort antwoord, misschien spuit 11.

* Queue.get() is blocking. Deze functie geeft geen Empty exception maar zal gewoon blijven wachten tot er iets in de queue komt.
* Queue.get_nowait() of Queue.get(False) is nonblocking en geeft wel een Empty exception als er op het moment van aanroepen niets in de queue zit.


Verder met Zeebonk. Het lijkt of je een CPU intensieve taak (hashing van een grote batch files) wilt versnellen door hem over een aantal threads te verdelen. Daarvoor zijn threads in Python niet geschikt. Door de GIL (Global Interpreter Lock) worden je verschillende CPU's niet goed belast. Er is 1 aparte GIL per proces, je zou de Multiprocessing of Subprocess module moeten bekijken voor deze toepassing.

Threading is vooral interessant wanneer de thread een aanzienlijk deel van zijn tijd in IOWAIT doorbrengt i.p.b. RUN. Denk aan database queries etc. Dan paralleliseer je de wait toestanden en dat levert snelheid op. Overigens zou je in jouw toepassing afhanlkelijk van de grootte van de files en de snelheid van je filesystem wel wat snelheid winnen, het lezen van een file is bevat immers ook IOWAIT.

[epic007]

Verwijderd schreef op woensdag 06 juni 2012 @ 21:18:
[...]


Moet ik deze reactie serieus nemen of heb je gewoon me code niet doorgelezen?!

Ik heb je code wel doorgelezen en het was een hint om inderdaad die 'name' parameter te gebruiken.
Zoiets:

code:

def scanFile(self, name):
            ftype = magic.from_file(name)
            mtype = magic.from_file(name, mime=True)
            btype = magic.from_buffer(open(name).read(1024))

Over het multithreading gedeelte, je programma doet nu ongeveer dit:

code:

for x
  Start een Scanner thread
  Wacht tot ie klaar is
end

for x
  Start een Hasher thread
  Wacht tot ie klaar is
end

Ik weet niet of je doel is om meerdere threads tegelijk te gebruiken zodat de performance omhoog gaat maar door steeds te wachten tot je thread klaar is (thread.join) bereik je dat niet.

[Verwijderd]

@epic007, Excuus voor mijn botte reactie nog...

Ben blij dat je me toch nog wilt helpen!!

Ik heb nu me code zo aangepast:

code:

def scanFile(self, name):
            #filepath = self.queue.get() #in commentaar, dus niet gebruikt
            ftype = magic.from_file(name)
            mtype = magic.from_file(name, mime=True)
            btype = magic.from_buffer(open(name).read(1024))

Dit draait hetzelfde in mijn ogen als voorheen, maar ik heb nog steeds dat hij blijft hangen...
Ik wil geen meerdere threads draaien. Ik wil dat ze na elkaar hun eigen taak afronden.
*Multitreading kan er altijd nog worden ingebouwd.

[Zeebonk]

Waarom gebruik je dan threads?

[Verwijderd]

Zeebonk schreef op maandag 11 juni 2012 @ 14:20:
Waarom gebruik je dan threads?

Omdat het in het vervolg makelijker om hem multitreaded te maken. Lijkt me logisch.
Doet er voor de rest niet toe. Gaat er gewoon om dat hij nog steeds vast loopt en niks verder ben....

Staat er niet ergens een voorbeeld? Kom er gewoon niet meer uit...

[Zeebonk]

Je zegt dat je geen meerdere threads wilt draaien, terwijl je in de door jou geposte code wel threads gebruikt. Dat is niet logisch. Ik heb je al twee keer naar een voorbeeld verwezen, waarbij ik 1x ook het voorbeeld heb gepost, als je daar nou gewoon eens naar kijkt, of als je al hebt gekeken verteld waarom het je niet help en wat je hebt gedaan om dat voorbeeld op jouw situatie toe te passen. Anders kunnen we natuurlijk weinig.

[epic007]

Kijk ook nog even goed hoe 'Queue.get()' en 'Queue.task_done()' werken en in welke volgorde ze moeten.

[Verwijderd]

epic007 schreef op maandag 11 juni 2012 @ 15:29:
Kijk ook nog even goed hoe 'Queue.get()' en 'Queue.task_done()' werken en in welke volgorde ze moeten.

Queue.get = Remove and return an item from the queue (Oftewel die verwijdert en zet ze terug in de queue)
Queue.task_done = Indicate that a formerly enqueued task is complete.(Oftewel geeft aan dat een eerder ge-queuede taak is voltooid)

For each get() used to fetch a task, a subsequent call to task_done() tells the queue that the processing on the task is complete.

Voor elke get() gebruikt om een taak op te halen. Een 2de aanroep van task_done geeft aan dat de taak voltooid is

Volgens mij is dit het

[Herko_ter_Horst]

Verwijderd schreef op donderdag 14 juni 2012 @ 17:55:
[...]


Queue.get = Remove and return an item from the queue (Oftewel die verwijdert en zet ze terug in de queue)

Zelfs zonder enige kennis van Python of de gebruikte library durf ik te zeggen dat dit niet klopt. "return" betekent niet "terugzetten", maar teruggeven als waarde van de method call.

[Verwijderd]

Herko_ter_Horst schreef op donderdag 14 juni 2012 @ 18:39:
[...]

Zelfs zonder enige kennis van Python of de gebruikte library durf ik te zeggen dat dit niet klopt. "return" betekent niet "terugzetten", maar teruggeven als waarde van de method call.

Klopt heb het heel snel geschreven maar je hebt gelijk inderdaad

[epic007]

Kijk eens goed naar regel 43 en 44 van je code. Je roept op regel 44 'task_done()' aan maar welke task is dan done? Misschien moet je ze omdraaien.

[Verwijderd]

epic007 schreef op vrijdag 15 juni 2012 @ 11:34:
Kijk eens goed naar regel 43 en 44 van je code. Je roept op regel 44 'task_done()' aan maar welke task is dan done? Misschien moet je ze omdraaien.

Bedankt voor reactie. Heb nu me code zo:

Python:

class Hasher(threading.Thread):
        def __init__(self, queue):
            threading.Thread.__init__(self)
            self.queue = queue
        
    def run(self):
        nextFile = self.queue.get()
        while nextFile:
            try:    
                        print "\n","\n","Trying " + nextFile #laatst geprint
                        self.scanFile(nextFile) #hier loopt hij vast en blijft hij hangen...
                print "\n","TEST1"
                        nextFile = self.queue.task_done()
                print "\n","TEST2"
                nextFile = self.queue.get(timeout=5, block=False)
                        print "\n","Hashing of this file successfull"
                    except:
                        print "Failed to scan file:", "/n",sys.exc_info()[0] #geeft geen exception
                        self.queue.task_done()
                nextFile = self.queue.get()
                        pass
            
        def scanFile(self, name):
        name = self.queue.get()
                ftype = magic.from_file(name)
                mtype = magic.from_file(name, mime=True)
                btype = magic.from_buffer(open(name).read(1024))
            print "\n","ScanFile doorlopen"
fileQueue = Queue.Queue()

Helaas blijft hij nog steeds hangen op waarschijnlijk het laatste element.

Krijg als output bij de laatste file:

code:

Hashing of this file successfull

Trying **BESTANDSNAAM**

Mijn prints met "ScanFile doorlopen", "TEST1" en "TEST2" worden niet geprint....

[Verwijderd]

iemand die me nog kan helpen??

[ValHallASW]

Je indenting op regel 24 en 28 (en, bij nader inzien: in je hele paste) klopt niet (gebruik je tabs en spaties door elkaar?), en je doet nog steeds een dubbele queue.get() die nergens voor nodig is. Maak voor jezelf anders eerst een een heel basale versie die niets meer doet dan de filename uitprinten.

Je kunt gebruik maken van sys._current_frames + traceback.print_trace om een idee te krijgen waar je deadlock plaatsvindt.

[ Voor 32% gewijzigd door ValHallASW op 18-06-2012 15:51 ]

[Zeebonk]

Ik snap niet waarom je nu niet gewoon kijkt naar de tips die er worden gegeven. Daar staat het allemaal in:

Je code loopt vast bij je één na laatste item omdat de je steeds 2x queue.get doet, 1x in run, en 1x in scanFile, in scanfile moet hij weg. Je slaat hierdoor sowieso de helft van je items in de queue over.

Als je dat hebt opgelost krijg je een exception bij de queue.get in run, omdat je timeout zal verstrijken als de queue leeg is en je de exceptie niet afvangt. Om er voor te zorgen dat je niet nodeloos zit te wachten op een timeout als de queue leeg is, is het slim om na alle bestandspaden een 'STOP' token in de queue te stoppen. Door steeds te checken of je 'STOP' uit de queue heb gehaald weet je direct wanneer de queue leeg is.

Uitgewerkt ziet dat er zo uit:

Python:

class Hasher(threading.Thread):

    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
        
    def run(self):
        path = self.queue.get()
        
        while path != 'STOP': # Controleer voor einde queue
            try:    
                self.scanFile(path)
            except:
                print "Failed to scan file:", "/n",sys.exc_info()[0]
                
            self.queue.task_done()
            path = self.queue.get()

            
    def scanFile(self, path):
        ftype = magic.from_file(path)
        mtype = magic.from_file(path, mime=True)
        btype = magic.from_buffer(open(path).read(1024))

Maar nogmaals, dit had je allemaal uit de tips en documentatie kunnen halen en snap ook niet waarom je niet op mijn tips hebt gereageerd.

[ Voor 3% gewijzigd door Zeebonk op 18-06-2012 16:21 ]

[Verwijderd]

ValHallASW schreef op maandag 18 juni 2012 @ 15:48:
Je indenting op regel 24 en 28 (en, bij nader inzien: in je hele paste) klopt niet (gebruik je tabs en spaties door elkaar?), en je doet nog steeds een dubbele queue.get() die nergens voor nodig is. Maak voor jezelf anders eerst een een heel basale versie die niets meer doet dan de filename uitprinten.

Je kunt gebruik maken van sys._current_frames + traceback.print_trace om een idee te krijgen waar je deadlock plaatsvindt.

Klopt me code is hier op tweakers niet goed overgekopieerd en geeft de verkeerde tabs weer.

Zeebonk schreef op maandag 18 juni 2012 @ 16:20:
Ik snap niet waarom je nu niet gewoon kijkt naar de tips die er worden gegeven. Daar staat het allemaal in:

Je code loopt vast bij je één na laatste item omdat de je steeds 2x queue.get doet, 1x in run, en 1x in scanFile, in scanfile moet hij weg. Je slaat hierdoor sowieso de helft van je items in de queue over.

Als je dat hebt opgelost krijg je een exception bij de queue.get in run, omdat je timeout zal verstrijken als de queue leeg is en je de exceptie niet afvangt. Om er voor te zorgen dat je niet nodeloos zit te wachten op een timeout als de queue leeg is, is het slim om na alle bestandspaden een 'STOP' token in de queue te stoppen. Door steeds te checken of je 'STOP' uit de queue heb gehaald weet je direct wanneer de queue leeg is.

Uitgewerkt ziet dat er zo uit:

Python:

class Hasher(threading.Thread):

    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue
        
    def run(self):
        path = self.queue.get()
        
        while path != 'STOP': # Controleer voor einde queue
            try:    
                self.scanFile(path)
            except:
                print "Failed to scan file:", "/n",sys.exc_info()[0]
                
            self.queue.task_done()
            path = self.queue.get()

            
    def scanFile(self, path):
        ftype = magic.from_file(path)
        mtype = magic.from_file(path, mime=True)
        btype = magic.from_buffer(open(path).read(1024))

Maar nogmaals, dit had je allemaal uit de tips en documentatie kunnen halen en snap ook niet waarom je niet op mijn tips hebt gereageerd.

Bedankt voor je hulp!! Heb nu jou deel letterlijk gekopieerd in me code.

Laatste ellement hangt hij nog steeds. Omdat de controle met de stop niet goed gaat.
De queue heeft geen STOP ellement in de queue volgens mij en blijft daardoor door gaan...

[ Voor 8% gewijzigd door Verwijderd op 18-06-2012 18:52 . Reden: Edit: Thread hangt ]

[Kajel]

Python kan heel slecht tegen het gebruik van persoonlijke voornaamwoorden waar eigenlijk bezittelijke voornaamwoorden horen.

[Verwijderd]

Kajel schreef op dinsdag 19 juni 2012 @ 00:08:
Python kan heel slecht tegen het gebruik van persoonlijke voornaamwoorden waar eigenlijk bezittelijke voornaamwoorden horen.

Wat word hier nu weer mee bedoelt? Snap er niet heel veel van...