python - ajout au même fichier à partir de plusieurs threads

Je suis en train d'écrire une application qui ajoute des lignes d'un même fichier à partir de plusieurs threads.

J'ai un problème dans lequel certaines lignes sont ajoutées sans une nouvelle ligne.

Aucune solution pour cela?

class PathThread(threading.Thread):
    def __init__(self, queue):
        threading.Thread.__init__(self)
        self.queue = queue

    def printfiles(self, p):
        for path, dirs, files in os.walk(p):
            for f in files:
                print(f, file=output)

    def run(self):
        while True:
            path = self.queue.get()
            self.printfiles(path)
            self.queue.task_done()


pathqueue = Queue.Queue()
paths = getThisFromSomeWhere()

output = codecs.open('file', 'a')

# spawn threads
for i in range(0, 5):
    t = PathThread(pathqueue)
    t.setDaemon(True)
    t.start()

# add paths to queue
for path in paths:
    pathqueue.put(path)

# wait for queue to get empty
pathqueue.join()

source d'informationauteur user1251654 | 2012-08-16

  1. 23

    La solution est d'écrire dans le fichier dans un seul thread.

    import Queue  # or queue in Python 3
import threading
class PrintThread(threading.Thread):
def __init__(self, queue):
threading.Thread.__init__(self)
self.queue = queue
def printfiles(self, p):
for path, dirs, files in os.walk(p):
for f in files:
print(f, file=output)
def run(self):
while True:
result = self.queue.get()
self.printfiles(result)
self.queue.task_done()
class ProcessThread(threading.Thread):
def __init__(self, in_queue, out_queue):
threading.Thread.__init__(self)
self.in_queue = in_queue
self.out_queue = out_queue
def run(self):
while True:
path = self.in_queue.get()
result = self.process(path)
self.out_queue.put(result)
self.in_queue.task_done()
def process(self, path):
# Do the processing job here
pathqueue = Queue.Queue()
resultqueue = Queue.Queue()
paths = getThisFromSomeWhere()
output = codecs.open('file', 'a')
# spawn threads to process
for i in range(0, 5):
t = ProcessThread(pathqueue, resultqueue)
t.setDaemon(True)
t.start()
# spawn threads to print
t = PrintThread(resultqueue)
t.setDaemon(True)
t.start()
# add paths to queue
for path in paths:
pathqueue.put(path)
# wait for queue to get empty
pathqueue.join()
resultqueue.join()
  2. 2

    le fait que vous ne verrez jamais pêle-texte sur la même ligne ou de nouvelles lignes dans le milieu d'une ligne est un indice que vous avez réellement n'avez pas besoin de synchroniser ajoutant dans le fichier. le problème est que vous utilisez l'impression d'écrire à un seul descripteur de fichier. je soupçonne print est en train de faire 2 opérations pour le descripteur de fichier en un seul appel et ces opérations sont en compétition entre les threads. fondamentalement print est en train de faire quelque chose comme:

    file_handle.write('whatever_text_you_pass_it')
file_handle.write(os.linesep)

    et parce que les différents threads de le faire simultanément sur le même descripteur de fichier parfois un thread va obtenir dans le premier à écrire et l'autre thread va obtenir alors que dans son premier à écrire et ensuite, vous aurez deux retours chariot dans une rangée. ou vraiment toute permutation de ces.

    la façon la plus simple de contourner ce problème consiste à cesser d'utiliser print et il suffit d'utiliser write directement. essayez quelque chose comme ceci:

    output.write(f + os.linesep)

    cela semble encore un danger pour moi. im pas sûr de ce que gaurantees vous pouvez vous attendre avec tous les threads en utilisant le même descripteur de fichier de l'objet et de la lutte pour la mémoire tampon interne. personnellement id pas de côté l'ensemble de la question et demandez à chaque thread a son propre descripteur de fichier. notez également que cela fonctionne parce que le défaut de la mémoire tampon d'écriture bouffées de chaleur est la ligne de tampon, de sorte que quand il fait une couleur pour le fichier, il se termine sur une os.linesep. pour forcer l'utilisation de la ligne de tampon envoyer un 1 comme troisième argument de open. vous pouvez le tester comme ceci:

    #!/usr/bin/env python
import os
import sys
import threading
def hello(file_name, message, count):
with open(file_name, 'a', 1) as f:
for i in range(0, count):
f.write(message + os.linesep)
if __name__ == '__main__':
#start a file
with open('some.txt', 'w') as f:
f.write('this is the beginning' + os.linesep)
#make 10 threads write a million lines to the same file at the same time
threads = []
for i in range(0, 10):
threads.append(threading.Thread(target=hello, args=('some.txt', 'hey im thread %d' % i, 1000000)))
threads[-1].start()
for t in threads:
t.join()
#check what the heck the file had
uniq_lines = set()
with open('some.txt', 'r') as f:
for l in f:
uniq_lines.add(l)
for u in uniq_lines:
sys.stdout.write(u)

    La sortie ressemble à ceci:

    hey im thread 6
hey im thread 7
hey im thread 9
hey im thread 8
hey im thread 3
this is the beginning
hey im thread 5
hey im thread 4
hey im thread 1
hey im thread 0
hey im thread 2
  3. 0

    Et peut-être plus des sauts de ligne où ils ne devraient pas l'être ?
    vous devriez avoir à l'esprit le fait que le partage de la ressource ne doit pas être consulté par plus d'un thread à un moment ou sinon des conséquences imprévisibles peuvent se produire. ( il est appelé à l'aide des 'opérations atomiques" alors que l'utilisation de threads )
    Jetez un oeil à cette page pour un peu d'intuition.

    Fil De Synchronisation