เพิ่มความเร็ว 300% ด้วย LRU Cache บน Python

เวลาเราเขียนโปรแกรม บางส่วนของโปรแกรม อาจจะมีการคำนวณที่เยอะมาก ๆ อยู่ ทำให้การเรียกเพียง 1 ครั้งก็ทำให้เราใช้เวลาในการทำงานพอสมควรเลย และถ้าเราเรียกหลาย ๆ ครั้ง ก็ยิ่งเสียเวลาเข้าไปใหญ่อีก ทำให้เราจะต้องทำอะไรบางอย่างเพื่อที่จะลดเวลา และ การคำนวณในส่วนนี้ลง ซึ่งอาจจะเป็นการ Optimise Method ด้วยวิธีการบางอย่าง แต่อีกวิธีที่สามารถทำได้อย่างง่ายดายใน Python คือการใช้ Cache ช่วยในการประมวลผลนั่นเอง ซึ่งในวันนี้เราจะพามาทำความรู้จักกับ LRU Cache ใน Python กัน

Caching คืออะไร

Caching มันเป็นเทคนิคนึงในการ Optimise โปรแกรมของเรา อย่างที่เราเล่าไปว่า บางครั้งเวลาเราเขียนโปรแกรม มันจะมีบางส่วนของโปรแกรมที่มันต้องการ การประมวลผล ที่เยอะมาก ๆ ทำให้เราเสียทั้งเวลา และ พลังงานเป็นจำนวนมาก เราจะต้องทำอะไรบางอย่าง ที่มาของการทำ Caching คือ การที่เราเข้าไปสังเกตการทำงานของโปรแกรมเรา

บางครั้ง เราจะเจอ Pattern ที่เรามักจะเรียก Function ด้วย Parameter เดิม ๆ ตัวอย่างง่าย ๆ เลยคือ เมื่อเราคิด Factorial เรารู้ว่า 5! = 5 x 4 x 3 x 2 x 1 และ 6! = 6 x 5 x 4 x 3 x 2 x 1 หรือถ้าเราดูดี ๆ 6! = 6 x 5! นั่นเอง สมมุติว่า เครื่องคอมพิวเตอร์ของเราคูณเลขช้ามาก ๆ การหา Factorial แบบนี้จะกลายเป็นเรื่องที่ปวดหัวขึ้นมาทันทีเลย

ถามว่า จากปัญหานี้ เราจะทำยังไงได้บ้าง ให้เราสามารถที่จะแก้ปัญหานี้ได้เร็วมากขึ้น ในรอบแรก เราทำการหาไปแล้วว่า 5! เท่ากับเท่าไหร่กันแน่ ดังนั้น ในตอนที่เราหา 6! เราก็สามารถเอาผลของ 5! มาคูณกับ 6 ได้เลย เราก็จะไม่จำเป็นต้องคูณเลขทั้งหมดแล้ว ทำให้การทำงานเร็วขึ้นเป็นอย่างมาก

ดังนั้น การที่เราจะทำแบบนี้ได้ เราจะต้องสอนให้เครื่องรู้จักการจำผลไว้ล่วงหน้าด้วย อาจจะจำใส่ลงไปใน Memory อะไรบางอย่าง เพื่อให้ครั้งหน้ามา เราก็สามารถที่จะเอาผลนั้นแหละ มาใช้งานได้เลย โดยที่ไม่ต้องทำ นั่นคือ หลักการของ Caching

การทำ Caching เราทำมันมานานมาก ๆ แล้วนะ ไม่ได้เกิดขึ้นใน Python ที่เราจะเล่าให้อ่านในวันนี้เป็นครั้งแรกเลย มันถูกใช้เยอะมาก ๆ เช่นใน CPU ของเราก็จะมี Cache เป็นหน่วยความจำขนาดเล็ก ๆ อยู่สำหรับการเก็บค่าไว้ เพื่อที่เครื่องจะใช้มันก็ดึงได้จาก Cache ที่มี Throughput สูงมาก ๆ ดีกว่าการที่ไปเรียกจาก Harddisk และ SSD ที่กว่าจะโหลดเข้าไปได้

Caching Strategies

อยากให้ลองนึกถึงตอนที่เราทำ Caching ขึ้นมา ถ้าเราเก็บผลของการรันไปเรื่อย ๆ เช่นเราบอกว่า เรามี Function ในการคำนวณ Factorial ทำให้ค่าที่เราจะใส่เข้าไป มันไปได้ไม่จำกัดเลย แต่ปัญหาคือ Memory เรามีจำกัด ทำให้เราไม่สามารถจำทุกผลลัพธ์ได้ทั้งหมด ทำให้เกิดคำถามว่า แล้วสรุปแล้ว เราควรจะให้มันจำอะไรบ้างละ เพื่อให้มั่นใจว่าเราจะได้ใช้ Cache ทั้งหมดอย่างมีประสิทธิภาพสูงสุด

ก่อนจะไปถึงวิธีการ เรามาคุยกันก่อนว่า เราจะรู้ได้อย่างไรว่า Caching ของเรามีประสิทธิภาพมากแค่ไหนกันละ เราจะใช้ค่าที่เรียกว่า Cache Hit Rate พูดง่าย ๆ ก็คือ อัตราส่วนของจำนวนครั้งที่เครื่องสามารถเรียกข้อมูลจาก Cache เทียบกับ ไม่เจอใน Cache แล้วต้องคิดใหม่ เอาหล่ะ แล้วทีนี้เราจะมีวิธีอะไรบ้างละที่จะเลือกว่าเราจะเก็บอะไรไว้บ้าง จริง ๆ มันมีเยอะมาก ๆ เลยนะ

Least Recently Used (LRU) เราขอเริ่มที่ตัวแรก เป็นตัวที่เราเขียนไว้ในหัวเรื่องก่อนละกัน LRU หลักการง่าย ๆ คือ อะไรที่ไม่ได้ใช้นานสุด เอามันออกไปแค่นั้นเลย ทำให้เวลาเราเก็บจริง ๆ เราก็จะต้องเก็บ Order ว่าอ่อ อันนี้นะ ใช้ล่าสุดแล้วนะ อันนี้เราไม่ได้ใช้มาหลังสุดแล้วนะ อาจจะทำเป็นลักษณะของตัวเลขกำกับ หรือจริง ๆ แล้วง่ายกว่านั้นเราก็อาจจะทำให้ลักษณะของ List ก็ได้ พอเราจะ Update เราก็หั่นมันออกแล้วย้ายขึ้นไปอันดับแรกก็ได้ แล้วพอเราจะเอาออก เราก็แค่เอาตัวสุดท้ายออกเท่านั้นเอง

Least Frequency Used (LFU) ไม่ต่างจาก LRU เท่าไหร่ แต่สิ่งที่ต่างคือ แทนที่เราจะคิดว่า ล่าสุดเราใช้อะไร เราเก็บจำนวนที่มันถูกเรียกใช้งานแทน ตัวไหนโดนเรียกใช้งานน้อยที่สุด ก็จะเป็นจุดอ่อนเอาออกไป

FIFO (First-in-First-Out) วิธีต่อไปมันง่ายกว่านั้นอีก ทำตามชื่อเลย ก็คือมาก่อน ไปก่อนเลยจ้า ไม่ได้สนใจเลยว่า ใครจะใช้บ่อย หรือไม่บ่อยอะไรยังไง แค่ว่า ใครมาก่อนไปก่อนเลย เหมือนกับเราต่อคิวซื้อข้าวอะไรพวกนั้นเลย ถ้าเป็นใน Data Structure ให้เรานึงถึง Queue

LIFO (Last-in-First-Out) วิธีนี้จะตรงกันข้ามกับ FIFO เลย เพราะใน LIFO ผู้ที่ถูกคัดออกก่อนเป็นตัวแรกจะเป็นตัวที่เข้ามาล่าสุด ถ้าเป็น Data Structure คือ Stack เลย ทำให้เราสามารถใช้ Stack นี่แหละในการ Implement ได้เลย

ไม่ว่าจะเป็นวิธีไหน มันก็ตอบยากว่า วิธีไหนเร็วที่สุด มันขึ้นกับลักษณะของการใช้ข้อมูลในงานนั้น ๆ ว่า มันมีธรรมชาติในการทำงานอย่างไร เช่น เราบอกว่า เราอาจจะทำ Caching สำหรับหนังบน Netflix ถ้าเราลองคิดดูว่า วิธีไหนละที่น่าจะดีสำหรับ Netflix ก็น่าจะเป็น LRU ก็ได้ เพราะถ้าเราสังเกตดู เราจะเห็นว่า หนังที่คนจะดูเยอะ ๆ มันก็จะเยอะอยู่ช่วงนึงเท่านั้นแหละ มันไม่ได้อยู่ตลอดกาล อย่างช่วงก่อนหน้าเมื่อไม่กี่อาทิตย์ก่อน Hometown Cha Cha Cha ก็อาจจะเป็นอันดับ 1 คนกดเข้าไปดูบ่อยมาก ๆ แต่เมื่อเวลาผ่านไปมันก็จะมีเรื่องใหม่เข้ามาแทน Cache เราก็ต้องเปลี่ยน ทำให้ LRU น่าจะเป็นวิธีการเลือกข้อมูลสำหรับการ Cache ที่ไม่เลวสำหรับ Application อย่าง Storage ของ Netflix เลย

ลอง Implement Cache ด้วยมือกัน

cache_storage = {}

def factorial (n) :
    if n in cache_storage :
        return cache_storage[n]
    else :
        result = 1
        for i in range(1,n+1):
            result *= i
        
        # Don't forget to store the result to the cache
        cache_storage[n] = result
        
        return result

เพื่อให้เข้าใจได้ง่ายขึ้น เราจะลองมาสร้าง Cache ด้วยมือเราเองก่อนดีกว่า ตัวอย่างด้านบนนี้ เราเริ่มจาก เราทำการสร้าง Data Structure สำหรับการเก็บ Cache โดยเราเลือกใช้ Dictionary ขึ้นมา จากนั้น เราก็ทำการสร้าง Function สำหรับการหา Factorial ขึ้นมา

โดยปกติแล้วเราก็อาจจะใช้ Recursion (ที่เราไม่ใช้ในตัวอย่างนี้ เพราะ Python ไม่ค่อยชอบ Recursion ที่ Stack ลึกเท่าไหร่) หรือ Loop กันตรง ๆ เลยก็ได้เหมือนกัน แต่ในเมื่อเรามี Cache แล้ว ก่อนที่เราจะเสียเวลาหา เราก็เข้าไปหาใน Cache ก่อนว่ามีมั้ย ถ้ามี เราก็ให้มันเอาจากใน Cache นี่แหละโยนกลับไป แต่ถ้าไม่มี เราก็ให้มันหามา แต่ก่อนที่เราจะเอาผลส่งกลับไป เราก็ให้มันเก็บลง Cache ไว้ใช้รอบหน้าไป

def fac_without_cache (n) :
    result = 1
    for i in range(1,n+1):
        result *= i    
    return result

ไหน ๆ ก็ไหน ๆ แล้ว เรามาลองดูใน Ideal Condition กันเลย ด้วยการที่เราให้ Cache Hit Rate เป็น 100% ไปเลย มาดูว่า Cache จะช่วยได้มากแค่ไหน เราจะสร้างอีก Function เป็นการหา Factorial เหมือนกัน แต่ตัด Process ของการยุ่งกับ Cache ออกไปเป็นการหา Factorial ตามปกติ

import time

start_time = time.time()
for _ in range(10000) :
    fac_without_cache(10000)
without_cache_time = time.time() - start_time

start_time = time.time()
for _ in range(10000) :
    factorial(10000)
cache_time = time.time() - start_time

print(f"Finding factorial without cache took {without_cache_time} sec(s)")
print(f"Finding factorial with cache took {cache_time} sec(s)")

จากนั้น เรามาเริ่มทำการทดสอบกันดีกว่า เราสร้างแบบทดสอบแบบ Ideal เลย ด้วยการ หา 10,000! ทั้งหมด 10,000 ครั้งด้วยกัน โดยแบ่งออกเป็น 2 การทดลองคือ แบบที่เราไม่ใช้ Cache และ ใช้ Cache ตามลำดับจากตัวอย่างด้านบน

Finding factorial without cache took 321.13297986984253 sec(s)
Finding factorial with cache took 0.03296780586242676 sec(s)

จากด้านบนคือผลที่ได้จากการรันโปรแกรม ตัวที่ไม่ได้ใช้ Cache กดไป 5 นาทีกว่า ๆ เลย แต่ในขณะที่ตัวที่อ่านจาก Cache ใช้ไปแค่ 0.03 วินาที หรือก็คือไม่ถึงวินาทีเท่านั้นเอง จะทำให้เราเห็นว่า การใช้ Cache มันมีผลจริง ๆ กับการทำงาน แต่นี่ก็คือ Ideal Condition ในการทำงานมากกว่า

from random import randint

problems = [randint(300,10_000) for _ in range(10_000)]

start_time = time.time()
for n in problems :
    fac_without_cache(n)
without_cache_time = time.time() - start_time

start_time = time.time()
for n in problems :
    factorial(n)
cache_time = time.time() - start_time

print(f"Finding factorial without cache took {without_cache_time} sec(s)")
print(f"Finding factorial with cache took {cache_time} sec(s) with {cache_hit/10000 * 100}% hit rate")

เพื่อให้เราจำลองให้สมจริงมากขึ้น เราลองกำหนดโจทย์ที่เลขไม่ซ้ำดีกว่า แทนที่เราจะหา Factorial ของ 10,000 หลาย ๆ รอบ เรามาลองสุ่มตัวเลขสัก 10,000 ตัวตั้งแต่ 300 จนไปถึง 10,000 ออกมากัน แล้วเอาตัวเลขที่สุ่มนี่แหละ เป็นโจทย์ให้ ทั้งสองการทดลอง ทำให้ Hit Rate น่าจะน้อยกว่า 100% เว้นแต่ Random Function มีปัญหา เราจะมาดูกันว่า ถ้ามันไม่ Ideal มันจะเป็นอย่างไร

Finding factorial without cache took 108.65645217895508 sec(s)
Finding factorial with cache took 66.88515496253967 sec(s) with 37.84% hit rate

จากผลการทดลองเราจะเห็นได้ว่า ถ้าเราไม่ได้ใช้ Cache เราจะกดอยู่ที่ 108 วินาที หรือเกือบ ๆ 2 นาทีได้ แต่เมื่อเราใช้ Cache ช่วย มันจะเหลืออยู่ 66 วินาที หรือนาทีนิด ๆ เท่านั้นเอง พร้อมกับ Hit Rate ที่ 37.84% ถือว่าไม่ได้เยอะ กลาง ๆ แต่ด้วย Hit Rate เพียงเท่านี้ แต่ทำให้ใช้เวลาในการคำนวณน้อยลงครึ่งนึงเลยก็ถือว่าได้ผลดีมาก ๆ เลยใช่ม้าา แต่ ๆ วิธีที่เราลองตอนนี้เอาจริง ๆ มันเอาไปใช้จริง ๆ ไม่ได้ เพราะเราไม่ได้กำหนดขนาดของ Cache ถ้าเราเอาไปใช้จริง ๆ รันไว้นาน ๆ ก็คือ Memory ที่ใช้ในการเก็บคือเละแน่นอน ไม่ดีแน่ ๆ ดังนั้นเราจะต้องกำหนดขนาดของ Cache และเลือกวิธีที่จะเลือกว่า เราจะ Cache อะไรไว้บ้าง

from typing import Dict, Union

class CachePool :
    def __init__(self) -> None:
        self.cache_hit = 0
        self.cache_called = 0
        self.cache_storage: Dict[int, int] = dict()
        
    def find_cache (self, n) -> Union[int, None] :
        self.cache_called += 1
        
        if n in self.cache_storage :
            self.cache_hit += 1
            return self.cache_storage[n]
        else :
            return None

    def cache_value (self, n, value) :
        self.cache_storage[n] = value
    
    def get_cache_hit_percentage (self) -> float:
        return self.cache_hit / self.cache_called * 100

เพื่อให้ง่ายกับการกำหนดขนาด และ กำหนด Caching Strategies ได้ง่ายขึ้น เราจะเขียน Cache Pool เป็น Class เลยละกัน ใน Class นี้ Constructor จะรับขนาดของ Cache เข้ามา โดยค่าเริ่มต้นจะตั้งไว้ที่ 100 พร้อมกับ มีอีก 2 Method คือ find_cache, cache_value และ get_cache_hit_percentage สำหรับการดึงข้อมูลจาก Cache, เขียนข้อมูลจาก Cache และดึง Hit Rate ตามลำดับ

class LFUCachePool :
    def __init__(self, cache_size:int = 100) -> None:
        self.cache_hit = 0
        self.cache_called = 0
        self.cache_storage: Dict[int, int] = dict()
        self.cache_freq : Dict[int, int] = dict()
        self.cache_size = cache_size
        
    def find_cache (self, n) -> Union[int, None] :
        self.cache_called += 1
        
        if n in self.cache_storage :
            self.cache_freq[n] += 1
            self.cache_hit += 1
            return self.cache_storage[n]
        else :
            return None

    def cache_value (self, n, value) -> None:
        if len(self.cache_storage) > self.cache_size :
            # The Least Freq needs to be removed
            min_freq_n = min(self.cache_freq, key=lambda k: self.cache_freq[k]) 
            del self.cache_freq[min_freq_n]
            del self.cache_storage[min_freq_n]
            
        self.cache_freq[n] = 1
        self.cache_storage[n] = value
    
    def get_cache_hit_percentage (self) -> float:
        return self.cache_hit / self.cache_called * 100

เพื่อให้สนุกกว่านั้น เรามาลองสร้าง Cache Pool แบบที่จำกัดขนาดกัน โดยวิธีที่เราเลือกใช้คือ LFU หรือก็คือ ใครที่ใช้บ่อยน้อยที่สุด ออกไปค่าาาา นั่นเอง ดังนั้นสิ่งแรกที่เราจะต้องมีคือ Dictionary สำหรับการเก็บจำนวนครั้งที่ Cache โดยเรียกใช้ พร้อมกับเวลาที่เรามีการเรียกใช้ และ มัน Hit เราก็ต้อง Update Frequency ไว้ด้วย

และสุดท้าย อันนี้แหละที่เป็นเงื่อนไข คือตอนที่เราจะเขียน Cache ลงไป ถ้าเกิดว่า Cache ของเราเต็ม เราก็จะมาเลือก เราเริ่มจากการหาก่อนว่า ใน Frequency Dictionary ตัวเลขไหนที่มีการใช้น้อยที่สุด เราก็จะได้ Index มา จากนั้น เราก็จะเอามาลบออกไปจากทั้ง Frequency Dictionay และ Cache Storage เพื่อให้มีที่สำหรับของใหม่ที่กำลังจะใส่เข้ามา หรือถ้า Cache ขนาดของเรายังไม่เกิน Limit Size เราก็เพิ่มลงไปได้เลย

Finding factorial with cache took 66.05996894836426 sec(s) with 37.6% hit rate
Finding factorial with LFU cache took 103.16957783699036 sec(s) with 0.86% hit rate

จะเห็นได้เลยว่า เมื่อเรากำหนดขนาดของ Cache เข้าไป และใช้ LFU ทำให้ Cache Hit Rate ต่ำลงอย่างมีนัยสำคัญมาก ๆ ทำให้เวลาในการทำงานก็เพิ่มขึ้นเช่นกัน จาก นาทีกว่า ๆ เมื่อเราไม่ได้กำหนดขนาดของ Cache เลย กับ เกือบ 2 นาทีเลย เมื่อเราใช้ LFU ทำให้การเลือกใช้ Caching Strategies ที่เหมาะสม เป็นเรื่องสำคัญมาก ๆ นอกจากนั้น Cache Size ก็เป็นอีกตัวแปร แต่เราจะไม่ได้ลองให้ดูนะ ไม่งั้นบทความเราจะยาวเกินไปมาก ๆ (ตอนนี้ก็ยาวชิบหายแล้ว !)

LRU Cache บน Python

ในที่สุดก็ถึงเรื่องหลักของเราสักที คือการใช้ LRU Cache สำเร็จรูปใน Python กัน ตัว Cache Pool เราจะสามารถเรียกใช้งานได้จาก Decorator ถ้าใครที่ยังไม่รู้จัก Decorator เราเคยเขียนไว้แล้วกดที่ด้านบนนี้ได้เลย

from functools import lru_cache
  
@lru_cache(maxsize = 100)
def factorial (n) :
    result = 1
    for i in range(1,n+1):
        result *= i    
    return result

การใช้งานนั้นง่ายแสนง่ายกว่าการ Implement เองหลายขุม ฮ่า ๆ ตัว LRU Cache จะอยู่ใน Module ของ functools ซึ่งเป็น Module ที่ใช้สำหรับการจัดการ High-Order Function ต่าง ๆ ในนั้นมีของเล่นเยอะมากจริง ๆ ลองไปหาดูกันได้

การใช้งานเราก็ใช้ lru_cache เป็น Decorator เข้ากับ Function ที่เราต้องการก็เรียบร้อยเลย โดยที่เราจะต้องกำหนดขนาดของ Cache ด้วย ในที่นี้เรากำหนดไว้ที่ 100 ก็เป็น Standard แล้วละ ขนาดของ Cache ขึ้นกับการใช้งานของเราเลย อาจจะต้อง Load โปรแกรมขึ้นมาแล้วค่อย ๆ Tune ไปเรื่อย ๆ จนกว่าจะหาขนาดของ Cache ที่ไม่ใหญ่เกินไป แต่ก็ยังคง Performance ที่ดีได้อยู่

>>> factorial.cache_info()
CacheInfo(hits=1989, misses=8011, maxsize=100, currsize=100)

ถามว่าเราจะรู้ได้ยังไงละว่า Cache Pool ที่เราเอาเข้ามาใช้มันทำงานได้ดีแค่ไหน เราก็สามารถขอดูได้จากคำสั่ง cache_info ได้เลย มันก็จะบอกข้อมูลมาหมดเลย ถ้าเราอยากรู้ Hit Rate เราก็เอาไปคำนวณต่อได้เลย ไม่ยากใช่มั้ย เอา hits หารด้วย hits + misses แล้วคูณด้วย 100% เราก็จะได้เป็น Percentage ออกมาพร้อมใช้งานเลย

สรุป

การใช้ Cache เป็นเพียงหนึ่งในวิธีการที่เราจะ Optimise Script หรือ Program ของเราให้สามารถทำงานได้เร็วขึ้นได้ ผ่านการจำผลลัพธ์ไว้ล่วงหน้าแล้ว ซึ่งในการใช้งานจริง เราก็จะมีวิธีการเลือกข้อมูลในการเก็บที่ไม่เหมือนกัน ซึ่งมันก็จะเหมาะกับการใช้งานที่แตกต่างกันอีก นอกจากนั้น ขนาดของ Cache ก็มีผลเช่นกัน ดังนั้นเพื่อให้ได้ประสิทธิภาพสูงสุด เราก็ต้องมาเลือก ทั้งวิธีการ และ ขนาดของ Cache เพื่อให้มั่นใจว่า เราจะได้ประสิทธิภาพในการทำงานที่สูงที่สุด โดยที่ไม่ได้ใช้ Resource มากเกินไปนั่นเอง ซึ่งหนึ่งใน Caching Technique ที่ Python เตรียมมาให้เราคือ LRU ที่เราสามารถเรียกได้เลยจาก functools โดยที่เราแค่จับมันไปใส่เป็น Decorator ก็เสร็จเลย ถือว่าง่ายมาก ๆ แนะนำให้ลองเอาไปใช้ดูได้

ปล. จากตัวอย่าง Factorial จริง ๆ มัน Optimise ได้มากกว่านี้อีกนี้ เราสามารถเก็บแยกเคสเป็นทีละขั้นตอนได้เลย เช่น 6! = 6 x 5! แล้วเราก็เอา 5! ไปเช็คได้อีกวิธีนี้ความเร็วจะเพิ่มขึ้นมหาศาลมาก ๆ เลย ปกติถ้าเราเขียนแบบ Recursive และใช้ Cache เราก็จะใช้วิธีแบบนี้แหละ แต่อันนี้เพื่อความเข้าใจง่ายเลยเขียนแบบที่มันอาจจะไม่ได้เขียนในวิธีที่ดีที่สุด