downloader: avoid duplicate downloads (#243)

GitHub: fix GH-242

Red Datasets sometimes handle large files. Downloading the same data
multiple times is not suitable for this use case. We should not download
per worker.

Add a cache check not only before acquiring the lock, but also after
acquiring the lock.

While checking only after acquiring the lock would be enough, it would
create a lock file unnecessarily, so we kept the pre-lock cache check as
well.

This patch will avoid duplicate downloads in the following case:

```mermaid
sequenceDiagram
  participant P1 as Process 1
  participant P2 as Process 2
  participant FS as File System

  P1->>FS: check output_path.exist? => false
  P2->>FS: check output_path.exist? => false
  P1->>FS: create lock => success
  P2->>FS: create lock => failure (sleep 1~10s)
  P1->>FS: download
  P1->>FS: delete lock
  P2->>FS: create lock => success
  Note over P2: No re-check after lock
  P2->>FS: download (duplicate)
  P2->>FS: delete lock
```

---------

Co-authored-by: Sutou Kouhei <kou@clear-code.com>

Author: tikkss

lib/datasets/downloader.rb

@@ -20,13 +20,12 @@ def initialize(url, *fallback_urls, http_method: nil, http_parameters: nil)
     end
 
     def download(output_path, &block)
-      if output_path.exist?
-        yield_chunks(output_path, &block) if block_given?
-        return
-      end
+      return if use_cache(output_path, &block)
 
       partial_output_path = Pathname.new("#{output_path}.partial")
       synchronize(output_path, partial_output_path) do
+        return if use_cache(output_path, &block)
+
         output_path.parent.mkpath
 
         n_retries = 0
@@ -94,6 +93,15 @@ def download(output_path, &block)
       url
     end
 
+    private def use_cache(output_path, &block)
+      if output_path.exist?
+        yield_chunks(output_path, &block) if block_given?
+        true
+      else
+        false
+      end
+    end
+
     private def synchronize(output_path, partial_output_path)
       begin
         Process.getpgid(Process.pid)