We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[Bug] Joblib launcher plugin doesn't properly convert some joblib arguments to integers
# 🐛 Bug
## Description

The following `joblib.Parallel` arguments are currently provided as strings to joblib when they should be integers:
* `pre_dispatch`
* `batch_size`
* `max_nbytes`

## Checklist
- [X] I checked on the latest version of Hydra
- [X] I created a minimal repro

## To reproduce

** Minimal Code/Config snippet to reproduce **

```bash
wget https://raw.githubusercontent.com/facebookresearch/hydra/master/examples/tutorials/basic/your_first_hydra_app/1_simple_cli/my_app.py
python my_app.py hydra.launcher.batch_size=1 hydra/launcher=joblib -m
```

** Stack trace/error message **
```
[2020-10-27 18:20:48,363][HYDRA] Joblib.Parallel(n_jobs=-1,backend=loky,prefer=processes,require=None,verbose=0,timeout=None,pre_dispatch=2*n_jobs,batch_size=1,temp_folder=None,max_nbytes=None,mmap_mode=r) is launching 1 jobs
[2020-10-27 18:20:48,363][HYDRA] Launching jobs, sweep output dir : multirun/2020-10-27/18-20-48
[2020-10-27 18:20:48,363][HYDRA]        #0 :
Traceback (most recent call last):
  File "/Users/odelalleau/src/hydra/hydra/_internal/utils.py", line 207, in run_and_report
    return func()
  File "/Users/odelalleau/src/hydra/hydra/_internal/utils.py", line 364, in <lambda>
    lambda: hydra.multirun(
  File "/Users/odelalleau/src/hydra/hydra/_internal/hydra.py", line 136, in multirun
    return sweeper.sweep(arguments=task_overrides)
  File "/Users/odelalleau/src/hydra/hydra/_internal/core_plugins/basic_sweeper.py", line 154, in sweep
    results = self.launcher.launch(batch, initial_job_idx=initial_job_idx)
  File "/Users/odelalleau/src/hydra/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/joblib_launcher.py", line 45, in launch
    return _core.launch(
  File "/Users/odelalleau/src/hydra/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py", line 89, in launch
    runs = Parallel(**joblib_cfg)(
  File "/Users/odelalleau/.local/lib/python3.8/site-packages/joblib/parallel.py", line 715, in __init__
    raise ValueError(
ValueError: batch_size must be 'auto' or a positive integer, got: '1'
```

(with `max_nbytes` it also crashes due to the string issue, `pre_dispatch` actually runs but not sure what happens exactly since it's given as a string)

## Expected Behavior

Running without error, providing appropriate integer arguments to `joblib.Parallel` when they should be integers (obviously there are situations where they should remain strings).

## System information
- **Hydra Version** :  master
- **Python version** : 3.8.5
- **Virtual environment type and version** : conda 4.8.5
- **Operating system** : MacOS


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py`
Content:
```
1 # Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved
2 from dataclasses import dataclass
3 from typing import Optional
4 
5 from hydra.core.config_store import ConfigStore
6 
7 
8 @dataclass
9 class JobLibLauncherConf:
10     _target_: str = "hydra_plugins.hydra_joblib_launcher.joblib_launcher.JoblibLauncher"
11 
12     # maximum number of concurrently running jobs. if -1, all CPUs are used
13     n_jobs: int = -1
14 
15     # allows to hard-code backend, otherwise inferred based on prefer and require
16     backend: Optional[str] = None
17 
18     # processes or threads, soft hint to choose backend
19     prefer: str = "processes"
20 
21     # null or sharedmem, sharedmem will select thread-based backend
22     require: Optional[str] = None
23 
24     # if greater than zero, prints progress messages
25     verbose: int = 0
26 
27     # timeout limit for each task
28     timeout: Optional[int] = None
29 
30     # number of batches to be pre-dispatched
31     pre_dispatch: str = "2*n_jobs"
32 
33     # number of atomic tasks to dispatch at once to each worker
34     batch_size: str = "auto"
35 
36     # path used for memmapping large arrays for sharing memory with workers
37     temp_folder: Optional[str] = None
38 
39     # thresholds size of arrays that triggers automated memmapping
40     max_nbytes: Optional[str] = None
41 
42     # memmapping mode for numpy arrays passed to workers
43     mmap_mode: str = "r"
44 
45 
46 ConfigStore.instance().store(
47     group="hydra/launcher",
48     name="joblib",
49     node=JobLibLauncherConf,
50     provider="joblib_launcher",
51 )
52
```

Path: `plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py`
Content:
```
1 # Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved
2 import logging
3 from pathlib import Path
4 from typing import Any, Dict, List, Sequence
5 
6 from hydra.core.config_loader import ConfigLoader
7 from hydra.core.hydra_config import HydraConfig
8 from hydra.core.singleton import Singleton
9 from hydra.core.utils import (
10     JobReturn,
11     configure_log,
12     filter_overrides,
13     run_job,
14     setup_globals,
15 )
16 from hydra.types import TaskFunction
17 from joblib import Parallel, delayed  # type: ignore
18 from omegaconf import DictConfig, open_dict
19 
20 from .joblib_launcher import JoblibLauncher
21 
22 log = logging.getLogger(__name__)
23 
24 
25 def execute_job(
26     idx: int,
27     overrides: Sequence[str],
28     config_loader: ConfigLoader,
29     config: DictConfig,
30     task_function: TaskFunction,
31     singleton_state: Dict[Any, Any],
32 ) -> JobReturn:
33     """Calls `run_job` in parallel"""
34     setup_globals()
35     Singleton.set_state(singleton_state)
36 
37     sweep_config = config_loader.load_sweep_config(config, list(overrides))
38     with open_dict(sweep_config):
39         sweep_config.hydra.job.id = "{}_{}".format(sweep_config.hydra.job.name, idx)
40         sweep_config.hydra.job.num = idx
41     HydraConfig.instance().set_config(sweep_config)
42 
43     ret = run_job(
44         config=sweep_config,
45         task_function=task_function,
46         job_dir_key="hydra.sweep.dir",
47         job_subdir_key="hydra.sweep.subdir",
48     )
49 
50     return ret
51 
52 
53 def launch(
54     launcher: JoblibLauncher,
55     job_overrides: Sequence[Sequence[str]],
56     initial_job_idx: int,
57 ) -> Sequence[JobReturn]:
58     """
59     :param job_overrides: a List of List<String>, where each inner list is the arguments for one job run.
60     :param initial_job_idx: Initial job idx in batch.
61     :return: an array of return values from run_job with indexes corresponding to the input list indexes.
62     """
63     setup_globals()
64     assert launcher.config is not None
65     assert launcher.config_loader is not None
66     assert launcher.task_function is not None
67 
68     configure_log(launcher.config.hydra.hydra_logging, launcher.config.hydra.verbose)
69     sweep_dir = Path(str(launcher.config.hydra.sweep.dir))
70     sweep_dir.mkdir(parents=True, exist_ok=True)
71 
72     # Joblib's backend is hard-coded to loky since the threading
73     # backend is incompatible with Hydra
74     joblib_cfg = launcher.joblib
75     joblib_cfg["backend"] = "loky"
76 
77     log.info(
78         "Joblib.Parallel({}) is launching {} jobs".format(
79             ",".join([f"{k}={v}" for k, v in joblib_cfg.items()]),
80             len(job_overrides),
81         )
82     )
83     log.info("Launching jobs, sweep output dir : {}".format(sweep_dir))
84     for idx, overrides in enumerate(job_overrides):
85         log.info("\t#{} : {}".format(idx, " ".join(filter_overrides(overrides))))
86 
87     singleton_state = Singleton.get_state()
88 
89     runs = Parallel(**joblib_cfg)(
90         delayed(execute_job)(
91             initial_job_idx + idx,
92             overrides,
93             launcher.config_loader,
94             launcher.config,
95             launcher.task_function,
96             singleton_state,
97         )
98         for idx, overrides in enumerate(job_overrides)
99     )
100 
101     assert isinstance(runs, List)
102     for run in runs:
103         assert isinstance(run, JobReturn)
104     return runs
105
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py
--- a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py
+++ b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py
@@ -50,6 +50,17 @@
     return ret
 
 
+def process_joblib_cfg(joblib_cfg: Dict[str, Any]) -> None:
+    for k in ["pre_dispatch", "batch_size", "max_nbytes"]:
+        if k in joblib_cfg.keys():
+            try:
+                val = joblib_cfg.get(k)
+                if val:
+                    joblib_cfg[k] = int(val)
+            except ValueError:
+                pass
+
+
 def launch(
     launcher: JoblibLauncher,
     job_overrides: Sequence[Sequence[str]],
@@ -73,6 +84,7 @@
     # backend is incompatible with Hydra
     joblib_cfg = launcher.joblib
     joblib_cfg["backend"] = "loky"
+    process_joblib_cfg(joblib_cfg)
 
     log.info(
         "Joblib.Parallel({}) is launching {} jobs".format(
diff --git a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py
--- a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py
+++ b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py
@@ -24,8 +24,8 @@
     # if greater than zero, prints progress messages
     verbose: int = 0
 
-    # timeout limit for each task
-    timeout: Optional[int] = None
+    # timeout limit for each task. Unit dependent on backend implementation; miliseconds for loky.
+    timeout: Optional[float] = None
 
     # number of batches to be pre-dispatched
     pre_dispatch: str = "2*n_jobs"

{"golden_diff": "diff --git a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py\n--- a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py\n+++ b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py\n@@ -50,6 +50,17 @@\n     return ret\n \n \n+def process_joblib_cfg(joblib_cfg: Dict[str, Any]) -> None:\n+    for k in [\"pre_dispatch\", \"batch_size\", \"max_nbytes\"]:\n+        if k in joblib_cfg.keys():\n+            try:\n+                val = joblib_cfg.get(k)\n+                if val:\n+                    joblib_cfg[k] = int(val)\n+            except ValueError:\n+                pass\n+\n+\n def launch(\n     launcher: JoblibLauncher,\n     job_overrides: Sequence[Sequence[str]],\n@@ -73,6 +84,7 @@\n     # backend is incompatible with Hydra\n     joblib_cfg = launcher.joblib\n     joblib_cfg[\"backend\"] = \"loky\"\n+    process_joblib_cfg(joblib_cfg)\n \n     log.info(\n         \"Joblib.Parallel({}) is launching {} jobs\".format(\ndiff --git a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py\n--- a/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py\n+++ b/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py\n@@ -24,8 +24,8 @@\n     # if greater than zero, prints progress messages\n     verbose: int = 0\n \n-    # timeout limit for each task\n-    timeout: Optional[int] = None\n+    # timeout limit for each task. Unit dependent on backend implementation; miliseconds for loky.\n+    timeout: Optional[float] = None\n \n     # number of batches to be pre-dispatched\n     pre_dispatch: str = \"2*n_jobs\"\n", "issue": "[Bug] Joblib launcher plugin doesn't properly convert some joblib arguments to integers\n# \ud83d\udc1b Bug\r\n## Description\r\n\r\nThe following `joblib.Parallel` arguments are currently provided as strings to joblib when they should be integers:\r\n* `pre_dispatch`\r\n* `batch_size`\r\n* `max_nbytes`\r\n\r\n## Checklist\r\n- [X] I checked on the latest version of Hydra\r\n- [X] I created a minimal repro\r\n\r\n## To reproduce\r\n\r\n** Minimal Code/Config snippet to reproduce **\r\n\r\n```bash\r\nwget https://raw.githubusercontent.com/facebookresearch/hydra/master/examples/tutorials/basic/your_first_hydra_app/1_simple_cli/my_app.py\r\npython my_app.py hydra.launcher.batch_size=1 hydra/launcher=joblib -m\r\n```\r\n\r\n** Stack trace/error message **\r\n```\r\n[2020-10-27 18:20:48,363][HYDRA] Joblib.Parallel(n_jobs=-1,backend=loky,prefer=processes,require=None,verbose=0,timeout=None,pre_dispatch=2*n_jobs,batch_size=1,temp_folder=None,max_nbytes=None,mmap_mode=r) is launching 1 jobs\r\n[2020-10-27 18:20:48,363][HYDRA] Launching jobs, sweep output dir : multirun/2020-10-27/18-20-48\r\n[2020-10-27 18:20:48,363][HYDRA]        #0 :\r\nTraceback (most recent call last):\r\n  File \"/Users/odelalleau/src/hydra/hydra/_internal/utils.py\", line 207, in run_and_report\r\n    return func()\r\n  File \"/Users/odelalleau/src/hydra/hydra/_internal/utils.py\", line 364, in <lambda>\r\n    lambda: hydra.multirun(\r\n  File \"/Users/odelalleau/src/hydra/hydra/_internal/hydra.py\", line 136, in multirun\r\n    return sweeper.sweep(arguments=task_overrides)\r\n  File \"/Users/odelalleau/src/hydra/hydra/_internal/core_plugins/basic_sweeper.py\", line 154, in sweep\r\n    results = self.launcher.launch(batch, initial_job_idx=initial_job_idx)\r\n  File \"/Users/odelalleau/src/hydra/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/joblib_launcher.py\", line 45, in launch\r\n    return _core.launch(\r\n  File \"/Users/odelalleau/src/hydra/plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py\", line 89, in launch\r\n    runs = Parallel(**joblib_cfg)(\r\n  File \"/Users/odelalleau/.local/lib/python3.8/site-packages/joblib/parallel.py\", line 715, in __init__\r\n    raise ValueError(\r\nValueError: batch_size must be 'auto' or a positive integer, got: '1'\r\n```\r\n\r\n(with `max_nbytes` it also crashes due to the string issue, `pre_dispatch` actually runs but not sure what happens exactly since it's given as a string)\r\n\r\n## Expected Behavior\r\n\r\nRunning without error, providing appropriate integer arguments to `joblib.Parallel` when they should be integers (obviously there are situations where they should remain strings).\r\n\r\n## System information\r\n- **Hydra Version** :  master\r\n- **Python version** : 3.8.5\r\n- **Virtual environment type and version** : conda 4.8.5\r\n- **Operating system** : MacOS\r\n\n", "before_files": [{"content": "# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved\nfrom dataclasses import dataclass\nfrom typing import Optional\n\nfrom hydra.core.config_store import ConfigStore\n\n\n@dataclass\nclass JobLibLauncherConf:\n    _target_: str = \"hydra_plugins.hydra_joblib_launcher.joblib_launcher.JoblibLauncher\"\n\n    # maximum number of concurrently running jobs. if -1, all CPUs are used\n    n_jobs: int = -1\n\n    # allows to hard-code backend, otherwise inferred based on prefer and require\n    backend: Optional[str] = None\n\n    # processes or threads, soft hint to choose backend\n    prefer: str = \"processes\"\n\n    # null or sharedmem, sharedmem will select thread-based backend\n    require: Optional[str] = None\n\n    # if greater than zero, prints progress messages\n    verbose: int = 0\n\n    # timeout limit for each task\n    timeout: Optional[int] = None\n\n    # number of batches to be pre-dispatched\n    pre_dispatch: str = \"2*n_jobs\"\n\n    # number of atomic tasks to dispatch at once to each worker\n    batch_size: str = \"auto\"\n\n    # path used for memmapping large arrays for sharing memory with workers\n    temp_folder: Optional[str] = None\n\n    # thresholds size of arrays that triggers automated memmapping\n    max_nbytes: Optional[str] = None\n\n    # memmapping mode for numpy arrays passed to workers\n    mmap_mode: str = \"r\"\n\n\nConfigStore.instance().store(\n    group=\"hydra/launcher\",\n    name=\"joblib\",\n    node=JobLibLauncherConf,\n    provider=\"joblib_launcher\",\n)\n", "path": "plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py"}, {"content": "# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved\nimport logging\nfrom pathlib import Path\nfrom typing import Any, Dict, List, Sequence\n\nfrom hydra.core.config_loader import ConfigLoader\nfrom hydra.core.hydra_config import HydraConfig\nfrom hydra.core.singleton import Singleton\nfrom hydra.core.utils import (\n    JobReturn,\n    configure_log,\n    filter_overrides,\n    run_job,\n    setup_globals,\n)\nfrom hydra.types import TaskFunction\nfrom joblib import Parallel, delayed  # type: ignore\nfrom omegaconf import DictConfig, open_dict\n\nfrom .joblib_launcher import JoblibLauncher\n\nlog = logging.getLogger(__name__)\n\n\ndef execute_job(\n    idx: int,\n    overrides: Sequence[str],\n    config_loader: ConfigLoader,\n    config: DictConfig,\n    task_function: TaskFunction,\n    singleton_state: Dict[Any, Any],\n) -> JobReturn:\n    \"\"\"Calls `run_job` in parallel\"\"\"\n    setup_globals()\n    Singleton.set_state(singleton_state)\n\n    sweep_config = config_loader.load_sweep_config(config, list(overrides))\n    with open_dict(sweep_config):\n        sweep_config.hydra.job.id = \"{}_{}\".format(sweep_config.hydra.job.name, idx)\n        sweep_config.hydra.job.num = idx\n    HydraConfig.instance().set_config(sweep_config)\n\n    ret = run_job(\n        config=sweep_config,\n        task_function=task_function,\n        job_dir_key=\"hydra.sweep.dir\",\n        job_subdir_key=\"hydra.sweep.subdir\",\n    )\n\n    return ret\n\n\ndef launch(\n    launcher: JoblibLauncher,\n    job_overrides: Sequence[Sequence[str]],\n    initial_job_idx: int,\n) -> Sequence[JobReturn]:\n    \"\"\"\n    :param job_overrides: a List of List<String>, where each inner list is the arguments for one job run.\n    :param initial_job_idx: Initial job idx in batch.\n    :return: an array of return values from run_job with indexes corresponding to the input list indexes.\n    \"\"\"\n    setup_globals()\n    assert launcher.config is not None\n    assert launcher.config_loader is not None\n    assert launcher.task_function is not None\n\n    configure_log(launcher.config.hydra.hydra_logging, launcher.config.hydra.verbose)\n    sweep_dir = Path(str(launcher.config.hydra.sweep.dir))\n    sweep_dir.mkdir(parents=True, exist_ok=True)\n\n    # Joblib's backend is hard-coded to loky since the threading\n    # backend is incompatible with Hydra\n    joblib_cfg = launcher.joblib\n    joblib_cfg[\"backend\"] = \"loky\"\n\n    log.info(\n        \"Joblib.Parallel({}) is launching {} jobs\".format(\n            \",\".join([f\"{k}={v}\" for k, v in joblib_cfg.items()]),\n            len(job_overrides),\n        )\n    )\n    log.info(\"Launching jobs, sweep output dir : {}\".format(sweep_dir))\n    for idx, overrides in enumerate(job_overrides):\n        log.info(\"\\t#{} : {}\".format(idx, \" \".join(filter_overrides(overrides))))\n\n    singleton_state = Singleton.get_state()\n\n    runs = Parallel(**joblib_cfg)(\n        delayed(execute_job)(\n            initial_job_idx + idx,\n            overrides,\n            launcher.config_loader,\n            launcher.config,\n            launcher.task_function,\n            singleton_state,\n        )\n        for idx, overrides in enumerate(job_overrides)\n    )\n\n    assert isinstance(runs, List)\n    for run in runs:\n        assert isinstance(run, JobReturn)\n    return runs\n", "path": "plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py"}], "after_files": [{"content": "# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved\nfrom dataclasses import dataclass\nfrom typing import Optional\n\nfrom hydra.core.config_store import ConfigStore\n\n\n@dataclass\nclass JobLibLauncherConf:\n    _target_: str = \"hydra_plugins.hydra_joblib_launcher.joblib_launcher.JoblibLauncher\"\n\n    # maximum number of concurrently running jobs. if -1, all CPUs are used\n    n_jobs: int = -1\n\n    # allows to hard-code backend, otherwise inferred based on prefer and require\n    backend: Optional[str] = None\n\n    # processes or threads, soft hint to choose backend\n    prefer: str = \"processes\"\n\n    # null or sharedmem, sharedmem will select thread-based backend\n    require: Optional[str] = None\n\n    # if greater than zero, prints progress messages\n    verbose: int = 0\n\n    # timeout limit for each task. Unit dependent on backend implementation; miliseconds for loky.\n    timeout: Optional[float] = None\n\n    # number of batches to be pre-dispatched\n    pre_dispatch: str = \"2*n_jobs\"\n\n    # number of atomic tasks to dispatch at once to each worker\n    batch_size: str = \"auto\"\n\n    # path used for memmapping large arrays for sharing memory with workers\n    temp_folder: Optional[str] = None\n\n    # thresholds size of arrays that triggers automated memmapping\n    max_nbytes: Optional[str] = None\n\n    # memmapping mode for numpy arrays passed to workers\n    mmap_mode: str = \"r\"\n\n\nConfigStore.instance().store(\n    group=\"hydra/launcher\",\n    name=\"joblib\",\n    node=JobLibLauncherConf,\n    provider=\"joblib_launcher\",\n)\n", "path": "plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/config.py"}, {"content": "# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved\nimport logging\nfrom pathlib import Path\nfrom typing import Any, Dict, List, Sequence\n\nfrom hydra.core.config_loader import ConfigLoader\nfrom hydra.core.hydra_config import HydraConfig\nfrom hydra.core.singleton import Singleton\nfrom hydra.core.utils import (\n    JobReturn,\n    configure_log,\n    filter_overrides,\n    run_job,\n    setup_globals,\n)\nfrom hydra.types import TaskFunction\nfrom joblib import Parallel, delayed  # type: ignore\nfrom omegaconf import DictConfig, open_dict\n\nfrom .joblib_launcher import JoblibLauncher\n\nlog = logging.getLogger(__name__)\n\n\ndef execute_job(\n    idx: int,\n    overrides: Sequence[str],\n    config_loader: ConfigLoader,\n    config: DictConfig,\n    task_function: TaskFunction,\n    singleton_state: Dict[Any, Any],\n) -> JobReturn:\n    \"\"\"Calls `run_job` in parallel\"\"\"\n    setup_globals()\n    Singleton.set_state(singleton_state)\n\n    sweep_config = config_loader.load_sweep_config(config, list(overrides))\n    with open_dict(sweep_config):\n        sweep_config.hydra.job.id = \"{}_{}\".format(sweep_config.hydra.job.name, idx)\n        sweep_config.hydra.job.num = idx\n    HydraConfig.instance().set_config(sweep_config)\n\n    ret = run_job(\n        config=sweep_config,\n        task_function=task_function,\n        job_dir_key=\"hydra.sweep.dir\",\n        job_subdir_key=\"hydra.sweep.subdir\",\n    )\n\n    return ret\n\n\ndef process_joblib_cfg(joblib_cfg: Dict[str, Any]) -> None:\n    for k in [\"pre_dispatch\", \"batch_size\", \"max_nbytes\"]:\n        if k in joblib_cfg.keys():\n            try:\n                val = joblib_cfg.get(k)\n                if val:\n                    joblib_cfg[k] = int(val)\n            except ValueError:\n                pass\n\n\ndef launch(\n    launcher: JoblibLauncher,\n    job_overrides: Sequence[Sequence[str]],\n    initial_job_idx: int,\n) -> Sequence[JobReturn]:\n    \"\"\"\n    :param job_overrides: a List of List<String>, where each inner list is the arguments for one job run.\n    :param initial_job_idx: Initial job idx in batch.\n    :return: an array of return values from run_job with indexes corresponding to the input list indexes.\n    \"\"\"\n    setup_globals()\n    assert launcher.config is not None\n    assert launcher.config_loader is not None\n    assert launcher.task_function is not None\n\n    configure_log(launcher.config.hydra.hydra_logging, launcher.config.hydra.verbose)\n    sweep_dir = Path(str(launcher.config.hydra.sweep.dir))\n    sweep_dir.mkdir(parents=True, exist_ok=True)\n\n    # Joblib's backend is hard-coded to loky since the threading\n    # backend is incompatible with Hydra\n    joblib_cfg = launcher.joblib\n    joblib_cfg[\"backend\"] = \"loky\"\n    process_joblib_cfg(joblib_cfg)\n\n    log.info(\n        \"Joblib.Parallel({}) is launching {} jobs\".format(\n            \",\".join([f\"{k}={v}\" for k, v in joblib_cfg.items()]),\n            len(job_overrides),\n        )\n    )\n    log.info(\"Launching jobs, sweep output dir : {}\".format(sweep_dir))\n    for idx, overrides in enumerate(job_overrides):\n        log.info(\"\\t#{} : {}\".format(idx, \" \".join(filter_overrides(overrides))))\n\n    singleton_state = Singleton.get_state()\n\n    runs = Parallel(**joblib_cfg)(\n        delayed(execute_job)(\n            initial_job_idx + idx,\n            overrides,\n            launcher.config_loader,\n            launcher.config,\n            launcher.task_function,\n            singleton_state,\n        )\n        for idx, overrides in enumerate(job_overrides)\n    )\n\n    assert isinstance(runs, List)\n    for run in runs:\n        assert isinstance(run, JobReturn)\n    return runs\n", "path": "plugins/hydra_joblib_launcher/hydra_plugins/hydra_joblib_launcher/_core.py"}]}