add error message for invalid multiprocess

Author: RichardZhu2

src/pyper/_core/task.py

@@ -48,9 +48,19 @@ def __init__(
             or inspect.iscoroutinefunction(func.__call__) \
             or inspect.isasyncgenfunction(func.__call__)
 
-        if self.is_async and multiprocess:
-            raise ValueError("multiprocess cannot be True for an async task")
-        
+        if multiprocess:
+            # Asynchronous functions cannot be multiprocessed
+            if self.is_async:
+                raise ValueError("multiprocess cannot be True for an async task")
+            
+            # The function needs to be globally accessible to be multiprocessed
+            # This excludes objects like lambdas and closures
+            # We capture these cases to throw a clear error message
+            module = inspect.getmodule(func)
+            if module is None or getattr(module, func.__name__, None) is not func:
+                raise RuntimeError(f"{func} cannot be multiprocessed because it is not globally accessible"
+                                   f" -- it must be a globally defined object accessible by the name {func.__name__}")
+            
         self.func = func if bind is None else functools.partial(func, *bind[0], **bind[1])
         self.branch = branch
         self.join = join

tests/test_task.py

@@ -74,14 +74,32 @@ def test_raise_for_invalid_func():
     else:
         raise AssertionError
 
-def test_raise_for_invalid_multiprocess():
+def test_raise_for_async_multiprocess():
     try:
         task(afunc, multiprocess=True)
     except Exception as e:
         assert isinstance(e, ValueError)
     else:
         raise AssertionError
 
+def test_raise_for_lambda_multiprocess():
+    try:
+        task(lambda x: x, multiprocess=True)
+    except Exception as e:
+        assert isinstance(e, RuntimeError)
+    else:
+        raise AssertionError
+    
+def test_raise_for_non_global_multiprocess():
+    try:
+        @task(multiprocess=True)
+        def f(x):
+            return x
+    except Exception as e:
+        assert isinstance(e, RuntimeError)
+    else:
+        raise AssertionError
+
 def test_async_task():
     p = task(afunc)
     assert isinstance(p, AsyncPipeline)