Is this the most I can get from Python multiprocess?

Note: This SharedMemory works only for Python >= 3.8 since it first appeared there

Start 3 kinds of processes: Reader, Processor(s), Writer.

Have Reader process read the file incrementally, sharing the result via shared_memory and Queue.

Have the Processor(s) consume the Queue, consume the shared_memory, and return the result(s) via another Queue. Again, as shared_memory.

Have the Writer process consume the second Queue, writing to the destination file.

Have them all communicate through, say, some Events or DictProxy, with the MainProcess who will act as the orchestrator.

Example:

import time
import random
import hashlib
import multiprocessing as MP

from queue import Queue, Empty

# noinspection PyCompatibility
from multiprocessing.shared_memory import SharedMemory

from typing import Dict, List


def readerfunc(
        shm_arr: List[SharedMemory], q_out: Queue, procr_ready: Dict[str, bool]
):
    numshm = len(shm_arr)
    for batch in range(1, 6):
        print(f"Reading batch #{batch}")
        for shm in shm_arr:
            #### Simulated Reading ####
            for j in range(0, shm.size):
                shm.buf[j] = random.randint(0, 255)
            #### ####
            q_out.put((batch, shm))
        # Need to sync here because we're reusing the same SharedMemory,
        # so gotta wait until all processors are done before sending the
        # next batch
        while not q_out.empty() or not all(procr_ready.values()):
            time.sleep(1.0)


def processorfunc(
        q_in: Queue, q_out: Queue, suicide: type(MP.Event()), procr_ready: Dict[str, bool]
):
    pname = MP.current_process().name
    procr_ready[pname] = False
    while True:
        time.sleep(1.0)
        procr_ready[pname] = True
        if q_in.empty() and suicide.is_set():
            break
        try:
            batch, shm = q_in.get_nowait()
        except Empty:
            continue
        print(pname, "got batch", batch)
        procr_ready[pname] = False
        #### Simulated Processing ####
        h = hashlib.blake2b(shm.buf, digest_size=4, person=b"processor")
        time.sleep(random.uniform(5.0, 7.0))
        #### ####
        q_out.put((pname, h.hexdigest()))


def writerfunc(q_in: Queue, suicide: type(MP.Event())):
    while True:
        time.sleep(1.0)
        if q_in.empty() and suicide.is_set():
            break
        try:
            pname, digest = q_in.get_nowait()
        except Empty:
            continue
        print("Writing", pname, digest)
        #### Simulated Writing ####
        time.sleep(random.uniform(3.0, 6.0))
        #### ####
        print("Writing", pname, digest, "done")


def main():
    shm_arr = [
        SharedMemory(create=True, size=1024)
        for _ in range(0, 5)
    ]
    q_read = MP.Queue()
    q_write = MP.Queue()
    procr_ready = MP.Manager().dict()
    poison = MP.Event()
    poison.clear()

    reader = MP.Process(target=readerfunc, args=(shm_arr, q_read, procr_ready))

    procrs = []
    for n in range(0, 3):
        p = MP.Process(
            target=processorfunc, name=f"Proc{n}", args=(q_read, q_write, poison, procr_ready)
        )
        procrs.append(p)

    writer = MP.Process(target=writerfunc, args=(q_write, poison))

    reader.start()
    [p.start() for p in procrs]
    writer.start()

    reader.join()
    print("Reader has ended")

    while not all(procr_ready.values()):
        time.sleep(5.0)
    poison.set()
    [p.join() for p in procrs]
    print("Processors have ended")

    writer.join()
    print("Writer has ended")

    [shm.close() for shm in shm_arr]
    [shm.unlink() for shm in shm_arr]


if __name__ == '__main__':
    main()

The first thing that comes to mind for the code is to run the saving function in the thread. By this we exclude the bottelneck of waiting disk writing. Like so:

executor = ThreadPoolExecutor(max_workers=2)
future = executor.submit(save_computed_data_to_disk, to_be_computed_filename, results)
saving_futures.append(future)
...
concurrent.futures.wait(saving_futures, return_when=ALL_COMPLETED)  # wait all saved to disk after processing
print("Done")

You say you have 8 cores, yet you have:

POOL_SIZE = 15 #nbcore - 1

Assuming you want to leave one processor free (presumably for the main process?) why wouldn't this number be 7? But why do you even want to read a processor free? You are making successive calls to map. While the main process is waiting for these calls to return, it requires know CPU. This is why if you do not specify a pool size when you instantiate your pool it defaults to the number of CPUs you have and not that number minus one. I will have more to say about this below.

Since you have a very large, in-memory list, is it possible that you are expending waisted cycles in your loop rewriting this list on each iteration of the loop. Instead, you can just take a slice of the list and pass that as the iterable argument to map:

POOL_SIZE = 15 # ????
PACKET_SIZE = 2000
data_lines = util.load_data_lines(to_be_computed_filename)
number_of_packets, remainder = divmod(number_of_lines, PACKET_SIZE)
with Pool(processes=POOL_SIZE) as pool:
    offset = 0
    for i in range(number_of_packets):
        results = pool.map(process, data_lines[offset:offset+PACKET_SIZE])
        offset += PACKET_SIZE
        save_computed_data_to_disk(to_be_computed_filename, results)
    if remainder:
        results = pool.map(process, data_lines[offset:offset+remainder])
        save_computed_data_to_disk(to_be_computed_filename, results)
print("Done")

Between each call to map the main process is writing out the results to to_be_computed_filename. In the meanwhile, every process in your pool is sitting idle. This should be given to another process (actually a thread running under the main process):

import multiprocessing
import queue
import threading

POOL_SIZE = 15 # ????
PACKET_SIZE = 2000
data_lines = util.load_data_lines(to_be_computed_filename)
number_of_packets, remainder = divmod(number_of_lines, PACKET_SIZE)

def save_data(q):
    while True:
        results = q.get()
        if results is None:
            return # signal to terminate
        save_computed_data_to_disk(to_be_computed_filename, results)

q = queue.Queue()
t = threading.Thread(target=save_data, args=(q,))
t.start()

with Pool(processes=POOL_SIZE) as pool:
    offset = 0
    for i in range(number_of_packets):
        results = pool.map(process, data_lines[offset:offset+PACKET_SIZE])
        offset += PACKET_SIZE
        q.put(results)
    if remainder:
        results = pool.map(process, data_lines[offset:offset+remainder])
        q.put(results)
q.put(None)
t.join() # wait for thread to terminate
print("Done")

I've chosen to run save_data in a thread of the main process. This could also be another process in which case you would need to use a multiprocessing.Queue instance. But I figured the main process thread is mostly waiting for the map to complete and there would not be competition for the GIL. Now if you do not leave a processor free for the threading job, save_data, it may end up doing most of the saving only after all of the results have been created. You would need to experiment a bit with this.

Ideally, I would also modify the reading of the input file so as to not have to first read it all into memory but rather read it line by line yielding 2000 line chunks and submitting those as jobs for map to process:

import multiprocessing
import queue
import threading

POOL_SIZE = 15 # ????
PACKET_SIZE = 2000


def save_data(q):
    while True:
        results = q.get()
        if results is None:
            return # signal to terminate
        save_computed_data_to_disk(to_be_computed_filename, results)


def read_data():
    """
    yield lists of PACKET_SIZE
    """
    lines = []
    with open(some_file, 'r') as f:
        for line in iter(f.readline(), ''):
            lines.append(line)
            if len(lines) == PACKET_SIZE:
                yield lines
                lines = []
        if lines:
            yield lines

q = queue.Queue()
t = threading.Thread(target=save_data, args=(q,))
t.start()

with Pool(processes=POOL_SIZE) as pool:
    for l in read_data():
        results = pool.map(process, l)
        q.put(results)
q.put(None)
t.join() # wait for thread to terminate
print("Done")

I made two assumptions: The writing is hitting the I/O bound, not the CPU bound - meaning that throwing more cores onto writing would not improve the performance. And the process function contains some heavy computations.

I would approach it differently:

1. Split up the large list into a list of list
2. Feed it than into the processes
3. Store the total result

Here is the example code:

import multiprocessing as mp

data_lines = [0]*10000 # read it from file
size = 2000

# Split the list into a list of list (with chunksize `size`)
work = [data_lines[i:i + size] for i in range(0, len(data_lines), size)]

def process(data):
    result = len(data) # some something fancy
    return result

with mp.Pool() as p:
    result = p.map(process, work)

save_computed_data_to_disk(file_name, result)

On meta: You may also have a look into numpy or pandas (depending on the data) because it sounds that you would like to do something into that direction.