SQLite에서 캐시 효율성

Source: Dev.to

개요

• 목표: NoSQL 캐시(Valkey)와 SQLite 내부 캐시 간의 읽기 속도를 비교합니다.
• 애플리케이션: Python + Flask 로 개발된 REST JSON 웹 서비스.
• 엔드포인트:
  1. Encurtar URL – 긴 URL을 받아 짧은 버전을 반환합니다.
  2. Resolver URL curta – 짧은 버전에서 원래 URL을 반환합니다.

왜 캐시가 중요한가?

생성된 캐시는 미리 단축된 URL을 읽는 데 사용됩니다. 구조화된 방식으로 URL을 저장하는 데이터베이스가 매 요청마다 파일을 검색할 필요가 없을 때, 웹 서버는 훨씬 더 효율적으로 응답할 수 있습니다.

사용된 기술

서버 라우트 (엔드포인트)

from flask import Flask, request, jsonify
from business.link_shortener import generate_new_link, get_link_by_hash_string

app = Flask(__name__)

# ----------------------------------------------------------------------
#  Health‑check
# ----------------------------------------------------------------------
@app.route("/heart-beat", methods=["GET"])
def heart_beat() -> dict:
    """애플리케이션이 실행 중인지 확인하는 간단한 엔드포인트."""
    return {"OK": True}


# ----------------------------------------------------------------------
#  Criação de link encurtado
# ----------------------------------------------------------------------
@app.route("/link", methods=["POST"])
def create_link() -> dict:
    """
    다음 형식의 JSON을 받습니다:
        { "url": "https://exemplo.com/alguma/pagina" }

    반환:
        { "short_link": "<hash>" }
    """
    data = request.get_json(silent=True) or {}
    url = data.get("url")

    if not url:
        return jsonify({"error": "Campo 'url' é obrigatório."}), 400

    short_link = generate_new_link(url)
    return {"short_link": short_link}, 201


# ----------------------------------------------------------------------
#  Resgate de link original a partir do hash
# ----------------------------------------------------------------------
@app.route("/link/<string:short_link>", methods=["GET"])
def get_original_link(short_link: str) -> dict:
    """
    제공된 해시와 연관된 원본 URL을 검색합니다.

    호출 예시:
        GET /link/abc123

    반환:
        { "link": "https://exemplo.com/alguma/pagina" }
    """
    original_url = get_link_by_hash_string(short_link)

    if not original_url:
        return jsonify({"error": "Link não encontrado."}), 404

    return {"link": original_url}, 200

관찰 사항

| | **실행** | 파일을app.py로 저장하고flask run을 실행하세요 (if name == “main”: app.run()가 있으면python app.py` 를 실행하세요). |

import sqlite3
import random
import string
import threading
import valkey
import json

# Thread‑local storage for database connections
_thread_local = threading.local()

# Flag that activates the key‑value database cache
kv_cache = True

# Global connection pool for Valkey
kv_pool = valkey.ConnectionPool(
    host='localhost',
    port=6379,
    db=0,
    max_connections=50,          # Match the number of concurrent threads
    socket_keepalive=True,
    socket_connect_timeout=5,
    retry_on_timeout=True,
    decode_responses=True       # Automatically decode responses to strings
)

# ----------------------------------------------------------------------
# Initialize database schema
# ----------------------------------------------------------------------
conn = sqlite3.connect('database.db')
cur = conn.cursor()
cur.execute('''
    CREATE TABLE IF NOT EXISTS links (
        id INTEGER PRIMARY KEY,
        link_hash TEXT,
        link_target TEXT
    )
''')
cur.execute('CREATE INDEX IF NOT EXISTS idx_links_id ON links (id)')
cur.execute('CREATE INDEX IF NOT EXISTS idx_links_link_hash ON links (link_hash)')
conn.commit()
conn.close()

def get_db_connection():
    """Return a thread‑local SQLite connection."""
    if not hasattr(_thread_local, 'conn'):
        _thread_local.conn = sqlite3.connect(
            'database.db',
            check_same_thread=False
        )
        # Disable SQLite's internal page cache and force temp files on disk
        _thread_local.conn.execute('PRAGMA cache_size = 0')
        _thread_local.conn.execute('PRAGMA temp_store = 0')
    return _thread_local.conn

def get_kv_db():
    """Return a thread‑local Valkey client."""
    if not hasattr(_thread_local, 'kv_db'):
        _thread_local.kv_db = valkey.Valkey(connection_pool=kv_pool)
    return _thread_local.kv_db

def generate_hash():
    """Generate a random 10‑character hash."""
    return ''.join(
        random.choices(string.ascii_letters + string.digits, k=10)
    )

def create(link):
    """Insert a new link into SQLite and return its hash."""
    conn = get_db_connection()
    cur = conn.cursor()
    link_hash = generate_hash()
    cur.execute(
        'INSERT INTO links (link_hash, link_target) VALUES (?, ?)',
        (link_hash, link)
    )
    conn.commit()
    return link_hash

def get_link_by_hash(hash):
    """Retrieve a link by its hash, using Valkey as a cache when enabled."""
    if kv_cache:
        cached = get_kv_db().get(hash)
        if cached:
            return cached                     # Already decoded (decode_responses=True)

    conn = get_db_connection()
    cur = conn.cursor()
    cur.execute(
        'SELECT link_target FROM links WHERE link_hash = ?',
        (hash,)
    )
    result = cur.fetchone()
    if result and result[0]:
        if kv_cache:
            get_kv_db().set(hash, result[0])
        return result[0]
    return None

주의:

• kv_cache = True (48번째 줄)에서 Valkey 캐시를 활성화할지 여부를 설정합니다.
• SQLite 내부 캐시를 비활성화하는 줄(PRAGMA cache_size = 0 및 PRAGMA temp_store = 0)은 SQLite가 자체 페이지 캐시를 사용하지 않게 하여 외부 캐시의 영향을 관찰할 수 있게 합니다.

스트레스 테스트 스크립트

아래 스크립트는 여러 스레드를 사용해 대량의 요청을 생성합니다. 짧은 링크를 생성한 뒤 해당 링크를 여러 번 읽어 지연 시간을 측정합니다.

import time
import requests
from concurrent.futures import ThreadPoolExecutor, as_completed
from statistics import mean, median

# -------------------------------------------------
# Configuration
# -------------------------------------------------
BASE_URL = "http://localhost:5000"
NUM_TASKS = 100          # Total number of concurrent tasks
NUM_THREADS = 30         # Number of worker threads
NUM_LINK_GETS = 80       # GET requests per task


def create_test_link() -> str | None:
    """Create a short link via the API."""
    try:
        resp = requests.post(f"{BASE_URL}/link", json={"url": "https://example.com"})
        if resp.status_code == 200:
            return resp.json().get("short_link")
    except Exception as exc:
        print(f"Error creating test link: {exc}")
    return None


def test_link_retrieval() -> list[dict]:
    """Perform `NUM_LINK_GETS` GET requests for a single short link."""
    start = time.time()
    short_link = create_test_link()
    results = []

    for _ in range(NUM_LINK_GETS):
        try:
            resp = requests.get(f"{BASE_URL}/link/{short_link}")
            elapsed = time.time() - start
            results.append(
                {
                    "success": resp.status_code == 200,
                    "time": elapsed,
                    "status_code": resp.status_code,
                }
            )
        except Exception as exc:
            results.append(
                {
                    "success": False,
                    "time": time.time() - start,
                    "error": str(exc),
                }
            )
    return results


def run_stress_test() -> None:
    print(f"Total tasks: {NUM_TASKS}")
    print(f"Concurrent threads: {NUM_THREADS}")

    with ThreadPoolExecutor(max_workers=NUM_THREADS) as executor:
        futures = [executor.submit(test_link_retrieval) for _ in range(NUM_TASKS)]

        all_results = []
        for future in as_completed(futures):
            all_results.extend(future.result())

    # -------------------------------------------------
    # Statistics
    # -------------------------------------------------
    times = [r["time"] for r in all_results if r["success"]]
    if times:
        print(f"Requests: {len(times)}")
        print(f"Mean latency: {mean(times):.4f}s")
        print(f"Median latency: {median(times):.4f}s")
        print(f"Min latency: {min(times):.4f}s")
        print(f"Max latency: {max(times):.4f}s")
    else:
        print("No successful requests recorded.")


if __name__ == "__main__":
    run_stress_test()

사용 방법

1. Flask 서버를 시작합니다 (예: python app.py).
2. 위 스크립트를 실행합니다.
3. kv_cache를 활성화 및 비활성화한 출력 결과를 비교하여 Valkey 캐시의 성능 영향을 확인합니다.

스트레스 테스트 스크립트 (Python)

import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from statistics import mean, median
from typing import List, Dict, Any

NUM_THREADS = 8   # Number of concurrent threads
NUM_TASKS = 1000   # Number of tasks to run


def test_link_retrieval() -> List[Dict[str, Any]]:
    """
    Simulate a request to the API endpoint that retrieves a link.

    Replace this stub with the actual request logic (e.g., using ``requests``).

    Returns
    -------
    list of dict
        Each dict should contain at least:
        - ``success`` (bool): whether the request succeeded
        - ``time`` (float): response time in seconds
        - optionally ``error`` (str) when ``success`` is ``False``
    """
    # TODO: implement real request
    # Example return value:
    # return [{'success': True, 'time': 0.123}, ...]
    return []


def run_stress_test() -> None:
    """Execute the stress test and print a summary of the results."""
    print(f"Current threads: {NUM_THREADS}\n")

    results: List[Dict[str, Any]] = []
    start_time = time.time()

    # ------------------------------------------------------------------
    # Execute the tasks concurrently
    # ------------------------------------------------------------------
    with ThreadPoolExecutor(max_workers=NUM_THREADS) as executor:
        futures = [executor.submit(test_link_retrieval) for _ in range(NUM_TASKS)]

        for i, future in enumerate(as_completed(futures), start=1):
            subresults = future.result()
            results.extend(subresults)

            if i % 100 == 0:
                print(f"Completed: {i}/{NUM_TASKS}")

    total_time = time.time() - start_time

    # ------------------------------------------------------------------
    # Analysis
    # ------------------------------------------------------------------
    successful = [r for r in results if r.get('success')]
    failed = [r for r in results if not r.get('success')]
    response_times = [r['time'] for r in successful]

    print("\n" + "=" * 60)
    print("STRESS TEST RESULTS")
    print("=" * 60)
    print(f"Total tasks: {NUM_TASKS}")
    print(f"Successful: {len(successful)}")
    print(f"Failed: {len(failed)}")
    print(f"Total time: {total_time:.2f}s")
    print(f"Tasks/second: {NUM_TASKS / total_time:.2f}")

    if response_times:
        print("\nResponse times:")
        print(f"  Min:    {min(response_times) * 1000:.2f} ms")
        print(f"  Max:    {max(response_times) * 1000:.2f} ms")
        print(f"  Mean:   {mean(response_times) * 1000:.2f} ms")
        print(f"  Median: {median(response_times) * 1000:.2f} ms")

    if failed:
        print("\nErrors (first 5):")
        for r in failed[:5]:
            print(f"  - {r.get('error', 'Unknown error')}")

    print("=" * 60)


if __name__ == "__main__":
    run_stress_test()

테스트 결과 (로컬 환경)

Note: 테스트는 SQLite의 인‑메모리 캐시를 비활성화하지 않은 단일 로컬 서버에서 수행되었습니다. 그럼에도 불구하고 SQLite는 Valkey(Redis) 캐시를 사용하든 사용하지 않든 거의 동일한 속도를 보였습니다.

SQLite 3 – Without Valkey (Redis) Cache

============================================================
STRESS TEST RESULTS
============================================================
Total tasks: 100
Successful: 8000
Failed: 0
Total time: 4.57s
Tasks/second: 21.89

Response times:
  Min: 10.92ms
  Max: 1534.86ms
  Mean: 664.45ms
  Median: 627.76ms
============================================================

SQLite 3 – With Valkey (Redis) Cache

============================================================
STRESS TEST RESULTS
============================================================
Total tasks: 100
Successful: 8000
Failed: 0
Total time: 5.24s
Tasks/second: 19.08

Response times:
  Min: 25.95ms
  Max: 1896.03ms
  Mean: 783.95ms
  Median: 747.06ms
============================================================

Valkey(Redis)에 도달하기 위해 필요한 네트워크 오버헤드를 고려하더라도, SQLite의 성능은 여전히 인상적입니다.

Observations

• SQLite는 일반적으로 대규모 프로덕션 워크로드에 사용되지 않으므로 이러한 로컬 결과는 전용 캐시 레이어가 유용한 실제 시나리오를 완전히 대변하지 않습니다.
• SQLite의 내부 캐시를 비활성화해도 운영 체제의 파일 시스템 캐시는 여전히 활성화되어 있어 Valkey와 같은 외부 캐시의 실제 영향을 가릴 수 있습니다.
• 가장 흥미로운 점은 Valkey를 캐시로 사용하는 개념 증명보다는 OS 수준 캐싱과 SQLite 자체 최적화가 성능에 어떻게 영향을 미치는가 입니다.

Source Code

이 개념 증명의 전체 소스 코드는 다음에서 확인할 수 있습니다:

memcache‑poc on GitHub