Pfthb

I am building a library for Python and noticed that one of my users was using the objects incorrectly, by sharing it between 2 Processes.

So, I wanted to bake in a way to prevent the user from doing that.

One way would be to add a "check" before every method call, using a decorator, but that adds a bit of overhead on each method call.

import multiprocessing
import threading


def is_parent():
 return (
 multiprocessing.current_process().name == "MainProcess"
 and threading.current_thread() == threading.main_thread()
 )


def process_unsafe(fn):
 def wrapper(*args, **kwargs):
 if not is_parent():
 raise RuntimeError("Not allowed!")
 fn(*args, **kwargs)

 return wrapper


class NonProcessSafe:
 @process_unsafe
 def foo(self):
 pass

 @process_unsafe
 def bar(self):
 pass


nps = NonProcessSafe()

nps.foo()
nps.bar()


def child():
 # complains
 nps.foo()


p = multiprocessing.Process(target=child)
p.start()
p.join()

Is there any more efficient way to achieve this?

As @timgeb mentioned in the comments, there's probably no way to close all the loopholes. Furthermore I'm not sure I fully understand your problem and process/thread/state sharing model...

Anyway, here's an Ansatz that might get you somewhere, at least with regards to processes (in the Unix/Linux sense):

import os

class State:

 def __init__(self):
 self._pid = os.getpid()
 self._state = 

 def __setitem__(self, key, value):
 if os.getpid() != self._pid:
 raise RuntimeError("this process must not change state")
 self._state[key] = value

The State instance remembers the ID of the process that created it, and can then check against that ID in all (relevant) methods. This will of course require the State be created by its "owner" process, in which case it is invisible to any parent or sibling process anyway, and visible only to itself and its children.

For threads, use threading.get_ident() instead of os.getpid().

If it is always the "main" process that creates the state and then forks worker children, instead you could (in the main process) temporarily mark the state as read-only, fork the worker child so it gets a read-only copy, and then in the main process switch state back to read-write:

class State:
 def __init__(self):
 self._read_only = False
 self._state = 
 def __setattr__(self, key, value):
 if self._read_only:
 raise RuntimeError("read only!")
 self._state[key] = value

In main process:

def new_worker(fct):
 state = State()
 state._read_only = True
 pid = os.fork() # or whatever you use to create a new process
 if pid:
 fct() # state is _read-only=True here in child process
 else:
 state._read_only = False
 state_list.append((pid, state))