DOC: Use DiskCache to prevent #317

example.ipynb

@@ -2265,37 +2265,54 @@
    "source": [
     "from concurrent.futures import ProcessPoolExecutor\n",
     "\n",
-    "for f in pipeline.functions:\n",
-    "    # Enable caching for all functions\n",
-    "    # See next section to only cache based on a certain parameter sweep\n",
-    "    f.cache = True\n",
     "\n",
-    "pf_e = pipeline.func(\"e\")\n",
-    "sequence = 10 * [{\"a\": 2, \"b\": 3, \"x\": 1}]\n",
-    "with ProcessPoolExecutor(max_workers=1) as executor:\n",
-    "    results = executor.map(pf_e.call_with_dict, sequence)\n",
+    "@pipefunc(output_name=\"y\", cache=True)\n",
+    "def my_function(a: int, b: int) -> int:\n",
+    "    time.sleep(1)  # Pretend this is a slow function\n",
+    "    print(\"Function is called!\")\n",
+    "    return a + b\n",
+    "\n",
+    "\n",
+    "# multiple cache_type options are available, e.g., \"lru\", \"hybrid\", \"disk\", and \"simple\"\n",
+    "pipeline_cache = Pipeline([my_function], cache_type=\"disk\")\n",
+    "\n",
+    "f = pipeline_cache.func(\"y\")\n",
+    "sequence = 10 * [{\"a\": 2, \"b\": 3}]  # lets call the function 10 times with the same arguments\n",
+    "with ProcessPoolExecutor(max_workers=1) as executor:  # inside a process pool executor\n",
+    "    results = executor.map(f.call_with_dict, sequence)\n",
     "    print(list(results))"
    ]
   },
   {
-   "cell_type": "markdown",
+   "cell_type": "code",
+   "execution_count": null,
    "id": "119",
    "metadata": {},
+   "outputs": [],
    "source": [
-    "The cache is populated _**even when using parallel execution**_. To see the cache, you can use the `cache` attribute on the pipeline.\n",
-    "\n",
-    "The keys of the cache are always in terms of the root arguments.\n"
+    "print(f\"Cache object: {pipeline_cache.cache}\")\n",
+    "pipeline_cache.cache.cache"
    ]
   },
   {
-   "cell_type": "code",
-   "execution_count": null,
+   "cell_type": "markdown",
    "id": "120",
    "metadata": {},
-   "outputs": [],
    "source": [
-    "print(f\"Cache object: {pipeline.cache}\")\n",
-    "pipeline.cache.cache"
+    "The cache is populated _**even when using parallel execution**_. To see the cache, you can use the `cache` attribute on the pipeline.\n",
+    "\n",
+    "```{note}\n",
+    "If calling the pipeline like a function (in contrast to using `pipeline.map`) keys of the cache are always in terms of the root arguments of the pipeline. When using `pipeline.map`, the keys are in terms of the arguments of the function.\n",
+    "\n",
+    "The key is constructed from the function name and the (root) arguments passed to the function. If the arguments are not hashable, the {class}`pipefunc.cache.to_hashable` function is used to *attempt* to convert them to a hashable form.\n",
+    "```\n",
+    "\n",
+    "One can also enable caching after the pipeline is created by setting the `cache` attribute to `True` for each function.\n",
+    "\n",
+    "```python\n",
+    "for f in pipeline.functions:\n",
+    "    f.cache = True\n",
+    "```"
    ]
   },
   {

pipefunc/_pipeline.py

@@ -14,6 +14,7 @@
 
 import functools
 import inspect
+import tempfile
 import time
 import warnings
 from dataclasses import dataclass, field
@@ -1760,6 +1761,7 @@ def _create_cache(
         return HybridCache(**cache_kwargs)
     if cache_type == "disk":
         cache_kwargs.setdefault("lru_shared", not lazy)
+        cache_kwargs.setdefault("cache_dir", tempfile.gettempdir())
         return DiskCache(**cache_kwargs)
     if cache_type == "simple":
         return SimpleCache()