Multiprocessing array с экземпляром класса

@Snowman8526 · Регистрация: 26.09.2017

Студворк — интернет-сервис помощи студентам

добрый день. Решил протестировать как будет вести себя программа в мульти поточных вычислениях. Начал разбираться нашёл модуль multiprocessing он устраивает своей простотой но к моей задаче Я не нашёл как прикрутить.
Вырезки из кода

Python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
#Описание класса
class point:
    temp = 0
    x_x = 0
    y_y = 0
    def assignTemp(self, tempId):
        self.temp = tempId
    def assignX_X(self, x):
        self.x_x = x
    def assignY_Y(self, y):
        self.y_y = y
 
#Создание экземпляра класса и добавление их в список их может быть очень много.
def createBuble(conf):
    array = []
    for x in range(3000):
        array.append(point())
    return array

Дальше пытаюсь экспериментировать хочу передать половину одному потоку половину другому (Пока топорно сейчас хочу довести чтобы просто работало)

Python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from multiprocessing import Process, Array
 
l=len(array)
 
arr1 = Array(array[:l/2])
arr2 = Array(array[l/2+1:])
 
t = Process(target=pointGo.tempGo, args=(p1, p2, p3, arr1)) 
v = Process(target=pointGo.tempGo, args=(p1, p2, p3, arr2)) 
 
t.start()
v.start()
 
t.join()
v.join()
array = arr1[:] + arr2[:]

выдаёт ошибку
arr1 = Array(array[:l/2])
TypeError: slice indices must be integers or None or have an __index__ method

либо
arr = Array(array)
TypeError: Array() missing 1 required positional argument: 'size_or_initializer'

На данный момент целостность списка важна.

@shsv · 06.10.2017, 18:56

По первой ошибке

Python
1
2
arr1 = array[:l//2]
arr2 = array[l//2:]

Второе. multiprocessing.Array - создает массив объектов ctypes. Это не то, что вам нужно. Список, разделяемый процессами, - это multiprocessing.Manager.list.

multiprocessing подразумевает обязательное использование конструкции if __name__ == '__main__': (чтобы «маринатор» не уронил скрипт в бездну бесконечного порождения новых и новых процессов... Впрочем, иначе multiprocessing бросит исключение, и вежливо попеняет, мол, вы, возможно, зыбыли использовать сию идиому).

На target=pointGo.tempGo будет ругаться «маринатор» (pickle), т.к. сериализовать такое не может (все аргументы target-функции тоже должны быть сериализуемыми).

Схематичное итого

Python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from multiprocessing import Process, Manager
 
 
class Point:
    def __init__(self, x=0, y=0):
        self.x = x
        self.y = y
 
 
def createBuble():
    array = []
    for x in range(3000):
        array.append(Point())
    return array
 
 
def f(x, y, lst):
    for i, it in enumerate(lst):
        it.x = x
        it.y = y
        lst[i] = it
 
 
if __name__ == '__main__':
    m = Manager()
    lst = createBuble()
    l = len(lst)
    arr1 = m.list(lst[:l//2])
    arr2 = m.list(lst[l//2:])
        
    t = Process(target=f, args=(-1, -1, arr1))
    v = Process(target=f, args=(-2, -2, arr2))
    
    t.start()
    v.start()
    
    t.join()
    v.join()
 
    arr = arr1 + arr2
    
    for it in arr:
        print(it.x, it.y)
    
    m.shutdown()

Гляньте multiprocessing.Pool и/или concurrent.futures.ProcessPoolExecutor (последний и попроще, и пошустрее). Вероятно они вам подойдут...

@Snowman8526 · 17.11.2017, 15:13 **[ТС]**

Пришлось отойти от проблемы покурить мануалы и столкнулся с проблемами.
Задача прежняя посчитать большой объём точек. Точки взаимодействуют друг с другом. Если условно разделить массив на 2-е части то будет вхождение вычисляемых точек как с одной стороны так и с другой. По этому мне очень хочется чтобы у всех процессов была общая память.

Python
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
from multiprocessing import Process, Array, Manager
#Стандартный класс для эксперимента
class Point:
    temp = 0
    #вставляем вводимое число в переменную
    def assignTemp(self, tempId):
        self.temp = tempId
    #Изменяем на переменную
    def chengeTemp(self, chTemp):
        self.temp = self.temp + chTemp
    #Возвращаем значение
    def recordTemp(self):
        return self.temp
#Создаем экземпляры класса
def createBuble(conf):
    array = []
    for x in range(conf):
        array.append(Buble())
    return array
#Изменяем всем значение temp
def temp(array, conf):
    for i in range(conf):
        array[i].assignTemp(24)
    return array
#Тест пытаюсь изменить значение на единицу
def test-1 (arr):
    #arr = array
    arr[1].chengeTemp(1)
    print("i", arr[1].recordTemp())
    
if __name__ == "__main__":
    array = createBuble(10)
    array = temp(array, 10)

Если бы мне нужен был бы неизменяемый массив.

Python
1
2
3
4
5
6
7
manager = Manager()
arr = manager.list(array)
print(arr[1].recordTemp())
pr = Process(target=test-1, args=(arr, ))
pr.start()
pr.join()
print(arr[1].recordTemp())

получаю

Bash
1
2
3
24
i 24
24

Подскажите как извернуться чтобы изменять общую память. Я уже думаю о копании в треды. делить потоки, передавать мультипроцессору часть задания и весь массив, заполнять manage.dict() а что вернётся переписать оригинальный массив. Но первая попытка не привела к успеху. Может есть простой путь? Если можно примеры.

@shsv · 17.11.2017, 18:11

Функция test_1 выглядеть должна так примерно

Python
1
2
3
4
5
def test_1(arr):
    it = arr[1]
    it.chengeTemp(1)
    arr[1] = it
    print("i", arr[1].recordTemp())

Штука в том, что в случае Manager «разделяемой памяти» как таковой - нет. Manager - это отдельный процесс, связанный с рабочими процессами пайпами. Когда вы, например, пишите arr = manager.list внутри него создается list. Непосредтвенного доступа к этому списку, разумеется, нет. Поэтому действие arr[1] = value приведёт к тому, что копия value будет отправлена по трубе manager'у, где и будет запихана в нужный список по требуемому индексу. И обратно, манипуляции с arr[1] - это действия над копией, содержащегося там объекта. Если объект - изменяемый (класс Point в нашем случае), arr[1].chengeTemp(1) приведёт к изменению его копии в текущем рабочем процессе, и никак не отразится на оригинале в «разделяемой памяти».

@Snowman8526 · 17.11.2017, 19:55 **[ТС]**

shsv, спасибо большое за объяснение буду экспериментировать.

@Snowman8526 · 21.11.2017, 09:30 **[ТС]**

shsv, Ваш пример помог но есть небольшая сложность. В одном цикле к списку который передаю обращений штук 60. Цикл выполняется примерно за пол секунды а то и быстрее. Получается что создавая 2-а процесса у меня работают 3-и процесса загружая 2-х ядерник под завязку. на 4-х ядерном в 5 процессов загрузка от 60% до 80% Предположил что слишком частое обращение к списку что до предела грузит канал связи процессов.

Решил уменьшить нагрузку на запросах на изменение всего 2-а в цикле можно и пожертвовать

Попробовал создать копию до создания процессов(arr2 = arr) и каждому передать не помогло. Так же попробовал создать после создания процесса. И пробовал создать копию с помощью модуля copy (arr2 = copy.copy(arr))

Может вы знаете метод как уменьшить нагрузку на канал связи?

@shsv · 21.11.2017, 18:43

Что-то вроде

Python
1
2
3
4
5
def test_1 (arr, k, a,b):
    it = arr[a:b]
    for i in it:
        i.chengeTemp(k)
    arr[a:b] = it

?

И всё же гляньте concurrent.futures.ProcessPoolExecutor...

Новые блоги и статьи Все статьи Все блоги /
сукцессия микоризы: основная теория в виде двух уравнений. anaschu 11.01.2026 https:/ / rutube. ru/ video/ 7a537f578d808e67a3c6fd818a44a5c4/	WordPad для Windows 11 Jel 10.01.2026 WordPad для Windows 11 — это приложение, которое восстанавливает классический текстовый редактор WordPad в операционной системе Windows 11. После того как Microsoft исключила WordPad из. . .	Classic Notepad for Windows 11 Jel 10.01.2026 Old Classic Notepad for Windows 11 Приложение для Windows 11, позволяющее пользователям вернуть классическую версию текстового редактора «Блокнот» из Windows 10. Программа предоставляет более. . .	Почему дизайн решает? Neotwalker 09.01.2026 В современном мире, где конкуренция за внимание потребителя достигла пика, дизайн становится мощным инструментом для успеха бренда. Это не просто красивый внешний вид продукта или сайта — это. . .
Модель микоризы: классовый агентный подход 3 anaschu 06.01.2026 aa0a7f55b50dd51c5ec569d2d10c54f6/ O1rJuneU_ls https:/ / vkvideo. ru/ video-115721503_456239114	Owen Logic: О недопустимости использования связки «аналоговый ПИД» + RegKZR ФедосеевПавел 06.01.2026 Owen Logic: О недопустимости использования связки «аналоговый ПИД» + RegKZR ВВЕДЕНИЕ Введу сокращения: аналоговый ПИД — ПИД регулятор с управляющим выходом в виде числа в диапазоне от 0% до. . .	Модель микоризы: классовый агентный подход 2 anaschu 06.01.2026 репозиторий https:/ / github. com/ shumilovas/ fungi ветка по-частям. коммит Create переделка под биомассу. txt вход sc, но sm считается внутри мицелия. кстати, обьем тоже должен там считаться. . . .	Расчёт токов в цепи постоянного тока igorrr37 05.01.2026 / * Дана цепь постоянного тока с сопротивлениями и источниками (напряжения, ЭДС и тока). Найти токи и напряжения во всех элементах. Программа составляет систему уравнений по 1 и 2 законам Кирхгофа и. . .

@Snowman8526 6 / 6 / 1 Регистрация: 26.09.2017 Сообщений: 68
	17.11.2017, 19:55 [ТС]
	shsv, спасибо большое за объяснение буду экспериментировать. 0

@Snowman8526 6 / 6 / 1 Регистрация: 26.09.2017 Сообщений: 68
	21.11.2017, 09:30 [ТС]
	shsv, Ваш пример помог но есть небольшая сложность. В одном цикле к списку который передаю обращений штук 60. Цикл выполняется примерно за пол секунды а то и быстрее. Получается что создавая 2-а процесса у меня работают 3-и процесса загружая 2-х ядерник под завязку. на 4-х ядерном в 5 процессов загрузка от 60% до 80% Предположил что слишком частое обращение к списку что до предела грузит канал связи процессов. Решил уменьшить нагрузку на запросах на изменение всего 2-а в цикле можно и пожертвовать Попробовал создать копию до создания процессов(arr2 = arr) и каждому передать не помогло. Так же попробовал создать после создания процесса. И пробовал создать копию с помощью модуля copy (arr2 = copy.copy(arr)) Может вы знаете метод как уменьшить нагрузку на канал связи? 0