diff --git a/CHANGES.rst b/CHANGES.rst
index cffa856..970d5f6 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -4,6 +4,16 @@
 Changes in Apache Libcloud in development
 -----------------------------------------
 
+Storage
+~~~~~~~
+
+- Optimize ``read_in_chunks()`` function implementation.
+
+  This should result in large performance speedups and lower memory usage when
+  uploading or downloading a large file with a mismatching chunk size.
+  (GITHUB-1847)
+  [Tobias Biester - @Tobi995]
+
 Changes in Apache Libcloud 3.7.0
 --------------------------------
 
diff --git a/libcloud/test/benchmarks/test_read_in_chunks.py b/libcloud/test/benchmarks/test_read_in_chunks.py
new file mode 100644
index 0000000..694e9fb
--- /dev/null
+++ b/libcloud/test/benchmarks/test_read_in_chunks.py
@@ -0,0 +1,153 @@
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from typing import Tuple, Callable
+
+import pytest
+
+from libcloud.utils.py3 import b, next
+from libcloud.utils.files import CHUNK_SIZE, read_in_chunks
+
+
+def _old_read_in_chunks(iterator, chunk_size=None, fill_size=False, yield_empty=False):
+    """
+    Old implementation of read_in_chunks without performance optimizations from #1847.
+
+    It's only here so we can directly measure compare performance of old and the new version.
+    """
+    chunk_size = chunk_size or CHUNK_SIZE
+
+    try:
+        get_data = iterator.read
+        args = (chunk_size,)
+    except AttributeError:
+        get_data = next
+        args = (iterator,)
+
+    data = b("")
+    empty = False
+
+    while not empty or len(data) > 0:
+        if not empty:
+            try:
+                chunk = b(get_data(*args))
+                if len(chunk) > 0:
+                    data += chunk
+                else:
+                    empty = True
+            except StopIteration:
+                empty = True
+
+        if len(data) == 0:
+            if empty and yield_empty:
+                yield b("")
+
+            return
+
+        if fill_size:
+            if empty or len(data) >= chunk_size:
+                yield data[:chunk_size]
+                data = data[chunk_size:]
+        else:
+            yield data
+            data = b("")
+
+
+# fmt: off
+@pytest.mark.parametrize(
+    "data_chunk_size_tuple",
+    [
+        (b"c" * (40 * 1024 * 1024), 1 * 1024 * 1024),
+        (b"c" * (40 * 1024 * 1024), 5 * 1024 * 1024),
+        (b"c" * (80 * 1024 * 1024), 1 * 1024 * 1024),
+    ],
+    ids=[
+        "40mb_data_1mb_chunk_size",
+        "40mb_data_5mb_chunk_size",
+        "80mb_data_1mb_chunk_size",
+    ],
+)
+@pytest.mark.parametrize(
+    "read_in_chunks_func",
+    [
+        _old_read_in_chunks,
+        read_in_chunks,
+    ],
+    ids=[
+        "old",
+        "new",
+    ],
+)
+# fmt: on
+def test_scenario_1(
+    benchmark, data_chunk_size_tuple: Tuple[bytes, int], read_in_chunks_func: Callable
+):
+    # similar to calling _upload_multipart_chunks with one large array of bytes
+    data, chunk_size = data_chunk_size_tuple
+
+    def run_benchmark():
+        for _ in read_in_chunks_func(iter([data]), chunk_size=chunk_size, fill_size=True):
+            pass
+
+    benchmark(run_benchmark)
+
+
+# fmt: off
+# NOTE: Because the old implementation is very slow when there is a chunk size mismatch, we need to
+# use smaller total objects size to prevent this benchmark from running for a very long time.
+@pytest.mark.parametrize(
+    "data_chunk_size_tuple",
+    [
+        (b"c" * (10 * 1024 * 1024), 8 * 1024),
+        (b"c" * (20 * 1024 * 1024), 1 * 1024 * 1024),
+        (b"c" * (30 * 1024 * 1024), 1 * 1024 * 1024),
+    ],
+    ids=[
+        "10mb_data_8k_chunk_size",
+        "20mb_data_1mb_chunk_size",
+        "30mb_data_1mb_chunk_size",
+    ],
+)
+@pytest.mark.parametrize(
+    "read_in_chunks_func",
+    [
+        _old_read_in_chunks,
+        read_in_chunks,
+    ],
+    ids=[
+        "old",
+        "new",
+    ],
+)
+# fmt: on
+def test_scenario_2(
+    benchmark, data_chunk_size_tuple: Tuple[bytes, int], read_in_chunks_func: Callable
+):
+    # similar to calling _upload_multipart_chunks with one large array of bytes
+    data, chunk_size = data_chunk_size_tuple
+    response_chunk = 5 * 1024 * 1024
+
+    # NOTE: It would be nice if we could also assert that data has been correctly add, but this
+    # would add additional overhead (since we would also measure accumulating this data) so we have
+    # those checks done separately in the unit tests.
+    def run_benchmark():
+        for a in read_in_chunks_func(
+            iter([data[i : i + response_chunk] for i in range(0, len(data), response_chunk)]),
+            chunk_size=chunk_size,
+            fill_size=True,
+        ):
+            pass
+
+    benchmark(run_benchmark)
diff --git a/libcloud/test/test_utils.py b/libcloud/test/test_utils.py
index c6fc161..395bfea 100644
--- a/libcloud/test/test_utils.py
+++ b/libcloud/test/test_utils.py
@@ -202,15 +202,42 @@
             for x in range(0, 1000):
                 yield "aa"
 
+        chunk_count = 0
         for result in libcloud.utils.files.read_in_chunks(
             iterator(), chunk_size=10, fill_size=False
         ):
+            chunk_count += 1
             self.assertEqual(result, b("aa"))
+        self.assertEqual(chunk_count, 1000)
 
+        chunk_count = 0
         for result in libcloud.utils.files.read_in_chunks(
             iterator(), chunk_size=10, fill_size=True
         ):
-            self.assertEqual(result, b("aaaaaaaaaa"))
+            chunk_count += 1
+            self.assertEqual(result, b("a") * 10)
+        self.assertEqual(chunk_count, 200)
+
+    def test_read_in_chunks_large_iterator_batches(self):
+        def iterator():
+            for x in range(0, 10):
+                yield "a" * 10_000
+
+        chunk_count = 0
+        for result in libcloud.utils.files.read_in_chunks(
+            iterator(), chunk_size=10, fill_size=False
+        ):
+            chunk_count += 1
+            self.assertEqual(result, b("a") * 10_000)
+        self.assertEqual(chunk_count, 10)
+
+        chunk_count = 0
+        for result in libcloud.utils.files.read_in_chunks(
+            iterator(), chunk_size=10, fill_size=True
+        ):
+            chunk_count += 1
+            self.assertEqual(result, b("a") * 10)
+        self.assertEqual(chunk_count, 10_000)
 
     def test_read_in_chunks_filelike(self):
         class FakeFile(file):
diff --git a/libcloud/utils/files.py b/libcloud/utils/files.py
index 3ab0bbc..fdf1ef7 100644
--- a/libcloud/utils/files.py
+++ b/libcloud/utils/files.py
@@ -15,6 +15,7 @@
 
 import os
 import mimetypes
+from typing import Generator
 
 from libcloud.utils.py3 import b, next
 
@@ -75,7 +76,9 @@
             return
 
         if fill_size:
-            if empty or len(data) >= chunk_size:
+            data = yield from _optimized_chunked_generator(data=data, chunk_size=chunk_size)
+            if empty:
+                # Yield last not completely filled chunk
                 yield data[:chunk_size]
                 data = data[chunk_size:]
         else:
@@ -83,6 +86,18 @@
             data = b("")
 
 
+def _optimized_chunked_generator(data: bytes, chunk_size: int) -> Generator[bytes, None, bytes]:
+    # We want to emit chunk_size large chunks, but chunk_size can be larger or smaller than the chunks returned
+    # by get_data. We need to yield in a loop to avoid large amounts of data piling up.
+    # The loop also avoids copying all data #chunks amount of times by keeping the original data as is.
+    chunk_start = 0
+    while chunk_start + chunk_size < len(data):
+        yield data[chunk_start : chunk_start + chunk_size]
+        chunk_start += chunk_size
+    data = data[chunk_start:]
+    return data
+
+
 def exhaust_iterator(iterator):
     """
     Exhaust an iterator and return all data returned by it.
diff --git a/tox.ini b/tox.ini
index 47db058..efdf9ec 100644
--- a/tox.ini
+++ b/tox.ini
@@ -387,6 +387,7 @@
 commands =
     cp libcloud/test/secrets.py-dist libcloud/test/secrets.py
     pytest -s -v --timeout 60 --benchmark-only --benchmark-name=short --benchmark-columns=min,max,mean,stddev,median,ops,rounds --benchmark-histogram=benchmark_histograms/benchmark  --benchmark-group-by=group,param:sort_objects libcloud/test/benchmarks/test_list_objects_filtering_performance.py
+    pytest -s -v --timeout 60 --benchmark-only --benchmark-name=short --benchmark-columns=min,max,mean,stddev,median,ops,rounds --benchmark-histogram=benchmark_histograms/benchmark --benchmark-group-by=group,func,param:read_in_chunks_func libcloud/test/benchmarks/test_read_in_chunks.py
 
 [testenv:import-timings]
 setenv =