Python GIL感知程式設計最佳實踐

import multiprocessing as mp
import numpy as np

def heavy_compute(chunk):
    # 對chunk執行CPU密集型計算
    return np.sum(np.sqrt(np.abs(chunk)))

def parallel_process(data, num_workers):
    chunk_size = len(data) // num_workers
    chunks = [data[i * chunk_size:(i + 1) * chunk_size] for i in range(num_workers)]
    with mp.Pool(processes=num_workers) as pool:
        results = pool.map(heavy_compute, chunks)
    return np.sum(results)

if __name__ == "__main__":
    data = np.random.randn(10**7)
    result = parallel_process(data, mp.cpu_count())
    print("最終結果：", result)

import asyncio
import aiohttp
from concurrent.futures import ThreadPoolExecutor

async def fetch(url):
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as response:
            return await response.text()

def parse_html(html):
    # CPU密集型的HTML解析可以解除安裝到執行緒池
    import re
    match = re.search(r'<title>(.*?)</title>', html)
    return match.group(1) if match else "無標題"

async def process_url(url, loop, executor):
    html = await fetch(url)
    title = await loop.run_in_executor(executor, parse_html, html)
    return title

async def main(urls):
    loop = asyncio.get_running_loop()
    with ThreadPoolExecutor(max_workers=4) as executor:
        tasks = [process_url(url, loop, executor) for url in urls]
        return await asyncio.gather(*tasks)

if __name__ == "__main__":
    urls = [
        "https://www.example.com",
        "https://www.python.org",
        "https://www.github.com",
        "https://www.stackoverflow.com"
    ]
    titles = asyncio.run(main(urls))
    for i, title in enumerate(titles):
        print(f"URL {i+1} 標題：{title}")

import asyncio

async def fetch_data(url):
    # 模擬網路請求
    await asyncio.sleep(1)
    return f"Data from {url}"

async def main():
    urls = ["http://example.com/page1", "http://example.com/page2", "http://example.com/page3"]
    tasks = [fetch_data(url) for url in urls]
    results = await asyncio.gather(*tasks)
    for result in results:
        print(result)

asyncio.run(main())

import multiprocessing
import time

def compute_sqrt(n):
    result = sum(i**0.5 for i in range(n))
    return result

def main():
    numbers = [10**7, 10**7, 10**7, 10**7]
    start_time = time.time()
    
    # 順序執行
    # results = [compute_sqrt(n) for n in numbers]
    
    # 平行執行
    with multiprocessing.Pool(processes=4) as pool:
        results = pool.map(compute_sqrt, numbers)
    
    print(f"Results: {results}")
    print(f"Time taken: {time.time() - start_time} seconds")

if __name__ == "__main__":
    main()

# cython: boundscheck=False, wraparound=False, nonecheck=False
cimport cython
from libc.math cimport sqrt

@cython.boundscheck(False)
@cython.wraparound(False)
def compute_sqrt(int n):
    cdef int i
    cdef double result = 0.0
    with nogil:
        for i in range(n):
            result += sqrt(i)
    return result

import threading
from ctypes import c_void_p

class Node:
    def __init__(self, value):
        self.value = value
        self.next = None

class LockFreeQueue:
    def __init__(self):
        self.head = Node(None)  # Dummy head node for simplified design
        self.tail = self.head
    
    def enqueue(self, value):
        new_node = Node(value)
        while True:
            tail = self.tail
            next_node = tail.next
            if tail is self.tail:  # Confirm consistency of tail reference
                if next_node is None:
                    # Attempt to link the new node at the tail
                    if self._cas_next(tail, None, new_node):
                        self._cas_tail(tail, new_node)
                        return
                else:
                    # Tail is behind, advance it
                    self._cas_tail(tail, next_node)
    
    def dequeue(self):
        while True:
            head = self.head
            tail = self.tail
            next_node = head.next
            if head is self.head:
                if head is tail:
                    if next_node is None:
                        return None  # Queue is empty
                    self._cas_tail(tail, next_node)
                else:
                    value = next_node.value
                    if self._cas_head(head, next_node):
                        return value
    
    def _cas_next(self, node, old, new):
        # Placeholder for the atomic compare-and-swap on node.next
        if node.next is old:
            node.next = new
            return True
        return False
    
    def _cas_tail(self, old, new):
        # Placeholder for the atomic compare-and-swap on self.tail
        if self.tail is old:
            self.tail = new
            return True
        return False
    
    def _cas_head(self, old, new):
        # Placeholder for the atomic compare-and-swap on self.head
        if self.head is old:
            self.head = new
            return True
        return False