Comment exécuter des fonctions en parallèle ?

Question

J'ai d'abord fait des recherches et je n'ai pas trouvé de réponse à ma question. J'essaie d'exécuter plusieurs fonctions en parallèle dans Python.

J'ai quelque chose comme ça :

files.py

import common #common is a util class that handles all the IO stuff

dir1 = 'C:\folder1'
dir2 = 'C:\folder2'
filename = 'test.txt'
addFiles = [25, 5, 15, 35, 45, 25, 5, 15, 35, 45]

def func1():
   c = common.Common()
   for i in range(len(addFiles)):
       c.createFiles(addFiles[i], filename, dir1)
       c.getFiles(dir1)
       time.sleep(10)
       c.removeFiles(addFiles[i], dir1)
       c.getFiles(dir1)

def func2():
   c = common.Common()
   for i in range(len(addFiles)):
       c.createFiles(addFiles[i], filename, dir2)
       c.getFiles(dir2)
       time.sleep(10)
       c.removeFiles(addFiles[i], dir2)
       c.getFiles(dir2)

Je veux appeler func1 et func2 et les faire fonctionner en même temps. Les fonctions n'interagissent pas entre elles ni sur le même objet. Pour l'instant, je dois attendre que func1 se termine avant que func2 ne démarre. Comment puis-je faire quelque chose comme ci-dessous :

process.py

from files import func1, func2

runBothFunc(func1(), func2())

Je veux pouvoir créer les deux répertoires à peu près en même temps, car je compte chaque minute le nombre de fichiers créés. Si le répertoire n'est pas là, cela perturbera mon timing.

Answer 1

Si vous êtes un utilisateur de Windows et que vous utilisez Python 3, alors ce billet vous aidera à faire de la programmation parallèle en Python.Lorsque vous exécutez une programmation en pool de la bibliothèque multiprocessus habituelle, vous obtiendrez une erreur concernant la fonction principale de votre programme. Ceci est dû au fait que Windows n'a pas de fonctionnalité fork(). L'article ci-dessous donne une solution à ce problème.

http://python.6.x6.nabble.com/Multiprocessing-Pool-woes-td5047050.html

Comme j'utilisais python 3, j'ai modifié le programme un peu comme suit :

from types import FunctionType
import marshal

def _applicable(*args, **kwargs):
  name = kwargs['__pw_name']
  code = marshal.loads(kwargs['__pw_code'])
  gbls = globals() #gbls = marshal.loads(kwargs['__pw_gbls'])
  defs = marshal.loads(kwargs['__pw_defs'])
  clsr = marshal.loads(kwargs['__pw_clsr'])
  fdct = marshal.loads(kwargs['__pw_fdct'])
  func = FunctionType(code, gbls, name, defs, clsr)
  func.fdct = fdct
  del kwargs['__pw_name']
  del kwargs['__pw_code']
  del kwargs['__pw_defs']
  del kwargs['__pw_clsr']
  del kwargs['__pw_fdct']
  return func(*args, **kwargs)

def make_applicable(f, *args, **kwargs):
  if not isinstance(f, FunctionType): raise ValueError('argument must be a function')
  kwargs['__pw_name'] = f.__name__  # edited
  kwargs['__pw_code'] = marshal.dumps(f.__code__)   # edited
  kwargs['__pw_defs'] = marshal.dumps(f.__defaults__)  # edited
  kwargs['__pw_clsr'] = marshal.dumps(f.__closure__)  # edited
  kwargs['__pw_fdct'] = marshal.dumps(f.__dict__)   # edited
  return _applicable, args, kwargs

def _mappable(x):
  x,name,code,defs,clsr,fdct = x
  code = marshal.loads(code)
  gbls = globals() #gbls = marshal.loads(gbls)
  defs = marshal.loads(defs)
  clsr = marshal.loads(clsr)
  fdct = marshal.loads(fdct)
  func = FunctionType(code, gbls, name, defs, clsr)
  func.fdct = fdct
  return func(x)

def make_mappable(f, iterable):
  if not isinstance(f, FunctionType): raise ValueError('argument must be a function')
  name = f.__name__    # edited
  code = marshal.dumps(f.__code__)   # edited
  defs = marshal.dumps(f.__defaults__)  # edited
  clsr = marshal.dumps(f.__closure__)  # edited
  fdct = marshal.dumps(f.__dict__)  # edited
  return _mappable, ((i,name,code,defs,clsr,fdct) for i in iterable)

Après cette fonction, le code du problème ci-dessus est également modifié un peu comme suit :

from multiprocessing import Pool
from poolable import make_applicable, make_mappable

def cube(x):
  return x**3

if __name__ == "__main__":
  pool    = Pool(processes=2)
  results = [pool.apply_async(*make_applicable(cube,x)) for x in range(1,7)]
  print([result.get(timeout=10) for result in results])

Et j'ai obtenu le résultat suivant :

[1, 8, 27, 64, 125, 216]

Je pense que cet article peut être utile à certains utilisateurs de Windows.

Answer 2

Il n'y a aucun moyen de garantir que deux fonctions s'exécuteront en même temps, ce qui semble être votre objectif.

Le mieux que vous puissiez faire est de diviser la fonction en plusieurs étapes, puis d'attendre que les deux se terminent à des points de synchronisation critiques en utilisant la fonction Process.join comme le mentionne la réponse de @aix.

C'est mieux que time.sleep(10) car il n'est pas possible de garantir des horaires exacts. Avec l'attente explicite, vous dites que les fonctions doivent avoir terminé l'exécution de cette étape avant de passer à la suivante, au lieu de supposer que cela sera fait dans les 10 ms, ce qui n'est pas garanti en fonction de ce qui se passe sur la machine.

Answer 3

(à propos de Comment puis-je exécuter simultanément deux fonctions (ou plus) en Python ? )

Con asyncio Les tâches de synchronisation et d'asynchronisation peuvent être exécutées simultanément par :

import asyncio
import time

def function1():
    # performing blocking tasks
    while True:
        print("function 1: blocking task ...")
        time.sleep(1)

async def function2():
    # perform non-blocking tasks
    while True:
        print("function 2: non-blocking task ...")
        await asyncio.sleep(1)

async def main():
    loop = asyncio.get_running_loop()

    await asyncio.gather(
        # https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.loop.run_in_executor
        loop.run_in_executor(None, function1),
        function2(),
    )

if __name__ == '__main__':
    asyncio.run(main())