Read the individuals archive without extracting it

frequency.py and mutation_overlap.py each extracted chr{N}n.tar.gz into
./chr{N}n before reading it. Every task for a chromosome shares that path -- one
per population in each of the two stages, ten of them at five populations -- and
tarfile.extractall truncates each target before rewriting it, so a task can read
a file another task is midway through replacing.

A reader that catches a file mid-rewrite sees a part-written line. frequency.py
takes the second field of each line and raises IndexError; mutation_overlap.py
splits the whole file and takes the result, so a short read passes silently into
its output. Three Q3 runs died this way, at two different tasks, and the fourth
passed -- and a directory with five concurrent extractors reproduces it at 4
IndexErrors in 29 reader passes, against none with a single extractor.

Both stages now read members straight out of the archive, which removes the
shared path and with it the race. It also drops ten redundant 16MB extractions
per chromosome, since every one of those tasks was already extracting the whole
archive only to overwrite what the others wrote.

Verified against the previous implementation on chr7: the text each stage builds
per individual, in both its own semantics, is identical for all 1,153
individuals across AFR and EUR, as is the sifted-mutation set that flows
downstream.

Worker image 1.6.

Author: balis

.github/workflows/equivalence.yml

@@ -76,7 +76,7 @@ jobs:
           make -C worker-image image
           docker build --platform linux/amd64 \
             -f engines/nextflow/worker-nf.Dockerfile \
-            -t 1000genome-worker-nf:1.5 engines/nextflow/
+            -t 1000genome-worker-nf:1.6 engines/nextflow/
 
       - name: Install Nextflow
         run: |

engines/hyperflow/harness/docker-compose.yml

@@ -32,7 +32,7 @@ services:
       - REDIS_URL=redis://redis:6379
 
       # HyperFlow configuration
-      - HF_VAR_WORKER_CONTAINER=${HF_VAR_WORKER_CONTAINER:-hyperflowwms/1000genome-worker:1.5-je1.4.2}
+      - HF_VAR_WORKER_CONTAINER=${HF_VAR_WORKER_CONTAINER:-hyperflowwms/1000genome-worker:1.6-je1.4.2}
       - HF_VAR_WORK_DIR=${WORKFLOW_DIR}
       - HF_VAR_HFLOW_IN_CONTAINER=true
       - HF_VAR_function=redisCommand

engines/nextflow/worker-nf.Dockerfile

@@ -6,7 +6,7 @@
 # Nextflow does not use. Both engines therefore run the same scripts from the
 # same layer, so "the science is identical" is a property of the image graph
 # rather than a claim about two copies staying in sync.
-FROM hyperflowwms/1000genome-worker-base:1.5
+FROM hyperflowwms/1000genome-worker-base:1.6
 
 # Nextflow requires bash in the container; the base is Alpine, which ships ash.
 RUN apk add --no-cache bash

worker-base-image/Makefile

@@ -1,6 +1,6 @@
 REPO_NAME = 1000genome-worker-base
 PREFIX = hyperflowwms
-VERSION = 1.5
+VERSION = 1.6
 TAG = $(VERSION)
 
 all: push

worker-base-image/scripts/frequency.py

@@ -55,12 +55,18 @@
 font = {'family': 'serif', 'size': 14}
 plt.rc('font', **font)
 
-# untar input data
+# Input data, read straight out of the archive. Extracting it would put every
+# task for this chromosome into one ./chr{N}n directory -- one task per
+# population here and another in mutation_overlap, ten of them at five
+# populations -- where each extraction truncates and rewrites files the others
+# are reading. A reader that catches a file mid-rewrite sees a part-written
+# line, which fails here on the second field and passes silently into
+# mutation_overlap's results. See archive.py.
 import tarfile
 
-tar = tarfile.open(chrom + 'n.tar.gz')
-tar.extractall(path='./' + chrom + 'n')
-tar.close()
+import archive
+
+individuals = dict(archive.read_archive(chrom + 'n.tar.gz'))
 
 
 class ReadData:
@@ -108,10 +114,8 @@ def read_individuals(self, ids, rs_numbers):
         tic = time.perf_counter()
         mutation_index_array = []
         for name in ids:
-            filename = data_dir + chrom + 'n/' + chrom + '.' + name
-            f = open(filename, 'r')
             text = []
-            for item in f:
+            for item in individuals[chrom + '.' + name]:
                 item = item.split()
                 text.append(item[1])
             sifted_mutations = list(set(rs_numbers).intersection(text))

worker-base-image/scripts/mutation_overlap.py

@@ -56,11 +56,17 @@
 plt.rc('font', **font)
 
 
-# untar input data
+# Input data, read straight out of the archive. Extracting it would put every
+# task for this chromosome into one ./chr{N}n directory -- one task per
+# population here and another in frequency, ten of them at five populations --
+# where each extraction truncates and rewrites files the others are reading. A
+# reader that catches a file mid-rewrite sees a part-written line, which passes
+# silently into the results here and fails outright in frequency. See archive.py.
 import tarfile
-tar = tarfile.open(chrom + 'n.tar.gz')
-tar.extractall(path='./' + chrom + 'n')
-tar.close()
+
+import archive
+
+individuals = dict(archive.read_archive(chrom + 'n.tar.gz'))
 
 tic = time.perf_counter()
 
@@ -115,11 +121,7 @@ def read_individuals(self, ids, rs_numbers) :
         total_mutations={}  
         total_mutations_list =[]    
         for name in ids :
-            filename = data_dir + chrom + 'n/' + chrom + '.' + name
-            f = open(filename, 'r')
-            text = f.read()
-            f.close()
-            text = text.split()
+            text = ''.join(individuals[chrom + '.' + name]).split()
             sifted_mutations = list(set(rs_numbers).intersection(text))
             mutation_index_array.append(sifted_mutations)
             total_mutations[name]= len(sifted_mutations)

worker-image/Dockerfile

@@ -1,4 +1,4 @@
-FROM hyperflowwms/1000genome-worker-base:1.5
+FROM hyperflowwms/1000genome-worker-base:1.6
 
 # Version of the job executor should be passed via docker build, e.g.:
 # docker build --build-arg hf_job_executor_version="1.3.4"

worker-image/Makefile

@@ -1,6 +1,6 @@
 REPO_NAME = 1000genome-worker
 PREFIX = hyperflowwms
-VERSION = 1.5
+VERSION = 1.6
 # ?= so CI and Renovate can override the version from the environment
 HF_JOB_EXECUTOR_VERSION ?= 1.4.2
 TAG = $(VERSION)-je$(HF_JOB_EXECUTOR_VERSION)