Cython и C (СИ) расширения Python для максимальной производительности

# Этот безобидный код может заморозить ваш компьютер
total = 0
for i in range(100_000_000):
    total += i * i

# Вместо медленного цикла
total = 0
for i in range(1_000_000):
    total += i * i
    
# Используем векторизованные операции NumPy
import numpy as np
total = np.sum(np.arange(1_000_000) [B] 2)

from numba import jit
 
@jit(nopython=True)
def sum_squares(n):
    total = 0
    for i in range(n):
        total += i * i
    return total
    
# Эта функция теперь работает со скоростью C-кода!

# Обычная Python-функция
def calculate_distance(point1, point2):
    x_diff = point2[0] - point1[0]
    y_diff = point2[1] - point1[1]
    return (x_diff[B]2 + y_diff[/B]2)**0.5
 
# Эквивалентная Cython-функция
def calculate_distance(double[:] point1, double[:] point2):
    cdef double x_diff = point2[0] - point1[0]
    cdef double y_diff = point2[1] - point1[1]
    return (x_diff[B]2 + y_diff[/B]2)**0.5

# fast_math.pyx
cdef int fibonacci(int n):
    cdef int a = 0, b = 1, i, temp
    if n <= 0:
        return 0
    for i in range(n - 1):
        temp = a + b
        a = b
        b = temp
    return b
 
# Экспортируем функцию в Python
def fib(n):
    return fibonacci(n)

from setuptools import setup
from Cython.Build import cythonize
 
setup(
    ext_modules = cythonize("fast_math.pyx")
)

import fast_math
result = fast_math.fib(30)  # Оч-чень быстро!

import cProfile
cProfile.run('my_expensive_function()')

import cython
 
@cython.cfunc
def calculate_sum(a: cython.int, b: cython.int) -> cython.int:
    return a + b

# Без типизации
def sum_squares_python(n):
    result = 0
    for i in range(n):
        result += i * i
    return result
 
# С типизацией
def sum_squares_cython(int n):
    cdef int i, result = 0
    for i in range(n):
        result += i * i
    return result

# file: slow_processing.py -> slow_processing.pyx
def process_data(data_list):
result = []
for item in data_list:
    # Какие-то сложные вычисления
    processed = item * item
    result.append(processed)
return result

def process_data(list data_list):
cdef list result = []
cdef int item
for item in data_list:
    # Теперь эти операции выполняются на C-уровне
    result.append(item * item)
return result

# Python version
def matrix_sum(matrix):
total = 0
for row in matrix:
    for val in row:
        total += val
return total
 
# Cython version
def matrix_sum_fast(double[:, :] matrix):
cdef double total = 0
cdef Py_ssize_t i, j
for i in range(matrix.shape[0]):
    for j in range(matrix.shape[1]):
        total += matrix[i, j]
return total

# Python version
def fib_py(n):
a, b = 0, 1
for i in range(n):
    a, b = b, a + b
return a
 
# Cython version
def fib_cy(int n):
cdef int a = 0, b = 1, i, temp
for i in range(n):
    temp = a + b
    a = b
    b = temp
return a

# Python version
def count_vowels(text):
vowels = "aeiouAEIOU"
count = 0
for char in text:
    if char in vowels:
        count += 1
return count
 
# Cython version
def count_vowels_fast(str text):
cdef str vowels = "aeiouAEIOU"
cdef int count = 0
cdef str char
for char in text:
    if char in vowels:
        count += 1
return count

from distutils.core import setup, Extension
 
module = Extension('mymath', sources=['mymath.c'])
 
setup(
    name='MyMath',
    version='1.0',
    description='Simple math extension',
    ext_modules=[module]
)

import mymath
result = mymath.add(5, 7)  # 12

%module mymath
 
%{
#include "mymath.h"
%}
 
extern int add(int a, int b);

from ctypes import cdll, c_int
 
# Загружаем стандартную библиотеку C
libc = cdll.LoadLibrary("libc.so.6")
 
# Вызываем функцию abs
result = libc.abs(c_int(-42))
print(result)  # 42

from cffi import FFI
 
ffi = FFI()
ffi.cdef("""
 int abs(int x);
""")
lib = ffi.dlopen("libc.so.6")
 
result = lib.abs(-42)
print(result)  # 42

# Pure Python
def quicksort_py(arr):
    if len(arr) <= 1:
        return arr
    pivot = arr[len(arr) // 2]
    left = [x for x in arr if x < pivot]
    middle = [x for x in arr if x == pivot]
    right = [x for x in arr if x > pivot]
    return quicksort_py(left) + middle + quicksort_py(right)

# Cython (quicksort.pyx)
def quicksort_cy(list arr):
    cdef int length = len(arr)
    if length <= 1:
        return arr
    cdef int pivot_idx = length // 2
    cdef int pivot = arr[pivot_idx]
    cdef list left = []
    cdef list middle = []
    cdef list right = []
    cdef int i, val
    
    for i in range(length):
        val = arr[i]
        if val < pivot:
            left.append(val)
        elif val == pivot:
            middle.append(val)
        else:
            right.append(val)
            
    return quicksort_cy(left) + middle + quicksort_cy(right)

# Python
def calc_energy_py(positions, charges):
    energy = 0.0
    n = len(positions)
    for i in range(n):
        for j in range(i+1, n):
            dx = positions[i][0] - positions[j][0]
            dy = positions[i][1] - positions[j][1]
            distance = (dx[B]2 + dy[/B]2)**0.5
            if distance > 0:
                energy += charges[i] * charges[j] / distance
    return energy

# Cython
def calc_energy_cy(double[:, :] positions, double[:] charges):
    cdef double energy = 0.0
    cdef int n = positions.shape[0]
    cdef int i, j
    cdef double dx, dy, distance
    
    for i in range(n):
        for j in range(i+1, n):
            dx = positions[i, 0] - positions[j, 0]
            dy = positions[i, 1] - positions[j, 1]
            distance = (dx*dx + dy*dy)**0.5
            if distance > 0:
                energy += charges[i] * charges[j] / distance
    return energy

# Python
def logistic_regression_py(X, y, lr=0.01, iters=1000):
    m, n = X.shape
    theta = np.zeros(n)
    
    for _ in range(iters):
        z = np.dot(X, theta)
        h = 1 / (1 + np.exp(-z))
        gradient = np.dot(X.T, (h - y)) / m
        theta -= lr * gradient
        
    return theta

# Cython
def logistic_regression_cy(double[:, :] X, double[:] y, double lr=0.01, int iters=1000):
    cdef int m = X.shape[0], n = X.shape[1]
    cdef double[:] theta = np.zeros(n)
    cdef double[:] z = np.zeros(m)
    cdef double[:] h = np.zeros(m)
    cdef double[:] gradient = np.zeros(n)
    cdef int i, j, it
    
    for it in range(iters):
        # Compute z = X*theta
        for i in range(m):
            z[i] = 0
            for j in range(n):
                z[i] += X[i, j] * theta[j]
        
        # Compute sigmoid
        for i in range(m):
            h[i] = 1.0 / (1.0 + exp(-z[i]))
        
        # Compute gradient
        for j in range(n):
            gradient[j] = 0
            for i in range(m):
                gradient[j] += X[i, j] * (h[i] - y[i])
            gradient[j] /= m
            
        # Update theta
        for j in range(n):
            theta[j] -= lr * gradient[j]
            
    return np.asarray(theta)

# Python
def word_count_py(text):
    words = text.lower().split()
    counter = {}
    for word in words:
        if word in counter:
            counter[word] += 1
        else:
            counter[word] = 1
    return counter

# Cython
def word_count_cy(str text):
    cdef list words = text.lower().split()
    cdef dict counter = {}
    cdef str word
    cdef int count
    
    for word in words:
        count = counter.get(word, 0)
        counter[word] = count + 1
        
    return counter

from collections import Counter
def word_count_counter(text):
    words = text.lower().split()
    return Counter(words)

# Неправильный бенчмарк
import time
 
def benchmark_wrong():
   start = time.time()
   result = my_function()  # Вызываем один раз
   end = time.time()
   return end - start

import timeit
 
# Лучший способ для микробенчмаркинга
result = timeit.timeit("my_function()", 
                       setup="from my_module import my_function", 
                       number=1000)

# Python
def parse_csv_py(filename):
   data = []
   with open(filename, 'r') as f:
       for line in f:
           if line.strip() and not line.startswith('#'):
               values = [float(x) for x in line.split(',')]
               data.append(values)
   return data

# Cython
def parse_csv_cy(str filename):
   cdef list data = []
   cdef list values
   cdef str line
   
   with open(filename, 'r') as f:
       for line in f:
           line = line.strip()
           if line and not line.startswith('#'):
               values = [float(x) for x in line.split(',')]
               data.append(values)
   return data

# fast_file_io.pyx
from libc.stdio cimport FILE, fopen, fclose, fread, fwrite, fseek, SEEK_SET, SEEK_END
 
def read_binary_fast(filename, size_t offset=0, size_t size=0):
    cdef FILE* f
    cdef bytes py_filename = filename.encode('utf8')
    cdef char* c_filename = py_filename
    cdef size_t file_size
    cdef bytes result
    
    f = fopen(c_filename, "rb")
    if f == NULL:
        raise IOError(f"Could not open {filename}")
    
    try:
        # Определяем размер файла, если не указан
        if size == 0:
            fseek(f, 0, SEEK_END)
            file_size = ftell(f)
            size = file_size - offset
        
        # Переходим к нужной позиции
        fseek(f, offset, SEEK_SET)
        
        # Выделяем буфер и читаем данные
        result = b"\0" * size
        fread(<char*>result, 1, size, f)
        return result
    finally:
        fclose(f)

# magic_speedup.py
# cython: language_level=3
import cython
 
@cython.ccall
def intensive_calculation(x, y, iterations):
    result = 0.0
    for i in range(iterations):
        result += (x[B]2 + y[/B]2) ** 0.5
    return result

import pyximport
pyximport.install()
import magic_speedup  # Автоматически компилируется!

# fast_array_ops.pyx
import numpy as np
cimport numpy as np
 
def fast_elementwise_multiply(double[:, :] a, double[:, :] b):
    cdef int i, j
    cdef int rows = a.shape[0]
    cdef int cols = a.shape[1]
    cdef double[:, :] result = np.zeros((rows, cols), dtype=np.float64)
    
    for i in range(rows):
        for j in range(cols):
            result[i, j] = a[i, j] * b[i, j]
    
    return np.asarray(result)

# memory_tricks.pyx
from libc.stdlib cimport malloc, free
from libc.string cimport memcpy
 
def create_and_manipulate_array(int size):
    cdef int* data = <int*>malloc(size * sizeof(int))
    if not data:
        raise MemoryError("Could not allocate memory")
    
    try:
        # Заполняем массив
        for i in range(size):
            data[i] = i * i
        
        # Создаём копию для возврата в Python
        result = []
        for i in range(size):
            result.append(data[i])
        
        return result
    finally:
        # Обязательно освобождаем память!
        free(data)

# parallel_cython.pyx
# cython: boundscheck=False, wraparound=False
# cython: cdivision=True
 
cimport cython
from cython.parallel import prange
 
def parallel_sum_of_squares(double[:] arr):
    cdef int i
    cdef double total = 0.0
    cdef int n = arr.shape[0]
    
    # Распараллеливаем цикл на все доступные ядра
    with nogil:  # Важно: освобождаем GIL!
        for i in prange(n, schedule='guided'):
            total += arr[i] * arr[i]
    
    return total

# cython: boundscheck=False, wraparound=False, cdivision=True, initializedcheck=False
 
@cython.nogil
@cython.cdivision(True)
def unsafe_but_blazing_fast(double[:] a, double[:] b):
    # Этот код отключает проверку границ массивов, 
    # деление на ноль и другие проверки
    # ...

# parallel_mandelbrot.pyx
import numpy as np
cimport numpy as np
from cython.parallel import prange
 
def mandelbrot_set_parallel(int width, int height, int max_iterations, double x_min, double x_max, double y_min, double y_max):
    cdef np.ndarray[np.int32_t, ndim=2] result = np.zeros((height, width), dtype=np.int32)
    cdef int i, j, n
    cdef double x, y, x0, y0, x_temp
    cdef double dx = (x_max - x_min) / width
    cdef double dy = (y_max - y_min) / height
    
    with nogil:  # Освобождаем GIL!
        for i in prange(height, schedule='dynamic'):
            y0 = y_min + i * dy
            for j in range(width):
                x0 = x_min + j * dx
                x, y = 0, 0
                n = 0
                while (x*x + y*y <= 4.0 and n < max_iterations):
                    x_temp = x*x - y*y + x0
                    y = 2*x*y + y0
                    x = x_temp
                    n += 1
                result[i, j] = n
    
    return result

for i in prange(n, schedule='static', chunksize=1000):
    # Обработка крупных блоков данных

# Плохой вариант: потоки конфликтуют из-за доступа к shared_array
def bad_parallel(double[:] shared_array, int n_threads):
    cdef int i, thread_id
    with nogil:
        for i in prange(n_threads, schedule='static'):
            thread_id = i
            shared_array[0] += thread_id  # Конфликт!
 
# Хороший вариант: каждый поток работает со своей областью массива
def good_parallel(double[:] array, int n_threads):
    cdef int i, thread_id
    with nogil:
        for i in prange(n_threads, schedule='static'):
            thread_id = i
            array[thread_id] += thread_id  # Нет конфликта

# cuda_extension.pyx
cdef extern from "cuda_kernels.h":
    void launch_vector_add(float* a, float* b, float* c, int n)
 
def vector_add_gpu(np.ndarray[np.float32_t, ndim=1] a,
                   np.ndarray[np.float32_t, ndim=1] b):
    cdef np.ndarray[np.float32_t, ndim=1] c = np.zeros_like(a)
    cdef int n = a.shape[0]
    
    # Вызов CUDA-функции из C-заголовка
    launch_vector_add(&a[0], &b[0], &c[0], n)
    
    return c

import cProfile
import pstats
from cython_profiling_example import heavy_computation
 
# Профилирование функции
cProfile.runctx('heavy_computation(1000)', globals(), locals(), 'stats.prof')
 
# Анализ результатов
stats = pstats.Stats('stats.prof')
stats.strip_dirs().sort_stats('cumulative').print_stats(10)

# time_profiling.pyx
from libc.time cimport time_t, time
 
def benchmark_section():
    cdef time_t start, end
    
    start = time(NULL)
    # Первая критичная секция
    end = time(NULL)
    print(f"Section 1: {end - start} seconds")
    
    start = time(NULL)
    # Вторая критичная секция
    end = time(NULL)
    print(f"Section 2: {end - start} seconds")

# reactive.pyx
from libc.stdlib cimport malloc, free
 
cdef struct DataBatch:
    double* values
    int size
 
cdef class ReactiveProcessor:
    cdef object subscribers
    
    def __init__(self):
        self.subscribers = []
    
    def subscribe(self, callback):
        self.subscribers.append(callback)
        return self  # Для цепочки вызовов
    
    def process_batch(self, double[:] batch):
        cdef DataBatch* processed = self._process_internal(batch)
        
        # Уведомляем подписчиков
        result = [processed.values[i] for i in range(processed.size)]
        for subscriber in self.subscribers:
            subscriber(result)
        
        # Освобождаем память
        free(processed.values)
        free(processed)
    
    cdef DataBatch* _process_internal(self, double[:] batch) nogil:
        cdef int i
        cdef int n = batch.shape[0]
        cdef DataBatch* result = <DataBatch*>malloc(sizeof(DataBatch))
        
        result.size = n
        result.values = <double*>malloc(n * sizeof(double))
        
        # Здесь выполняем тяжёлые вычисления
        for i in range(n):
            result.values[i] = batch[i] * batch[i]  # Например, возведение в квадрат
        
        return result

# adaptive_sort.pyx
import numpy as np
cimport numpy as np
from cython.parallel import parallel, prange
 
def adaptive_sort(double[:] arr):
    cdef int n = arr.shape[0]
    
    if n < 1000:
        # Для маленьких массивов используем быстрый последовательный алгоритм
        return insertion_sort(arr)
    elif n < 100000:
        # Для средних массивов - обычный quicksort
        return quicksort(arr, 0, n-1)
    else:
        # Для больших массивов - параллельный mergesort
        return parallel_mergesort(arr)

# adaptive_parallel.pyx
import multiprocessing
from cython.parallel import parallel, prange
 
cdef int get_optimal_thread_count():
    cdef int cpu_count = multiprocessing.cpu_count()
    # Оставляем одно ядро для ОС и других задач
    return max(1, cpu_count - 1)
 
def parallel_compute(double[:] data):
    cdef int n = data.shape[0]
    cdef int num_threads = get_optimal_thread_count()
    cdef double[:] result = np.zeros_like(data)
    
    with nogil, parallel(num_threads=num_threads):
        for i in prange(n, schedule='dynamic'):
            result[i] = complex_computation(data[i])
    
    return np.asarray(result)

# resource_management.pyx
from libc.stdlib cimport malloc, free
 
cdef class Buffer:
    cdef double* data
    cdef int size
    
    def __cinit__(self, int size):
        self.size = size
        self.data = <double*>malloc(size * sizeof(double))
        if not self.data:
            raise MemoryError("Failed to allocate memory")
    
    def __dealloc__(self):
        if self.data:
            free(self.data)
    
    def set_value(self, int index, double value):
        if index < 0 or index >= self.size:
            raise IndexError("Index out of range")
        self.data[index] = value
    
    def get_value(self, int index):
        if index < 0 or index >= self.size:
            raise IndexError("Index out of range")
        return self.data[index]