cpython/Lib/multiprocessing/sharedctypes.py at main · python/cpython · GitHub

Repository navigation

History

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

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

# Module which supports allocation of ctypes objects from shared memory

# multiprocessing/sharedctypes.py

# Licensed to PSF under a Contributor Agreement.

import ctypes

import weakref

from . import heap

from . import get_context

from .context import reduction, assert_spawning

_ForkingPickler = reduction.ForkingPickler

__all__ = ['RawValue', 'RawArray', 'Value', 'Array', 'copy', 'synchronized']

typecode_to_type = {

'c': ctypes.c_char, 'u': ctypes.c_wchar,

'b': ctypes.c_byte, 'B': ctypes.c_ubyte,

'h': ctypes.c_short, 'H': ctypes.c_ushort,

'i': ctypes.c_int, 'I': ctypes.c_uint,

'l': ctypes.c_long, 'L': ctypes.c_ulong,

'q': ctypes.c_longlong, 'Q': ctypes.c_ulonglong,

'f': ctypes.c_float, 'd': ctypes.c_double

}

def _new_value(type_):

try:

size = ctypes.sizeof(type_)

except TypeError as e:

raise TypeError("bad typecode (must be a ctypes type or one of "

"c, b, B, u, h, H, i, I, l, L, q, Q, f or d)") from e

wrapper = heap.BufferWrapper(size)

return rebuild_ctype(type_, wrapper, None)

def RawValue(typecode_or_type, *args):

'''

Returns a ctypes object allocated from shared memory

'''

type_ = typecode_to_type.get(typecode_or_type, typecode_or_type)

obj = _new_value(type_)

ctypes.memset(ctypes.addressof(obj), 0, ctypes.sizeof(obj))

obj.__init__(*args)

return obj

def RawArray(typecode_or_type, size_or_initializer):

'''

Returns a ctypes array allocated from shared memory

'''

type_ = typecode_to_type.get(typecode_or_type, typecode_or_type)

if isinstance(size_or_initializer, int):

type_ = type_ * size_or_initializer

obj = _new_value(type_)

ctypes.memset(ctypes.addressof(obj), 0, ctypes.sizeof(obj))

return obj

else:

type_ = type_ * len(size_or_initializer)

result = _new_value(type_)

result.__init__(*size_or_initializer)

return result

def Value(typecode_or_type, *args, lock=True, ctx=None):

'''

Return a synchronization wrapper for a Value

'''

obj = RawValue(typecode_or_type, *args)

if lock is False:

return obj

if lock in (True, None):

ctx = ctx or get_context()

lock = ctx.RLock()

if not hasattr(lock, 'acquire'):

raise AttributeError("%r has no method 'acquire'" % lock)

return synchronized(obj, lock, ctx=ctx)

def Array(typecode_or_type, size_or_initializer, *, lock=True, ctx=None):

'''

Return a synchronization wrapper for a RawArray

'''

obj = RawArray(typecode_or_type, size_or_initializer)

if lock is False:

return obj

if lock in (True, None):

ctx = ctx or get_context()

lock = ctx.RLock()

if not hasattr(lock, 'acquire'):

raise AttributeError("%r has no method 'acquire'" % lock)

return synchronized(obj, lock, ctx=ctx)

def copy(obj):

new_obj = _new_value(type(obj))

ctypes.pointer(new_obj)[0] = obj

return new_obj

def synchronized(obj, lock=None, ctx=None):

assert not isinstance(obj, SynchronizedBase), 'object already synchronized'

ctx = ctx or get_context()

if isinstance(obj, ctypes._SimpleCData):

return Synchronized(obj, lock, ctx)

elif isinstance(obj, ctypes.Array):

if obj._type_ is ctypes.c_char:

return SynchronizedString(obj, lock, ctx)

return SynchronizedArray(obj, lock, ctx)

else:

cls = type(obj)

try:

scls = class_cache[cls]

except KeyError:

names = [field[0] for field in cls._fields_]

d = {name: make_property(name) for name in names}

classname = 'Synchronized' + cls.__name__

scls = class_cache[cls] = type(classname, (SynchronizedBase,), d)

return scls(obj, lock, ctx)

# Functions for pickling/unpickling

def reduce_ctype(obj):

assert_spawning(obj)

if isinstance(obj, ctypes.Array):

return rebuild_ctype, (obj._type_, obj._wrapper, obj._length_)

else:

return rebuild_ctype, (type(obj), obj._wrapper, None)

def rebuild_ctype(type_, wrapper, length):

if length is not None:

type_ = type_ * length

_ForkingPickler.register(type_, reduce_ctype)

buf = wrapper.create_memoryview()

obj = type_.from_buffer(buf)

obj._wrapper = wrapper

return obj

# Function to create properties

def make_property(name):

try:

return prop_cache[name]

except KeyError:

d = {}

exec(template % ((name,)*7), d)

prop_cache[name] = d[name]

return d[name]

template = '''

def get%s(self):

self.acquire()

try:

return self._obj.%s

finally:

self.release()

def set%s(self, value):

self.acquire()

try:

self._obj.%s = value

finally:

self.release()

%s = property(get%s, set%s)

'''

prop_cache = {}

class_cache = weakref.WeakKeyDictionary()

# Synchronized wrappers

class SynchronizedBase(object):

def __init__(self, obj, lock=None, ctx=None):

self._obj = obj

if lock:

self._lock = lock

else:

ctx = ctx or get_context(force=True)

self._lock = ctx.RLock()

self.acquire = self._lock.acquire

self.release = self._lock.release

def __enter__(self):

return self._lock.__enter__()

def __exit__(self, *args):

return self._lock.__exit__(*args)

def __reduce__(self):

assert_spawning(self)

return synchronized, (self._obj, self._lock)

def get_obj(self):

return self._obj

def get_lock(self):

return self._lock

def __repr__(self):

return '<%s wrapper for %s>' % (type(self).__name__, self._obj)

class Synchronized(SynchronizedBase):

value = make_property('value')

class SynchronizedArray(SynchronizedBase):

def __len__(self):

return len(self._obj)

def __getitem__(self, i):

with self:

return self._obj[i]

def __setitem__(self, i, value):

with self:

self._obj[i] = value

def __getslice__(self, start, stop):

with self:

return self._obj[start:stop]

def __setslice__(self, start, stop, values):

with self:

self._obj[start:stop] = values

class SynchronizedString(SynchronizedArray):

value = make_property('value')

raw = make_property('raw')

cpython/Lib/multiprocessing/sharedctypes.py at main · python/cpython · GitHub

Repository navigation

Expand file tree

Breadcrumbs

sharedctypes.py

Latest commit

History

Breadcrumbs

sharedctypes.py

File metadata and controls