Skip to content

Fix multiprocessing warnings when runnign tests on Python 3.12 #16564

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 1 commit into from
Nov 27, 2023
Merged

Fix multiprocessing warnings when runnign tests on Python 3.12 #16564

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions mypy/moduleinspect.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@
import pkgutil
import queue
import sys
from multiprocessing import Process, Queue
from multiprocessing import Queue, get_context
from types import ModuleType


Expand Down Expand Up @@ -123,9 +123,13 @@ def __init__(self) -> None:
self._start()

def _start(self) -> None:
self.tasks: Queue[str] = Queue()
self.results: Queue[ModuleProperties | str] = Queue()
self.proc = Process(target=worker, args=(self.tasks, self.results, sys.path))
if sys.platform == "linux":
ctx = get_context("forkserver")
else:
ctx = get_context("spawn")
self.tasks: Queue[str] = ctx.Queue()
self.results: Queue[ModuleProperties | str] = ctx.Queue()
self.proc = ctx.Process(target=worker, args=(self.tasks, self.results, sys.path))
self.proc.start()
self.counter = 0 # Number of successful roundtrips

Expand Down
21 changes: 14 additions & 7 deletions mypy/test/testipc.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

import sys
import time
from multiprocessing import Process, Queue
from multiprocessing import Queue, get_context
from unittest import TestCase, main

import pytest
Expand Down Expand Up @@ -35,10 +35,17 @@ def server_multi_message_echo(q: Queue[str]) -> None:


class IPCTests(TestCase):
def setUp(self) -> None:
if sys.platform == "linux":
# The default "fork" start method is potentially unsafe
self.ctx = get_context("forkserver")
else:
self.ctx = get_context("spawn")

def test_transaction_large(self) -> None:
queue: Queue[str] = Queue()
queue: Queue[str] = self.ctx.Queue()
msg = "t" * 200000 # longer than the max read size of 100_000
p = Process(target=server, args=(msg, queue), daemon=True)
p = self.ctx.Process(target=server, args=(msg, queue), daemon=True)
p.start()
connection_name = queue.get()
with IPCClient(connection_name, timeout=1) as client:
Expand All @@ -49,9 +56,9 @@ def test_transaction_large(self) -> None:
p.join()

def test_connect_twice(self) -> None:
queue: Queue[str] = Queue()
queue: Queue[str] = self.ctx.Queue()
msg = "this is a test message"
p = Process(target=server, args=(msg, queue), daemon=True)
p = self.ctx.Process(target=server, args=(msg, queue), daemon=True)
p.start()
connection_name = queue.get()
with IPCClient(connection_name, timeout=1) as client:
Expand All @@ -67,8 +74,8 @@ def test_connect_twice(self) -> None:
assert p.exitcode == 0

def test_multiple_messages(self) -> None:
queue: Queue[str] = Queue()
p = Process(target=server_multi_message_echo, args=(queue,), daemon=True)
queue: Queue[str] = self.ctx.Queue()
p = self.ctx.Process(target=server_multi_message_echo, args=(queue,), daemon=True)
p.start()
connection_name = queue.get()
with IPCClient(connection_name, timeout=1) as client:
Expand Down