I am trying to share a string between processes with multiprocessing.Value(c_wchar_p, ""). Program exits unexpectedly without exception with code (0xC0000005). I am guessing that problem is somehow caused by main process trying to access memory of child process. Issue occurs only when ctypes c_char_p/c_wchar_p are used. Thread works fine. Is this expected beheavor?
Steps to reproduce:
import multiprocessing
import threading
import ctypes
import time
class ProcessExample:
def __init__(self, proc: bool):
self.shared_bool = multiprocessing.Value(ctypes.c_bool, False)
self.shared_string = multiprocessing.Value(ctypes.c_wchar_p, "one")
self.shared_byte = multiprocessing.Value(ctypes.c_char_p, b"one")
if proc:
self.th_proc = multiprocessing.Process(target=self.target, args=(self.shared_bool, self.shared_string, self.shared_byte))
else:
self.th_proc = threading.Thread(target=self.target, args=(self.shared_bool, self.shared_string, self.shared_byte))
self.th_proc.start()
@staticmethod
def target(shared_bool, shared_string, shared_byte):
shared_bool.value = True
shared_string.value = "two"
shared_byte.value = b"two"
def print_values(self):
self.th_proc.join()
print(self.shared_bool.value)
print(self.shared_string.value) # Exits
print(self.shared_byte.value) # Exits
if __name__ == "__main__":
# example = ProcessExample(False) # Works
example = ProcessExample(True)
time.sleep(1)
example.print_values()
Output
True
Process finished with exit code -1073741819 (0xC0000005)
I think the problem is in multiprocessing.Value(ctypes.c_wchar_p, "one")
and multiprocessing.Value(ctypes.c_char_p, b"one")
. To have string/byte string initialized correctly, use multiprocessing.Array
:
import multiprocessing
import threading
import ctypes
import time
class ProcessExample:
def __init__(self, proc: bool):
self.shared_bool = multiprocessing.Value(ctypes.c_bool, False)
self.shared_string = multiprocessing.Array(ctypes.c_wchar, "one")
self.shared_byte = multiprocessing.Array(ctypes.c_char, b"one")
if proc:
self.th_proc = multiprocessing.Process(
target=self.target,
args=(self.shared_bool, self.shared_string, self.shared_byte),
)
else:
self.th_proc = threading.Thread(
target=self.target,
args=(self.shared_bool, self.shared_string, self.shared_byte),
)
self.th_proc.start()
@staticmethod
def target(shared_bool, shared_string, shared_byte):
shared_bool.value = True
shared_string[:] = "two"
shared_byte[:] = b"two"
def print_values(self):
self.th_proc.join()
print(self.shared_bool.value)
print(self.shared_string[:])
print(self.shared_byte[:])
if __name__ == "__main__":
example = ProcessExample(True)
time.sleep(1)
example.print_values()
Prints:
True
two
b'two'