Table of Contents
바인딩(Binding) & Reference Count & Garbage Collector
바인딩(Binding)
•
Python에서는 모든 데이터가 ‘객체’로 생성되어 메모리에 할당됨
◦
위의 그림에서 3은 정수 타입의 객체로 생성되어 ‘4396325856’ 메모리 주소에 할당됨
•
바인딩: 이후 변수를 할당해서 메모리에 할당된 객체의 주소를 가리키도록 함
Reference Count
Reference Count: 어떤 객체를 변수가 참조하거나 함수의 인자로 전달될 때, 또는 데이터 구조에 포함될 때 해당 객체의 Reference Count가 증가
import sys
# Using a large number outside Python's integer cache
a = 1000000
print(f"Initial reference count: {sys.getrefcount(a)}")
b = a # Create another reference
print(f"After b = a: {sys.getrefcount(a)}")
my_list = [1000000] # Add to list
print(f"After list creation: {sys.getrefcount(a)}")
def use_number(x):
print(f"In function: {sys.getrefcount(x)}")
use_number(a)
del b # Remove reference
print(f"After del b: {sys.getrefcount(a)}")
my_list.clear() # Clear list
print(f"After list clear: {sys.getrefcount(a)}")
Python
복사
•
Initial reference count: 4
◦
변수가 객체를 바인딩 했음으로 reference count가 증가 (+1)
◦
그 외에는 파이썬 인터프리터 (코드 객체가 객체를 참조해, (코드 객체는 함수, 모듈, 클래스 등 실행 가능한 코드 블록))에서 참조가 증가
•
After b = a: 5
◦
다른 변수가 같은 데이터 객체 바인딩
•
After list creation: 6
•
In function: 7
◦
함수 내부에서 a의 참조 횟수를 출력할 때, 함수 호출 자체가 참조 횟수를 증가
•
After del b: 5
◦
“5”를 바인딩 하는 변수 b 삭제 후 참조 횟수 출력
•
After list clear: 4
#### Integer interning
import sys
a = 0
print(f"Initial reference count: {sys.getrefcount(a)}")
b = a
print(f"After b = 0, reference count: {sys.getrefcount(a)}")
my_list = [a]
print(f"After adding to list, reference count: {sys.getrefcount(a)}")
Python
복사
•
Initial reference count for 0: 4294967295
•
After b = 0, reference count: 4294967295
•
After adding to list, reference count: 4294967295
•
Python은 자주 사용되는 작은 정수들(-5에서 256)을 메모리에 미리 생성해두고 캐싱
◦
초기화 시점에 자주 사용되는 정수들(-5에서 256)을 위한 integer object pool을 생성 → 객체들은 PyLongObject 타입으로 생성되어 특별한 메모리 영역에 저장
•
Integer object pool은 파이썬 인터프리터의 수명과 같음
◦
즉, 실제 참조 카운트를 추적하지 않고 최대값(2^32 - 1 = 4294967295)으로 설정
◦
이 이유는 refcount()를 줄이지 않아 절대 메모리에서 해제되지 않아야 하기 때문
Garbage Collector
Reference Counting
•
파이썬에서는 객체를 생성하고, 어떤 변수가 해당 객체를 바인딩하는 구조라고 했는데, 객체의 참조 카운트가 0이 되면 즉시 메모리에서 해제
•
자동이며 즉각적이며, CPython의 기본적인 메모리 관리 방식
Cyclic Garbage Collector
•
순환 참조와 같이 Reference Counting으로 해결할 수 없는 경우를 처리
•
주기적으로 실행
class Node:
def __init__(self, name):
self.name = name
self.next = None
node1 = Node("A")
node2 = Node("B")
node1.next = node2
node2.next = node1
del node1 # 외부 참조 제거
del node2 # 외부 참조 제거
Python
복사
node1 ─────> [Node A] ───┐
▲ │
│ │
│ ▼
node2 ─────> [Node B] ───┘
Python
복사
•
del node1, del node2 후에는 외부에서 이 객체들을 참조하는 변수가 없음 (외부에서 해당 객체들을 바인딩 하는 변수 X)
•
하지만 두 Node 객체는 여전히 서로의 next 속성을 통해 참조 중 (메모리 상에서 상호 참조중)
•
이런 경우 가비지 컬렉터가 주기적으로 검사하여 "도달 불가능한 순환 참조"로 판단하고 제거
참고사항: 세대 기반 가비지 컬렉션
•
3개의 세대로 관리 (0, 1, 2)
•
젊은 세대(0)가 더 자주 검사됨
•
각 세대마다 다른 임계값 적용
•
gc 모듈로 수동 제어 가능
Immutable vs. Mutable
a = "hello"
b = ["hello", "python"]
Python
복사
Memory in Python
•
객체(Object): 파이썬에서 문자열, 숫자, 리스트 등 모든 데이터는 객체로 생성됨
◦
각 객체는 고유의 메모리 주소를 가짐
◦
위의 예시에서, a = "hello"라고 선언하면, 문자열 객체 "hello"가 메모리 상에 생성되고, 변수 a는 이 객체를 참조
•
그렇다면 변수 ‘a’와 메모리 상에 올라간 객체 ‘hello’는 어떤 관계일까?
◦
"hello"라는 문자열 객체가 메모리(힙 메모리)에 생성
◦
변수 a는 이 객체를 가리키는 참조(객체의 메모리 주소)를 저장
▪
id(a)를 확인해보면 “hello”의 메모리 주소를 확인할 수 있음
◦
변수 자체는 메모리에 저장되지 않는 것일까?
▪
변수 자체도 메모리에 저장이 된다. 일반적으로 파이썬의 심볼 테이블(Symbol Table)에서 저장이 이루어 짐.
▪
정리해보면,
•
a는 심볼 테이블에서 관리
•
"hello"는 힙 메모리에 저장되고, 심볼 테이블에서 a가 이 객체를 가리키는 구조
▪
python은 기본적으로 심볼 테이블을 사용해 변수와 객체를 연결
•
전역 변수: 전역 심볼 테이블(Global Symbol Table)에 저장
•
지역 변수: 함수나 클래스 내부에서 선언된 변수는 로컬 심볼 테이블(Local Symbol Table)에 저장
•
심볼 테이블은 변수 이름(문자열)을 키로, 객체에 대한 참조를 값으로 가지도록 설계
Mutable vs. Immutable
•
Mutable(변경 가능) 객체: 생성된 후에도 내부 상태나 데이터를 변경할 수 있는 객체
•
Immutable(변경 불가능) 객체: 생성된 후에는 내부 상태나 데이터를 변경할 수 없는 객체
•
둘의 차이점을 볼 수 있는 관점은 아래와 같다.
◦
메모리 사용
▪
Immutable 객체를 변경하려고 하면 새로운 객체가 생성되어 메모리를 사용하지만, Mutable 객체는 동일한 객체 내에서 변경이 이루어짐
◦
해시 가능성
▪
Immutable 객체는 해시 가능하여 딕셔너리의 키나 집합의 요소로 사용할 수 있지만, Mutable 객체는 일반적으로 해시 불가능하여 이러한 용도로 사용X
•
예시
◦
Mutable 객체 예시
▪
리스트(list)
my_list = ["python2", "python3"]
my_list.append("python4")
print(my_list) # 출력: ["python2", "python3","python4]
Python
복사
•
list에 append를 수행해도 list 객체 자체 시작 주소 값은 변하지 않음을 알 수 있음, 반면 [0]번, [1]번, [2]번이 list 원소인 문자열 객체를 다시 바인딩하는 구조로 갖음. 따라서 list 원소를 추가하거나 삭제해도 리스트 객체의 시작 주소는 변하지 X.
▪
딕셔너리(dict)
my_dict = {'a': 1, 'b': 2}
my_dict['c'] = 3
print(my_dict) # 출력: {'a': 1, 'b': 2, 'c': 3}
Python
복사
▪
집합(set)
my_set = {1, 2, 3}
my_set.add(4)
print(my_set) # 출력: {1, 2, 3, 4}
Python
복사
◦
Immutable 객체 예시
▪
정수(int)
x = 10
x += 5 # 새로운 객체가 생성됨
print(x) # 출력: 15
Python
복사
▪
문자열(str)
s = "hello"
s += " world" # 새로운 문자열 객체가 생성됨
print(s) # 출력: "hello world"
s = "python2"
s = "python3" # 새로운 문자열 객체가 생성됨
print(s) # 출력: "python3"
Python
복사
•
문자열 객체는 수정 불가능하기 때문에 기존 객체(”hello”, “python2”)는 그대로 있고 새로운 문자열 객체가 생성
•
변수가 새로 생성된 문자열 객체("hello world", "python3")를 바인딩하게 되면 기존 문자열 객체는 가비지 컬렉터에 의해 자동으로 소멸
▪
튜플(tuple)
t = (1, 2, 3)
# t[0] = 4 # 오류 발생: 튜플은 변경 불가능
Python
복사
•
Example
◦
"t"*10000
▪
한 번의 연산으로 길이 n의 문자열을 생성
1.
메모리 사용량: O(n) (최종 문자열 하나만 생성)
2.
시간복잡도: O(n) (문자열의 길이에 비례)
◦
for i in range(10000): t += 't'
▪
이전에도 언급했듯, string은 불변(immutable) 객체, 따라서 문자열을 변경하려고 하면 실제로는 매 iteration마다 새로운 문자열 객체를 생성해야 함
t = '' # 초기 상태: 빈 문자열
for i in range(4):
t += 't'
print(f"단계 {i+1}: t = '{t}'")
Python
복사
•
단계 1: t = '' + 't' → 새로운 문자열 't' 생성
•
단계 2: t = 't' + 't' → 새로운 문자열 'tt' 생성
◦
이때, immutable 속성 때문에, 새로운 문자열을 만들기 위해서는 기존 문자열의 내용을 복사하여 새로운 메모리 공간에 저장.
◦
즉, 이때 기존의 't'가 새로운 메모리 공간에 복사
▪
추후 가비지 컬렉터에 의해 회수 가능.
◦
't'와 't'를 합쳐 새로운 문자열 'tt'가 생성
•
단계 3: t = 'tt' + 't' → 새로운 문자열 'ttt' 생성
•
단계 4: t = 'ttt' + 't' → 새로운 문자열 'tttt' 생성
1.
총 추가로 사용된 메모리 사용량은 약 n(n+1)/2에 비례하여 증가
a.
생성된 모든 문자열의 총 길이: 1 + 2 + 3 + ... + n = n(n + 1) / 2
2.
문자열의 길이에 비례하는 시간복잡도를 가짐
a.
총 시간 = O(1 + 2 + 3 + ... + n) = O(n(n + 1)/2) = O(n^2)
IS vs. ==
•
터미널에서 REPL(파이썬 인터프리터를 대화형 모드(interactive mode)) 실행시, 위와 같은 결과를 얻을 수 있음
•
두 값을 비교할 때 == 연산자를 사용
•
하지만, 만약 두 객체가 동일한 주소에 할당된 객체임을 비교하려면 is 연산자를 사용
◦
당연히, Integer Interning에 의해 특정 데이터 객체는 서로 다른 변수가 바인딩해도, 미리 지정된 메모리에 할당을 해놓기 때문에 id(a)==id(b) 가 True 를 반환
a = 1000
b = 1000
print(id(a), id(b))
print(a is b)
s1 = "this_is_a_long_string_example"
s2 = "this_is_a_long_string_example"
print(id(s1), id(s2))
print(s1 is s2)
Python
복사
•
하지만, 스크립트에서 실행하면 아래와 같은 결과를 얻는다
4381652720 4381652720
True
4382562368 4382562368
True
Python
복사
•
GPT 설명으로는 (1) 파이썬 최적화:스크립트 실행 환경에서 동일한 코드 블록 안에서 같은 값이 반복 사용되면, 파이썬은 동일 객체를 재사용하기 위해 일한 값(예: 1000)이 자주 사용될 경우 캐싱한다고 합니다. 또한 (2) REPL(대화형 쉘)과 달리, 스크립트 실행에서는 코드가 한 번에 컴파일되므로 파이썬 인터프리터가 정수를 최적화(캐싱)할 가능성이 더 높아집니다.
(정확하지는 않아서 추후에 더 공부하고 수정 예정입니다!)
Dictionary
•
zip()
◦
여러 개의 iterable(예: 리스트, 튜플 등)을 인덱스별로 묶어서 튜플을 생성하는 이터레이터를 반환
name = ['merona', 'gugucon'] # 이름 리스트
price = [500, 1000] # 가격 리스트
z = zip(name, price) # name과 price 리스트를 쌍으로 묶음
print(list(z)) # zip 객체를 리스트로 변환 후 출력
Python
복사
•
dictionary
아이스크림1 = {"메로나": 500, "구구콘": 1000}
아이스크림2 = dict(메로나=500, 구구콘=1000) # Class instantiation
아이스크림3 = dict([("메로나", 500), ("구구콘", 1000)])
Python
복사
◦
생성 방법
name = ['merona', 'gugucon']
price = [500, 1000]
z = zip(name, price) # zip 객체 생성
print(list(z)) # 이터레이터를 리스트로 변환 (튜플 리스트 생성)
# [('merona', 500), ('gugucon', 1000)]
z = zip(name, price) # zip 객체는 한 번 순회 후 재생성 필요
icecream = dict(z) # dict()는 zip 객체를 직접 순회하며 딕셔너리 생성
print(icecream)
# {'merona': 500, 'gugucon': 1000}
Python
복사
▪
zip() 함수는 **zip 객체(이터레이터)**를 반환 & 이터레이터는 데이터가 필요할 때 한 번에 하나씩 생성
▪
dict() 생성자는 입력으로 반복 가능한 객체(iterable)를 받을 수 있다. zip()이 반환한 이터레이터도 반복 가능한 객체에 해당하므로, dict()에서 바로 사용이 가능하다.
▪
ETC
name = ['merona', 'gugucon']
price = [500, 1000]
icecream = {k:v for k, v in zip(name, price)}
print(icecream)
Python
복사
⇒ 이거보단 iterable 객체인 zip()을 바로 dict()의 args로 넘겨 dictionary 객체를 생성하는게 더 간단하고 높은 가독성
name = ['merona', 'gugucon']
price = [500, 1000]
icecream2 = {k:v*2 for k, v in zip(name, price)}
print(icecream2)
Python
복사
⇒ 위처럼 생성도중에 value값을 조정하거나 조건을 걸때는 위의 방법이 용이
◦
setdefault()
data = {}
# 1. 첫 번째 setdefault() 호출
ret = data.setdefault('a', 123)
print(ret, data)
# 2. 두 번째 setdefault() 호출
ret = data.setdefault('a', 1)
print(ret, data)
'''
123 {'a': 123}
123 {'a': 123}
'''
Python
복사
▪
만약 key가 딕셔너리에 존재하지 않으면:
•
key를 추가하고, default 값을 해당 키의 값으로 설정
•
설정된 default 값을 반환
▪
만약 key가 딕셔너리에 이미 존재하면:
•
아무 작업도 하지 않고, 기존 값을 반환
iterable object & iterator object & generator
class MyIterable:
def __init__(self):
self.data = [1, 2, 3]
def __iter__(self):
return MyIterator(self.data)
class MyIterator:
def __init__(self, data):
self.data = data
self.index = 0
def __next__(self):
if self.index < len(self.data):
value = self.data[self.index]
self.index += 1
return value
else:
raise StopIteration
Python
복사
1. iterable object
iterable object는 반복 가능한 객체를 의미함
•
특징: __iter__() 메서드를 포함
•
내장 함수 iter(): iter() 함수는 iterable object의 __iter__() 메서드를 호출하여 iterator object를 반환합니다.
•
예시: 리스트, 튜플, 문자열, 집합, 딕셔너리 등은 모두 기본적으로 iterable object.
2. iterator object
iterator object는 iterable object에서 만들어진, 값을 차례차례 반환할 준비가 된 객체
•
특징:
◦
반드시 __next__() 메서드를 포함
◦
__next__()는 호출될 때마다 다음 값을 반환 & 만약 반환할 값이 없으면 StopIteration 예외를 발생
•
내장 함수 next(): next() 함수는 iterator의 __next__() 메서드를 호출
•
차이점 정리
특성 | iterable object | iterator object |
조건 | __iter__() 메서드를 포함 | __iter__()와 __next__() 포함 |
반환값 | __iter__() → iterator object | __iter__() → 자기 자신 반환 |
next() 호출 | 불가능 (직접 호출 불가) | 가능 |
예시 | 리스트, 튜플, 문자열 등 기본 자료형 | 리스트 이터레이터, 제너레이터 |
•
모든 기본 iterable객체는 __iter__() 메서드를 호출하면 그 객체에 대응되는 이터레이터(iterator) 객체를 반환하는가? ⇒ Yes!!!
iterable object | iterator object | 설명 |
리스트 (list) | list_iterator | 리스트의 요소를 순차적으로 반환 |
튜플 (tuple) | tuple_iterator | 튜플의 요소를 순차적으로 반환 |
문자열 (str) | str_iterator | 문자열의 각 문자를 순차적으로 반환 |
딕셔너리 (dict) | dict_keyiterator | 딕셔너리의 키를 순차적으로 반환 (기본 동작) |
집합 (set) | set_iterator | 집합의 요소를 순차적으로 반환 |
레인지 (range) | range_iterator | range 객체의 숫자를 순차적으로 반환 |
◦
dictionary
연산 | 설명 |
iter(my_dict) | 키의 iterator를 반환합니다 (dict_keyiterator). |
iter(my_dict.keys()) | 키의 iterator를 반환합니다. |
iter(my_dict.values()) | 값의 iterator를 반환합니다. |
iter(my_dict.items()) | 키-값 쌍의 iterator를 반환합니다 (dict_itemiterator). |
3. Generator
•
generator는 iterator를 생성하는 함수 또는 표현식
•
__iter__()와 __next__() 메서드를 자동으로 구현하므로, generator 객체는 직접적으로 iterator로 사용됩니다.
•
generator는 값을 필요할 때마다 생성하므로 메모리를 효율적으로 사용
•
generator는 일반적인 함수처럼 정의되지만, return 대신 yield를 사용하여 값을 반환
generator object는 iterator object이다.
◦
generator 함수 또는 generator 표현식을 통해 생성된 객체는 iterator 객체
◦
따라서 __iter__()와 __next__() 메서드를 자동으로 구현하며, 반복 가능한 데이터 소스로 사용할 수 있음
def my_generator():
yield 1 # 첫 번째 값 반환
yield 2 # 두 번째 값 반환
yield 3 # 세 번째 값 반환
gen = my_generator() # 제너레이터 객체 생성
print(next(gen)) # 1
print(next(gen)) # 2
print(next(gen)) # 3
# print(next(gen)) # StopIteration 예외 발생
Python
복사
generator 함수의 특징
1.
함수 정의에 yield 키워드를 포함
2.
함수가 호출되면, 값을 즉시 반환하지 않고, generator 객체를 반환
(아래 그림에서 generator는 인스턴스임)
3.
generator 객체는 호출 시 상태를 유지하며, next()가 호출될 때마다 yield 지점에서 실행을 재개
generator 표현식
gen = (x * x for x in range(5)) # 제너레이터 표현식
print(next(gen)) # 0
print(next(gen)) # 1
print(next(gen)) # 4
Python
복사
•
(x * x for x in range(5))는 리스트 컴프리헨션과 비슷하지만, generator 객체를 반환합니다.
•
값을 필요할 때 생성하므로 메모리 효율적
•
일반 함수와 generator 함수와의 비교
특징 | 일반 함수 | generator 함수 |
반환 방식 | 모든 값을 한 번에 반환 | yield를 통해 값을 하나씩 반환 |
반환 시점 | 함수 호출 시 즉시 반환 | next() 호출 시 반환 |
상태 유지 | 함수가 종료되면 상태가 사라짐 | 호출마다 상태를 유지함 |
메모리 효율성 | 모든 데이터를 메모리에 저장 | 값을 필요할 때 생성 |
◦
예시
▪
일반함수 사용 → def 빵만들기(n)에서 100개의 정수를 담은 파이썬 리스트를 리턴
def 빵만들기(n):
빵쟁반 = []
for i in range(n):
빵 = "빵" + str(i) # 빵0, 빵1, ..., 빵99
빵쟁반.append(빵)
return 빵쟁반
def 빵포장(빵):
print("{} 포장완료".format(빵))
for i in 빵만들기(100):
빵포장(i)
Python
복사
▪
generator 함수 사용 빵포장(빵)을 호출할대마다 빵변수가 빵X를 메모리에서 바인딩 (미리 모든 빵들을 메모리에 로드할 필요가 X)
def 빵만들기(n):
for i in range(n):
빵 = "빵" + str(i) # 빵0, 빵1, ..., 빵99
yield 빵
def 빵포장(빵):
print("{} 포장완료".format(빵))
for i in 빵만들기(100):
빵포장(i)
Python
복사
•
정리
◦
generator는 Python의 iterator를 생성하는 특별한 방법
◦
일반 함수와 달리 yield 키워드를 사용하여 값을 하나씩 반환하며, 메모리를 효율적으로 사용
◦
generator는 Python의 이터레이션 프로토콜(__iter__와 __next__)을 자동으로 구현하여 iterator 객체로 사용
◦
이러한 특성 때문에 generator는 파일 읽기, 데이터 처리, 스트림 처리 등 메모리 효율성이 중요한 작업에서 널리 사용
Decorator
decorator
def logging_decorator(func):
def wrapper(*args, **kwargs):
print(f"함수 {func.__name__} 실행 시작")
result = func(*args, **kwargs)
print(f"함수 {func.__name__} 실행 종료")
return result
return wrapper
# 실제 비즈니스 로직은 아래 함수들에서 구현
@logging_decorator
def calculate_price(quantity, price):
return quantity * price
@logging_decorator
def get_user_info(user_id):
return f"User {user_id} info"
Python
복사
•
데코레이터는 여러 함수간의 종속(?) 관계를 편하게 구현하기 위해 정의되었다.
◦
크게 (1) decorator 역할을 하는 함수 (2) decorator의 인자로 들어가는 함수 이렇게 구분할 수 있다.
▪
decorator 역할을 하는 함수는
•
@ 뒤에 오는 표현식의 최종 결과가 함수를 인자로 받을 수 있어야 한다.
•
이는 직접적으로 함수를 인자로 받는 함수이거나 (e.g., logging_decorator )
•
함수를 인자로 받는 함수를 반환하는 함수일 수 있습니다.
def deco_with_args(arg1, arg2): # 1. 일반 인자를 받는 함수
def real_deco(func): # 2. 함수를 인자로 받는 내부 함수
def wrapper(): # 3. 실제 래퍼 함수
print(f"인자들: {arg1}, {arg2}")
func()
return wrapper
return real_deco
@deco_with_args("hello", 123)
def my_func():
print("함수 실행")
Python
복사
◦
따라서 decorator는 보통 반복되는 패턴이나 공통 기능을 정의하는데 많이 사용된다. (일종의 skeleton function의 역할)
◦
실제로 구현하고 싶은 로직은 개별 함수에서 구현하는 방식이 코드 재사용성과 유지보수 측면에서 매우 효율적이다. (calculate_price(), get_user_info(), my_func() )
@property
•
@property는 메서드를 속성(attribute)처럼 사용할 수 있게 해주는 데코레이터이다.
class Car:
def __init__(self, model):
self._model = model # 내부 변수임을 표시
@property
def model(self):
return self._model # 실제 데이터는 _model에 있지만
# 외부에서는 그냥 car.model로 접근
car = Car("GV80")
print(car.model) # 내부 구현은 숨기고 깔끔한 인터페이스 제공
Python
복사
⇒ @property는 메서드를 읽기 전용 속성(attribute)처럼 만들어주는 역할을 하기 때문에 일반적인 클래스 속성이 항상 어떤 값을 가지고 있는 것처럼, property도 항상 어떤 값을 가지고 있어야 한다.
__closure__
•
__closure__ 는 외부 함수 안에 내부 함수를 정의하고 정의된 내부 함수를 리턴하는 구조를 가졌는데,
◦
외부 함수를 어떤 객체로 선언하는 순간 → 선언한 변수가 외부함수 호출을 통해 반환될 내부 함수를 바인딩
◦
내부 함수 내부의 __closure__ 속성을 통해 내부 함수가 참조하는 외부 함수의 지역 변수의 값을 바인딩하고 있으며 이 값은 inner 함수가 호출(선언한 변수를 가지고 내부함수를 호출할때)될 때 사용
def outer(out1):
def inner(in1):
print("inner function called")
print("outer argument: ", out1)
print("inner argument: ", in1)
return inner
f = outer(1)
f(10)
Python
복사
Logging
•
Unix/Linux 시스템의 output redirection
def hap(a, b):
ret = a + b
print(a, b, ret) # 함수의 입/출력 확인을 위한 print 구문
return ret
result = hap(3, 4)
python run.py > log.txt
Python
복사
◦
스크립트내 아래의 내용이 로깅됨
▪
print() 함수로 출력되는 모든 내용
▪
sys.stdout.write()로 출력되는 내용
▪
에러 메시지는 표준 에러(stderr)로 출력되므로 log.txt에 저장되지 않기 때문에 에러 메세지도 같이 로깅하기 위해서는 python run.py > log.txt 2>&1
•
logging 모듈
◦
print 대신에 logging.info() 함수를 사용
◦
logging 레벨에 따라 로깅되는 값의 차이가 존재함
레벨 | 로깅 함수 | 사용할 때 |
DEBUG | debug() | 상세한 정보를 출력 |
INFO | info() | 예상대로 작동하는지를 확인 |
WARNING | warning() | 소프트웨어는 정상 동작하는데 예상치 못한 일이 발생한 것에 대해 표시 |
ERROR | error() | 소프트웨어의 일부가 정상적으로 동작하지 않는 경우에 대해 표시 |
CRITICAL | critical() | 심각한 에러 상황에 대해 표시 |
◦
WARNING이 default이기에 level을 INFO로 변경하고 아래 파일을 실행하면 콘솔에 info()의 내용이 출력됨
▪
info에는 str type만 인자로 받을 수 있음
import logging
logging.basicConfig(level=logging.INFO)
def hap(a, b):
ret = a + b
logging.info(f"input: {a} {b}, output={ret}")
return ret
result = hap(3, 4)
Python
복사
▪
콘솔출력을 filename으로 파일출력으로 변경가능
import logging
logging.basicConfig(filename="mylog.txt", level=logging.INFO)
def hap(a, b):
ret = a + b
logging.info(f"input: {a} {b}, output={ret}")
return ret
result = hap(3, 4)
Python
복사
Unit Test
•
유닛 테스트는 프로그램 구현할 때 유닛 단위로 테스트를 진행하는 것을 의미
◦
유닛은 함수이며,
◦
함수를 구현한 후 함수의 입력과 예상되는 출력을 비교함으로써 함수를 테스트
•
파이썬 유닛 테스트에서는 주로 두 가지 모듈이 사용된다. (unittest, pytest)
◦
검증할 유닛/함수
'''math_mine'''
def average(a):
return sum(a) / len(a)
Python
복사
◦
테스트 함수
▪
pytest는 test_ 라는 접두사 혹은 _test 라는 접미사로 끝나는 파일을 찾아서 파일을 실행
'''math_test'''
from math_mine import *
def test_average():
assert average([1, 2, 3]) == 1
Python
복사
(base) ~ % pytest
============================================================================ test session starts ============================================================================
platform darwin -- Python 3.12.4, pytest-7.4.4, pluggy-1.0.0
rootdir: ~
plugins: anyio-4.2.0
collected 1 item
math_test.py . [100%]
============================================================================= 1 passed in 0.46s =============================================================================
Python
복사
Concurrency Programming & Thread
•
Concurrency (동시성)
◦
하나의 CPU 코어가 여러 작업을 번갈아가면서 처리하는 방식 (멀티태스킹)
•
Parallelism (병렬성)
◦
여러개의 CPU 코어가 여러 작업을 동시에 처리하는 방식
•
병렬성이 항상 동시성보다 좋다고 할 수 있는가?
◦
코어의 물리적인 개수 확장에 제한이 있기 때문에 task에 따라서 동시성을 활용할것인지, 병렬성을 활용할 것인지 선택해야한다.
◦
I/O 위주의 작업이라면 실제로 CPU가 처리해야하는 시간은 상대적으로 적기에 동시성이 유리하다.
▪
I/O Bound
# I/O 작업의 예
with open('large_file.txt', 'r') as f:
content = f.read() # 이 부분에서 대부분의 시간 소요
# CPU는 파일이 디스크에서 메모리로 로딩될 때까지 대기
Python
복사
1.
CPU: "파일 읽어와!" 명령 전달 (0.1ms)
2.
디스크: 파일 찾고 읽어오는 중... (100ms)
3.
CPU: 대기... 대기... 대기... (이 시간동안 CPU는 놀고 있음)
4.
디스크: "자, 다 읽었어요!"
5.
CPU: 데이터 처리 (0.1ms)
⇒ 즉, 전체 100.2ms 중에서:
•
CPU가 실제로 일하는 시간: 0.2ms
•
CPU가 그냥 대기하는 시간: 100ms
▪
CPU Bound
# CPU 작업의 예
result = 0
for i in range(1000000):
result += i * i # CPU가 계속 계산에 바쁨
Python
복사
⇒ CPU가 계속 일을 하고 있어서 대기 시간이 거의 없다.
▪
I/O 위주의 작업은 CPU가 대기 시간 동안 다른 작업을 처리하도록 두는 것이 효율적이다.
•
Thread
◦
프로세스의 실행 단위
◦
스레드는 프로세스의 가상메모리 공간을 공유
(ex1) 서브 스레드들보다 메인 스레드가 더 빨리 실행되었음에도 메인 스레드가 서브 스레드의 작업이 종료될떄까지 기다렸다가 서브 스레드의 작업이 모두 완료되면 종료
import threading
import time
class Worker(threading.Thread):
def __init__(self, name):
super().__init__()
self.name = name # thread 이름 지정
def run(self):
print("sub thread start ", threading.currentThread().getName())
time.sleep(3)
print("sub thread end ", threading.currentThread().getName())
print("main thread start")
for i in range(5):
name = "thread {}".format(i)
t = Worker(name) # sub thread 생성
t.start() # sub thread의 run 메서드를 호출
print("main thread end")
'''
main thread start
sub thread start thread 0
sub thread start thread 1
sub thread start thread 2
sub thread start thread 3
sub thread start thread 4
main thread end
sub thread end thread 0
sub thread end thread 1
sub thread end sub thread end thread 2
thread 4
sub thread end thread 3
'''
Python
복사
(ex2) 데몬 스레드: 메인 스레드가 종료되면 서브 스레드가 자신의 실행 상태와 상관없이 종료되는 서브 스레드 (예를 들어, 파일을 서브 스레드로 받는데, 메인스레드가 종료되면 (e.g., 메인프로세스 킬) 그대로 서브 스레드를 죽여야함. 이때 서브스레드를 데몬 스레드로 생성)
import threading
import time
class Worker(threading.Thread):
def __init__(self, name):
super().__init__()
self.name = name # thread 이름 지정
def run(self):
print("sub thread start ", threading.currentThread().getName())
time.sleep(3)
print("sub thread end ", threading.currentThread().getName())
print("main thread start")
for i in range(5):
name = "thread {}".format(i)
t = Worker(name) # sub thread 생성
t.daemon = True
t.start() # sub thread의 run 메서드를 호출
print("main thread end")
'''
main thread start
sub thread start thread 0
sub thread start thread 1
sub thread start thread 2
sub thread start thread 3
sub thread start thread 4
main thread end
'''
Python
복사
◦
Join
▪
서브 스레드의 동기화, 메인 스레드가 서브스레드에서 처리한 값을 처리해야하는 경우, 메인 스레드를 서브 스레드 끝날때까지 대기시키는 method
def worker(name):
print(f"{name} 작업 시작")
time.sleep(3) # 작업 중...
print(f"{name} 작업 끝")
# 스레드 생성 및 시작
t1 = threading.Thread(target=worker, args=("Thread-1",))
t2 = threading.Thread(target=worker, args=("Thread-2",))
t1.start() # Thread-1 시작
t2.start() # Thread-2 시작
# 이 시점에서:
# - Thread-1은 작업 실행 중
# - Thread-2도 작업 실행 중
# - 메인 스레드도 실행 중
t1.join() # 메인 스레드만 Thread-1이 끝날 때까지 대기
# Thread-1은 계속 작업 실행
# Thread-2도 계속 작업 실행
t2.join() # 메인 스레드만 Thread-2가 끝날 때까지 대기
# Thread-2는 계속 작업 실행
Python
복사
▪
메인 스레드는 실행 중에 join() 문을 만나는 순간 (e.g., t1.join(), "아, 이 시점에서 이 스레드가 끝날 때까지 기다려야겠구나"라고 인식하고 대기
▪
Join 과정의 시각화
threads = []
# 3개의 스레드 생성 및 시작
for i in range(3):
thread = Worker(i)
thread.start()
threads.append(thread)
# 각 스레드에 대해 join() 호출
for thread in threads:
thread.join()
print("main thread end") # 모든 스레드가 완료된 후에만 실행
Python
복사
⇒ 아래 보이는것처럼 join()을 만난 순간 메인 스레드는 서브스레드가 완료될때까지 대기사태로 있는 것을 확인할 수 있음
Thread 1: [실행중...] [완료]
Thread 2: [실행중........] [완료]
Thread 3: [실행중....] [완료]
Main: [대기중... join()... 대기중...] [계속 실행]
↑ ↑ ↑
T1 T2 T3
완료 완료 완료
Python
복사
Multiprocessing
Process
•
개별 프로세스를 직접 생성하고 관리
•
단일 작업 수행에 적합
Pool
•
미리 지정된 개수만큼의 프로세스를 생성하여 유지하는 작업자 프로세스 풀
import multiprocessing as mp
# 시작 방식 설정
mp.set_start_method('spawn') # 또는 'fork', 'forkserver'
# Pool 생성
with Pool(4) as pool:
pool.map(work, data)
Python
복사
Spawn vs Fork
Fork (Unix/Linux 기본)
•
부모 프로세스의 모든 자원을 복사
•
메모리를 복사하므로 시작이 빠름
•
예측하기 어려운 동작 가능성 있음
부모 프로세스
├─ 전체 메모리 상태
├─ 열린 파일
├─ 스레드
└─ 기타 모든 자원
↓ (모두 복사)
자식 프로세스
Python
복사
Spawn (Windows 기본)
•
새로운 Python 인터프리터로 시작
•
필요한 자원만 가져옴
•
더 안전하고 예측 가능한 동작
•
시작이 상대적으로 느림
부모 프로세스 자식 프로세스
└─ 필요한 자원 정보 전달 ─→ (새로운 파이썬 인터프리터로 시작)
└─ 필요한 자원만 초기화
Python
복사
Spawn vs Fork example
file = open("some_file.txt") # 부모 프로세스에서 파일 열기
def worker():
# fork: 파일 핸들러가 복사되어 예상치 못한 동작 가능
# spawn: 새로 시작하므로 파일 핸들러가 없음 - 더 예측 가능
if __name__ == "__main__":
p = Process(target=worker)
p.start()
Python
복사
. join()
•
해당 프로세스가 종료될 때까지 메인 프로세스를 대기시키는 blocking 메서드
import multiprocessing as mp
from multiprocessing import Pool
class Worker:
def __init__(self):
pass
def run(self, value):
pname = mp.current_process().name
print(pname, value)
if __name__ == "__main__":
w = Worker()
# 4개의 프로세스를 가진 풀 생성
with Pool(4) as pool:
# 여러 값을 병렬 처리할 수 있도록 수정
values = ["hello", "world", "python", "pool"]
# w.run 메서드를 values의 각 항목에 대해 실행
pool.map(w.run, values)
print("Main Process")
Python
복사
•
모든 작업이 완료될 때까지 자동으로 기다림 (pool의 __exit__에서 자동으로 join 수행)
◦
join()을 만나면 메인 프로세스는 대기 상태로 들어감
◦
서브 프로세스는 계속 실행됨
◦
풀의 모든 프로세스가 종료된 후 "Main Process" 출력
•
도식화
메인 프로세스 (1개)
│
├── Worker 프로세스 1
├── Worker 프로세스 2
├── Worker 프로세스 3
└── Worker 프로세스 4
Python
복사
서브 프로세스 상태 및 종료
import multiprocessing as mp
import time
def work():
while True:
print("sub process is running")
time.sleep(1)
if __name__ == "__main__":
p = mp.Process(target=work, name="SubProcess")
p.start()
time.sleep(5)
p.kill()
print("Status right after kill(): ", p.is_alive())
# 잠시 대기하여 프로세스가 실제로 종료되기를 기다림
time.sleep(0.1)
print("Status after small delay: ", p.is_alive())
p.join()
print("Status after join(): ", p.is_alive())
Python
복사
⇒ p.kill()을 해도 곧바로 프로세스가 종료되진 않음
멀티프로세싱과 PyQt
•
I/O 작업 위주의 경우 스레드를 사용해도 성능에 문제가 없음 (I/O 위주의 작업의 경우 실제로 CPU가 처리하는 시간보다 인터넷이나 하드 드라이브에서 데이터를 읽는데 많은 시간을 사용하게 때문)
•
실제로 CPU를 사용하는 복잡한 계산에 의한 것이라면 스레드보다는 프로세스를 사용하는 것이 좋음. 특히 파이썬은 GIL (Global Interpeter Lock)의 제한이 있어서 스레드를 여러개 사용하는 멀티 스레드의 성능이 더 좋지 않음
•
producer는 어떤 데이터를 생성하는 역할로 메인 프로세스와 별도로 생성된 프로세스
•
consumer는 main process내 스레드로 queue를 계속해서 지켜보면서 producer에 가져와 gui에 emit()하는 스레드
◦
데이터의 흐름 도식화
[Producer Process] → [Queue] → [Main Process/Consumer Thread]
시간 데이터 put() get() GUI 표시
Python
복사
◦
아래 코드의 흐름에 대한 다이어그램
▪
스레드 인스턴스화하고 gui랑 연결시킨 후 thread run해서 queue로부터 계속 데이터를 가져오도록 하는 과정
Consumer Thread (QThread)
│
▼
[while True 루프]
│
▼
q.empty()? ──── Yes ──► 다시 체크
│
│ No
▼
q.get() 데이터 가져오기
│
▼
poped.emit(data) 시그널 발생
│
│ connect()로 연결됨
│ │
└────────────┘
▼
print_data 실행
│
▼
statusBar에 데이터 표시
Python
복사
◦
Code
# 필요한 라이브러리 임포트
from PyQt5.QtWidgets import * # PyQt5의 위젯 관련 모듈
from PyQt5.QtCore import * # PyQt5의 코어 기능 모듈
from multiprocessing import Process, Queue # 멀티프로세싱 관련 모듈
# 생산자 함수 정의
def producer(q):
proc = mp.current_process()
print(proc.name) # 현재 프로세스 이름 출력
while True:
now = datetime.datetime.now() # 현재 시간 가져오기
data = str(now) # 시간을 문자열로 변환
q.put(data) # Queue에 데이터 삽입
time.sleep(1) # 1초 대기
# 소비자 클래스 정의 (QThread 상속)
class Consumer(QThread):
poped = pyqtSignal(str) # 문자열을 전달하는 시그널 정의
def __init__(self, q):
super().__init__()
self.q = q # Queue 객체 저장
def run(self):
while True:
if not self.q.empty(): # Queue가 비어있지 않으면
data = q.get() # 데이터 가져오기
self.poped.emit(data) # 시그널 발생
# 메인 윈도우 클래스
class MyWindow(QMainWindow):
def __init__(self, q):
super().__init__()
self.setGeometry(200, 200, 300, 200) # 윈도우 크기와 위치 설정
# 소비자 스레드 생성 및 시작
self.consumer = Consumer(q) # 이 시점에서는 아직 실행되지 않음
self.consumer.poped.connect(self.print_data) # 시그널-슬롯 연결 # 실행 전에 연결 설정
self.consumer.start() # 이때 run() 메서드가 별도 스레드에서 실행 시작
@pyqtSlot(str)
def print_data(self, data):
self.statusBar().showMessage(data) # 상태바에 데이터 표시
if __name__ == "__main__":
q = Queue() # 프로세스 간 통신을 위한 Queue 생성
# 생산자 프로세스 생성 및 시작
p = Process(name="producer", target=producer, args=(q,), daemon=True)
p.start()
# GUI 메인 프로세스
app = QApplication(sys.argv) # Qt 애플리케이션 생성
mywindow = MyWindow(q) # 메인 윈도우 생성
mywindow.show() # 윈도우 표시
app.exec_() # 이벤트 루프 시작
Python
복사
Function
#### 함수
def hello():
print("hello")
Python
복사
•
함수 이름은 함수 객체를 바인딩 (변수가 객체 값을 바인딩 하듯이)
#### 가변인자
•
args
◦
함수를 정의할 때 인자값의 개수를 가변적으로 정의해야 하는 경우 위치 가변 인자인 args를 인자로 전달
◦
args라는 변수는 여러 개의 입력에 대해 튜플로 저장한 후 이 튜플 객체를 바인딩
def foo(*args):
print(args)
data = [1,2,3]
foo(*data)
foo(1, 2, 3)
foo(1, 2, 3, 4)
Python
복사
•
kwargs
◦
키워드 가변 인자는 **를 붙여주고 키워드와 값의 형태로 인자를 전달
◦
호출부에서 a=1, b=2, c=3와 같은 형태로 인자를 전달하면 kwargs라는 변수가 딕셔너리 객체를 바인딩함
◦
딕셔너리 형태의 자료구조를 **로 풀어서 kwargs에 바인딩 시키면 알아서 함수에 키워드 가변 인자로 전달됨
def foo(**kwargs):
print(kwargs)
foo(a=1, b=2, c=3)
params = {'a': 1, 'b': 2}
foo(**params)
Python
복사
#### lambda
•
함수식의 간단한 표현
•
바인딩할 함수이름 = lambda 인자 : return할 값
•
a = lambda x : 5 * x
◦
a가 함수 객체를 바인딩함 (기존 함수랑 똑같은 로직)
#### hasattr & getattr
class Car:
def __init__(self):
self.wheels = 4
def drive(self):
print("drive")
mycar = Car()
Python
복사
•
hasattr(객체, 이름) 내장 함수의 인자로 넘겨주는 문자열 타입의 이름이 객체에 존재하면 True이고 그렇지 않으면 False
hasattr(mycar, "wheels") > True
hasattr(mycar, "drive") > True
Python
복사
•
getattr(객체, 이름) 내장 함수의 인자로 넘겨주는 문자열 타입의 이름을 가진 attribute를 가져오는 것
print(getattr(mycar, "wheels")) > 4
method = getattr(mycar, "drive")
method() > drive
Python
복사
#### Closure
•
클로저는 함수 안에 함수를 정의하고, 내부 함수가 외부 함수의 변수를 기억하고 접근할 수 있게 하는 기능
def outer():
num = 3
def inner():
print(num)
return inner
f = outer()
f()
Python
복사
◦
위의 함수 안의 함수에서 f라는 변수가 바인딩하는 함수 객체는 print(num)이라는 코드를 바인딩.
◦
f()를 통해서 f 가 바인딩하는 함수 객체를 출력하면 num 값은 에러가 나야하는데, closure에 의해서 3이 출력됨.
•
클로저는
◦
내부 함수가 외부 함수의 변수를 참조
◦
외부 함수가 종료된 후에도 내부 함수는 외부 함수의 변수를 기억
◦
각 클로저는 독립적인 환경을 가짐
def create_counter():
count = 0 # 클래스의 인스턴스 변수 같은거임, 클래스 변수
def increment():
nonlocal count # 외부 변수를 수정하기 위해 nonlocal 사용
count += 1
return count
return increment
# 카운터 생성
counter = create_counter()
print(counter()) # 출력: 1
print(counter()) # 출력: 2
print(counter()) # 출력: 3
# 새로운 카운터 생성 (독립적으로 동작)
counter2 = create_counter()
print(counter2()) # 출력: 1
Python
복사
Appendix
•
lst.append(4)
[1, 2, 3] -> [1, 2, 3, 4]
(바로 기존 리스트 끝에 4를 추가)
Python
복사
→ append는 단일 객체를 리스트 끝에 추가하는 전용 메소드
→ 임시 리스트를 생성X
→ 메모리 재할당이 필요한 경우, Python은 보통 현재 크기의 약 2배 정도로 미리 공간을 할당
•
lst += [4]
[1, 2, 3] -> [1, 2, 3] + [4] -> [1, 2, 3, 4]
↑ ↑ ↑
원본 임시리스트 생성 후 extend
Python
복사
→ += 연산자는 Python 내부적으로 list.__iadd__([a]) 메소드를 호출
→ [a]라는 새로운 임시 리스트가 생성되었다가 소멸됩니다
→ 여러 요소를 한 번에 추가할 때 유용합니다 (예: list += [1,2,3])
•
list의 mutable 특성
lst = lst + [4]
•
새로운 리스트를 생성
•
기존 리스트와 [4]를 합친 새 리스트를 만듦
•
원본 리스트는 그대로 두고 새 리스트의 참조를 lst에 할당