DEV Community: Lucas Leal da Costa

Recursion vs Loop: a low-level analysis

Lucas Leal da Costa — Fri, 12 Jul 2024 00:16:38 +0000

Introduction

Sometimes when writing code we have to decide between using loop or recursion. How do both work under the hood? In terms of performance which one is the best to pick and why?

PS.: CPU architecture -> x86_64 | RAM -> DDR4 16G

1. Study Case 📝

Let's implement a function that receives a number and adding it one by one, arrives at the value received.

1.1. Recursion

int sum(int n) {
    if (n == 0) {
        return 0;
    }

    return 1 + sum(n - 1);
}

1.2. Loop

int sum(int n) {
    int result = 0;

    for (int i = 1; i <= n; i++) {
        result += 1;
    }

    return result;
}

1.3. Time each function

#include <stdio.h>
#include <time.h>

// sum functions here

int main() {
    int n = 70000;

    clock_t start = clock();

    sum(n);

    clock_t end = clock();

    double cpu_time_used = ((double) (end - start)) / CLOCKS_PER_SEC;

    printf("sum(%d) Time elapsed = %.7f\n", n, cpu_time_used);

    return 0;
}

2. Compiling the code

Let's compile but not assemble it. We can do it by using the gcc compiler with the flag -S. Since we are interested in the recursive and loop functions, we will only examine these assembly instructions:

2.1. Recursion

sum:
.LFB0:
    .cfi_startproc
    endbr64
    pushq   %rbp
    .cfi_def_cfa_offset 16
    .cfi_offset 6, -16
    movq    %rsp, %rbp
    .cfi_def_cfa_register 6
    subq    $16, %rsp
    movl    %edi, -4(%rbp)
    cmpl    $0, -4(%rbp)
    jne .L2
    movl    $0, %eax
    jmp .L3
.L2:
    movl    -4(%rbp), %eax
    subl    $1, %eax
    movl    %eax, %edi
    call    sum
    addl    $1, %eax
.L3:
    leave
    .cfi_def_cfa 7, 8
    ret
    .cfi_endproc
.LFE0:
    .size   sum, .-sum
    .section    .rodata

2.2. Loop

sum:
.LFB0:
    .cfi_startproc
    endbr64
    pushq   %rbp
    .cfi_def_cfa_offset 16
    .cfi_offset 6, -16
    movq    %rsp, %rbp
    .cfi_def_cfa_register 6
    movl    %edi, -20(%rbp)
    movl    $0, -8(%rbp)
    movl    $1, -4(%rbp)
    jmp .L2
.L3:
    addl    $1, -8(%rbp)
    addl    $1, -4(%rbp)
.L2:
    movl    -4(%rbp), %eax
    cmpl    -20(%rbp), %eax
    jle .L3
    movl    -8(%rbp), %eax
    popq    %rbp
    .cfi_def_cfa 7, 8
    ret
    .cfi_endproc
.LFE0:
    .size   sum, .-sum
    .section    .rodata

3. Executing both programs

Execution of sum(70000) using both approaches 5 times each. Elapsed time in seconds:

Loop	Recursion
0.0001380	0.0012100
0.0001230	0.0011630
0.0001370	0.0011390
0.0001710	0.0012150
0.0001380	0.0011810

Why is the loop approach 10 times faster?

4. Cause of the performance penalty 📉

At the assembly code, we have to pay attention to some instructions in order to understand the overhead of the recursion approach. These are:

pushq %rbp
movq %rsp, %rbp
leave or popq %rbp

In short, pushq %rbp saves the base pointer of the previous function. movq %rsp, %rbp initializes the base pointer to the current function, subq $16, %rsp alocates 16 bytes on the top of the stackframe - by subtracting 16 bytes from the initial %rsp, since it is initialy equal to the %rbp: the stack grows from high memory addresses to the lower ones. leave sets %rsp to %rbp, then pops the top of the stack, thus restoring the function that called it.

This process is repeated in each function call. It significantly increases the interaction with memory - L caches cannot help that much in this case - the performance penalty comes precisely from that.

5. Comparing the instructions in both cases

5.1. Loop

movl    %edi, -20(%rbp)
movl    $0, -8(%rbp)
movl    $1, -4(%rbp)

These three instructions load into memory the initial values of n, result and i. %edi is the register used to pass the argument to the function, so the value of n is saved into the address -20(%rbp). i and result are initially set to 1 and 0, that is why $1 and $0 are loaded into memory addresses -4(%rbp) and -8(%rbp), respectively.

The loop starts at jmp .L2. movl -4(%rbp), %eax loads data into %eax register so it can be used to compare in cmpl -20(%rbp), %eax. jle .L3 jumps to .L3 if %eax is less or equal to -20(%rbp), namely i <= n;.

In L3., addl $1, -8(%rbp) will increase -8(%rbp) by one - result -, then do the same to i: addl $1, -4(%rbp), i.e., i++

This process is executed until the comparison is false, then popq %rbp and ret are executed, respectively poping the top of the stack and returning sum.

5.2. Recursion

Recursion instructions:

movl %edi, -4(%rbp) takes the argument - sent via %edi register - and saves it 4 bytes bellow the value of the pointer saved in %rbp. cmpl $0, -4(%rbp) takes the value and compares it to 0, jne .L2 sets that, if the value in -4(%rbp) is not equal to 0, the code jumps to .L2 block.

At .L2, movl -4(%rbp), %eax will load the value of -4(%rbp) to %eax, then subl $1, %eax subtract it by 1 and save it in the register itself. In movl %eax, %edi, the value of register %eax is loaded into another register: %edi. As we saw, this register is responsible for passing arguments to functions. The argument is passed to call sum, so allocating more addresses onto the top of the stackframe and repeting the whole process of recursion. addl $1, %eax will increase by 1 and save the value in the register %eax. When recursion reach the base case, movl $0, %eax - so placing 0 as return of sum(n - 1) - will be executed, then jmp.L3 will jump to .L3 and execute leave and ret.

6. Segmentation Fault using recursion

Once it is clear how recursion works under the hood, it is much easier to see how a Segmentation Fault can take place. Each process allocates a finite space in memory, since recursions can grown the stackframe indefinitely, therefore Segmentation Fault can occur.

E.g., let's set int n = 1000000; and see how it performs:

linux@x86_64:~$ ./recursive_sum
Segmentation fault (core dumped)

Once using loop approach there is no indefinite function calls, there is no risk of the Stack Overflow that happens under recursion.

Solving Race Conditions in Python: A Mutex Approach for Efficiency

Lucas Leal da Costa — Tue, 31 Oct 2023 22:50:22 +0000

Introduction

When working with concurrent programs, it is crucial to prevent race conditions. In this article, we will demonstrate the implementation of a scalable and efficient mutex to facilitate the management of multiple threads.

As will be presented, it was able to reduce the time up to 90.02% when implementing the mutex instead of simple lock solution.

Source code can be found at: GitHub

PS.: Python version used: 3.10
PS.: Notebook configs used in these tests: Intel i7-1165G7 2.80 GHz | 16 Gb RAM

1. Study Case 📝

By exploring a race condition problem, we will develop a mutex to address the problem efficiently.

1.1. Race Condition Example

We have multiple objects to count, a function to increase the counter and multiple threads.

1.1.1. Counters



from dataclasses import dataclass


@dataclass
class Counter:
    value: int
    name: str


first_counter = Counter(value=0, name="first_counter")
second_counter = Counter(value=0, name="second_counter")
third_counter = Counter(value=0, name="third_counter")
fourth_counter = Counter(value=0, name="fourth_counter")
fifth_counter = Counter(value=0, name="fifth_counter")
sixth_counter = Counter(value=0, name="sixth_counter")

1.1.2. Increase Counter Function



from time import sleep

def increase(by: int, counter: Counter):

    local_counter = counter.value
    local_counter += by

    sleep(0.1) # processing time simulation

    counter.value = local_counter

1.1.3. Execution



import threading
from time import time

if __name__ == "__main__":

    threads = [
        threading.Thread(target=increase, args=(1,first_counter)),
        threading.Thread(target=increase, args=(1,fourth_counter)),
        threading.Thread(target=increase, args=(1,sixth_counter)),
        threading.Thread(target=increase, args=(1,third_counter)),
        threading.Thread(target=increase, args=(1,first_counter)),
        threading.Thread(target=increase, args=(1,fifth_counter)),
        threading.Thread(target=increase, args=(1,first_counter)),
        threading.Thread(target=increase, args=(1,second_counter)),
        threading.Thread(target=increase, args=(1,fourth_counter)),
        threading.Thread(target=increase, args=(1,first_counter)),
        threading.Thread(target=increase, args=(1,second_counter)),
        threading.Thread(target=increase, args=(1,first_counter)),
        threading.Thread(target=increase, args=(1,fourth_counter)),
        threading.Thread(target=increase, args=(1,first_counter)),
        threading.Thread(target=increase, args=(1,sixth_counter)),
        threading.Thread(target=increase, args=(1,third_counter)),
        threading.Thread(target=increase, args=(1,third_counter)),
        threading.Thread(target=increase, args=(1,fifth_counter)),
        threading.Thread(target=increase, args=(1,third_counter)),
        threading.Thread(target=increase, args=(1,sixth_counter)),
        threading.Thread(target=increase, args=(1,third_counter)),
    ]

    initial_time = time()

    for t in threads:
        t.start()

    for t in threads:
        t.join()

    final_time = time()

    print(f"\nFinal value {first_counter.name}:", first_counter.value)
    print(f"Final value {second_counter.name}:", second_counter.value)
    print(f"Final value {third_counter.name}:", third_counter.value)
    print(f"Final value {fourth_counter.name}:", fourth_counter.value)
    print(f"Final value {fifth_counter.name}:", fifth_counter.value)
    print(f"Final value {sixth_counter.name}:", sixth_counter.value)

    print(f"\nExecution time: {round(final_time - initial_time, 3)} seconds")

1.1.4. Execution Results



Final value first_counter: 1
Final value second_counter: 1
Final value third_counter: 1
Final value fourth_counter: 1
Final value fifth_counter: 1
Final value sixth_counter: 1

Execution time: 0.102 seconds

It wasn't what we expected. Although multiple threads were modifying the counters, they all have the same value: 1. This is an example of race conditions. But how can we solve this?

1.2. Solving the Problem

What if we could lock the counter so each thread increase the counter at a time? So let's insert a lock in this function so that any thread has to wait for its turn to increase the counter.

We can do that by implementing a Lock in the increase function:



import threading
from time import sleep

lock = threading.Lock()

def increase(by: int, counter: Counter):

    with lock:

        local_counter = counter.value
        local_counter += by

        sleep(0.1) # processing time simulation

        counter.value = local_counter

Whenever a thread enters the context manager it locks for the rest. The lock is only released when thread exits the context manager.

1.2.1. Results After Simple Lock



Final value first_counter: 6
Final value second_counter: 2
Final value third_counter: 5
Final value fourth_counter: 3
Final value fifth_counter: 2
Final value sixth_counter: 3

Execution time: 2.108 seconds

As we can see, it has taken a lot longer to finish execution after inserting the lock. How can we improve it? That is what our mutex is going to do.

2. Our Goal

The mutex has to check some requirements:

Be efficient
Be easy to maintain
Be scalable - see Open–closed principle

For the sake of efficiency, our mutex will have multiple locks each associated with a specific key; Once the thread provides the key, the mutex will create a lock associated with the key or enqueues the predefined key if the lock already exists. Since our mutex has to be available to the entire system, the mutex needs to be a thread safe Singleton.

3. Mutex 🚦

Now let's implement what we have defined.

3.1. Thread Safe Singleton



import threading


class SingletonThreadSafeMeta(type):
    """Thread safe Singleton"""

    __instances = {}
    __lock = threading.Lock()

    def __call__(cls, *args, **kwargs):
        with cls.__lock:
            if cls not in cls.__instances:
                instance = super().__call__(*args, **kwargs)
                cls.__instances[cls] = instance
        return cls.__instances[cls]

3.2. Key Based Mutex



import threading

from singleton_thread_safe import SingletonThreadSafeMeta


class CustomMutex(metaclass=SingletonThreadSafeMeta):

    def __init__(self):
        self.__lockers: dict[str, threading.Lock] = {}
        self.__lockers_counter: dict[str, int] = {}

    def __decrease_counter(self, key: str) -> None:
        self.__lockers_counter[key] -= 1

        if self.__lockers_counter == 0:
            del self.__lockers[key]
            del self.__lockers_counter[key]

    def __get_thread_locker(self, key: str) -> threading.Lock:
        if key in self.__lockers:
            self.__lockers_counter[key] += 1
            return self.__lockers[key]

        self.__lockers_counter[key] = 1
        thread_locker = threading.Lock()
        self.__lockers[key] = thread_locker

        return thread_locker

    def lock_with_key(self, key: str) -> None:
        """Locks the thread associated with the key"""

        thread_locker = self.__get_thread_locker(key)
        thread_locker.acquire()

    def unlock_with_key(self, key: str) -> None:
        """Unlocks the thread associated with the key"""

        if key in self.__lockers:

            thread_locker = self.__lockers[key]
            self.__decrease_counter(key)
            thread_locker.release()

The Mutex class will be responsible for managing the relation between the locks and the keys; this relation is defined in the self.__lockers attribute.

One might be asking where is the queue so we can handle the threads waiting for the lock to be released. The queue is implemented under the hood. As a thread reaches thread_lock.acquire() the threading module will implement a FIFO queue. Therefore, there is no need for us to develop a queue.

4. Workers 🦺

Since we have the mutex it is important to create the workers so they can lock threads based on keys. Each thread will have an instance of a worker, every worker will report to the mutex in order to lock and unlock threads.

Each worker is responsible to manage timeouts in case a thread takes too long to unlock the queue in the mutex.

4.1. Abstract Worker

Since we can have multiple workers it is important to define an interface so they can all work in the same way.



from abc import ABC, abstractmethod

from mutex import CustomMutex


class MutexWorkerAbstract(ABC):
    @abstractmethod
    def __enter__(self) -> "MutexWorkerAbstract":
        """implement context manager"""

    @abstractmethod
    def __exit__(self, exc_type, exc_value, exc_tb) -> None:
        """implement context manager"""

    @property
    @abstractmethod
    def custom_mutex(self) -> CustomMutex:
        """Enforcing composition with CustomMutex"""

As we dive into the worker's structure the context manager implementation will get clearer.

4.2. Worker Counter



import threading

from mutex import CustomMutex
from worker_abstract import MutexWorkerAbstract


class CounterMutexWorker(MutexWorkerAbstract):

    def __init__(self):
        self.__general_mutex = CustomMutex()
        self.__counter_key_lock: int | None = None
        self.__timeout = 5 # seconds
        self.__timeout_timer = threading.Timer(
            self.__timeout, 
            function=self.__timeout_callback
        )

    @property
    def custom_mutex(self) -> CustomMutex:
        return self.__general_mutex

    def __enter__(self) -> "CounterMutexWorker":
        return self

    def __exit__(self, exc_type, exc_value, exc_tb) -> None:
        self.unlock_by_counter()

    def __format_key_lock_thread(self, counter_name: int) -> str:
        key = f"COUNTER_WORKER_KEY_{counter_name}"
        return key

    def __timeout_callback(self) -> None:
        print(f"Lock timeout! worker=CounterMutexWorker | number={self.__counter_key_lock}")
        print(f"Releasing Lock... worker=CounterMutexWorker number={self.__counter_key_lock}")

        self.unlock_by_counter()

    def lock_by_counter(self, counter_name: str) -> None:
        """Locks thread by counter name"""

        key = self.__format_key_lock_thread(counter_name)
        self.custom_mutex.lock_with_key(key=key)
        self.__timeout_timer.start()
        self.__counter_key_lock = counter_name

    def unlock_by_counter(self) -> None:
        """Unlocks thread by counter name"""

        if self.__timeout_timer.is_alive():
            self.__timeout_timer.cancel()

        if self.__counter_key_lock:
            key = self.__format_key_lock_thread(self.__counter_key_lock)
            self.custom_mutex.unlock_with_key(key=key)
            self.__counter_key_lock = None

PS.: Please note that two different workers cannot share the same key formatter

Now we can understand the importance of the context manager: if there is an exception before unlocking the thread, the queue is freed previous to the timeout expires.

Each worker has their own custom key. As soon as a thread locks the mutex, the worker initiates another thread to monitor the timeout, if the timeout exceeds the queue is released.

5. Testing

It's time to implement in the previous problem what we have built. Then let's check the results.

5.1. Implementing Worker

Applying the worker to the increase function.



from worker import CounterMutexWorker

def increase(by: int, counter: Counter):

    worker = CounterMutexWorker()

    with worker:
        worker.lock_by_counter(counter.name)

        local_counter = counter.value
        local_counter += by

        sleep(0.1) # simulating process time

        counter.value = local_counter

        worker.unlock_by_counter()

5.1.1. Results After Mutex



Final value first_counter: 6
Final value second_counter: 2
Final value third_counter: 5
Final value fourth_counter: 3
Final value fifth_counter: 2
Final value sixth_counter: 3

Execution time: 0.608 seconds

5.2. Testing timeout

As timeout was implemented, let's check how it performs. We can test it by decreasing self.__timeout from 5 to 0.02 seconds, since our sleep time is 0.1 seconds.



class CounterMutexWorker(MutexWorkerAbstract):

    def __init__(self):
        self.__general_mutex = CustomMutex()
        self.__counter_key_lock: int | None = None
        self.__timeout = 0.02 # seconds
        self.__timeout_timer = threading.Timer(
            self.__timeout, 
            function=self.__timeout_callback
        )
.
.
.

5.2.1. Timeout Results



Lock timeout! worker=CounterMutexWorker | key=first_counter
Releasing Lock... worker=CounterMutexWorker key=first_counter
Lock timeout! worker=CounterMutexWorker | key=fourth_counter
Releasing Lock... worker=CounterMutexWorker key=fourth_counter
Lock timeout! worker=CounterMutexWorker | key=third_counter
Releasing Lock... worker=CounterMutexWorker key=third_counter
Lock timeout! worker=CounterMutexWorker | key=sixth_counter
Lock timeout! worker=CounterMutexWorker | key=fifth_counter
Lock timeout! worker=CounterMutexWorker | key=second_counter
Releasing Lock... worker=CounterMutexWorker key=sixth_counter
Releasing Lock... worker=CounterMutexWorker key=fifth_counter
Releasing Lock... worker=CounterMutexWorker key=second_counter
Lock timeout! worker=CounterMutexWorker | key=first_counter
Releasing Lock... worker=CounterMutexWorker key=first_counter
Lock timeout! worker=CounterMutexWorker | key=third_counter
Releasing Lock... worker=CounterMutexWorker key=third_counter
Lock timeout! worker=CounterMutexWorker | key=fourth_counter
Releasing Lock... worker=CounterMutexWorker key=fourth_counter
Lock timeout! worker=CounterMutexWorker | key=sixth_counter
Releasing Lock... worker=CounterMutexWorker key=sixth_counter
Lock timeout! worker=CounterMutexWorker | key=fifth_counter
Releasing Lock... worker=CounterMutexWorker key=fifth_counter
Lock timeout! worker=CounterMutexWorker | key=second_counter
Releasing Lock... worker=CounterMutexWorker key=second_counter
Lock timeout! worker=CounterMutexWorker | key=first_counter
Releasing Lock... worker=CounterMutexWorker key=first_counter
Lock timeout! worker=CounterMutexWorker | key=third_counter
Releasing Lock... worker=CounterMutexWorker key=third_counter
Lock timeout! worker=CounterMutexWorker | key=fourth_counter
Releasing Lock... worker=CounterMutexWorker key=fourth_counter
Lock timeout! worker=CounterMutexWorker | key=sixth_counter
Releasing Lock... worker=CounterMutexWorker key=sixth_counter
Lock timeout! worker=CounterMutexWorker | key=first_counter
Releasing Lock... worker=CounterMutexWorker key=first_counter
Lock timeout! worker=CounterMutexWorker | key=third_counter
Releasing Lock... worker=CounterMutexWorker key=third_counter
Lock timeout! worker=CounterMutexWorker | key=first_counter
Releasing Lock... worker=CounterMutexWorker key=first_counter
Lock timeout! worker=CounterMutexWorker | key=third_counter
Releasing Lock... worker=CounterMutexWorker key=third_counter
Lock timeout! worker=CounterMutexWorker | key=first_counter
Releasing Lock... worker=CounterMutexWorker key=first_counter

Final value first_counter: 2
Final value second_counter: 1
Final value third_counter: 1
Final value fourth_counter: 1
Final value fifth_counter: 1
Final value sixth_counter: 1

Execution time: 0.203 seconds

It took only 0.2 seconds because as timeouts were exceeding, the locker's queue was being released.

5.3. Load Test

What if we increase the number of threads and counters? For the purpose of testing the performance of the mutex, let's set 1000 threads and 12 counters.

5.3.1. Execution



import threading
from random import randint
from time import time

if __name__ == "__main__":
    counters = [
        Counter(value=0, name="first_counter"),
        Counter(value=0, name="second_counter"),
        Counter(value=0, name="third_counter"),
        Counter(value=0, name="fourth_counter"),
        Counter(value=0, name="fifth_counter"),
        Counter(value=0, name="sixth_counter"),
        Counter(value=0, name="seventh_counter"),
        Counter(value=0, name="eighth_counter"),
        Counter(value=0, name="ninth_counter"),
        Counter(value=0, name="tenth_counter"),
        Counter(value=0, name="eleventh_counter"),
        Counter(value=0, name="twelfth_counter"),
    ]

    amount_of_threads = 1000

    threads = [
        threading.Thread(
            target=increase, 
            args=(
                1, 
                counters[randint(0, len(counters) -1)]
            )
        )
        for _ in range(amount_of_threads)
    ]

    initial_time = time()

    for t in threads:
        t.start()

    for t in threads:
        t.join()

    final_time = time()

    print("\n=======================================\n")

    for c in counters:
        print(f"Final value {c.name}:", c.value)    

    print("\n=======================================\n")

    print(f"Execution time: {round(final_time - initial_time, 3)} seconds")

5.3.2. Results Using Simple Lock



=======================================

Final value first_counter: 74
Final value second_counter: 85
Final value third_counter: 85
Final value fourth_counter: 90
Final value fifth_counter: 92
Final value sixth_counter: 87
Final value seventh_counter: 85
Final value eighth_counter: 78
Final value ninth_counter: 85
Final value tenth_counter: 85
Final value eleventh_counter: 82
Final value twelfth_counter: 72

=======================================

Execution time: 100.403 seconds

5.3.3. Results Using Mutex



=======================================

Final value first_counter: 74
Final value second_counter: 85
Final value third_counter: 85
Final value fourth_counter: 90
Final value fifth_counter: 92
Final value sixth_counter: 87
Final value seventh_counter: 85
Final value eighth_counter: 78
Final value ninth_counter: 85
Final value tenth_counter: 85
Final value eleventh_counter: 82
Final value twelfth_counter: 72

=======================================

Execution time: 9.815 seconds

6. Conclusion

After implementing the mutex we managed to reduced the time considerably.

21 threads and 6 counters -> from 2.108 seconds to 0.608 seconds. Time reduction of 71.1%.
1000 threads and 12 counters -> from 100.403 seconds to 9.815 seconds. Time reduction of 90.02%.

As more threads and more counters were introduced these values got even better. So we can state that this is an excellent implementation for large systems.

Furthermore, the combination mutex + worker is scalable: If there is a demand for another lock in the same fashion, simply implement a new worker conforming to the class MutexWorkerAbstract.

Maintainable Code: Building Components of Large Scale Projects

Lucas Leal da Costa — Tue, 08 Aug 2023 02:36:12 +0000

Introduction

Every developer at some point has faced problems when it comes to implementing new features into legacy code. Your component needs to have another functionality implemented. What if the project that you are working on is getting bigger each day, how would you assure that creating new functionalities wouldn't be painful? By exploring a simple example we will understand what maintainable code really is.

PS.: I choose Python due to its simplicity, what will be shown can be easily applied in any other modern language.

Case Study 📝

So you are a developer building the back-end of a two numbers calculator. The most straightforward approach would be something like

def calculate(first_number, second_number, operation):

    result = 0

    if operation == "addition":
        result = first_number + second_number

    elif operation == "subtraction":
        result = first_number - second_number

    elif operation == "multiplication":
        result = first_number * second_number

    elif operation == "division":
        result = first_number / second_number

    return result

What if you need to implement another operation, like raising the power of a number? Would you create another elif statement operation == "power"? What if you had to implement more complicated operations?

There are definitely better ways of writing the back-end of this calculator, that is what we are going to dive into.

Refactoring the Code 💻

Whenever we think about large scale projects, it is always important to build components that can easily scale modifying as little as possible the existing code. Your new feature should be an extension of what is already there, not a restructure of it. That is what the 'O' of the S.O.L.I.D. principles stands for:

software entities (classes, modules, functions, etc.) should be open for extension, but closed for modification

Our goal is not to explain SOLID Principles or Design Patterns, but providing a solution for the problem we've pointed. For the moment let's keep these principles under the hood...

1. Creating Enum

It is important to define in our system which operations our calculator will handle, we can do it by creating an Enum.

from enum import Enum

class Operations(Enum):
    ADDITION="addition"
    SUBTRACTION="subtraction"
    MULTIPLICATION="multiplication"
    DIVISION="division"

As soon as we have our Enum let's rewrite our function using the type we created

def calculate(first_number: float, second_number: float, operation: Operations) -> float:

    result = 0

    if operation == Operations.ADDITION:
        result = first_number + second_number

    elif operation == Operations.SUBTRACTION:
        result = first_number - second_number

    elif operation == Operations.MULTIPLICATION:
        result = first_number * second_number

    elif operation == Operations.DIVISION:
        result = first_number / second_number

    return result

We are no longer manipulating strings to define which operation has to be performed. From now on every operation is going to be strictly defined in the Enum created.

2. Segregating Responsibilities

As we discussed earlier what if we want to implement more complicated operations? It does not seem like a good idea to let this function get bigger and bigger as we need to grow our project. So let's decouple the main calculator function from its functionalities

2.1. Operations

def addition(first_number: float, second_number: float) -> float:
    return first_number + second_number

def subtraction(first_number: float, second_number: float) -> float:
    return first_number - second_number

def multiplication(first_number: float, second_number: float) -> float:
    return first_number * second_number

def division(first_number: float, second_number: float) -> float:
    return first_number / second_number

2.2. Calculator

def calculate(first_number: float, second_number: float, operation: Operations) -> float:

    if operation == Operations.ADDITION:
        return addition(first_number, second_number)

    if operation == Operations.SUBTRACTION:
        return subtraction(first_number, second_number)

    if operation == Operations.MULTIPLICATION:
        return multiplication(first_number, second_number)

    if operation == Operations.DIVISION:
        return division(first_number, second_number)

    raise ValueError(f"{Operations} value not supported => {operation}")

As we can see, now it is much easier to create and apply new operations to our calculator. We also took the opportunity to create error handling. If an invalid value for operation is inputted, an error will be raised so we can more easily track unexpected arguments so avoiding unwanted behaviors.

However, our initial proposal was building it as if it were made for large scale projects, so let's take a step further and improve this function...

3. Taking a Step Further

In our example we have simple mathematical operations to be implemented. Nevertheless if we were building features which depend on complex business logic, we should make our code more generic, so to speak. We must structure our components so they rely on interfaces, not on concrete implementations. This is also a SOLID principle, so it's highly recommended to know these concepts by heart.

3.1. Creating Classes for Implementations

Since we are taking into account complex implementations, let's create classes to our calculator operations; these classes will have a public method called calculate. For the sake of simplicity only one example will be shown

class AdditionOperation(CalculatorOperationsInterface):

    def __init__(self, first_number: float, second_number: float):
        self.__first_number = first_number
        self.__second_number = second_number

    # Here you can create more methods to apply business logic    

    def calculate(self) -> float:
        return self.__first_number + self.__second_number

One might be wondering what is CalculatorOperationsInterface. That is the interface we are applying to all of our operations.

3.2. Creating Interface

Now we need an interface for all our operations to conform. This interface requires a method which returns a float. That is how we create interfaces in Python

from abc import ABC, abstractmethod

class CalculatorOperationsInterface(ABC):

    @abstractmethod
    def calculate(init) -> float:
        '''calculate operations'''

Every class that implements this interface must have the calculate method.

3.3. Factory Pattern

In order to make it more generic, we need to have a component that will create the objects of the implementation classes. The component just mentioned is a Factory Design Pattern. This is how it is gonna work: depending on the operation - Enum - we're going to provide, our Factory has to create an object in which we can call the method calculate.

Since new Python versions 3.10+ have introduced Switch statements, that is what we are going to use, however you are free to use if statements too

class CalculatorOperationsFactory:

    @staticmethod
    def create_operation_object(
        operation: Operations, 
        first_number: float, 
        second_number: float
    ) -> CalculatorOperationsInterface:

        match operation:

            case Operations.ADDITION:
                return AdditionOperation(
                    first_number=first_number,
                    second_number=second_number
                )

            case Operations.SUBTRACTION:
                return SubtrationOperation(
                    first_number=first_number,
                    second_number=second_number
                ) 

            case Operations.MULTIPLICATION:
                return MultiplicationOperation(
                    first_number=first_number,
                    second_number=second_number
                ) 

            case Operations.DIVISION:
                return DivisionOperation(
                    first_number=first_number,
                    second_number=second_number
                )

            case _: # default case
                raise ValueError(f"Enter a valid {Operations} => '{operation}' was entered")

3.4. Refactored Calculator Function

Once we have created our components, we can present our new calculator function

def calculate(first_number: float, second_number: float, operation: Operations) -> float:

    operation_object = CalculatorOperationsFactory.create_operation_object(
        operation=operation,
        first_number=first_number,
        second_number=second_number,
    )

    return operation_object.calculate()

4. What We Have Built

Okay, you might be thinking that this whole structure for a simple calculator is kind of overkill, and you are not completely wrong. However, what we built is much more than a calculator. What we have accomplished is a maintainable solution to apply to large scale systems whenever we need to implement new features to an existing component, as long as it is possible to gather those input conditions inside an Enum.

4.1. How to Create New Features?

Creating new features is as simple as:

1. Creating a new Enum value
2. Creating the feature class implementing the interface created
3. Creating a new case in Factory

As you can perceive almost no existing code is modified to implement new features. This is ideal when we are in a project in which dozens of developers are contributing to, it reduces the possibility of conflicts and makes it easier to maintain.