Wie kann die Speichernutzung im Multithreading gesteuert werden?

Question

Ich verwende Multi-Thread zur Bildverarbeitung.

Es funktioniert gut auf meinem Computer, der genug Speicher (erhöht 2 ~ 3 GB bei der Verarbeitung von vielen Bildern), aber mein Server hat nur 1 GB Speicher und der Code nicht richtig funktioniert.

Manchmal mit Segmentation fault enden, manchmal:

Exception in thread Thread-13: 
Traceback (most recent call last): 
    File "/usr/lib/python2.7/threading.py", line 810, in __bootstrap_inner 
    self.run() 
    File "/usr/lib/python2.7/threading.py", line 763, in run 
    self.__target(*self.__args, **self.__kwargs) 
    File "passportRecognizeNew.py", line 267, in doSomething 
    ... 

Code:

import threading 

def doSomething(image): 
    # picture processing code 
    print("processing over") 

threads = [] 

for i in range(20): 
    thread = threading.Thread(target=doSomething, args=("image",)) 
    threads.append(thread) 

for t in threads: 
    t.setDaemon(True) 
    t.start() 

t.join() 

print("All over") 

Wie diese oder irgendeine Art und Weise zu lösen, die Speichernutzung zu kontrollieren?

Answer 1

Ich denke, Sie sehen dies aus dem falschen Blickwinkel. Ihr Code wird aktiviert n Threads. Diese Threads führen dann Arbeit aus, die Sie für sie definiert haben.

Wenn diese Arbeit erfordert, dass sie eine Menge Speicher zuweisen - was sollte irgendetwas "außerhalb" dieses Kontexts dazu tun? Was soll passieren? Sollten einige der Threads getötet werden? Sollte irgendwo tief in C-Code eine malloc ... nicht passieren ... und dann?

Was ich sage ist: Ihr Problem ist höchstwahrscheinlich, dass Sie einfach zu viele dieser Threads feuern.

So die Antwort ist: nicht versuchen, zu beheben Dinge nach Sie sie brach - besser sicherstellen, dass Sie sie gar nicht brechen:

• vorsichtig Profilierung tun, um Ihre Anwendung zu verstehen ; So können Sie beurteilen, wie viel Speicher ein einzelner Thread benötigt, um seine "Arbeit" zu erledigen
• dann ändern Sie Ihr "Main" -Programm, um die Hardware abzufragen, auf der es läuft (wie: nach verfügbarem Speicher und Anzahl der physischen CPUs suchen verfügbar)
• und anhand dieser Beurteilung, dass die Anzahl der Threads starten, die sollte die oben genannten Hardware-Details angegeben Arbeit

Darüber hinaus: das ist sehr häufig Muster. Der Entwickler hat eine "leistungsstarke" Maschine, an der er arbeitet; und er setzt implizit voraus, dass jedes Zielsystem, das sein Produkt ausführt, dieselben oder bessere Eigenschaften hat. Und das ist einfach nicht wahr.

Mit anderen Worten: Wenn Sie nicht wissen, wie die Hardware Code wie folgt aussieht auf läuft - dann gibt es nur eine vernünftige Sache zu tun: Erstens, dieses Wissen zu erwerben. Danach verschiedene Dinge zu tun, basierend auf realen Daten.

Answer 2

Mit der GhostCat Hilfe, verwende ich folgenden Code, um Speicherbelegung Problem zu lösen.

import Queue 
import threading 
import multiprocessing 
import time 
import psutil 


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

    def run(self): 
     while True: 
      # check available memory 
      virtualMemoryInfo = psutil.virtual_memory() 
      availableMemory = virtualMemoryInfo.available 

      print(str(availableMemory/1025/1024)+"M") 

      if availableMemory > MEMORY_WARNING: 
       # image from queue 
       image = self.queue.get() 

       # do something 
       doSomething(image) 

       # signals to queue job is done 
       self.queue.task_done() 
      else: 
       print("memory warning!") 

def doSomething(image): 
    # picture processing code, cost time and memory 
    print("processing over") 

# After testing, there seems no use to create threads more than CPU_COUNT, 
# execution time is not reduce. 
CPU_COUNT = multiprocessing.cpu_count() 
MEMORY_WARNING = 200*1024*1024 # 200M 

images = ["1.png", "2.png", "3.png", "4.png", "5.png"] 
queue = Queue.Queue() 

def main(): 
    # spawn a pool of threads, and pass them queue instance 
    for i in range(CPU_COUNT): 
     t = ThreadSomething(queue) 
     t.setDaemon(True) 
     t.start() 

    # populate queue with data 
     for image in images: 
      queue.put(image) 

    # wait on the queue until everything has been processed 
    queue.join() 

start = time.time() 
main() 
print 'All over. Elapsed Time: %s' % (time.time() - start) 

Ich benutze psutil Modul verfügbaren Speicher zu erhalten.

Referenzcode: yosemitebandit/ibm_queue.py

Der Code in meiner Frage hat ein Problem der Threads mehr als CPU_COUNT zu schaffen.