I would suggest looking at Pools to easily create processes on the fly. If you need to have some shared state, in this case a boolean indicating a non-black image has been found, look at Managers.

Update: Here is an example of what I mean.

import multiprocessing.Manager as Manager
import multiprocessing.Pool as Pool

m = Manager()
p = Pool(processes=5)

state_info = m.dict()
state_info['image_found'] = False

def processImage(img):

    # ... Process Image ...

    if imageIsBlack(img):
        state_info['image_found'] = True
        p.terminate()

 p.apply(processImage, imageList)

 if state_info['image_found']:
     print 'There was a black image!!'
 else:
     print 'No black images were found.'

I'd do something like this, starting with dependencies:

from multiprocessing import Pool
import numpy as np
from PIL import Image

# and some for testing
from random import random
from time import sleep

first I define a function to divide an image up into "chunks", sort of as you talked about:

def chunkit(ys, xs, blocksize=64):
    for y in range(0, ys, blocksize):
        yt = (y, min(ys, y + blocksize))
        for x in range(0, xs, blocksize):
            xt = (x, min(xs, x + blocksize))
            yield yt, xt

it's a lazy iterator, so this can go on for a while.

I then define my worker function:

def dumb_func(cc):
    (y0,y1), (x0,x1) = cc
    # convert to floats for ease of processing
    chunk = image[y0:y1,x0:x1] / 255.
    # random slow down for testing
    # sleep(random() ** 6)
    res = chunk ** 2
    # convert back to bytes for efficiency
    return cc, (res * 255).astype(np.uint8)

I make sure the source array stays as close to original format as possible for efficiency and send it back in the same format (this might take some fiddling, if you're dealing with other pixel formats obviously).

then I put it together:

if __name__ == '__main__':
    source = Image.open('tmp.jpeg')
    image = np.asarray(source)
    print("loaded", image.shape, image.dtype)

    with Pool() as pool:
        resit = pool.imap_unordered(
            dumb_func, chunkit(*image.shape[:2]))

        output = np.empty_like(image)
        for cc, res in resit:
            (y0,y1), (x0,x1) = cc
            output[y0:y1,x0:x1] = res

    im = Image.fromarray(output, 'RGB')
    im.save('out.jpeg')

this churns through a 15Mpixel image in a couple of seconds, with most of that spent loading/saving the image. it could probably be a lot more clever with array strides and cache friendliness, but hope that helps!

note: I think this code relies on CPython Unix style process forking semantics to make sure the image is shared between processes efficiently. not sure what would happen if you ran it on something else

You have to create list with pairs (filename, path)

data = [(img, img_dir) for img in images]

and then map will run every pair in separated process.

But you have to get args in start_pipeline

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

And you have to use () to create instance of class segmenter_class to use start_pipeline

pool.map(segmenter_class().start_pipeline, data)

BTW: In example code I also return result from process.

import os
import multiprocessing

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self):
        print('init')

    def read_image(self):
        print('read')

    def write_image(self):
        print('write')

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        print('ok HSV')
        return "result HSV"
    
class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        print('ok LAB')
        return "result LAB"

if __name__ == '__main__':

    procedure = 'hsv'
    
    segmenter_class = {
        'hsv': HSV_Segmenter,
        'lab': LabSegmenter,
    }.get(procedure)
    
    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data = [(img, img_dir) for img in images]
    
    pool = multiprocessing.Pool(3)

    # example 1

    results = pool.map(segmenter_class().start_pipeline, data)
    print('Results:', results)

    # example 2

    for result in pool.map(segmenter_class().start_pipeline, data):
        print('result:', result)

    pool.terminate()

EDIT:

You can also create function which gets procedure and data and then use it in map - this way every process will create own instance of procedure or you can send different procedures to different processes.

import os
import multiprocessing

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self):
        print('init')

    def read_image(self):
        print('read')

    def write_image(self):
        print('write')

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        print('ok HSV')
        return "result HSV"
    
class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        print('ok LAB')
        return "result LAB"

segmenters = {
    'hsv': HSV_Segmenter,
    'lab': LabSegmenter,
}

def start_process(args):

    procedure = args[0]
    data = args[1:]

    segmenter_class = segmenters.get(procedure)
    result = segmenter_class().start_pipeline(data)

    return result
    
if __name__ == '__main__':

    procedure = 'hsv'
    
    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data = [(procedure, img, img_dir) for img in images]
    
    pool = multiprocessing.Pool(3)

    # example 1

    results = pool.map(start_process, data)
    print('Results:', results)

    # example 2

    for result in pool.map(segmenter_class().start_pipeline, data):
        print('result:', result)

    pool.terminate()

Example with different procedures

if __name__ == '__main__':

    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    pool = multiprocessing.Pool(3)

    data = [('hsv', img, img_dir) for img in images]
    results = pool.map(start_process, data)
    print('Results HSV:', results)

    data = [('lab', img, img_dir) for img in images]
    results = pool.map(start_process, data)
    print('Results LAB:', results)

    pool.terminate()

And the same with one map(). There are 6 processes to start and Pool(3) so it will run only 3 process at the same time and when it will have free process then map will get next values from list and run process.

if __name__ == '__main__':

    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data_hsv = [('hsv', img, img_dir) for img in images]
    data_lab = [('lab', img, img_dir) for img in images]
    
    data = data_hsv + data_lab

    pool = multiprocessing.Pool(3)

    # example 1

    results = pool.map(start_process, data)
    print('Results:', results)

    # example 2

    for result in pool.map(start_process, data):
        print('results:', result)

    pool.terminate()

EDIT:

It works also with Ray

It needs only

from ray.util import multiprocessing

instead of

import multiprocessing

I didn't tested it with Dask, PySpark or Joblib

EDIT:

Example with Joblib

from joblib import Parallel, delayed

class SegmentationType(object):
    DISPLAY_NAME = "invalid"

    def __init__(self):
        print('init')

    def read_image(self):
        print('read')

    def write_image(self):
        print('write')

    def process(self):
        # override in derived classes to perform an actual segmentation
        pass

    def start_pipeline(self, args):
        print('ok starting')
        
        filename, path = args
        print('filename: {}\npath: {}'.format(filename, path))
        
        return self.process()

class HSV_Segmenter(SegmentationType):
    DISPLAY_NAME = 'HSV'

    def process(self):
        print('ok HSV')
        return "result HSV"

class LabSegmenter(SegmentationType):
    DISPLAY_NAME = 'LAB'

    def process(self):
        print('ok LAB')
        return "result LAB"

segmenters = {
    'hsv': HSV_Segmenter,
    'lab': LabSegmenter,
}

def start_process(args):
    
    procedure = args[0]
    data = args[1:]
    
    segmenter_class = segmenters.get(procedure)
    result = segmenter_class().start_pipeline(data)
    
    return result

if __name__ == '__main__':

    images = ['01.png', '02.png', '03.png']
    img_dir = 'C:/'
    
    data_hsv = [('hsv', img, img_dir) for img in images]
    data_lab = [('lab', img, img_dir) for img in images]
    
    data = data_hsv + data_lab

    # --- version 1 ---

    #pool = Parallel(n_jobs=3, backend='threading')
    #pool = Parallel(n_jobs=3, backend='multiprocessing')
    pool = Parallel(n_jobs=3)
    
    # example 1
    
    results = pool( delayed(start_process)(args) for args in data )
    print('Results:', results)

    # example 2
    
    for result in pool( delayed(start_process)(args) for args in data ):
        print('result:', result)

    # --- version 2 ---
    
    #with Parallel(n_jobs=3, backend='threading') as pool:
    #with Parallel(n_jobs=3, backend='multiprocessing') as pool:
    with Parallel(n_jobs=3) as pool:

        # example 1
        
        results = pool( delayed(start_process)(args) for args in data )
        print('Results:', results)

        # example 1

        for result in pool( delayed(start_process)(args) for args in data ):
            print('result:', result)
        

The error you're seeing is because you're calling a method that doesn't exist, probably because it's not part of self.

You'll likely see better performance with multithreading than multiprocessing. Multiprocessing is best for CPU-bound tasks for the simple reason that Python uses a global lock for all operations. Multiprocessing is a hack to get around this lock. It's nastier to work with than threading and it's best to avoid it unless absolutely necessary.

Multithreading is likely enough for your use case and it won't create lots of gotchas for a new programmer. Here's a working sample setup using Python 3's Futures API that will easily scale your problem size, just add your arguments and actual save code in the marked places.

import concurrent.futures

# Save single image
def save_image(image_arg):
    # your image save code goes here
    print("Working on image {}...".format(image_arg))
    return True

# max_workers specifies the number of threads. If None then use 5x your CPU count
with concurrent.futures.ThreadPoolExecutor(max_workers=None) as executor:
    # Images we'll save. Depending on how you generate your images you might not
    # want to materialize a list like this to avoid running out of memory.
    image_args = ["image1", "image2", "image3"]

    # Submit futures to the executor pool.
    # Map each future back to the arguments used to create that future. That way
    # if one fails we know which image it was that failed.
    future_to_args = {executor.submit(save_image, image_arg): image_arg for image_arg in image_args}

    # Images are being saved in worker threads. They will complete in any order.
    for future in concurrent.futures.as_completed(future_to_args):
        image_arg = future_to_args[future]
        try:
            result = future.result()
        except Exception as exc:
            print("Saving image {} generated an exception: {}".format(image_arg, exc))
        else:
            print("Image {} saved successfully.".format(image_arg))

If you insist on multiprocessing, just use ProcessPoolExecutor instead. That might be worthwhile if you also want to generate your images in parallel.

Whether ThreadPoolExecutor or ProcessPoolExecutor is better depends a lot on what the rest of your workload is and how you structured it. Try both to see which works better for you. Note that multiprocessing places restrictions on communication and sharing state between workers, hence why I suggest trying threads first.