在 python 中使用多处理来创建对二维矩阵切片进行操作的进程

Question

我有一个包含整数的二维矩阵（比如 5000 行 x 8000 列）。我想在 python 中使用多处理将矩阵的每个元素乘以 2，以便每个进程获得一组要处理的行并获得一个目标函数“array_mult”，该函数在它已发送的矩阵的分区上完成工作。

Array has been partitioned by rows and each partition sent to a (sub)process

import time,os
import multiprocessing as mp


A=[[1,2,3],[4,5,6],[7,8,9]]
global arr 
''' I am trying to use a global variable to write the output of the function so that
the storage is persistent and the output doesn't vanish when the process ends'''

def array_mult(a): 
    '''This is the function which is supposed to 
    multiply each element of input matrix a'''
    print("array is =",a)
    for i in range(len(a)):
        print("counter is",i)
        a[i]=a[i]*2
    print(a,os.getpid())
    arr.append(a)

if __name__ == '__main__':


    starttime = time.time()
    array_proc=list()
    for i in range(3):
        p=mp.Process(target=array_mult, args=(A[i], )) ### I am trying to send partitions of the list as the arg to the function array_mult
        array_proc.append(p)
        p.start()

    for process in array_proc:
        process.join()


    print(time.time() - starttime)
    print(A)
    print(arr)

**

• CONSTRAINT - 不能使用 python 核心模块之外的任何东西或低于 python 3.6 的任何功能

** 使用 ctypes 库和使用 RawArray 有用吗？如果是这样，我该如何使用它？ 保存二维矩阵的任何其他想法？ （我不想使用 numpy，因为它不是核心包）

Answer 1

'''Created on 12-Jun-2020

@author: Shouvik 
'''
import time,os
import multiprocessing as mp
from multiprocessing import sharedctypes
from ctypes import Structure,c_int




num_of_columns=10000
num_of_rows=10000
num_of_cpu=len(os.sched_getaffinity(0))

class Row_Vector(Structure):
    _fields_ = [("column", c_int * num_of_columns)]

class array_2d(Structure):
    _fields_ = [("Rectangular_Matrix", Row_Vector * num_of_rows)]



'''create_row_boundaries returns a list with the row-numbers which partition the matrix row-wise '''
def create_row_boundaries(num_of_rows):
    num_of_cpu=len(os.sched_getaffinity(0))
    row_boundary=[0,(int(num_of_rows/num_of_cpu) if num_of_rows>1 else 1)]

    'we add num_of_rows/num_of_cpu to each partition' 
    index=2
    while row_boundary[index-1]< (num_of_rows-1):

            row_boundary.append(int(row_boundary[index-1])+1)

            '''After adding integer number of 'num_of_rows/num_of_cpu' we can be left with any of 0,1,2... (num_of_cpu-1) rows 
            which should be added to the last element'''
            if num_of_rows-num_of_cpu<row_boundary[index-1]+int(num_of_rows/num_of_cpu)<=num_of_rows :
                #print("num of rows- num of cpu is",num_of_rows-num_of_cpu,"row_boundary[index]+int(num_of_rows/num_of_cpu)", row_boundary[index]+int(num_of_rows/num_of_cpu))
                row_boundary.append(num_of_rows-1)
            else:
                row_boundary.append(int(row_boundary[index-1])+ int(num_of_rows/num_of_cpu))
            index =index+2

    return row_boundary



'''matrix_operation operates on each element of the matrix (type RawArray) that is passed to it'''
def matrix_operation(a,row_initial, row_final,column_initial, column_final):
    print("this instance of 'matrix_operation' is going to work from row number {} to row number{}".format(row_initial,row_final))
    for i in range(row_initial,row_final+1,1):
        for j in range(column_initial,column_final):
            a[i].column[j]=2 * a[i].column[j]

#     print("Row_initial{} to Row_final{} done by process id {}".format(row_initial,row_final,os.getpid()))
#     print("The process which operates on below matrix is",os.getpid())
#     for i in range(row_initial,row_final+1,1):
#         print([a[i].column[j] for j in range(column_initial,column_final)],"pid is {}".format(os.getpid()))
    return 





if __name__ == '__main__':

    '''We create a matrix of type Raw_Array having num_of_rows rows 
    and each row is a column vector having num_of_columns columns and operate using multiprocessing'''
    m1=sharedctypes.RawArray(Row_Vector, num_of_rows)

    '''We create another matrix;operate on it sequentially on a single core of the cpu;time the operation'''
    m2=sharedctypes.RawArray(Row_Vector, num_of_rows)



    '''the two nested for loops below simply populate the matrices m1 and m2 with some values'''
    for i in range(num_of_rows):
        for j in range(num_of_columns):
            m1[i].column[j]=i*j
            m2[i].column[j]=i*j

    '''The for loop and the print statement below print the matrix row-wise'''      
#     for i in range(num_of_rows):
#         print([m1[i].column[j] for j in range(num_of_columns)] )



    matrix_partition_by_row_num=create_row_boundaries(num_of_rows)
    index=0
    array_proc=list()

    starttime = time.time()

    for i in range(int(len(matrix_partition_by_row_num)/2)):    #originally range had num_of_cpu as the argument

        p=mp.Process(target=matrix_operation, args=(m1, matrix_partition_by_row_num[index],matrix_partition_by_row_num[index+1],0,num_of_columns)) 
        '''We pass the matrix m1 and it's various partitioning rows to the 'matrix_operation' function'''
        array_proc.append(p)
        p.start()
        index=index+2

    for process in array_proc:
        process.join()

    print("Time taken for concurrent operation is {:e}".format(time.time() - starttime))

#     for i in range(num_of_rows):
#         print([m[i].column[j] for j in range(num_of_columns)] )

    print("no of cpu",num_of_cpu,"matrix partition values",matrix_partition_by_row_num)
#     for i in range(num_of_rows):
#         print([m[i].column[j] for j in range(num_of_columns)])
    '''We will simply input the entire matrix to the matrix_operation function and time the process'''
    sequential_process_starttime=time.time()
    matrix_operation(m2, 0, num_of_rows-1, 0, num_of_columns)
    print("time taken for sequential operation is {:e}".format(time.time()-sequential_process_starttime))

    Is_matrix_operation_correct= True
    for i in range(num_of_rows):
        for j in range(num_of_columns):
            if (m1[i].column[j] != m2[i].column[j]) :
                Is_matrix_operation_correct=False
                break

    print("Is matrix operation correct: {}".format(Is_matrix_operation_correct) )