multiprocessing.Value.value needs to be defined

[remdragon]

The following sample code:

import multiprocessing as mp
x = mp.Value( 'i', 0 )
print( f'{x.value=}' )
x.value = 1
print( f'{x.value=}' )

Produces this output when run:

x.value=0
x.value=1

But mypy complains:

mypy_bug_multiprocessing_value.py:3: error: "SynchronizedBase[Any]" has no attribute "value"  [attr-defined]
mypy_bug_multiprocessing_value.py:4: error: "SynchronizedBase[Any]" has no attribute "value"  [attr-defined]
mypy_bug_multiprocessing_value.py:5: error: "SynchronizedBase[Any]" has no attribute "value"  [attr-defined]

Python version: Python 3.10.9 (tags/v3.10.9:1dd9be6, Dec 6 2022, 20:01:21) [MSC v.1934 64 bit (AMD64)] on win32
mypy version: mypy==1.3.0

[Daverball]

I've tripped over this recently myself, the two levels of indirection between SynchronizedBase (and its subtypes), ctypes and python types makes multiprocessing.Value kind of a pain to annotate correctly, the value field only exists on Synchronized and SynchronizedString which are only used for simple ctypes, but technically you could pass in pretty much any arbitrarily complex ctype with a custom layout and make it a synchronized value.

So the choice kinda is, add about 30 overloads/complicated type unions to handle all the basic ctypes and their corresponding string literals correctly (but probably still have a lot of edge cases that are inferred wrong, due to inheritance) or just return SynchronizedBase which the result is always guaranteed to be derived from, typeshed's maintainers probably went for the second option in order to prioritize correctness.

To get around it you can do what I ended up doing and force cast it to multiprocessing.sharedctypes.Synchronized[int], since that is what you will get in this case or only use what's implemented on the SynchronizedBase type and do what the value property on Synchronized does under the hood:

import multiprocessing as mp
x = mp.Value('i', 0)
with x:  # aquires lock
    x.get_obj().value = 0

It's obviously a lot more verbose, but it will work without mypy complaining. (You probably should bind SynchronizedBase to the correct c_type though, so mypy can infer the return value of get_obj(), but at least you won't need to force an unsafe cast)

Another option would be to use multiprocessing.sharedctypes.synchronized on a RawValue to construct its synchronized counterpart manually (the overloads for synchronized and RawValue are a lot easier to get right). You just have to be careful about not using the RawValue directly, unless you know it's safe to access it without first acquiring an exclusive lock on it.

So the following will be inferred correctly by mypy:

import multiprocessing as mp
from ctypes import c_int
from multiprocessing.sharedctypes import synchronized

x = synchronized(mp.RawValue(c_int, 0))
reveal_type(x)   # Revealed type is "multiprocessing.sharedctypes.Synchronized[builtins.int]"

Value under the hood just either returns what RawValue gives you back if you set lock=False or returns synchronized(RawValue(*args)) if lock=True, which is the default. It's quite likely that you don't care about the lock in some cases, so sometimes it's even useful to have both a raw and a synchronized version of the variable. (although you can always get the raw value back by calling .get_obj() on the synchronized one)

There's also some caveats to using the property on Synchronized, since for example an increment won't be atomic i.e. x += 1 will acquire and release the lock twice for reading and setting the value, so you will be better off going with the more verbose option in that case anyways i.e. using the context manager and performing the increment on the raw value within.