En raison du déménagement des baies serveurs, les services gitlab.irstea.fr et mattermost.irstea.fr seront interrompus le samedi 2 octobre 2021 au matin. Ils devraient revenir à la normale dans la journée.

Unverified Commit 61094506 authored by Julien Veyssier's avatar Julien Veyssier
Browse files

disable multiprocessing on windows in step2


Signed-off-by: default avatarJulien Veyssier <eneiluj@posteo.net>
parent 29a156ac
......@@ -13,7 +13,7 @@
# to keep python2 compatibility
from __future__ import print_function
import string, os, sys, glob, types, time
import string, os, sys, glob, types, time, platform
import numpy as np
try:
import ConfigParser
......@@ -28,8 +28,9 @@ from osgeo.gdalnumeric import *
from gdalconst import *
from osgeo import ogr
import multiprocessing
from multiprocessing import Pool, cpu_count
if platform.system() != 'Windows':
import multiprocessing
from multiprocessing import Pool, cpu_count
MY_ABS_PATH=os.path.abspath(__file__)
MY_DIR=os.path.dirname(MY_ABS_PATH)
......@@ -397,12 +398,16 @@ def cut_streams_at_gauges(directory_out, gauges_reloc_name):
station_lyr = station_shp.CopyLayer(gauges_lyr, gauges_reloc_name)
def processReach(params):
id, configFileDir, nbProc = params
id, configFileDir, nbProc, useMultiProcess = params
# define which grass env we are using
current = multiprocessing.current_process()
processN = int(current._identity[0])
# because multiprocess does not reset workers id numbers and we still use [1, N]
processN = (processN % nbProc) + 1
if useMultiProcess:
current = multiprocessing.current_process()
processN = int(current._identity[0])
# because multiprocess does not reset workers id numbers and we still use [1, N]
processN = (processN % nbProc) + 1
else:
processN = 1
location = 'hru-delin_%s' % (processN)
os.environ['GISRC'] = os.path.join(configFileDir, 'grass_db', 'grassdata', location, '.grassrc')
......@@ -418,6 +423,9 @@ def processReach(params):
'''
def main(parms_file, nbProc, generator=False):
# multiprocessing does not work on windows
if platform.system() == 'Windows':
nbProc = 1
print('---------- HRU-delin Step 2 started ---------------------------------------------')
# TODO remove dirty global variables
......@@ -626,18 +634,25 @@ def main(parms_file, nbProc, generator=False):
importRastersInEnv(rastersForWorkers, grassDbPath, location)
nbReachs = len(reach_ids)
# the locks are here to prevent concurrent terminal tqdm writing
# this is the interesting part, launching N processes in parallel to process basins
if generator:
if platform.system() == 'Windows':
# on windows (qgis or batch) we don't use multiprocessing
params = [(id, configFileDir, nbProc, False) for (i, id) in enumerate(reach_ids)]
results = []
for i, parm in enumerate(params, 1):
results.append(processReach(parm))
yield (i/nbReachs*100)
elif generator:
with Pool(nbProc) as p:
params = [(id, configFileDir, nbProc) for (i, id) in enumerate(reach_ids)]
params = [(id, configFileDir, nbProc, True) for (i, id) in enumerate(reach_ids)]
results = []
for i, _ in enumerate(p.imap_unordered(processReach, params), 1):
results.append(_)
yield (i/nbReachs*100)
else:
# this is the interesting part, launching N processes in parallel to process basins
# the locks are here to prevent concurrent terminal tqdm writing
with Pool(nbProc, initializer=tqdm.set_lock, initargs=(tqdm.get_lock(),)) as p:
params = [(id, configFileDir, nbProc) for (i, id) in enumerate(reach_ids)]
params = [(id, configFileDir, nbProc, True) for (i, id) in enumerate(reach_ids)]
results = list(tqdm(p.imap_unordered(processReach, params),
desc='[main process] get reach id => subbasins id [%s process] ' % nbProc,
total=nbReachs,
......@@ -725,11 +740,15 @@ if __name__ == '__main__':
parms_file = sys.argv[1]
if len(sys.argv) > 2:
nbProcArg = sys.argv[2]
# determine how many processes we can launch
if str(nbProcArg).isnumeric() and int(nbProcArg) > 0:
nbProc = int(nbProcArg)
if platform.system() != 'Windows':
# determine how many processes we can launch
if str(nbProcArg).isnumeric() and int(nbProcArg) > 0:
nbProc = int(nbProcArg)
else:
nbProc = cpu_count()
else:
nbProc = cpu_count()
nbProc = 1
# main is a generator but we don't use it here
for pc in main(parms_file, nbProc, False):
......@@ -741,4 +760,4 @@ if __name__ == '__main__':
pass
else:
from .parallel import buildGrassEnv, buildGrassLocation, exportRasters, importRastersInEnv
from .progressColors import *
from .progressColors import *
\ No newline at end of file
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment