Threading, Multiprocessing & SQLite

**_Froggy_** · 31/08/2023, 22h28

Hello tout le monde,

Actuellement j'ai le programme suivant :
1/ Un module déclenche une interruption
2/ L'interruption est détectée via GPIO.wait_for_edge
3/ Le raspberry pi zero récupère les données
4/ Le raspberry pi zero écrit les données dans une bdd SQLite sur sa carte SD

Le problème, c'est que cela prend grosso modo 60ms dont 50ms pour la partie 4 et potentiellement, le module peut balancer une interruption toutes les 20ms et actuellement, j'ai une probabilité de perdre des données.
Dit autrement, sur 1s, je peux avoir entre 50 données ou 0.

En gros mon idée, c'est de créer d'autres process qui ne feraient que de l'écriture en parallèle pour que la priorité reste à la récupération des données.

Mais pour le coup, entre threading & multiprocessing, j'ai un peu du mal à voir lequel utilisé. Sachant que comme c'est une BDD SQLite, si les threads/process pouvaient avoir une fonction lock, cela serait cool.

Question annexe (pour du multicore), est-ce qu'il est possible de dire à python de s'executer sur le core 2 par exemple ?

Merci

**tyrtamos** · 01/09/2023, 05h55

Bonjour

A ma connaissance, sqlite3 n'étant pas un serveur, ne supporte pas les accès simultanés.

Threading et multiprocessing comportent des verrous. Pour multiprocessing, qui a été intelligemment construit de façon similaire à threading, les verrous sont ici: https://docs.python.org/fr/3.10/libr...rocessing.Lock. Il ne reste plus qu'à apprendre à s'en servir...

A ma connaissance aussi, python ne peut pas choisir un cœur du CPU: c'est l'OS qui le fait. A noter que si on parle d'une exécution multi-core du CPU, il faut choisir multiprocessing, et non threading qui ne s'exécute que sur un seul cœur.

**fred1599** · 01/09/2023, 10h20

Bonjour,

Avec SQLite3 on peut faire de l'asynchronisme : aiosqlite

Ensuite, il n'y a pas 36 solutions, déjà faut savoir qu'il n'y a pas plus rapide que la RAM. Conclusion : Minimiser le temps d'accès à la base de données !

Stocker les données en mémoire dans une file d'attente.
Écrire toutes les données en une seule transaction dans la base de données (on appelle cela des transactions batch) avec une boucle for par exemple et une fois le bloc terminé, on commit l'insertion.

Autre solution: Passer par un serveur Redis, beaucoup plus efficace que SQLite3 en écriture, puis une fois toutes les données stockées, les transférer sur la base de données SQLite.

Peut-être qu'une solution moins simple que les précédentes comme un système de messagerie (kafka, RabbitMQ, ...) pourrait faire l'affaire, mais je pense que la complexité dépasse peut-être le besoin actuel.

**wiztricks** · 01/09/2023, 10h23

Envoyé par _Froggy_

En gros mon idée, c'est de créer d'autres process qui ne feraient que de l'écriture en parallèle pour que la priorité reste à la récupération des données.

C'est une bonne idée. Mais côté réalisation, çà veut dire déconnecter le temps mis pour récupérer les données avec celui pris pour les écrire via un "buffer" qu'on peut remplir aussi vite que nécessaire et regrouper les écritures pour qu'elles prennent un minimum de temps (histoire que le buffer ait le temps de se vider).
note: on ne parallélise pas les écritures, on les regroupe pour que çà prenne le moins de temps possible: s'il faut une seconde pour recevoir 5 messages toutes les 20ms, on espère pouvoir qu'écrire ces 5 messages d'un seul coup prendra (beaucoup moins).

Reste à savoir quelle devra être la nature du "buffer": ce peut être de multiples fichiers séquentiels, des sockets (on expédie ce qu'il y a à écrire à un serveur), de la mémoire partagée (entre threads ou entre process).

Mais c'est la conception de votre application qu'il faut étudier: python n'a pas grand chose à faire là dedans sinon dans les détails de réalisation.

- W

**_Froggy_** · 01/09/2023, 11h41

Envoyé par fred1599

[*]Stocker les données en mémoire dans une file d'attente.

Dans ma tête, la création d'un autre thread/process faisait justement fonction de stockage temporaire en mémoire sans que j'ai à faire de gestion d'un buffer d'écriture. J'ai par ailleurs pas de nécessité d'écriture FIFO. Il est évident qu'une vraie gestion de buffer serait plus optimise. Je vois bien comment en faire un dans ma tête : un second process à qui je donne les données en écriture, mais je vois pas comment le concevoir en python autrement qu'avec un process "serveur".
Pour l'instant, le Pi Zero est un monocore au départ l'idée était d'utilisé un Pi Zero 2 4 core mais les prix sont encore totalement délirants.

Question annexe, si j'utilise threading, comment se réparti l'utilisation processeur ? C'est à dire que si j'ai :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
def ecriture_donnees():
 bdd.insert()
 bdd.commit()
 
try:
 while(true)
  gestion_interruption()
  threading (ecriture_donnees())

Le bdd.commit() du thread me prend 50ms (temps d'écriture sur la SD je pense). Est-ce que pendant ces 50ms d'attente d'I/O, l'interpréteur python continue la boucle try: ?

Après relecture de la doc SQLite, il y a une gestion interne à SQLite de l'écriture en parallèle, donc pas besoin de gestion de lock par les threads/process.

Je pense qu'à terme, je ferais la gestion de l'IRQ en C, pour optimiser le bouzin en mode *boum* quand il y a une interruption, plus qu'un wait_for_edge

Autre solution: Passer par un serveur Redis, beaucoup plus efficace que SQLite3 en écriture, puis une fois toutes les données stockées, les transférer sur la base de données SQLite.

J'avais réfléchis à cela en version plus garbage, le fichier stocker sur la SD, copier dans une ramFS au démarrage et un crontab qui sauvegarde toutes les x minutes. Je connaissais pas Redis, je vais voir ça

Envoyé par wiztricks

Mais c'est la conception de votre application qu'il faut étudier: python n'a pas grand chose à faire là dedans sinon dans les détails de réalisation.

- W

Dans ma tête c'est plus ou moins claire, la question c'est comment le réaliser. Comme répondu au dessus, le mieux serait un second process qui ne gère que l'écriture. Donc s'il y a 3 trames de données pendant une écriture, à l'écriture suivante, il écrit les 3 trames d'un coup. Mais je ne vois pas comment l'écrire.

------

Merci pour vos réponses

**wiztricks** · 01/09/2023, 12h08

Envoyé par _Froggy_

Le bdd.commit() du thread me prend 50ms (temps d'écriture sur la SD je pense). Est-ce que pendant ces 50ms d'attente d'I/O, l'interpréteur python continue la boucle try: ?

Ce sera une mixture d'entrées sorties (ou le thread perd le CPU pour faire avancer l'autre) et de CPU pour les effectuer.
Maintenant, vous n'allez pas démarrer un thread à chaque fois mais mettre un tampon/buffer entre les deux.

Envoyé par _Froggy_

Dans ma tête c'est plus ou moins claire, la question c'est comment le réaliser. Comme répondu au dessus, le mieux serait un second process qui ne gère que l'écriture. Donc s'il y a 3 trames de données pendant une écriture, à l'écriture suivante, il écrit les 3 trames d'un coup. Mais je ne vois pas comment l'écrire.

Comme vous n'avez qu'un CPU, il va falloir faire avec threads ou asyncio. Et si vous faites avec des threads, il y a Queue pour créer ce tampon. Ceci dit comme vous ne contrôlez pas grand chose, ça ne marchera peut être pas (et Python n'est pas forcément adapté pour être réactif).

- W

**umfred** · 01/09/2023, 13h07

Est-ce que tu as testé de mettre ta base sur une clé USB plutôt que sur la SD ? Il me semble que c'est plus rapide.

**fred1599** · 01/09/2023, 13h18

Le code ci-dessous est déjà un problème de performance, d'où mon intervention en faisant de l'insertion "batch"

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
def ecriture_donnees(): 
    bdd.insert()
    bdd.commit()

ce que je proposerai serait plutôt du style :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
def ecriture_donnees(donnees):
    for donnee in donnees:
        bdd.insert(...)
    bdd.commit()

donnees est l'ensemble des données en mémoire (par exemple une liste ou un ensemble (set) python), ça implique une seule écriture en base, plutôt que de multiples comme tu le proposes.

Perso, avant de voir du côté du C, je pense qu'il faut faire des essais et voir ce que ça donne, mais des essais optimisés en python, car même en C, si tes idées sont présentées de la même manière que ce que tu fais en python, tu feras du C non optimisé.

**wiztricks** · 01/09/2023, 14h37

Envoyé par fred1599

donnees est l'ensemble des données en mémoire (par exemple une liste ou un ensemble (set) python), ça implique une seule écriture en base, plutôt que de multiples comme tu le proposes.

L’intérêt d'un "bulk" insert est de remplacer la boucle:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
   for donnee in donnees:
        bdd.insert(...)

par:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

   bdd.bulk_insert(données)

note: avec sqlite, bulk_insert = executemany.

On réduit le temps CPU en remplaçant N appels au SGDB par un seul et on permet au SGDB d'optimiser les écritures disques (un SGDB est un peu plus compliqué qu'un fichier séquentiel).

- W

**fred1599** · 01/09/2023, 14h57

Envoyé par wiztricks

L’intérêt d'un "bulk" insert est de...

Exactement c'est le principe que je voulais montrer, je connais pas assez SQLite, je suis plutôt PostgreSQL, mais executemany est bien la solution adéquate pour SQLite.

Envoyé par wiztricks

On réduit le temps CPU en remplaçant N appels au SGDB par un seul et on permet au SGDB d'optimiser les écritures disques (un SGDB est un peu plus compliqué qu'un fichier séquentiel).

Tout à fait !

**wiztricks** · 01/09/2023, 15h30

Envoyé par fred1599

Exactement c'est le principe que je voulais montrer, je connais pas assez SQLite, je suis plutôt PostgreSQL, mais executemany est bien la solution adéquate pour SQLite.

Avec postgresql, ça dépend de l'interface offerte par le pilote. pyscopg2 étant le plus populaire connaît executemany et execute_values (beaucoup plus rapide).
Ceci dit, reste le soucis SQLIte vs. accès concurrents car si on y écrit des choses, il va bien falloir les relire de temps en temps.

- W

**jurassic pork** · 01/09/2023, 15h43

Hello,
A lire ceci aussi : Optimizing SQLite on Raspberry Pi

dont la conclusion est :

If your goal is to improve performance using SQLite, definitely try to group as many operations as you can into a single transaction.

ce qui rejoint ce que propose fred1599.

Ami calmant, J.P

**fred1599** · 01/09/2023, 17h51

Envoyé par wiztricks

Avec postgresql, ça dépend de l'interface offerte par le pilote. pyscopg2 étant le plus populaire connaît executemany et execute_values (beaucoup plus rapide).

Je passe par un intermédiaire, plus précisément l'ORM Django

**wiztricks** · 01/09/2023, 18h21

Envoyé par fred1599

Je passe par un intermédiaire, plus précisément l'ORM Django

Dans ce cas, Django utilise le SGDB de façon très particulière et on ne va pas toujours pouvoir bénéficier des optimisations possibles.
Mais comme on est dans des domaines où on ne compte pas les millisecondes, on s'intéressera pas trop à l'optimisation des ressources de base (CPU, disque, mémoire): on en ajoute et on répartit la charge.

- W

**binarygirl** · 01/09/2023, 19h19

Je pense que ce serait bien d'avoir un exemple de code représentatif. Car les goulets d'étranglement ne se situent pas toujours où on le pense.
Pour mesurer le temps d'exécution j'utilise personnellement le package codetiming en Python. Des explications ici: https://realpython.com/python-timer/...hon-timer-code

On peut optimiser les performances de SQLite, par exemple en utilisant PRAGMA journal_mode = MEMORY'.
Cela devrait augmenter la rapidité mais il y a des inconvénients: en cas de coupure de courant on peut se retrouver potentiellement avec une DB corrompue. Ça dépend donc de la valeur des données.

The MEMORY journaling mode stores the rollback journal in volatile RAM. This saves disk I/O but at the expense of database safety and integrity. If the application using SQLite crashes in the middle of a transaction when the MEMORY journaling mode is set, then the database file will very likely go corrupt.

Il y a d'autres options similaires qui tournent autour de la journalisation, à étudier au cas par cas.

Le support physique induit forcément de la latence, mais on peut créer une DB en mémoire aussi. C'est justifiable quand on n'a pas besoin de stocker les données à long terme.

**papajoker** · 01/09/2023, 20h01

Envoyé par wiztricks

Dans ce cas, Django utilise le SGDB de façon très particulière et on ne va pas toujours pouvoir bénéficier des optimisations possibles.

Avec django, j'utilise toujours pour les restaurations de bd Bulk, qui indique dans la doc : pour sqlite une limite de 999 variables ...

----------

Si je reprends ton pseudo code :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
datas = []
try:
    while(true)
        nombre = len(datas)
        if nombre >= 200:
             # un seul commit pour 200 entrées
             ok, size = sauvegarder_in_db_dans_autre_tache(datas, nombre) 
             if ok :
                 # supprimer les premières entrées traitées
                 datas = datas[size:] 
             # sinon ?
        data = gestion_interruption()
        datas.append(data)
finally:
    sauvegarder_in_db(datas, len(datas))

**wiztricks** · 02/09/2023, 16h10

Envoyé par binarygirl

Pour mesurer le temps d'exécution j'utilise personnellement le package codetiming en Python. Des explications ici: https://realpython.com/python-timer/...hon-timer-code

Ici, on s'en fout un peu du temps d'exécution: on doit faire une saisie toutes les 20ms et les écrire dans une table sqlite3 (ou autre) sans perdre de données. Si on part à explorer les bulk write, la question est de savoir combien d'enregistrements écrire d'un coup.

Et pour savoir si ça marche, comme on a mis une Queue entre la lecture et l'écriture, il suffit de s'assurer qu'on ne remplit pas la Queue qui sert de tampon.

Ce qui donne un "prototype" qui pourrait être:

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
from threading import Thread, Event
from queue import Queue, Empty, Full
from time import sleep
 
import sqlite3
import os
 
def writer(queue, chunk_size, db_path='./test.sqlite3'):
 
    if os.path.exists(db_path):
        os.remove(db_path)
    cnx = sqlite3.connect(db_path)
    cnx.execute('create table test(a, b, c)')
 
    buffer = []
    while True:
        try:
            item = queue.get(timeout=0.5)
        except Empty:
            if done.is_set():
                break
        else:
            buffer.append(item)
            if len(buffer) == chunk_size:
                with cnx:
                    cnx.executemany(
                     'insert into test values(?, ?, ?)', 
                     buffer)
                buffer.clear()
 
def reader(queue, count=100, delay=0.02):
    for x in range(count):
        queue.put_nowait((x, f'yyy-{x}', f'yyy-{x}') )
        sleep(delay)
 
if __name__ == '__main__':
 
    CHUNK_SIZE = 10
    COUNT = 1000
    DELAY = 0.002
 
    done = Event()
    queue = Queue(CHUNK_SIZE)
 
    th = Thread(target=writer, args=(queue, CHUNK_SIZE))
    th.start()
    try:
        reader(queue, count=COUNT, delay=DELAY)
    except Full:
        print('failed')
    except Exception as e:
        print('unexpected exception', e)
    else:
        print('Done.')
    finally:
        done.set()
        th.join()

où j'écris toutes les 0.002ms (10*plus rapide que 20ms pour avoir de la marge).
Sur ma machine, çà plante avec des CHUNK_SIZE < 5 et ça fonctionne au delà.
note: reste à tester avec des enregistrements plus proches de ceux du PO que ce que j'ai choisi.... puis de mettre de vraies lectures et faire tourner ça sur un Pi Zero.

- W

**fred1599** · 03/09/2023, 11h11

Hello,

@wiztricks,

Ok pour la solution, mais me vient une question sur le choix de concevoir ce type de solution.

Pourquoi utiliser les threads et non les tâches asynchrones par exemple ? Il me semble avoir déjà lu que les threads sont assez consommateurs mémoire, sachant qu'on est sur un support mémoire limité (Raspberry Pi Zero).

**wiztricks** · 03/09/2023, 14h00

Envoyé par fred1599

Pourquoi utiliser les threads et non les tâches asynchrones par exemple ? Il me semble avoir déjà lu que les threads sont assez consommateurs mémoire, sachant qu'on est sur un support mémoire limité ([/SIZE]Raspberry Pi Zero).

Pour que çà ait un réel avantage, il faudrait avoir à multiplexer un tas d'entrées sorties. Ici, on a juste la lecture du port GPIO (qui chez moi est simulé par un time.sleep et l'ajout d'un tuple dans le buffer) et l'écriture "bulk" dans la base de donnée. Par contre, çà a des inconvénients: utiliser une bibliothèque externe pour sqlite3, je ne sais pas ce qu'on peut faire côté GPIO, un style de programmation différent (on passe de séquentiel à évènementiel plus ardu),...

Si on veut faire mieux côté PiZero, je n'écrirais pas en BDD locale mais me contenterai de pousser les données sur le réseau et de les récupérer sur un serveur. On pourra aussi les regrouper si nécessaire, mais côté process, on attend le temps d'une copie mémoire: pas évident qu'il soit nécessaire d'avoir un asynchronisme explicite (dans le code): une boucle avec des conditions...

- W

**fred1599** · 03/09/2023, 15h01

Super, merci !

J'aime bien ton principe KISS... le fait de donner une solution simple avant de rentrer dans des considérations plus complexes.

Externaliser le stockage me semble une excellente idée aussi !

Dans la complexité des tâches asynchrones qui sont faut avouer, plus sur la partie apprentissage, on pourrait aussi dire que les threads peuvent être complexes sur leur gestion (problème de concurrences).

Aussi le GIL est une partie importante à ne pas prendre à la légère, on a pas un vrai parallélisme, car un seul thread à la fois sera utilisé sur le processus, ce qui veut dire qu'on pourrait complexifier en ajoutant des modules tels que multiprocessing par exemple.
Malgré le fait que le GIL n'affecte pas autant les tâches qui sont I/O bound, c'est un point que l'on ne devrait pas négliger.

Threading, Multiprocessing & SQLite [Python 3.X]

Python

Discussions similaires

Partager

Partager