SQLite 簡易サーバー ２

先日の記事「SQLite 簡易サーバー １」で動かし始めた「簡易サーバー」について、私の使い方の範囲で試験を行いました。

負荷試験と呼ぶには甘々ですが、これ位使えるなら積極的に使ってみようと思います。

「簡易サーバー」の概要

SQLite と socket 通信を使って、一つのデータベースを複数のクライアントからアクセス出来る簡易サーバーです。Webで検索しても具体的なお手本を見つけられなかったので、試してみました。よく調べると、同様の方法で使っている方は結構いらっしゃるようです。socket でやり取りする文字列を特定のアプリケーション向けに工夫する場合が多いようです。

私が見つけられなかったのは、汎用性がないからなのか、「誰でも考え付くでしょう！」と言ったレベルなのでわざわざ書く必要がないからなのかも知れません。

「簡易サーバー」はsocket（TCP)でSQL文を送信すると返信して来ます。
SELECT に対してはリスト形式を、INSERT, UPDATE に対しては整数形式 0 を返します。

例
SELECT * FROM temp where time = "11:40:10"
[['200505', '11:40:10', 28.6, 45.2]]

UPDATE temp SET T0 = 28.6 where time = "11:40:10"
0

例外：
SQLite でエラーが発生した時には整数形式 1 を返します。
空文または先頭文字が d または D の時は無視します。

リスト形式、整数形式は後述のスクリプト１のように、ast.literal_eval または eval でリストや整数値に変換可能な文字列と言う意味で使いました。

試験内容と結果

以下、３台のRasPiを使いました。
１．「簡易サーバー」Raspberry Pi 3B+
２．クライアント１　Raspberry Pi 4B
　　役割：SELECTを繰り返して時間測定する
３．クライアント２　Raspberry Pi Zero WH
　　役割：INSERTを繰り返して負荷になる

クライアント１は全力で同じSELECT文を3000回送ります。（約30秒）
クライアント２は0.5秒間隔でINSERT文を120回送ります。（約60秒）

結果は測定の都度時間が変動して判然としませんでした。0.5秒間隔のINSERTは大して負荷になっていないようです。

クライアント１の3000回所要時間
回目　クライアント２停止中　クライアント２動作中
 1 　　　　　29.25 　　　　　　　　29.33
 2 　　　　　26.46 　　　　　　　　29.95
 3 　　　　　27.95 　　　　　　　　30.19
 4 　　　　　27.50 　　　　　　　　29.52
 5 　　　　　27.82 　　　　　　　　32.92
 6 　　　　　29.27 　　　　　　　　31.82
 7 　　　　　31.60 　　　　　　　　30.52
 8 　　　　　30.85 　　　　　　　　30.32
 9 　　　　　27.87 　　　　　　　　31.61
10　　　　　28.29 　　　　　　　　31.49

１０回の試験後にINSERTされたレコード数は1200でした。

想定している環境
非同期で複数（20台程度）のクライアントから、平均して数秒に１回程度のINSERT, UPDATE, SELECT が実施される。
これと並行しデータ集計等の目的で、1時間に1回程度、連続してSELECTを実施するクライアントが居ると言った環境を想定しています。

稀にこれ以上の負荷がかかるとしても余り気にする必要はなさそうです。
複数行のSELECTを実施しないようにすれば、私の用途には十分使えると思います。

実験に使ったスクリプト

「簡易サーバー」

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

# socket サーバを作成
import socket
import sqlite3
dbname = '/media/pi/SQLite/DBs/T_sens.db'

# AF = IPv4 という意味
# TCP/IP の場合は、SOCK_STREAM を使う
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:

   # IPアドレスとポートを指定
   #s .bind(('127.0.0.1', 50007))
   s.bind(('192.168.11.202', 50007))
   # 1 接続
   s.listen(5)
   # connection するまで待つ
   while True:
       # 誰かがアクセスしてきたら、コネクションとアドレスを入れる
       sock, addr = s.accept()
       with sock:
           while True:
               # データを受け取る
               data = sock.recv(1024)
               if not data:
                   break
               d = data.decode('utf8')
               if d[:1] == 'd' or d[:1] == 'D':
                   break
               #print (d)
               try:
                   conn = sqlite3.connect(dbname)
                   cur = conn.cursor()
                   
                   # 受信したSQL文をSQLiteに渡す
                   cur.execute(d)
                   
                   if d[:6] == 'SELECT' or d[:6] == 'select':
                       # レスポンスを取得するfetchall()
                       res = cur.fetchall()
                   else:
                       #res  = [(0,)]
                       res = '0'
                       
                   conn.commit()
                   cur.close()
                   conn.close()
               except:
                   #res  = [('Error',)]
                   res = '1'
                   print(d)
                   
               # クライアントにデータを返す(b -> byte でないといけない)
               # list の要素がタプルなので list に変換
               if type(res) == str:
                   out = res
               else:
                   out = ['',] * len(res)
                   for i in range(len(res)):
                       out[i] = list(res[i])
               #print (out)
               
               sock.sendall(bytes(str(out).encode('utf8')))

クライアント１

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import socket
import ast
from time import time

addr = '192.168.11.202'
port = 50007
# クライアントを作成
   
stim = time()
for i in range(3000):
   with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
       # サーバを指定
       s.connect((addr, port))
       cmd = 'SELECT * FROM temp where time = "11:40:10"'
       
       cmd = cmd.encode('utf8')
       #print (cmd)
       s.sendall(cmd)
       # ネットワークのバッファサイズは1024。サーバからの文字列を取得する
       data = s.recv(1024)
       # str(リスト) を リストに戻す str(整数) も　整数に戻す
       rdat = ast.literal_eval(data.decode('utf8'))
       #print (type(rdat), rdat)
   
   
etim = time()
print(type(rdat), rdat)
print('start time = ' + str(stim))
print('end time = ' + str(etim))
print('differencs = ' + str(etim-stim))

クライアント２

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

import datetime
import signal
from time import sleep, time
import socket
import ast

addr = '192.168.11.202'
port = 50007

def scheduler(arg1, args2):
   
   global sigcnt
   
   sigcnt += 1
   now = datetime.datetime.now()
   dt = now.strftime('%y%m%d')
   tm = now.strftime('%H:%M:%S')
   val = '("{}", "{}", 20.0, 50.0)'.format(dt,tm)
   cmd = 'INSERT INTO temp(date,time, T0, hum) values' + val
   
   with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
       s.connect((addr, port))
       cmd = cmd.encode('utf8')
       #print (cmd)
       s.sendall(cmd)
       # ネットワークのバッファサイズは1024。サーバからの文字列を取得する
       data = s.recv(1024)
       # str(リスト) を リストに戻す str(整数) も　整数に戻す
       rdat = ast.literal_eval(data.decode('utf8'))
       print(type(rdat), rdat)
       
   
   
def main():
   global sigcnt
   sigcnt = 0
   sigcm = 0

   signal.signal(signal.SIGALRM, scheduler)
   signal.setitimer(signal.ITIMER_REAL, 0.5, 0.5)

   stim = time()
   
   while(sigcnt < 120):
       if sigcm != sigcnt:
           print(sigcnt)
           sigcm = sigcnt
       sleep(0.01)

   etim = time()
   signal.setitimer(signal.ITIMER_REAL, 0)
   print('start time = ' + str(stim))
   print('end time = ' + str(etim))
   print('differencs = ' + str(etim-stim))

if __name__ == "__main__":
   main()

今後の予定

特に問題なく使えそうなので、積極的に使って行きます。

必要に応じてUSBメモリーをHDD等に変更します。
CRON等で定期的（毎日夜間）にデータベースのバックアップを取りながら運用するつもりです。（コピー中のアクセスはどうなるんでしょう？）

何か変化があったら続報します。
今後とも宜しくお願い致します。