We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
using default_scheduler_url as a mounting point (not root `http://address/mount`)  behind proxy not working
<!---
We use GitHub issues mainly for tracking bugs and feature requests.
Questions for how to use luigi can be sent to the mailing list.

Currently, there are no strict procedures or guidelines for submitting issues.
In short, please just use common sense.

Common sense includes this at bare-minimum:

 * search for similar issues posted before creating a new issue.
 * Use markdown to format all code/logs. Issues which are hard to read
   when rendered on GitHub might be closed with a friendly reminder of this.
 * If applicable, reading relevant parts of the documentation.

Also, add steps to reproduce the bug, if applicable. Sample code would be nice too :)

For more information on how to submit valuable contributions,
see https://opensource.guide/how-to-contribute/#how-to-submit-a-contribution
-->
I need to mount `luigi` behind a nginx, for example luigi at `http://address/mount`.
For that I configure:
```ini
[core]
default_scheduler_url=http://address/mount
....
```

GUI is ok and works but, CLI not due to url resolution. it happens there https://github.com/spotify/luigi/blob/c13566418c92de3e4d8d33ead4e7c936511afae1/luigi/rpc.py#L54

To understand what happened:
```python
parsed=urlparse('http://address/mount')
url='/api/add_task'
urljoin(parsed.geturl(), url)
# ==> give 'http://address/api/add_task'
# expected http://address/mount/api/add_task
```
What I must do for working - slash at the end of mount point, no slash for url -:
```python
parsed=urlparse('http://address/mount/')
url='api/add_task'
urljoin(parsed.geturl(), url)
# ==> http://address/mount/api/add_task
```

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `luigi/rpc.py`
Content:
```
1 # -*- coding: utf-8 -*-
2 #
3 # Copyright 2012-2015 Spotify AB
4 #
5 # Licensed under the Apache License, Version 2.0 (the "License");
6 # you may not use this file except in compliance with the License.
7 # You may obtain a copy of the License at
8 #
9 # http://www.apache.org/licenses/LICENSE-2.0
10 #
11 # Unless required by applicable law or agreed to in writing, software
12 # distributed under the License is distributed on an "AS IS" BASIS,
13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 # See the License for the specific language governing permissions and
15 # limitations under the License.
16 #
17 """
18 Implementation of the REST interface between the workers and the server.
19 rpc.py implements the client side of it, server.py implements the server side.
20 See :doc:`/central_scheduler` for more info.
21 """
22 import abc
23 import os
24 import json
25 import logging
26 import socket
27 import base64
28 
29 from urllib.parse import urljoin, urlencode, urlparse
30 from urllib.request import urlopen, Request
31 from urllib.error import URLError
32 
33 from tenacity import Retrying, wait_fixed, stop_after_attempt
34 from luigi import configuration
35 from luigi.scheduler import RPC_METHODS
36 
37 HAS_UNIX_SOCKET = True
38 HAS_REQUESTS = True
39 
40 
41 try:
42     import requests_unixsocket as requests
43 except ImportError:
44     HAS_UNIX_SOCKET = False
45     try:
46         import requests
47     except ImportError:
48         HAS_REQUESTS = False
49 
50 
51 logger = logging.getLogger('luigi-interface')  # TODO: 'interface'?
52 
53 
54 def _urljoin(base, url):
55     """
56     Join relative URLs to base URLs like urllib.parse.urljoin but support
57     arbitrary URIs (esp. 'http+unix://').
58     """
59     parsed = urlparse(base)
60     scheme = parsed.scheme
61     return urlparse(
62         urljoin(parsed._replace(scheme='http').geturl(), url)
63     )._replace(scheme=scheme).geturl()
64 
65 
66 class RPCError(Exception):
67 
68     def __init__(self, message, sub_exception=None):
69         super(RPCError, self).__init__(message)
70         self.sub_exception = sub_exception
71 
72 
73 class _FetcherInterface(metaclass=abc.ABCMeta):
74     @abc.abstractmethod
75     def fetch(self, full_url, body, timeout):
76         pass
77 
78     @abc.abstractmethod
79     def close(self):
80         pass
81 
82 
83 class URLLibFetcher(_FetcherInterface):
84     raises = (URLError, socket.timeout)
85 
86     def _create_request(self, full_url, body=None):
87         # when full_url contains basic auth info, extract it and set the Authorization header
88         url = urlparse(full_url)
89         if url.username:
90             # base64 encoding of username:password
91             auth = base64.b64encode('{}:{}'.format(url.username, url.password or '').encode('utf-8'))
92             auth = auth.decode('utf-8')
93             # update full_url and create a request object with the auth header set
94             full_url = url._replace(netloc=url.netloc.split('@', 1)[-1]).geturl()
95             req = Request(full_url)
96             req.add_header('Authorization', 'Basic {}'.format(auth))
97         else:
98             req = Request(full_url)
99 
100         # add the request body
101         if body:
102             req.data = urlencode(body).encode('utf-8')
103 
104         return req
105 
106     def fetch(self, full_url, body, timeout):
107         req = self._create_request(full_url, body=body)
108         return urlopen(req, timeout=timeout).read().decode('utf-8')
109 
110     def close(self):
111         pass
112 
113 
114 class RequestsFetcher(_FetcherInterface):
115     def __init__(self):
116         from requests import exceptions as requests_exceptions
117         self.raises = requests_exceptions.RequestException
118         self.session = requests.Session()
119         self.process_id = os.getpid()
120 
121     def check_pid(self):
122         # if the process id change changed from when the session was created
123         # a new session needs to be setup since requests isn't multiprocessing safe.
124         if os.getpid() != self.process_id:
125             self.session = requests.Session()
126             self.process_id = os.getpid()
127 
128     def fetch(self, full_url, body, timeout):
129         self.check_pid()
130         resp = self.session.post(full_url, data=body, timeout=timeout)
131         resp.raise_for_status()
132         return resp.text
133 
134     def close(self):
135         self.session.close()
136 
137 
138 class RemoteScheduler:
139     """
140     Scheduler proxy object. Talks to a RemoteSchedulerResponder.
141     """
142 
143     def __init__(self, url='http://localhost:8082/', connect_timeout=None):
144         assert not url.startswith('http+unix://') or HAS_UNIX_SOCKET, (
145             'You need to install requests-unixsocket for Unix socket support.'
146         )
147 
148         self._url = url.rstrip('/')
149         config = configuration.get_config()
150 
151         if connect_timeout is None:
152             connect_timeout = config.getfloat('core', 'rpc-connect-timeout', 10.0)
153         self._connect_timeout = connect_timeout
154 
155         self._rpc_retry_attempts = config.getint('core', 'rpc-retry-attempts', 3)
156         self._rpc_retry_wait = config.getint('core', 'rpc-retry-wait', 30)
157         self._rpc_log_retries = config.getboolean('core', 'rpc-log-retries', True)
158 
159         if HAS_REQUESTS:
160             self._fetcher = RequestsFetcher()
161         else:
162             self._fetcher = URLLibFetcher()
163 
164     def close(self):
165         self._fetcher.close()
166 
167     def _get_retryer(self):
168         def retry_logging(retry_state):
169             if self._rpc_log_retries:
170                 logger.warning("Failed connecting to remote scheduler %r", self._url, exc_info=True)
171                 logger.info("Retrying attempt %r of %r (max)" % (retry_state.attempt_number + 1, self._rpc_retry_attempts))
172                 logger.info("Wait for %d seconds" % self._rpc_retry_wait)
173 
174         return Retrying(wait=wait_fixed(self._rpc_retry_wait),
175                         stop=stop_after_attempt(self._rpc_retry_attempts),
176                         reraise=True,
177                         after=retry_logging)
178 
179     def _fetch(self, url_suffix, body):
180         full_url = _urljoin(self._url, url_suffix)
181         scheduler_retry = self._get_retryer()
182 
183         try:
184             response = scheduler_retry(self._fetcher.fetch, full_url, body, self._connect_timeout)
185         except self._fetcher.raises as e:
186             raise RPCError(
187                 "Errors (%d attempts) when connecting to remote scheduler %r" %
188                 (self._rpc_retry_attempts, self._url),
189                 e
190             )
191         return response
192 
193     def _request(self, url, data, attempts=3, allow_null=True):
194         body = {'data': json.dumps(data)}
195 
196         for _ in range(attempts):
197             page = self._fetch(url, body)
198             response = json.loads(page)["response"]
199             if allow_null or response is not None:
200                 return response
201         raise RPCError("Received null response from remote scheduler %r" % self._url)
202 
203 
204 for method_name, method in RPC_METHODS.items():
205     setattr(RemoteScheduler, method_name, method)
206
```
--- 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/luigi/rpc.py b/luigi/rpc.py
--- a/luigi/rpc.py
+++ b/luigi/rpc.py
@@ -55,11 +55,12 @@
     """
     Join relative URLs to base URLs like urllib.parse.urljoin but support
     arbitrary URIs (esp. 'http+unix://').
+    base part is fixed or mounted point, every url contains full base part.
     """
     parsed = urlparse(base)
     scheme = parsed.scheme
     return urlparse(
-        urljoin(parsed._replace(scheme='http').geturl(), url)
+        urljoin(parsed._replace(scheme='http').geturl(), parsed.path + (url if url[0] == '/' else '/' + url))
     )._replace(scheme=scheme).geturl()

{"golden_diff": "diff --git a/luigi/rpc.py b/luigi/rpc.py\n--- a/luigi/rpc.py\n+++ b/luigi/rpc.py\n@@ -55,11 +55,12 @@\n     \"\"\"\n     Join relative URLs to base URLs like urllib.parse.urljoin but support\n     arbitrary URIs (esp. 'http+unix://').\n+    base part is fixed or mounted point, every url contains full base part.\n     \"\"\"\n     parsed = urlparse(base)\n     scheme = parsed.scheme\n     return urlparse(\n-        urljoin(parsed._replace(scheme='http').geturl(), url)\n+        urljoin(parsed._replace(scheme='http').geturl(), parsed.path + (url if url[0] == '/' else '/' + url))\n     )._replace(scheme=scheme).geturl()\n", "issue": "using default_scheduler_url as a mounting point (not root `http://address/mount`)  behind proxy not working\n<!---\r\nWe use GitHub issues mainly for tracking bugs and feature requests.\r\nQuestions for how to use luigi can be sent to the mailing list.\r\n\r\nCurrently, there are no strict procedures or guidelines for submitting issues.\r\nIn short, please just use common sense.\r\n\r\nCommon sense includes this at bare-minimum:\r\n\r\n * search for similar issues posted before creating a new issue.\r\n * Use markdown to format all code/logs. Issues which are hard to read\r\n   when rendered on GitHub might be closed with a friendly reminder of this.\r\n * If applicable, reading relevant parts of the documentation.\r\n\r\nAlso, add steps to reproduce the bug, if applicable. Sample code would be nice too :)\r\n\r\nFor more information on how to submit valuable contributions,\r\nsee https://opensource.guide/how-to-contribute/#how-to-submit-a-contribution\r\n-->\r\nI need to mount `luigi` behind a nginx, for example luigi at `http://address/mount`.\r\nFor that I configure:\r\n```ini\r\n[core]\r\ndefault_scheduler_url=http://address/mount\r\n....\r\n```\r\n\r\nGUI is ok and works but, CLI not due to url resolution. it happens there https://github.com/spotify/luigi/blob/c13566418c92de3e4d8d33ead4e7c936511afae1/luigi/rpc.py#L54\r\n\r\nTo understand what happened:\r\n```python\r\nparsed=urlparse('http://address/mount')\r\nurl='/api/add_task'\r\nurljoin(parsed.geturl(), url)\r\n# ==> give 'http://address/api/add_task'\r\n# expected http://address/mount/api/add_task\r\n```\r\nWhat I must do for working - slash at the end of mount point, no slash for url -:\r\n```python\r\nparsed=urlparse('http://address/mount/')\r\nurl='api/add_task'\r\nurljoin(parsed.geturl(), url)\r\n# ==> http://address/mount/api/add_task\r\n```\n", "before_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nImplementation of the REST interface between the workers and the server.\nrpc.py implements the client side of it, server.py implements the server side.\nSee :doc:`/central_scheduler` for more info.\n\"\"\"\nimport abc\nimport os\nimport json\nimport logging\nimport socket\nimport base64\n\nfrom urllib.parse import urljoin, urlencode, urlparse\nfrom urllib.request import urlopen, Request\nfrom urllib.error import URLError\n\nfrom tenacity import Retrying, wait_fixed, stop_after_attempt\nfrom luigi import configuration\nfrom luigi.scheduler import RPC_METHODS\n\nHAS_UNIX_SOCKET = True\nHAS_REQUESTS = True\n\n\ntry:\n    import requests_unixsocket as requests\nexcept ImportError:\n    HAS_UNIX_SOCKET = False\n    try:\n        import requests\n    except ImportError:\n        HAS_REQUESTS = False\n\n\nlogger = logging.getLogger('luigi-interface')  # TODO: 'interface'?\n\n\ndef _urljoin(base, url):\n    \"\"\"\n    Join relative URLs to base URLs like urllib.parse.urljoin but support\n    arbitrary URIs (esp. 'http+unix://').\n    \"\"\"\n    parsed = urlparse(base)\n    scheme = parsed.scheme\n    return urlparse(\n        urljoin(parsed._replace(scheme='http').geturl(), url)\n    )._replace(scheme=scheme).geturl()\n\n\nclass RPCError(Exception):\n\n    def __init__(self, message, sub_exception=None):\n        super(RPCError, self).__init__(message)\n        self.sub_exception = sub_exception\n\n\nclass _FetcherInterface(metaclass=abc.ABCMeta):\n    @abc.abstractmethod\n    def fetch(self, full_url, body, timeout):\n        pass\n\n    @abc.abstractmethod\n    def close(self):\n        pass\n\n\nclass URLLibFetcher(_FetcherInterface):\n    raises = (URLError, socket.timeout)\n\n    def _create_request(self, full_url, body=None):\n        # when full_url contains basic auth info, extract it and set the Authorization header\n        url = urlparse(full_url)\n        if url.username:\n            # base64 encoding of username:password\n            auth = base64.b64encode('{}:{}'.format(url.username, url.password or '').encode('utf-8'))\n            auth = auth.decode('utf-8')\n            # update full_url and create a request object with the auth header set\n            full_url = url._replace(netloc=url.netloc.split('@', 1)[-1]).geturl()\n            req = Request(full_url)\n            req.add_header('Authorization', 'Basic {}'.format(auth))\n        else:\n            req = Request(full_url)\n\n        # add the request body\n        if body:\n            req.data = urlencode(body).encode('utf-8')\n\n        return req\n\n    def fetch(self, full_url, body, timeout):\n        req = self._create_request(full_url, body=body)\n        return urlopen(req, timeout=timeout).read().decode('utf-8')\n\n    def close(self):\n        pass\n\n\nclass RequestsFetcher(_FetcherInterface):\n    def __init__(self):\n        from requests import exceptions as requests_exceptions\n        self.raises = requests_exceptions.RequestException\n        self.session = requests.Session()\n        self.process_id = os.getpid()\n\n    def check_pid(self):\n        # if the process id change changed from when the session was created\n        # a new session needs to be setup since requests isn't multiprocessing safe.\n        if os.getpid() != self.process_id:\n            self.session = requests.Session()\n            self.process_id = os.getpid()\n\n    def fetch(self, full_url, body, timeout):\n        self.check_pid()\n        resp = self.session.post(full_url, data=body, timeout=timeout)\n        resp.raise_for_status()\n        return resp.text\n\n    def close(self):\n        self.session.close()\n\n\nclass RemoteScheduler:\n    \"\"\"\n    Scheduler proxy object. Talks to a RemoteSchedulerResponder.\n    \"\"\"\n\n    def __init__(self, url='http://localhost:8082/', connect_timeout=None):\n        assert not url.startswith('http+unix://') or HAS_UNIX_SOCKET, (\n            'You need to install requests-unixsocket for Unix socket support.'\n        )\n\n        self._url = url.rstrip('/')\n        config = configuration.get_config()\n\n        if connect_timeout is None:\n            connect_timeout = config.getfloat('core', 'rpc-connect-timeout', 10.0)\n        self._connect_timeout = connect_timeout\n\n        self._rpc_retry_attempts = config.getint('core', 'rpc-retry-attempts', 3)\n        self._rpc_retry_wait = config.getint('core', 'rpc-retry-wait', 30)\n        self._rpc_log_retries = config.getboolean('core', 'rpc-log-retries', True)\n\n        if HAS_REQUESTS:\n            self._fetcher = RequestsFetcher()\n        else:\n            self._fetcher = URLLibFetcher()\n\n    def close(self):\n        self._fetcher.close()\n\n    def _get_retryer(self):\n        def retry_logging(retry_state):\n            if self._rpc_log_retries:\n                logger.warning(\"Failed connecting to remote scheduler %r\", self._url, exc_info=True)\n                logger.info(\"Retrying attempt %r of %r (max)\" % (retry_state.attempt_number + 1, self._rpc_retry_attempts))\n                logger.info(\"Wait for %d seconds\" % self._rpc_retry_wait)\n\n        return Retrying(wait=wait_fixed(self._rpc_retry_wait),\n                        stop=stop_after_attempt(self._rpc_retry_attempts),\n                        reraise=True,\n                        after=retry_logging)\n\n    def _fetch(self, url_suffix, body):\n        full_url = _urljoin(self._url, url_suffix)\n        scheduler_retry = self._get_retryer()\n\n        try:\n            response = scheduler_retry(self._fetcher.fetch, full_url, body, self._connect_timeout)\n        except self._fetcher.raises as e:\n            raise RPCError(\n                \"Errors (%d attempts) when connecting to remote scheduler %r\" %\n                (self._rpc_retry_attempts, self._url),\n                e\n            )\n        return response\n\n    def _request(self, url, data, attempts=3, allow_null=True):\n        body = {'data': json.dumps(data)}\n\n        for _ in range(attempts):\n            page = self._fetch(url, body)\n            response = json.loads(page)[\"response\"]\n            if allow_null or response is not None:\n                return response\n        raise RPCError(\"Received null response from remote scheduler %r\" % self._url)\n\n\nfor method_name, method in RPC_METHODS.items():\n    setattr(RemoteScheduler, method_name, method)\n", "path": "luigi/rpc.py"}], "after_files": [{"content": "# -*- coding: utf-8 -*-\n#\n# Copyright 2012-2015 Spotify AB\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n# http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n#\n\"\"\"\nImplementation of the REST interface between the workers and the server.\nrpc.py implements the client side of it, server.py implements the server side.\nSee :doc:`/central_scheduler` for more info.\n\"\"\"\nimport abc\nimport os\nimport json\nimport logging\nimport socket\nimport base64\n\nfrom urllib.parse import urljoin, urlencode, urlparse\nfrom urllib.request import urlopen, Request\nfrom urllib.error import URLError\n\nfrom tenacity import Retrying, wait_fixed, stop_after_attempt\nfrom luigi import configuration\nfrom luigi.scheduler import RPC_METHODS\n\nHAS_UNIX_SOCKET = True\nHAS_REQUESTS = True\n\n\ntry:\n    import requests_unixsocket as requests\nexcept ImportError:\n    HAS_UNIX_SOCKET = False\n    try:\n        import requests\n    except ImportError:\n        HAS_REQUESTS = False\n\n\nlogger = logging.getLogger('luigi-interface')  # TODO: 'interface'?\n\n\ndef _urljoin(base, url):\n    \"\"\"\n    Join relative URLs to base URLs like urllib.parse.urljoin but support\n    arbitrary URIs (esp. 'http+unix://').\n    base part is fixed or mounted point, every url contains full base part.\n    \"\"\"\n    parsed = urlparse(base)\n    scheme = parsed.scheme\n    return urlparse(\n        urljoin(parsed._replace(scheme='http').geturl(), parsed.path + (url if url[0] == '/' else '/' + url))\n    )._replace(scheme=scheme).geturl()\n\n\nclass RPCError(Exception):\n\n    def __init__(self, message, sub_exception=None):\n        super(RPCError, self).__init__(message)\n        self.sub_exception = sub_exception\n\n\nclass _FetcherInterface(metaclass=abc.ABCMeta):\n    @abc.abstractmethod\n    def fetch(self, full_url, body, timeout):\n        pass\n\n    @abc.abstractmethod\n    def close(self):\n        pass\n\n\nclass URLLibFetcher(_FetcherInterface):\n    raises = (URLError, socket.timeout)\n\n    def _create_request(self, full_url, body=None):\n        # when full_url contains basic auth info, extract it and set the Authorization header\n        url = urlparse(full_url)\n        if url.username:\n            # base64 encoding of username:password\n            auth = base64.b64encode('{}:{}'.format(url.username, url.password or '').encode('utf-8'))\n            auth = auth.decode('utf-8')\n            # update full_url and create a request object with the auth header set\n            full_url = url._replace(netloc=url.netloc.split('@', 1)[-1]).geturl()\n            req = Request(full_url)\n            req.add_header('Authorization', 'Basic {}'.format(auth))\n        else:\n            req = Request(full_url)\n\n        # add the request body\n        if body:\n            req.data = urlencode(body).encode('utf-8')\n\n        return req\n\n    def fetch(self, full_url, body, timeout):\n        req = self._create_request(full_url, body=body)\n        return urlopen(req, timeout=timeout).read().decode('utf-8')\n\n    def close(self):\n        pass\n\n\nclass RequestsFetcher(_FetcherInterface):\n    def __init__(self):\n        from requests import exceptions as requests_exceptions\n        self.raises = requests_exceptions.RequestException\n        self.session = requests.Session()\n        self.process_id = os.getpid()\n\n    def check_pid(self):\n        # if the process id change changed from when the session was created\n        # a new session needs to be setup since requests isn't multiprocessing safe.\n        if os.getpid() != self.process_id:\n            self.session = requests.Session()\n            self.process_id = os.getpid()\n\n    def fetch(self, full_url, body, timeout):\n        self.check_pid()\n        resp = self.session.post(full_url, data=body, timeout=timeout)\n        resp.raise_for_status()\n        return resp.text\n\n    def close(self):\n        self.session.close()\n\n\nclass RemoteScheduler:\n    \"\"\"\n    Scheduler proxy object. Talks to a RemoteSchedulerResponder.\n    \"\"\"\n\n    def __init__(self, url='http://localhost:8082/', connect_timeout=None):\n        assert not url.startswith('http+unix://') or HAS_UNIX_SOCKET, (\n            'You need to install requests-unixsocket for Unix socket support.'\n        )\n\n        self._url = url.rstrip('/')\n        config = configuration.get_config()\n\n        if connect_timeout is None:\n            connect_timeout = config.getfloat('core', 'rpc-connect-timeout', 10.0)\n        self._connect_timeout = connect_timeout\n\n        self._rpc_retry_attempts = config.getint('core', 'rpc-retry-attempts', 3)\n        self._rpc_retry_wait = config.getint('core', 'rpc-retry-wait', 30)\n        self._rpc_log_retries = config.getboolean('core', 'rpc-log-retries', True)\n\n        if HAS_REQUESTS:\n            self._fetcher = RequestsFetcher()\n        else:\n            self._fetcher = URLLibFetcher()\n\n    def close(self):\n        self._fetcher.close()\n\n    def _get_retryer(self):\n        def retry_logging(retry_state):\n            if self._rpc_log_retries:\n                logger.warning(\"Failed connecting to remote scheduler %r\", self._url, exc_info=True)\n                logger.info(\"Retrying attempt %r of %r (max)\" % (retry_state.attempt_number + 1, self._rpc_retry_attempts))\n                logger.info(\"Wait for %d seconds\" % self._rpc_retry_wait)\n\n        return Retrying(wait=wait_fixed(self._rpc_retry_wait),\n                        stop=stop_after_attempt(self._rpc_retry_attempts),\n                        reraise=True,\n                        after=retry_logging)\n\n    def _fetch(self, url_suffix, body):\n        full_url = _urljoin(self._url, url_suffix)\n        scheduler_retry = self._get_retryer()\n\n        try:\n            response = scheduler_retry(self._fetcher.fetch, full_url, body, self._connect_timeout)\n        except self._fetcher.raises as e:\n            raise RPCError(\n                \"Errors (%d attempts) when connecting to remote scheduler %r\" %\n                (self._rpc_retry_attempts, self._url),\n                e\n            )\n        return response\n\n    def _request(self, url, data, attempts=3, allow_null=True):\n        body = {'data': json.dumps(data)}\n\n        for _ in range(attempts):\n            page = self._fetch(url, body)\n            response = json.loads(page)[\"response\"]\n            if allow_null or response is not None:\n                return response\n        raise RPCError(\"Received null response from remote scheduler %r\" % self._url)\n\n\nfor method_name, method in RPC_METHODS.items():\n    setattr(RemoteScheduler, method_name, method)\n", "path": "luigi/rpc.py"}]}