Sending Numpy structured arrays with mpi4py

[GV]

Hi nice people,

I'm sending back from MPI workers to root a number of large python dictionaries (~1.5 million keys each). These dictionaries come back in the form of a list, as in

list_of_dictionaries = [{'68eef0f0-9385-11e3-baa8-0800200c9a66' : 1287612, '533c5680-938a-11e3-baa8-0800200c9a66': 98234, . . . }, {'66339ecc-db8d-440a-b015-897db84ffb6d': 188892, '7ed7847d-63fa-4cc1-9ddd-46e07978888a': 234324, . . . }, . . .]

I'm gathering everything back in root by using: 

d = comm.gather(list_of_dictionaries, root=0)

This approach works good for smaller subsets of the dictionaries, however for the whole set I'm getting the error:

Traceback (most recent call last):

  File "4_SCRIPTS/MPI/11_ranking_artists_album_tracks_DUP.py", line 138, in <module>

    dictionary_list_gathered = comm.gather(dict_this_rank, root=0)

  File "Comm.pyx", line 869, in mpi4py.MPI.Comm.gather (src/mpi4py.MPI.c:73266)

  File "pickled.pxi", line 614, in mpi4py.MPI.PyMPI_gather (src/mpi4py.MPI.c:33592)

  File "pickled.pxi", line 146, in mpi4py.MPI._p_Pickle.allocv (src/mpi4py.MPI.c:28517)

  File "pickled.pxi", line 95, in mpi4py.MPI._p_Pickle.alloc (src/mpi4py.MPI.c:27832)

SystemError: Negative size passed to PyString_FromStringAndSize

I searched into this forum and realized that Lisandro Dalcin posted that he recommends not using gather() for large messages, but to use Gather(). I implemented a toy example using Numpy structured arrays for not sending an actual Python dictionary, but a list:

import mpi4py.MPI as mpi

import numpy as np

comm = mpi.COMM_WORLD

rank = comm.Get_rank()

x = np.array(('ba78f30e-9339-4a98-9851-abee0ef60c36', 102809), dtype=('a36,i4'))

if rank == 0: 

x_all = np.zeros((4,), dtype = x.dtype)

else:

x_all = None 

comm.Gather(x, x_all, root=0) 

if rank == 0: 

print x_all

But I'm getting the error:

Traceback (most recent call last):

  File "mpi_Gather_test.py", line 19, in <module>

    comm.Gather(x, x_all, root=0)

  File "Comm.pyx", line 415, in mpi4py.MPI.Comm.Gather (src/mpi4py.MPI.c:66916)

  File "message.pxi", line 429, in mpi4py.MPI._p_msg_cco.for_gather (src/mpi4py.MPI.c:23582)

  File "message.pxi", line 369, in mpi4py.MPI._p_msg_cco.for_cco_recv (src/mpi4py.MPI.c:23059)

  File "message.pxi", line 355, in mpi4py.MPI._p_msg_cco.for_cco_send (src/mpi4py.MPI.c:22959)

  File "message.pxi", line 111, in mpi4py.MPI.message_simple (src/mpi4py.MPI.c:20516)

  File "message.pxi", line 58, in mpi4py.MPI.message_basic (src/mpi4py.MPI.c:19723)

KeyError: 'T{36s:f0:i:f1:}'

MPI returns the error when the Numpy structured array has mixed datatypes (in this case: [('f0', 'S36'), ('f1', '<i4')]).

My questions are: is there anyway of overcoming the limit of the pickle serialized objects? As I guess the answer is no: Is there any way to send a structured array with mixed datatypes using mpi4py? 

Thank you very much,

G.

[Lev Givon]

Received from GV on Tue, Feb 11, 2014 at 09:31:24PM EST:

> Hi nice people,
>
> I'm sending back from MPI workers to root a number of large python
> dictionaries (~1.5 million keys each). These dictionaries come back in the
> form of a list, as in

(snip)

> My questions are: is there anyway of overcoming the limit of the pickle
> serialized objects? As I guess the answer is no: Is there any way to send a
> structured array with mixed datatypes using mpi4py?

Recalling an approach Lisandro once mentioned on the list [1], you may wish to try
overriding the pickle serialization routines used by mpi4py with some other
serializer such as dill [2] or the msgpack Python bindings [3].

[1] https://groups.google.com/forum/#!msg/mpi4py/1fd4FwdgpWY/019QT3S-HfkJ
[2] https://pypi.python.org/pypi/dill
[3] https://pypi.python.org/pypi/msgpack-python
--
Lev Givon
Bionet Group
http://www.columbia.edu/~lev/
http://lebedov.github.io/

[Lisandro Dalcin]

On 12 February 2014 05:31, GV <gab...@vigliensoni.com> wrote:
> My questions are: is there anyway of overcoming the limit of the pickle
> serialized objects?

Overcoming this limit is usually not easy, it is related to MPI using
the "int" C datatype to represent message counts, thus hitting the 2G
entries limit. Fixing this issue would be possible, requires some
implementation work on mpi4py's side, but perhaps at the expense of
extra communications. In short: do not expect this to be fixed in the
near future.

> As I guess the answer is no: Is there any way to send a
> structured array with mixed datatypes using mpi4py?

You have to use user-defined MPI datatypes. Here you have an example,
I've used a structured datatype with 3 fields to make is more clear
what block lengths and displacements are for
MPI.Datatype.Create_struct().

from mpi4py import MPI
import numpy as np

npydt = np.dtype('S36,i4,f8') # NumPy structured datatype

mpidt = MPI.Datatype.Create_struct( # MPI user-defined datatype
[36, 1, 1], # block lengths
[0, 36, 40], # displacements in bytes
[MPI.CHAR, MPI.INT, MPI.DOUBLE], # MPI datatypes
).Commit() # don't forget to call Commit() after creation !!!

comm = MPI.COMM_WORLD
rank = comm.Get_rank()
x = np.array(('ba78f30e-9339-4a98-9851-abee0ef60c36', 102809, 3.14),
dtype=npydt)
if rank == 0:
x_all = np.zeros((comm.size,), dtype = x.dtype)
else:
x_all = None
comm.Gather([x, mpidt], [x_all, mpidt], root=0) # specify messages as
[numpy_array, mpi_datatype]

if rank == 0:
print x_all

--
Lisandro Dalcin
---------------
CIMEC (UNL/CONICET)
Predio CONICET-Santa Fe
Colectora RN 168 Km 472, Paraje El Pozo
3000 Santa Fe, Argentina
Tel: +54-342-4511594 (ext 1016)
Tel/Fax: +54-342-4511169

[GV]

Thanks Lisandro, that was exactly was I was looking for.

G.

[GV]

Hi Lisandro,

I'm still having a minor problem when trying to allocate memory for the user-defined structured datatype. As you suggested, now I'm doing:

comm = MPI.COMM_WORLD 

rank = comm.Get_rank() 

size = comm.size

# NumPy structured datatype 

npydt = np.dtype({'mbid': ('S36', 0), 'freq': (np.int64, 36)})

npydt_size = npydt.itemsize

print 'Numpy datatype size: ', npydt_size

# MPI structured datatype 

mpidt = MPI.Datatype.Create_struct( 

    [36, 1], # 36-byte char

    [0, 36], # 8-byte integer

    [MPI.CHAR, MPI.LONG], 

).Commit() 

print 'MPI datatype size: ', mpidt.Get_size()

# Generating data in each processor

no_data = 100

x = np.array([(random.choice("abcdefgh"), random.randint(0, 100)) for e in xrange(no_data)], dtype = npydt)

sendmsg = x.size 

x_len = comm.reduce(sendmsg, op=MPI.SUM, root = 0)

if rank == 0:

    # The global array size has to be multiple of number of processes 

    x_all = np.zeros( size * ( x_len + x_len % size ), dtype = npydt) 

else:

    x_all = None

comm.Gather(sendbuf = [x, mpidt], recvbuf = [x_all, mpidt], root = 0)

However, when running the code I'm getting:

Numpy datatype size:  44

MPI datatype size:  44

Traceback (most recent call last):

  File "datatype_checker.py", line 45, in <module>

    comm.Gather(sendbuf = [x, mpidt], recvbuf = [x_all, mpidt], root = 0)

  File "Comm.pyx", line 415, in mpi4py.MPI.Comm.Gather (src/mpi4py.MPI.c:66916)

  File "message.pxi", line 420, in mpi4py.MPI._p_msg_cco.for_gather (src/mpi4py.MPI.c:23507)

  File "message.pxi", line 369, in mpi4py.MPI._p_msg_cco.for_cco_recv (src/mpi4py.MPI.c:23059)

  File "message.pxi", line 148, in mpi4py.MPI.message_simple (src/mpi4py.MPI.c:20934)

ValueError: message: cannot guess count, buffer length 4400 is not a multiple of datatype extent 48 (lb:0, ub:48)

Which is very strange, because the Numpy and MPI datatype sizes are 44 bytes, as were printed at the beginning of the code.

Thank you very much,

GV

[Lisandro Dalcin]

On 18 February 2014 21:23, GV <vigli...@gmail.com> wrote:
> mpidt = MPI.Datatype.Create_struct(
> [36, 1], # 36-byte char
> [0, 36], # 8-byte integer
> [MPI.CHAR, MPI.LONG],
> ).Commit()
> print 'MPI datatype size: ', mpidt.Get_size()

If you print mpidt.extent, it is 48. This is related to the address of
the 64 bit integer not being multiple of 8 (IOW, an alignment issue).
I don't remember right now what the MPI standard says about this.

A way to "fix" this issue is to add padding between "char[36]" and
"long" as I show below, and adjust the NumPy datatype accordingly. Of
course, you "waste" 4 bytes of memory per entry in your arrays.

mpidt = MPI.Datatype.Create_struct(
[36, 4, 1],
[0, 36, 40],
[MPI.CHAR, MPI.BYTE, MPI.LONG],
).Commit()

[Lisandro Dalcin]

The definitive "solution" for this issue is to "fix" the extent using
Create_resized(), as shown below.

from mpi4py import MPI
import numpy as np

npydt = np.dtype('S36,i8')

mpidt_aux = MPI.Datatype.Create_struct(
[36, 1], # block lengths
[0, 36], # displacements in bytes
[MPI.CHAR, MPI.LONG], # MPI datatypes
)
mpidt = mpidt_aux.Create_resized(0, 44).Commit()
mpidt_aux.Free()

comm = MPI.COMM_WORLD
rank = comm.Get_rank()

x = np.array(('ba78f30e-9339-4a98-9851-abee0ef60c36', 102809), dtype=npydt)

if rank == 0:
x_all = np.zeros((comm.size,), dtype = x.dtype)
else:
x_all = None
comm.Gather([x, mpidt], [x_all, mpidt], root=0)

if rank == 0:
print x_all

[ivo wolff gersberg]

Caro Lisandro,

Muito obrigado. Sua ajuda foi valiosa. 

Grato pela gentileza,

Ivo Wolff Gersberg

COPPE/Universidade Federal do Rio de Janeiro

Brasil