prompt: You will be provided with a partial code base and an issue statement explaining a problem to resolve.
  <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.
  
  
  
  
  </issue>
  <code>
  [start of hub/collections/client_manager.py]
  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 of hub/collections/client_manager.py]
  </code>
  
  I need you to solve this issue by generating a single patch file that I can apply directly to this repository using git apply. Please respond with a single patch file in the following format.
  <patch>
  diff --git a/file.py b/file.py
  --- a/file.py
  +++ b/file.py
  @@ -1,27 +1,35 @@
   def euclidean(a, b):
  -    while b:
  -        a, b = b, a % b
  -    return a
  +    if b == 0:
  +        return a
  +    return euclidean(b, a % b)
   
   
   def bresenham(x0, y0, x1, y1):
       points = []
       dx = abs(x1 - x0)
       dy = abs(y1 - y0)
  -    sx = 1 if x0 < x1 else -1
  -    sy = 1 if y0 < y1 else -1
  -    err = dx - dy
  +    x, y = x0, y0
  +    sx = -1 if x0 > x1 else 1
  +    sy = -1 if y0 > y1 else 1
   
  -    while True:
  -        points.append((x0, y0))
  -        if x0 == x1 and y0 == y1:
  -            break
  -        e2 = 2 * err
  -        if e2 > -dy:
  -            err -= dy
  -            x0 += sx
  -        if e2 < dx:
  -            err += dx
  -            y0 += sy
  +    if dx > dy:
  +        err = dx / 2.0
  +        while x != x1:
  +            points.append((x, y))
  +            err -= dy
  +            if err < 0:
  +                y += sy
  +                err += dx
  +            x += sx
  +    else:
  +        err = dy / 2.0
  +        while y != y1:
  +            points.append((x, y))
  +            err -= dx
  +            if err < 0:
  +                x += sx
  +                err += dy
  +            y += sy
  +
  +    points.append((x, y))
       return points
  
  </patch>
golden_diff: 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
  
verification_info: {"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"}]}