lockfile in download_decompress (#1332)

* lockfile in download_decompress

* Revert "lockfile in download_decompress"

This reverts commit 6e67de8.

* refactor: download synchronization with filelock

* filelock logs and poll interval

* remove lockfile.unlink() cause of race condition

* feat: filelock log level increased

Co-authored-by: Fedor Ignatov <[email protected]>

.gitignore

@@ -125,3 +125,8 @@ download/
 
 # project data
 /data/
+
+# local dockerfiles
+/Dockerfile
+/entrypoint.sh
+/.dockerignore

deeppavlov/core/data/utils.py

@@ -97,7 +97,6 @@ def simple_download(url: str, destination: Union[Path, str]) -> None:
         return s3_download(url, str(destination))
 
     chunk_size = 32 * 1024
-
     temporary = destination.with_suffix(destination.suffix + '.part')
 
     headers = {'dp-token': _get_download_token()}

deeppavlov/download.py

@@ -19,8 +19,10 @@
 from logging import getLogger
 from pathlib import Path
 from typing import Union, Optional, Dict, Iterable, Set, Tuple, List
+from urllib.parse import urlparse
 
 import requests
+from filelock import FileLock
 
 import deeppavlov
 from deeppavlov.core.commands.utils import expand_path, parse_config
@@ -124,19 +126,21 @@ def check_md5(url: str, dest_paths: List[Path]) -> bool:
     return True
 
 
-def download_resource(url: str, dest_paths: Iterable[Union[Path, str]]) \
-        -> None:
+def download_resource(url: str, dest_paths: Iterable[Union[Path, str]]) -> None:
     dest_paths = [Path(dest) for dest in dest_paths]
-
-    if check_md5(url, dest_paths):
-        log.info(f'Skipped {url} download because of matching hashes')
-    elif any(ext in url for ext in ('.tar.gz', '.gz', '.zip')):
-        download_path = dest_paths[0].parent
-        download_decompress(url, download_path, dest_paths)
-    else:
-        file_name = url.split('/')[-1].split('?')[0]
-        dest_files = [dest_path / file_name for dest_path in dest_paths]
-        download(dest_files, url)
+    download_path = dest_paths[0].parent
+    download_path.mkdir(parents=True, exist_ok=True)
+    file_name = urlparse(url).path.split('/')[-1]
+    lockfile = download_path / f'.{file_name}.lock'
+
+    with FileLock(lockfile).acquire(poll_intervall=10):
+        if check_md5(url, dest_paths):
+            log.info(f'Skipped {url} download because of matching hashes')
+        elif any(ext in url for ext in ('.tar.gz', '.gz', '.zip')):
+            download_decompress(url, download_path, dest_paths)
+        else:
+            dest_files = [dest_path / file_name for dest_path in dest_paths]
+            download(dest_files, url)
 
 
 def download_resources(args: Namespace) -> None:

deeppavlov/utils/settings/log_config.json

@@ -22,6 +22,13 @@
         "uvicorn_handler"
       ],
       "propagate": true
+    },
+    "filelock": {
+      "level": "WARNING",
+      "handlers": [
+        "stdout"
+      ],
+      "propagate": true
     }
   },
   "formatters": {

requirements.txt

@@ -1,6 +1,7 @@
 aio-pika==6.4.1
 Cython==0.29.14
 fastapi==0.47.1
+filelock==3.0.12
 h5py==2.10.0
 nltk==3.4.5
 numpy==1.18.0