Test for behaviour of GC in python multiprocessing using regular/shared objects

multiprocess_gc.py

import multiprocessing
import time
import sys

process_identifier = 'main'


class Collectible:
    def __init__(self, identifier):
        self.identifier = identifier
        self.collections = multiprocessing.Value('i')

    def __del__(self):
        with self.collections.get_lock():
            self.collections.value += 1
        print('__del__ %r in %s [%d]' % (self.identifier, process_identifier, self.collections.value))

    def __repr__(self):
        return '%s(%s, collections=%d)' % (self.__class__.__name__, self.identifier, self.collections.value)


def drop_refs(child_id, payload: Collectible):
    global process_identifier
    process_identifier = child_id
    print('start', child_id, 'payload', payload)
    time.sleep(0.2)
    print('drop', child_id)
    del payload
    time.sleep(0.2)
    print('done', child_id)


def local_ref(drop_early=False):
    local_child = Collectible('local')
    procs = [
        multiprocessing.Process(target=drop_refs, args=(i, local_child))
        for i in range(4)
    ]
    print('start children')
    if drop_early:
        print('drop main')
        del local_child
    for proc in procs:
        proc.start()
    for proc in procs:
        proc.join()
    print('done all')


if __name__ == '__main__':
    method = sys.argv[1] if len(sys.argv) > 1 else None
    multiprocessing.set_start_method(method)
    print('drop late....')
    local_ref()
    print('drop early....')
    local_ref(drop_early=True)