Contexto
En una publicación anterior se desarrollo un código en python para generar hashes de tarjetas de crédito, con el fin de mostrar la diferencia entre integridad y confidencialidad. Igualmente, se menciona que "el código desarrollado es sólo una aproximación, dado que se puede hacer más eficiente aplicando procesamiento en paralelo". El objetivo de esta publicación es explorar el procesamiento en paralelo tanto en Python, con su última versión 3.14t que habilita paralelismo real al eliminar GIL, el cual en resumen es es un mecanismo de protección que asegura que solo un hilo de ejecución pueda ejecutar código a la vez, permitiendo simplificar la gestión de la memoria y prevenir condiciones de carrera, pero impide el verdadero paralelismo en aplicaciones multihilo, incluso en procesadores multinúcleo.
Adicionalmente se quiere introducir Rust para comparar su velocidad de procesamiento frente a Python, y más allá de eso resaltar sus capacidades como un lenguaje seguro. Rust implementa un sistema propio llamado Ownership y Borrowing (Propiedad y Préstamo) en el cual el compilador rastrea quién posee un dato en memoria, y cuando esa variable sale de su ámbito la memoria se libera inmediatamente; dando como resultado que es imposible tener errores comunes como buffer overflows, dangling pointers o double frees. También detecta las condiciones de carrera en tiempo de compilación y tiene un sistema de tipos que elimina el error del puntero nulo, al este no existe de forma nativa.
En otras palabras si el código tiene un riesgo de memoria, simplemente no compila, y teniendo en cuenta que los errores en la gestión de la memoria causan el 70 % de las vulnerabilidades, se actúa de manera preventiva en la mitigación de estas. Rust se convierte así es un lenguaje a tener en cuenta para el desarrollo de soluciones que requieran un rendimiento alto en aplicaciones críticas, como motores de browser, sistemas operativos, criptografía, drivers, entre otros.
Comparación de desempeño
A continuación se muestra el resultado del procesamiento de cálculo de hashes para python en procesamiento paralelo, rust en un solo hilo, y rust en procesamiento paralelo. El código de cada uno, se encuentra al final de este mismo post en la sección Anexos.
El siguiente gráfico detalla el rendimiento del procesamiento de los 20 lotes de datos después de implementar las técnicas de paralelismo No-GIL en Python, combinadas con la estrategia de Productor-Consumidor y batching para gestionar la alta carga de E/S. Cada barra representa el tiempo (en segundos) que tardó en procesarse, hashearse y escribirse en disco un lote de 10 millones de hashes. Se observa que la mayoría de los archivos se procesan en un rango muy estrecho de 14.2 a 14.6 segundos.
El rendimiento observado en rust en procesamiento mono-hilo muestra una diferencia fundamental en la filosofía de ejecución de ambos entornos. Mientras que la versión de Rust en un solo hilo es extremadamente rápida debido a su cercanía al hardware y la ausencia de overhead de interpretación, la versión de Python No-GIL depende totalmente de la paralelización masiva para intentar alcanzar cifras similares (en este caso en particular).
La implementación multihilo en Rust marca un punto de inflexión en el ejercicio, mientras que las versiones anteriores (Rust mono-hilo y Python No-GIL) están alrededor de los 14 segundos, la arquitectura de Productor-Consumidor optimizada en Rust muestra un rendimiento consistente, lo que indica que el software no está generando cuellos de botella, dónde ya el hardware es el que no puede ir más rápido.
Tiempos promedio por archivo y estimados para procesamiento total
| Python (1-hilo) | Rust (1-hilo) | Python (m-hilo) | Rust (m-hilo) | |
|---|---|---|---|---|
| PROMEDIO (s) | 79,47 | 13,81 | 14,48 | 3,49 |
| ESTIMADO (s) | 7947 | 1381,1 | 1448 | 349 |
| ESTIMADO (h) | 02:12:27 | 00:23:01 | 00:24:08 | 00:05:48 |
Matriz de comparación de desempeño
| Python (1-hilo) | Rust (1-hilo) | Python (m-hilo) | Rust (m-hilo) | |
|---|---|---|---|---|
| Python (1-hilo) | 100,0% | 17,4% | 18,2% | 4,4% |
| Rust (1-hilo) | 575,4% | 100,0% | 104,9% | 25,2% |
| Python (m-hilo) | 548,7% | 95,4% | 100,0% | 24,1% |
| Rust (m-hilo) | 2280,3% | 396,3% | 415,6% | 100,0% |
Conclusiones
El ejercicio de generación masiva de hashes de tarjetas de crédito nos permite extraer algunas conclusiones sobre los lenguajes de programación en cuestión y la gestión de recursos de hardware.
La superioridad de Rust no reside únicamente en su velocidad, sino en su filosofía de diseño, mientras que en otros lenguajes el rendimiento suele comprometer la seguridad (exponiendo al sistema a buffer overflows o condiciones de carrera), Rust mitiga preventivamente alrededor del 70% de las vulnerabilidades comunes mediante su sistema de Ownership, lo cual lo hace una alternativa a considerar.
Por otro lado, la eliminación del GIL en Python muestra que logra una mejora del 548% respecto a su versión mono-hilo, sin embargo en este caso, requiere de toda la potencia multihilo del procesador para apenas igualar lo que Rust logra con un solo núcleo. Esto subraya que, aunque Python es ahora capaz de paralelismo real, el overhead de su interpretación sigue siendo un factor importante frente a lenguajes compilados.
La versión de Rust multihilo es la opción ganadora, procesando cada lote en apenas 3.49 segundos de promedio, este código ha alcanzado el límite físico del hardware, siendo limitante la velocidad de transferencia del disco (E/S).
Para aplicaciones que requieren alta disponibilidad, seguridad de memoria y procesamiento masivo de datos, Rust es la opción lógica. Python, con su nueva versión No-GIL, reduce la brecha y se vuelve mucho más competitivo para tareas paralelas, pero Rust sigue manteniendo el liderazgo en eficiencia y protección proactiva contra vulnerabilidades.
Anexo: Código para Rust en un único hilo, Python multihilo y Rust multihilo
Rust en un único hilo
Se desarrollo el siguiente código en Rust para el cálculo de hashes de las tarjetas de crédito, ejecutando la misma funcionalidad desarrollada en python en el post anterior.
use sha2::{Sha256, Digest};
use std::fs::File;
use std::io::{BufWriter, Write};
use std::time::Instant;
// 1. Implementación del Algoritmo de Luhn
// A diferencia de Python, aquí trabajamos con bytes directamente, ahorrando memoria.
fn calcular_digito_luhn(numero_str: &str) -> u32 {
let mut suma_total = 0;
// Iteramos los caracteres en reverso
// .chars().rev() es un iterador perezoso (lazy), no crea una lista nueva en memoria como Python
for (i, c) in numero_str.chars().rev().enumerate() {
let mut digito = c.to_digit(10).unwrap(); // Convertir char a u32
if i % 2 == 1 {
digito *= 2;
if digito > 9 {
digito -= 9;
}
}
suma_total += digito;
}
(10 - (suma_total % 10)) % 10
}
// 2. Función para hashear (aplica SHA256)
fn aplicar_sha256(numero: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(numero.as_bytes());
hex::encode(hasher.finalize())
}
fn main() -> std::io::Result<()> {
let start_time = Instant::now();
let numero_fijo = "491511";
let batch_size = 10_000_000;
let total_limit = 1_000_000_000;
let mut batch_num = 1;
let mut count = 0;
// Bucle principal
let mut iterador = 0..total_limit;
loop {
let filename = format!("hashes_batch_{}.csv", batch_num);
// Usamos BufWriter para escribir en disco de forma eficiente (bufferizada)
let file = File::create(&filename)?;
let mut writer = BufWriter::new(file);
writeln!(writer, "Número,Hash SHA-256")?; // Escribir cabecera
for _ in 0..batch_size {
// Obtenemos el siguiente número del iterador
let num = match iterador.next() {
Some(n) => n,
None => break,
};
// Construcción del número (equivalente a str(num).zfill(9))
// format! macro es segura y verifica tipos.
let numero_cliente = format!("{:09}", num);
let base = format!("{}{}", numero_fijo, numero_cliente);
let digito_luhn = calcular_digito_luhn(&base);
let numero_completo = format!("{}{}", base, digito_luhn);
let hash_result = aplicar_sha256(&numero_completo);
writeln!(writer, "{},{}", numero_completo, hash_result)?;
count += 1;
}
// Asegurarse de que el buffer se vacíe a disco antes de cerrar el archivo
writer.flush()?;
let elapsed = start_time.elapsed();
println!("Archivo {} completo. Total procesado: {}. Tiempo parcial: {:.2}s",
batch_num, count, elapsed.as_secs_f64());
batch_num += 1;
// Si el iterador se agotó o alcanzamos el límite
if count >= total_limit {
break;
}
// Si el último batch fue parcial, salir para evitar un archivo vacío.
if total_limit - (count - batch_size) < batch_size {
break;
}
}
let duration = start_time.elapsed();
println!("Proceso completo.");
println!("Tiempo total de ejecución: {:.2} segundos", duration.as_secs_f64());
Ok(())
}
Python en procesamiento paralelo real
Ahora con la versión 3.14t se cuenta con procesamiento paralelo real, se muestra a continuación el código desarrollado.
import hashlib
import time
import os
import queue
import threading
from concurrent.futures import ThreadPoolExecutor, as_completed
# --- Funciones de lógica pura (CPU Bound) ---
def calcular_digito_luhn(numero_str):
"""Calcula el dígito verificador de Luhn."""
digitos = [int(d) for d in numero_str][::-1]
suma_total = 0
for i, digito in enumerate(digitos):
if i % 2 == 1:
digito *= 2
if digito > 9:
digito -= 9
suma_total += digito
return (10 - (suma_total % 10)) % 10
def aplicar_sha256(numero):
"""Aplica el hash SHA-256 al número."""
hash_obj = hashlib.sha256()
hash_obj.update(numero.encode())
return hash_obj.hexdigest()
# --- Función Worker para los hilos (Produce lista de listas) ---
def procesar_sub_lote(rango_numeros, numero_fijo):
"""Procesa un rango de números y devuelve los resultados."""
resultados = []
numero_fijo_str = str(numero_fijo)
for num in rango_numeros:
# Lógica de generación y hashing (CPU-bound)
numero_cliente = str(num).zfill(9)
base = numero_fijo_str + numero_cliente
digito_luhn = calcular_digito_luhn(base)
numero_completo = base + str(digito_luhn)
hash_result = aplicar_sha256(numero_completo)
# Guardamos en memoria temporal del hilo
resultados.append([numero_completo, hash_result])
return resultados
# --- Hilo Consumidor (Escritor en Disco usando F-Strings) ---
def file_writer_worker(write_queue, filename):
"""
Hilo dedicado a escribir datos del CSV en disco.
OPTIMIZACIÓN: Usa F-strings y buffering manual para I/O rápida.
"""
try:
# Usamos buffering=1024*1024 (1MB) para mejorar el rendimiento de I/O
with open(filename, 'w', newline='', buffering=1024*1024) as file:
# Escribir cabecera manualmente
file.write('Número,Hash SHA-256\n')
while True:
# Obtener el chunk (lista de listas) de la cola.
chunk = write_queue.get()
# Usamos None como valor centinela para terminar el hilo
if chunk is None:
break
# OPTIMIZACIÓN DE FORMATO: Usamos F-strings para la construcción rápida de la cadena.
csv_lines = [f'{row[0]},{row[1]}' for row in chunk]
data_string = '\n'.join(csv_lines) + '\n'
# Escritura en disco del buffer gigante
file.write(data_string)
write_queue.task_done()
except Exception as e:
print(f"Error fatal en el escritor de archivos {filename}: {e}")
# --- Lógica Principal ---
def main():
start_time = time.time()
numero_fijo = 491511
total_limit = 200_000_000
batch_size_csv = 10_000_000 # Tamaño del archivo CSV
# Tamaño del trabajo que le damos a cada hilo (Chunking)
thread_chunk_size = 100_000
# Usamos todos los núcleos disponibles
max_workers = os.cpu_count()
print(f"Ejecutando en Python {os.sys.version.split()[0]} (Verificar que sea 3.14t o similar)")
print(f"Usando {max_workers} hilos para cálculo (Sin GIL).")
print("Usando patrón Productor-Consumidor con I/O nativa optimizada.")
count_global = 0
batch_num = 1
# Iteramos para crear los archivos CSV (Lotes grandes)
for batch_start in range(0, total_limit, batch_size_csv):
filename = f'hashes_batch_{batch_num}_py.csv'
batch_end = min(batch_start + batch_size_csv, total_limit)
print(f"Generando {filename} (Rango {batch_start} - {batch_end})...")
batch_start_time = time.time()
# 1. Creamos la cola de comunicación
write_queue = queue.Queue()
# 2. Iniciamos el HILO ESCRITOR (Consumidor)
writer_thread = threading.Thread(
target=file_writer_worker,
args=(write_queue, filename)
)
writer_thread.start()
# --- CÁLCULO PARALELO (Productores) ---
with ThreadPoolExecutor(max_workers=max_workers) as executor:
futures = []
# Dividimos el lote del CSV en tareas más pequeñas
for chunk_start in range(batch_start, batch_end, thread_chunk_size):
chunk_end_adjusted = min(chunk_start + thread_chunk_size, batch_end)
rango = range(chunk_start, chunk_end_adjusted)
# Enviamos la tarea al pool
futures.append(executor.submit(procesar_sub_lote, rango, numero_fijo))
# 3. Recogemos resultados FUERA DE ORDEN y los ENVIAMOS A LA COLA
for future in as_completed(futures):
try:
# El resultado es ahora una lista de listas
resultados_chunk = future.result()
# Envía el chunk a la cola inmediatamente
write_queue.put(resultados_chunk)
count_global += len(resultados_chunk)
except Exception as e:
print(f"Error en el futuro: {e}")
# 4. Finalización del Lote: Señalar al escritor que termine
write_queue.put(None) # Centinela de terminación
writer_thread.join() # Esperar a que el escritor termine y cierre el archivo
elapsed = time.time() - batch_start_time
print(f"Archivo {batch_num} terminado en {elapsed:.2f}s. Velocidad: {batch_size_csv/elapsed:.0f} h/s")
batch_num += 1
end_time = time.time()
print("Proceso completo.")
print(f"Tiempo total: {end_time - start_time:.2f} segundos")
if __name__ == "__main__":
main()
Rust en procesamiento paralelo
Aquí se detalla el código utilizado para habilitar el procesamiento paralelo en Rust, agregando las dependencias rayon y crossbeam_channel para tal fin.
use sha2::{Sha256, Digest};
use std::fs::File;
use std::io::{BufWriter, Write};
use std::time::Instant;
// Importamos el preludio de Rayon para tener acceso a las funciones paralelas
use rayon::prelude::*;
// Usamos crossbeam-channel para canales de alto rendimiento
use crossbeam_channel::{unbounded, Receiver, Sender};
// --- CONFIGURACIÓN DE OPTIMIZACIÓN ---
// Tamaño de los lotes de datos que se envían por el canal.
// Esto reduce el overhead de la comunicación entre hilos.
const RAYON_CHUNK_SIZE: u64 = 100_000;
// --- Funciones de lógica pura (CPU Bound) ---
fn calcular_digito_luhn(numero_str: &str) -> u32 {
let mut suma_total = 0;
for (i, c) in numero_str.chars().rev().enumerate() {
let mut digito = c.to_digit(10).unwrap();
if i % 2 == 1 {
digito *= 2;
if digito > 9 { digito -= 9; }
}
suma_total += digito;
}
(10 - (suma_total % 10)) % 10
}
fn aplicar_sha256(numero: &str) -> String {
let mut hasher = Sha256::new();
hasher.update(numero.as_bytes());
hex::encode(hasher.finalize())
}
// --- Hilo Consumidor (Escritor en Disco) ---
// El canal ahora espera un vector de cadenas (Vec<String>)
fn writer_thread(rx: Receiver<Vec<String>>, filename: String) -> std::io::Result<()> {
// Usamos un buffer grande (8MB)
let file = File::create(&filename)?;
let mut writer = BufWriter::with_capacity(8 * 1024 * 1024, file);
writeln!(writer, "Número,Hash SHA-256")?;
// Consume el lote completo de resultados
while let Ok(chunk) = rx.recv() {
// Itera sobre el vector y escribe todas las líneas
for linea in chunk {
writeln!(writer, "{}", linea)?;
}
}
// El BufWriter se vacía y cierra al salir del scope
Ok(())
}
fn main() -> std::io::Result<()> {
let start_time = Instant::now();
let numero_fijo = "491511";
let batch_size = 10_000_000;
let total_limit: u64 = 200_000_000;
let mut batch_num = 1;
// Iteramos por "chunks" (lotes) del tamaño de batch_size
for start in (0..total_limit).step_by(batch_size as usize) {
let end = std::cmp::min(start + batch_size as u64, total_limit);
let filename = format!("hashes_batch_{}.csv", batch_num);
println!("Procesando lote {} (Números {} a {})...", batch_num, start, end);
let batch_start = Instant::now();
// 1. Crear el canal MPSC (Multi-Productor, Single-Consumer)
// El canal lleva ahora Vec<String>
let (tx, rx): (Sender<Vec<String>>, Receiver<Vec<String>>) = unbounded();
// 2. Iniciar el hilo de escritura (Consumidor I/O)
let writer_handle = std::thread::spawn({
let filename_cloned = filename.clone();
move || writer_thread(rx, filename_cloned)
});
// --- INICIO DE LA SECCIÓN PARALELA (Productores CPU con Batching) ---
// Creamos una lista de índices de inicio para los chunks de Rayon
let chunk_starts: Vec<u64> = (start..end).step_by(RAYON_CHUNK_SIZE as usize).collect();
// Clonamos 'tx' para que cada hilo productor pueda enviar el Vec<String>
let tx_clone_for_rayon = tx.clone();
// 3. Paralelizamos sobre los puntos de inicio de los chunks
chunk_starts.into_par_iter()
.for_each_with(tx_clone_for_rayon, |tx_cloned, chunk_start| {
let chunk_end = std::cmp::min(chunk_start + RAYON_CHUNK_SIZE, end);
// Creamos un vector para almacenar los resultados de este chunk
let mut results: Vec<String> = Vec::with_capacity((chunk_end - chunk_start) as usize);
// Calculamos secuencialmente dentro de este hilo para este chunk
// Rayon garantiza que esta iteración es ejecutada por un solo hilo
for num in chunk_start..chunk_end {
let numero_cliente = format!("{:09}", num);
let base = format!("{}{}", numero_fijo, numero_cliente);
let digito_luhn = calcular_digito_luhn(&base);
let numero_completo = format!("{}{}", base, digito_luhn);
let hash_result = aplicar_sha256(&numero_completo);
results.push(format!("{},{}", numero_completo, hash_result));
}
// Enviar el lote COMPLETO de resultados por el canal (bajo overhead)
let _ = tx_cloned.send(results);
});
// 4. Rayon ha terminado. Cerramos el transmisor principal para notificar al receptor.
drop(tx);
// 5. Esperar a que el hilo escritor termine
writer_handle.join().unwrap()?;
// --- FIN DE LA SECCIÓN PARALELA Y ESCRITURA CONCURRENTE ---
let elapsed = batch_start.elapsed();
println!("Lote {} escrito. Tiempo: {:.2}s. Velocidad: {:.0} hashes/seg",
batch_num, elapsed.as_secs_f64(), (batch_size as f64 / elapsed.as_secs_f64()));
batch_num += 1;
}
let duration = start_time.elapsed();
println!("Tiempo total de ejecución: {:.2} segundos", duration.as_secs_f64());
Ok(())
}
Top comments (0)