We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
PermissionException on AWS
Facing issues with ds.store() on AWS while the same code works properly locally.
Error : `hub.exceptions.PermissionException: No permision to store the dataset at s3://snark-hub/public/abhinav/ds`

For now, got it working using `sudo rm -rf /tmp/dask-worker-space/`.
A proper fix is needed.




--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `hub/collections/client_manager.py`
Content:
```
1 import psutil
2 
3 import dask
4 import hub
5 from dask.cache import Cache
6 
7 from dask.distributed import Client
8 from hub import config
9 from multiprocessing import current_process
10 
11 from dask.callbacks import Callback
12 from timeit import default_timer
13 from numbers import Number
14 import sys
15 
16 import psutil, os, time
17 
18 _client = None
19 
20 
21 def get_client():
22     global _client
23     if _client is None:
24         _client = init()
25     return _client
26 
27 
28 def init(
29     token: str = "",
30     cloud=False,
31     n_workers=1,
32     memory_limit=None,
33     processes=False,
34     threads_per_worker=1,
35     distributed=True,
36 ):
37     """Initializes cluster either local or on the cloud
38         
39         Parameters
40         ----------
41         token: str
42             token provided by snark
43         cache: float
44             Amount on local memory to cache locally, default 2e9 (2GB)
45         cloud: bool
46             Should be run locally or on the cloud
47         n_workers: int
48             number of concurrent workers, default to1
49         threads_per_worker: int
50             Number of threads per each worker
51     """
52     print("initialized")
53     global _client
54     if _client is not None:
55         _client.close()
56 
57     if cloud:
58         raise NotImplementedError
59     elif not distributed:
60         client = None
61         dask.config.set(scheduler="threading")
62         hub.config.DISTRIBUTED = False
63     else:
64         n_workers = n_workers if n_workers is not None else psutil.cpu_count()
65         memory_limit = (
66             memory_limit
67             if memory_limit is not None
68             else psutil.virtual_memory().available
69         )
70         client = Client(
71             n_workers=n_workers,
72             processes=processes,
73             memory_limit=memory_limit,
74             threads_per_worker=threads_per_worker,
75             local_directory="/tmp/",
76         )
77         config.DISTRIBUTED = True
78 
79     _client = client
80     return client
81 
82 
83 overhead = sys.getsizeof(1.23) * 4 + sys.getsizeof(()) * 4
84 
85 
86 class HubCache(Cache):
87     def _posttask(self, key, value, dsk, state, id):
88         duration = default_timer() - self.starttimes[key]
89         deps = state["dependencies"][key]
90         if deps:
91             duration += max(self.durations.get(k, 0) for k in deps)
92         self.durations[key] = duration
93         nb = self._nbytes(value) + overhead + sys.getsizeof(key) * 4
94 
95         # _cost calculation has been fixed to avoid memory leak
96         _cost = duration
97         self.cache.put(key, value, cost=_cost, nbytes=nb)
98 
99 
100 # cache = HubCache(2e9)
101 # cache.register()
102
```
--- 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/hub/collections/client_manager.py b/hub/collections/client_manager.py
--- a/hub/collections/client_manager.py
+++ b/hub/collections/client_manager.py
@@ -35,7 +35,7 @@
     distributed=True,
 ):
     """Initializes cluster either local or on the cloud
-        
+
         Parameters
         ----------
         token: str
@@ -67,12 +67,20 @@
             if memory_limit is not None
             else psutil.virtual_memory().available
         )
+
+        local_directory = os.path.join(
+            os.path.expanduser('~'),
+            '.activeloop',
+            'tmp',
+        )
+        if not os.path.exists(local_directory):
+            os.makedirs(local_directory)
         client = Client(
             n_workers=n_workers,
             processes=processes,
             memory_limit=memory_limit,
             threads_per_worker=threads_per_worker,
-            local_directory="/tmp/",
+            local_directory=local_directory,
         )
         config.DISTRIBUTED = True

{"golden_diff": "diff --git a/hub/collections/client_manager.py b/hub/collections/client_manager.py\n--- a/hub/collections/client_manager.py\n+++ b/hub/collections/client_manager.py\n@@ -35,7 +35,7 @@\n     distributed=True,\n ):\n     \"\"\"Initializes cluster either local or on the cloud\n-        \n+\n         Parameters\n         ----------\n         token: str\n@@ -67,12 +67,20 @@\n             if memory_limit is not None\n             else psutil.virtual_memory().available\n         )\n+\n+        local_directory = os.path.join(\n+            os.path.expanduser('~'),\n+            '.activeloop',\n+            'tmp',\n+        )\n+        if not os.path.exists(local_directory):\n+            os.makedirs(local_directory)\n         client = Client(\n             n_workers=n_workers,\n             processes=processes,\n             memory_limit=memory_limit,\n             threads_per_worker=threads_per_worker,\n-            local_directory=\"/tmp/\",\n+            local_directory=local_directory,\n         )\n         config.DISTRIBUTED = True\n", "issue": "PermissionException on AWS\nFacing issues with ds.store() on AWS while the same code works properly locally.\r\nError : `hub.exceptions.PermissionException: No permision to store the dataset at s3://snark-hub/public/abhinav/ds`\r\n\r\nFor now, got it working using `sudo rm -rf /tmp/dask-worker-space/`.\r\nA proper fix is needed.\r\n\r\n\r\n\n", "before_files": [{"content": "import psutil\n\nimport dask\nimport hub\nfrom dask.cache import Cache\n\nfrom dask.distributed import Client\nfrom hub import config\nfrom multiprocessing import current_process\n\nfrom dask.callbacks import Callback\nfrom timeit import default_timer\nfrom numbers import Number\nimport sys\n\nimport psutil, os, time\n\n_client = None\n\n\ndef get_client():\n    global _client\n    if _client is None:\n        _client = init()\n    return _client\n\n\ndef init(\n    token: str = \"\",\n    cloud=False,\n    n_workers=1,\n    memory_limit=None,\n    processes=False,\n    threads_per_worker=1,\n    distributed=True,\n):\n    \"\"\"Initializes cluster either local or on the cloud\n        \n        Parameters\n        ----------\n        token: str\n            token provided by snark\n        cache: float\n            Amount on local memory to cache locally, default 2e9 (2GB)\n        cloud: bool\n            Should be run locally or on the cloud\n        n_workers: int\n            number of concurrent workers, default to1\n        threads_per_worker: int\n            Number of threads per each worker\n    \"\"\"\n    print(\"initialized\")\n    global _client\n    if _client is not None:\n        _client.close()\n\n    if cloud:\n        raise NotImplementedError\n    elif not distributed:\n        client = None\n        dask.config.set(scheduler=\"threading\")\n        hub.config.DISTRIBUTED = False\n    else:\n        n_workers = n_workers if n_workers is not None else psutil.cpu_count()\n        memory_limit = (\n            memory_limit\n            if memory_limit is not None\n            else psutil.virtual_memory().available\n        )\n        client = Client(\n            n_workers=n_workers,\n            processes=processes,\n            memory_limit=memory_limit,\n            threads_per_worker=threads_per_worker,\n            local_directory=\"/tmp/\",\n        )\n        config.DISTRIBUTED = True\n\n    _client = client\n    return client\n\n\noverhead = sys.getsizeof(1.23) * 4 + sys.getsizeof(()) * 4\n\n\nclass HubCache(Cache):\n    def _posttask(self, key, value, dsk, state, id):\n        duration = default_timer() - self.starttimes[key]\n        deps = state[\"dependencies\"][key]\n        if deps:\n            duration += max(self.durations.get(k, 0) for k in deps)\n        self.durations[key] = duration\n        nb = self._nbytes(value) + overhead + sys.getsizeof(key) * 4\n\n        # _cost calculation has been fixed to avoid memory leak\n        _cost = duration\n        self.cache.put(key, value, cost=_cost, nbytes=nb)\n\n\n# cache = HubCache(2e9)\n# cache.register()\n", "path": "hub/collections/client_manager.py"}], "after_files": [{"content": "import psutil\n\nimport dask\nimport hub\nfrom dask.cache import Cache\n\nfrom dask.distributed import Client\nfrom hub import config\nfrom multiprocessing import current_process\n\nfrom dask.callbacks import Callback\nfrom timeit import default_timer\nfrom numbers import Number\nimport sys\n\nimport psutil, os, time\n\n_client = None\n\n\ndef get_client():\n    global _client\n    if _client is None:\n        _client = init()\n    return _client\n\n\ndef init(\n    token: str = \"\",\n    cloud=False,\n    n_workers=1,\n    memory_limit=None,\n    processes=False,\n    threads_per_worker=1,\n    distributed=True,\n):\n    \"\"\"Initializes cluster either local or on the cloud\n\n        Parameters\n        ----------\n        token: str\n            token provided by snark\n        cache: float\n            Amount on local memory to cache locally, default 2e9 (2GB)\n        cloud: bool\n            Should be run locally or on the cloud\n        n_workers: int\n            number of concurrent workers, default to1\n        threads_per_worker: int\n            Number of threads per each worker\n    \"\"\"\n    print(\"initialized\")\n    global _client\n    if _client is not None:\n        _client.close()\n\n    if cloud:\n        raise NotImplementedError\n    elif not distributed:\n        client = None\n        dask.config.set(scheduler=\"threading\")\n        hub.config.DISTRIBUTED = False\n    else:\n        n_workers = n_workers if n_workers is not None else psutil.cpu_count()\n        memory_limit = (\n            memory_limit\n            if memory_limit is not None\n            else psutil.virtual_memory().available\n        )\n\n        local_directory = os.path.join(\n            os.path.expanduser('~'),\n            '.activeloop',\n            'tmp',\n        )\n        if not os.path.exists(local_directory):\n            os.makedirs(local_directory)\n        client = Client(\n            n_workers=n_workers,\n            processes=processes,\n            memory_limit=memory_limit,\n            threads_per_worker=threads_per_worker,\n            local_directory=local_directory,\n        )\n        config.DISTRIBUTED = True\n\n    _client = client\n    return client\n\n\noverhead = sys.getsizeof(1.23) * 4 + sys.getsizeof(()) * 4\n\n\nclass HubCache(Cache):\n    def _posttask(self, key, value, dsk, state, id):\n        duration = default_timer() - self.starttimes[key]\n        deps = state[\"dependencies\"][key]\n        if deps:\n            duration += max(self.durations.get(k, 0) for k in deps)\n        self.durations[key] = duration\n        nb = self._nbytes(value) + overhead + sys.getsizeof(key) * 4\n\n        # _cost calculation has been fixed to avoid memory leak\n        _cost = duration\n        self.cache.put(key, value, cost=_cost, nbytes=nb)\n\n\n# cache = HubCache(2e9)\n# cache.register()\n", "path": "hub/collections/client_manager.py"}]}