Let's say I have a simple way to train a model with N cores:
def train(cores = 1):
reg = RandomForestRegressor(n_jobs=cores).fit(X, y)
return reg
and a simple dataset:
X = np.random.random(size=[25_000, 10])
y = np.sum(X, axis = 1)
Training on 2 cores takes 7.6 seconds, and on 8 cores takes 3.4 seconds. So as expected when I train a model 4 times with 8 cores it takes 13.6 seconds:
models = []
for i in range(4):
models.append(train(8))
However when I try to train them concurrently using ProcessPoolExecutor it takes 14.2 seconds (of which 0.2 seconds are the job submission time):
start = time()
models = []
with ProcessPoolExecutor(max_workers=4) as executor:
for i in range(4):
models.append(executor.submit(train, 2))
print("Submission Time:", time() - start)
print(time() - start)
I understand there are overheads and I could potentially try using joblib or some other way to parallelize models. However I expect that if it takes 7.6 seconds on 2 cores, then training 4 on 4 workers (on a machine with 8 cores) would take less than 10 seconds, not 14 seconds.
Is the slow-down really due to processing overhead? Or is there some oversubscription I'm not noticing? Or is it perhaps possible that the models are attempting to utilize the same cores (I used htop and it seems to max out all cores). I'm not quite sure what's going on here and how to remedy it or best train models in parallel. The goal is to train on machines with up to 100+ cores and up to 100+ models in parallel, this won't function with slowdowns of this scale.
Note I also tried joblib which took 14 seconds as well:
start = time()
result = Parallel(n_jobs=4)(delayed(train)(2) for _ in range(4))
print(time() - start)
EDIT Here is another full script this time 16 models either 1 model per core or 8 cores per model sequentially:
import numpy as np
from time import time, sleep
from sklearn.ensemble import RandomForestRegressor
from concurrent.futures import ProcessPoolExecutor
X = np.random.random(size=[25_000, 10])
y = np.sum(X, axis=1)
def train(cores=1):
reg = RandomForestRegressor(n_jobs=cores).fit(X, y)
return reg
start = time()
models = []
with ProcessPoolExecutor(max_workers=8) as executor:
for i in range(16):
models.append(executor.submit(train, 1))
print("Submission Time:", time() - start)
print(time() - start)
start = time()
models = []
for i in range(16):
models.append(train(8))
print(time() - start)
Further EDIT:
I have also tried the loky joblib backend:
from joblib import parallel_backend, Parallel, delayed
start = time()
with parallel_backend("loky", n_jobs=4):
Parallel()(delayed(train)(2, X.copy(), y.copy()) for _ in range(8))
print("joblib loky 8 Models:", time() - start)