小编Bha*_*rel的帖子

我遇到了这种奇怪的行为并且无法解释它.这些是基准:

py -3 -m timeit "tuple(range(2000)) == tuple(range(2000))"
10000 loops, best of 3: 97.7 usec per loop
py -3 -m timeit "a = tuple(range(2000));  b = tuple(range(2000)); a==b"
10000 loops, best of 3: 70.7 usec per loop

为什么与变量赋值的比较比使用临时变量的单个衬里快27%以上？

通过Python文档,在timeit期间禁用垃圾收集,因此它不可能.这是某种优化吗？

结果也可以在Python 2.x中重现,但程度较小.

运行Windows 7,CPython 3.5.1,Intel i7 3.40 GHz,64位操作系统和Python.看起来像我尝试在Intel i7 3.60 GHz上使用Python 3.5.0运行的另一台机器不能重现结果.

使用与timeit.timeit()@ 10000循环相同的Python进程运行分别产生0.703和0.804.仍显示尽管程度较轻.(〜12.5%)

这个问题是由我的另一个问题推动的:如何在cdef中等待？

网上有大量的文章和博客文章asyncio,但它们都非常肤浅.我找不到任何有关如何asyncio实际实现的信息,以及I/O异步的原因.我试图阅读源代码,但它是成千上万行不是最高级别的C代码,其中很多处理辅助对象,但最重要的是,很难在Python语法和它将翻译的C代码之间建立连接成.

Asycnio自己的文档甚至没那么有用.没有关于它是如何工作的信息,只有关于如何使用它的一些指导,这些指导有时也会误导/写得很差.

我熟悉Go的coroutines实现,并且希望Python做同样的事情.如果是这种情况,我在上面链接的帖子中出现的代码就可以了.既然没有,我现在正试图找出原因.到目前为止我最好的猜测如下,请纠正我错在哪里:

1. 表单的过程定义async def foo(): ...实际上被解释为类继承的方法coroutine.
2. 也许,async def实际上是通过await语句拆分成多个方法,其中调用这些方法的对象能够跟踪到目前为止通过执行所做的进度.
3. 如果上述情况属实,那么,从本质上讲,协程的执行归结为一些全局管理器调用协程对象的方法(循环？).
4. 全局管理器以某种方式(如何？)了解I/O操作何时由Python(仅？)代码执行,并且能够在当前执行方法放弃控制之后选择一个待执行的协程方法执行(命中await语句) ).

换句话说,这是我尝试将某些asyncio语法"贬低"为更容易理解的东西:

async def coro(name):
    print('before', name)
    await asyncio.sleep()
    print('after', name)

asyncio.gather(coro('first'), coro('second'))

# translated from async def coro(name)
class Coro(coroutine):
    def before(self, name):
        print('before', name)

    def after(self, name):
        print('after', name)

    def __init__(self, name):
        self.name = name
        self.parts = self.before, self.after
        self.pos = 0

    def __call__():
        self.parts[self.pos](self.name)
        self.pos += 1

    def …

我有一个程序,我目前正在使用concurrent.futures.ThreadPoolExecutor同时运行多个任务.这些任务通常受I/O限制,涉及对本地数据库和远程REST API的访问.但是,这些任务本身可以分成子任务,这也可以从并发中受益.

我希望在任务中使用concurrent.futures.ThreadPoolExecutor是安全的.我编写了一个玩具示例,它似乎有效:

import concurrent.futures


def inner(i, j):
    return i, j, i**j


def outer(i):
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(inner, i, j): j for j in range(5)}
        results = []
        for future in concurrent.futures.as_completed(futures):
            results.append(future.result())
    return results


def main():
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = {executor.submit(outer, i): i for i in range(10)}
        results = []
        for future in concurrent.futures.as_completed(futures):
            results.extend(future.result())
    print(results)


if __name__ == "__main__":
    main()

尽管这个玩具示例似乎有效,但我还是有信心这是故意的.我希望它是,因为否则使用执行程序执行任意代码是不安全的,以防它也使用concurrent.futures来利用并发.

它似乎

import Queue

Queue.Queue().get(timeout=10)

是键盘可中断(ctrl-c)而

import Queue

Queue.Queue().get()

不是.我总是可以创建一个循环;

import Queue
q = Queue()

while True:
    try:
        q.get(timeout=1000)
    except Queue.Empty:
        pass

但这似乎是一件奇怪的事情.

那么,是否有一种方法可以无限期地等待但是键盘可以中断Queue.get()？

下面的代码失败mypy并显示error: Overloaded function signatures 1 and 2 overlap with incompatible return types.

@overload
def test_overload(x: str) -> str: ...

@overload
def test_overload(x: object) -> int: ...

def test_overload(x) -> Union[str, int]:
    if isinstance(x, str):
        return x
    else:
        return 1

我想要表达的是：“这个函数接受一个任意的Python对象。如果该对象是一个字符串，它返回一个字符串。如果它是任何其他类型，它返回一个整数。注意这个特定的例子是为了表示一般情况。

可以用重载来表达吗？

示例代码：

async def test():
  """
  >>> await test()
  hello world
  """
  print("hello world")

尝试使用 doctests 运行它会导致SyntaxError: 'await' outside function.

假设我有这些解析器:

parsers = {
    ".foo": parse_foo,
    ".bar", parse_bar
}

parse_foo并且parse_bar都是逐个产生行的生成器.如果我想创建一个调度函数,我会这样做:

def parse(ext):
    yield from parsers[ext]()

语法的收益使我能够轻松地在生成器上下传递信息.

有没有办法在修改产量结果的同时保持隧道效应？
在破坏隧道时这样做很容易:

def parse(ext):
    for result in parsers[ext]():
        # Add the extension to the result
        result.ext = ext
        yield result

但这种方式我无法使用.send()或.throw()一直到解析器.

我想到的唯一方法是做一些丑陋的事情try: ... except Exception: ...并将异常传递出去,同时做同样的事情.send().它很难看,很杂乱,容易出错.

pip install cmake在 alpine linux (WSL) 上运行时出现上述错误。

完整错误：

/home/user# pip install cmake
Collecting cmake
  Using cached cmake-3.18.4.post1.tar.gz (28 kB)
  Installing build dependencies ... done
  Getting requirements to build wheel ... done
    Preparing wheel metadata ... done
Building wheels for collected packages: cmake
  Building wheel for cmake (PEP 517) ... error
  ERROR: Command errored out with exit status 1:
   command: /usr/bin/python3 /usr/lib/python3.8/site-packages/pip/_vendor/pep517/_in_process.py build_wheel /tmp/tmp4q90dnrm
       cwd: /tmp/pip-install-2d2lze1g/cmake_b52ab822a1764f9f9b60c93c8713e39b
  Complete output (9 lines):
    File "/tmp/pip-build-env-szooj3fd/overlay/lib/python3.8/site-packages/skbuild/setuptools_wrap.py", line 560, in setup
      cmkr = cmaker.CMaker(cmake_executable)
    File …

我想在缩进代码中搜索不以井号(#)开头的行.

目前,我正在使用^\s*([^\s#].*)带有多行选项的正则表达式.

我的问题是在非注释行上它完美地工作.

在注释行上,正则表达式引擎执行回溯,因为\s*从注释符号一直到行的开头,这有时会导致40或50个回溯步骤.

正则表达式完全适用于python代码.由于发动机引起的回溯,它效率不高.

关于如何避免它的任何想法？

奖励:有趣的是,正则表达式引擎无法识别它正在[^\s]逐个搜索\s*并导致这种回溯的事实.使重新发动机工作的挑战是什么？

奖励2:仅使用stdlib re模块.因为我无法添加第三方.(我在技术上使用sublime文本搜索,但想知道如何在Python中一般地使用它)

我有一个脚本，可以删除很多 dict 并最终对其进行迭代。

我设法将其简化为一个简单的基准：

> py -m timeit -s "a = {i:i for i in range(10000000)};[a.pop(i) for i in range(10000000-1)]" "next(iter(a))"
10 loops, best of 5: 30.8 msec per loop

为什么在删除所有以前的值后迭代单个键变得很慢？