python使用ThreadPoolExecutor、ProcessPoolExecutor进行任务并行

在Python中,可以使用concurrent.futures.ThreadPoolExecutor来实现多线程调用函数。

from concurrent.futures import ThreadPoolExecutor

def my_function(arg1, arg2):
    print(f"Processing {arg1} and {arg2}")

with ThreadPoolExecutor(max_workers=3) as executor:
    for i in range(5):
        executor.submit(my_function, i, i*2)

需要获取返回值,可通过future.result()获取


from concurrent.futures import ThreadPoolExecutor
import time

def calculate_square(number):
    time.sleep(1)  # 模拟耗时操作
    return number * number

with ThreadPoolExecutor(max_workers=3) as executor:
    # 提交任务并保存future对象
    future1 = executor.submit(calculate_square, 2)
    future2 = executor.submit(calculate_square, 3)
    future3 = executor.submit(calculate_square, 4)
    
    # 获取结果(会阻塞直到任务完成)
    print(f"2的平方: {future1.result()}")
    print(f"3的平方: {future2.result()}")
    print(f"4的平方: {future3.result()}")

使用多少线程数合理?对于cpu计算密集型,可以使用cpu核心数作为线程数。对于I/O密集型,可以使用核心数x2。使用如下代码可以获得核心数

import os
cpu_count = os.cpu_count()  # 返回逻辑核心数(含超线程)‌
print(f"CPU核心数: {cpu_count}")

但是对于cpu密集型的任务,更推荐使用多进程的ProcessPoolExecutor,接口和ThreadPoolExecutor一样。但是要说明的是,开启进程、进程间通信和销毁进程要比同样的线程步骤耗时更大,如果要执行的任务不是明显的cpu密集型,那么使用ProcessPoolExecutor并不一定要快于ThreadPoolExecutor

ipdb调试python代码

在不使用调试模型运行python代码时,有时程序会发生我们意想不到的错误,想要复现错误来调试需要很多时间,或者错误是偶发性的,这时候我们可以借助ipdb这个工具来在运行代码时进行调试。

安装

pip install ipdb

在可能出错的地方,插入语句。一般可以配合try except语句一起使用,在捕获到错误的地方插入语句,程序在运行到这个地方时会进入到断点,然后在命令行里可以交互式调试代码

import ipdb; ipdb.set_trace()

常用调试命令

  • c:继续执行代码,直到遇到下一个断点或程序结束。
  • n:单步执行下一行代码(不会进入函数内部)。
  • s:单步进入下一行代码(如果有函数调用,则进入函数内部)。
  • q:退出调试器并终止程序的执行。
  • l:查看当前位置附近的代码。
  • p:打印变量的值,例如p variable_name
  • h:查看帮助信息,例如h command_name
  • w:查看当前的调用栈。
  • u:向上移动一层调用栈。
  • d:向下移动一层调用栈。

numba 操作pytorch GPU tensor

numba可以对numpy的array使用gpu进行并行计算,从而进行加速。

import numpy as np
from numba import cuda

自定义numba cuda函数
@cuda.jit
def mat_add(A, B, C):
    j, i = cuda.grid(2)
    C[i, j] = A[i, j] + B[i, j]

a = np.arange(12).reshape(3, 4)
b = np.arange(12, 24).reshape(3, 4)
c = cuda.device_array_like(a_in)

a_dev = cuda.to_device(a)
b_dev = cuda.to_device(b)

blocks = (1, 1)
threads_per_block = (4, 3)

mat_add[blocks, threads_per_block](a_dev, b_dev, c)
print(c.copy_to_host())

其它

stride_j, stride_i = cuda.gridsize(2)

使用共享内存
tmp = cuda.shared.array((32, 32), numba.types.int32)
对共享内存的读写要注意进行同步
cuda.syncthreads()

pytorch现在提供给numba一个接口,可以使用numba直接对pytorch的gpu tensor进行操作,从而可以比较方便地实现自己想要的一些运算。

import torch
import numba.cuda

a = torch.arange(12).reshape(3,4).cuda()
a_nb = numba.cuda.as_cuda_array(a)
# 这里a_nb和a的数据是同一片内存空间的,因此a_nb对数据修改,a的数据也会同样修改,反之亦然。
# 得到a_nb后,就可以用自定义的并行运算函数进行操作了

python调用cv2.imshow出现错误The function is not implemented

错误信息为

cv2.error: OpenCV(4.2.0) C:\projects\opencv-python\opencv\modules\highgui\src\window.cpp:717: error: (-2:Unspecified error) The function is not implemented. Rebuild the library with Windows, GTK+ 2.x or Cocoa support. If you are on Ubuntu or Debian, install libgtk2.0-dev and pkg-config, then re-run cmake or configure script in function 'cvWaitKey'

解决方式是pip 安装 opencv-contrib-python
pip install opencv-contrib-python

python内置函数(三)eval、exec

python中这两个内置的函数eval和exec都可以执行str形式的python表达式,比如说”print(a + b)” 。有时候可以派上用场。

eval函数形式

eval(source, globals=None, locals=None)

其中gobals和locals分别表示全局命名空间和局部命名空间,必须是dict形式。如果不提供则使用当前调用环境下的全局命名空间和局部命名空间。eval的返回值就是其中表达式的返回值。注意eval中的表达式不能是赋值表达式!!!

a = 10
def foo():
    b = 12
    c = eval("a+b")
    print("c:",c)
    d = eval("a+b",{"a":1,"b":2})
    print("d:",d)
    e = eval("a+b",{"a":1,"b":2},{"b":3})
    print("e:",e)
    f = eval("print(a+b)")
    print("f:",f)

foo()

结果为

c: 22
d: 3
e: 4
22
f: None

exec函数形式

exec(source, globals=None, locals=None)

形式上同eval函数一样。exec函数中可以是赋值表达式。但exec与eval最主要的区别是它可以运行代码段,而eval中只能是单个表达式。 另外exec函数返回值永远是None!

snippet = 
"""
a = 1
b = 2
c = a+b
print(c)
print("this is a code snippet")
""" 

a = 10
def foo():
    b = 12
    c = exec("a+b")
    print("c:",c)
    d = exec("d=a+b\nprint('d:',d)")
    exec(snippet)

foo()

结果为

c: None
d: 22
3
this is a code snippet

python内置函数(二)dir

当参数为空时,返回当前范围内的属性、变量、类型、 方法、 导入模块的列表

dir()

['__builtins__', '__doc__', '__file__', '__loader__', '__name__', '__package__', '__spec__', 'a', 'b', 'my_class', 'my_print', 'os', 'rdn', 'sys']

包含参数时,返回参数的属性、方法列表

dir([])
[‘__add__’, ‘__class__’, ‘__contains__’, ‘__delattr__’, ‘__delitem__’, ‘__dir__’, ‘__doc__’, ‘__eq__’, ‘__format__’, ‘__ge__’, ‘__getattribute__’, ‘__getitem__’, ‘__gt__’, ‘__hash__’, ‘__iadd__’, ‘__imul__’, ‘__init__’, ‘__init_subclass__’, ‘__iter__’, ‘__le__’, ‘__len__’, ‘__lt__’, ‘__mul__’, ‘__ne__’, ‘__new__’, ‘__reduce__’, ‘__reduce_ex__’, ‘__repr__’, ‘__reversed__’, ‘__rmul__’, ‘__setattr__’, ‘__setitem__’, ‘__sizeof__’, ‘__str__’, ‘__subclasshook__’, ‘append’, ‘clear’, ‘copy’, ‘count’, ‘extend’, ‘index’, ‘insert’, ‘pop’, ‘remove’, ‘reverse’, ‘sort’]

python内置函数(一)help

help函数是python内置的一个比较有用的函数,它主要可以获取模块、数据类型、函数等的帮助信息。

1、获取模块的信息。注意如果该模块之前没有被import进来,那么需要在模块名外面加上引号。

help('os')  #获取os模块的帮助信息
import os
help(os) #等同于上面一种方式

2、获取数据类型的帮助。如果直接是数据类型的名称,那需要加上引号。或者是该数据类型的实例

help('int')
a = 1
help(a)

输出信息如下(部分)

Help on class int in module builtins:
class int(object)
| int([x]) -> integer
| int(x, base=10) -> integer
|
| Convert a number or string to an integer, or return 0 if no arguments
| are given. If x is a number, return x.int(). For floating point
| numbers, this truncates towards zero.

可以看出这些内置的数据类型(int、float、str等)其实都是类

3、获取函数的帮助信息。直接是函数名就可以

help(help)
help([].append)

os.makedirs

os.makedirs(path, exist_ok=False)

该语句可以建立用于迭代建立文件夹。exist_ok为bool值。当path不存在时,则会根据path迭代(划重点)的建立文件夹。当path存在时,若exist_ok为True时,则什么都不做,若为False,则会报文件夹已存在的错误。该语句可以很方便地替代以下语句:

if not os.path.exists(path):
os.mkdir(path)