( )⚙ D12279 cleanup: directly use concurrent.futures instead of via pycompat

This is an archive of the discontinued Mercurial Phabricator instance.

Differential D12279

cleanup: directly use concurrent.futures instead of via pycompat
ClosedPublic

Authored by durin42 on Mar 2 2022, 10:44 AM.

Details

Reviewers
indygreg
Group Reviewers
hg-reviewers
Commits
rHGa0da5075bca3: cleanup: directly use concurrent.futures instead of via pycompat
rHGd0d411bcfcee: cleanup: directly use concurrent.futures instead of via pycompat
Summary

Python 2 is gone.

Diff Detail

Repository
rHG Mercurial
Branch
default
Lint
No Linters Available
Unit
No Unit Test Coverage

Event Timeline

durin42 created this revision.Mar 2 2022, 10:44 AM
indygreg accepted this revision.Mar 2 2022, 11:20 AM
This revision is now accepted and ready to land.Mar 2 2022, 11:20 AM
indygreg added inline comments.Mar 2 2022, 6:18 PM
mercurial/pycompat.py
48

In hindsight, we shouldn't have removed this, as the pycompat.futures symbol is part of the pycompat API, which we want to preserve to help extensions.

I'll restore this in D12247.

Revision Contents
Changeset List

Diff 32370

mercurial/httppeer.py

# httppeer.py - HTTP repository proxy classes for mercurial # httppeer.py - HTTP repository proxy classes for mercurial
# #
# Copyright 2005, 2006 Olivia Mackall <olivia@selenic.com> # Copyright 2005, 2006 Olivia Mackall <olivia@selenic.com>
# Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.com> # Copyright 2006 Vadim Gelfer <vadim.gelfer@gmail.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import errno import errno
import io import io
import os import os
import socket import socket
import struct import struct
from concurrent import futures
from .i18n import _ from .i18n import _
from .pycompat import getattr from .pycompat import getattr
from . import ( from . import (
bundle2, bundle2,
error, error,
httpconnection, httpconnection,
pycompat, pycompat,
statichttprepo, statichttprepo,
def _callcompressable(self, cmd, **args): def _callcompressable(self, cmd, **args):
return self._callstream(cmd, _compressible=True, **args) return self._callstream(cmd, _compressible=True, **args)
def _abort(self, exception): def _abort(self, exception):
raise exception raise exception
class queuedcommandfuture(pycompat.futures.Future): class queuedcommandfuture(futures.Future):
"""Wraps result() on command futures to trigger submission on call.""" """Wraps result() on command futures to trigger submission on call."""
def result(self, timeout=None): def result(self, timeout=None):
if self.done(): if self.done():
return pycompat.futures.Future.result(self, timeout) return futures.Future.result(self, timeout)
self._peerexecutor.sendcommands() self._peerexecutor.sendcommands()
# sendcommands() will restore the original __class__ and self.result # sendcommands() will restore the original __class__ and self.result
# will resolve to Future.result. # will resolve to Future.result.
return self.result(timeout) return self.result(timeout)

mercurial/localrepo.py

import errno import errno
import functools import functools
import os import os
import random import random
import sys import sys
import time import time
import weakref import weakref
from concurrent import futures
from .i18n import _ from .i18n import _
from .node import ( from .node import (
bin, bin,
hex, hex,
nullrev, nullrev,
sha1nodeconstants, sha1nodeconstants,
short, short,
) )
raise error.ProgrammingError( raise error.ProgrammingError(
b'callcommand() cannot be used after close()' b'callcommand() cannot be used after close()'
) )
# We don't need to support anything fancy. Just call the named # We don't need to support anything fancy. Just call the named
# method on the peer and return a resolved future. # method on the peer and return a resolved future.
fn = getattr(self._peer, pycompat.sysstr(command)) fn = getattr(self._peer, pycompat.sysstr(command))
f = pycompat.futures.Future() f = futures.Future()
try: try:
result = fn(**pycompat.strkwargs(args)) result = fn(**pycompat.strkwargs(args))
except Exception: except Exception:
pycompat.future_set_exception_info(f, sys.exc_info()[1:]) pycompat.future_set_exception_info(f, sys.exc_info()[1:])
else: else:
f.set_result(result) f.set_result(result)

mercurial/pycompat.py

if not ispy3: if not ispy3:
import cookielib import cookielib
import cPickle as pickle import cPickle as pickle
import httplib import httplib
import Queue as queue import Queue as queue
import SocketServer as socketserver import SocketServer as socketserver
import xmlrpclib import xmlrpclib
from .thirdparty.concurrent import futures
def future_set_exception_info(f, exc_info): def future_set_exception_info(f, exc_info):
f.set_exception_info(*exc_info) f.set_exception_info(*exc_info)
# this is close enough for our usage # this is close enough for our usage
FileNotFoundError = OSError FileNotFoundError = OSError
else: else:
import builtins import builtins
import concurrent.futures as futures
indygregUnsubmitted
Not Done

In hindsight, we shouldn't have removed this, as the pycompat.futures symbol is part of the pycompat API, which we want to preserve to help extensions.

I'll restore this in D12247.

indygreg: In hindsight, we shouldn't have removed this, as the `pycompat.futures` symbol is part of the…
import http.cookiejar as cookielib import http.cookiejar as cookielib
import http.client as httplib import http.client as httplib
import pickle import pickle
import queue as queue import queue as queue
import socketserver import socketserver
import xmlrpc.client as xmlrpclib import xmlrpc.client as xmlrpclib
def future_set_exception_info(f, exc_info): def future_set_exception_info(f, exc_info):

mercurial/wireprotov1peer.py

# wireprotov1peer.py - Client-side functionality for wire protocol version 1. # wireprotov1peer.py - Client-side functionality for wire protocol version 1.
# #
# Copyright 2005-2010 Olivia Mackall <olivia@selenic.com> # Copyright 2005-2010 Olivia Mackall <olivia@selenic.com>
# #
# This software may be used and distributed according to the terms of the # This software may be used and distributed according to the terms of the
# GNU General Public License version 2 or any later version. # GNU General Public License version 2 or any later version.
from __future__ import absolute_import from __future__ import absolute_import
import sys import sys
import weakref import weakref
from concurrent import futures
from .i18n import _ from .i18n import _
from .node import bin from .node import bin
from .pycompat import ( from .pycompat import (
getattr, getattr,
setattr, setattr,
) )
from . import ( from . import (
bundle2, bundle2,
b'%s=%s' % (escapearg(k), escapearg(v)) b'%s=%s' % (escapearg(k), escapearg(v))
for k, v in pycompat.iteritems(argsdict) for k, v in pycompat.iteritems(argsdict)
) )
cmds.append(b'%s %s' % (op, args)) cmds.append(b'%s %s' % (op, args))
return b';'.join(cmds) return b';'.join(cmds)
class unsentfuture(pycompat.futures.Future): class unsentfuture(futures.Future):
"""A Future variation to represent an unsent command. """A Future variation to represent an unsent command.
Because we buffer commands and don't submit them immediately, calling Because we buffer commands and don't submit them immediately, calling
``result()`` on an unsent future could deadlock. Futures for buffered ``result()`` on an unsent future could deadlock. Futures for buffered
commands are represented by this type, which wraps ``result()`` to commands are represented by this type, which wraps ``result()`` to
call ``sendcommands()``. call ``sendcommands()``.
""" """
def result(self, timeout=None): def result(self, timeout=None):
if self.done(): if self.done():
return pycompat.futures.Future.result(self, timeout) return futures.Future.result(self, timeout)
self._peerexecutor.sendcommands() self._peerexecutor.sendcommands()
# This looks like it will infinitely recurse. However, # This looks like it will infinitely recurse. However,
# sendcommands() should modify __class__. This call serves as a check # sendcommands() should modify __class__. This call serves as a check
# on that. # on that.
return self.result(timeout) return self.result(timeout)
# Commands are either batchable or they aren't. If a command # Commands are either batchable or they aren't. If a command
# isn't batchable, we send it immediately because the executor # isn't batchable, we send it immediately because the executor
# can no longer accept new commands after a non-batchable command. # can no longer accept new commands after a non-batchable command.
# If a command is batchable, we queue it for later. But we have # If a command is batchable, we queue it for later. But we have
# to account for the case of a non-batchable command arriving after # to account for the case of a non-batchable command arriving after
# a batchable one and refuse to service it. # a batchable one and refuse to service it.
def addcall(): def addcall():
f = pycompat.futures.Future() f = futures.Future()
self._futures.add(f) self._futures.add(f)
self._calls.append((command, args, fn, f)) self._calls.append((command, args, fn, f))
return f return f
if getattr(fn, 'batchable', False): if getattr(fn, 'batchable', False):
f = addcall() f = addcall()
# But since we don't issue it immediately, we wrap its result() # But since we don't issue it immediately, we wrap its result()
return return
self._sent = True self._sent = True
# Unhack any future types so caller seens a clean type and to break # Unhack any future types so caller seens a clean type and to break
# cycle between us and futures. # cycle between us and futures.
for f in self._futures: for f in self._futures:
if isinstance(f, unsentfuture): if isinstance(f, unsentfuture):
f.__class__ = pycompat.futures.Future f.__class__ = futures.Future
f._peerexecutor = None f._peerexecutor = None
calls = self._calls calls = self._calls
# Mainly to destroy references to futures. # Mainly to destroy references to futures.
self._calls = None self._calls = None
# Simple case of a single command. We call it synchronously. # Simple case of a single command. We call it synchronously.
if len(calls) == 1: if len(calls) == 1:
# This will emit responses in order they were executed. # This will emit responses in order they were executed.
wireresults = self._peer._submitbatch(requests) wireresults = self._peer._submitbatch(requests)
# The use of a thread pool executor here is a bit weird for something # The use of a thread pool executor here is a bit weird for something
# that only spins up a single thread. However, thread management is # that only spins up a single thread. However, thread management is
# hard and it is easy to encounter race conditions, deadlocks, etc. # hard and it is easy to encounter race conditions, deadlocks, etc.
# concurrent.futures already solves these problems and its thread pool # concurrent.futures already solves these problems and its thread pool
# executor has minimal overhead. So we use it. # executor has minimal overhead. So we use it.
self._responseexecutor = pycompat.futures.ThreadPoolExecutor(1) self._responseexecutor = futures.ThreadPoolExecutor(1)
self._responsef = self._responseexecutor.submit( self._responsef = self._responseexecutor.submit(
self._readbatchresponse, states, wireresults self._readbatchresponse, states, wireresults
) )
def close(self): def close(self):
self.sendcommands() self.sendcommands()
if self._closed: if self._closed: