Voorbeeld van Python multiprocessing

In onze vorige tutorial hebben we geleerd over het Python CSV Voorbeeld. In deze tutorial gaan we Python Multiprocessing leren met voorbeelden.

Python Multiprocessing

Parallelle verwerking krijgt tegenwoordig meer aandacht. Als je nog steeds niet weet over parallelle verwerking, leer dan van wikipedia. Nu CPU-fabrikanten steeds meer kernen aan hun processors toevoegen, is het maken van parallelle code een geweldige manier om de prestaties te verbeteren. Python introduceerde de multiprocessing module om ons parallelle code te laten schrijven. Om de belangrijkste motivatie van deze module te begrijpen, moeten we enkele basisprincipes van parallel programmeren kennen. Na het lezen van dit artikel hopen we dat je enige kennis over dit onderwerp hebt opgedaan.

Python Multiprocessing Process, Queue en Locks

Er zijn genoeg klassen in de Python multiprocessing-module om een parallel programma te bouwen. Onder hen zijn drie basis klassen Process, Queue en Lock. Deze klassen zullen je helpen om een parallel programma te bouwen. Maar voordat we daarover praten, laten we dit onderwerp initiëren met eenvoudige code. Om een parallel programma nuttig te maken, moet je weten hoeveel kernen er in je pc zijn. De Python multiprocessing-module stelt je in staat om dat te weten. De volgende eenvoudige code zal het aantal kernen op je pc afdrukken.

import multiprocessing

print("Number of cpu : ", multiprocessing.cpu_count())

De volgende output kan variëren voor jouw pc. Voor mij is het aantal kernen 8.

Python multiprocessing Process-klasse

De Python multiprocessing Process klasse is een abstractie die een andere Python proces opzet, het de code laat uitvoeren en een manier biedt voor de ouderapplicatie om de uitvoering te controleren. Er zijn twee belangrijke functies die bij de Process klasse horen – start() en join() functie. Eerst moeten we een functie schrijven die door het proces zal worden uitgevoerd. Vervolgens moeten we een procesobject instantiëren. Als we een procesobject maken, gebeurt er niets totdat we het vertellen om te beginnen met verwerken via de start() functie. Dan zal het proces draaien en het resultaat retourneren. Daarna vertellen we het proces om te voltooien via de join() functie. Zonder de join() functie-aanroep blijft het proces inactief en wordt het niet beëindigd. Dus als je veel processen maakt en ze niet beëindigt, kun je te maken krijgen met schaarste aan middelen. Dan moet je ze mogelijk handmatig doden. Een belangrijk ding is, als je een argument wilt doorgeven via het proces, moet je het args sleutelwoordargument gebruiken. De volgende code zal nuttig zijn om het gebruik van de Process klasse te begrijpen.

from multiprocessing import Process


def print_func(continent='Asia'):
    print('The name of continent is : ', continent)

if __name__ == "__main__":  # confirms that the code is under main function
    names = ['America', 'Europe', 'Africa']
    procs = []
    proc = Process(target=print_func)  # instantiating without any argument
    procs.append(proc)
    proc.start()

    # instantiëren van proces met argumenten
    for name in names:
        # print(name)
        proc = Process(target=print_func, args=(name,))
        procs.append(proc)
        proc.start()

    # voltooi de processen
    for proc in procs:
        proc.join()

De output van de volgende code zal zijn:

Python multiprocessing Queue klasse

Je hebt basiskennis over computergegevensstructuren, je kent waarschijnlijk de Wachtrij. De Python Multiprocessing-modules bieden de Queue-klasse die precies een First-In-First-Out-gegevensstructuur is. Ze kunnen elk Python-object van het type pickle opslaan (hoewel eenvoudige objecten het beste zijn) en zijn uiterst nuttig voor het delen van gegevens tussen processen. Wachtrijen zijn vooral handig wanneer ze als parameter worden doorgegeven aan de doelfunctie van een Process om de Process in staat te stellen gegevens te consumeren. Door de put()-functie te gebruiken kunnen we gegevens invoegen in de wachtrij en met get() kunnen we items uit wachtrijen halen. Zie de volgende code voor een snel voorbeeld.

from multiprocessing import Queue

colors = ['red', 'green', 'blue', 'black']
cnt = 1
# het instantiëren van een wachtrij-object
queue = Queue()
print('pushing items to queue:')
for color in colors:
    print('item no: ', cnt, ' ', color)
    queue.put(color)
    cnt += 1

print('\npopping items from queue:')
cnt = 0
while not queue.empty():
    print('item no: ', cnt, ' ', queue.get())
    cnt += 1

Python multiprocessing Lock-klasse

De taak van de Lock-klasse is vrij eenvoudig. Het maakt het mogelijk om code te vergrendelen zodat geen enkel ander proces dezelfde code kan uitvoeren totdat de vergrendeling is vrijgegeven. De taak van de Lock-klasse is dus voornamelijk tweeledig. Eenmaal vergrendeld wordt de acquire()-functie gebruikt en om de vergrendeling vrij te geven wordt de release()-functie gebruikt.

Voorbeeld van Python multiprocessing

In dit voorbeeld van Python multiprocessing zullen we al onze kennis samenbrengen. Stel dat we enkele taken moeten uitvoeren. Om die taak te volbrengen, zullen we verschillende processen gebruiken. Dus zullen we twee wachtrijen bijhouden. De ene bevat de taken en de andere bevat het logboek van voltooide taken. Vervolgens instantiëren we de processen om de taak te voltooien. Let op dat de Python Queue-klasse al gesynchroniseerd is. Dat betekent dat we de Lock-klasse niet hoeven te gebruiken om te voorkomen dat meerdere processen toegang krijgen tot hetzelfde wachtrijobject. Daarom hoeven we in dit geval de Lock-klasse niet te gebruiken. Hieronder staat de implementatie waarbij we taken toevoegen aan de wachtrij, vervolgens processen creëren en starten, en vervolgens join() gebruiken om de processen te voltooien. Ten slotte printen we het logboek vanuit de tweede wachtrij.

from multiprocessing import Lock, Process, Queue, current_process
import time
import queue # imported for using queue.Empty exception


def do_job(tasks_to_accomplish, tasks_that_are_done):
    while True:
        try:
            '''
                try to get task from the queue. get_nowait() function will 
                raise queue.Empty exception if the queue is empty. 
                queue(False) function would do the same task also.
            '''
            task = tasks_to_accomplish.get_nowait()
        except queue.Empty:

            break
        else:
            '''
                if no exception has been raised, add the task completion 
                message to task_that_are_done queue
            '''
            print(task)
            tasks_that_are_done.put(task + ' is done by ' + current_process().name)
            time.sleep(.5)
    return True


def main():
    number_of_task = 10
    number_of_processes = 4
    tasks_to_accomplish = Queue()
    tasks_that_are_done = Queue()
    processes = []

    for i in range(number_of_task):
        tasks_to_accomplish.put("Task no " + str(i))

    # processen creëren
    for w in range(number_of_processes):
        p = Process(target=do_job, args=(tasks_to_accomplish, tasks_that_are_done))
        processes.append(p)
        p.start()

    # processen voltooien
    for p in processes:
        p.join()

    # de uitvoer printen
    while not tasks_that_are_done.empty():
        print(tasks_that_are_done.get())

    return True


if __name__ == '__main__':
    main()

Afhankelijk van het aantal taken zal de code even tijd nodig hebben om de uitvoer te tonen. De uitvoer van de volgende code zal van tijd tot tijd verschillen.

Python multiprocessing Pool

Python multiprocessing Pool kan worden gebruikt voor parallelle uitvoering van een functie over meerdere invoerwaarden, waarbij de invoerdata over processen worden verdeeld (gegevensparallelisme). Hieronder staat een eenvoudig voorbeeld van Python multiprocessing Pool.

from multiprocessing import Pool

import time

work = (["A", 5], ["B", 2], ["C", 1], ["D", 3])


def work_log(work_data):
    print(" Process %s waiting %s seconds" % (work_data[0], work_data[1]))
    time.sleep(int(work_data[1]))
    print(" Process %s Finished." % work_data[0])


def pool_handler():
    p = Pool(2)
    p.map(work_log, work)


if __name__ == '__main__':
    pool_handler()

Onderstaande afbeelding toont de uitvoer van het bovenstaande programma. Let op dat de poolgrootte 2 is, dus er vinden twee uitvoeringen van de functie work_log gelijktijdig plaats. Wanneer een van de functies klaar is met verwerken, neemt deze het volgende argument enzovoort. Dat is alles voor de python multiprocessing-module. Referentie: Officiële documentatie