joblib 1.5.3: Handle closure serialization failures and backend fallback when switching joblib from loky to multiprocessing

# Notes on joblib Multiprocessing and Serialization

## Prior Search Result
`search_known_solution` returned **MISS** (no relevant verified samples existed for `joblib` multiprocessing backend pickling and start methods; returned unrelated ecosystem samples).

## What a Naive Model Would Write
A naive model would write `Parallel(n_jobs=2, backend="multiprocessing")(delayed(lambda x: x * 2)(i) for i in range(10))` assuming `multiprocessing` behaves identically to the default `loky` backend.

## How the Wrong Version Fails
Loudly at runtime with `AttributeError: Can't pickle local object '<locals>.<lambda>'` or `pickle.PicklingError` because standard `multiprocessing` uses `pickle` rather than `cloudpickle` and cannot serialize dynamically defined closures without `@wrap_non_picklable_objects`.

{"case":{"caseId":"case:sha256:2eb36abd3109f462c3268b9f075a5a682959b51a5d44fcd3342fc0761e222107","constraints":{"runtime":"python"},"contract":["assert jp.DEFAULT_BACKEND == 'loky' and isinstance(jp.get_active_backend()[0], jp.LokyBackend)","assert 'loky' in loky_ctx.START_METHODS","assert Parallel(n_jobs=2, backend='loky')(delayed(lambda x: x + 100)(i) for i in range(3)) == [100, 101, 102]","assert Parallel(n_jobs=2, backend='multiprocessing')(delayed(lambda x: x + 100)(i)) raises (AttributeError, pickle.PicklingError)","assert Parallel(n_jobs=2, backend='multiprocessing')(delayed(wrap_non_picklable_objects(lambda x: x + 100))(i) for i in range(3)) == [100, 101, 102]","assert Parallel(n_jobs=2, backend='threading')(delayed(worker_with_lock)(lock, i) for i in range(3)) == [0, 2, 4]","assert isinstance(Parallel(n_jobs=2, require='sharedmem')._backend, jp.ThreadingBackend)","assert Parallel(backend='multiprocessing', require='sharedmem') raises ValueError"],"goal":"Handle closure serialization failures and backend fallback when switching joblib from loky to multiprocessing","kind":"HOW","packages":["pkg:pypi/joblib@1.5.3"],"schemaVersion":1,"symbols":["joblib.Parallel","joblib.delayed","joblib.wrap_non_picklable_objects","joblib.parallel_backend","joblib.parallel.DEFAULT_BACKEND","joblib.parallel.get_active_backend","joblib.externals.loky.backend.context.START_METHODS"]},"contractCommand":["python","test/contract.py"],"environment":{"arch":"x64","ecosystem":"pypi","executionContext":"python","language":"python","os":"linux","packageManager":"pip","runtime":"python","schemaVersion":1},"license":"MIT-0","packages":["pkg:pypi/joblib@1.5.3"],"schemaVersion":1,"symbols":["joblib.Parallel","joblib.delayed","joblib.wrap_non_picklable_objects","joblib.parallel_backend","joblib.parallel.DEFAULT_BACKEND","joblib.parallel.get_active_backend","joblib.externals.loky.backend.context.START_METHODS"],"verifierAdapter":"python@1"}

[build-system]
requires = ["setuptools>=61.0"]
build-backend = "setuptools.build_meta"

[project]
name = "sample-joblib-start-method"
version = "0.1.0"
dependencies = [
    "joblib==1.5.3",
]

joblib==1.5.3

joblib==1.5.3

import pickle
import threading
from joblib import Parallel, delayed, parallel_backend, wrap_non_picklable_objects
import joblib.parallel as jp
import joblib.externals.loky.backend.context as loky_ctx


def _thread_worker(lock, item):
    with lock:
        return item * 2


def main():
    # 1. Joblib defaults to 'loky' backend, NOT 'multiprocessing'
    active_backend, _ = jp.get_active_backend()
    assert jp.DEFAULT_BACKEND == "loky"
    assert isinstance(active_backend, jp.LokyBackend)

    # 2. Loky backend uses dedicated process context and cloudpickle serialization for closures
    assert "loky" in loky_ctx.START_METHODS
    base = 100
    local_closure = lambda x: x + base
    res_loky = Parallel(n_jobs=2, backend="loky")(
        delayed(local_closure)(i) for i in range(3)
    )
    assert res_loky == [100, 101, 102]

    # 3. Switching to 'multiprocessing' uses standard pickle which fails on raw local closures
    mp_failed = False
    try:
        Parallel(n_jobs=2, backend="multiprocessing")(
            delayed(local_closure)(i) for i in range(3)
        )
    except (AttributeError, pickle.PicklingError):
        mp_failed = True
    assert mp_failed, "backend='multiprocessing' must fail on un-picklable local closure"

    # 4. wrap_non_picklable_objects enables raw closures to work with backend='multiprocessing'
    wrapped = wrap_non_picklable_objects(local_closure)
    res_mp_wrapped = Parallel(n_jobs=2, backend="multiprocessing")(
        delayed(wrapped)(i) for i in range(3)
    )
    assert res_mp_wrapped == [100, 101, 102]

    # 5. Non-serializable state (like threading.Lock) fails in process backends but works in threading
    lock = threading.Lock()
    loky_lock_failed = False
    try:
        Parallel(n_jobs=2, backend="loky")(
            delayed(_thread_worker)(lock, i) for i in range(3)
        )
    except Exception:
        loky_lock_failed = True
    assert loky_lock_failed, "loky process pool cannot serialize threading.Lock"

    res_threads = Parallel(n_jobs=2, backend="threading")(
        delayed(_thread_worker)(lock, i) for i in range(3)
    )
    assert res_threads == [0, 2, 4]

    # 6. require='sharedmem' automatically switches backend context to ThreadingBackend
    with parallel_backend("multiprocessing"):
        p_shared = Parallel(n_jobs=2, require="sharedmem")
        assert isinstance(p_shared._backend, jp.ThreadingBackend)
        res_shared = p_shared(delayed(_thread_worker)(lock, i) for i in range(3))
        assert res_shared == [0, 2, 4]

    # 7. Explicit contradictory arguments raise ValueError
    contradiction_caught = False
    try:
        Parallel(backend="multiprocessing", require="sharedmem")
    except ValueError:
        contradiction_caught = True
    assert contradiction_caught, "Contradictory backend and require='sharedmem' must raise ValueError"


if __name__ == "__main__":
    main()