Traitement de données à haute fréquence sur des serveurs dédiés

Le traitement de données à haute fréquence a une définition précise : il s'agit de systèmes qui doivent ingérer, traiter et exploiter des flux de données à des débits qui dépassent les capacités d'une infrastructure d'hébergement web classique. Des flux de données des marchés financiers arrivant à un rythme de 100 000 mises à jour par seconde, des réseaux de capteurs industriels transmettant simultanément les données télémétriques de milliers d'appareils, des pipelines d'agrégation en temps réel qui doivent réduire des millions d'événements par minute en résumés consultables : ces charges de travail nécessitent du matériel bare metal dédié pour des raisons qui vont bien au-delà de la simple capacité du processeur.

Pourquoi le « bare metal » est-il si important pour les charges de travail à haute fréquence ?

L'infrastructure virtualisée introduit une latence non déterministe aux pires moments possibles dans un pipeline de traitement à haute fréquence. C'est le planificateur de l'hyperviseur qui décide quand les processeurs virtuels de la machine virtuelle s'exécutent. Sous charge, une machine virtuelle qui se dispute le temps CPU physique avec d'autres locataires subit des retards de planification de 1 à 10 ms. Pour les applications web, une variation de 5 ms dans la planification est imperceptible. Pour un processeur de flux de données financières qui doit réagir aux événements du marché en moins d'une milliseconde, une variation de 5 ms dans la planification est un problème rédhibitoire.

Les serveurs dédiés « bare metal » suppriment complètement la couche d'hyperviseur. Tes processus s'exécutent directement sur le processeur physique, sans intermédiaire de planification. Associés aux options du noyau Linux en temps réel, à l'affinité CPU fixe et à l'allocation de mémoire adaptée au modèle NUMA, les serveurs dédiés peuvent atteindre une latence de traitement inférieure à la milliseconde pour les charges de travail à haute fréquence, ce que les infrastructures virtualisées ne peuvent pas égaler de manière fiable.

La documentation sur l'architecture des processeurs EPYC d'AMD indique que la conception en chiplets du 4545P garantit une latence d'accès à la mémoire homogène sur tous les cœurs – ce qui est particulièrement important pour les charges de travail à haute fréquence sensibles au modèle NUMA, où les schémas d'accès à la mémoire peuvent avoir une influence déterminante sur le temps de traitement.

Cas d'utilisation n° 1 : Flux de données sur les marchés financiers

Les fournisseurs de données financières (Bloomberg, Refinitiv, CME Group) publient des données de marché à des cadences qui nécessitent une infrastructure de traitement dédiée. Un flux de données boursières en période de forte activité peut générer entre 50 000 et 500 000 mises à jour par seconde pour des milliers d'instruments.

Conditions de traitement :

• Pile réseau à faible latence : les solutions de mise en réseau contournant le noyau (DPDK, RDMA) éliminent la surcharge liée à la pile TCP pour les implémentations les plus sensibles à la latence ; la mise en réseau standard via le noyau suffit pour la plupart des cas d'utilisation en dessous d'un million de messages par seconde
• Structures de données sans verrouillage : les files d'attente traditionnelles basées sur des mutex génèrent des conflits d'accès lorsque le débit de messages est élevé ; les tampons circulaires sans verrouillage permettent aux threads producteurs et consommateurs de fonctionner sans se bloquer
• Affinité CPU : attribue le thread de réception réseau et les threads de traitement à des cœurs de processeur spécifiques pour éliminer les fluctuations dans la planification

Implémentation Python de base d'une file d'attente de messages à haut débit utilisant le multiprocessing :

import multiprocessing as mp
from collections import deque
import time
 
class HighFrequencyProcessor:
    def __init__(self, num_workers=8):
        self.queue = mp.Queue(maxsize=100000)
        self.results = mp.Queue()
        self.workers = []
        
        # Pin workers to specific cores for consistent latency
        for i in range(num_workers):
            p = mp.Process(
                target=self._worker,
                args=(self.queue, self.results, i),
                daemon=True
            )
            p.start()
            self.workers.append(p)
    
    def _worker(self, queue, results, worker_id):
        # Set CPU affinity if psutil available
        try:
            import psutil
            psutil.Process().cpu_affinity([worker_id % mp.cpu_count()])
        except ImportError:
            pass
        
        while True:
            try:
                message = queue.get(timeout=0.001)
                result = self._process_message(message)
                results.put(result)
            except Exception:
                continue
    
    def _process_message(self, message):
        # Application-specific processing logic
        return {
            'timestamp': time.time_ns(),
            'symbol': message.get('symbol'),
            'price': message.get('price'),
            'processed': True
        }
    
    def ingest(self, message):
        try:
            self.queue.put_nowait(message)
            return True
        except mp.queues.Full:
            # Queue full - implement backpressure or drop strategy
            return False

import multiprocessing as mp
from collections import deque
import time
 
class HighFrequencyProcessor:
    def __init__(self, num_workers=8):
        self.queue = mp.Queue(maxsize=100000)
        self.results = mp.Queue()
        self.workers = []
        
        # Pin workers to specific cores for consistent latency
        for i in range(num_workers):
            p = mp.Process(
                target=self._worker,
                args=(self.queue, self.results, i),
                daemon=True
            )
            p.start()
            self.workers.append(p)
    
    def _worker(self, queue, results, worker_id):
        # Set CPU affinity if psutil available
        try:
            import psutil
            psutil.Process().cpu_affinity([worker_id % mp.cpu_count()])
        except ImportError:
            pass
        
        while True:
            try:
                message = queue.get(timeout=0.001)
                result = self._process_message(message)
                results.put(result)
            except Exception:
                continue
    
    def _process_message(self, message):
        # Application-specific processing logic
        return {
            'timestamp': time.time_ns(),
            'symbol': message.get('symbol'),
            'price': message.get('price'),
            'processed': True
        }
    
    def ingest(self, message):
        try:
            self.queue.put_nowait(message)
            return True
        except mp.queues.Full:
            # Queue full - implement backpressure or drop strategy
            return False

Pour les applications où la latence de l'ordre de la microseconde est cruciale, Rouille c'est le langage de prédilection sous Linux. Son modèle de gestion de la mémoire élimine les pauses liées au ramasse-miettes qui, autrement, provoqueraient des pics de latence imprévisibles aux moments les plus inopportuns. Le modèle de tampon circulaire de LMAX Disruptor propose une architecture de file d'attente sans verrouillage qui a fait ses preuves, avec des implémentations open source disponibles en Java (l'implémentation de référence) et en Rust. C'est parti est une alternative pratique pour les équipes qui ont besoin d'un débit en temps quasi réel avec des primitives de concurrence plus simples ; son goroutine Le planificateur gère des milliers de gestionnaires de messages simultanés sans nécessiter la gestion manuelle des threads qu'exige Python.

Cas d'utilisation n° 2 : réseaux de capteurs industriels

Les réseaux de capteurs IoT issus des équipements de production, des infrastructures de réseaux intelligents ou des systèmes de surveillance environnementale génèrent d'énormes volumes de données télémétriques qui doivent être collectées, validées et agrégées en temps réel.

Un déploiement IoT industriel typique peut compter 10 000 capteurs transmettant des données toutes les secondes – soit 10 000 messages par seconde en flux continu, avec des pics lors des événements de détection d'anomalies. Le traitement de chaque message implique la normalisation de l'horodatage, la conversion des unités, la validation des plages de valeurs et l'agrégation dans un système de stockage de séries chronologiques.

InfluxDB est la base de données de séries chronologiques de référence pour les données de capteurs à haute fréquence. Son format de protocole « line » est optimisé pour les écritures à haut débit :

# Enregistrer plusieurs points dans une seule requête HTTP (enregistrements par lots)
curl -i -XPOST 'http://localhost:8086/write?db=sensors&precision=ns' \
  --data-binary '
sensor_data,facility=plant1,device=temp_sensor_001 temperature=72.4,humidity=45.2 1675000000000000000
sensor_data,facility=plant1,device=temp_sensor_002 temperature=71.8,humidity=44.9 1675000000000000001
sensor_data,facility=plant1,device=pressure_001 pression=14,7,débit=125,3 1675000000000000002'

# Write multiple points in a single HTTP request (batch writes)
curl -i -XPOST 'http://localhost:8086/write?db=sensors&precision=ns' \
  --data-binary '
sensor_data,facility=plant1,device=temp_sensor_001 temperature=72.4,humidity=45.2 1675000000000000000
sensor_data,facility=plant1,device=temp_sensor_002 temperature=71.8,humidity=44.9 1675000000000000001
sensor_data,facility=plant1,device=pressure_001 pressure=14.7,flow_rate=125.3 1675000000000000002'

Les écritures par lots sont nettement plus performantes que les écritures individuelles lorsque le débit de messages est élevé. La documentation d'InfluxDB sur les performances d'écriture recommande des lots de 5 000 à 10 000 points par requête d'écriture pour un débit maximal.

Kafka se situe en amont d'InfluxDB dans la plupart des pipelines de capteurs en production ; il sert de tampon de messages durable qui absorbe les pics d'ingestion et permet à plusieurs consommateurs de traiter le même flux de données à des fins différentes :

# Créer un sujet Kafka pour les données des capteurs avec un partitionnement adapté
kafka-topics.sh --create \
  --topic sensor-readings \
  --partitions 32 \          # Une partition par thread de traitement
  --replication-factor 1 \   # Déploiement sur un seul serveur
  --bootstrap-server localhost:9092

# Create a Kafka topic for sensor data with appropriate partitioning
kafka-topics.sh --create \
  --topic sensor-readings \
  --partitions 32 \          # One partition per processing thread
  --replication-factor 1 \   # Single-server deployment
  --bootstrap-server localhost:9092

Les 32 partitions permettent à 32 threads consommateurs parallèles de traiter simultanément les données des capteurs. Sur le serveur Extreme équipé d'un processeur EPYC à 16 cœurs (32 threads), cela correspond parfaitement au parallélisme maximal sans sursouscription.

Cas d'utilisation n° 3 : Pipelines d'agrégation en temps réel

Les pipelines d'agrégation transforment les flux d'événements à haut débit en résumés exploitables : nombre de pages vues par minute, total des transactions par heure, sessions d'utilisateurs actifs par région. Le défi consiste à calculer ces agrégations en temps réel tout en traitant des millions d'événements bruts par heure.

Apache Flink et Apache Kafka Streams sont les outils open source de référence pour l'agrégation en flux continu à grande échelle. Pour les déploiements sur un seul serveur avec du matériel dédié, Kafka Streams est plus simple à utiliser (pas besoin de cluster séparé) tout en offrant pratiquement les mêmes capacités d'agrégation.

Un pipeline d'agrégation Kafka Streams en Java :

StreamsBuilder builder = new StreamsBuilder();
 
// Read from input topic
KStream<String, PageViewEvent> pageViews = builder.stream("page-views");
 
// Aggregate into 1-minute tumbling windows
KTable<Windowed<String>, Long> viewCounts = pageViews
    .groupBy((key, value) -> value.getPageId())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("page-view-counts"));
 
// Write aggregated results to output topic
viewCounts.toStream()
    .map((windowedKey, count) -> KeyValue.pair(
        windowedKey.key(),
        new AggregatedCount(windowedKey.window().startTime(), count)
    ))
    .to("page-view-aggregates");

StreamsBuilder builder = new StreamsBuilder();
 
// Read from input topic
KStream<String, PageViewEvent> pageViews = builder.stream("page-views");
 
// Aggregate into 1-minute tumbling windows
KTable<Windowed<String>, Long> viewCounts = pageViews
    .groupBy((key, value) -> value.getPageId())
    .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(1)))
    .count(Materialized.as("page-view-counts"));
 
// Write aggregated results to output topic
viewCounts.toStream()
    .map((windowedKey, count) -> KeyValue.pair(
        windowedKey.key(),
        new AggregatedCount(windowedKey.window().startTime(), count)
    ))
    .to("page-view-aggregates");

Les états de stockage pour les agrégations par fenêtre consomment beaucoup de mémoire. Un pipeline gérant des fenêtres glissantes d'une heure sur 100 000 identifiants de page uniques nécessite environ 1 à 2 Go d'espace de stockage par étape du pipeline. Les 192 Go de mémoire vive DDR5 du serveur Extreme offrent une marge suffisante pour exécuter plusieurs étapes d'agrégation avec une allocation généreuse d'espace de stockage, sans risque de saturation de la mémoire.

Optimisation matérielle pour les charges de travail à haute fréquence sous Linux

Plusieurs options de configuration du noyau Linux et du matériel sont particulièrement adaptées aux charges de travail impliquant un traitement à haute fréquence.

Réglage de la fréquence du processeur : le traitement à haute fréquence nécessite des vitesses d'horloge stables. Désactive le réglage de la fréquence pour éviter que les cœurs ne fonctionnent à une fréquence réduite entre les rafales :

# Définir le régulateur de performance (fonctionner toujours à la fréquence maximale)
for cpu in /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor; do
    echo performance > $cpu
done
 
# Rendre cette configuration permanente via cpupower
cpupower frequency-set -g performance

# Set performance governor (run at maximum frequency always)
for cpu in /sys/devices/system/cpu/cpu*/cpufreq/scaling_governor; do
    echo performance > $cpu
done
 
# Make persistent via cpupower
cpupower frequency-set -g performance

Prise en compte du NUMA : L'AMD EPYC 4545P utilise une architecture à chiplets, dans laquelle la latence d'accès à la mémoire varie en fonction du nœud NUMA auquel la mémoire est attribuée, par rapport au cœur qui y accède. Pour les charges de travail sensibles à la latence, affecte les threads de traitement à des cœurs situés dans le même nœud NUMA que la mémoire à laquelle ils accèdent :

# Vérifier la topologie NUMA
numactl --hardware
 
# Lancer un processus avec une affinité NUMA (liée aux processeurs et à la mémoire du nœud 0)
numactl --cpunodebind=0 --membind=0 ./your_processor

# Check NUMA topology
numactl --hardware
 
# Run a process with NUMA affinity (bind to node 0 CPUs and memory)
numactl --cpunodebind=0 --membind=0 ./your_processor

Pages géantes : les pages mémoire de 4 Ko par défaut du noyau Linux nécessitent de nombreuses entrées TLB pour les grands ensembles de travail. L'activation des pages géantes de 2 Mo réduit les échecs TLB lors des traitements gourmands en mémoire :

# Allouer 512 pages géantes (512 × 2 Mo = 1 Go)
echo 512 > /proc/sys/vm/nr_hugepages
 
# Persistance après les redémarrages
echo "vm.nr_hugepages = 512" >> /etc/sysctl.conf

# Allocate 512 huge pages (512 x 2MB = 1GB)
echo 512 > /proc/sys/vm/nr_hugepages
 
# Persistent across reboots
echo "vm.nr_hugepages = 512" >> /etc/sysctl.conf

Affinité IRQ : pour un traitement réseau à haut débit, attribue la gestion des interruptions réseau à des cœurs de processeur spécifiques afin d'éviter le « cache thrashing » lorsque les interruptions sont gérées sur différents cœurs :

# Attribuer les interruptions de la carte réseau aux cœurs 0 à 3
# Commence par identifier les numéros d'interruption de la carte réseau
cat /proc/interrupts | grep eth0
 
# Définir l'affinité (exemple : l'interruption 23 vers le cœur 0)
echo 1 > /proc/irq/23/smp_affinity

# Pin NIC interrupts to cores 0-3
# First identify NIC interrupt numbers
cat /proc/interrupts | grep eth0
 
# Set affinity (example for interrupt 23 to core 0)
echo 1 > /proc/irq/23/smp_affinity

Stockage pour les données à haute fréquence

Les charges de travail à haute fréquence génèrent souvent des volumes de données considérables. Un flux de données financières traitant 100 000 mises à jour par seconde, avec chaque événement occupant 200 octets, génère 20 Mo par seconde, soit 1,7 To par jour.

Le serveur Extreme InMotion Hostingcomprend deux NVMe de 3,84 To chacun, ce qui offre environ 4 jours de stockage brut à ce rythme avant qu'un archivage ne soit nécessaire. Pour une conservation plus longue, mets en place une stratégie de stockage à plusieurs niveaux :

• Stockage à accès fréquent (NVMe) : dernières 48 à 72 heures de données brutes, entièrement consultables
• Stockage à accès modéré (stockage d'objets) : 30 à 90 jours, compressé, consultable avec un certain temps de latence
• Stockage à froid (archives) : au-delà de 90 jours, compressé, récupération lente

Le format Apache Parquet offre une compression en colonnes qui réduit les données financières et les séries chronologiques de capteurs à 10-20 % de leur taille brute, tout en permettant leur interrogation par des outils analytiques tels qu'Apache Spark, DuckDB ou ClickHouse.

L'infrastructure dédiée InMotion Hostingpour les charges de travail à haute fréquence

La configuration du serveur Extreme, qui combine un processeur AMD EPYC 4545P (16 cœurs, 32 threads), 192 Go de mémoire vive DDR5 ECC,SSD NVMe de 3,84 To et une vitesse de port de base de 3 Gbps (extensible à 10 Gbps), répond aux contraintes spécifiques du traitement de données à haute fréquence : le parallélisme du processeur pour le traitement simultané des messages, la bande passante mémoire pour les grands magasins d'états, NVMe pour les écritures à grande vitesse et la capacité réseau pour l'ingestion de données provenant de sources externes.

Le port de base à 3 Gbps est particulièrement adapté aux déploiements de réseaux de capteurs et aux agrégateurs de flux financiers, où le volume de données entrantes est constant plutôt que sporadique. Les équipes qui ont besoin d'un débit garanti plutôt que d'une marge de manœuvre pour les pics de trafic peuvent augmenter la vitesse du port par paliers de 1 Gbps.

Le fait qu'il s'agisse d'un serveur « bare metal » élimine les fluctuations de planification de l'hyperviseur — une caractéristique qui rend les serveurs dédiés particulièrement adaptés aux charges de travail sensibles à la latence, que les machines virtuelles dans le cloud ne peuvent pas prendre en charge de manière fiable. Pour les applications où la latence de traitement se mesure en microsecondes plutôt qu'en millisecondes, la gamme de serveurs dédiés d'InMotion offre la base matérielle dont les charges de travail à haute fréquence ont besoin.