We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[Feature Request] RQ Launcher SSL support
# 🚀 Feature Request
RQ (Redis-Queue) has [support](https://github.com/rq/rq/blob/master/rq/cli/helpers.py#L56) for SSL connections. We can easily port this over to the plugin.

## Motivation
Allows users to use SSL Redis connections in the RQ Hydra Launcher. e.g. some cloud providers require you to connect with SSL (`rediss` instead of `redis`) and don't let you connect otherwise.

## Pitch
Will open PR 👍🏻

## Additional context
Currently Hydra RQ launcher only supports the following options:
```shell
export REDIS_HOST=localhost
export REDIS_PORT=6379
export REDIS_DB=0
export REDIS_PASSWORD=
python run.py hydra/launcher=rq random_state=0,1,2,3 --multirun
```
→ no way to configure SSL if server requires so.

Therefore, we require an option to configure Redis connections over SSL, e.g.:
```shell
export REDIS_SSL=True
python run.py hydra/launcher=rq random_state=0,1,2,3 --multirun
```

✨

--- 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_rq_launcher/hydra_plugins/hydra_rq_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 from omegaconf import II
7 
8 
9 @dataclass
10 class RedisConf:
11     # host address via REDIS_HOST environment variable, default: localhost
12     host: str = II("oc.env:REDIS_HOST,localhost")
13     # port via REDIS_PORT environment variable, default: 6379
14     port: int = II("oc.env:REDIS_PORT,'6379'")
15     # database via REDIS_DB environment variable, default: 0
16     db: Optional[str] = II("oc.env:REDIS_DB,'0'")
17     # password via REDIS_PASSWORD environment variable, default: no password
18     password: Optional[str] = II("oc.env:REDIS_PASSWORD,null")
19     # switch to run without redis server in single thread, for testing purposes only
20     mock: bool = II("oc.env:REDIS_MOCK,'False'")
21 
22 
23 @dataclass
24 class EnqueueConf:
25     # maximum runtime of the job before it's killed (e.g. "1d" for 1 day, units: d/h/m/s), default: no limit
26     job_timeout: Optional[str] = None
27     # maximum queued time before the job before is discarded (e.g. "1d" for 1 day, units: d/h/m/s), default: no limit
28     ttl: Optional[str] = None
29     # how long successful jobs and their results are kept (e.g. "1d" for 1 day, units: d/h/m/s), default: no limit
30     result_ttl: Optional[str] = None
31     # specifies how long failed jobs are kept (e.g. "1d" for 1 day, units: d/h/m/s), default: no limit
32     failure_ttl: Optional[str] = None
33     # place job at the front of the queue, instead of the back
34     at_front: bool = False
35     # job id, will be overidden automatically by a uuid unless specified explicitly
36     job_id: Optional[str] = None
37     # description, will be overidden automatically unless specified explicitly
38     description: Optional[str] = None
39 
40 
41 @dataclass
42 class RQLauncherConf:
43     _target_: str = "hydra_plugins.hydra_rq_launcher.rq_launcher.RQLauncher"
44     # enqueue configuration
45     enqueue: EnqueueConf = EnqueueConf()
46     # queue name
47     queue: str = "default"
48     # redis configuration
49     redis: RedisConf = RedisConf()
50     # stop after enqueueing by raising custom exception
51     stop_after_enqueue: bool = False
52     # wait time in seconds when polling results
53     wait_polling: float = 1.0
54 
55 
56 ConfigStore.instance().store(
57     group="hydra/launcher", name="rq", node=RQLauncherConf, provider="rq_launcher"
58 )
59
```

Path: `plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py`
Content:
```
1 # Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved
2 import logging
3 import time
4 import uuid
5 from pathlib import Path
6 from typing import Any, Dict, List, Sequence
7 
8 import cloudpickle  # type: ignore
9 from fakeredis import FakeStrictRedis  # type: ignore
10 from hydra.core.hydra_config import HydraConfig
11 from hydra.core.singleton import Singleton
12 from hydra.core.utils import (
13     JobReturn,
14     configure_log,
15     filter_overrides,
16     run_job,
17     setup_globals,
18 )
19 from hydra.types import HydraContext, TaskFunction
20 from omegaconf import DictConfig, OmegaConf, open_dict
21 from redis import Redis
22 from rq import Queue  # type: ignore
23 
24 from .rq_launcher import RQLauncher
25 
26 log = logging.getLogger(__name__)
27 
28 
29 def execute_job(
30     hydra_context: HydraContext,
31     sweep_config: DictConfig,
32     task_function: TaskFunction,
33     singleton_state: Dict[Any, Any],
34 ) -> JobReturn:
35     setup_globals()
36     Singleton.set_state(singleton_state)
37 
38     HydraConfig.instance().set_config(sweep_config)
39 
40     ret = run_job(
41         hydra_context=hydra_context,
42         task_function=task_function,
43         config=sweep_config,
44         job_dir_key="hydra.sweep.dir",
45         job_subdir_key="hydra.sweep.subdir",
46     )
47 
48     return ret
49 
50 
51 def launch(
52     launcher: RQLauncher, job_overrides: Sequence[Sequence[str]], initial_job_idx: int
53 ) -> Sequence[JobReturn]:
54     """
55     :param job_overrides: a List of List<String>, where each inner list is the arguments for one job run.
56     :param initial_job_idx: Initial job idx in batch.
57     :return: an array of return values from run_job with indexes corresponding to the input list indexes.
58     """
59     setup_globals()
60     assert launcher.config is not None
61     assert launcher.task_function is not None
62     assert launcher.hydra_context is not None
63 
64     configure_log(launcher.config.hydra.hydra_logging, launcher.config.hydra.verbose)
65     sweep_dir = Path(str(launcher.config.hydra.sweep.dir))
66     sweep_dir.mkdir(parents=True, exist_ok=True)
67 
68     # RQ configuration
69     rq_cfg = launcher.rq
70 
71     # Redis configuration
72     is_async = not rq_cfg.redis.mock
73     if is_async:
74         connection = Redis(
75             host=rq_cfg.redis.host,
76             port=rq_cfg.redis.port,
77             db=rq_cfg.redis.db,
78             password=rq_cfg.redis.password,
79         )
80     else:
81         log.info("Running in synchronous mode")
82         connection = FakeStrictRedis()
83     queue = Queue(
84         name=rq_cfg.queue,
85         connection=connection,
86         is_async=is_async,
87         serializer=cloudpickle,
88     )
89 
90     # Enqueue jobs
91     jobs: List[Any] = []
92     singleton_state = Singleton.get_state()
93     log.info(
94         f"RQ Launcher is enqueuing {len(job_overrides)} job(s) in queue : {rq_cfg.queue}"
95     )
96     log.info("Sweep output dir : {}".format(sweep_dir))
97     if not sweep_dir.is_absolute():
98         log.warn(
99             "Using relative sweep dir: Please be aware that dir will be relative to where workers are started from."
100         )
101 
102     for idx, overrides in enumerate(job_overrides):
103         description = " ".join(filter_overrides(overrides))
104 
105         enqueue_keywords = OmegaConf.to_container(rq_cfg.enqueue, resolve=True)
106         assert isinstance(enqueue_keywords, dict)
107         if enqueue_keywords["job_timeout"] is None:
108             enqueue_keywords["job_timeout"] = -1
109         if enqueue_keywords["result_ttl"] is None:
110             enqueue_keywords["result_ttl"] = -1
111         if enqueue_keywords["failure_ttl"] is None:
112             enqueue_keywords["failure_ttl"] = -1
113         if enqueue_keywords["job_id"] is None:
114             enqueue_keywords["job_id"] = str(uuid.uuid4())
115         if enqueue_keywords["description"] is None:
116             enqueue_keywords["description"] = description
117 
118         sweep_config = launcher.hydra_context.config_loader.load_sweep_config(
119             launcher.config, list(overrides)
120         )
121         with open_dict(sweep_config):
122             sweep_config.hydra.job.id = enqueue_keywords["job_id"]
123             sweep_config.hydra.job.num = initial_job_idx + idx
124 
125         job = queue.enqueue(
126             execute_job,
127             hydra_context=launcher.hydra_context,
128             sweep_config=sweep_config,
129             task_function=launcher.task_function,
130             singleton_state=singleton_state,
131             **enqueue_keywords,
132         )
133         jobs.append(job)
134 
135         log.info(f"Enqueued {job.get_id()}")
136         log.info(f"\t#{idx+1} : {description}")
137 
138     log.info("Finished enqueuing")
139     if rq_cfg.stop_after_enqueue:
140         raise StopAfterEnqueue
141 
142     log.info(f"Polling job statuses every {rq_cfg.wait_polling} sec")
143     while True:
144         job_ids_done = [
145             job.get_id() for job in jobs if job.get_status() in ["finished", "failed"]
146         ]
147         if len(job_ids_done) == len(jobs):
148             break
149         else:
150             time.sleep(rq_cfg.wait_polling)
151 
152     runs: List[JobReturn] = []
153     for job in jobs:
154         result = job.result if job.result is not None else None
155         runs.append(result)
156 
157     return runs
158 
159 
160 class StopAfterEnqueue(Exception):
161     pass
162
```
--- 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_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py
--- a/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py
+++ b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py
@@ -76,6 +76,8 @@
             port=rq_cfg.redis.port,
             db=rq_cfg.redis.db,
             password=rq_cfg.redis.password,
+            ssl=rq_cfg.redis.ssl,
+            ssl_ca_certs=rq_cfg.redis.ssl_ca_certs,
         )
     else:
         log.info("Running in synchronous mode")
diff --git a/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py
--- a/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py
+++ b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py
@@ -16,6 +16,10 @@
     db: Optional[str] = II("oc.env:REDIS_DB,'0'")
     # password via REDIS_PASSWORD environment variable, default: no password
     password: Optional[str] = II("oc.env:REDIS_PASSWORD,null")
+    # enable/disable SSL, via REDIS_SSL environment variable, default False
+    ssl: bool = II("oc.env:REDIS_SSL,'False'")
+    # path to custom certs, via REDIS_SSL_CA_CERTS env veriable, default none
+    ssl_ca_certs: Optional[str] = II("oc.env:REDIS_SSL_CA_CERTS,null")
     # switch to run without redis server in single thread, for testing purposes only
     mock: bool = II("oc.env:REDIS_MOCK,'False'")

{"golden_diff": "diff --git a/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py\n--- a/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py\n+++ b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py\n@@ -76,6 +76,8 @@\n             port=rq_cfg.redis.port,\n             db=rq_cfg.redis.db,\n             password=rq_cfg.redis.password,\n+            ssl=rq_cfg.redis.ssl,\n+            ssl_ca_certs=rq_cfg.redis.ssl_ca_certs,\n         )\n     else:\n         log.info(\"Running in synchronous mode\")\ndiff --git a/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py\n--- a/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py\n+++ b/plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py\n@@ -16,6 +16,10 @@\n     db: Optional[str] = II(\"oc.env:REDIS_DB,'0'\")\n     # password via REDIS_PASSWORD environment variable, default: no password\n     password: Optional[str] = II(\"oc.env:REDIS_PASSWORD,null\")\n+    # enable/disable SSL, via REDIS_SSL environment variable, default False\n+    ssl: bool = II(\"oc.env:REDIS_SSL,'False'\")\n+    # path to custom certs, via REDIS_SSL_CA_CERTS env veriable, default none\n+    ssl_ca_certs: Optional[str] = II(\"oc.env:REDIS_SSL_CA_CERTS,null\")\n     # switch to run without redis server in single thread, for testing purposes only\n     mock: bool = II(\"oc.env:REDIS_MOCK,'False'\")\n", "issue": "[Feature Request] RQ Launcher SSL support\n# \ud83d\ude80 Feature Request\r\nRQ (Redis-Queue) has [support](https://github.com/rq/rq/blob/master/rq/cli/helpers.py#L56) for SSL connections. We can easily port this over to the plugin.\r\n\r\n## Motivation\r\nAllows users to use SSL Redis connections in the RQ Hydra Launcher. e.g. some cloud providers require you to connect with SSL (`rediss` instead of `redis`) and don't let you connect otherwise.\r\n\r\n## Pitch\r\nWill open PR \ud83d\udc4d\ud83c\udffb\r\n\r\n## Additional context\r\nCurrently Hydra RQ launcher only supports the following options:\r\n```shell\r\nexport REDIS_HOST=localhost\r\nexport REDIS_PORT=6379\r\nexport REDIS_DB=0\r\nexport REDIS_PASSWORD=\r\npython run.py hydra/launcher=rq random_state=0,1,2,3 --multirun\r\n```\r\n\u2192 no way to configure SSL if server requires so.\r\n\r\nTherefore, we require an option to configure Redis connections over SSL, e.g.:\r\n```shell\r\nexport REDIS_SSL=True\r\npython run.py hydra/launcher=rq random_state=0,1,2,3 --multirun\r\n```\r\n\r\n\u2728\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\nfrom omegaconf import II\n\n\n@dataclass\nclass RedisConf:\n    # host address via REDIS_HOST environment variable, default: localhost\n    host: str = II(\"oc.env:REDIS_HOST,localhost\")\n    # port via REDIS_PORT environment variable, default: 6379\n    port: int = II(\"oc.env:REDIS_PORT,'6379'\")\n    # database via REDIS_DB environment variable, default: 0\n    db: Optional[str] = II(\"oc.env:REDIS_DB,'0'\")\n    # password via REDIS_PASSWORD environment variable, default: no password\n    password: Optional[str] = II(\"oc.env:REDIS_PASSWORD,null\")\n    # switch to run without redis server in single thread, for testing purposes only\n    mock: bool = II(\"oc.env:REDIS_MOCK,'False'\")\n\n\n@dataclass\nclass EnqueueConf:\n    # maximum runtime of the job before it's killed (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    job_timeout: Optional[str] = None\n    # maximum queued time before the job before is discarded (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    ttl: Optional[str] = None\n    # how long successful jobs and their results are kept (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    result_ttl: Optional[str] = None\n    # specifies how long failed jobs are kept (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    failure_ttl: Optional[str] = None\n    # place job at the front of the queue, instead of the back\n    at_front: bool = False\n    # job id, will be overidden automatically by a uuid unless specified explicitly\n    job_id: Optional[str] = None\n    # description, will be overidden automatically unless specified explicitly\n    description: Optional[str] = None\n\n\n@dataclass\nclass RQLauncherConf:\n    _target_: str = \"hydra_plugins.hydra_rq_launcher.rq_launcher.RQLauncher\"\n    # enqueue configuration\n    enqueue: EnqueueConf = EnqueueConf()\n    # queue name\n    queue: str = \"default\"\n    # redis configuration\n    redis: RedisConf = RedisConf()\n    # stop after enqueueing by raising custom exception\n    stop_after_enqueue: bool = False\n    # wait time in seconds when polling results\n    wait_polling: float = 1.0\n\n\nConfigStore.instance().store(\n    group=\"hydra/launcher\", name=\"rq\", node=RQLauncherConf, provider=\"rq_launcher\"\n)\n", "path": "plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py"}, {"content": "# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved\nimport logging\nimport time\nimport uuid\nfrom pathlib import Path\nfrom typing import Any, Dict, List, Sequence\n\nimport cloudpickle  # type: ignore\nfrom fakeredis import FakeStrictRedis  # type: ignore\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 HydraContext, TaskFunction\nfrom omegaconf import DictConfig, OmegaConf, open_dict\nfrom redis import Redis\nfrom rq import Queue  # type: ignore\n\nfrom .rq_launcher import RQLauncher\n\nlog = logging.getLogger(__name__)\n\n\ndef execute_job(\n    hydra_context: HydraContext,\n    sweep_config: DictConfig,\n    task_function: TaskFunction,\n    singleton_state: Dict[Any, Any],\n) -> JobReturn:\n    setup_globals()\n    Singleton.set_state(singleton_state)\n\n    HydraConfig.instance().set_config(sweep_config)\n\n    ret = run_job(\n        hydra_context=hydra_context,\n        task_function=task_function,\n        config=sweep_config,\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: RQLauncher, job_overrides: Sequence[Sequence[str]], 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.task_function is not None\n    assert launcher.hydra_context 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    # RQ configuration\n    rq_cfg = launcher.rq\n\n    # Redis configuration\n    is_async = not rq_cfg.redis.mock\n    if is_async:\n        connection = Redis(\n            host=rq_cfg.redis.host,\n            port=rq_cfg.redis.port,\n            db=rq_cfg.redis.db,\n            password=rq_cfg.redis.password,\n        )\n    else:\n        log.info(\"Running in synchronous mode\")\n        connection = FakeStrictRedis()\n    queue = Queue(\n        name=rq_cfg.queue,\n        connection=connection,\n        is_async=is_async,\n        serializer=cloudpickle,\n    )\n\n    # Enqueue jobs\n    jobs: List[Any] = []\n    singleton_state = Singleton.get_state()\n    log.info(\n        f\"RQ Launcher is enqueuing {len(job_overrides)} job(s) in queue : {rq_cfg.queue}\"\n    )\n    log.info(\"Sweep output dir : {}\".format(sweep_dir))\n    if not sweep_dir.is_absolute():\n        log.warn(\n            \"Using relative sweep dir: Please be aware that dir will be relative to where workers are started from.\"\n        )\n\n    for idx, overrides in enumerate(job_overrides):\n        description = \" \".join(filter_overrides(overrides))\n\n        enqueue_keywords = OmegaConf.to_container(rq_cfg.enqueue, resolve=True)\n        assert isinstance(enqueue_keywords, dict)\n        if enqueue_keywords[\"job_timeout\"] is None:\n            enqueue_keywords[\"job_timeout\"] = -1\n        if enqueue_keywords[\"result_ttl\"] is None:\n            enqueue_keywords[\"result_ttl\"] = -1\n        if enqueue_keywords[\"failure_ttl\"] is None:\n            enqueue_keywords[\"failure_ttl\"] = -1\n        if enqueue_keywords[\"job_id\"] is None:\n            enqueue_keywords[\"job_id\"] = str(uuid.uuid4())\n        if enqueue_keywords[\"description\"] is None:\n            enqueue_keywords[\"description\"] = description\n\n        sweep_config = launcher.hydra_context.config_loader.load_sweep_config(\n            launcher.config, list(overrides)\n        )\n        with open_dict(sweep_config):\n            sweep_config.hydra.job.id = enqueue_keywords[\"job_id\"]\n            sweep_config.hydra.job.num = initial_job_idx + idx\n\n        job = queue.enqueue(\n            execute_job,\n            hydra_context=launcher.hydra_context,\n            sweep_config=sweep_config,\n            task_function=launcher.task_function,\n            singleton_state=singleton_state,\n            **enqueue_keywords,\n        )\n        jobs.append(job)\n\n        log.info(f\"Enqueued {job.get_id()}\")\n        log.info(f\"\\t#{idx+1} : {description}\")\n\n    log.info(\"Finished enqueuing\")\n    if rq_cfg.stop_after_enqueue:\n        raise StopAfterEnqueue\n\n    log.info(f\"Polling job statuses every {rq_cfg.wait_polling} sec\")\n    while True:\n        job_ids_done = [\n            job.get_id() for job in jobs if job.get_status() in [\"finished\", \"failed\"]\n        ]\n        if len(job_ids_done) == len(jobs):\n            break\n        else:\n            time.sleep(rq_cfg.wait_polling)\n\n    runs: List[JobReturn] = []\n    for job in jobs:\n        result = job.result if job.result is not None else None\n        runs.append(result)\n\n    return runs\n\n\nclass StopAfterEnqueue(Exception):\n    pass\n", "path": "plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_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\nfrom omegaconf import II\n\n\n@dataclass\nclass RedisConf:\n    # host address via REDIS_HOST environment variable, default: localhost\n    host: str = II(\"oc.env:REDIS_HOST,localhost\")\n    # port via REDIS_PORT environment variable, default: 6379\n    port: int = II(\"oc.env:REDIS_PORT,'6379'\")\n    # database via REDIS_DB environment variable, default: 0\n    db: Optional[str] = II(\"oc.env:REDIS_DB,'0'\")\n    # password via REDIS_PASSWORD environment variable, default: no password\n    password: Optional[str] = II(\"oc.env:REDIS_PASSWORD,null\")\n    # enable/disable SSL, via REDIS_SSL environment variable, default False\n    ssl: bool = II(\"oc.env:REDIS_SSL,'False'\")\n    # path to custom certs, via REDIS_SSL_CA_CERTS env veriable, default none\n    ssl_ca_certs: Optional[str] = II(\"oc.env:REDIS_SSL_CA_CERTS,null\")\n    # switch to run without redis server in single thread, for testing purposes only\n    mock: bool = II(\"oc.env:REDIS_MOCK,'False'\")\n\n\n@dataclass\nclass EnqueueConf:\n    # maximum runtime of the job before it's killed (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    job_timeout: Optional[str] = None\n    # maximum queued time before the job before is discarded (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    ttl: Optional[str] = None\n    # how long successful jobs and their results are kept (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    result_ttl: Optional[str] = None\n    # specifies how long failed jobs are kept (e.g. \"1d\" for 1 day, units: d/h/m/s), default: no limit\n    failure_ttl: Optional[str] = None\n    # place job at the front of the queue, instead of the back\n    at_front: bool = False\n    # job id, will be overidden automatically by a uuid unless specified explicitly\n    job_id: Optional[str] = None\n    # description, will be overidden automatically unless specified explicitly\n    description: Optional[str] = None\n\n\n@dataclass\nclass RQLauncherConf:\n    _target_: str = \"hydra_plugins.hydra_rq_launcher.rq_launcher.RQLauncher\"\n    # enqueue configuration\n    enqueue: EnqueueConf = EnqueueConf()\n    # queue name\n    queue: str = \"default\"\n    # redis configuration\n    redis: RedisConf = RedisConf()\n    # stop after enqueueing by raising custom exception\n    stop_after_enqueue: bool = False\n    # wait time in seconds when polling results\n    wait_polling: float = 1.0\n\n\nConfigStore.instance().store(\n    group=\"hydra/launcher\", name=\"rq\", node=RQLauncherConf, provider=\"rq_launcher\"\n)\n", "path": "plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/config.py"}, {"content": "# Copyright (c) Facebook, Inc. and its affiliates. All Rights Reserved\nimport logging\nimport time\nimport uuid\nfrom pathlib import Path\nfrom typing import Any, Dict, List, Sequence\n\nimport cloudpickle  # type: ignore\nfrom fakeredis import FakeStrictRedis  # type: ignore\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 HydraContext, TaskFunction\nfrom omegaconf import DictConfig, OmegaConf, open_dict\nfrom redis import Redis\nfrom rq import Queue  # type: ignore\n\nfrom .rq_launcher import RQLauncher\n\nlog = logging.getLogger(__name__)\n\n\ndef execute_job(\n    hydra_context: HydraContext,\n    sweep_config: DictConfig,\n    task_function: TaskFunction,\n    singleton_state: Dict[Any, Any],\n) -> JobReturn:\n    setup_globals()\n    Singleton.set_state(singleton_state)\n\n    HydraConfig.instance().set_config(sweep_config)\n\n    ret = run_job(\n        hydra_context=hydra_context,\n        task_function=task_function,\n        config=sweep_config,\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: RQLauncher, job_overrides: Sequence[Sequence[str]], 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.task_function is not None\n    assert launcher.hydra_context 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    # RQ configuration\n    rq_cfg = launcher.rq\n\n    # Redis configuration\n    is_async = not rq_cfg.redis.mock\n    if is_async:\n        connection = Redis(\n            host=rq_cfg.redis.host,\n            port=rq_cfg.redis.port,\n            db=rq_cfg.redis.db,\n            password=rq_cfg.redis.password,\n            ssl=rq_cfg.redis.ssl,\n            ssl_ca_certs=rq_cfg.redis.ssl_ca_certs,\n        )\n    else:\n        log.info(\"Running in synchronous mode\")\n        connection = FakeStrictRedis()\n    queue = Queue(\n        name=rq_cfg.queue,\n        connection=connection,\n        is_async=is_async,\n        serializer=cloudpickle,\n    )\n\n    # Enqueue jobs\n    jobs: List[Any] = []\n    singleton_state = Singleton.get_state()\n    log.info(\n        f\"RQ Launcher is enqueuing {len(job_overrides)} job(s) in queue : {rq_cfg.queue}\"\n    )\n    log.info(\"Sweep output dir : {}\".format(sweep_dir))\n    if not sweep_dir.is_absolute():\n        log.warn(\n            \"Using relative sweep dir: Please be aware that dir will be relative to where workers are started from.\"\n        )\n\n    for idx, overrides in enumerate(job_overrides):\n        description = \" \".join(filter_overrides(overrides))\n\n        enqueue_keywords = OmegaConf.to_container(rq_cfg.enqueue, resolve=True)\n        assert isinstance(enqueue_keywords, dict)\n        if enqueue_keywords[\"job_timeout\"] is None:\n            enqueue_keywords[\"job_timeout\"] = -1\n        if enqueue_keywords[\"result_ttl\"] is None:\n            enqueue_keywords[\"result_ttl\"] = -1\n        if enqueue_keywords[\"failure_ttl\"] is None:\n            enqueue_keywords[\"failure_ttl\"] = -1\n        if enqueue_keywords[\"job_id\"] is None:\n            enqueue_keywords[\"job_id\"] = str(uuid.uuid4())\n        if enqueue_keywords[\"description\"] is None:\n            enqueue_keywords[\"description\"] = description\n\n        sweep_config = launcher.hydra_context.config_loader.load_sweep_config(\n            launcher.config, list(overrides)\n        )\n        with open_dict(sweep_config):\n            sweep_config.hydra.job.id = enqueue_keywords[\"job_id\"]\n            sweep_config.hydra.job.num = initial_job_idx + idx\n\n        job = queue.enqueue(\n            execute_job,\n            hydra_context=launcher.hydra_context,\n            sweep_config=sweep_config,\n            task_function=launcher.task_function,\n            singleton_state=singleton_state,\n            **enqueue_keywords,\n        )\n        jobs.append(job)\n\n        log.info(f\"Enqueued {job.get_id()}\")\n        log.info(f\"\\t#{idx+1} : {description}\")\n\n    log.info(\"Finished enqueuing\")\n    if rq_cfg.stop_after_enqueue:\n        raise StopAfterEnqueue\n\n    log.info(f\"Polling job statuses every {rq_cfg.wait_polling} sec\")\n    while True:\n        job_ids_done = [\n            job.get_id() for job in jobs if job.get_status() in [\"finished\", \"failed\"]\n        ]\n        if len(job_ids_done) == len(jobs):\n            break\n        else:\n            time.sleep(rq_cfg.wait_polling)\n\n    runs: List[JobReturn] = []\n    for job in jobs:\n        result = job.result if job.result is not None else None\n        runs.append(result)\n\n    return runs\n\n\nclass StopAfterEnqueue(Exception):\n    pass\n", "path": "plugins/hydra_rq_launcher/hydra_plugins/hydra_rq_launcher/_core.py"}]}