From f49df91b9dcdc192d3bf0b168c65b2bc245a2ae0 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 2 Jun 2023 16:41:39 +0100
Subject: [PATCH 01/36] resume feature added

---
 .gitignore                          |  4 ++
 requirements.txt                    |  1 +
 src/resume.py                       | 32 ++++++++++++++
 src/s3_md5.py                       | 65 +++++++++++++++++++----------
 test/test_parse_file_md5.py         |  2 +-
 test/test_resumed_parse_file_md5.py | 26 ++++++++++++
 6 files changed, 108 insertions(+), 22 deletions(-)
 create mode 100644 src/resume.py
 create mode 100644 test/test_resumed_parse_file_md5.py

diff --git a/.gitignore b/.gitignore
index fbd405a..2d7c511 100644
--- a/.gitignore
+++ b/.gitignore
@@ -128,3 +128,7 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
+
+# Src specific
+.state.block
+.state.pickle
diff --git a/requirements.txt b/requirements.txt
index adb565a..a87d8bc 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,2 +1,3 @@
 boto3-stubs[s3]
 boto3==1.26.41
+rehash==1.0.1
diff --git a/src/resume.py b/src/resume.py
new file mode 100644
index 0000000..e6330d0
--- /dev/null
+++ b/src/resume.py
@@ -0,0 +1,32 @@
+"""
+This module is responsible for loading the state of the hashing process.
+"""
+import pickle
+from typing import Tuple
+from src.logger import logger
+import rehash
+
+
+def load_state() -> Tuple[int, rehash.ResumableHasher]:
+    '''try and load existing state'''
+    try:
+        with open('.state.block', 'r',  encoding='utf-8') as previous_block_number, \
+                open('.state.pickle', 'rb') as previous_hash:
+
+            hash_object = pickle.load(previous_hash)
+            start_block = int(previous_block_number.read()) + 1
+            return start_block, hash_object
+    except FileNotFoundError:
+        return 0, rehash.new('md5')
+
+
+def save_state(block_number: int, hash_object: rehash.ResumableHasher) -> None:
+    '''try and save state'''
+    try:
+        with open('.state.block', 'w',  encoding='utf-8') as current_block_number, \
+                open('.state.pickle', 'wb') as current_hash:
+            pickle.dump(hash_object, current_hash)
+            current_block_number.write(str(block_number))
+
+    except OSError as exception:
+        raise OSError('Could not save state') from exception
diff --git a/src/s3_md5.py b/src/s3_md5.py
index 0cf9397..2c37a3a 100644
--- a/src/s3_md5.py
+++ b/src/s3_md5.py
@@ -1,11 +1,12 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
 from concurrent.futures import ThreadPoolExecutor
-from hashlib import md5
+from os import remove
 
 from mypy_boto3_s3 import S3Client
 
 from src.logger import logger
 from src.s3_file import S3FileHelper
+from src.resume import load_state, save_state
 
 
 def parse_file_md5(s3_client: S3Client,
@@ -20,23 +21,45 @@ def parse_file_md5(s3_client: S3Client,
     if file_size < chunk_size:
         raise AssertionError('file size cannot be smaller than chunk size')
     logger.info(f'file size {file_size} bytes')
-    file_chunk_count = file_size // chunk_size
-    logger.info(f'file chunk count {file_chunk_count}')
-
-    with ThreadPoolExecutor(max_workers=workers) as thread_executor:
-        def wrapper(part_number: int):
-            ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
-                part_number, chunk_size, file_chunk_count)
-            logger.info(f"downloading {ranged_bytes_string}")
-            ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
-            logger.info(f"downloaded {ranged_bytes_string}")
-            return ranged_bytes
-
-        logger.info('downloading file')
-        results = thread_executor.map(wrapper,
-                                      range(file_chunk_count))
-
-        hash_object = md5()
-        for result in results:
-            hash_object.update(result)
-        return hash_object.hexdigest()
+
+    chunk_count = file_size // chunk_size
+    logger.info(f'chunk count {chunk_count}')
+
+    if chunk_count < workers:
+        raise AssertionError('chunk size too large')
+
+    block_count = chunk_count // workers
+    logger.info(f'block count {block_count}')
+
+    chunk_count_per_block = chunk_count // block_count
+    logger.info(f'chunk to get per block {chunk_count_per_block}')
+
+    start_block, hash_object = load_state()
+    if start_block != 0:
+        logger.info(f"resuming from block {start_block}")
+
+    for block_number in range(start_block, block_count):
+        logger.info(f"processing block {block_number}")
+        start_part_number = block_number * chunk_count_per_block
+        end_part_number = chunk_count - 1 if block_number == block_count - \
+            1 else (start_part_number + chunk_count_per_block) - 1
+        logger.info(f"part number {start_part_number}-{end_part_number}")
+
+        with ThreadPoolExecutor(max_workers=workers) as thread_executor:
+            def wrapper(part_number: int):
+                ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
+                    part_number, chunk_size, chunk_count)
+                logger.debug(f"downloading {ranged_bytes_string}")
+                ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
+                logger.debug(f"downloaded {ranged_bytes_string}")
+                return ranged_bytes
+
+            for result in thread_executor.map(wrapper,
+                                              range(start_part_number, end_part_number + 1)):
+                hash_object.update(result)
+            save_state(block_number, hash_object)
+
+    remove('.state.block')
+    remove('.state.pickle')
+
+    return hash_object.hexdigest()
diff --git a/test/test_parse_file_md5.py b/test/test_parse_file_md5.py
index a1d0931..c89cee9 100644
--- a/test/test_parse_file_md5.py
+++ b/test/test_parse_file_md5.py
@@ -7,7 +7,7 @@
 from src.s3_md5 import parse_file_md5
 
 
-def test_get_md5_hash(s3_setup: Tuple[S3Client, str, str, str]):
+def test_parse_file_md5(s3_setup: Tuple[S3Client, str, str, str]):
     '''test function'''
     s3_client, test_bucket, test_file_name, test_body = s3_setup
     md5_hash = parse_file_md5(s3_client, test_bucket, test_file_name, 1, 1)
diff --git a/test/test_resumed_parse_file_md5.py b/test/test_resumed_parse_file_md5.py
new file mode 100644
index 0000000..ad8531d
--- /dev/null
+++ b/test/test_resumed_parse_file_md5.py
@@ -0,0 +1,26 @@
+'''tests the driver function'''
+from hashlib import md5
+from pickle import dump
+from typing import Tuple
+
+from mypy_boto3_s3 import S3Client
+from rehash import new
+
+from src.s3_md5 import parse_file_md5
+
+
+def test_resumed_parse_file_md5(s3_setup: Tuple[S3Client, str, str, str]):
+    '''test function'''
+    s3_client, test_bucket, test_file_name, test_body = s3_setup
+
+    with open('.state.block', 'w', encoding="utf-8") as previous_block_number:
+        previous_block_number.write('1')
+
+    hash_object = new('md5')
+    hash_object.update(bytes(test_body[0:2], 'utf-8'))
+
+    with open('.state.pickle', 'wb') as previous_md5_hash:
+        dump(hash_object, previous_md5_hash)
+
+    md5_hash = parse_file_md5(s3_client, test_bucket, test_file_name, 1, 1)
+    assert md5_hash == md5(bytes(test_body, 'utf-8')).hexdigest()

From 13ffc51c0305519f9bbf434ff8ce77f6b2fa2748 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 2 Jun 2023 16:43:48 +0100
Subject: [PATCH 02/36] lib update

---
 dev-requirements.txt | 1 +
 1 file changed, 1 insertion(+)

diff --git a/dev-requirements.txt b/dev-requirements.txt
index 13f18ee..501db24 100644
--- a/dev-requirements.txt
+++ b/dev-requirements.txt
@@ -4,3 +4,4 @@ boto3==1.26.41
 moto==4.0.12
 mypy==1.1.1
 pytest==7.2.0
+rehash==1.0.1

From 36673c7db1c0812072d3dcdd9bdc7b253ded646b Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 2 Jun 2023 16:50:40 +0100
Subject: [PATCH 03/36] openssl install

---
 .github/workflows/main.yaml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index a068116..df56f9d 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -11,7 +11,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v4
         with:
-          python-version: '3.10'
+          python-version: '3.8'
 
       - name: Install Dependencies
         run: pip install -r dev-requirements.txt

From 40a576d118a3016cb354d900420e41b6996cd723 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 10:05:35 +0100
Subject: [PATCH 04/36] readme

---
 README.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/README.md b/README.md
index 7ab6e88..5518925 100644
--- a/README.md
+++ b/README.md
@@ -4,6 +4,7 @@ Get fast md5 hashes for an s3 file.
 - python 3.10
 - boto3
 - boto3-stubs[s3]
+- rehash
 
 ## how to use
 From the command line run

From 9b722459248456a6e40aa542eb1132040599b7cf Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 10:56:26 +0100
Subject: [PATCH 05/36] wip

---
 src/s3_md5.py | 46 +++++++++++++++++++++++++++-------------------
 1 file changed, 27 insertions(+), 19 deletions(-)

diff --git a/src/s3_md5.py b/src/s3_md5.py
index 2c37a3a..673baed 100644
--- a/src/s3_md5.py
+++ b/src/s3_md5.py
@@ -1,12 +1,19 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
-from concurrent.futures import ThreadPoolExecutor
-from os import remove
+from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, Future
+from hashlib import md5
+from typing import Iterator, List, Any
 
 from mypy_boto3_s3 import S3Client
 
 from src.logger import logger
 from src.s3_file import S3FileHelper
-from src.resume import load_state, save_state
+
+
+def process_md5_block(block: Iterator[bytes], hash_object):
+    '''generate md5 hash for each block'''
+    logger.info("running")
+    for chunk in block:
+        hash_object.update(chunk)
 
 
 def parse_file_md5(s3_client: S3Client,
@@ -34,10 +41,11 @@ def parse_file_md5(s3_client: S3Client,
     chunk_count_per_block = chunk_count // block_count
     logger.info(f'chunk to get per block {chunk_count_per_block}')
 
-    start_block, hash_object = load_state()
+    start_block, hash_object = 0, md5()
     if start_block != 0:
         logger.info(f"resuming from block {start_block}")
 
+    block_calculator: List[Future[Any]] = []
     for block_number in range(start_block, block_count):
         logger.info(f"processing block {block_number}")
         start_part_number = block_number * chunk_count_per_block
@@ -45,21 +53,21 @@ def parse_file_md5(s3_client: S3Client,
             1 else (start_part_number + chunk_count_per_block) - 1
         logger.info(f"part number {start_part_number}-{end_part_number}")
 
-        with ThreadPoolExecutor(max_workers=workers) as thread_executor:
-            def wrapper(part_number: int):
-                ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
-                    part_number, chunk_size, chunk_count)
-                logger.debug(f"downloading {ranged_bytes_string}")
-                ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
-                logger.debug(f"downloaded {ranged_bytes_string}")
-                return ranged_bytes
-
-            for result in thread_executor.map(wrapper,
-                                              range(start_part_number, end_part_number + 1)):
-                hash_object.update(result)
-            save_state(block_number, hash_object)
+        with ProcessPoolExecutor(max_workers=workers//2) as process_executor:
+            with ThreadPoolExecutor(max_workers=workers) as thread_executor:
+                def wrapper(part_number: int):
+                    ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
+                        part_number, chunk_size, chunk_count)
+                    logger.debug(f"downloading {ranged_bytes_string}")
+                    ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
+                    logger.debug(f"downloaded {ranged_bytes_string}")
+                    return ranged_bytes
 
-    remove('.state.block')
-    remove('.state.pickle')
+                results = thread_executor.map(wrapper, range(
+                    start_part_number, end_part_number + 1))
 
+    #             block_calculator.append(process_executor.submit(
+    #                 process_md5_block, results, hash_object))
+    # for p in block_calculator:
+    #     print(p.done())
     return hash_object.hexdigest()

From 0df70825b60ec8ef303c79692755953e31d5165f Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 12:35:35 +0100
Subject: [PATCH 06/36] using separate process

---
 .tool-versions       |   2 +-
 README.md            |   3 +-
 dev-requirements.txt |   1 -
 src/resume.py        |  32 --------------
 src/s3_md5.py        | 101 +++++++++++++++++++++++++++----------------
 5 files changed, 65 insertions(+), 74 deletions(-)
 delete mode 100644 src/resume.py

diff --git a/.tool-versions b/.tool-versions
index 411c63d..5922314 100644
--- a/.tool-versions
+++ b/.tool-versions
@@ -1 +1 @@
-python 3.8.16
+python 3.10.0
diff --git a/README.md b/README.md
index 5518925..17769ad 100644
--- a/README.md
+++ b/README.md
@@ -1,10 +1,9 @@
 # s3-md5
-Get fast md5 hashes for an s3 file.
+Get fast md5 hashes for an s3 file. It a process that listens on the incoming chunks via a process queue.
 ## installation
 - python 3.10
 - boto3
 - boto3-stubs[s3]
-- rehash
 
 ## how to use
 From the command line run
diff --git a/dev-requirements.txt b/dev-requirements.txt
index 501db24..13f18ee 100644
--- a/dev-requirements.txt
+++ b/dev-requirements.txt
@@ -4,4 +4,3 @@ boto3==1.26.41
 moto==4.0.12
 mypy==1.1.1
 pytest==7.2.0
-rehash==1.0.1
diff --git a/src/resume.py b/src/resume.py
deleted file mode 100644
index e6330d0..0000000
--- a/src/resume.py
+++ /dev/null
@@ -1,32 +0,0 @@
-"""
-This module is responsible for loading the state of the hashing process.
-"""
-import pickle
-from typing import Tuple
-from src.logger import logger
-import rehash
-
-
-def load_state() -> Tuple[int, rehash.ResumableHasher]:
-    '''try and load existing state'''
-    try:
-        with open('.state.block', 'r',  encoding='utf-8') as previous_block_number, \
-                open('.state.pickle', 'rb') as previous_hash:
-
-            hash_object = pickle.load(previous_hash)
-            start_block = int(previous_block_number.read()) + 1
-            return start_block, hash_object
-    except FileNotFoundError:
-        return 0, rehash.new('md5')
-
-
-def save_state(block_number: int, hash_object: rehash.ResumableHasher) -> None:
-    '''try and save state'''
-    try:
-        with open('.state.block', 'w',  encoding='utf-8') as current_block_number, \
-                open('.state.pickle', 'wb') as current_hash:
-            pickle.dump(hash_object, current_hash)
-            current_block_number.write(str(block_number))
-
-    except OSError as exception:
-        raise OSError('Could not save state') from exception
diff --git a/src/s3_md5.py b/src/s3_md5.py
index 673baed..8784e78 100644
--- a/src/s3_md5.py
+++ b/src/s3_md5.py
@@ -1,19 +1,60 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
-from concurrent.futures import ProcessPoolExecutor, ThreadPoolExecutor, Future
+from concurrent.futures import ThreadPoolExecutor
 from hashlib import md5
-from typing import Iterator, List, Any
+from multiprocessing import Process, Queue, Manager
+from multiprocessing.managers import ValueProxy
+from typing import Type
 
 from mypy_boto3_s3 import S3Client
 
 from src.logger import logger
+
 from src.s3_file import S3FileHelper
 
 
-def process_md5_block(block: Iterator[bytes], hash_object):
-    '''generate md5 hash for each block'''
-    logger.info("running")
-    for chunk in block:
-        hash_object.update(chunk)
+def consumer(queue: Queue, variable: ValueProxy[str]):
+    '''a process that subscribes to the queue and processes md5'''
+    hasher = md5()
+    while True:
+        item = queue.get()
+        if item is None:
+            break
+        logger.info("consuming range")
+        hasher.update(item)
+    md5_hash = hasher.hexdigest()
+    logger.info(md5_hash)
+    variable.value = md5_hash
+
+
+def process_block(block_number: int,
+                  block_count: int,
+                  chunk_count: int,
+                  chunk_size: int,
+                  chunk_count_per_block: int,
+                  workers: int,
+                  queue: Queue,
+                  s3_file: S3FileHelper):
+    '''runs individual blocks into a separate process'''
+    logger.info(f"processing block {block_number}")
+    start_part_number = block_number * chunk_count_per_block
+    end_part_number = chunk_count - 1 if block_number == block_count - \
+        1 else (start_part_number + chunk_count_per_block) - 1
+    logger.info(f"part number {start_part_number}-{end_part_number}")
+
+    with ThreadPoolExecutor(max_workers=workers) as thread_executor:
+        def wrapper(part_number: int):
+            ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
+                part_number, chunk_size, chunk_count)
+            logger.debug(f"downloading {ranged_bytes_string}")
+            ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
+            logger.debug(f"downloaded {ranged_bytes_string}")
+            return ranged_bytes
+
+        results = thread_executor.map(wrapper, range(
+            start_part_number, end_part_number + 1))
+
+        for result in results:
+            queue.put(result)
 
 
 def parse_file_md5(s3_client: S3Client,
@@ -28,12 +69,14 @@ def parse_file_md5(s3_client: S3Client,
     if file_size < chunk_size:
         raise AssertionError('file size cannot be smaller than chunk size')
     logger.info(f'file size {file_size} bytes')
+    logger.info(f'chunk size {chunk_size} bytes')
+    logger.info(f'workers {workers}')
 
     chunk_count = file_size // chunk_size
     logger.info(f'chunk count {chunk_count}')
 
     if chunk_count < workers:
-        raise AssertionError('chunk size too large')
+        raise AssertionError('chunk count cannot be smaller than workers')
 
     block_count = chunk_count // workers
     logger.info(f'block count {block_count}')
@@ -41,33 +84,15 @@ def parse_file_md5(s3_client: S3Client,
     chunk_count_per_block = chunk_count // block_count
     logger.info(f'chunk to get per block {chunk_count_per_block}')
 
-    start_block, hash_object = 0, md5()
-    if start_block != 0:
-        logger.info(f"resuming from block {start_block}")
-
-    block_calculator: List[Future[Any]] = []
-    for block_number in range(start_block, block_count):
-        logger.info(f"processing block {block_number}")
-        start_part_number = block_number * chunk_count_per_block
-        end_part_number = chunk_count - 1 if block_number == block_count - \
-            1 else (start_part_number + chunk_count_per_block) - 1
-        logger.info(f"part number {start_part_number}-{end_part_number}")
-
-        with ProcessPoolExecutor(max_workers=workers//2) as process_executor:
-            with ThreadPoolExecutor(max_workers=workers) as thread_executor:
-                def wrapper(part_number: int):
-                    ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
-                        part_number, chunk_size, chunk_count)
-                    logger.debug(f"downloading {ranged_bytes_string}")
-                    ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
-                    logger.debug(f"downloaded {ranged_bytes_string}")
-                    return ranged_bytes
-
-                results = thread_executor.map(wrapper, range(
-                    start_part_number, end_part_number + 1))
-
-    #             block_calculator.append(process_executor.submit(
-    #                 process_md5_block, results, hash_object))
-    # for p in block_calculator:
-    #     print(p.done())
-    return hash_object.hexdigest()
+    queue = Queue()
+    variable = Manager().Value(str, None)
+
+    consumer_process = Process(target=consumer, args=(queue, variable))
+    consumer_process.start()
+
+    for block_number in range(0, block_count):
+        process_block(block_number, block_count, chunk_count,
+                      chunk_size, chunk_count_per_block, workers // 2, queue, s3_file)
+    queue.put(None)
+    consumer_process.join()
+    return variable.value

From 9a915ce72acb03f85d2810924af58e6203807842 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 12:39:54 +0100
Subject: [PATCH 07/36] typo on python version

---
 .github/workflows/main.yaml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/.github/workflows/main.yaml b/.github/workflows/main.yaml
index df56f9d..a068116 100644
--- a/.github/workflows/main.yaml
+++ b/.github/workflows/main.yaml
@@ -11,7 +11,7 @@ jobs:
       - name: Setup Python
         uses: actions/setup-python@v4
         with:
-          python-version: '3.8'
+          python-version: '3.10'
 
       - name: Install Dependencies
         run: pip install -r dev-requirements.txt

From 9c66ff2299b459ad040a7bc1784aeb803df38693 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 12:40:31 +0100
Subject: [PATCH 08/36] gitignore update

---
 .gitignore | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/.gitignore b/.gitignore
index 2d7c511..fbd405a 100644
--- a/.gitignore
+++ b/.gitignore
@@ -128,7 +128,3 @@ dmypy.json
 
 # Pyre type checker
 .pyre/
-
-# Src specific
-.state.block
-.state.pickle

From 41a5e25a5dec98171dbfa9b15689a6230cf5a019 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 12:40:54 +0100
Subject: [PATCH 09/36] req update

---
 requirements.txt | 1 -
 1 file changed, 1 deletion(-)

diff --git a/requirements.txt b/requirements.txt
index a87d8bc..adb565a 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,3 +1,2 @@
 boto3-stubs[s3]
 boto3==1.26.41
-rehash==1.0.1

From e4da486b826d828b460179435c850a56ce95a2a4 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 12:42:54 +0100
Subject: [PATCH 10/36] test removed

---
 src/s3_md5.py                       |  4 +---
 test/test_resumed_parse_file_md5.py | 26 --------------------------
 2 files changed, 1 insertion(+), 29 deletions(-)
 delete mode 100644 test/test_resumed_parse_file_md5.py

diff --git a/src/s3_md5.py b/src/s3_md5.py
index 8784e78..165490d 100644
--- a/src/s3_md5.py
+++ b/src/s3_md5.py
@@ -1,14 +1,12 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
 from concurrent.futures import ThreadPoolExecutor
 from hashlib import md5
-from multiprocessing import Process, Queue, Manager
+from multiprocessing import Manager, Process, Queue
 from multiprocessing.managers import ValueProxy
-from typing import Type
 
 from mypy_boto3_s3 import S3Client
 
 from src.logger import logger
-
 from src.s3_file import S3FileHelper
 
 
diff --git a/test/test_resumed_parse_file_md5.py b/test/test_resumed_parse_file_md5.py
deleted file mode 100644
index ad8531d..0000000
--- a/test/test_resumed_parse_file_md5.py
+++ /dev/null
@@ -1,26 +0,0 @@
-'''tests the driver function'''
-from hashlib import md5
-from pickle import dump
-from typing import Tuple
-
-from mypy_boto3_s3 import S3Client
-from rehash import new
-
-from src.s3_md5 import parse_file_md5
-
-
-def test_resumed_parse_file_md5(s3_setup: Tuple[S3Client, str, str, str]):
-    '''test function'''
-    s3_client, test_bucket, test_file_name, test_body = s3_setup
-
-    with open('.state.block', 'w', encoding="utf-8") as previous_block_number:
-        previous_block_number.write('1')
-
-    hash_object = new('md5')
-    hash_object.update(bytes(test_body[0:2], 'utf-8'))
-
-    with open('.state.pickle', 'wb') as previous_md5_hash:
-        dump(hash_object, previous_md5_hash)
-
-    md5_hash = parse_file_md5(s3_client, test_bucket, test_file_name, 1, 1)
-    assert md5_hash == md5(bytes(test_body, 'utf-8')).hexdigest()

From 44aa2dcd3f9dd8d70bf2b29a1d11de787ad3ab69 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 12:46:30 +0100
Subject: [PATCH 11/36] test fix

---
 test/test_parse_file_md5.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/test/test_parse_file_md5.py b/test/test_parse_file_md5.py
index c89cee9..2fed6a1 100644
--- a/test/test_parse_file_md5.py
+++ b/test/test_parse_file_md5.py
@@ -10,5 +10,5 @@
 def test_parse_file_md5(s3_setup: Tuple[S3Client, str, str, str]):
     '''test function'''
     s3_client, test_bucket, test_file_name, test_body = s3_setup
-    md5_hash = parse_file_md5(s3_client, test_bucket, test_file_name, 1, 1)
+    md5_hash = parse_file_md5(s3_client, test_bucket, test_file_name, 1, 2)
     assert md5_hash == md5(bytes(test_body, 'utf-8')).hexdigest()

From 8b8d196ec21075f51de98524d5cb9148f5e53679 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Thu, 24 Aug 2023 12:47:20 +0100
Subject: [PATCH 12/36] init value

---
 src/s3_md5.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/s3_md5.py b/src/s3_md5.py
index 165490d..6277264 100644
--- a/src/s3_md5.py
+++ b/src/s3_md5.py
@@ -83,7 +83,7 @@ def parse_file_md5(s3_client: S3Client,
     logger.info(f'chunk to get per block {chunk_count_per_block}')
 
     queue = Queue()
-    variable = Manager().Value(str, None)
+    variable = Manager().Value(str, '')
 
     consumer_process = Process(target=consumer, args=(queue, variable))
     consumer_process.start()

From c9923918ca96e5c823f8f6c1dce43eeb12da043f Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 25 Aug 2023 15:09:56 +0100
Subject: [PATCH 13/36] wip

---
 src/s3_md5.py  | 33 +++++++++++++++++++++++----------
 src/signals.py |  3 +++
 2 files changed, 26 insertions(+), 10 deletions(-)
 create mode 100644 src/signals.py

diff --git a/src/s3_md5.py b/src/s3_md5.py
index 6277264..e874647 100644
--- a/src/s3_md5.py
+++ b/src/s3_md5.py
@@ -8,17 +8,24 @@
 
 from src.logger import logger
 from src.s3_file import S3FileHelper
+from src.signals import CANCEL, COMPLETE
 
 
 def consumer(queue: Queue, variable: ValueProxy[str]):
     '''a process that subscribes to the queue and processes md5'''
     hasher = md5()
     while True:
-        item = queue.get()
-        if item is None:
-            break
-        logger.info("consuming range")
-        hasher.update(item)
+        try:
+            item = queue.get()
+            if item == COMPLETE:
+                break
+            if item == CANCEL:
+                raise ValueError("one of the processes failed")
+            logger.info("consuming range")
+            hasher.update(item)
+        except Exception as exception:  # pylint: disable=broad-except
+            logger.error(exception)
+            return
     md5_hash = hasher.hexdigest()
     logger.info(md5_hash)
     variable.value = md5_hash
@@ -43,9 +50,9 @@ def process_block(block_number: int,
         def wrapper(part_number: int):
             ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
                 part_number, chunk_size, chunk_count)
-            logger.debug(f"downloading {ranged_bytes_string}")
+            logger.info(f"downloading {ranged_bytes_string}")
             ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
-            logger.debug(f"downloaded {ranged_bytes_string}")
+            logger.info(f"downloaded {ranged_bytes_string}")
             return ranged_bytes
 
         results = thread_executor.map(wrapper, range(
@@ -89,8 +96,14 @@ def parse_file_md5(s3_client: S3Client,
     consumer_process.start()
 
     for block_number in range(0, block_count):
-        process_block(block_number, block_count, chunk_count,
-                      chunk_size, chunk_count_per_block, workers // 2, queue, s3_file)
-    queue.put(None)
+        try:
+            process_block(block_number, block_count, chunk_count,
+                          chunk_size, chunk_count_per_block, workers // 2, queue, s3_file)
+        except Exception as exception:  # pylint: disable=broad-except
+            logger.error(exception)
+            queue.put(CANCEL)
+            raise ValueError from exception
+
+    queue.put(COMPLETE)
     consumer_process.join()
     return variable.value
diff --git a/src/signals.py b/src/signals.py
new file mode 100644
index 0000000..bbca98a
--- /dev/null
+++ b/src/signals.py
@@ -0,0 +1,3 @@
+'''signal values'''
+CANCEL = "CANCEL"
+COMPLETE = "COMPLETE"

From 7840e1128c8c953da6fa2949ce02389a8b306da4 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sat, 30 Mar 2024 22:15:31 +0000
Subject: [PATCH 14/36] relative import updates

---
 s3_md5/src/s3_md5.py | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index e874647..038640e 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -6,9 +6,9 @@
 
 from mypy_boto3_s3 import S3Client
 
-from src.logger import logger
-from src.s3_file import S3FileHelper
-from src.signals import CANCEL, COMPLETE
+from .logger import logger
+from .s3_file import S3FileHelper
+from .signals import CANCEL, COMPLETE
 
 
 def consumer(queue: Queue, variable: ValueProxy[str]):

From 4232de0d0f3a4a63c247da5c2ce80531d679947f Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sat, 30 Mar 2024 23:05:03 +0000
Subject: [PATCH 15/36] updated types

---
 s3_md5/src/s3_md5.py  | 31 ++++++++++++++++---------------
 s3_md5/src/signals.py |  8 ++++++--
 2 files changed, 22 insertions(+), 17 deletions(-)

diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 038640e..ae8fc6c 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -8,10 +8,12 @@
 
 from .logger import logger
 from .s3_file import S3FileHelper
-from .signals import CANCEL, COMPLETE
+from .signals import CANCEL, COMPLETE, cancel_type, complete_type
 
+queue_type = Queue[bytes | cancel_type | complete_type]
 
-def consumer(queue: Queue, variable: ValueProxy[str]):
+
+def consumer(queue: queue_type, variable: ValueProxy[str]):
     '''a process that subscribes to the queue and processes md5'''
     hasher = md5()
     while True:
@@ -31,14 +33,13 @@ def consumer(queue: Queue, variable: ValueProxy[str]):
     variable.value = md5_hash
 
 
-def process_block(block_number: int,
-                  block_count: int,
-                  chunk_count: int,
-                  chunk_size: int,
-                  chunk_count_per_block: int,
-                  workers: int,
-                  queue: Queue,
-                  s3_file: S3FileHelper):
+def fetch_block(block_number: int,
+                block_count: int,
+                chunk_count: int,
+                chunk_size: int,
+                chunk_count_per_block: int,
+                workers: int,
+                s3_file: S3FileHelper):
     '''runs individual blocks into a separate process'''
     logger.info(f"processing block {block_number}")
     start_part_number = block_number * chunk_count_per_block
@@ -58,8 +59,7 @@ def wrapper(part_number: int):
         results = thread_executor.map(wrapper, range(
             start_part_number, end_part_number + 1))
 
-        for result in results:
-            queue.put(result)
+        return results
 
 
 def parse_file_md5(s3_client: S3Client,
@@ -89,7 +89,7 @@ def parse_file_md5(s3_client: S3Client,
     chunk_count_per_block = chunk_count // block_count
     logger.info(f'chunk to get per block {chunk_count_per_block}')
 
-    queue = Queue()
+    queue: queue_type = Queue()
     variable = Manager().Value(str, '')
 
     consumer_process = Process(target=consumer, args=(queue, variable))
@@ -97,8 +97,9 @@ def parse_file_md5(s3_client: S3Client,
 
     for block_number in range(0, block_count):
         try:
-            process_block(block_number, block_count, chunk_count,
-                          chunk_size, chunk_count_per_block, workers // 2, queue, s3_file)
+            for item in fetch_block(block_number, block_count, chunk_count,
+                                    chunk_size, chunk_count_per_block, workers // 2, s3_file):
+                queue.put(item)
         except Exception as exception:  # pylint: disable=broad-except
             logger.error(exception)
             queue.put(CANCEL)
diff --git a/s3_md5/src/signals.py b/s3_md5/src/signals.py
index bbca98a..eedf9fa 100644
--- a/s3_md5/src/signals.py
+++ b/s3_md5/src/signals.py
@@ -1,3 +1,7 @@
+from typing import Literal
+
 '''signal values'''
-CANCEL = "CANCEL"
-COMPLETE = "COMPLETE"
+cancel_type = Literal['CANCEL']
+complete_type = Literal['COMPLETE']
+CANCEL: cancel_type = "CANCEL"
+COMPLETE: complete_type = "COMPLETE"

From a1bc71108ec47fda9728f185783509bcd4e61fe8 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 09:36:45 +0100
Subject: [PATCH 16/36] type fixes

---
 s3_md5/src/s3_md5.py  | 8 +++-----
 s3_md5/src/signals.py | 9 +++------
 2 files changed, 6 insertions(+), 11 deletions(-)

diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index ae8fc6c..f6a12f1 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -8,12 +8,10 @@
 
 from .logger import logger
 from .s3_file import S3FileHelper
-from .signals import CANCEL, COMPLETE, cancel_type, complete_type
+from .signals import CANCEL, COMPLETE
 
-queue_type = Queue[bytes | cancel_type | complete_type]
 
-
-def consumer(queue: queue_type, variable: ValueProxy[str]):
+def consumer(queue: Queue, variable: ValueProxy[str]):
     '''a process that subscribes to the queue and processes md5'''
     hasher = md5()
     while True:
@@ -89,7 +87,7 @@ def parse_file_md5(s3_client: S3Client,
     chunk_count_per_block = chunk_count // block_count
     logger.info(f'chunk to get per block {chunk_count_per_block}')
 
-    queue: queue_type = Queue()
+    queue = Queue()
     variable = Manager().Value(str, '')
 
     consumer_process = Process(target=consumer, args=(queue, variable))
diff --git a/s3_md5/src/signals.py b/s3_md5/src/signals.py
index eedf9fa..6c9ed98 100644
--- a/s3_md5/src/signals.py
+++ b/s3_md5/src/signals.py
@@ -1,7 +1,4 @@
-from typing import Literal
-
 '''signal values'''
-cancel_type = Literal['CANCEL']
-complete_type = Literal['COMPLETE']
-CANCEL: cancel_type = "CANCEL"
-COMPLETE: complete_type = "COMPLETE"
+
+CANCEL = "CANCEL"
+COMPLETE = "COMPLETE"

From 7d40e6fc07c2cd2874dc71f751f366f4e5c5432a Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 12:16:49 +0100
Subject: [PATCH 17/36] updated readme

---
 README.md | 16 ++++++++++++----
 1 file changed, 12 insertions(+), 4 deletions(-)

diff --git a/README.md b/README.md
index abe4e73..693524a 100644
--- a/README.md
+++ b/README.md
@@ -2,11 +2,11 @@
 
 Get fast md5 hashes for an s3 file.
 
-## installation
+## Requirements
 
 -   python 3.10
 
-## how to use
+## Usage
 
 You can use the tool as a command line argument. You can download the latest release from [here](https://github.com/sakibstark11/s3-md5-python/releases). You can also build the wheel file yourself by running the following command.
 
@@ -38,11 +38,19 @@ Or you can directly invoke the script by running
 python s3_md5/main.py <bucket_name> <file_name>
 ```
 
+### Arguments
+
 There are two _optional_ arguments that you may want to provide
 
 -   `-w` or workers sets the number of python threads to use for downloading purposes, by default its set to the following equation `number of cpu cores * 2 - 1`
--   `-c` or chunk size in **bytes** sets the individual download size on each get request sent to s3, by default its set to `1000000`
+-   `-c` or chunk size in **bytes** sets the individual download size on each get request sent to s3, by default its set to `1000000`. For optimal performance you need to saturate the available bandwidth on your machine. Try running a few sample file to work out what chunk size would work best.
+
+### Example
+
+for a file size of `1048576000` bytes
+on a 250 mpbs bandwidth
+a chunk size of `4000000` works the best as it completes it within ~90 seconds
 
-## caveats
+## Caveats
 
 -   File size can not be smaller than the default chunk size of `1000000`, if yes, then the chunk size must be manually provided or it will raise an assertion error.

From 2fd79a143c5ff4f191659f4525308f169040832c Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 12:33:57 +0100
Subject: [PATCH 18/36] updated readme

---
 README.md | 1 +
 1 file changed, 1 insertion(+)

diff --git a/README.md b/README.md
index 693524a..a004fa8 100644
--- a/README.md
+++ b/README.md
@@ -49,6 +49,7 @@ There are two _optional_ arguments that you may want to provide
 
 for a file size of `1048576000` bytes
 on a 250 mpbs bandwidth
+on a macbook m1 8 core cpu
 a chunk size of `4000000` works the best as it completes it within ~90 seconds
 
 ## Caveats

From f8e532ef280b1cf9621a21b3f36f6d7b219759eb Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 15:48:36 +0100
Subject: [PATCH 19/36] implemented async fetch

---
 s3_md5/src/consumer.py |  23 ++++++++++
 s3_md5/src/s3_md5.py   | 100 ++++++++++++-----------------------------
 s3_md5/src/signals.py  |   4 --
 setup.py               |   1 +
 4 files changed, 52 insertions(+), 76 deletions(-)
 create mode 100644 s3_md5/src/consumer.py
 delete mode 100644 s3_md5/src/signals.py

diff --git a/s3_md5/src/consumer.py b/s3_md5/src/consumer.py
new file mode 100644
index 0000000..60d03a8
--- /dev/null
+++ b/s3_md5/src/consumer.py
@@ -0,0 +1,23 @@
+from hashlib import md5
+from multiprocessing.managers import DictProxy, ValueProxy
+
+from .logger import logger
+
+
+def consumer(store: DictProxy, variable: ValueProxy[str], chunk_count: int):
+    '''a process that subscribes to the queue and processes md5'''
+    hasher = md5()
+    logger.info("consumer started")
+    element_to_consume = 0
+    while element_to_consume < chunk_count:
+        potential_item = store.get(element_to_consume)
+        if potential_item is not None:
+            hasher.update(potential_item)
+            logger.info(
+                f"consumed chunk {element_to_consume}")
+            logger.info(
+                f"remaining chunk {chunk_count - (element_to_consume + 1)}")
+            element_to_consume += 1
+    logger.info("calculating md5 hash")
+    md5_hash = hasher.hexdigest()
+    variable.value = md5_hash
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index f6a12f1..7a47d5c 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -1,63 +1,17 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
 from concurrent.futures import ThreadPoolExecutor
-from hashlib import md5
-from multiprocessing import Manager, Process, Queue
-from multiprocessing.managers import ValueProxy
+from multiprocessing import Manager, Process
+from multiprocessing.managers import DictProxy, ValueProxy
+from typing import Any
 
 from mypy_boto3_s3 import S3Client
+from setproctitle import setproctitle
 
+from .consumer import consumer
 from .logger import logger
 from .s3_file import S3FileHelper
-from .signals import CANCEL, COMPLETE
 
-
-def consumer(queue: Queue, variable: ValueProxy[str]):
-    '''a process that subscribes to the queue and processes md5'''
-    hasher = md5()
-    while True:
-        try:
-            item = queue.get()
-            if item == COMPLETE:
-                break
-            if item == CANCEL:
-                raise ValueError("one of the processes failed")
-            logger.info("consuming range")
-            hasher.update(item)
-        except Exception as exception:  # pylint: disable=broad-except
-            logger.error(exception)
-            return
-    md5_hash = hasher.hexdigest()
-    logger.info(md5_hash)
-    variable.value = md5_hash
-
-
-def fetch_block(block_number: int,
-                block_count: int,
-                chunk_count: int,
-                chunk_size: int,
-                chunk_count_per_block: int,
-                workers: int,
-                s3_file: S3FileHelper):
-    '''runs individual blocks into a separate process'''
-    logger.info(f"processing block {block_number}")
-    start_part_number = block_number * chunk_count_per_block
-    end_part_number = chunk_count - 1 if block_number == block_count - \
-        1 else (start_part_number + chunk_count_per_block) - 1
-    logger.info(f"part number {start_part_number}-{end_part_number}")
-
-    with ThreadPoolExecutor(max_workers=workers) as thread_executor:
-        def wrapper(part_number: int):
-            ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
-                part_number, chunk_size, chunk_count)
-            logger.info(f"downloading {ranged_bytes_string}")
-            ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
-            logger.info(f"downloaded {ranged_bytes_string}")
-            return ranged_bytes
-
-        results = thread_executor.map(wrapper, range(
-            start_part_number, end_part_number + 1))
-
-        return results
+setproctitle('s3-md5: main process')
 
 
 def parse_file_md5(s3_client: S3Client,
@@ -81,28 +35,30 @@ def parse_file_md5(s3_client: S3Client,
     if chunk_count < workers:
         raise AssertionError('chunk count cannot be smaller than workers')
 
-    block_count = chunk_count // workers
-    logger.info(f'block count {block_count}')
-
-    chunk_count_per_block = chunk_count // block_count
-    logger.info(f'chunk to get per block {chunk_count_per_block}')
-
-    queue = Queue()
-    variable = Manager().Value(str, '')
+    md5_store = Manager().Value(str, '')
+    byte_store = Manager().dict()
 
-    consumer_process = Process(target=consumer, args=(queue, variable))
+    consumer_process = Process(target=consumer, args=(
+        byte_store, md5_store, chunk_count), name="s3-md5: sub process")
     consumer_process.start()
+    chunk_count = file_size // chunk_size
 
-    for block_number in range(0, block_count):
-        try:
-            for item in fetch_block(block_number, block_count, chunk_count,
-                                    chunk_size, chunk_count_per_block, workers // 2, s3_file):
-                queue.put(item)
-        except Exception as exception:  # pylint: disable=broad-except
-            logger.error(exception)
-            queue.put(CANCEL)
-            raise ValueError from exception
+    with ThreadPoolExecutor(max_workers=workers) as thread_executor:
+        def wrapper(part_number: int):
+            ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
+                part_number, chunk_size, chunk_count)
+            logger.info(f"downloading {ranged_bytes_string}")
+            ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
+            logger.info(f"downloaded {ranged_bytes_string}")
+            byte_store[part_number] = ranged_bytes
+        for part_number in range(chunk_count):
+            try:
+                thread_executor.submit(wrapper, part_number)
+            except Exception as exception:
+                logger.error(f"parse_file_md5 {exception}")
+                consumer_process.kill()
+                raise exception
+        thread_executor.shutdown()
 
-    queue.put(COMPLETE)
     consumer_process.join()
-    return variable.value
+    return md5_store.value
diff --git a/s3_md5/src/signals.py b/s3_md5/src/signals.py
deleted file mode 100644
index 6c9ed98..0000000
--- a/s3_md5/src/signals.py
+++ /dev/null
@@ -1,4 +0,0 @@
-'''signal values'''
-
-CANCEL = "CANCEL"
-COMPLETE = "COMPLETE"
diff --git a/setup.py b/setup.py
index 504ab79..ec50f54 100644
--- a/setup.py
+++ b/setup.py
@@ -14,6 +14,7 @@
     install_requires=[
         "boto3==1.26.41",
         "boto3-stubs[s3]",
+        "setproctitle==1.3.3"
     ],
     extras_require={
         "develop": [

From 41fb8df34ad2eb92c93e9ba561245cf07a004ed6 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 16:23:47 +0100
Subject: [PATCH 20/36] adding a progress bar

---
 s3_md5/cmd.py          |  1 +
 s3_md5/src/consumer.py | 28 +++++++++++++++-------------
 s3_md5/src/logger.py   | 10 +++++++++-
 s3_md5/src/s3_md5.py   | 16 +++++++++-------
 setup.py               |  3 ++-
 5 files changed, 36 insertions(+), 22 deletions(-)

diff --git a/s3_md5/cmd.py b/s3_md5/cmd.py
index f5e273b..41bb09b 100644
--- a/s3_md5/cmd.py
+++ b/s3_md5/cmd.py
@@ -9,6 +9,7 @@
 
 
 def run():
+    '''runs the script'''
     start_time = perf_counter()
     args = parse_args()
     main_s3_client = client('s3')
diff --git a/s3_md5/src/consumer.py b/s3_md5/src/consumer.py
index 60d03a8..19cf855 100644
--- a/s3_md5/src/consumer.py
+++ b/s3_md5/src/consumer.py
@@ -1,23 +1,25 @@
 from hashlib import md5
 from multiprocessing.managers import DictProxy, ValueProxy
 
+from tqdm import tqdm
+
 from .logger import logger
 
 
 def consumer(store: DictProxy, variable: ValueProxy[str], chunk_count: int):
     '''a process that subscribes to the queue and processes md5'''
     hasher = md5()
-    logger.info("consumer started")
+    logger.debug("consumer started")
     element_to_consume = 0
-    while element_to_consume < chunk_count:
-        potential_item = store.get(element_to_consume)
-        if potential_item is not None:
-            hasher.update(potential_item)
-            logger.info(
-                f"consumed chunk {element_to_consume}")
-            logger.info(
-                f"remaining chunk {chunk_count - (element_to_consume + 1)}")
-            element_to_consume += 1
-    logger.info("calculating md5 hash")
-    md5_hash = hasher.hexdigest()
-    variable.value = md5_hash
+    with tqdm(total=chunk_count) as progress_bar:
+        while element_to_consume < chunk_count:
+            potential_item = store.get(element_to_consume)
+            if potential_item is not None:
+                hasher.update(potential_item)
+                logger.debug(
+                    f"consumed chunk {element_to_consume} left {chunk_count - element_to_consume + 1}")
+                element_to_consume += 1
+                progress_bar.update(1)
+        logger.debug("calculating md5 hash")
+        md5_hash = hasher.hexdigest()
+        variable.value = md5_hash
diff --git a/s3_md5/src/logger.py b/s3_md5/src/logger.py
index 0d5741c..b2c3106 100644
--- a/s3_md5/src/logger.py
+++ b/s3_md5/src/logger.py
@@ -1,9 +1,17 @@
 '''creates a logger'''
 import logging
+import os
 import sys
 
+levels = {
+    'CRITICAL': logging.CRITICAL,
+    'ERROR': logging.ERROR,
+    'WARNING': logging.WARNING,
+    'INFO': logging.INFO,
+    'DEBUG': logging.DEBUG
+}
 logger = logging.getLogger(__name__)
-logger.setLevel(logging.INFO)
+logger.setLevel(levels.get(os.getenv('LOG_LEVEL', None), logging.INFO))
 
 stream_handler = logging.StreamHandler(sys.stdout)
 formatter = logging.Formatter(
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 7a47d5c..d13f16a 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -1,4 +1,5 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
+import sys
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import Manager, Process
 from multiprocessing.managers import DictProxy, ValueProxy
@@ -25,12 +26,12 @@ def parse_file_md5(s3_client: S3Client,
     file_size = s3_file.get_file_size()
     if file_size < chunk_size:
         raise AssertionError('file size cannot be smaller than chunk size')
-    logger.info(f'file size {file_size} bytes')
-    logger.info(f'chunk size {chunk_size} bytes')
-    logger.info(f'workers {workers}')
+    logger.debug(f'file size {file_size} bytes')
+    logger.debug(f'chunk size {chunk_size} bytes')
+    logger.debug(f'workers {workers}')
 
     chunk_count = file_size // chunk_size
-    logger.info(f'chunk count {chunk_count}')
+    logger.debug(f'chunk count {chunk_count}')
 
     if chunk_count < workers:
         raise AssertionError('chunk count cannot be smaller than workers')
@@ -47,9 +48,9 @@ def parse_file_md5(s3_client: S3Client,
         def wrapper(part_number: int):
             ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
                 part_number, chunk_size, chunk_count)
-            logger.info(f"downloading {ranged_bytes_string}")
+            logger.debug(f"downloading {ranged_bytes_string}")
             ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
-            logger.info(f"downloaded {ranged_bytes_string}")
+            logger.debug(f"downloaded {ranged_bytes_string}")
             byte_store[part_number] = ranged_bytes
         for part_number in range(chunk_count):
             try:
@@ -57,7 +58,8 @@ def wrapper(part_number: int):
             except Exception as exception:
                 logger.error(f"parse_file_md5 {exception}")
                 consumer_process.kill()
-                raise exception
+                thread_executor.shutdown(wait=False, cancel_futures=True)
+                sys.exit(1)
         thread_executor.shutdown()
 
     consumer_process.join()
diff --git a/setup.py b/setup.py
index ec50f54..076fb45 100644
--- a/setup.py
+++ b/setup.py
@@ -14,7 +14,8 @@
     install_requires=[
         "boto3==1.26.41",
         "boto3-stubs[s3]",
-        "setproctitle==1.3.3"
+        "setproctitle==1.3.3",
+        "tqdm==4.66.2"
     ],
     extras_require={
         "develop": [

From 57ce9e96b1c00e3225d102107e122c2637afba94 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 21:30:10 +0100
Subject: [PATCH 21/36] remove unused modules

---
 s3_md5/src/s3_md5.py | 3 +--
 1 file changed, 1 insertion(+), 2 deletions(-)

diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index d13f16a..3e0b47e 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -2,8 +2,6 @@
 import sys
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import Manager, Process
-from multiprocessing.managers import DictProxy, ValueProxy
-from typing import Any
 
 from mypy_boto3_s3 import S3Client
 from setproctitle import setproctitle
@@ -55,6 +53,7 @@ def wrapper(part_number: int):
         for part_number in range(chunk_count):
             try:
                 thread_executor.submit(wrapper, part_number)
+            # pylint: disable=broad-exception-caught
             except Exception as exception:
                 logger.error(f"parse_file_md5 {exception}")
                 consumer_process.kill()

From f50b75bfa0197f70224c7c63571c0ca67da8c570 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 22:07:36 +0100
Subject: [PATCH 22/36] readme update

---
 README.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/README.md b/README.md
index a004fa8..aba8039 100644
--- a/README.md
+++ b/README.md
@@ -1,6 +1,6 @@
 # s3-md5
 
-Get fast md5 hashes for an s3 file.
+Get fast md5 hashes for an s3 file. This works by utilizing a process to fetch chunks into by doing them in threads while having another process consuming the shared dictionary. Given MD5 hashes need to be processed sequentially, it keeps looking for the expected chunk to ensure the order is right.
 
 ## Requirements
 
@@ -50,7 +50,7 @@ There are two _optional_ arguments that you may want to provide
 for a file size of `1048576000` bytes
 on a 250 mpbs bandwidth
 on a macbook m1 8 core cpu
-a chunk size of `4000000` works the best as it completes it within ~90 seconds
+a chunk size of `4000000` works the best as it completes it within ~100 seconds
 
 ## Caveats
 

From 14a6a6e3e2fdf1a966d7242235d0e90607b8c1cb Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Sun, 31 Mar 2024 23:00:31 +0100
Subject: [PATCH 23/36] automatic chunk selection

---
 README.md            |  6 +-----
 s3_md5/src/cli.py    | 16 +++++++++++++---
 s3_md5/src/s3_md5.py | 10 +++++-----
 setup.py             |  3 ++-
 4 files changed, 21 insertions(+), 14 deletions(-)

diff --git a/README.md b/README.md
index aba8039..8333e4a 100644
--- a/README.md
+++ b/README.md
@@ -43,7 +43,7 @@ python s3_md5/main.py <bucket_name> <file_name>
 There are two _optional_ arguments that you may want to provide
 
 -   `-w` or workers sets the number of python threads to use for downloading purposes, by default its set to the following equation `number of cpu cores * 2 - 1`
--   `-c` or chunk size in **bytes** sets the individual download size on each get request sent to s3, by default its set to `1000000`. For optimal performance you need to saturate the available bandwidth on your machine. Try running a few sample file to work out what chunk size would work best.
+-   `-c` or chunk size in **bytes** sets the individual download size on each get request sent to s3, by default it will use [speedtest-cli](https://pypi.org/project/speedtest-cli/) to determine the network speed.
 
 ### Example
 
@@ -51,7 +51,3 @@ for a file size of `1048576000` bytes
 on a 250 mpbs bandwidth
 on a macbook m1 8 core cpu
 a chunk size of `4000000` works the best as it completes it within ~100 seconds
-
-## Caveats
-
--   File size can not be smaller than the default chunk size of `1000000`, if yes, then the chunk size must be manually provided or it will raise an assertion error.
diff --git a/s3_md5/src/cli.py b/s3_md5/src/cli.py
index c240a92..e4ccadd 100644
--- a/s3_md5/src/cli.py
+++ b/s3_md5/src/cli.py
@@ -2,11 +2,15 @@
 from argparse import ArgumentParser
 from multiprocessing import cpu_count
 
+from speedtest import Speedtest
+
+from .logger import logger
+
 
 def parse_args():
     '''parses command line arguments'''
     DEFAULT_WORKERS = cpu_count() * 2 - 1
-    DEFAULT_CHUNK_SIZE = 1000000
+    BIT_IN_BYTE = 0.125
 
     parser = ArgumentParser(description='parse md5 of an s3 object')
     parser.add_argument('bucket',
@@ -19,6 +23,12 @@ def parse_args():
                         default=DEFAULT_WORKERS,
                         help='number of cpu threads to use for downloading')
     parser.add_argument('-c', '--chunk_size', type=int,
-                        default=DEFAULT_CHUNK_SIZE,
+                        default=None,
                         help='chunk size to download on each request')
-    return parser.parse_args()
+    parsed_args = parser.parse_args()
+    if parsed_args.chunk_size is None:
+        logger.info("picking chunk size")
+        speed_test = Speedtest()
+        parsed_args.chunk_size = int(
+            speed_test.download() * BIT_IN_BYTE) // parsed_args.workers
+    return parsed_args
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 3e0b47e..9b6881c 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -22,17 +22,17 @@ def parse_file_md5(s3_client: S3Client,
     s3_file = S3FileHelper(s3_client, bucket, file_name)
 
     file_size = s3_file.get_file_size()
+    logger.info(f'file size {file_size} bytes')
     if file_size < chunk_size:
-        raise AssertionError('file size cannot be smaller than chunk size')
-    logger.debug(f'file size {file_size} bytes')
-    logger.debug(f'chunk size {chunk_size} bytes')
-    logger.debug(f'workers {workers}')
+        chunk_size = file_size
+    logger.info(f'chunk size {chunk_size} bytes')
 
     chunk_count = file_size // chunk_size
     logger.debug(f'chunk count {chunk_count}')
 
     if chunk_count < workers:
-        raise AssertionError('chunk count cannot be smaller than workers')
+        workers = chunk_count
+    logger.info(f'workers {workers}')
 
     md5_store = Manager().Value(str, '')
     byte_store = Manager().dict()
diff --git a/setup.py b/setup.py
index 076fb45..7bc72ff 100644
--- a/setup.py
+++ b/setup.py
@@ -15,7 +15,8 @@
         "boto3==1.26.41",
         "boto3-stubs[s3]",
         "setproctitle==1.3.3",
-        "tqdm==4.66.2"
+        "tqdm==4.66.2",
+        "speedtest-cli==2.1.3"
     ],
     extras_require={
         "develop": [

From 52a539fcd1f890437bc48096f6d809cde707d8b4 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Mon, 1 Apr 2024 15:15:24 +0100
Subject: [PATCH 24/36] clean up

---
 s3_md5/cmd.py          |  4 ++--
 s3_md5/src/cli.py      | 28 +++++++++++++++++++++-------
 s3_md5/src/consumer.py | 28 +++++++++++++++++++---------
 s3_md5/src/s3_file.py  |  2 +-
 s3_md5/src/s3_md5.py   | 10 +++++-----
 5 files changed, 48 insertions(+), 24 deletions(-)

diff --git a/s3_md5/cmd.py b/s3_md5/cmd.py
index 41bb09b..c2344a6 100644
--- a/s3_md5/cmd.py
+++ b/s3_md5/cmd.py
@@ -20,8 +20,8 @@ def run():
         args.chunk_size,
         args.workers
     )
-    logger.info(f'md5 hash {md5_hash}')
-    logger.info(f'took {perf_counter() - start_time} seconds')
+    logger.info(f"md5 hash {md5_hash}")
+    logger.info(f"took {perf_counter() - start_time} seconds")
 
 
 if __name__ == "__main__":
diff --git a/s3_md5/src/cli.py b/s3_md5/src/cli.py
index e4ccadd..91fd4a3 100644
--- a/s3_md5/src/cli.py
+++ b/s3_md5/src/cli.py
@@ -6,12 +6,29 @@
 
 from .logger import logger
 
+DEFAULT_WORKERS = cpu_count() * 2 - 1
+BIT_IN_BYTE = 0.125
+DEFAULT_CHUNK_SIZE = 1000000
+
+
+def get_download_speed(workers: int):
+    '''uses speed test to get download speed'''
+    logger.info("picking chunk size")
+    try:
+        speed_test = Speedtest()
+        download_speed = speed_test.download()
+        chunk_size = int(download_speed * BIT_IN_BYTE) // workers
+        return chunk_size
+    # pylint: disable=broad-exception-caught
+    except Exception as exception:
+        logger.debug(f"get_download_speed {exception}")
+        logger.warning(
+            "will use default chunk size as automatic chunk size calculation failed")
+        return DEFAULT_CHUNK_SIZE
+
 
 def parse_args():
     '''parses command line arguments'''
-    DEFAULT_WORKERS = cpu_count() * 2 - 1
-    BIT_IN_BYTE = 0.125
-
     parser = ArgumentParser(description='parse md5 of an s3 object')
     parser.add_argument('bucket',
                         type=str,
@@ -27,8 +44,5 @@ def parse_args():
                         help='chunk size to download on each request')
     parsed_args = parser.parse_args()
     if parsed_args.chunk_size is None:
-        logger.info("picking chunk size")
-        speed_test = Speedtest()
-        parsed_args.chunk_size = int(
-            speed_test.download() * BIT_IN_BYTE) // parsed_args.workers
+        parsed_args.chunk_size = get_download_speed(parsed_args.workers)
     return parsed_args
diff --git a/s3_md5/src/consumer.py b/s3_md5/src/consumer.py
index 19cf855..48a51db 100644
--- a/s3_md5/src/consumer.py
+++ b/s3_md5/src/consumer.py
@@ -1,25 +1,35 @@
+import sys
 from hashlib import md5
-from multiprocessing.managers import DictProxy, ValueProxy
+from multiprocessing.managers import ValueProxy
+from typing import Dict
 
 from tqdm import tqdm
 
 from .logger import logger
 
 
-def consumer(store: DictProxy, variable: ValueProxy[str], chunk_count: int):
+def consumer(store: Dict[int, bytes], variable: ValueProxy[str], chunk_count: int):
     '''a process that subscribes to the queue and processes md5'''
     hasher = md5()
     logger.debug("consumer started")
     element_to_consume = 0
     with tqdm(total=chunk_count) as progress_bar:
         while element_to_consume < chunk_count:
-            potential_item = store.get(element_to_consume)
-            if potential_item is not None:
-                hasher.update(potential_item)
-                logger.debug(
-                    f"consumed chunk {element_to_consume} left {chunk_count - element_to_consume + 1}")
-                element_to_consume += 1
-                progress_bar.update(1)
+            try:
+                potential_item = store.get(element_to_consume)
+                if potential_item is not None:
+                    hasher.update(potential_item)
+                    logger.debug(
+                        f"consumed chunk {element_to_consume}"
+                        + " " +
+                        f"left {chunk_count - element_to_consume + 1}")
+                    element_to_consume += 1
+                    progress_bar.update(1)
+            # pylint: disable=broad-exception-caught
+            except Exception as exception:
+                logger.error(f"consumer {exception}")
+                sys.exit(1)
+
         logger.debug("calculating md5 hash")
         md5_hash = hasher.hexdigest()
         variable.value = md5_hash
diff --git a/s3_md5/src/s3_file.py b/s3_md5/src/s3_file.py
index d38fae8..2fcaf19 100644
--- a/s3_md5/src/s3_file.py
+++ b/s3_md5/src/s3_file.py
@@ -33,7 +33,7 @@ def calculate_range_bytes_from_part_number(self, part_number: int,
 
         end_bytes: int = self.__file_size if part_number + \
             1 == file_chunk_count else (((part_number * chunk_size) + chunk_size) - 1)
-        return f'bytes={start_bytes}-{end_bytes}'
+        return f"bytes={start_bytes}-{end_bytes}"
 
     def get_range_bytes(self, range_string: str) -> bytes:
         '''fetches the range bytes requested from s3'''
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 9b6881c..b446b92 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -22,17 +22,17 @@ def parse_file_md5(s3_client: S3Client,
     s3_file = S3FileHelper(s3_client, bucket, file_name)
 
     file_size = s3_file.get_file_size()
-    logger.info(f'file size {file_size} bytes')
+    logger.info(f"file size {file_size} bytes")
     if file_size < chunk_size:
         chunk_size = file_size
-    logger.info(f'chunk size {chunk_size} bytes')
+    logger.info(f"chunk size {chunk_size} bytes")
 
     chunk_count = file_size // chunk_size
-    logger.debug(f'chunk count {chunk_count}')
+    logger.debug(f"chunk count {chunk_count}")
 
     if chunk_count < workers:
         workers = chunk_count
-    logger.info(f'workers {workers}')
+    logger.info(f"workers {workers}")
 
     md5_store = Manager().Value(str, '')
     byte_store = Manager().dict()
@@ -56,8 +56,8 @@ def wrapper(part_number: int):
             # pylint: disable=broad-exception-caught
             except Exception as exception:
                 logger.error(f"parse_file_md5 {exception}")
-                consumer_process.kill()
                 thread_executor.shutdown(wait=False, cancel_futures=True)
+                consumer_process.terminate()
                 sys.exit(1)
         thread_executor.shutdown()
 

From 76846ac031c1f1ceef860a83fe5ac130ffe08879 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Mon, 1 Apr 2024 17:56:47 +0100
Subject: [PATCH 25/36] added better death handling

---
 s3_md5/src/s3_md5.py | 16 +++++++++++++++-
 1 file changed, 15 insertions(+), 1 deletion(-)

diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index b446b92..580f13d 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -2,6 +2,8 @@
 import sys
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import Manager, Process
+from signal import SIGCHLD, signal
+from typing import Any
 
 from mypy_boto3_s3 import S3Client
 from setproctitle import setproctitle
@@ -13,6 +15,16 @@
 setproctitle('s3-md5: main process')
 
 
+def consumer_death_strategy(signal_number: int, stack: Any, process: Process, thread_executor: ThreadPoolExecutor):
+    '''handler to call when consumer process dies'''
+    logger.error(f"consumer died with {signal_number}")
+    logger.error(f"consumer stack {stack}")
+    logger.warning("will exit")
+    process.terminate()
+    thread_executor.shutdown(wait=False, cancel_futures=True)
+    sys.exit(1)
+
+
 def parse_file_md5(s3_client: S3Client,
                    bucket: str,
                    file_name: str,
@@ -40,9 +52,11 @@ def parse_file_md5(s3_client: S3Client,
     consumer_process = Process(target=consumer, args=(
         byte_store, md5_store, chunk_count), name="s3-md5: sub process")
     consumer_process.start()
-    chunk_count = file_size // chunk_size
 
     with ThreadPoolExecutor(max_workers=workers) as thread_executor:
+        signal(SIGCHLD, lambda signal_number, stack: consumer_death_strategy(
+            signal_number, stack, consumer_process, thread_executor))
+
         def wrapper(part_number: int):
             ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
                 part_number, chunk_size, chunk_count)

From bba8af99a93e5e85b7a92f9f36e1e18cbd51c7b7 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Tue, 2 Apr 2024 11:37:09 +0100
Subject: [PATCH 26/36] added erorr logging for investigation

---
 s3_md5/src/consumer.py |  1 +
 s3_md5/src/logger.py   | 11 ++++++-----
 s3_md5/src/s3_md5.py   | 26 +++++++++++++++++---------
 3 files changed, 24 insertions(+), 14 deletions(-)

diff --git a/s3_md5/src/consumer.py b/s3_md5/src/consumer.py
index 48a51db..43242c8 100644
--- a/s3_md5/src/consumer.py
+++ b/s3_md5/src/consumer.py
@@ -33,3 +33,4 @@ def consumer(store: Dict[int, bytes], variable: ValueProxy[str], chunk_count: in
         logger.debug("calculating md5 hash")
         md5_hash = hasher.hexdigest()
         variable.value = md5_hash
+        sys.exit()
diff --git a/s3_md5/src/logger.py b/s3_md5/src/logger.py
index b2c3106..12062e8 100644
--- a/s3_md5/src/logger.py
+++ b/s3_md5/src/logger.py
@@ -3,15 +3,16 @@
 import os
 import sys
 
-levels = {
+LOG_LEVELS = {
     'CRITICAL': logging.CRITICAL,
-    'ERROR': logging.ERROR,
     'WARNING': logging.WARNING,
-    'INFO': logging.INFO,
-    'DEBUG': logging.DEBUG
+    'ERROR': logging.ERROR,
+    'DEBUG': logging.DEBUG,
+    'INFO': logging.INFO
 }
 logger = logging.getLogger(__name__)
-logger.setLevel(levels.get(os.getenv('LOG_LEVEL', None), logging.INFO))
+log_level = LOG_LEVELS[os.getenv('LOG_LEVEL', 'INFO')]
+logger.setLevel(log_level)
 
 stream_handler = logging.StreamHandler(sys.stdout)
 formatter = logging.Formatter(
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 580f13d..72553a2 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -12,17 +12,23 @@
 from .logger import logger
 from .s3_file import S3FileHelper
 
-setproctitle('s3-md5: main process')
+setproctitle('s3-md5')
 
 
-def consumer_death_strategy(signal_number: int, stack: Any, process: Process, thread_executor: ThreadPoolExecutor):
+def consumer_death_strategy(signal_number: int,
+                            stack: Any,
+                            process: Process,
+                            thread_executor: ThreadPoolExecutor):
     '''handler to call when consumer process dies'''
-    logger.error(f"consumer died with {signal_number}")
-    logger.error(f"consumer stack {stack}")
-    logger.warning("will exit")
-    process.terminate()
-    thread_executor.shutdown(wait=False, cancel_futures=True)
-    sys.exit(1)
+    if process.exitcode != 0:
+        logger.error(
+            f"consumer died with signal number {signal_number} exit code {process.exitcode}")
+        logger.error(f"consumer stack {stack}")
+        logger.warning("will exit")
+        process.terminate()
+        thread_executor.shutdown(wait=False, cancel_futures=True)
+        sys.exit(1)
+    logger.debug("consumer process finished")
 
 
 def parse_file_md5(s3_client: S3Client,
@@ -50,7 +56,7 @@ def parse_file_md5(s3_client: S3Client,
     byte_store = Manager().dict()
 
     consumer_process = Process(target=consumer, args=(
-        byte_store, md5_store, chunk_count), name="s3-md5: sub process")
+        byte_store, md5_store, chunk_count))
     consumer_process.start()
 
     with ThreadPoolExecutor(max_workers=workers) as thread_executor:
@@ -64,6 +70,7 @@ def wrapper(part_number: int):
             ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
             logger.debug(f"downloaded {ranged_bytes_string}")
             byte_store[part_number] = ranged_bytes
+
         for part_number in range(chunk_count):
             try:
                 thread_executor.submit(wrapper, part_number)
@@ -73,6 +80,7 @@ def wrapper(part_number: int):
                 thread_executor.shutdown(wait=False, cancel_futures=True)
                 consumer_process.terminate()
                 sys.exit(1)
+
         thread_executor.shutdown()
 
     consumer_process.join()

From a2c1186d19120ea0a77a4d38a70e3e66c2f89881 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Tue, 2 Apr 2024 12:06:56 +0100
Subject: [PATCH 27/36] async enhancement

---
 s3_md5/src/s3_md5.py | 43 ++++++++++++++++++++-----------------------
 1 file changed, 20 insertions(+), 23 deletions(-)

diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 72553a2..92bebe0 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -1,4 +1,5 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
+import asyncio
 import sys
 from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import Manager, Process
@@ -59,29 +60,25 @@ def parse_file_md5(s3_client: S3Client,
         byte_store, md5_store, chunk_count))
     consumer_process.start()
 
-    with ThreadPoolExecutor(max_workers=workers) as thread_executor:
-        signal(SIGCHLD, lambda signal_number, stack: consumer_death_strategy(
-            signal_number, stack, consumer_process, thread_executor))
-
-        def wrapper(part_number: int):
-            ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
-                part_number, chunk_size, chunk_count)
-            logger.debug(f"downloading {ranged_bytes_string}")
-            ranged_bytes = s3_file.get_range_bytes(ranged_bytes_string)
-            logger.debug(f"downloaded {ranged_bytes_string}")
-            byte_store[part_number] = ranged_bytes
-
-        for part_number in range(chunk_count):
-            try:
-                thread_executor.submit(wrapper, part_number)
-            # pylint: disable=broad-exception-caught
-            except Exception as exception:
-                logger.error(f"parse_file_md5 {exception}")
-                thread_executor.shutdown(wait=False, cancel_futures=True)
-                consumer_process.terminate()
-                sys.exit(1)
-
-        thread_executor.shutdown()
+    signal(SIGCHLD, lambda signal_number, stack: consumer_death_strategy(
+        signal_number, stack, consumer_process, None))
+
+    async def wrapper(part_number: int):
+        ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
+            part_number, chunk_size, chunk_count)
+        logger.debug(f"downloading {ranged_bytes_string}")
+        ranged_bytes = await asyncio.to_thread(s3_file.get_range_bytes, ranged_bytes_string)
+        logger.debug(f"downloaded {ranged_bytes_string}")
+        byte_store[part_number] = ranged_bytes
+
+    loop = asyncio.get_event_loop()
+    try:
+        loop.run_until_complete(asyncio.gather(
+            *[wrapper(part_number) for part_number in range(chunk_count)]))
+    except Exception as exception:
+        logger.error(f"parse_file_md5 {exception}")
+        consumer_process.terminate()
+        sys.exit(1)
 
     consumer_process.join()
     return md5_store.value

From 2c479477caabc0f617b47336c9d171630f357067 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Tue, 2 Apr 2024 12:57:44 +0100
Subject: [PATCH 28/36] wip

---
 s3_md5/cmd.py        |  1 -
 s3_md5/src/cli.py    | 11 ++++-------
 s3_md5/src/s3_md5.py | 15 ++++-----------
 3 files changed, 8 insertions(+), 19 deletions(-)

diff --git a/s3_md5/cmd.py b/s3_md5/cmd.py
index c2344a6..41a79df 100644
--- a/s3_md5/cmd.py
+++ b/s3_md5/cmd.py
@@ -18,7 +18,6 @@ def run():
         args.bucket,
         args.file_name,
         args.chunk_size,
-        args.workers
     )
     logger.info(f"md5 hash {md5_hash}")
     logger.info(f"took {perf_counter() - start_time} seconds")
diff --git a/s3_md5/src/cli.py b/s3_md5/src/cli.py
index 91fd4a3..8caebad 100644
--- a/s3_md5/src/cli.py
+++ b/s3_md5/src/cli.py
@@ -11,13 +11,13 @@
 DEFAULT_CHUNK_SIZE = 1000000
 
 
-def get_download_speed(workers: int):
+def get_download_speed():
     '''uses speed test to get download speed'''
     logger.info("picking chunk size")
     try:
         speed_test = Speedtest()
-        download_speed = speed_test.download()
-        chunk_size = int(download_speed * BIT_IN_BYTE) // workers
+        download_speed = speed_test.download(threads=1)
+        chunk_size = int(download_speed * BIT_IN_BYTE)
         return chunk_size
     # pylint: disable=broad-exception-caught
     except Exception as exception:
@@ -36,13 +36,10 @@ def parse_args():
     parser.add_argument('file_name',
                         help='file name',
                         type=str)
-    parser.add_argument('-w', '--workers', type=int,
-                        default=DEFAULT_WORKERS,
-                        help='number of cpu threads to use for downloading')
     parser.add_argument('-c', '--chunk_size', type=int,
                         default=None,
                         help='chunk size to download on each request')
     parsed_args = parser.parse_args()
     if parsed_args.chunk_size is None:
-        parsed_args.chunk_size = get_download_speed(parsed_args.workers)
+        parsed_args.chunk_size = get_download_speed()
     return parsed_args
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 92bebe0..c1c0b46 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -1,7 +1,6 @@
 '''module uses threads to download file from s3 and generates md5 hash'''
 import asyncio
 import sys
-from concurrent.futures import ThreadPoolExecutor
 from multiprocessing import Manager, Process
 from signal import SIGCHLD, signal
 from typing import Any
@@ -18,8 +17,7 @@
 
 def consumer_death_strategy(signal_number: int,
                             stack: Any,
-                            process: Process,
-                            thread_executor: ThreadPoolExecutor):
+                            process: Process):
     '''handler to call when consumer process dies'''
     if process.exitcode != 0:
         logger.error(
@@ -27,7 +25,6 @@ def consumer_death_strategy(signal_number: int,
         logger.error(f"consumer stack {stack}")
         logger.warning("will exit")
         process.terminate()
-        thread_executor.shutdown(wait=False, cancel_futures=True)
         sys.exit(1)
     logger.debug("consumer process finished")
 
@@ -35,8 +32,7 @@ def consumer_death_strategy(signal_number: int,
 def parse_file_md5(s3_client: S3Client,
                    bucket: str,
                    file_name: str,
-                   chunk_size: int,
-                   workers: int) -> str:
+                   chunk_size: int) -> str:
     '''main function to orchestrate the md5 generation of s3 object'''
     s3_file = S3FileHelper(s3_client, bucket, file_name)
 
@@ -49,10 +45,6 @@ def parse_file_md5(s3_client: S3Client,
     chunk_count = file_size // chunk_size
     logger.debug(f"chunk count {chunk_count}")
 
-    if chunk_count < workers:
-        workers = chunk_count
-    logger.info(f"workers {workers}")
-
     md5_store = Manager().Value(str, '')
     byte_store = Manager().dict()
 
@@ -61,7 +53,7 @@ def parse_file_md5(s3_client: S3Client,
     consumer_process.start()
 
     signal(SIGCHLD, lambda signal_number, stack: consumer_death_strategy(
-        signal_number, stack, consumer_process, None))
+        signal_number, stack, consumer_process))
 
     async def wrapper(part_number: int):
         ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
@@ -75,6 +67,7 @@ async def wrapper(part_number: int):
     try:
         loop.run_until_complete(asyncio.gather(
             *[wrapper(part_number) for part_number in range(chunk_count)]))
+    # pylint: disable=broad-exception-caught
     except Exception as exception:
         logger.error(f"parse_file_md5 {exception}")
         consumer_process.terminate()

From 83e536f8750f1173da4934e4358af66a3c12b0e7 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Tue, 2 Apr 2024 18:42:46 +0100
Subject: [PATCH 29/36] wip

---
 s3_md5/cmd.py         | 26 ++++++++++++++------------
 s3_md5/src/s3_file.py | 18 +++++++++++-------
 s3_md5/src/s3_md5.py  | 20 ++++++++++----------
 setup.py              |  3 ++-
 4 files changed, 37 insertions(+), 30 deletions(-)

diff --git a/s3_md5/cmd.py b/s3_md5/cmd.py
index 41a79df..5a2b9b1 100644
--- a/s3_md5/cmd.py
+++ b/s3_md5/cmd.py
@@ -1,27 +1,29 @@
 '''driver'''
+from asyncio import run as asyncio_run
 from time import perf_counter
 
-from boto3 import client
+from aioboto3 import Session
 
 from s3_md5.src.cli import parse_args
 from s3_md5.src.logger import logger
 from s3_md5.src.s3_md5 import parse_file_md5
 
 
-def run():
+async def run():
     '''runs the script'''
     start_time = perf_counter()
     args = parse_args()
-    main_s3_client = client('s3')
-    md5_hash = parse_file_md5(
-        main_s3_client,
-        args.bucket,
-        args.file_name,
-        args.chunk_size,
-    )
-    logger.info(f"md5 hash {md5_hash}")
-    logger.info(f"took {perf_counter() - start_time} seconds")
+    main_s3_session = Session()
+    async with main_s3_session.client('s3') as s3_client:
+        md5_hash = await parse_file_md5(
+            s3_client,
+            args.bucket,
+            args.file_name,
+            args.chunk_size,
+        )
+        logger.info(f"md5 hash {md5_hash}")
+        logger.info(f"took {perf_counter() - start_time} seconds")
 
 
 if __name__ == "__main__":
-    run()
+    asyncio_run(run())
diff --git a/s3_md5/src/s3_file.py b/s3_md5/src/s3_file.py
index 2fcaf19..7c75c2c 100644
--- a/s3_md5/src/s3_file.py
+++ b/s3_md5/src/s3_file.py
@@ -1,4 +1,6 @@
 '''S3 file helper module'''
+from typing import Awaitable
+
 from mypy_boto3_s3 import S3Client
 
 
@@ -11,10 +13,10 @@ def __init__(self, s3_client: S3Client, bucket: str, file_name: str) -> None:
         self.bucket = bucket
         self.file_name = file_name
 
-    def get_file_size(self) -> int:
+    async def get_file_size(self) -> Awaitable[int]:
         '''makes a head object request to get file size in bytes'''
-        s3_object = self.s3_client.head_object(Bucket=self.bucket,
-                                               Key=self.file_name)
+        s3_object = await self.s3_client.head_object(Bucket=self.bucket,
+                                                     Key=self.file_name)
         self.__file_size = s3_object['ContentLength']
         return self.__file_size
 
@@ -35,8 +37,10 @@ def calculate_range_bytes_from_part_number(self, part_number: int,
             1 == file_chunk_count else (((part_number * chunk_size) + chunk_size) - 1)
         return f"bytes={start_bytes}-{end_bytes}"
 
-    def get_range_bytes(self, range_string: str) -> bytes:
+    async def get_range_bytes(self, range_string: str) -> bytes:
         '''fetches the range bytes requested from s3'''
-        return self.s3_client.get_object(Bucket=self.bucket,
-                                         Key=self.file_name,
-                                         Range=range_string)['Body'].read()
+        s3_object = await self.s3_client.get_object(Bucket=self.bucket,
+                                                    Key=self.file_name,
+                                                    Range=range_string)
+        async with s3_object['Body'] as stream:
+            return await stream.read()
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index c1c0b46..7ba1839 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -3,7 +3,7 @@
 import sys
 from multiprocessing import Manager, Process
 from signal import SIGCHLD, signal
-from typing import Any
+from typing import Any, Awaitable
 
 from mypy_boto3_s3 import S3Client
 from setproctitle import setproctitle
@@ -29,14 +29,14 @@ def consumer_death_strategy(signal_number: int,
     logger.debug("consumer process finished")
 
 
-def parse_file_md5(s3_client: S3Client,
-                   bucket: str,
-                   file_name: str,
-                   chunk_size: int) -> str:
+async def parse_file_md5(s3_client: S3Client,
+                         bucket: str,
+                         file_name: str,
+                         chunk_size: int):
     '''main function to orchestrate the md5 generation of s3 object'''
     s3_file = S3FileHelper(s3_client, bucket, file_name)
 
-    file_size = s3_file.get_file_size()
+    file_size = await s3_file.get_file_size()
     logger.info(f"file size {file_size} bytes")
     if file_size < chunk_size:
         chunk_size = file_size
@@ -59,14 +59,14 @@ async def wrapper(part_number: int):
         ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
             part_number, chunk_size, chunk_count)
         logger.debug(f"downloading {ranged_bytes_string}")
-        ranged_bytes = await asyncio.to_thread(s3_file.get_range_bytes, ranged_bytes_string)
+        ranged_bytes = await s3_file.get_range_bytes(ranged_bytes_string)
         logger.debug(f"downloaded {ranged_bytes_string}")
         byte_store[part_number] = ranged_bytes
 
-    loop = asyncio.get_event_loop()
+    tasks = [asyncio.create_task(wrapper(part_number))
+             for part_number in range(chunk_count)]
     try:
-        loop.run_until_complete(asyncio.gather(
-            *[wrapper(part_number) for part_number in range(chunk_count)]))
+        await asyncio.gather(*tasks)
     # pylint: disable=broad-exception-caught
     except Exception as exception:
         logger.error(f"parse_file_md5 {exception}")
diff --git a/setup.py b/setup.py
index 7bc72ff..77bc137 100644
--- a/setup.py
+++ b/setup.py
@@ -16,7 +16,8 @@
         "boto3-stubs[s3]",
         "setproctitle==1.3.3",
         "tqdm==4.66.2",
-        "speedtest-cli==2.1.3"
+        "speedtest-cli==2.1.3",
+        "aioboto3==12.3.0"
     ],
     extras_require={
         "develop": [

From a1a7f2113772483c1fa8ac879652931e31288109 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 19 Apr 2024 11:03:54 +0100
Subject: [PATCH 30/36] added semaphore

---
 README.md            |  2 +-
 s3_md5/cmd.py        |  5 ++++-
 s3_md5/src/cli.py    |  4 ++++
 s3_md5/src/s3_md5.py | 26 ++++++++++++++++----------
 s3_md5/src/utils.py  | 11 +++++++++++
 setup.py             |  1 -
 6 files changed, 36 insertions(+), 13 deletions(-)
 create mode 100644 s3_md5/src/utils.py

diff --git a/README.md b/README.md
index 8333e4a..0df7b26 100644
--- a/README.md
+++ b/README.md
@@ -42,8 +42,8 @@ python s3_md5/main.py <bucket_name> <file_name>
 
 There are two _optional_ arguments that you may want to provide
 
--   `-w` or workers sets the number of python threads to use for downloading purposes, by default its set to the following equation `number of cpu cores * 2 - 1`
 -   `-c` or chunk size in **bytes** sets the individual download size on each get request sent to s3, by default it will use [speedtest-cli](https://pypi.org/project/speedtest-cli/) to determine the network speed.
+-   `-b` or block size to determine the number of maximum concurrent requests sent to s3 to protect against rate limiting. By default it is set to **10**. Please change this as this is related to your aws account s3 api rate limits.
 
 ### Example
 
diff --git a/s3_md5/cmd.py b/s3_md5/cmd.py
index 5a2b9b1..086046e 100644
--- a/s3_md5/cmd.py
+++ b/s3_md5/cmd.py
@@ -7,6 +7,7 @@
 from s3_md5.src.cli import parse_args
 from s3_md5.src.logger import logger
 from s3_md5.src.s3_md5 import parse_file_md5
+from s3_md5.src.utils import seconds_to_minutes
 
 
 async def run():
@@ -20,9 +21,11 @@ async def run():
             args.bucket,
             args.file_name,
             args.chunk_size,
+            args.block_size
         )
         logger.info(f"md5 hash {md5_hash}")
-        logger.info(f"took {perf_counter() - start_time} seconds")
+        logger.info(
+            f"took {seconds_to_minutes(perf_counter() - start_time)} minute(s)")
 
 
 if __name__ == "__main__":
diff --git a/s3_md5/src/cli.py b/s3_md5/src/cli.py
index 8caebad..7205b63 100644
--- a/s3_md5/src/cli.py
+++ b/s3_md5/src/cli.py
@@ -9,6 +9,7 @@
 DEFAULT_WORKERS = cpu_count() * 2 - 1
 BIT_IN_BYTE = 0.125
 DEFAULT_CHUNK_SIZE = 1000000
+DEFAULT_BLOCK_SIZE = 10
 
 
 def get_download_speed():
@@ -39,6 +40,9 @@ def parse_args():
     parser.add_argument('-c', '--chunk_size', type=int,
                         default=None,
                         help='chunk size to download on each request')
+    parser.add_argument('-b', '--block_size', type=int,
+                        default=DEFAULT_BLOCK_SIZE,
+                        help='maximum concurrent request')
     parsed_args = parser.parse_args()
     if parsed_args.chunk_size is None:
         parsed_args.chunk_size = get_download_speed()
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index 7ba1839..a38d220 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -3,7 +3,7 @@
 import sys
 from multiprocessing import Manager, Process
 from signal import SIGCHLD, signal
-from typing import Any, Awaitable
+from typing import Any
 
 from mypy_boto3_s3 import S3Client
 from setproctitle import setproctitle
@@ -11,6 +11,7 @@
 from .consumer import consumer
 from .logger import logger
 from .s3_file import S3FileHelper
+from .utils import bytes_to_mega_bytes
 
 setproctitle('s3-md5')
 
@@ -32,21 +33,25 @@ def consumer_death_strategy(signal_number: int,
 async def parse_file_md5(s3_client: S3Client,
                          bucket: str,
                          file_name: str,
-                         chunk_size: int):
+                         chunk_size: int,
+                         block_size: int):
     '''main function to orchestrate the md5 generation of s3 object'''
     s3_file = S3FileHelper(s3_client, bucket, file_name)
 
     file_size = await s3_file.get_file_size()
-    logger.info(f"file size {file_size} bytes")
+    logger.info(f"file size {bytes_to_mega_bytes(file_size)} megabyte(s)")
     if file_size < chunk_size:
         chunk_size = file_size
-    logger.info(f"chunk size {chunk_size} bytes")
+    logger.info(f"chunk size {bytes_to_mega_bytes(chunk_size)} megabyte(s)")
 
     chunk_count = file_size // chunk_size
     logger.debug(f"chunk count {chunk_count}")
 
+    logger.info(f"block size {block_size}")
+
     md5_store = Manager().Value(str, '')
     byte_store = Manager().dict()
+    semaphore = asyncio.Semaphore(block_size)
 
     consumer_process = Process(target=consumer, args=(
         byte_store, md5_store, chunk_count))
@@ -56,12 +61,13 @@ async def parse_file_md5(s3_client: S3Client,
         signal_number, stack, consumer_process))
 
     async def wrapper(part_number: int):
-        ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
-            part_number, chunk_size, chunk_count)
-        logger.debug(f"downloading {ranged_bytes_string}")
-        ranged_bytes = await s3_file.get_range_bytes(ranged_bytes_string)
-        logger.debug(f"downloaded {ranged_bytes_string}")
-        byte_store[part_number] = ranged_bytes
+        async with semaphore:
+            ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
+                part_number, chunk_size, chunk_count)
+            logger.debug(f"downloading {ranged_bytes_string}")
+            ranged_bytes = await s3_file.get_range_bytes(ranged_bytes_string)
+            logger.debug(f"downloaded {ranged_bytes_string}")
+            byte_store[part_number] = ranged_bytes
 
     tasks = [asyncio.create_task(wrapper(part_number))
              for part_number in range(chunk_count)]
diff --git a/s3_md5/src/utils.py b/s3_md5/src/utils.py
new file mode 100644
index 0000000..19f4288
--- /dev/null
+++ b/s3_md5/src/utils.py
@@ -0,0 +1,11 @@
+'''basic converter utilities'''
+
+
+def bytes_to_mega_bytes(value: int) -> float:
+    '''convert bytes to megabytes'''
+    return value / (1024 * 1024)
+
+
+def seconds_to_minutes(value: float) -> float:
+    '''convert seconds to minutes'''
+    return value / 60
diff --git a/setup.py b/setup.py
index 77bc137..acbfd1d 100644
--- a/setup.py
+++ b/setup.py
@@ -12,7 +12,6 @@
     author_email="16sakib@gmail.com",
     license="MIT",
     install_requires=[
-        "boto3==1.26.41",
         "boto3-stubs[s3]",
         "setproctitle==1.3.3",
         "tqdm==4.66.2",

From a0c159b5b174ec942f520cd8cece520f8bfa13f2 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 19 Apr 2024 11:18:09 +0100
Subject: [PATCH 31/36] fix byte to mb conversion

---
 s3_md5/src/utils.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/s3_md5/src/utils.py b/s3_md5/src/utils.py
index 19f4288..a2ae509 100644
--- a/s3_md5/src/utils.py
+++ b/s3_md5/src/utils.py
@@ -3,7 +3,7 @@
 
 def bytes_to_mega_bytes(value: int) -> float:
     '''convert bytes to megabytes'''
-    return value / (1024 * 1024)
+    return value / (1000 * 1000)
 
 
 def seconds_to_minutes(value: float) -> float:

From f7042401568f9c71e3adf3bc5eeeb3213d7dda88 Mon Sep 17 00:00:00 2001
From: Sakib Alam <47223230+sakibstark11@users.noreply.github.com>
Date: Fri, 2 Aug 2024 15:13:38 +0100
Subject: [PATCH 32/36] delete dictionary on consumption

---
 s3_md5/src/consumer.py | 1 +
 1 file changed, 1 insertion(+)

diff --git a/s3_md5/src/consumer.py b/s3_md5/src/consumer.py
index 43242c8..3895397 100644
--- a/s3_md5/src/consumer.py
+++ b/s3_md5/src/consumer.py
@@ -23,6 +23,7 @@ def consumer(store: Dict[int, bytes], variable: ValueProxy[str], chunk_count: in
                         f"consumed chunk {element_to_consume}"
                         + " " +
                         f"left {chunk_count - element_to_consume + 1}")
+                    del store[element_to_consume]
                     element_to_consume += 1
                     progress_bar.update(1)
             # pylint: disable=broad-exception-caught

From 63dd1f88cdb2653a5a657d0f19a3fc1da1101634 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 2 Aug 2024 15:29:18 +0100
Subject: [PATCH 33/36] readme fix

---
 README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/README.md b/README.md
index 0df7b26..4f123da 100644
--- a/README.md
+++ b/README.md
@@ -35,7 +35,7 @@ s3-md5 <bucket_name> <file_name>
 Or you can directly invoke the script by running
 
 ```sh
-python s3_md5/main.py <bucket_name> <file_name>
+python s3_md5/cmd.py <bucket_name> <file_name>
 ```
 
 ### Arguments

From 4dcbdd245e459f4d0d6ca20eccee2e9eb422b30e Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 2 Aug 2024 19:03:16 +0100
Subject: [PATCH 34/36] more progress bar, semaphore, block downloads

---
 s3_md5/src/consumer.py |  6 +++---
 s3_md5/src/s3_md5.py   | 36 +++++++++++++++++++++---------------
 2 files changed, 24 insertions(+), 18 deletions(-)

diff --git a/s3_md5/src/consumer.py b/s3_md5/src/consumer.py
index 3895397..f81405b 100644
--- a/s3_md5/src/consumer.py
+++ b/s3_md5/src/consumer.py
@@ -13,16 +13,16 @@ def consumer(store: Dict[int, bytes], variable: ValueProxy[str], chunk_count: in
     hasher = md5()
     logger.debug("consumer started")
     element_to_consume = 0
-    with tqdm(total=chunk_count) as progress_bar:
+    with tqdm(total=chunk_count, position=1, desc="consumed") as progress_bar:
         while element_to_consume < chunk_count:
             try:
                 potential_item = store.get(element_to_consume)
                 if potential_item is not None:
                     hasher.update(potential_item)
                     logger.debug(
-                        f"consumed chunk {element_to_consume}"
+                        f"consumed chunk {element_to_consume + 1}"
                         + " " +
-                        f"left {chunk_count - element_to_consume + 1}")
+                        f"left {chunk_count - (element_to_consume + 1)}")
                     del store[element_to_consume]
                     element_to_consume += 1
                     progress_bar.update(1)
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index a38d220..eedeb5e 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -7,6 +7,7 @@
 
 from mypy_boto3_s3 import S3Client
 from setproctitle import setproctitle
+from tqdm import tqdm
 
 from .consumer import consumer
 from .logger import logger
@@ -45,7 +46,7 @@ async def parse_file_md5(s3_client: S3Client,
     logger.info(f"chunk size {bytes_to_mega_bytes(chunk_size)} megabyte(s)")
 
     chunk_count = file_size // chunk_size
-    logger.debug(f"chunk count {chunk_count}")
+    logger.info(f"chunk count {chunk_count}")
 
     logger.info(f"block size {block_size}")
 
@@ -59,25 +60,30 @@ async def parse_file_md5(s3_client: S3Client,
 
     signal(SIGCHLD, lambda signal_number, stack: consumer_death_strategy(
         signal_number, stack, consumer_process))
-
-    async def wrapper(part_number: int):
-        async with semaphore:
+    with tqdm(total=chunk_count, position=0, desc="downloaded") as progress_bar:
+        async def wrapper(part_number: int):
             ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
                 part_number, chunk_size, chunk_count)
-            logger.debug(f"downloading {ranged_bytes_string}")
+            logger.debug(
+                f"downloading {part_number + 1} {ranged_bytes_string}")
             ranged_bytes = await s3_file.get_range_bytes(ranged_bytes_string)
-            logger.debug(f"downloaded {ranged_bytes_string}")
+            logger.debug(f"downloaded {part_number + 1} {ranged_bytes_string}")
+            progress_bar.update(1)
             byte_store[part_number] = ranged_bytes
 
-    tasks = [asyncio.create_task(wrapper(part_number))
-             for part_number in range(chunk_count)]
-    try:
-        await asyncio.gather(*tasks)
-    # pylint: disable=broad-exception-caught
-    except Exception as exception:
-        logger.error(f"parse_file_md5 {exception}")
-        consumer_process.terminate()
-        sys.exit(1)
+        # Process tasks in blocks
+        async with semaphore:
+            for i in range(0, chunk_count, block_size):
+                block_end = min(i + block_size, chunk_count)
+                block_tasks = [wrapper(part_number)
+                               for part_number in range(i, block_end)]
+                try:
+                    await asyncio.gather(*block_tasks)
+                # pylint: disable=broad-exception-caught
+                except Exception as exception:
+                    logger.error(f"parse_file_md5 {exception}")
+                    consumer_process.terminate()
+                    sys.exit(1)
 
     consumer_process.join()
     return md5_store.value

From 1e1a828878cfdfc3fbdbc84451b07f0d7079a92c Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 2 Aug 2024 19:56:43 +0100
Subject: [PATCH 35/36] wipped

---
 .github/workflows/package.yaml |  2 +-
 .github/workflows/test.yaml    |  2 +-
 .tool-versions                 |  2 +-
 s3_md5/cmd.py                  |  6 +++-
 s3_md5/src/s3_md5.py           | 59 ++++++++++++++++------------------
 5 files changed, 35 insertions(+), 36 deletions(-)

diff --git a/.github/workflows/package.yaml b/.github/workflows/package.yaml
index e76cc5a..7e36e54 100644
--- a/.github/workflows/package.yaml
+++ b/.github/workflows/package.yaml
@@ -22,7 +22,7 @@ jobs:
             - name: Setup Python
               uses: actions/setup-python@v4
               with:
-                  python-version: "3.10"
+                  python-version: "3.12"
 
             - name: Create Tag
               id: tag_generator
diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
index a78391a..fff1d3a 100644
--- a/.github/workflows/test.yaml
+++ b/.github/workflows/test.yaml
@@ -11,7 +11,7 @@ jobs:
             - name: Setup Python
               uses: actions/setup-python@v4
               with:
-                  python-version: "3.10"
+                  python-version: "3.12"
 
             - name: Install Dependencies
               run: pip install ".[develop]"
diff --git a/.tool-versions b/.tool-versions
index c4f2bfc..69e5cf7 100644
--- a/.tool-versions
+++ b/.tool-versions
@@ -1 +1 @@
-python 3.10.12
+python 3.12.2
diff --git a/s3_md5/cmd.py b/s3_md5/cmd.py
index 086046e..0f13384 100644
--- a/s3_md5/cmd.py
+++ b/s3_md5/cmd.py
@@ -29,4 +29,8 @@ async def run():
 
 
 if __name__ == "__main__":
-    asyncio_run(run())
+    try:
+        asyncio_run(run())
+    # pylint: disable=broad-exception-caught
+    except Exception as exception:
+        logger.error(f"cmd {exception}")
diff --git a/s3_md5/src/s3_md5.py b/s3_md5/src/s3_md5.py
index eedeb5e..afe3b9a 100644
--- a/s3_md5/src/s3_md5.py
+++ b/s3_md5/src/s3_md5.py
@@ -1,4 +1,3 @@
-'''module uses threads to download file from s3 and generates md5 hash'''
 import asyncio
 import sys
 from multiprocessing import Manager, Process
@@ -20,15 +19,15 @@
 def consumer_death_strategy(signal_number: int,
                             stack: Any,
                             process: Process):
-    '''handler to call when consumer process dies'''
+    '''Handler to call when consumer process dies'''
     if process.exitcode != 0:
         logger.error(
-            f"consumer died with signal number {signal_number} exit code {process.exitcode}")
-        logger.error(f"consumer stack {stack}")
-        logger.warning("will exit")
+            f"Consumer died with signal number {signal_number} exit code {process.exitcode}")
+        logger.error(f"Consumer stack {stack}")
+        logger.warning("Will exit")
         process.terminate()
         sys.exit(1)
-    logger.debug("consumer process finished")
+    logger.debug("Consumer process finished")
 
 
 async def parse_file_md5(s3_client: S3Client,
@@ -36,19 +35,19 @@ async def parse_file_md5(s3_client: S3Client,
                          file_name: str,
                          chunk_size: int,
                          block_size: int):
-    '''main function to orchestrate the md5 generation of s3 object'''
+    '''Main function to orchestrate the MD5 generation of S3 object'''
     s3_file = S3FileHelper(s3_client, bucket, file_name)
 
     file_size = await s3_file.get_file_size()
-    logger.info(f"file size {bytes_to_mega_bytes(file_size)} megabyte(s)")
+    logger.info(f"File size {bytes_to_mega_bytes(file_size)} megabyte(s)")
     if file_size < chunk_size:
         chunk_size = file_size
-    logger.info(f"chunk size {bytes_to_mega_bytes(chunk_size)} megabyte(s)")
+    logger.info(f"Chunk size {bytes_to_mega_bytes(chunk_size)} megabyte(s)")
 
     chunk_count = file_size // chunk_size
-    logger.info(f"chunk count {chunk_count}")
+    logger.info(f"Chunk count {chunk_count}")
 
-    logger.info(f"block size {block_size}")
+    logger.info(f"Block size {block_size}")
 
     md5_store = Manager().Value(str, '')
     byte_store = Manager().dict()
@@ -60,30 +59,26 @@ async def parse_file_md5(s3_client: S3Client,
 
     signal(SIGCHLD, lambda signal_number, stack: consumer_death_strategy(
         signal_number, stack, consumer_process))
+
     with tqdm(total=chunk_count, position=0, desc="downloaded") as progress_bar:
-        async def wrapper(part_number: int):
-            ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
-                part_number, chunk_size, chunk_count)
-            logger.debug(
-                f"downloading {part_number + 1} {ranged_bytes_string}")
-            ranged_bytes = await s3_file.get_range_bytes(ranged_bytes_string)
-            logger.debug(f"downloaded {part_number + 1} {ranged_bytes_string}")
-            progress_bar.update(1)
-            byte_store[part_number] = ranged_bytes
-
-        # Process tasks in blocks
-        async with semaphore:
+        async with asyncio.TaskGroup() as task_group:
+            async def wrapper(part_number: int):
+                ranged_bytes_string = s3_file.calculate_range_bytes_from_part_number(
+                    part_number, chunk_size, chunk_count)
+                async with semaphore:
+                    logger.debug(
+                        f"Downloading {part_number + 1} {ranged_bytes_string}")
+                    ranged_bytes = await s3_file.get_range_bytes(ranged_bytes_string)
+                    logger.debug(
+                        f"Downloaded {part_number + 1} {ranged_bytes_string}")
+                    progress_bar.update(1)
+                    byte_store[part_number] = ranged_bytes
+
+            # Process tasks in blocks
             for i in range(0, chunk_count, block_size):
                 block_end = min(i + block_size, chunk_count)
-                block_tasks = [wrapper(part_number)
-                               for part_number in range(i, block_end)]
-                try:
-                    await asyncio.gather(*block_tasks)
-                # pylint: disable=broad-exception-caught
-                except Exception as exception:
-                    logger.error(f"parse_file_md5 {exception}")
-                    consumer_process.terminate()
-                    sys.exit(1)
+                for part_number in range(i, block_end):
+                    task_group.create_task(wrapper(part_number))
 
     consumer_process.join()
     return md5_store.value

From 0d181a3eab0242ca59442a7c39115a1ce3847501 Mon Sep 17 00:00:00 2001
From: sakibstark11 <16sakib@gmail.com>
Date: Fri, 2 Aug 2024 20:49:43 +0100
Subject: [PATCH 36/36] attempted to fix tests

---
 pytest.ini                                    |  2 ++
 setup.py                                      |  1 +
 test/conftest.py                              | 32 +++++++++++--------
 ..._calculate_range_bytes_from_part_number.py |  6 ++--
 test/test_get_file_size.py                    |  6 ++--
 test/test_get_range_bytes.py                  |  8 +++--
 test/test_parse_file_md5.py                   |  7 ++--
 7 files changed, 39 insertions(+), 23 deletions(-)
 create mode 100644 pytest.ini

diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..4088045
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,2 @@
+[pytest]
+asyncio_mode=auto
diff --git a/setup.py b/setup.py
index acbfd1d..d325893 100644
--- a/setup.py
+++ b/setup.py
@@ -24,6 +24,7 @@
             "moto==4.0.12",
             "pytest==7.2.0",
             "pylint==3.1.0",
+            "pytest-asyncio==0.23.8"
         ],
         "release": ["wheel==0.43.0"]
     },
diff --git a/test/conftest.py b/test/conftest.py
index 2cb6019..6d7c38e 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -1,20 +1,24 @@
-'''common resource for testing'''
-from boto3 import client
+import aioboto3
 from moto import mock_s3
 from pytest import fixture
 
 
 @fixture
-def s3_setup():
+async def s3_setup():
     with mock_s3():
-        s3_client = client('s3')
-        test_bucket = 'bucket'
-        test_file_name = 'key'
-        test_body = '0123456789'
-        s3_client.create_bucket(Bucket=test_bucket, CreateBucketConfiguration={
-            'LocationConstraint': 'ap-east-1',
-        },)
-        s3_client.put_object(Bucket=test_bucket,
-                             Key=test_file_name,
-                             Body=test_body)
-        yield s3_client, test_bucket, test_file_name, test_body
+        async with aioboto3.Session().client('s3', region_name='ap-east-1') as s3_client:
+            test_bucket = 'bucket'
+            test_file_name = 'key'
+            test_body = '0123456789'
+
+            await s3_client.create_bucket(
+                Bucket=test_bucket,
+                CreateBucketConfiguration={'LocationConstraint': 'ap-east-1'}
+            )
+            await s3_client.put_object(
+                Bucket=test_bucket,
+                Key=test_file_name,
+                Body=test_body
+            )
+
+            yield s3_client, test_bucket, test_file_name, test_body
diff --git a/test/test_calculate_range_bytes_from_part_number.py b/test/test_calculate_range_bytes_from_part_number.py
index 16f4c4a..1c198c9 100644
--- a/test/test_calculate_range_bytes_from_part_number.py
+++ b/test/test_calculate_range_bytes_from_part_number.py
@@ -1,14 +1,16 @@
 '''tests range byte calculator function'''
 from typing import Tuple
+
 from mypy_boto3_s3 import S3Client
+
 from s3_md5.src.s3_file import S3FileHelper
 
 
-def test_calculate_range_bytes_from_part_number(s3_setup: Tuple[S3Client, str, str, str]):
+async def test_calculate_range_bytes_from_part_number(s3_setup: Tuple[S3Client, str, str, str]):
     '''test function'''
     s3_client, test_bucket, test_file_name, _ = s3_setup
     s3_file = S3FileHelper(s3_client, test_bucket, test_file_name)
-    s3_file.get_file_size()
+    await s3_file.get_file_size()
     part_number = 1
     chunk_size = 1000000
     file_chunk_count = 10
diff --git a/test/test_get_file_size.py b/test/test_get_file_size.py
index 22d5e77..975f691 100644
--- a/test/test_get_file_size.py
+++ b/test/test_get_file_size.py
@@ -1,12 +1,14 @@
 '''tests file size method'''
 from typing import Tuple
+
 from mypy_boto3_s3 import S3Client
+
 from s3_md5.src.s3_file import S3FileHelper
 
 
-def test_get_file_size(s3_setup: Tuple[S3Client, str, str, str]):
+async def test_get_file_size(s3_setup: Tuple[S3Client, str, str, str]):
     '''test function'''
     s3_client, test_bucket, test_file_name, test_body = s3_setup
     s3_file = S3FileHelper(s3_client, test_bucket, test_file_name)
-    file_size = s3_file.get_file_size()
+    file_size = await s3_file.get_file_size()
     assert file_size == len(test_body)
diff --git a/test/test_get_range_bytes.py b/test/test_get_range_bytes.py
index 6afb26c..cb884ce 100644
--- a/test/test_get_range_bytes.py
+++ b/test/test_get_range_bytes.py
@@ -1,14 +1,16 @@
 '''tests range fetching function'''
 from typing import Tuple
+
 from mypy_boto3_s3 import S3Client
+
 from s3_md5.src.s3_file import S3FileHelper
 
 
-def test_get_range_bytes(s3_setup: Tuple[S3Client, str, str, str]):
+async def test_get_range_bytes(s3_setup: Tuple[S3Client, str, str, str]):
     '''test function'''
     s3_client, test_bucket, test_file_name, test_body = s3_setup
     s3_file = S3FileHelper(s3_client, test_bucket, test_file_name)
-    s3_file.get_file_size()
+    await s3_file.get_file_size()
     range_string = 'bytes=0-0'
-    data = s3_file.get_range_bytes(range_string)
+    data = await s3_file.get_range_bytes(range_string)
     assert data == bytes(test_body[0], 'utf-8')
diff --git a/test/test_parse_file_md5.py b/test/test_parse_file_md5.py
index 72d800c..cba25a8 100644
--- a/test/test_parse_file_md5.py
+++ b/test/test_parse_file_md5.py
@@ -1,12 +1,15 @@
 '''tests the driver function'''
 from hashlib import md5
 from typing import Tuple
+
 from mypy_boto3_s3 import S3Client
+
 from s3_md5.src.s3_md5 import parse_file_md5
 
 
-def test_parse_file_md5(s3_setup: Tuple[S3Client, str, str, str]):
+async def test_parse_file_md5(s3_setup: Tuple[S3Client, str, str, str]):
     '''test function'''
     s3_client, test_bucket, test_file_name, test_body = s3_setup
-    md5_hash = parse_file_md5(s3_client, test_bucket, test_file_name, 1, 2)
+    md5_hash = await parse_file_md5(
+        s3_client, test_bucket, test_file_name, 1, 2)
     assert md5_hash == md5(bytes(test_body, 'utf-8')).hexdigest()