python--线程stack_size

源码: tests/stack_size.py

# -.- coding:utf-8 -.-
#
# thread stack_size 指的是创建线程时最小内存占用, 用于确保
# 拥有足够的内存来启动线程不会因为内存不足而报错.
#
# thread stack_size 并不意味着它不能大于指定值, 因为线程接
# 下来要做什么, 要占用多少内存, 程序是不知道的, 它知道的是
# 启动这个线程要消耗多少内存, 因此它正能保证线程能顺利启动,
# 线程启动后使用多少内存是交给操作系统来动态分配的.
#
# windows default thread stack size use 1M.  refer: https://docs.microsoft.com/en-us/windows/desktop/procthread/thread-stack-size
# linux default threads stack size use 8M.   refer: https://unix.stackexchange.com/questions/127602/default-stack-size-for-pthreads
#
# 由于python threading 模块没有提供查看默认线程堆栈大小的函数,
# 因此只能通过线程启动数量来大约评估每个线程占用多少内存空间.
#
# 在测试windows和linux的默认线程堆栈大小之前, 需要先确认指定
# 指定的堆栈大小和线程数量是否与现有操作系统内存成正比例.
from __future__ import print_function
import time
import unittest
import threading
import traceback
from psutil import virtual_memory


class SpecifiedStackSize(unittest.TestCase):

    # TODO: python2 只能运行到1.7G就会报错, 无法再创建新的线程.
    # TODO: python3 可以运行到创建预期的 1626 个线程.
    # TODO: 这是为什么, 等待以后有机会能解答.
    def test_can_launch_threads_use_all_memory(self):
        # 假设电脑有16G内存, 线程堆栈大小设定为10M
        # 那么启动一个线程就要占用10M内存,
        # 也就是大概能启动1626个线程.
        #
        # 测试办法:
        # 启动1626个线程, 如果不报错, 那么就验证成功,
        # 如果报错那么就验证失败.
        fixed_size = 10 * 1024 * 1024                        # 10M
        threading.stack_size(fixed_size)
        mem = virtual_memory()                               # 17058336768
        should_launch_threads = int(mem.total / fixed_size)  # 1626

        def the_function(argument):
            argument += 1
            time.sleep(10)                                   # 持续占用线程资源

        # print(mem)

        try:
            for i in range(should_launch_threads):
                # print(i, i * 10)
                threading.Thread(target=the_function, args=(10, )).start()
        except threading.ThreadError as e:
            traceback.print_exc(e)

        threading.stack_size(0)
        self.assertTrue(True)

测试: tests/main.py

import unittest


TEST_MODULE = [
    "ln_threading.tests.stack_size",
]


if __name__ == '__main__':
    suite = unittest.defaultTestLoader.loadTestsFromNames(TEST_MODULE)
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(suite)