DEV Community: mengqinyuan

My github projects

mengqinyuan — Fri, 25 Oct 2024 07:54:56 +0000

Recent I make a project on github:

https://github.com/mengqinyuan/RedHorse

You can make some PR or give me some suggestions.

Speed up `shutil.copytree` !

mengqinyuan — Tue, 27 Aug 2024 06:36:09 +0000

Discuss of speeding up shutil.copytree

Write here

This is a discussion on , see: https://discuss.python.org/t/speed-up-shutil-copytree/62078. If you have any ideas, send me please !

Background

shutil is a very useful moudle in Python. You can find it in github: https://github.com/python/cpython/blob/master/Lib/shutil.py

shutil.copytree is a function that copies a folder to another folder.

In this function, it calls _copytree function to copy.

What does _copytree do ?

Ignoring specified files/directories.
Creating destination directories.
Copying files or directories while handling symbolic links.
Collecting and eventually raising errors encountered (e.g., permission issues).
Replicating metadata of the source directory to the destination directory.

Problems

_copytree speed is not very fast when the numbers of files are large or the file size is large.

Test here:

import os
import shutil

os.mkdir('test')
os.mkdir('test/source')

def bench_mark(func, *args):
    import time
    start = time.time()
    func(*args)
    end = time.time()
    print(f'{func.__name__} takes {end - start} seconds')
    return end - start

# write in 3000 files
def write_in_5000_files():
    for i in range(5000):
        with open(f'test/source/{i}.txt', 'w') as f:
            f.write('Hello World' + os.urandom(24).hex())
            f.close()

bench_mark(write_in_5000_files)

def copy():
    shutil.copytree('test/source', 'test/destination')

bench_mark(copy)

The result is:

write_in_5000_files takes 4.084963083267212 seconds
copy takes 27.12768316268921 seconds

What I done

Multithreading

I use multithread to speed up the copying process. And I rename the funtion _copytree_single_threaded add a new function _copytree_multithreaded. Here is the copytree_multithreaded:

def _copytree_multithreaded(src, dst, symlinks=False, ignore=None, copy_function=shutil.copy2,
                            ignore_dangling_symlinks=False, dirs_exist_ok=False, max_workers=4):
    """Recursively copy a directory tree using multiple threads."""
    sys.audit("shutil.copytree", src, dst)

    # get the entries to copy
    entries = list(os.scandir(src))

    # make the pool
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # submit the tasks
        futures = [
            executor.submit(_copytree_single_threaded, entries=[entry], src=src, dst=dst,
                            symlinks=symlinks, ignore=ignore, copy_function=copy_function,
                            ignore_dangling_symlinks=ignore_dangling_symlinks,
                            dirs_exist_ok=dirs_exist_ok)
            for entry in entries
        ]

        # wait for the tasks
        for future in as_completed(futures):
            try:
                future.result()
            except Exception as e:
                print(f"Failed to copy: {e}")
                raise

I add a judgement to choose use multithread or not.

if len(entries) >= 100 or sum(os.path.getsize(entry.path) for entry in entries) >= 100*1024*1024:
        # multithreaded version
        return _copytree_multithreaded(src, dst, symlinks=symlinks, ignore=ignore,
                                        copy_function=copy_function,
                                        ignore_dangling_symlinks=ignore_dangling_symlinks,
                                        dirs_exist_ok=dirs_exist_ok)

else:
    # single threaded version
    return _copytree_single_threaded(entries=entries, src=src, dst=dst,
                                        symlinks=symlinks, ignore=ignore,
                                        copy_function=copy_function,
                                        ignore_dangling_symlinks=ignore_dangling_symlinks,
                                        dirs_exist_ok=dirs_exist_ok)

Test

I write 50000 files in the source folder. Bench Mark:


def bench_mark(func, *args):
    import time
    start = time.perf_counter()
    func(*args)
    end = time.perf_counter()
    print(f"{func.__name__} costs {end - start}s")

Write in:


import os
os.mkdir("Test")
os.mkdir("Test/source")

# write in 50000 files
def write_in_file():
    for i in range(50000):
         with open(f"Test/source/{i}.txt", 'w') as f:
             f.write(f"{i}")
             f.close()

Two comparing:


def copy1():
    import shutil
    shutil.copytree('test/source', 'test/destination1')

def copy2():
    import my_shutil
    my_shutil.copytree('test/source', 'test/destination2')

"my_shutil" is my modified version of shutil.

copy1 costs 173.04780609999943s
copy2 costs 155.81321870000102s

copy2 is faster than copy1 a lot. You can run many times.

Advantages & Disadvantages

Use multithread can speed up the copying process. But it will increase the memory usage. But we do not need to rewrite the multithread in the code.

Async

Thanks to "Barry Scott". I will follow his/her suggestion :

You might get the same improvement for less overhead by using async I/O.

I write these code:


import os
import shutil
import asyncio
from concurrent.futures import ThreadPoolExecutor
import time


# create directory
def create_target_directory(dst):
    os.makedirs(dst, exist_ok=True)

# copy 1 file
async def copy_file_async(src, dst):
    loop = asyncio.get_event_loop()
    await loop.run_in_executor(None, shutil.copy2, src, dst)

# copy directory
async def copy_directory_async(src, dst, symlinks=False, ignore=None, dirs_exist_ok=False):
    entries = os.scandir(src)
    create_target_directory(dst)

    tasks = []
    for entry in entries:
        src_path = entry.path
        dst_path = os.path.join(dst, entry.name)

        if entry.is_dir(follow_symlinks=not symlinks):
            tasks.append(copy_directory_async(src_path, dst_path, symlinks, ignore, dirs_exist_ok))
        else:
            tasks.append(copy_file_async(src_path, dst_path))

    await asyncio.gather(*tasks)
# choose copy method
def choose_copy_method(entries, src, dst, **kwargs):
    if len(entries) >= 100 or sum(os.path.getsize(entry.path) for entry in entries) >= 100 * 1024 * 1024:
        # async version
        asyncio.run(copy_directory_async(src, dst, **kwargs))
    else:
        # single thread version
        shutil.copytree(src, dst, **kwargs)
# test function
def bench_mark(func, *args):
    start = time.perf_counter()
    func(*args)
    end = time.perf_counter()
    print(f"{func.__name__} costs {end - start:.2f}s")

# write in 50000 files
def write_in_50000_files():
    for i in range(50000):
        with open(f"Test/source/{i}.txt", 'w') as f:
            f.write(f"{i}")

def main():
    os.makedirs('Test/source', exist_ok=True)
    write_in_50000_files()

    # 单线程复制
    def copy1():
        shutil.copytree('Test/source', 'Test/destination1')

    def copy2():
        shutil.copytree('Test/source', 'Test/destination2')

    # async
    def copy3():
        entries = list(os.scandir('Test/source'))
        choose_copy_method(entries, 'Test/source', 'Test/destination3')

    bench_mark(copy1)
    bench_mark(copy2)
    bench_mark(copy3)

    shutil.rmtree('Test')

if __name__ == "__main__":
    main()

Output:

copy1 costs 187.21s
copy2 costs 244.33s
copy3 costs 111.27s

You can see that the async version is faster than the single thread version. But the single thread version is faster than the multi-thread version. ( Maybe my test environment is not very good, you can try and send your result as a reply to me )

Thank you Barry Scott !

Advantages & Disadvantages

Async is a good choice. But no solution is perfect. If you find some problem, you can send me as a reply.

End

This is my first time to write discussion on python.org. If there is any problem, please let me know. Thank you.

My Github: https://github.com/mengqinyuan
My Dev.to: https://dev.to/mengqinyuan

How Far You Can Optimize a Program to Compute the Fibonacci Sequence?

mengqinyuan — Wed, 21 Aug 2024 05:31:31 +0000

How Far Can You Optimize a Program to Compute the Fibonacci Sequence?

Introduction

When I was learning Python, our teacher gave us a homework -- calculate the Nth number of Fibonacci Sequence.

I think it's very easy, so I write this code:

def fib(n):
    if n == 0:
        return 0
    elif n == 1:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)

Later, I know this kind of solution cost too much time.

When you read here, you may think of the Binet, but we need to consider the float problems.

$$ [ F(n) = \frac{1}{\sqrt{5}} \left( \left(\frac{1 + \sqrt{5}}{2}\right)^n - \left(\frac{1 - \sqrt{5}}{2}\right)^n \right) ] $$

Optimize a Program

I change the solution to iteration.

def fib(n):
    ls = [1,1]
    for i in range(2,n):
        ls.append(ls[i-1]+ls[i-2])

    return ls[n-1]

I use matplotlib draw the time it cost:

import time
import matplotlib.pyplot as plt


def bench_mark(func, *args):
    # calculate the time
    start = time.perf_counter()
    result = func(*args)
    end = time.perf_counter()

    return end - start, result  # return the time and the result

def fib(n):
    ls = [1,1]
    for i in range(2,n):
        ls.append(ls[i-1]+ls[i-2])

    return ls[n-1]

mark_list = []

for i in range(1,10000):
    mark = bench_mark(fib,i)
    mark_list.append(mark[0])
    print(f"size : {i} , time : {mark[0]}")

plt.plot(range(1, 10000), mark_list)
plt.xlabel('Input Size')
plt.ylabel('Execution Time (seconds)')
plt.title('Test fot fibonacci')
plt.grid(True)
plt.show()

Here is the result:

The time it cost is very short.

But I write fib(300000), cost 5.719049899998936 seconds. It's too long.

When I grow up, I learn CACHE, so I change the solution to use dict to store the result.

from functools import lru_cache
@lru_cache(maxsize=None)
def fib(n):
    if n < 2:
        return 1
    else:
        return fib(n - 1) + fib(n - 2)

Or we can write the CACHE by ourself.

def fib(n, cache={}):
    if n in cache:
        return cache[n]
    elif n < 2:
        return 1
    else:
        ls = [1, 1]
        for i in range(2, n):
            next_value = ls[-1] + ls[-2]
            ls.append(next_value)
            cache[i] = next_value
        cache[n-1] = ls[-1]
        return ls[-1]

I compare two kinds of fib with cache, and one without cache:

from functools import lru_cache
import matplotlib.pyplot as plt
import sys
sys.set_int_max_str_digits(7500)
sys.setrecursionlimit(190000)
@lru_cache(maxsize=None)
def fib_with_lru_cache(n):
    if n < 2:
        return 1
    else:
        return fib_with_lru_cache(n - 1) + fib_with_lru_cache(n - 2)

def fib_with_hand_writing_cache(n, cache={}):
    if n in cache:
        return cache[n]
    elif n < 2:
        return 1
    else:
        ls = [1, 1]
        for i in range(2, n):
            next_value = ls[-1] + ls[-2]
            ls.append(next_value)
            cache[i] = next_value
        cache[n-1] = ls[-1]
        return ls[-1]

def fib_with_no_cache(n) :
    ls = [1, 1]
    for i in range(2, n):
        next_value = ls[-1] + ls[-2]
        ls.append(next_value)
    return ls[-1]

def bench_mark(func, *args):
    import time
    start = time.perf_counter()
    func(*args)
    end = time.perf_counter()
    print(f'{func.__name__} took {end - start} seconds to run.')
    return end - start

lru_cache_time = []
hand_writing_cache_time = []
no_cache_time = []
for i in range(1, 18000):
    lru_cache_time.append(bench_mark(fib_with_lru_cache, i))
    hand_writing_cache_time.append(bench_mark(fib_with_hand_writing_cache, i))
    no_cache_time.append(bench_mark(fib_with_no_cache, i))

plt.plot(lru_cache_time, label='lru_cache')
plt.plot(hand_writing_cache_time, label='hand_writing_cache')
plt.plot(no_cache_time, label='no_cache')
plt.legend()
plt.show()

I calculate from fib(1) to fib(18000) in order to see the difference clearly.

Here is the result:

Later, I know the MATRIX EXPONENTIAL, so I change the solution to use matrix.
Here is the code:


def matrix_power(matrix, power):
    if power == 1:
        return matrix
    elif power % 2 == 0:
        half_power = matrix_power(matrix, power // 2)
        return matrix_multiply(half_power, half_power)
    else:
        return matrix_multiply(matrix, matrix_power(matrix, power - 1))

def matrix_multiply(a, b):
    a11, a12, a21, a22 = a
    b11, b12, b21, b22 = b
    return (
        a11 * b11 + a12 * b21,
        a11 * b12 + a12 * b22,
        a21 * b11 + a22 * b21,
        a21 * b12 + a22 * b22
    )

def fibonacci(n):
    if n <= 1:
        return n
    fib_matrix = (1, 1, 1, 0)
    result_matrix = matrix_power(fib_matrix, n - 1)
    return result_matrix[0] + result_matrix[2]

I make the compare with the previous solution:

What is the next step?

There are many kinds of solution to solve this problem. I think the next step is to learn more about the algorithm.
And do not give up finding the solution. Maybe you can have more ideas.

Additional

A person named "Jon Randy 🎖️" add my idea, here is the reply :

JS function to calculate Nth Fibonacci number in O(1) - well, maybe O(log N) - depends on how the system implements some of the underlying maths functions:


const fibonacci = n => {
  const r5 = 5**.5
  return Math.floor((((1+r5)/2)**n - ((1-r5)/2)**n)/r5)
}

And HE/SHE gives a explanation (very detailed) :
https://akuli.github.io/math-tutorial/fib.html

End

My Github : https://github.com/mengqinyuan
My Dev.to : https://dev.to/mengqinyuan

Python PDB Code Review Report

mengqinyuan — Wed, 07 Aug 2024 10:30:37 +0000

Python PDB Code Review Report

TIME: 2024/08/07
AUTHOR: QINYUAN MENG
EMAIL: njbj1210@sina.com
GITHUB: https://github.com/mengqinyuan/
DEV.TO： https://dev.to/mengqinyuan/

Code Review Report

FILE_LOCATION: C:\Program Files\WindowsApps\PythonSoftwareFoundation.Python.3.11_3.11.2544.0_x64__qbz5n2kfra8p0\Lib\bdb.py

USAGE

enable, disable: Toggle a breakpoint's activity.
bpprint: Print breakpoint information.
bpformat: Format breakpoint details as a string.
__str__: Return a concise breakpoint description.
checkfuncname: Determine if a breakpoint should be set based on function name or line number.
effective: Decide if a breakpoint should be set at a given file and line, and if it's temporary.

I find some problems in the code.

Problems

The provided code snippet defines a Breakpoint class for managing breakpoints, along with helper functions and a test case. Here are the translations and refinements for the identified potential issues and optimization directions:

Potential Issues and Optimization Directions

1. `del self.bplist[index]`

Potential Issue:

bplist is a class attribute, and directly deleting elements from it may lead to unexpected results, especially in multi-threaded environments.

Optimization Suggestion:

Ensure that access to and modification of bplist are thread-safe, or explicitly state that the class does not support multi-threading.

2. `bpprint` method using `sys.stdout`

Potential Issue:

Directly using sys.stdout can cause output to be mixed with that of external callers.

Optimization Suggestion:

Provide an option to specify the output stream, allowing users to direct the output to a specific location.

3. Static Methods and Class Attributes

Potential Issue:

Static methods and class attributes like Breakpoint.clearBreakpoints() and Breakpoint.next can lead to shared state issues between different instances of Bdb.

Optimization Suggestion:

Consider using instance attributes and methods instead of static methods and class attributes to avoid shared state issues.

4. Exception Handling in `effective` Function

Potential Issue:

The exception handling in the effective function catches all exceptions, which might not be ideal.

Optimization Suggestion:

Catch specific exceptions to handle them appropriately, and log the exception details for debugging purposes.

5. Use of `sys.settrace` and `BdbQuit`

Potential Issue:

The use of sys.settrace and raising BdbQuit can interfere with the normal flow of the program.

Optimization Suggestion:

Document the implications of using these mechanisms and provide guidance on how to properly integrate the debugger into applications.

6. Test Case Implementation

Potential Issue:

The test case (test) uses a global Tdb instance, which might not be suitable for all scenarios.

Optimization Suggestion:

Encapsulate the test case within a function or a class to ensure that the test environment is isolated and does not affect other parts of the application.

New Code

bpformat function:

def bpformat(self):
    """Return a string with information about the breakpoint."""
    disp = f'del  ' if self.temporary else f'keep '
    disp += 'yes  ' if self.enabled else 'no   '
    ret = f'{self.number:<4}breakpoint   {disp}at {self.file}:{self.line}'
    if self.cond:
        ret += f'\n\tstop only if {self.cond}'
    if self.ignore:
        ret += f'\n\tignore next {self.ignore} hits'
    if self.hits:
        ss = 's' if self.hits > 1 else ''
        ret += f'\n\tbreakpoint already hit {self.hits} time{ss}'
    return ret

effective function:

def effective(file, line, frame):
    """Return (active breakpoint, delete temporary flag) or (None, None) as
       breakpoint to act upon.
    """
    possibles = Breakpoint.bplist[file, line]
    for b in possibles:
        if not b.enabled:
            continue
        if not checkfuncname(b, frame):
            continue
        b.hits += 1
        if not b.cond:
            if b.ignore > 0:
                b.ignore -= 1
                continue
            return (b, True)
        else:
            try:
                val = eval(b.cond, frame.f_globals, frame.f_locals)
                if val:
                    if b.ignore > 0:
                        b.ignore -= 1
                        continue
                    return (b, True)
            except NameError as e:
                print(f"Error evaluating condition: {e}")
                return (b, False)
    return (None, None)

Summary

This analysis provides insights into potential issues and optimization directions for the Breakpoint class and related functionalities. Implementing the suggested optimizations can improve the robustness and maintainability of the code.

“Be wary of time complexity pitfalls"

mengqinyuan — Thu, 01 Aug 2024 09:57:57 +0000

Be wary of time complexity pitfalls

Write Here

A bilibili vedio also shows this : https://www.bilibili.com/video/BV16u4m1c7cU/?spm_id_from=333.999.0.0 and I think it's a good vedio, but its language is Chinese.

time complexity

What does time complexity mean?
Time complexity is a measure of how much time an algorithm takes to run as a function of the size of its input. It is a way to describe the efficiency of an algorithm, and it is used to compare different algorithms and determine which one is the most efficient.
How to calculate time complexity?
To calculate time complexity, we need to consider the number of operations performed by the algorithm as a function of the size of the input. The most common way to measure the number of operations is by counting the number of times a particular operation is performed.
What are some common pitfalls when calculating time complexity?
1. Not considering the input size: If we only consider the number of operations performed by the algorithm, we may underestimate the time complexity. For example, if we count the number of times a loop runs, but we don't take into account the size of the input, then the time complexity may be too high.
1. Not considering the algorithm's efficiency: Some algorithms may perform many operations even when the input size is small, which can lead to high time complexity. For example, sorting algorithms like bubble sort and selection sort have quadratic time complexity, even though they may only need to swap two elements in a small array.
1. Not considering the algorithm's worst-case scenario: Some algorithms have a best-case scenario where they perform fewer operations than the worst-case scenario. For example, searching algorithms like binary search have a best-case scenario where they find the target element in logarithmic time, but they have a worst-case scenario where they need to examine all elements in the array.

Let's see some examples of time complexity

Here is a question:
Find the max 10 integers in a list.

import random
ls = [random.randint(1, 100) for _ in range(n)]

Here is the test function:

import time
def benchmark(func, *args) -> float:
    # bench mark for testing
    start = time.perf_counter()
    func(*args)
    end = time.perf_counter()
    return end - start

Solution1: Use heap

Here is the solution which uses the heapq module:

def find_max_n(ls, n):
    import heapq
    return heapq.nlargest(n, ls)

Or we write it in the python way:

def find_largest_n(nums, n):
    if n <= 0:
        return []

    max_heap = []

    for num in nums:
        if len(max_heap) < n:
            max_heap.append(num)
            # call sift_down
            for i in range((len(max_heap) - 2) // 2, -1, -1):
                _sift_down(max_heap, i)
        elif num > max_heap[0]:
            max_heap[0] = num
            _sift_down(max_heap, 0)

    return max_heap

def _sift_down(heap, index):
    left = 2 * index + 1
    right = 2 * index + 2
    largest = index

    if left < len(heap) and heap[left] > heap[largest]:
        largest = left

    if right < len(heap) and heap[right] > heap[largest]:
        largest = right

    if largest != index:
        heap[index], heap[largest] = heap[largest], heap[index]
        _sift_down(heap, largest)

Solution2: Use sort

Here is the solution which uses the sort function:

def find_max_n(ls, n):
    return sorted(ls, reverse=True)[:n]

Testing

Almost everyone know that The time complexity of the heap, is O(n log k), and the time complexity of the sort function is O(n log n).

It seems that the First solution is better than the second one. But it is not true in python.

Look at the final code:

import time
def benchmark(func, *args) -> float:
    start = time.perf_counter()
    func(*args)
    end = time.perf_counter()
    return end - start

def find_max_n_heapq(ls, n):
    import heapq
    return heapq.nlargest(n, ls)

def find_max_n_python_heap(nums, n):
    if n <= 0:
        return []

    max_heap = []

    for num in nums:
        if len(max_heap) < n:
            max_heap.append(num)
            # call sift_down
            for i in range((len(max_heap) - 2) // 2, -1, -1):
                _sift_down(max_heap, i)
        elif num > max_heap[0]:
            max_heap[0] = num
            _sift_down(max_heap, 0)

    return max_heap

def _sift_down(heap, index):
    left = 2 * index + 1
    right = 2 * index + 2
    largest = index

    if left < len(heap) and heap[left] > heap[largest]:
        largest = left

    if right < len(heap) and heap[right] > heap[largest]:
        largest = right

    if largest != index:
        heap[index], heap[largest] = heap[largest], heap[index]
        _sift_down(heap, largest)


def find_max_n_sorted(ls, n):
    return sorted(ls, reverse=True)[:n]

# test
import random
for n in range(10, 10000, 100):
    ls = [random.randint(1, 100) for _ in range(n)]
    print(f"n = {n}")
    print(f"Use    Heapq: {benchmark(find_max_n_heapq, ls, n)}")
    print(f"Python Heapq: {benchmark(find_max_n_python_heap, ls, n)}")
    print(f"Sorted      : {benchmark(find_max_n_sorted, ls, n)}")

I run it in Python3.11 VScode, Here is the result:

n is not big

n = 900
Use Heapq: 0.002430099993944168
Python Heapq: 0.06343129999004304
Sorted : 9.280000813305378e-02
n = 910
Use Heapq: 9.220000356435776e-05
Python Heapq: 0.07715500006452203
Sorted : 9.360001422464848e-05
n = 920
Use Heapq: 9.440002031624317e-05
Python Heapq: 0.06573969998862594
Sorted : 0.00012450001668185
n = 930
Use Heapq: 9.689992293715477e-05
Python Heapq: 0.06760239996947348
Sorted : 9.66999214142561e-05
n = 940
Use Heapq: 9.600003249943256e-05
Python Heapq: 0.07372559991199523

Sorted : 9.680003859102726e-05

n = 950
Use Heapq: 9.770004544407129e-05
Python Heapq: 0.07306570000946522
Sorted : 0.000119799980893731
n = 960
Use Heapq: 9.980006143450737e-05
Python Heapq: 0.0771204000338912
Sorted : 0.00022829999215900898
n = 970
Use Heapq: 0.0001601999392732978
Python Heapq: 0.07493270002305508
Sorted : 0.00010840001050382853
n = 980
Use Heapq: 9.949994273483753e-05
Python Heapq: 0.07698320003692061
Sorted : 0.00010300008580088615
n = 990
Use Heapq: 9.979994501918554e-05
Python Heapq: 0.0848745999392122
Sorted : 0.00012620002962648869

if n is big ?

n = 10000
Use Heapq: 0.003642000025138259
Python Heapq: 9.698883199947886
Sorted : 0.0010799999581649
n = 11000
Use Heapq: 0.0014836000045761466
Python Heapq: 10.537632800056599
Sorted : 0.0012236000038683414
n = 12000
Use Heapq: 0.001384599949233234
Python Heapq: 12.328411899972707
Sorted : 0.001322699943557381
n = 13000
Use Heapq: 0.0020017001079395413
Python Heapq: 15.637207800056785
Sorted : 0.0015075999544933438
n = 14000
Use Heapq: 0.0017026999266818166
Python Heapq: 17.298848500009626
Sorted : 0.0016967999981716275
n = 15000
Use Heapq: 0.0017773000290617347
Python Heapq: 20.780625900020823
Sorted : 0.0017105999868363142

What I find & how to improve it

When n is big, Sorted costs a little time (sometimes even better than use heapq), but Python Heapq costs a lot of time.

Why Sorted costs a little time, but Python Heap costs a lot of time?
Because sorted() is a bulit-in function in Python, you can find python office document about it.

Bulit-in function is faster than heapq because it is written in C, which is a compiled language.

How to improve it?
You can use the built-in function sorted() instead of heapq.nlargest() to improve the performance of your code. The sorted() function is a built-in function in Python, which is implemented in C and is therefore much faster than heapq.sort()

Maybe heapq.nlargest is faster in some situatios, but I cannot make sure that.

What about use other language?

Use other language like C++, C#, Java, etc.

#include <iostream>
#include <vector>
#include <algorithm>
#include <chrono>
#include <random>

// C++ benchmark function
double benchmark(std::function<void(std::vector<int>, int)> func, std::vector<int> ls, int n) {
    auto start = std::chrono::high_resolution_clock::now();
    func(ls, n);
    auto end = std::chrono::high_resolution_clock::now();
    return std::chrono::duration<double>(end - start).count();
}

// Using std::partial_sort to find the n largest elements
void find_max_n_sorted(std::vector<int>& ls, int n) {
    std::partial_sort(ls.begin(), ls.begin() + n, ls.end(), std::greater<int>());
    ls.resize(n);
}

// Manual heap implementation
void _sift_down(std::vector<int>& heap, int index) {
    int left = 2 * index + 1;
    int right = 2 * index + 2;
    int largest = index;

    if (left < heap.size() && heap[left] > heap[largest])
        largest = left;

    if (right < heap.size() && heap[right] > heap[largest])
        largest = right;

    if (largest != index) {
        std::swap(heap[index], heap[largest]);
        _sift_down(heap, largest);
    }
}

std::vector<int> find_max_n_python_heap(std::vector<int> nums, int n) {
    if (n <= 0)
        return {};

    std::vector<int> max_heap;

    for (int num : nums) {
        if (max_heap.size() < n) {
            max_heap.push_back(num);
            for (int i = (max_heap.size() - 2) / 2; i >= 0; --i)
                _sift_down(max_heap, i);
        } else if (num > max_heap[0]) {
            max_heap[0] = num;
            _sift_down(max_heap, 0);
        }
    }

    return max_heap;
}

int main() {
    std::mt19937 gen(std::random_device{}());
    std::uniform_int_distribution<> dis(1, 100);

    for (int n = 10; n <= 10000; n += 100) {
        std::vector<int> ls(n);
        std::generate(ls.begin(), ls.end(), [&]() { return dis(gen); });

        std::vector<int> ls_copy1 = ls;
        std::vector<int> ls_copy2 = ls;

        std::cout << "n = " << n << std::endl;
        std::cout << "Use    Heapq: " << benchmark(find_max_n_sorted, ls_copy1, n) << std::endl;
        std::cout << "Python Heapq: " << benchmark(find_max_n_python_heap, ls_copy2, n) << std::endl;

        // Note: C++ std::nth_element or std::partial_sort could be used here for comparison
    }

    return 0;
}

import java.util.*;
import java.time.Duration;
import java.time.Instant;

public class MaxElementsBenchmark {

    public static double benchmark(Runnable func) {
        Instant start = Instant.now();
        func.run();
        Instant end = Instant.now();
        return Duration.between(start, end).toNanos() / 1e9;
    }

    public static List<Integer> findMaxNHeap(ArrayList<Integer> ls, int n) {
        PriorityQueue<Integer> minHeap = new PriorityQueue<>();
        for (int num : ls) {
            if (minHeap.size() < n) {
                minHeap.offer(-num);
            } else if (-num > minHeap.peek()) {
                minHeap.poll();
                minHeap.offer(-num);
            }
        }
        List<Integer> result = new ArrayList<>(minHeap);
        Collections.sort(result);
        Collections.reverse(result);
        return result;
    }

    public static List<Integer> findMaxNManualHeap(ArrayList<Integer> nums, int n) {
        if (n <= 0) return new ArrayList<>();

        ArrayList<Integer> maxHeap = new ArrayList<>();
        for (int num : nums) {
            if (maxHeap.size() < n) {
                maxHeap.add(num);
                siftDown(maxHeap, maxHeap.size() - 1);
            } else if (num > maxHeap.get(0)) {
                maxHeap.set(0, num);
                siftDown(maxHeap, 0);
            }
        }
        return maxHeap;
    }

    private static void siftDown(ArrayList<Integer> heap, int index) {
        int left = 2 * index + 1;
        int right = 2 * index + 2;
        int largest = index;

        if (left < heap.size() && heap.get(left) > heap.get(largest))
            largest = left;

        if (right < heap.size() && heap.get(right) > heap.get(largest))
            largest = right;

        if (largest != index) {
            Collections.swap(heap, index, largest);
            siftDown(heap, largest);
        }
    }

    public static List<Integer> findMaxNSorted(ArrayList<Integer> ls, int n) {
        Collections.sort(ls, Collections.reverseOrder());
        return ls.subList(0, Math.min(n, ls.size()));
    }

    public static void main(String[] args) {
        Random rand = new Random();
        for (int n = 10; n <= 10000; n += 100) {
            ArrayList<Integer> ls = new ArrayList<>();
            for (int i = 0; i < n; i++) {
                ls.add(rand.nextInt(100) + 1);
            }

            System.out.println("n = " + n);
            System.out.println("Use    Heapq: " + benchmark(() -> findMaxNHeap(ls, n)));
            System.out.println("Python Heapq: " + benchmark(() -> findMaxNManualHeap(ls, n)));
            System.out.println("Sorted      : " + benchmark(() -> findMaxNSorted(ls, n)));
        }
    }
}

The result is different from the C++ version. The C++ version is faster than the Java version. This is because the C++ version uses a built-in function, while the Java version uses a manual heap implementation.

Conclusion

When we are dealing with large data, we should use built-in function instead of heapq.sort() to improve the performance of our code. We must be wary of time complexity pitfalls when dealing with large data. Sometimes time complexity pitfalls are unavoidable, but we should try to avoid them as much as possible.

About Me

Hello, I'm mengqinyuan. I'm a student. I love to learn new things.
You can see my github: [MengQinYuan's Github][https://github.com/mengqinyuan]

Other things

Eight Queens Problem: [https://www.geeksforgeeks.org/8-queen-problem/]
Time & Space Complexity: [https://www.geeksforgeeks.org/time-complexity-and-space-complexity/]
Heap: [https://www.geeksforgeeks.org/heap-data-structure/]

DEV Community: mengqinyuan

My github projects

Speed up `shutil.copytree` !

Discuss of speeding up shutil.copytree

Write here

Background

Problems

What I done

Multithreading

Test

Advantages & Disadvantages

Async

Advantages & Disadvantages

End

How Far You Can Optimize a Program to Compute the Fibonacci Sequence?

How Far Can You Optimize a Program to Compute the Fibonacci Sequence?

Introduction

Optimize a Program

What is the next step?

Additional

End

Python PDB Code Review Report

Python PDB Code Review Report

Code Review Report

USAGE

Problems

Potential Issues and Optimization Directions

1. del self.bplist[index]

2. bpprint method using sys.stdout

3. Static Methods and Class Attributes

4. Exception Handling in effective Function

5. Use of sys.settrace and BdbQuit

6. Test Case Implementation

New Code

Summary

“Be wary of time complexity pitfalls"

Be wary of time complexity pitfalls

Write Here

time complexity

Let's see some examples of time complexity

Solution1: Use heap

Solution2: Use sort

Testing

n is not big

Sorted : 9.680003859102726e-05

if n is big ?

What I find & how to improve it

What about use other language?

Conclusion

About Me

Other things

1. `del self.bplist[index]`

2. `bpprint` method using `sys.stdout`

4. Exception Handling in `effective` Function

5. Use of `sys.settrace` and `BdbQuit`