DZone
Thanks for visiting DZone today,
Edit Profile
  • Manage Email Subscriptions
  • How to Post to DZone
  • Article Submission Guidelines
Sign Out View Profile
  • Post an Article
  • Manage My Drafts
Over 2 million developers have joined DZone.
Log In / Join
Refcards Trend Reports
Events Video Library
Over 2 million developers have joined DZone. Join Today! Thanks for visiting DZone today,
Edit Profile Manage Email Subscriptions Moderation Admin Console How to Post to DZone Article Submission Guidelines
View Profile
Sign Out
Refcards
Trend Reports
Events
View Events Video Library
Zones
Culture and Methodologies Agile Career Development Methodologies Team Management
Data Engineering AI/ML Big Data Data Databases IoT
Software Design and Architecture Cloud Architecture Containers Integration Microservices Performance Security
Coding Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks
Culture and Methodologies
Agile Career Development Methodologies Team Management
Data Engineering
AI/ML Big Data Data Databases IoT
Software Design and Architecture
Cloud Architecture Containers Integration Microservices Performance Security
Coding
Frameworks Java JavaScript Languages Tools
Testing, Deployment, and Maintenance
Deployment DevOps and CI/CD Maintenance Monitoring and Observability Testing, Tools, and Frameworks

Mobile Database Essentials: Assess data needs, storage requirements, and more when leveraging databases for cloud and edge applications.

Full-Stack Observability Essentials: Explore the fundamentals of system-wide observability and key components of the OpenTelemetry standard.

Monitoring and Observability for LLMs: Datadog and Google Cloud discuss how to achieve optimal AI model performance.

Automated Testing: The latest on architecture, TDD, and the benefits of AI and low-code tools.

Related

  • Rethinking Quality Assurance in the Age of Generative AI
  • A Continuous Testing Approach to Performance
  • Top Five AI-Powered Tools for Test Automation
  • Selecting the Right Automated Tests

Trending

  • Supercharge Your Communication With Twilio and Ballerina
  • SQL Query Performance Tuning in MySQL
  • A Complete Guide to Open-Source LLMs
  • Exploring Different Continuous Integration Servers: Streamlining Software Development

Undoing Gevent’s Monkey-Patching

A. Jesse Jiryu Davis user avatar by
A. Jesse Jiryu Davis
·
Apr. 06, 12 · Interview
Like (0)
Save
Tweet
Share
4.99K Views

Join the DZone community and get the full member experience.

Join For Free

I ran into an odd problem while testing the next release of PyMongo, the Python driver for MongoDB which I help develop. We’re improving its support for Gevent, so we’re of course doing additional tests that begin with:

from gevent import monkey; monkey.patch_socket()

Now, some tests rely on this patching, and some rely on not being patched. Gevent doesn’t provide an unpatch_socket, so I had a clever idea: I’ll fork a subprocess with multiprocessing, do the test there, and return its result to the parent process in a multiprocessing.Value. Then subsequent tests won’t be affected by the patching.

SUCCESS = 1
FAILURE = 0

def my_test(outcome):
    from gevent import monkey; monkey.patch_socket()
    # do the test ....
    outcome.value = SUCCESS

class Test(unittest.TestCase):
    def test(self):
        outcome = multiprocessing.Value('i', FAILURE)
        multiprocessing.Process(target=my_test, args=(outcome,)).start().join()
        self.assertEqual(SUCCESS, outcome.value)

Nice and straightforward, right? In sane operating systems this worked great. On Windows it broke horribly. When I did python setup.py test, instead of executing my_test(), multiprocessing on Windows restarted the whole test suite, which started another whole test suite, … Apparently, since Windows can’t fork(), multiprocessing re-imports your script and attempts to execute the proper function within it. If the test suite is begun with python setup.py test, then everything goes haywire. This problem with multiprocessing and unittests on Windows was discussed on the Python mailing list last February.

After some gloomy minutes, I decided to look at what patch_socket() is doing. Turns out it’s simple, so I wrote a version which allows unpatching:

def patch_socket(aggressive=True):
    """Like gevent.monkey.patch_socket(), but stores old socket attributes for
    unpatching.
    """
    from gevent import socket
    _socket = __import__('socket')

    old_attrs = {}
    for attr in (
        'socket', 'SocketType', 'create_connection', 'socketpair', 'fromfd'
    ):
        if hasattr(_socket, attr):
            old_attrs[attr] = getattr(_socket, attr)
            setattr(_socket, attr, getattr(socket, attr))

    try:
        from gevent.socket import ssl, sslerror
        old_attrs['ssl'] = _socket.ssl
        _socket.ssl = ssl
        old_attrs['sslerror'] = _socket.sslerror
        _socket.sslerror = sslerror
    except ImportError:
        if aggressive:
            try:
                del _socket.ssl
            except AttributeError:
                pass

    return old_attrs


def unpatch_socket(old_attrs):
    """Take output of patch_socket() and undo patching."""
    _socket = __import__('socket')

    for attr in old_attrs:
        if hasattr(_socket, attr):
            setattr(_socket, attr, old_attrs[attr])


def patch_dns():
    """Like gevent.monkey.patch_dns(), but stores old socket attributes for
    unpatching.
    """
    from gevent.socket import gethostbyname, getaddrinfo
    _socket = __import__('socket')

    old_attrs = {}
    old_attrs['getaddrinfo'] = _socket.getaddrinfo
    _socket.getaddrinfo = getaddrinfo
    old_attrs['gethostbyname'] = _socket.gethostbyname
    _socket.gethostbyname = gethostbyname

    return old_attrs


def unpatch_dns(old_attrs):
    """Take output of patch_dns() and undo patching."""
    _socket = __import__('socket')

    for attr in old_attrs:
        setattr(_socket, attr, old_attrs[attr])

In Gevent’s version, calling patch_socket() calls patch_dns() implicitly, in mine you must call both:

class Test(unittest.TestCase):
    def test(self):
        old_socket_attrs = patch_socket()
        old_dns_attrs = patch_dns()

        try:
            # do test ...
        finally:
            unpatch_dns(old_dns_attrs)
            unpatch_socket(old_socket_attrs)

Now I don’t need multiprocessing at all.

 

Testing

Published at DZone with permission of A. Jesse Jiryu Davis, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.

Related

  • Rethinking Quality Assurance in the Age of Generative AI
  • A Continuous Testing Approach to Performance
  • Top Five AI-Powered Tools for Test Automation
  • Selecting the Right Automated Tests

Comments

Partner Resources

X

ABOUT US

  • About DZone
  • Send feedback
  • Careers
  • Sitemap

ADVERTISE

  • Advertise with DZone

CONTRIBUTE ON DZONE

  • Article Submission Guidelines
  • Become a Contributor
  • Visit the Writers' Zone

LEGAL

  • Terms of Service
  • Privacy Policy

CONTACT US

  • 3343 Perimeter Hill Drive
  • Suite 100
  • Nashville, TN 37211
  • support@dzone.com

Let's be friends: