We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
increase rpc api version and check it
due to the locking changes, the rpc api slightly changed and this leads to borgbackup crashing if server and client are different.


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `borg/remote.py`
Content:
```
1 import errno
2 import fcntl
3 import os
4 import select
5 import shlex
6 from subprocess import Popen, PIPE
7 import sys
8 import tempfile
9 import traceback
10 
11 from . import __version__
12 
13 from .helpers import Error, IntegrityError
14 from .repository import Repository
15 
16 import msgpack
17 
18 BUFSIZE = 10 * 1024 * 1024
19 
20 
21 class ConnectionClosed(Error):
22     """Connection closed by remote host"""
23 
24 
25 class ConnectionClosedWithHint(ConnectionClosed):
26     """Connection closed by remote host. {}"""
27 
28 
29 class PathNotAllowed(Error):
30     """Repository path not allowed"""
31 
32 
33 class InvalidRPCMethod(Error):
34     """RPC method {} is not valid"""
35 
36 
37 class RepositoryServer:  # pragma: no cover
38     rpc_methods = (
39         '__len__',
40         'check',
41         'commit',
42         'delete',
43         'destroy',
44         'get',
45         'list',
46         'negotiate',
47         'open',
48         'put',
49         'repair',
50         'rollback',
51         'save_key',
52         'load_key',
53         'break_lock',
54     )
55 
56     def __init__(self, restrict_to_paths):
57         self.repository = None
58         self.restrict_to_paths = restrict_to_paths
59 
60     def serve(self):
61         stdin_fd = sys.stdin.fileno()
62         stdout_fd = sys.stdout.fileno()
63         # Make stdin non-blocking
64         fl = fcntl.fcntl(stdin_fd, fcntl.F_GETFL)
65         fcntl.fcntl(stdin_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)
66         # Make stdout blocking
67         fl = fcntl.fcntl(stdout_fd, fcntl.F_GETFL)
68         fcntl.fcntl(stdout_fd, fcntl.F_SETFL, fl & ~os.O_NONBLOCK)
69         unpacker = msgpack.Unpacker(use_list=False)
70         while True:
71             r, w, es = select.select([stdin_fd], [], [], 10)
72             if r:
73                 data = os.read(stdin_fd, BUFSIZE)
74                 if not data:
75                     return
76                 unpacker.feed(data)
77                 for unpacked in unpacker:
78                     if not (isinstance(unpacked, tuple) and len(unpacked) == 4):
79                         raise Exception("Unexpected RPC data format.")
80                     type, msgid, method, args = unpacked
81                     method = method.decode('ascii')
82                     try:
83                         if method not in self.rpc_methods:
84                             raise InvalidRPCMethod(method)
85                         try:
86                             f = getattr(self, method)
87                         except AttributeError:
88                             f = getattr(self.repository, method)
89                         res = f(*args)
90                     except BaseException as e:
91                         exc = "Remote Traceback by Borg %s%s%s" % (__version__, os.linesep, traceback.format_exc())
92                         os.write(stdout_fd, msgpack.packb((1, msgid, e.__class__.__name__, exc)))
93                     else:
94                         os.write(stdout_fd, msgpack.packb((1, msgid, None, res)))
95             if es:
96                 return
97 
98     def negotiate(self, versions):
99         return 1
100 
101     def open(self, path, create=False, lock_wait=None, lock=True):
102         path = os.fsdecode(path)
103         if path.startswith('/~'):
104             path = path[1:]
105         path = os.path.realpath(os.path.expanduser(path))
106         if self.restrict_to_paths:
107             for restrict_to_path in self.restrict_to_paths:
108                 if path.startswith(os.path.realpath(restrict_to_path)):
109                     break
110             else:
111                 raise PathNotAllowed(path)
112         self.repository = Repository(path, create, lock_wait=lock_wait, lock=lock)
113         return self.repository.id
114 
115 
116 class RemoteRepository:
117     extra_test_args = []
118     remote_path = 'borg'
119     # default umask, overriden by --umask, defaults to read/write only for owner
120     umask = 0o077
121 
122     class RPCError(Exception):
123         def __init__(self, name):
124             self.name = name
125 
126     def __init__(self, location, create=False, lock_wait=None, lock=True):
127         self.location = location
128         self.preload_ids = []
129         self.msgid = 0
130         self.to_send = b''
131         self.cache = {}
132         self.ignore_responses = set()
133         self.responses = {}
134         self.unpacker = msgpack.Unpacker(use_list=False)
135         self.p = None
136         # XXX: ideally, the testsuite would subclass Repository and
137         # override ssh_cmd() instead of this crude hack, although
138         # __testsuite__ is not a valid domain name so this is pretty
139         # safe.
140         if location.host == '__testsuite__':
141             args = [sys.executable, '-m', 'borg.archiver', 'serve' ] + self.extra_test_args
142         else:  # pragma: no cover
143             args = self.ssh_cmd(location)
144         self.p = Popen(args, bufsize=0, stdin=PIPE, stdout=PIPE)
145         self.stdin_fd = self.p.stdin.fileno()
146         self.stdout_fd = self.p.stdout.fileno()
147         fcntl.fcntl(self.stdin_fd, fcntl.F_SETFL, fcntl.fcntl(self.stdin_fd, fcntl.F_GETFL) | os.O_NONBLOCK)
148         fcntl.fcntl(self.stdout_fd, fcntl.F_SETFL, fcntl.fcntl(self.stdout_fd, fcntl.F_GETFL) | os.O_NONBLOCK)
149         self.r_fds = [self.stdout_fd]
150         self.x_fds = [self.stdin_fd, self.stdout_fd]
151 
152         try:
153             version = self.call('negotiate', 1)
154         except ConnectionClosed:
155             raise ConnectionClosedWithHint('Is borg working on the server?')
156         if version != 1:
157             raise Exception('Server insisted on using unsupported protocol version %d' % version)
158         self.id = self.call('open', location.path, create, lock_wait, lock)
159 
160     def __del__(self):
161         self.close()
162 
163     def __repr__(self):
164         return '<%s %s>' % (self.__class__.__name__, self.location.canonical_path())
165 
166     def umask_flag(self):
167         return ['--umask', '%03o' % self.umask]
168 
169     def ssh_cmd(self, location):
170         args = shlex.split(os.environ.get('BORG_RSH', 'ssh'))
171         if location.port:
172             args += ['-p', str(location.port)]
173         if location.user:
174             args.append('%s@%s' % (location.user, location.host))
175         else:
176             args.append('%s' % location.host)
177         # use local umask also for the remote process
178         args += [self.remote_path, 'serve'] + self.umask_flag()
179         return args
180 
181     def call(self, cmd, *args, **kw):
182         for resp in self.call_many(cmd, [args], **kw):
183             return resp
184 
185     def call_many(self, cmd, calls, wait=True, is_preloaded=False):
186         if not calls:
187             return
188 
189         def fetch_from_cache(args):
190             msgid = self.cache[args].pop(0)
191             if not self.cache[args]:
192                 del self.cache[args]
193             return msgid
194 
195         calls = list(calls)
196         waiting_for = []
197         w_fds = [self.stdin_fd]
198         while wait or calls:
199             while waiting_for:
200                 try:
201                     error, res = self.responses.pop(waiting_for[0])
202                     waiting_for.pop(0)
203                     if error:
204                         if error == b'DoesNotExist':
205                             raise Repository.DoesNotExist(self.location.orig)
206                         elif error == b'AlreadyExists':
207                             raise Repository.AlreadyExists(self.location.orig)
208                         elif error == b'CheckNeeded':
209                             raise Repository.CheckNeeded(self.location.orig)
210                         elif error == b'IntegrityError':
211                             raise IntegrityError(res)
212                         elif error == b'PathNotAllowed':
213                             raise PathNotAllowed(*res)
214                         elif error == b'ObjectNotFound':
215                             raise Repository.ObjectNotFound(res[0], self.location.orig)
216                         elif error == b'InvalidRPCMethod':
217                             raise InvalidRPCMethod(*res)
218                         else:
219                             raise self.RPCError(res.decode('utf-8'))
220                     else:
221                         yield res
222                         if not waiting_for and not calls:
223                             return
224                 except KeyError:
225                     break
226             r, w, x = select.select(self.r_fds, w_fds, self.x_fds, 1)
227             if x:
228                 raise Exception('FD exception occurred')
229             if r:
230                 data = os.read(self.stdout_fd, BUFSIZE)
231                 if not data:
232                     raise ConnectionClosed()
233                 self.unpacker.feed(data)
234                 for unpacked in self.unpacker:
235                     if not (isinstance(unpacked, tuple) and len(unpacked) == 4):
236                         raise Exception("Unexpected RPC data format.")
237                     type, msgid, error, res = unpacked
238                     if msgid in self.ignore_responses:
239                         self.ignore_responses.remove(msgid)
240                     else:
241                         self.responses[msgid] = error, res
242             if w:
243                 while not self.to_send and (calls or self.preload_ids) and len(waiting_for) < 100:
244                     if calls:
245                         if is_preloaded:
246                             if calls[0] in self.cache:
247                                 waiting_for.append(fetch_from_cache(calls.pop(0)))
248                         else:
249                             args = calls.pop(0)
250                             if cmd == 'get' and args in self.cache:
251                                 waiting_for.append(fetch_from_cache(args))
252                             else:
253                                 self.msgid += 1
254                                 waiting_for.append(self.msgid)
255                                 self.to_send = msgpack.packb((1, self.msgid, cmd, args))
256                     if not self.to_send and self.preload_ids:
257                         args = (self.preload_ids.pop(0),)
258                         self.msgid += 1
259                         self.cache.setdefault(args, []).append(self.msgid)
260                         self.to_send = msgpack.packb((1, self.msgid, cmd, args))
261 
262                 if self.to_send:
263                     try:
264                         self.to_send = self.to_send[os.write(self.stdin_fd, self.to_send):]
265                     except OSError as e:
266                         # io.write might raise EAGAIN even though select indicates
267                         # that the fd should be writable
268                         if e.errno != errno.EAGAIN:
269                             raise
270                 if not self.to_send and not (calls or self.preload_ids):
271                     w_fds = []
272         self.ignore_responses |= set(waiting_for)
273 
274     def check(self, repair=False):
275         return self.call('check', repair)
276 
277     def commit(self, *args):
278         return self.call('commit')
279 
280     def rollback(self, *args):
281         return self.call('rollback')
282 
283     def destroy(self):
284         return self.call('destroy')
285 
286     def __len__(self):
287         return self.call('__len__')
288 
289     def list(self, limit=None, marker=None):
290         return self.call('list', limit, marker)
291 
292     def get(self, id_):
293         for resp in self.get_many([id_]):
294             return resp
295 
296     def get_many(self, ids, is_preloaded=False):
297         for resp in self.call_many('get', [(id_,) for id_ in ids], is_preloaded=is_preloaded):
298             yield resp
299 
300     def put(self, id_, data, wait=True):
301         return self.call('put', id_, data, wait=wait)
302 
303     def delete(self, id_, wait=True):
304         return self.call('delete', id_, wait=wait)
305 
306     def save_key(self, keydata):
307         return self.call('save_key', keydata)
308 
309     def load_key(self):
310         return self.call('load_key')
311 
312     def break_lock(self):
313         return self.call('break_lock')
314 
315     def close(self):
316         if self.p:
317             self.p.stdin.close()
318             self.p.stdout.close()
319             self.p.wait()
320             self.p = None
321 
322     def preload(self, ids):
323         self.preload_ids += ids
324 
325 
326 class RepositoryCache:
327     """A caching Repository wrapper
328 
329     Caches Repository GET operations using a local temporary Repository.
330     """
331     def __init__(self, repository):
332         self.repository = repository
333         tmppath = tempfile.mkdtemp(prefix='borg-tmp')
334         self.caching_repo = Repository(tmppath, create=True, exclusive=True)
335 
336     def __del__(self):
337         self.caching_repo.destroy()
338 
339     def get(self, key):
340         return next(self.get_many([key]))
341 
342     def get_many(self, keys):
343         unknown_keys = [key for key in keys if key not in self.caching_repo]
344         repository_iterator = zip(unknown_keys, self.repository.get_many(unknown_keys))
345         for key in keys:
346             try:
347                 yield self.caching_repo.get(key)
348             except Repository.ObjectNotFound:
349                 for key_, data in repository_iterator:
350                     if key_ == key:
351                         self.caching_repo.put(key, data)
352                         yield data
353                         break
354         # Consume any pending requests
355         for _ in repository_iterator:
356             pass
357 
358 
359 def cache_if_remote(repository):
360     if isinstance(repository, RemoteRepository):
361         return RepositoryCache(repository)
362     return repository
363
```
--- 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/borg/remote.py b/borg/remote.py
--- a/borg/remote.py
+++ b/borg/remote.py
@@ -15,6 +15,8 @@
 
 import msgpack
 
+RPC_PROTOCOL_VERSION = 2
+
 BUFSIZE = 10 * 1024 * 1024
 
 
@@ -96,7 +98,7 @@
                 return
 
     def negotiate(self, versions):
-        return 1
+        return RPC_PROTOCOL_VERSION
 
     def open(self, path, create=False, lock_wait=None, lock=True):
         path = os.fsdecode(path)
@@ -150,10 +152,10 @@
         self.x_fds = [self.stdin_fd, self.stdout_fd]
 
         try:
-            version = self.call('negotiate', 1)
+            version = self.call('negotiate', RPC_PROTOCOL_VERSION)
         except ConnectionClosed:
             raise ConnectionClosedWithHint('Is borg working on the server?')
-        if version != 1:
+        if version != RPC_PROTOCOL_VERSION:
             raise Exception('Server insisted on using unsupported protocol version %d' % version)
         self.id = self.call('open', location.path, create, lock_wait, lock)

{"golden_diff": "diff --git a/borg/remote.py b/borg/remote.py\n--- a/borg/remote.py\n+++ b/borg/remote.py\n@@ -15,6 +15,8 @@\n \n import msgpack\n \n+RPC_PROTOCOL_VERSION = 2\n+\n BUFSIZE = 10 * 1024 * 1024\n \n \n@@ -96,7 +98,7 @@\n                 return\n \n     def negotiate(self, versions):\n-        return 1\n+        return RPC_PROTOCOL_VERSION\n \n     def open(self, path, create=False, lock_wait=None, lock=True):\n         path = os.fsdecode(path)\n@@ -150,10 +152,10 @@\n         self.x_fds = [self.stdin_fd, self.stdout_fd]\n \n         try:\n-            version = self.call('negotiate', 1)\n+            version = self.call('negotiate', RPC_PROTOCOL_VERSION)\n         except ConnectionClosed:\n             raise ConnectionClosedWithHint('Is borg working on the server?')\n-        if version != 1:\n+        if version != RPC_PROTOCOL_VERSION:\n             raise Exception('Server insisted on using unsupported protocol version %d' % version)\n         self.id = self.call('open', location.path, create, lock_wait, lock)\n", "issue": "increase rpc api version and check it\ndue to the locking changes, the rpc api slightly changed and this leads to borgbackup crashing if server and client are different.\n\n", "before_files": [{"content": "import errno\nimport fcntl\nimport os\nimport select\nimport shlex\nfrom subprocess import Popen, PIPE\nimport sys\nimport tempfile\nimport traceback\n\nfrom . import __version__\n\nfrom .helpers import Error, IntegrityError\nfrom .repository import Repository\n\nimport msgpack\n\nBUFSIZE = 10 * 1024 * 1024\n\n\nclass ConnectionClosed(Error):\n    \"\"\"Connection closed by remote host\"\"\"\n\n\nclass ConnectionClosedWithHint(ConnectionClosed):\n    \"\"\"Connection closed by remote host. {}\"\"\"\n\n\nclass PathNotAllowed(Error):\n    \"\"\"Repository path not allowed\"\"\"\n\n\nclass InvalidRPCMethod(Error):\n    \"\"\"RPC method {} is not valid\"\"\"\n\n\nclass RepositoryServer:  # pragma: no cover\n    rpc_methods = (\n        '__len__',\n        'check',\n        'commit',\n        'delete',\n        'destroy',\n        'get',\n        'list',\n        'negotiate',\n        'open',\n        'put',\n        'repair',\n        'rollback',\n        'save_key',\n        'load_key',\n        'break_lock',\n    )\n\n    def __init__(self, restrict_to_paths):\n        self.repository = None\n        self.restrict_to_paths = restrict_to_paths\n\n    def serve(self):\n        stdin_fd = sys.stdin.fileno()\n        stdout_fd = sys.stdout.fileno()\n        # Make stdin non-blocking\n        fl = fcntl.fcntl(stdin_fd, fcntl.F_GETFL)\n        fcntl.fcntl(stdin_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)\n        # Make stdout blocking\n        fl = fcntl.fcntl(stdout_fd, fcntl.F_GETFL)\n        fcntl.fcntl(stdout_fd, fcntl.F_SETFL, fl & ~os.O_NONBLOCK)\n        unpacker = msgpack.Unpacker(use_list=False)\n        while True:\n            r, w, es = select.select([stdin_fd], [], [], 10)\n            if r:\n                data = os.read(stdin_fd, BUFSIZE)\n                if not data:\n                    return\n                unpacker.feed(data)\n                for unpacked in unpacker:\n                    if not (isinstance(unpacked, tuple) and len(unpacked) == 4):\n                        raise Exception(\"Unexpected RPC data format.\")\n                    type, msgid, method, args = unpacked\n                    method = method.decode('ascii')\n                    try:\n                        if method not in self.rpc_methods:\n                            raise InvalidRPCMethod(method)\n                        try:\n                            f = getattr(self, method)\n                        except AttributeError:\n                            f = getattr(self.repository, method)\n                        res = f(*args)\n                    except BaseException as e:\n                        exc = \"Remote Traceback by Borg %s%s%s\" % (__version__, os.linesep, traceback.format_exc())\n                        os.write(stdout_fd, msgpack.packb((1, msgid, e.__class__.__name__, exc)))\n                    else:\n                        os.write(stdout_fd, msgpack.packb((1, msgid, None, res)))\n            if es:\n                return\n\n    def negotiate(self, versions):\n        return 1\n\n    def open(self, path, create=False, lock_wait=None, lock=True):\n        path = os.fsdecode(path)\n        if path.startswith('/~'):\n            path = path[1:]\n        path = os.path.realpath(os.path.expanduser(path))\n        if self.restrict_to_paths:\n            for restrict_to_path in self.restrict_to_paths:\n                if path.startswith(os.path.realpath(restrict_to_path)):\n                    break\n            else:\n                raise PathNotAllowed(path)\n        self.repository = Repository(path, create, lock_wait=lock_wait, lock=lock)\n        return self.repository.id\n\n\nclass RemoteRepository:\n    extra_test_args = []\n    remote_path = 'borg'\n    # default umask, overriden by --umask, defaults to read/write only for owner\n    umask = 0o077\n\n    class RPCError(Exception):\n        def __init__(self, name):\n            self.name = name\n\n    def __init__(self, location, create=False, lock_wait=None, lock=True):\n        self.location = location\n        self.preload_ids = []\n        self.msgid = 0\n        self.to_send = b''\n        self.cache = {}\n        self.ignore_responses = set()\n        self.responses = {}\n        self.unpacker = msgpack.Unpacker(use_list=False)\n        self.p = None\n        # XXX: ideally, the testsuite would subclass Repository and\n        # override ssh_cmd() instead of this crude hack, although\n        # __testsuite__ is not a valid domain name so this is pretty\n        # safe.\n        if location.host == '__testsuite__':\n            args = [sys.executable, '-m', 'borg.archiver', 'serve' ] + self.extra_test_args\n        else:  # pragma: no cover\n            args = self.ssh_cmd(location)\n        self.p = Popen(args, bufsize=0, stdin=PIPE, stdout=PIPE)\n        self.stdin_fd = self.p.stdin.fileno()\n        self.stdout_fd = self.p.stdout.fileno()\n        fcntl.fcntl(self.stdin_fd, fcntl.F_SETFL, fcntl.fcntl(self.stdin_fd, fcntl.F_GETFL) | os.O_NONBLOCK)\n        fcntl.fcntl(self.stdout_fd, fcntl.F_SETFL, fcntl.fcntl(self.stdout_fd, fcntl.F_GETFL) | os.O_NONBLOCK)\n        self.r_fds = [self.stdout_fd]\n        self.x_fds = [self.stdin_fd, self.stdout_fd]\n\n        try:\n            version = self.call('negotiate', 1)\n        except ConnectionClosed:\n            raise ConnectionClosedWithHint('Is borg working on the server?')\n        if version != 1:\n            raise Exception('Server insisted on using unsupported protocol version %d' % version)\n        self.id = self.call('open', location.path, create, lock_wait, lock)\n\n    def __del__(self):\n        self.close()\n\n    def __repr__(self):\n        return '<%s %s>' % (self.__class__.__name__, self.location.canonical_path())\n\n    def umask_flag(self):\n        return ['--umask', '%03o' % self.umask]\n\n    def ssh_cmd(self, location):\n        args = shlex.split(os.environ.get('BORG_RSH', 'ssh'))\n        if location.port:\n            args += ['-p', str(location.port)]\n        if location.user:\n            args.append('%s@%s' % (location.user, location.host))\n        else:\n            args.append('%s' % location.host)\n        # use local umask also for the remote process\n        args += [self.remote_path, 'serve'] + self.umask_flag()\n        return args\n\n    def call(self, cmd, *args, **kw):\n        for resp in self.call_many(cmd, [args], **kw):\n            return resp\n\n    def call_many(self, cmd, calls, wait=True, is_preloaded=False):\n        if not calls:\n            return\n\n        def fetch_from_cache(args):\n            msgid = self.cache[args].pop(0)\n            if not self.cache[args]:\n                del self.cache[args]\n            return msgid\n\n        calls = list(calls)\n        waiting_for = []\n        w_fds = [self.stdin_fd]\n        while wait or calls:\n            while waiting_for:\n                try:\n                    error, res = self.responses.pop(waiting_for[0])\n                    waiting_for.pop(0)\n                    if error:\n                        if error == b'DoesNotExist':\n                            raise Repository.DoesNotExist(self.location.orig)\n                        elif error == b'AlreadyExists':\n                            raise Repository.AlreadyExists(self.location.orig)\n                        elif error == b'CheckNeeded':\n                            raise Repository.CheckNeeded(self.location.orig)\n                        elif error == b'IntegrityError':\n                            raise IntegrityError(res)\n                        elif error == b'PathNotAllowed':\n                            raise PathNotAllowed(*res)\n                        elif error == b'ObjectNotFound':\n                            raise Repository.ObjectNotFound(res[0], self.location.orig)\n                        elif error == b'InvalidRPCMethod':\n                            raise InvalidRPCMethod(*res)\n                        else:\n                            raise self.RPCError(res.decode('utf-8'))\n                    else:\n                        yield res\n                        if not waiting_for and not calls:\n                            return\n                except KeyError:\n                    break\n            r, w, x = select.select(self.r_fds, w_fds, self.x_fds, 1)\n            if x:\n                raise Exception('FD exception occurred')\n            if r:\n                data = os.read(self.stdout_fd, BUFSIZE)\n                if not data:\n                    raise ConnectionClosed()\n                self.unpacker.feed(data)\n                for unpacked in self.unpacker:\n                    if not (isinstance(unpacked, tuple) and len(unpacked) == 4):\n                        raise Exception(\"Unexpected RPC data format.\")\n                    type, msgid, error, res = unpacked\n                    if msgid in self.ignore_responses:\n                        self.ignore_responses.remove(msgid)\n                    else:\n                        self.responses[msgid] = error, res\n            if w:\n                while not self.to_send and (calls or self.preload_ids) and len(waiting_for) < 100:\n                    if calls:\n                        if is_preloaded:\n                            if calls[0] in self.cache:\n                                waiting_for.append(fetch_from_cache(calls.pop(0)))\n                        else:\n                            args = calls.pop(0)\n                            if cmd == 'get' and args in self.cache:\n                                waiting_for.append(fetch_from_cache(args))\n                            else:\n                                self.msgid += 1\n                                waiting_for.append(self.msgid)\n                                self.to_send = msgpack.packb((1, self.msgid, cmd, args))\n                    if not self.to_send and self.preload_ids:\n                        args = (self.preload_ids.pop(0),)\n                        self.msgid += 1\n                        self.cache.setdefault(args, []).append(self.msgid)\n                        self.to_send = msgpack.packb((1, self.msgid, cmd, args))\n\n                if self.to_send:\n                    try:\n                        self.to_send = self.to_send[os.write(self.stdin_fd, self.to_send):]\n                    except OSError as e:\n                        # io.write might raise EAGAIN even though select indicates\n                        # that the fd should be writable\n                        if e.errno != errno.EAGAIN:\n                            raise\n                if not self.to_send and not (calls or self.preload_ids):\n                    w_fds = []\n        self.ignore_responses |= set(waiting_for)\n\n    def check(self, repair=False):\n        return self.call('check', repair)\n\n    def commit(self, *args):\n        return self.call('commit')\n\n    def rollback(self, *args):\n        return self.call('rollback')\n\n    def destroy(self):\n        return self.call('destroy')\n\n    def __len__(self):\n        return self.call('__len__')\n\n    def list(self, limit=None, marker=None):\n        return self.call('list', limit, marker)\n\n    def get(self, id_):\n        for resp in self.get_many([id_]):\n            return resp\n\n    def get_many(self, ids, is_preloaded=False):\n        for resp in self.call_many('get', [(id_,) for id_ in ids], is_preloaded=is_preloaded):\n            yield resp\n\n    def put(self, id_, data, wait=True):\n        return self.call('put', id_, data, wait=wait)\n\n    def delete(self, id_, wait=True):\n        return self.call('delete', id_, wait=wait)\n\n    def save_key(self, keydata):\n        return self.call('save_key', keydata)\n\n    def load_key(self):\n        return self.call('load_key')\n\n    def break_lock(self):\n        return self.call('break_lock')\n\n    def close(self):\n        if self.p:\n            self.p.stdin.close()\n            self.p.stdout.close()\n            self.p.wait()\n            self.p = None\n\n    def preload(self, ids):\n        self.preload_ids += ids\n\n\nclass RepositoryCache:\n    \"\"\"A caching Repository wrapper\n\n    Caches Repository GET operations using a local temporary Repository.\n    \"\"\"\n    def __init__(self, repository):\n        self.repository = repository\n        tmppath = tempfile.mkdtemp(prefix='borg-tmp')\n        self.caching_repo = Repository(tmppath, create=True, exclusive=True)\n\n    def __del__(self):\n        self.caching_repo.destroy()\n\n    def get(self, key):\n        return next(self.get_many([key]))\n\n    def get_many(self, keys):\n        unknown_keys = [key for key in keys if key not in self.caching_repo]\n        repository_iterator = zip(unknown_keys, self.repository.get_many(unknown_keys))\n        for key in keys:\n            try:\n                yield self.caching_repo.get(key)\n            except Repository.ObjectNotFound:\n                for key_, data in repository_iterator:\n                    if key_ == key:\n                        self.caching_repo.put(key, data)\n                        yield data\n                        break\n        # Consume any pending requests\n        for _ in repository_iterator:\n            pass\n\n\ndef cache_if_remote(repository):\n    if isinstance(repository, RemoteRepository):\n        return RepositoryCache(repository)\n    return repository\n", "path": "borg/remote.py"}], "after_files": [{"content": "import errno\nimport fcntl\nimport os\nimport select\nimport shlex\nfrom subprocess import Popen, PIPE\nimport sys\nimport tempfile\nimport traceback\n\nfrom . import __version__\n\nfrom .helpers import Error, IntegrityError\nfrom .repository import Repository\n\nimport msgpack\n\nRPC_PROTOCOL_VERSION = 2\n\nBUFSIZE = 10 * 1024 * 1024\n\n\nclass ConnectionClosed(Error):\n    \"\"\"Connection closed by remote host\"\"\"\n\n\nclass ConnectionClosedWithHint(ConnectionClosed):\n    \"\"\"Connection closed by remote host. {}\"\"\"\n\n\nclass PathNotAllowed(Error):\n    \"\"\"Repository path not allowed\"\"\"\n\n\nclass InvalidRPCMethod(Error):\n    \"\"\"RPC method {} is not valid\"\"\"\n\n\nclass RepositoryServer:  # pragma: no cover\n    rpc_methods = (\n        '__len__',\n        'check',\n        'commit',\n        'delete',\n        'destroy',\n        'get',\n        'list',\n        'negotiate',\n        'open',\n        'put',\n        'repair',\n        'rollback',\n        'save_key',\n        'load_key',\n        'break_lock',\n    )\n\n    def __init__(self, restrict_to_paths):\n        self.repository = None\n        self.restrict_to_paths = restrict_to_paths\n\n    def serve(self):\n        stdin_fd = sys.stdin.fileno()\n        stdout_fd = sys.stdout.fileno()\n        # Make stdin non-blocking\n        fl = fcntl.fcntl(stdin_fd, fcntl.F_GETFL)\n        fcntl.fcntl(stdin_fd, fcntl.F_SETFL, fl | os.O_NONBLOCK)\n        # Make stdout blocking\n        fl = fcntl.fcntl(stdout_fd, fcntl.F_GETFL)\n        fcntl.fcntl(stdout_fd, fcntl.F_SETFL, fl & ~os.O_NONBLOCK)\n        unpacker = msgpack.Unpacker(use_list=False)\n        while True:\n            r, w, es = select.select([stdin_fd], [], [], 10)\n            if r:\n                data = os.read(stdin_fd, BUFSIZE)\n                if not data:\n                    return\n                unpacker.feed(data)\n                for unpacked in unpacker:\n                    if not (isinstance(unpacked, tuple) and len(unpacked) == 4):\n                        raise Exception(\"Unexpected RPC data format.\")\n                    type, msgid, method, args = unpacked\n                    method = method.decode('ascii')\n                    try:\n                        if method not in self.rpc_methods:\n                            raise InvalidRPCMethod(method)\n                        try:\n                            f = getattr(self, method)\n                        except AttributeError:\n                            f = getattr(self.repository, method)\n                        res = f(*args)\n                    except BaseException as e:\n                        exc = \"Remote Traceback by Borg %s%s%s\" % (__version__, os.linesep, traceback.format_exc())\n                        os.write(stdout_fd, msgpack.packb((1, msgid, e.__class__.__name__, exc)))\n                    else:\n                        os.write(stdout_fd, msgpack.packb((1, msgid, None, res)))\n            if es:\n                return\n\n    def negotiate(self, versions):\n        return RPC_PROTOCOL_VERSION\n\n    def open(self, path, create=False, lock_wait=None, lock=True):\n        path = os.fsdecode(path)\n        if path.startswith('/~'):\n            path = path[1:]\n        path = os.path.realpath(os.path.expanduser(path))\n        if self.restrict_to_paths:\n            for restrict_to_path in self.restrict_to_paths:\n                if path.startswith(os.path.realpath(restrict_to_path)):\n                    break\n            else:\n                raise PathNotAllowed(path)\n        self.repository = Repository(path, create, lock_wait=lock_wait, lock=lock)\n        return self.repository.id\n\n\nclass RemoteRepository:\n    extra_test_args = []\n    remote_path = 'borg'\n    # default umask, overriden by --umask, defaults to read/write only for owner\n    umask = 0o077\n\n    class RPCError(Exception):\n        def __init__(self, name):\n            self.name = name\n\n    def __init__(self, location, create=False, lock_wait=None, lock=True):\n        self.location = location\n        self.preload_ids = []\n        self.msgid = 0\n        self.to_send = b''\n        self.cache = {}\n        self.ignore_responses = set()\n        self.responses = {}\n        self.unpacker = msgpack.Unpacker(use_list=False)\n        self.p = None\n        # XXX: ideally, the testsuite would subclass Repository and\n        # override ssh_cmd() instead of this crude hack, although\n        # __testsuite__ is not a valid domain name so this is pretty\n        # safe.\n        if location.host == '__testsuite__':\n            args = [sys.executable, '-m', 'borg.archiver', 'serve' ] + self.extra_test_args\n        else:  # pragma: no cover\n            args = self.ssh_cmd(location)\n        self.p = Popen(args, bufsize=0, stdin=PIPE, stdout=PIPE)\n        self.stdin_fd = self.p.stdin.fileno()\n        self.stdout_fd = self.p.stdout.fileno()\n        fcntl.fcntl(self.stdin_fd, fcntl.F_SETFL, fcntl.fcntl(self.stdin_fd, fcntl.F_GETFL) | os.O_NONBLOCK)\n        fcntl.fcntl(self.stdout_fd, fcntl.F_SETFL, fcntl.fcntl(self.stdout_fd, fcntl.F_GETFL) | os.O_NONBLOCK)\n        self.r_fds = [self.stdout_fd]\n        self.x_fds = [self.stdin_fd, self.stdout_fd]\n\n        try:\n            version = self.call('negotiate', RPC_PROTOCOL_VERSION)\n        except ConnectionClosed:\n            raise ConnectionClosedWithHint('Is borg working on the server?')\n        if version != RPC_PROTOCOL_VERSION:\n            raise Exception('Server insisted on using unsupported protocol version %d' % version)\n        self.id = self.call('open', location.path, create, lock_wait, lock)\n\n    def __del__(self):\n        self.close()\n\n    def __repr__(self):\n        return '<%s %s>' % (self.__class__.__name__, self.location.canonical_path())\n\n    def umask_flag(self):\n        return ['--umask', '%03o' % self.umask]\n\n    def ssh_cmd(self, location):\n        args = shlex.split(os.environ.get('BORG_RSH', 'ssh'))\n        if location.port:\n            args += ['-p', str(location.port)]\n        if location.user:\n            args.append('%s@%s' % (location.user, location.host))\n        else:\n            args.append('%s' % location.host)\n        # use local umask also for the remote process\n        args += [self.remote_path, 'serve'] + self.umask_flag()\n        return args\n\n    def call(self, cmd, *args, **kw):\n        for resp in self.call_many(cmd, [args], **kw):\n            return resp\n\n    def call_many(self, cmd, calls, wait=True, is_preloaded=False):\n        if not calls:\n            return\n\n        def fetch_from_cache(args):\n            msgid = self.cache[args].pop(0)\n            if not self.cache[args]:\n                del self.cache[args]\n            return msgid\n\n        calls = list(calls)\n        waiting_for = []\n        w_fds = [self.stdin_fd]\n        while wait or calls:\n            while waiting_for:\n                try:\n                    error, res = self.responses.pop(waiting_for[0])\n                    waiting_for.pop(0)\n                    if error:\n                        if error == b'DoesNotExist':\n                            raise Repository.DoesNotExist(self.location.orig)\n                        elif error == b'AlreadyExists':\n                            raise Repository.AlreadyExists(self.location.orig)\n                        elif error == b'CheckNeeded':\n                            raise Repository.CheckNeeded(self.location.orig)\n                        elif error == b'IntegrityError':\n                            raise IntegrityError(res)\n                        elif error == b'PathNotAllowed':\n                            raise PathNotAllowed(*res)\n                        elif error == b'ObjectNotFound':\n                            raise Repository.ObjectNotFound(res[0], self.location.orig)\n                        elif error == b'InvalidRPCMethod':\n                            raise InvalidRPCMethod(*res)\n                        else:\n                            raise self.RPCError(res.decode('utf-8'))\n                    else:\n                        yield res\n                        if not waiting_for and not calls:\n                            return\n                except KeyError:\n                    break\n            r, w, x = select.select(self.r_fds, w_fds, self.x_fds, 1)\n            if x:\n                raise Exception('FD exception occurred')\n            if r:\n                data = os.read(self.stdout_fd, BUFSIZE)\n                if not data:\n                    raise ConnectionClosed()\n                self.unpacker.feed(data)\n                for unpacked in self.unpacker:\n                    if not (isinstance(unpacked, tuple) and len(unpacked) == 4):\n                        raise Exception(\"Unexpected RPC data format.\")\n                    type, msgid, error, res = unpacked\n                    if msgid in self.ignore_responses:\n                        self.ignore_responses.remove(msgid)\n                    else:\n                        self.responses[msgid] = error, res\n            if w:\n                while not self.to_send and (calls or self.preload_ids) and len(waiting_for) < 100:\n                    if calls:\n                        if is_preloaded:\n                            if calls[0] in self.cache:\n                                waiting_for.append(fetch_from_cache(calls.pop(0)))\n                        else:\n                            args = calls.pop(0)\n                            if cmd == 'get' and args in self.cache:\n                                waiting_for.append(fetch_from_cache(args))\n                            else:\n                                self.msgid += 1\n                                waiting_for.append(self.msgid)\n                                self.to_send = msgpack.packb((1, self.msgid, cmd, args))\n                    if not self.to_send and self.preload_ids:\n                        args = (self.preload_ids.pop(0),)\n                        self.msgid += 1\n                        self.cache.setdefault(args, []).append(self.msgid)\n                        self.to_send = msgpack.packb((1, self.msgid, cmd, args))\n\n                if self.to_send:\n                    try:\n                        self.to_send = self.to_send[os.write(self.stdin_fd, self.to_send):]\n                    except OSError as e:\n                        # io.write might raise EAGAIN even though select indicates\n                        # that the fd should be writable\n                        if e.errno != errno.EAGAIN:\n                            raise\n                if not self.to_send and not (calls or self.preload_ids):\n                    w_fds = []\n        self.ignore_responses |= set(waiting_for)\n\n    def check(self, repair=False):\n        return self.call('check', repair)\n\n    def commit(self, *args):\n        return self.call('commit')\n\n    def rollback(self, *args):\n        return self.call('rollback')\n\n    def destroy(self):\n        return self.call('destroy')\n\n    def __len__(self):\n        return self.call('__len__')\n\n    def list(self, limit=None, marker=None):\n        return self.call('list', limit, marker)\n\n    def get(self, id_):\n        for resp in self.get_many([id_]):\n            return resp\n\n    def get_many(self, ids, is_preloaded=False):\n        for resp in self.call_many('get', [(id_,) for id_ in ids], is_preloaded=is_preloaded):\n            yield resp\n\n    def put(self, id_, data, wait=True):\n        return self.call('put', id_, data, wait=wait)\n\n    def delete(self, id_, wait=True):\n        return self.call('delete', id_, wait=wait)\n\n    def save_key(self, keydata):\n        return self.call('save_key', keydata)\n\n    def load_key(self):\n        return self.call('load_key')\n\n    def break_lock(self):\n        return self.call('break_lock')\n\n    def close(self):\n        if self.p:\n            self.p.stdin.close()\n            self.p.stdout.close()\n            self.p.wait()\n            self.p = None\n\n    def preload(self, ids):\n        self.preload_ids += ids\n\n\nclass RepositoryCache:\n    \"\"\"A caching Repository wrapper\n\n    Caches Repository GET operations using a local temporary Repository.\n    \"\"\"\n    def __init__(self, repository):\n        self.repository = repository\n        tmppath = tempfile.mkdtemp(prefix='borg-tmp')\n        self.caching_repo = Repository(tmppath, create=True, exclusive=True)\n\n    def __del__(self):\n        self.caching_repo.destroy()\n\n    def get(self, key):\n        return next(self.get_many([key]))\n\n    def get_many(self, keys):\n        unknown_keys = [key for key in keys if key not in self.caching_repo]\n        repository_iterator = zip(unknown_keys, self.repository.get_many(unknown_keys))\n        for key in keys:\n            try:\n                yield self.caching_repo.get(key)\n            except Repository.ObjectNotFound:\n                for key_, data in repository_iterator:\n                    if key_ == key:\n                        self.caching_repo.put(key, data)\n                        yield data\n                        break\n        # Consume any pending requests\n        for _ in repository_iterator:\n            pass\n\n\ndef cache_if_remote(repository):\n    if isinstance(repository, RemoteRepository):\n        return RepositoryCache(repository)\n    return repository\n", "path": "borg/remote.py"}]}