blocks|key|747925|text|Array对象有一个与之关联的get_obj()方法，该方法返回ctypes数组，该数组提供一个buffer接口。我认为下面的方法应该行得通。|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|747926|from+multiprocessing+import+Process,+Array
import+scipy
import+numpy

def+f(a):
++++a[0]+=+-a[0]

if+__name__+==+'__main__':
++++#+Create+the+array
++++N+=+int(10)
++++unshared_arr+=+scipy.rand(N)
++++a+=+Array('d',+unshared_arr)
++++print+"Originally,+the+first+two+elements+of+arr+=+%25s"%25(a[:2])

++++#+Create,+start,+and+finish+the+child+process
++++p+=+Process(target=f,+args=(a,))
++++p.start()
++++p.join()

++++#+Print+out+the+changed+values
++++print+"Now,+the+first+two+elements+of+arr+=+%25s"%25a[:2]

++++b+=+numpy.frombuffer(a.get_obj())

++++b[0]+=+10.0
++++print+a[0]|code-block|syntax|javascript|747927|运行时，这会打印出a的第一个元素，现在是10.0，显示a和b只是同一内存中的两个视图。|747928|为了确保它仍然是多处理器安全的，我相信您必须使用存在于Array对象、a及其内置锁上的acquire和release方法，以确保安全地访问它(尽管我不是多处理器模块的专家)。|747929|entityMap^0|0|5|F|9|0|0|9|1|R|1|T|1|0|R|5|Z|1|17|7|1F|7|0^^$0|@$1|2|3|4|5|6|7|Q|8|@$9|R|A|S|B|C]|$9|T|A|U|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|V|8|@]|D|@]|E|$I|J]]|$1|K|3|L|5|6|7|W|8|@$9|X|A|Y|B|C]|$9|Z|A|10|B|C]|$9|11|A|12|B|C]]|D|@]|E|$]]|$1|M|3|N|5|6|7|13|8|@$9|14|A|15|B|C]|$9|16|A|17|B|C]|$9|18|A|19|B|C]|$9|1A|A|1B|B|C]]|D|@]|E|$]]|$1|O|3|-4|5|6|7|1C|8|@]|D|@]|E|$]]]|P|$]]

The <code>Array</code> object has a <code>get_obj()</code> method associated with it, which returns the ctypes array which presents a buffer interface. I think the following should work...

<pre><code>from multiprocessing import Process, Array
import scipy
import numpy

def f(a):
 a[0] = -a[0]

if __name__ == '__main__':
 # Create the array
 N = int(10)
 unshared_arr = scipy.rand(N)
 a = Array('d', unshared_arr)
 print "Originally, the first two elements of arr = %s"%(a[:2])

 # Create, start, and finish the child process
 p = Process(target=f, args=(a,))
 p.start()
 p.join()

 # Print out the changed values
 print "Now, the first two elements of arr = %s"%a[:2]

 b = numpy.frombuffer(a.get_obj())

 b[0] = 10.0
 print a[0]
</code></pre>

When run, this prints out the first element of <code>a</code> now being 10.0, showing <code>a</code> and <code>b</code> are just two views into the same memory.

In order to make sure it is still multiprocessor safe, I believe you will have to use the <code>acquire</code> and <code>release</code> methods that exist on the <code>Array</code> object, <code>a</code>, and its built in lock to make sure its all safely accessed (though I'm not an expert on the multiprocessor module).

blocks|key|531589|text|添加到@unutbu(不再提供)和@Henry+Gomersall的答案中。您可以在需要时使用shared_arr.get_lock()同步访问：|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|531590|shared_arr+=+mp.Array(ctypes.c_double,+N)
#+...
def+f(i):+#+could+be+anything+numpy+accepts+as+an+index+such+another+numpy+array
++++with+shared_arr.get_lock():+#+synchronize+access
++++++++arr+=+np.frombuffer(shared_arr.get_obj())+#+no+data+copying
++++++++arr[i]+=+-arr[i]|code-block|syntax|javascript|531591|示例|531592|import+ctypes
import+logging
import+multiprocessing+as+mp

from+contextlib+import+closing

import+numpy+as+np

info+=+mp.get_logger().info

def+main():
++++logger+=+mp.log_to_stderr()
++++logger.setLevel(logging.INFO)

++++#+create+shared+array
++++N,+M+=+100,+11
++++shared_arr+=+mp.Array(ctypes.c_double,+N)
++++arr+=+tonumpyarray(shared_arr)

++++#+fill+with+random+values
++++arr[:]+=+np.random.uniform(size=N)
++++arr_orig+=+arr.copy()

++++#+write+to+arr+from+different+processes
++++with+closing(mp.Pool(initializer=init,+initargs=(shared_arr,)))+as+p:
++++++++#+many+processes+access+the+same+slice
++++++++stop_f+=+N+//+10
++++++++p.map_async(f,+[slice(stop_f)]*M)

++++++++#+many+processes+access+different+slices+of+the+same+array
++++++++assert+M+%25+2+#+odd
++++++++step+=+N+//+10
++++++++p.map_async(g,+[slice(i,+i+%2B+step)+for+i+in+range(stop_f,+N,+step)])
++++p.join()
++++assert+np.allclose(((-1)**M)*tonumpyarray(shared_arr),+arr_orig)

def+init(shared_arr_):
++++global+shared_arr
++++shared_arr+=+shared_arr_+#+must+be+inherited,+not+passed+as+an+argument

def+tonumpyarray(mp_arr):
++++return+np.frombuffer(mp_arr.get_obj())

def+f(i):
++++"""synchronized."""
++++with+shared_arr.get_lock():+#+synchronize+access
++++++++g(i)

def+g(i):
++++"""no+synchronization."""
++++info("start+%25s"+%25+(i,))
++++arr+=+tonumpyarray(shared_arr)
++++arr[i]+=+-1+*+arr[i]
++++info("end+++%25s"+%25+(i,))

if+__name__+==+'__main__':
++++mp.freeze_support()
++++main()|531593|如果您不需要同步访问，或者您创建了自己的锁，那么mp.Array()是不必要的。在这种情况下，您可以使用mp.sharedctypes.RawArray。|531594|entityMap^0|1B|L|0|0|0|0|O|A|1G|O|0^^$0|@$1|2|3|4|5|6|7|S|8|@$9|T|A|U|B|C]]|D|@]|E|$]]|$1|F|3|G|5|H|7|V|8|@]|D|@]|E|$I|J]]|$1|K|3|L|5|6|7|W|8|@]|D|@]|E|$]]|$1|M|3|N|5|H|7|X|8|@]|D|@]|E|$I|J]]|$1|O|3|P|5|6|7|Y|8|@$9|Z|A|10|B|C]|$9|11|A|12|B|C]]|D|@]|E|$]]|$1|Q|3|-4|5|6|7|13|8|@]|D|@]|E|$]]]|R|$]]

To add to @unutbu's (not available anymore) and @Henry Gomersall's answers. You could use <code>shared_arr.get_lock()</code> to synchronize access when needed:

<pre><code>shared_arr = mp.Array(ctypes.c_double, N)
# ...
def f(i): # could be anything numpy accepts as an index such another numpy array
 with shared_arr.get_lock(): # synchronize access
 arr = np.frombuffer(shared_arr.get_obj()) # no data copying
 arr[i] = -arr[i]
</code></pre>

<h3>Example</h3>

<pre><code>import ctypes
import logging
import multiprocessing as mp

from contextlib import closing

import numpy as np

info = mp.get_logger().info

def main():
 logger = mp.log_to_stderr()
 logger.setLevel(logging.INFO)

 # create shared array
 N, M = 100, 11
 shared_arr = mp.Array(ctypes.c_double, N)
 arr = tonumpyarray(shared_arr)

 # fill with random values
 arr[:] = np.random.uniform(size=N)
 arr_orig = arr.copy()

 # write to arr from different processes
 with closing(mp.Pool(initializer=init, initargs=(shared_arr,))) as p:
 # many processes access the same slice
 stop_f = N // 10
 p.map_async(f, [slice(stop_f)]*M)

 # many processes access different slices of the same array
 assert M % 2 # odd
 step = N // 10
 p.map_async(g, [slice(i, i + step) for i in range(stop_f, N, step)])
 p.join()
 assert np.allclose(((-1)**M)*tonumpyarray(shared_arr), arr_orig)

def init(shared_arr_):
 global shared_arr
 shared_arr = shared_arr_ # must be inherited, not passed as an argument

def tonumpyarray(mp_arr):
 return np.frombuffer(mp_arr.get_obj())

def f(i):
 """synchronized."""
 with shared_arr.get_lock(): # synchronize access
 g(i)

def g(i):
 """no synchronization."""
 info("start %s" % (i,))
 arr = tonumpyarray(shared_arr)
 arr[i] = -1 * arr[i]
 info("end %s" % (i,))

if __name__ == '__main__':
 mp.freeze_support()
 main()
</code></pre>

If you don't need synchronized access or you create your own locks then <code>mp.Array()</code> is unnecessary. You could use <code>mp.sharedctypes.RawArray</code> in this case.

blocks|key|531644|text|您可以使用sharedmem模块：https://bitbucket.org/cleemesser/numpy-sharedmem|type|unstyled|depth|inlineStyleRanges|offset|length|style|CODE|entityRanges|data|531645|下面是您的原始代码，这一次使用了行为类似于NumPy数组的共享内存(注意调用NumPy+sum()函数的最后一条额外语句)：|531646|from+multiprocessing+import+Process
import+sharedmem
import+scipy

def+f(a):
++++a[0]+=+-a[0]

if+__name__+==+'__main__':
++++#+Create+the+array
++++N+=+int(10)
++++unshared_arr+=+scipy.rand(N)
++++arr+=+sharedmem.empty(N)
++++arr[:]+=+unshared_arr.copy()
++++print+"Originally,+the+first+two+elements+of+arr+=+%25s"%25(arr[:2])

++++#+Create,+start,+and+finish+the+child+process
++++p+=+Process(target=f,+args=(arr,))
++++p.start()
++++p.join()

++++#+Print+out+the+changed+values
++++print+"Now,+the+first+two+elements+of+arr+=+%25s"%25arr[:2]

++++#+Perform+some+NumPy+operation
++++print+arr.sum()|code-block|syntax|javascript|531647|entityMap|0|LINK|mutability|MUTABLE|url|https://bitbucket.org/cleemesser/numpy-sharedmem^0|5|9|H|1C|0|0|18|5|0|0^^$0|@$1|2|3|4|5|6|7|U|8|@$9|V|A|W|B|C]]|D|@$9|X|A|Y|1|Z]]|E|$]]|$1|F|3|G|5|6|7|10|8|@$9|11|A|12|B|C]]|D|@]|E|$]]|$1|H|3|I|5|J|7|13|8|@]|D|@]|E|$K|L]]|$1|M|3|-4|5|6|7|14|8|@]|D|@]|E|$]]]|N|$O|$5|P|Q|R|E|$S|T]]]]

You can use the <code>sharedmem</code> module: <a href="https://bitbucket.org/cleemesser/numpy-sharedmem">https://bitbucket.org/cleemesser/numpy-sharedmem</a>

Here's your original code then, this time using shared memory that behaves like a NumPy array (note the additional last statement calling a NumPy <code>sum()</code> function):

<pre><code>from multiprocessing import Process
import sharedmem
import scipy

def f(a):
 a[0] = -a[0]

if __name__ == '__main__':
 # Create the array
 N = int(10)
 unshared_arr = scipy.rand(N)
 arr = sharedmem.empty(N)
 arr[:] = unshared_arr.copy()
 print "Originally, the first two elements of arr = %s"%(arr[:2])

 # Create, start, and finish the child process
 p = Process(target=f, args=(arr,))
 p.start()
 p.join()

 # Print out the changed values
 print "Now, the first two elements of arr = %s"%arr[:2]

 # Perform some NumPy operation
 print arr.sum()
</code></pre>

blocks|key|531700|text|我已经编写了一个小的python模块，它使用POSIX共享内存在python解释器之间共享numpy数组。也许你会发现它很方便。|type|unstyled|depth|inlineStyleRanges|entityRanges|data|531701|https://pypi.python.org/pypi/SharedArray|offset|length|531702|下面是它的工作原理：|531703|import+numpy+as+np
import+SharedArray+as+sa

#+Create+an+array+in+shared+memory
a+=+sa.create("test1",+10)

#+Attach+it+as+a+different+array.+This+can+be+done+from+another
#+python+interpreter+as+long+as+it+runs+on+the+same+computer.
b+=+sa.attach("test1")

#+See+how+they+are+actually+sharing+the+same+memory+block
a[0]+=+42
print(b[0])

#+Destroying+a+does+not+affect+b.
del+a
print(b[0])

#+See+how+"test1"+is+still+present+in+shared+memory+even+though+we
#+destroyed+the+array+a.
sa.list()

#+Now+destroy+the+array+"test1"+from+memory.
sa.delete("test1")

#+The+array+b+is+not+affected,+but+once+you+destroy+it+then+the
#+data+are+lost.
print(b[0])|code-block|syntax|javascript|531704|entityMap|0|LINK|mutability|MUTABLE|url^0|0|0|14|0|0|0|0^^$0|@$1|2|3|4|5|6|7|T|8|@]|9|@]|A|$]]|$1|B|3|C|5|6|7|U|8|@]|9|@$D|V|E|W|1|X]]|A|$]]|$1|F|3|G|5|6|7|Y|8|@]|9|@]|A|$]]|$1|H|3|I|5|J|7|Z|8|@]|9|@]|A|$K|L]]|$1|M|3|-4|5|6|7|10|8|@]|9|@]|A|$]]]|N|$O|$5|P|Q|R|A|$S|C]]]]

I've written a small python module that uses POSIX shared memory to share numpy arrays between python interpreters. Maybe you will find it handy.

<a href="https://pypi.python.org/pypi/SharedArray" rel="noreferrer">https://pypi.python.org/pypi/SharedArray</a>

Here's how it works:

<pre><code>import numpy as np
import SharedArray as sa

# Create an array in shared memory
a = sa.create("test1", 10)

# Attach it as a different array. This can be done from another
# python interpreter as long as it runs on the same computer.
b = sa.attach("test1")

# See how they are actually sharing the same memory block
a[0] = 42
print(b[0])

# Destroying a does not affect b.
del a
print(b[0])

# See how "test1" is still present in shared memory even though we
# destroyed the array a.
sa.list()

# Now destroy the array "test1" from memory.
sa.delete("test1")

# The array b is not affected, but once you destroy it then the
# data are lost.
print(b[0])
</code></pre>

I would like to use a numpy array in shared memory for use with the multiprocessing module. The difficulty is using it like a numpy array, and not just as a ctypes array.

<pre><code>from multiprocessing import Process, Array
import scipy

def f(a):
 a[0] = -a[0]

if __name__ == '__main__':
 # Create the array
 N = int(10)
 unshared_arr = scipy.rand(N)
 arr = Array('d', unshared_arr)
 print "Originally, the first two elements of arr = %s"%(arr[:2])

 # Create, start, and finish the child processes
 p = Process(target=f, args=(arr,))
 p.start()
 p.join()

 # Printing out the changed values
 print "Now, the first two elements of arr = %s"%arr[:2]
</code></pre>

This produces output such as:

<pre><code>Originally, the first two elements of arr = [0.3518653236697369, 0.517794725524976]
Now, the first two elements of arr = [-0.3518653236697369, 0.517794725524976]
</code></pre>

The array can be accessed in a ctypes manner, e.g. <code>arr[i]</code> makes sense. However, it is not a numpy array, and I cannot perform operations such as <code>-1*arr</code>, or <code>arr.sum()</code>. I suppose a solution would be to convert the ctypes array into a numpy array. However (besides not being able to make this work), I don't believe it would be shared anymore.

It seems there would be a standard solution to what has to be a common problem.

Use numpy array in shared memory for multiprocessing

解释器

我想在共享内存中使用numpy数组，以便与多处理模块一起使用。困难之处在于将其用作numpy数组，而不仅仅是ctypes数组。from multiprocessing import Process, Arrayimport scipydef f(a):    a[0] = -a[0]if __name__ == '__...

问在共享内存中使用numpy数组进行多处理
EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问在共享内存中使用numpy数组进行多处理EN

社区

活动

资源

关于

腾讯云开发者

热门产品

热门推荐

更多推荐

问在共享内存中使用numpy数组进行多处理
EN