Paralel Radix Sort in Python

Question

im learning parallel programming in python nad ive got code like this

import multiprocessing
import random
import time

def merge_buckets(final_bucket,bucket):
    helper=0
    for i in bucket:
        for j in i:
            final_bucket[helper].append(j)
        helper += 1
    return final_bucket
def radix_sort(array,start,end,digit,helper,result_queue):
        bucket = [[] for _ in range(10)]
        #cisklus ktory prechadza polom cisel a rozdelujem ich do bucketov
        for k in range(start,end):

            num = array[k]
            print(num)
            string_number = str(num)
            if(num < digit):
                bucket[0].append(num)
            else:
                #print((max_digits-helper),"=",num)
                bucket[int(str(num)[len(str(num))-helper])].append(num)
        result_queue.put(bucket)
        #reset pola cisel do ktoreho sa znovu vyhodia novo usporiadane cisla z bucketu
        #musi byt dakde ale ne tu dakde v dajakom inom proces
        #navysenie pocitadiel

if __name__ == '__main__':
    cores = int(input("Cores : "))
    array = [random.randint(0, 100) for _ in range(1000)]
    lock = multiprocessing.Lock()
    #arr = multiprocessing.Array('i',[3, 5, 1, 18, 25, 4, 17, 22, 31, 6, 89, 92, 7], lock=lock)
    #bucket = multiprocessing.Array ('i',[[] for _ in range(10)],lock=lock)
    manager = multiprocessing.Manager()
    arr = manager.list([3, 5, 1, 18, 25, 4, 17, 22, 31, 6, 89, 92, 7])

    # vybere maximalnu hodnotu z pola
    max_val = max(array)
    print("Maximum of numbers: ", max_val)
    # digit znamena ze ci polom prechadzame ze 1 , 10 , 100 ako ide radix zoraduje od poslednej cislice
    digit = 1



    num = len(array)//cores

    zvysok = len(array)%cores

    parts = []
    for i in range(cores):
        if(zvysok != 0):
            parts.append(num + 1)
            zvysok -= 1
        else:
            parts.append(num)

    result_queue = multiprocessing.Queue()
    start = time.time()
    helper = 1
    while(max_val // digit > 0):
        print(array[:20])
        results = []
        result_queue = multiprocessing.Queue()
        processes = []
        j = 0
        bucket = [[] for _ in range(10)]
        for o in parts:
            p = multiprocessing.Process(target=radix_sort, args=(array,j,j+o,digit,helper,result_queue))
            p.start()
            processes.append(p)
            j += o

        for p in processes:
            p.join()

        results = []
        for _ in range(len(parts)):
            results.append(result_queue.get())
        for result in results:
            bucket = merge_buckets(bucket,result)
        array = []
        for i in bucket:
            for j in i:
                array.append(j)
        for p in processes:
            p.terminate()


        digit *= 10
        helper = helper + 1
    end = time.time()
    #array = radix_sort(array)
    print(array)
    print("Cas : ", end - start)

I want to ask you how would i make this working, now asyou can see im making threads for every postion in numbers so like if there is 2 igit number and i set 8 cores program makes 16. And if there are 3 digits i make 24. So i know that in multiprocessing there are som synchronization tools but i dont know how to apply them on radix sort. Cant someone think of way how to do this. I will be really glad.

Answer 1

I don't know how Python's multi-threading works, but for a multi-threaded radix sort, you just need to start and wait for completion of each thread.

A typical implementation of a multi-threaded radix sort first performs a single threaded initial most significant digit pass to split up the data into buckets, for example base 256 to create 256 buckets. It will help if enough enough buckets are created so that the average bucket size fits within a thread's local cache, typically each core's L1 and L2 cache.

Assume that a processor efficiently supports K threads. For best efficiency, a thread pool of K threads is used. To implement a thread pool, something similar to Windows Wait For Multiple Objects is needed:

https://learn.microsoft.com/en-us/windows/win32/api/synchapi/nf-synchapi-waitformultipleobjects

Linux doesn't have native support for this, but there are third party libraries that implement the equivalent. I don't know about Python.

Each of the K threads is started to sort a bucket. Once any of the threads completes, that thread is then used to sort the next unsorted bucket. Once all buckets are sorted, the threads are shut down and the buckets concatenated to form a sorted array.

---

A simpler and possibly faster alternative would be to split the array into K parts, and do K radix sorts in parallel, then use parallel merges to merge the sorted sub-arrays. For example if K = 8, then after the radix sorts complete, 4 threads to merge 8 sub-arrays into 4, 2 threads to merge 4 sub-arrays into 2, then a single thread to do the final merge of 2 sub-arrays into 1.