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...

Новые блоги и статьи Все статьи Все блоги /
SDL3 для Web (WebAssembly): Синхронизация спрайтов SDL3 и тел Box2D 8Observer8 04.03.2026 Содержание блога Финальная демка в браузере. Итоговый код: finish-sync-physics-sprites-sdl3-c. zip На первой гифке отладочные линии отключены, а на второй включены:. . .	SDL3 для Web (WebAssembly): Идентификация объектов на Box2D v3 - использование userData и событий коллизий 8Observer8 02.03.2026 Содержание блога Финальная демка в браузере. Итоговый код: finish-collision-events-sdl3-c. zip https:/ / www. cyberforum. ru/ blog_attachment. php?attachmentid=11680&d=1772460536 Одним из. . .	Реалии Hrethgir 01.03.2026 Нет, я не закончил до сих пор симулятор. Эта задача сложнее. Не получилось уйти в плавсостав, но оно и к лучшему, возможно. Точнее получалось - но сварщиком в палубную команду, а это значит, в моём. . .	Ритм жизни kumehtar 27.02.2026 Иногда приходится жить в ритме, где дел становится всё больше, а вовлечения в происходящее — всё меньше. Плотный график не даёт вниманию закрепиться ни на одном событии. Утро начинается с быстрых,. . .
SDL3 для Web (WebAssembly): Сборка библиотек: SDL3, Box2D, FreeType, SDL3_ttf, SDL3_mixer и SDL3_image из исходников с помощью CMake и Emscripten 8Observer8 27.02.2026 Недавно вышла версия 3. 4. 2 библиотеки SDL3. На странице официальной релиза доступны исходники, готовые DLL (для x86, x64, arm64), а также библиотеки для разработки под Android, MinGW и Visual Studio. . . .	SDL3 для Web (WebAssembly): Реализация движения на Box2D v3 - трение и коллизии с повёрнутыми стенами 8Observer8 20.02.2026 Содержание блога Box2D позволяет легко создать главного героя, который не проходит сквозь стены и перемещается с заданным трением о препятствия, которые можно располагать под углом, как верхнее. . .	Конвертировать закладки radiotray-ng в m3u-плейлист damix 19.02.2026 Это можно сделать скриптом для PowerShell. Использование . \СonvertRadiotrayToM3U. ps1 <path_to_bookmarks. json> Рядом с файлом bookmarks. json появится файл bookmarks. m3u с результатом. # Check if. . .	Семь CDC на одном интерфейсе: 5 U[S]ARTов, 1 CAN и 1 SSI Eddy_Em 18.02.2026 Постепенно допиливаю свою "многоинтерфейсную плату". Выглядит вот так: https:/ / www. cyberforum. ru/ blog_attachment. php?attachmentid=11617&stc=1&d=1771445347 Основана на STM32F303RBT6. На борту пять. . .

@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