We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[2.0] Lock class uses blocking sleep
aioredis/lock.py calls `time.sleep` instead of `asyncio.sleep`. Pull request on its way...

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `aioredis/lock.py`
Content:
```
1 import threading
2 import time as mod_time
3 import uuid
4 from types import SimpleNamespace
5 from typing import TYPE_CHECKING, Awaitable, NoReturn, Union
6 
7 from aioredis.exceptions import LockError, LockNotOwnedError
8 
9 if TYPE_CHECKING:
10     from aioredis import Redis
11 
12 
13 class Lock:
14     """
15     A shared, distributed Lock. Using Redis for locking allows the Lock
16     to be shared across processes and/or machines.
17 
18     It's left to the user to resolve deadlock issues and make sure
19     multiple clients play nicely together.
20     """
21 
22     lua_release = None
23     lua_extend = None
24     lua_reacquire = None
25 
26     # KEYS[1] - lock name
27     # ARGV[1] - token
28     # return 1 if the lock was released, otherwise 0
29     LUA_RELEASE_SCRIPT = """
30         local token = redis.call('get', KEYS[1])
31         if not token or token ~= ARGV[1] then
32             return 0
33         end
34         redis.call('del', KEYS[1])
35         return 1
36     """
37 
38     # KEYS[1] - lock name
39     # ARGV[1] - token
40     # ARGV[2] - additional milliseconds
41     # ARGV[3] - "0" if the additional time should be added to the lock's
42     #           existing ttl or "1" if the existing ttl should be replaced
43     # return 1 if the locks time was extended, otherwise 0
44     LUA_EXTEND_SCRIPT = """
45         local token = redis.call('get', KEYS[1])
46         if not token or token ~= ARGV[1] then
47             return 0
48         end
49         local expiration = redis.call('pttl', KEYS[1])
50         if not expiration then
51             expiration = 0
52         end
53         if expiration < 0 then
54             return 0
55         end
56 
57         local newttl = ARGV[2]
58         if ARGV[3] == "0" then
59             newttl = ARGV[2] + expiration
60         end
61         redis.call('pexpire', KEYS[1], newttl)
62         return 1
63     """
64 
65     # KEYS[1] - lock name
66     # ARGV[1] - token
67     # ARGV[2] - milliseconds
68     # return 1 if the locks time was reacquired, otherwise 0
69     LUA_REACQUIRE_SCRIPT = """
70         local token = redis.call('get', KEYS[1])
71         if not token or token ~= ARGV[1] then
72             return 0
73         end
74         redis.call('pexpire', KEYS[1], ARGV[2])
75         return 1
76     """
77 
78     def __init__(
79         self,
80         redis: "Redis",
81         name: str,
82         timeout: float = None,
83         sleep: float = 0.1,
84         blocking: bool = True,
85         blocking_timeout: float = None,
86         thread_local: bool = True,
87     ):
88         """
89         Create a new Lock instance named ``name`` using the Redis client
90         supplied by ``redis``.
91 
92         ``timeout`` indicates a maximum life for the lock.
93         By default, it will remain locked until release() is called.
94         ``timeout`` can be specified as a float or integer, both representing
95         the number of seconds to wait.
96 
97         ``sleep`` indicates the amount of time to sleep per loop iteration
98         when the lock is in blocking mode and another client is currently
99         holding the lock.
100 
101         ``blocking`` indicates whether calling ``acquire`` should block until
102         the lock has been acquired or to fail immediately, causing ``acquire``
103         to return False and the lock not being acquired. Defaults to True.
104         Note this value can be overridden by passing a ``blocking``
105         argument to ``acquire``.
106 
107         ``blocking_timeout`` indicates the maximum amount of time in seconds to
108         spend trying to acquire the lock. A value of ``None`` indicates
109         continue trying forever. ``blocking_timeout`` can be specified as a
110         float or integer, both representing the number of seconds to wait.
111 
112         ``thread_local`` indicates whether the lock token is placed in
113         thread-local storage. By default, the token is placed in thread local
114         storage so that a thread only sees its token, not a token set by
115         another thread. Consider the following timeline:
116 
117             time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
118                      thread-1 sets the token to "abc"
119             time: 1, thread-2 blocks trying to acquire `my-lock` using the
120                      Lock instance.
121             time: 5, thread-1 has not yet completed. redis expires the lock
122                      key.
123             time: 5, thread-2 acquired `my-lock` now that it's available.
124                      thread-2 sets the token to "xyz"
125             time: 6, thread-1 finishes its work and calls release(). if the
126                      token is *not* stored in thread local storage, then
127                      thread-1 would see the token value as "xyz" and would be
128                      able to successfully release the thread-2's lock.
129 
130         In some use cases it's necessary to disable thread local storage. For
131         example, if you have code where one thread acquires a lock and passes
132         that lock instance to a worker thread to release later. If thread
133         local storage isn't disabled in this case, the worker thread won't see
134         the token set by the thread that acquired the lock. Our assumption
135         is that these cases aren't common and as such default to using
136         thread local storage.
137         """
138         self.redis = redis
139         self.name = name
140         self.timeout = timeout
141         self.sleep = sleep
142         self.blocking = blocking
143         self.blocking_timeout = blocking_timeout
144         self.thread_local = bool(thread_local)
145         self.local = threading.local() if self.thread_local else SimpleNamespace()
146         self.local.token = None
147         self.register_scripts()
148 
149     def register_scripts(self):
150         cls = self.__class__
151         client = self.redis
152         if cls.lua_release is None:
153             cls.lua_release = client.register_script(cls.LUA_RELEASE_SCRIPT)
154         if cls.lua_extend is None:
155             cls.lua_extend = client.register_script(cls.LUA_EXTEND_SCRIPT)
156         if cls.lua_reacquire is None:
157             cls.lua_reacquire = client.register_script(cls.LUA_REACQUIRE_SCRIPT)
158 
159     async def __aenter__(self):
160         # force blocking, as otherwise the user would have to check whether
161         # the lock was actually acquired or not.
162         if await self.acquire(blocking=True):
163             return self
164         raise LockError("Unable to acquire lock within the time specified")
165 
166     async def __aexit__(self, exc_type, exc_value, traceback):
167         await self.release()
168 
169     async def acquire(
170         self,
171         blocking: bool = None,
172         blocking_timeout: float = None,
173         token: Union[str, bytes] = None,
174     ):
175         """
176         Use Redis to hold a shared, distributed lock named ``name``.
177         Returns True once the lock is acquired.
178 
179         If ``blocking`` is False, always return immediately. If the lock
180         was acquired, return True, otherwise return False.
181 
182         ``blocking_timeout`` specifies the maximum number of seconds to
183         wait trying to acquire the lock.
184 
185         ``token`` specifies the token value to be used. If provided, token
186         must be a bytes object or a string that can be encoded to a bytes
187         object with the default encoding. If a token isn't specified, a UUID
188         will be generated.
189         """
190         sleep = self.sleep
191         if token is None:
192             token = uuid.uuid1().hex.encode()
193         else:
194             encoder = self.redis.connection_pool.get_encoder()
195             token = encoder.encode(token)
196         if blocking is None:
197             blocking = self.blocking
198         if blocking_timeout is None:
199             blocking_timeout = self.blocking_timeout
200         stop_trying_at = None
201         if blocking_timeout is not None:
202             stop_trying_at = mod_time.monotonic() + blocking_timeout
203         while True:
204             if await self.do_acquire(token):
205                 self.local.token = token
206                 return True
207             if not blocking:
208                 return False
209             next_try_at = mod_time.monotonic() + sleep
210             if stop_trying_at is not None and next_try_at > stop_trying_at:
211                 return False
212             mod_time.sleep(sleep)
213 
214     async def do_acquire(self, token: Union[str, bytes]) -> bool:
215         if self.timeout:
216             # convert to milliseconds
217             timeout = int(self.timeout * 1000)
218         else:
219             timeout = None
220         if await self.redis.set(self.name, token, nx=True, px=timeout):
221             return True
222         return False
223 
224     async def locked(self) -> bool:
225         """
226         Returns True if this key is locked by any process, otherwise False.
227         """
228         return await self.redis.get(self.name) is not None
229 
230     async def owned(self) -> bool:
231         """
232         Returns True if this key is locked by this lock, otherwise False.
233         """
234         stored_token = await self.redis.get(self.name)
235         # need to always compare bytes to bytes
236         # TODO: this can be simplified when the context manager is finished
237         if stored_token and not isinstance(stored_token, bytes):
238             encoder = self.redis.connection_pool.get_encoder()
239             stored_token = encoder.encode(stored_token)
240         return self.local.token is not None and stored_token == self.local.token
241 
242     def release(self) -> Awaitable[NoReturn]:
243         """Releases the already acquired lock"""
244         expected_token = self.local.token
245         if expected_token is None:
246             raise LockError("Cannot release an unlocked lock")
247         self.local.token = None
248         return self.do_release(expected_token)
249 
250     async def do_release(self, expected_token: bytes):
251         if not bool(
252             await self.lua_release(
253                 keys=[self.name], args=[expected_token], client=self.redis
254             )
255         ):
256             raise LockNotOwnedError("Cannot release a lock" " that's no longer owned")
257 
258     def extend(
259         self, additional_time: float, replace_ttl: bool = False
260     ) -> Awaitable[bool]:
261         """
262         Adds more time to an already acquired lock.
263 
264         ``additional_time`` can be specified as an integer or a float, both
265         representing the number of seconds to add.
266 
267         ``replace_ttl`` if False (the default), add `additional_time` to
268         the lock's existing ttl. If True, replace the lock's ttl with
269         `additional_time`.
270         """
271         if self.local.token is None:
272             raise LockError("Cannot extend an unlocked lock")
273         if self.timeout is None:
274             raise LockError("Cannot extend a lock with no timeout")
275         return self.do_extend(additional_time, replace_ttl)
276 
277     async def do_extend(self, additional_time, replace_ttl) -> bool:
278         additional_time = int(additional_time * 1000)
279         if not bool(
280             await self.lua_extend(
281                 keys=[self.name],
282                 args=[self.local.token, additional_time, replace_ttl and "1" or "0"],
283                 client=self.redis,
284             )
285         ):
286             raise LockNotOwnedError("Cannot extend a lock that's" " no longer owned")
287         return True
288 
289     def reacquire(self) -> Awaitable[bool]:
290         """
291         Resets a TTL of an already acquired lock back to a timeout value.
292         """
293         if self.local.token is None:
294             raise LockError("Cannot reacquire an unlocked lock")
295         if self.timeout is None:
296             raise LockError("Cannot reacquire a lock with no timeout")
297         return self.do_reacquire()
298 
299     async def do_reacquire(self) -> bool:
300         timeout = int(self.timeout * 1000)
301         if not bool(
302             await self.lua_reacquire(
303                 keys=[self.name], args=[self.local.token, timeout], client=self.redis
304             )
305         ):
306             raise LockNotOwnedError("Cannot reacquire a lock that's" " no longer owned")
307         return True
308
```
--- 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/aioredis/lock.py b/aioredis/lock.py
--- a/aioredis/lock.py
+++ b/aioredis/lock.py
@@ -1,3 +1,4 @@
+import asyncio
 import threading
 import time as mod_time
 import uuid
@@ -209,7 +210,7 @@
             next_try_at = mod_time.monotonic() + sleep
             if stop_trying_at is not None and next_try_at > stop_trying_at:
                 return False
-            mod_time.sleep(sleep)
+            await asyncio.sleep(sleep)
 
     async def do_acquire(self, token: Union[str, bytes]) -> bool:
         if self.timeout:

{"golden_diff": "diff --git a/aioredis/lock.py b/aioredis/lock.py\n--- a/aioredis/lock.py\n+++ b/aioredis/lock.py\n@@ -1,3 +1,4 @@\n+import asyncio\n import threading\n import time as mod_time\n import uuid\n@@ -209,7 +210,7 @@\n             next_try_at = mod_time.monotonic() + sleep\n             if stop_trying_at is not None and next_try_at > stop_trying_at:\n                 return False\n-            mod_time.sleep(sleep)\n+            await asyncio.sleep(sleep)\n \n     async def do_acquire(self, token: Union[str, bytes]) -> bool:\n         if self.timeout:\n", "issue": "[2.0] Lock class uses blocking sleep\naioredis/lock.py calls `time.sleep` instead of `asyncio.sleep`. Pull request on its way...\n", "before_files": [{"content": "import threading\nimport time as mod_time\nimport uuid\nfrom types import SimpleNamespace\nfrom typing import TYPE_CHECKING, Awaitable, NoReturn, Union\n\nfrom aioredis.exceptions import LockError, LockNotOwnedError\n\nif TYPE_CHECKING:\n    from aioredis import Redis\n\n\nclass Lock:\n    \"\"\"\n    A shared, distributed Lock. Using Redis for locking allows the Lock\n    to be shared across processes and/or machines.\n\n    It's left to the user to resolve deadlock issues and make sure\n    multiple clients play nicely together.\n    \"\"\"\n\n    lua_release = None\n    lua_extend = None\n    lua_reacquire = None\n\n    # KEYS[1] - lock name\n    # ARGV[1] - token\n    # return 1 if the lock was released, otherwise 0\n    LUA_RELEASE_SCRIPT = \"\"\"\n        local token = redis.call('get', KEYS[1])\n        if not token or token ~= ARGV[1] then\n            return 0\n        end\n        redis.call('del', KEYS[1])\n        return 1\n    \"\"\"\n\n    # KEYS[1] - lock name\n    # ARGV[1] - token\n    # ARGV[2] - additional milliseconds\n    # ARGV[3] - \"0\" if the additional time should be added to the lock's\n    #           existing ttl or \"1\" if the existing ttl should be replaced\n    # return 1 if the locks time was extended, otherwise 0\n    LUA_EXTEND_SCRIPT = \"\"\"\n        local token = redis.call('get', KEYS[1])\n        if not token or token ~= ARGV[1] then\n            return 0\n        end\n        local expiration = redis.call('pttl', KEYS[1])\n        if not expiration then\n            expiration = 0\n        end\n        if expiration < 0 then\n            return 0\n        end\n\n        local newttl = ARGV[2]\n        if ARGV[3] == \"0\" then\n            newttl = ARGV[2] + expiration\n        end\n        redis.call('pexpire', KEYS[1], newttl)\n        return 1\n    \"\"\"\n\n    # KEYS[1] - lock name\n    # ARGV[1] - token\n    # ARGV[2] - milliseconds\n    # return 1 if the locks time was reacquired, otherwise 0\n    LUA_REACQUIRE_SCRIPT = \"\"\"\n        local token = redis.call('get', KEYS[1])\n        if not token or token ~= ARGV[1] then\n            return 0\n        end\n        redis.call('pexpire', KEYS[1], ARGV[2])\n        return 1\n    \"\"\"\n\n    def __init__(\n        self,\n        redis: \"Redis\",\n        name: str,\n        timeout: float = None,\n        sleep: float = 0.1,\n        blocking: bool = True,\n        blocking_timeout: float = None,\n        thread_local: bool = True,\n    ):\n        \"\"\"\n        Create a new Lock instance named ``name`` using the Redis client\n        supplied by ``redis``.\n\n        ``timeout`` indicates a maximum life for the lock.\n        By default, it will remain locked until release() is called.\n        ``timeout`` can be specified as a float or integer, both representing\n        the number of seconds to wait.\n\n        ``sleep`` indicates the amount of time to sleep per loop iteration\n        when the lock is in blocking mode and another client is currently\n        holding the lock.\n\n        ``blocking`` indicates whether calling ``acquire`` should block until\n        the lock has been acquired or to fail immediately, causing ``acquire``\n        to return False and the lock not being acquired. Defaults to True.\n        Note this value can be overridden by passing a ``blocking``\n        argument to ``acquire``.\n\n        ``blocking_timeout`` indicates the maximum amount of time in seconds to\n        spend trying to acquire the lock. A value of ``None`` indicates\n        continue trying forever. ``blocking_timeout`` can be specified as a\n        float or integer, both representing the number of seconds to wait.\n\n        ``thread_local`` indicates whether the lock token is placed in\n        thread-local storage. By default, the token is placed in thread local\n        storage so that a thread only sees its token, not a token set by\n        another thread. Consider the following timeline:\n\n            time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.\n                     thread-1 sets the token to \"abc\"\n            time: 1, thread-2 blocks trying to acquire `my-lock` using the\n                     Lock instance.\n            time: 5, thread-1 has not yet completed. redis expires the lock\n                     key.\n            time: 5, thread-2 acquired `my-lock` now that it's available.\n                     thread-2 sets the token to \"xyz\"\n            time: 6, thread-1 finishes its work and calls release(). if the\n                     token is *not* stored in thread local storage, then\n                     thread-1 would see the token value as \"xyz\" and would be\n                     able to successfully release the thread-2's lock.\n\n        In some use cases it's necessary to disable thread local storage. For\n        example, if you have code where one thread acquires a lock and passes\n        that lock instance to a worker thread to release later. If thread\n        local storage isn't disabled in this case, the worker thread won't see\n        the token set by the thread that acquired the lock. Our assumption\n        is that these cases aren't common and as such default to using\n        thread local storage.\n        \"\"\"\n        self.redis = redis\n        self.name = name\n        self.timeout = timeout\n        self.sleep = sleep\n        self.blocking = blocking\n        self.blocking_timeout = blocking_timeout\n        self.thread_local = bool(thread_local)\n        self.local = threading.local() if self.thread_local else SimpleNamespace()\n        self.local.token = None\n        self.register_scripts()\n\n    def register_scripts(self):\n        cls = self.__class__\n        client = self.redis\n        if cls.lua_release is None:\n            cls.lua_release = client.register_script(cls.LUA_RELEASE_SCRIPT)\n        if cls.lua_extend is None:\n            cls.lua_extend = client.register_script(cls.LUA_EXTEND_SCRIPT)\n        if cls.lua_reacquire is None:\n            cls.lua_reacquire = client.register_script(cls.LUA_REACQUIRE_SCRIPT)\n\n    async def __aenter__(self):\n        # force blocking, as otherwise the user would have to check whether\n        # the lock was actually acquired or not.\n        if await self.acquire(blocking=True):\n            return self\n        raise LockError(\"Unable to acquire lock within the time specified\")\n\n    async def __aexit__(self, exc_type, exc_value, traceback):\n        await self.release()\n\n    async def acquire(\n        self,\n        blocking: bool = None,\n        blocking_timeout: float = None,\n        token: Union[str, bytes] = None,\n    ):\n        \"\"\"\n        Use Redis to hold a shared, distributed lock named ``name``.\n        Returns True once the lock is acquired.\n\n        If ``blocking`` is False, always return immediately. If the lock\n        was acquired, return True, otherwise return False.\n\n        ``blocking_timeout`` specifies the maximum number of seconds to\n        wait trying to acquire the lock.\n\n        ``token`` specifies the token value to be used. If provided, token\n        must be a bytes object or a string that can be encoded to a bytes\n        object with the default encoding. If a token isn't specified, a UUID\n        will be generated.\n        \"\"\"\n        sleep = self.sleep\n        if token is None:\n            token = uuid.uuid1().hex.encode()\n        else:\n            encoder = self.redis.connection_pool.get_encoder()\n            token = encoder.encode(token)\n        if blocking is None:\n            blocking = self.blocking\n        if blocking_timeout is None:\n            blocking_timeout = self.blocking_timeout\n        stop_trying_at = None\n        if blocking_timeout is not None:\n            stop_trying_at = mod_time.monotonic() + blocking_timeout\n        while True:\n            if await self.do_acquire(token):\n                self.local.token = token\n                return True\n            if not blocking:\n                return False\n            next_try_at = mod_time.monotonic() + sleep\n            if stop_trying_at is not None and next_try_at > stop_trying_at:\n                return False\n            mod_time.sleep(sleep)\n\n    async def do_acquire(self, token: Union[str, bytes]) -> bool:\n        if self.timeout:\n            # convert to milliseconds\n            timeout = int(self.timeout * 1000)\n        else:\n            timeout = None\n        if await self.redis.set(self.name, token, nx=True, px=timeout):\n            return True\n        return False\n\n    async def locked(self) -> bool:\n        \"\"\"\n        Returns True if this key is locked by any process, otherwise False.\n        \"\"\"\n        return await self.redis.get(self.name) is not None\n\n    async def owned(self) -> bool:\n        \"\"\"\n        Returns True if this key is locked by this lock, otherwise False.\n        \"\"\"\n        stored_token = await self.redis.get(self.name)\n        # need to always compare bytes to bytes\n        # TODO: this can be simplified when the context manager is finished\n        if stored_token and not isinstance(stored_token, bytes):\n            encoder = self.redis.connection_pool.get_encoder()\n            stored_token = encoder.encode(stored_token)\n        return self.local.token is not None and stored_token == self.local.token\n\n    def release(self) -> Awaitable[NoReturn]:\n        \"\"\"Releases the already acquired lock\"\"\"\n        expected_token = self.local.token\n        if expected_token is None:\n            raise LockError(\"Cannot release an unlocked lock\")\n        self.local.token = None\n        return self.do_release(expected_token)\n\n    async def do_release(self, expected_token: bytes):\n        if not bool(\n            await self.lua_release(\n                keys=[self.name], args=[expected_token], client=self.redis\n            )\n        ):\n            raise LockNotOwnedError(\"Cannot release a lock\" \" that's no longer owned\")\n\n    def extend(\n        self, additional_time: float, replace_ttl: bool = False\n    ) -> Awaitable[bool]:\n        \"\"\"\n        Adds more time to an already acquired lock.\n\n        ``additional_time`` can be specified as an integer or a float, both\n        representing the number of seconds to add.\n\n        ``replace_ttl`` if False (the default), add `additional_time` to\n        the lock's existing ttl. If True, replace the lock's ttl with\n        `additional_time`.\n        \"\"\"\n        if self.local.token is None:\n            raise LockError(\"Cannot extend an unlocked lock\")\n        if self.timeout is None:\n            raise LockError(\"Cannot extend a lock with no timeout\")\n        return self.do_extend(additional_time, replace_ttl)\n\n    async def do_extend(self, additional_time, replace_ttl) -> bool:\n        additional_time = int(additional_time * 1000)\n        if not bool(\n            await self.lua_extend(\n                keys=[self.name],\n                args=[self.local.token, additional_time, replace_ttl and \"1\" or \"0\"],\n                client=self.redis,\n            )\n        ):\n            raise LockNotOwnedError(\"Cannot extend a lock that's\" \" no longer owned\")\n        return True\n\n    def reacquire(self) -> Awaitable[bool]:\n        \"\"\"\n        Resets a TTL of an already acquired lock back to a timeout value.\n        \"\"\"\n        if self.local.token is None:\n            raise LockError(\"Cannot reacquire an unlocked lock\")\n        if self.timeout is None:\n            raise LockError(\"Cannot reacquire a lock with no timeout\")\n        return self.do_reacquire()\n\n    async def do_reacquire(self) -> bool:\n        timeout = int(self.timeout * 1000)\n        if not bool(\n            await self.lua_reacquire(\n                keys=[self.name], args=[self.local.token, timeout], client=self.redis\n            )\n        ):\n            raise LockNotOwnedError(\"Cannot reacquire a lock that's\" \" no longer owned\")\n        return True\n", "path": "aioredis/lock.py"}], "after_files": [{"content": "import asyncio\nimport threading\nimport time as mod_time\nimport uuid\nfrom types import SimpleNamespace\nfrom typing import TYPE_CHECKING, Awaitable, NoReturn, Union\n\nfrom aioredis.exceptions import LockError, LockNotOwnedError\n\nif TYPE_CHECKING:\n    from aioredis import Redis\n\n\nclass Lock:\n    \"\"\"\n    A shared, distributed Lock. Using Redis for locking allows the Lock\n    to be shared across processes and/or machines.\n\n    It's left to the user to resolve deadlock issues and make sure\n    multiple clients play nicely together.\n    \"\"\"\n\n    lua_release = None\n    lua_extend = None\n    lua_reacquire = None\n\n    # KEYS[1] - lock name\n    # ARGV[1] - token\n    # return 1 if the lock was released, otherwise 0\n    LUA_RELEASE_SCRIPT = \"\"\"\n        local token = redis.call('get', KEYS[1])\n        if not token or token ~= ARGV[1] then\n            return 0\n        end\n        redis.call('del', KEYS[1])\n        return 1\n    \"\"\"\n\n    # KEYS[1] - lock name\n    # ARGV[1] - token\n    # ARGV[2] - additional milliseconds\n    # ARGV[3] - \"0\" if the additional time should be added to the lock's\n    #           existing ttl or \"1\" if the existing ttl should be replaced\n    # return 1 if the locks time was extended, otherwise 0\n    LUA_EXTEND_SCRIPT = \"\"\"\n        local token = redis.call('get', KEYS[1])\n        if not token or token ~= ARGV[1] then\n            return 0\n        end\n        local expiration = redis.call('pttl', KEYS[1])\n        if not expiration then\n            expiration = 0\n        end\n        if expiration < 0 then\n            return 0\n        end\n\n        local newttl = ARGV[2]\n        if ARGV[3] == \"0\" then\n            newttl = ARGV[2] + expiration\n        end\n        redis.call('pexpire', KEYS[1], newttl)\n        return 1\n    \"\"\"\n\n    # KEYS[1] - lock name\n    # ARGV[1] - token\n    # ARGV[2] - milliseconds\n    # return 1 if the locks time was reacquired, otherwise 0\n    LUA_REACQUIRE_SCRIPT = \"\"\"\n        local token = redis.call('get', KEYS[1])\n        if not token or token ~= ARGV[1] then\n            return 0\n        end\n        redis.call('pexpire', KEYS[1], ARGV[2])\n        return 1\n    \"\"\"\n\n    def __init__(\n        self,\n        redis: \"Redis\",\n        name: str,\n        timeout: float = None,\n        sleep: float = 0.1,\n        blocking: bool = True,\n        blocking_timeout: float = None,\n        thread_local: bool = True,\n    ):\n        \"\"\"\n        Create a new Lock instance named ``name`` using the Redis client\n        supplied by ``redis``.\n\n        ``timeout`` indicates a maximum life for the lock.\n        By default, it will remain locked until release() is called.\n        ``timeout`` can be specified as a float or integer, both representing\n        the number of seconds to wait.\n\n        ``sleep`` indicates the amount of time to sleep per loop iteration\n        when the lock is in blocking mode and another client is currently\n        holding the lock.\n\n        ``blocking`` indicates whether calling ``acquire`` should block until\n        the lock has been acquired or to fail immediately, causing ``acquire``\n        to return False and the lock not being acquired. Defaults to True.\n        Note this value can be overridden by passing a ``blocking``\n        argument to ``acquire``.\n\n        ``blocking_timeout`` indicates the maximum amount of time in seconds to\n        spend trying to acquire the lock. A value of ``None`` indicates\n        continue trying forever. ``blocking_timeout`` can be specified as a\n        float or integer, both representing the number of seconds to wait.\n\n        ``thread_local`` indicates whether the lock token is placed in\n        thread-local storage. By default, the token is placed in thread local\n        storage so that a thread only sees its token, not a token set by\n        another thread. Consider the following timeline:\n\n            time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.\n                     thread-1 sets the token to \"abc\"\n            time: 1, thread-2 blocks trying to acquire `my-lock` using the\n                     Lock instance.\n            time: 5, thread-1 has not yet completed. redis expires the lock\n                     key.\n            time: 5, thread-2 acquired `my-lock` now that it's available.\n                     thread-2 sets the token to \"xyz\"\n            time: 6, thread-1 finishes its work and calls release(). if the\n                     token is *not* stored in thread local storage, then\n                     thread-1 would see the token value as \"xyz\" and would be\n                     able to successfully release the thread-2's lock.\n\n        In some use cases it's necessary to disable thread local storage. For\n        example, if you have code where one thread acquires a lock and passes\n        that lock instance to a worker thread to release later. If thread\n        local storage isn't disabled in this case, the worker thread won't see\n        the token set by the thread that acquired the lock. Our assumption\n        is that these cases aren't common and as such default to using\n        thread local storage.\n        \"\"\"\n        self.redis = redis\n        self.name = name\n        self.timeout = timeout\n        self.sleep = sleep\n        self.blocking = blocking\n        self.blocking_timeout = blocking_timeout\n        self.thread_local = bool(thread_local)\n        self.local = threading.local() if self.thread_local else SimpleNamespace()\n        self.local.token = None\n        self.register_scripts()\n\n    def register_scripts(self):\n        cls = self.__class__\n        client = self.redis\n        if cls.lua_release is None:\n            cls.lua_release = client.register_script(cls.LUA_RELEASE_SCRIPT)\n        if cls.lua_extend is None:\n            cls.lua_extend = client.register_script(cls.LUA_EXTEND_SCRIPT)\n        if cls.lua_reacquire is None:\n            cls.lua_reacquire = client.register_script(cls.LUA_REACQUIRE_SCRIPT)\n\n    async def __aenter__(self):\n        # force blocking, as otherwise the user would have to check whether\n        # the lock was actually acquired or not.\n        if await self.acquire(blocking=True):\n            return self\n        raise LockError(\"Unable to acquire lock within the time specified\")\n\n    async def __aexit__(self, exc_type, exc_value, traceback):\n        await self.release()\n\n    async def acquire(\n        self,\n        blocking: bool = None,\n        blocking_timeout: float = None,\n        token: Union[str, bytes] = None,\n    ):\n        \"\"\"\n        Use Redis to hold a shared, distributed lock named ``name``.\n        Returns True once the lock is acquired.\n\n        If ``blocking`` is False, always return immediately. If the lock\n        was acquired, return True, otherwise return False.\n\n        ``blocking_timeout`` specifies the maximum number of seconds to\n        wait trying to acquire the lock.\n\n        ``token`` specifies the token value to be used. If provided, token\n        must be a bytes object or a string that can be encoded to a bytes\n        object with the default encoding. If a token isn't specified, a UUID\n        will be generated.\n        \"\"\"\n        sleep = self.sleep\n        if token is None:\n            token = uuid.uuid1().hex.encode()\n        else:\n            encoder = self.redis.connection_pool.get_encoder()\n            token = encoder.encode(token)\n        if blocking is None:\n            blocking = self.blocking\n        if blocking_timeout is None:\n            blocking_timeout = self.blocking_timeout\n        stop_trying_at = None\n        if blocking_timeout is not None:\n            stop_trying_at = mod_time.monotonic() + blocking_timeout\n        while True:\n            if await self.do_acquire(token):\n                self.local.token = token\n                return True\n            if not blocking:\n                return False\n            next_try_at = mod_time.monotonic() + sleep\n            if stop_trying_at is not None and next_try_at > stop_trying_at:\n                return False\n            await asyncio.sleep(sleep)\n\n    async def do_acquire(self, token: Union[str, bytes]) -> bool:\n        if self.timeout:\n            # convert to milliseconds\n            timeout = int(self.timeout * 1000)\n        else:\n            timeout = None\n        if await self.redis.set(self.name, token, nx=True, px=timeout):\n            return True\n        return False\n\n    async def locked(self) -> bool:\n        \"\"\"\n        Returns True if this key is locked by any process, otherwise False.\n        \"\"\"\n        return await self.redis.get(self.name) is not None\n\n    async def owned(self) -> bool:\n        \"\"\"\n        Returns True if this key is locked by this lock, otherwise False.\n        \"\"\"\n        stored_token = await self.redis.get(self.name)\n        # need to always compare bytes to bytes\n        # TODO: this can be simplified when the context manager is finished\n        if stored_token and not isinstance(stored_token, bytes):\n            encoder = self.redis.connection_pool.get_encoder()\n            stored_token = encoder.encode(stored_token)\n        return self.local.token is not None and stored_token == self.local.token\n\n    def release(self) -> Awaitable[NoReturn]:\n        \"\"\"Releases the already acquired lock\"\"\"\n        expected_token = self.local.token\n        if expected_token is None:\n            raise LockError(\"Cannot release an unlocked lock\")\n        self.local.token = None\n        return self.do_release(expected_token)\n\n    async def do_release(self, expected_token: bytes):\n        if not bool(\n            await self.lua_release(\n                keys=[self.name], args=[expected_token], client=self.redis\n            )\n        ):\n            raise LockNotOwnedError(\"Cannot release a lock\" \" that's no longer owned\")\n\n    def extend(\n        self, additional_time: float, replace_ttl: bool = False\n    ) -> Awaitable[bool]:\n        \"\"\"\n        Adds more time to an already acquired lock.\n\n        ``additional_time`` can be specified as an integer or a float, both\n        representing the number of seconds to add.\n\n        ``replace_ttl`` if False (the default), add `additional_time` to\n        the lock's existing ttl. If True, replace the lock's ttl with\n        `additional_time`.\n        \"\"\"\n        if self.local.token is None:\n            raise LockError(\"Cannot extend an unlocked lock\")\n        if self.timeout is None:\n            raise LockError(\"Cannot extend a lock with no timeout\")\n        return self.do_extend(additional_time, replace_ttl)\n\n    async def do_extend(self, additional_time, replace_ttl) -> bool:\n        additional_time = int(additional_time * 1000)\n        if not bool(\n            await self.lua_extend(\n                keys=[self.name],\n                args=[self.local.token, additional_time, replace_ttl and \"1\" or \"0\"],\n                client=self.redis,\n            )\n        ):\n            raise LockNotOwnedError(\"Cannot extend a lock that's\" \" no longer owned\")\n        return True\n\n    def reacquire(self) -> Awaitable[bool]:\n        \"\"\"\n        Resets a TTL of an already acquired lock back to a timeout value.\n        \"\"\"\n        if self.local.token is None:\n            raise LockError(\"Cannot reacquire an unlocked lock\")\n        if self.timeout is None:\n            raise LockError(\"Cannot reacquire a lock with no timeout\")\n        return self.do_reacquire()\n\n    async def do_reacquire(self) -> bool:\n        timeout = int(self.timeout * 1000)\n        if not bool(\n            await self.lua_reacquire(\n                keys=[self.name], args=[self.local.token, timeout], client=self.redis\n            )\n        ):\n            raise LockNotOwnedError(\"Cannot reacquire a lock that's\" \" no longer owned\")\n        return True\n", "path": "aioredis/lock.py"}]}