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
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
from mpi4py import MPI
import mpiunittest as unittest
import arrayimpl
class BaseTestPack(object):
COMM = MPI.COMM_NULL
def testPackSize(self):
for array in arrayimpl.ArrayTypes:
for typecode, datatype in arrayimpl.TypeMap.items():
itemsize = array(0, typecode).itemsize
overhead = datatype.Pack_size(0, self.COMM)
for count in range(10):
pack_size = datatype.Pack_size(count, self.COMM)
self.assertEqual(pack_size - overhead, count*itemsize)
def testPackUnpack(self):
for array in arrayimpl.ArrayTypes:
for typecode1, datatype1 in arrayimpl.TypeMap.items():
for typecode2, datatype2 in arrayimpl.TypeMap.items():
for items in range(10):
# input and output arrays
iarray1 = array(range(items), typecode1)
iarray2 = array(range(items), typecode2)
oarray1 = array(items, typecode1, items)
oarray2 = array(items, typecode2, items)
# temp array for packing
size1 = datatype1.Pack_size(len(iarray1), self.COMM)
size2 = datatype2.Pack_size(len(iarray2), self.COMM)
tmpbuf = array(0, 'b', size1 + size2 + 1)
# pack input arrays
position = 0
position = datatype1.Pack(iarray1, tmpbuf, position, self.COMM)
position = datatype2.Pack(iarray2, tmpbuf, position, self.COMM)
# unpack output arrays
position = 0
position = datatype1.Unpack(tmpbuf, position, oarray1, self.COMM)
position = datatype2.Unpack(tmpbuf, position, oarray2, self.COMM)
# test
equal = array.allclose
self.assertTrue(equal(iarray1, oarray1))
self.assertTrue(equal(iarray2, oarray2))
EXT32 = 'external32'
class BaseTestPackExternal(object):
byteswap = False
skipdtype = []
def testPackSize(self):
for array in arrayimpl.ArrayTypes:
for typecode, datatype in arrayimpl.TypeMap.items():
itemsize = array(0, typecode).itemsize
overhead = datatype.Pack_external_size(EXT32, 0)
for count in range(10):
pack_size = datatype.Pack_external_size(EXT32, count)
real_size = pack_size - overhead
def testPackUnpackExternal(self):
for array in arrayimpl.ArrayTypes:
for typecode1, datatype1 in arrayimpl.TypeMap.items():
for typecode2, datatype2 in arrayimpl.TypeMap.items():
for items in range(1, 10):
if typecode1 in self.skipdtype: continue
if typecode2 in self.skipdtype: continue
# input and output arrays
if typecode1 == 'b':
iarray1 = array(127, typecode1, items)
else:
iarray1 = array(255, typecode1, items)
iarray2 = array(range(items), typecode2)
oarray1 = array(-1, typecode1, items)
oarray2 = array(-1, typecode2, items)
# temp array for packing
size1 = datatype1.Pack_external_size(EXT32, iarray1.size)
size2 = datatype2.Pack_external_size(EXT32, iarray2.size)
tmpbuf = array(0, 'b', size1 + size2 + 1)
# pack input arrays
position = 0
position = datatype1.Pack_external(EXT32, iarray1, tmpbuf, position)
position = datatype2.Pack_external(EXT32, iarray2, tmpbuf, position)
# unpack output arrays
position = 0
position = datatype1.Unpack_external(EXT32, tmpbuf, position, oarray1)
position = datatype2.Unpack_external(EXT32, tmpbuf, position, oarray2)
# test result
if self.byteswap:
oarray1 = oarray1.byteswap()
oarray2 = oarray2.byteswap()
#return
equal = array.allclose
self.assertTrue(equal(iarray1, oarray1))
self.assertTrue(equal(iarray2, oarray2))
class TestPackSelf(BaseTestPack, unittest.TestCase):
COMM = MPI.COMM_SELF
class TestPackWorld(BaseTestPack, unittest.TestCase):
COMM = MPI.COMM_SELF
class TestPackExternal(BaseTestPackExternal, unittest.TestCase):
pass
_name, _version = MPI.get_vendor()
if _name == 'Open MPI':
if _version < (1, 3, 0):
from sys import byteorder as sys_byteorder
if sys_byteorder == 'little':
BaseTestPackExternal.byteswap = True
elif _name in ('MPICH2', 'DeinoMPI'):
BaseTestPackExternal.skipdtype += ['l']
BaseTestPackExternal.skipdtype += ['d']
pass
else:
try:
MPI.BYTE.Pack_external_size(EXT32, 0)
except NotImplementedError:
del BaseTestPackExternal
del TestPackExternal
if __name__ == '__main__':
unittest.main()
|
