-
I/O multiplexing이 가능한 채팅 프로그램Side Project 2023. 3. 26. 23:43
채팅 프로그램은 입사하고 처음 과제로 받았던 파일럿 프로젝트인데 이번 기회에 OOP로 리팩토링을 하면서 다시 들여다본다. 당시에 국비로 머신러닝, 데이터 분석을 6개월 배우고 입사했기 때문에 개발에 대한 경험은 데이터 파이프라인을 구축해 본 경험이 전부였다. 이 분야에서 뭘 잘하는지도 모르겠고 일단 부딪히면서 생각해보자라는 주의로 입사했는데 지금 생각해보면 어디서 그런 밑도끝도 없는 용기가 나왔는지 신기하다. 결론은 실무를 하며 배우는 부분이 참 많고 여러가지 경험을 하고 있기 때문에 잘 한 선택이라고 생각한다.
각설하고 본론으로 들어가면 이 포스팅은 다중 client가 하나의 서버에서 대화를 주고받을 수 있는 채팅 프로그램에 대한 글이다. 채팅 서버는 client들의 요청을 비동기적으로 수행한다. 서버가 client의 요청을 동기적으로 처리한다면 client 1의 요청을 처리할 때는 client 2의 요청은 처리할 수가 없을 것이다.
갖추고 있는 기능은 아래와 같다.
1. 클라이언트는 채팅방 번호와 닉네임을 입력하고 채팅방에 들어가며, 중복 닉네임은 허용하지 않는다.
2. 입장한 채팅방 안에서만 채팅할 수 있다.
3. 클라이언트는 특정 유저와 귓속말을 할 수 있다. 귓속말을 할 때는 대화 내용이 다른 유저들에게 공개되지 않는다.
4. 클라이언트는 아이디를 변경할 수 있다.
5. 클라이언트는 소속된 채팅방의 모든 유저들의 닉네임을 확인할 수 있다.가장 핵심이 되는 모듈은 select이다. select는 동시에 여러 입출력을 처리함으로써 I/O multiplexing을 가능하게 한다. threading을 사용할 수도 있었는데 thread는 그 수가 늘어날수록 시스템에 부담이 되므로 효율적이지 않지만 select는 처음에 블로킹되어 있다가 특정 이벤트가 발생하면 작동하는 방식이라고 해서 select를 선택했다. 시야가 조금 더 넓어지니 selectors라는 모듈도 보인다. 후,,😂 (selectors는 select를 확장하여 고수준 I/O 멀티플렉싱을 가능하도록 한 모듈로, select 대신 사용하도록 권장하는 모듈이다. - 점프 투 파이썬 https://wikidocs.net/125716)
여하튼 select는 서버 소켓에 클라이언트가 접속하는지 감시하며, 아래 코드와 같이 30초마다 블로킹을 해제할 수 있도록 설정했다. 참고로 read_sockets는 수신한 데이터를 가진 소켓, write_sockets는 블로킹되지 않고 데이터를 전송할 수 있는 소켓, error_sockets는 예외상황이 발생한 소켓을 의미한다. 여기서는 write_sockets와 error_sockets를 고려하지 않았다.
read_sockets, write_sockets, error_sockets = select.select(connection_list, [], [], 30)server.py를 실행시키면 IPv4에 대해 TCP 통신이 가능한 소켓 객체를 얻어 HOST와 PORT를 바인딩한 후 해당 주소로 들어오는 클라이언트의 요청을 listen한다.
그 다음으로 client.py를 실행시켜 server에 연결을 시도한다.
서버에 연결하기 위해 클라이언트 application에서 connect() 시스템 콜을 사용하면 TCP 스택에 연결되고 TCP 스택은 3-way handshaking을 시작한다. 3-way handshaking 과정을 거쳐 서버는 accept() 상태가 되고 서버와 클라이언트는 서로 데이터를 송수신할 수 있는 상태가 된다. 서버의 accept() 시스템 콜은 연결이 완전히 수립된 이후에 클라이언트와의 통신을 위한 새로운 소켓 파일 디스크립터를 반환해 서버 어플리케이션에게 제공한다.
3-way handshaking 과정을 조금 더 자세히 살펴보면 ① 클라이언트가 서버로 연결을 시작할 때 SYN 패킷을 보내며 SYN-SENT 상태가 된다. ② 서버는 SYN 패킷을 수신하면 SYN-RECEIVED 상태가 되고 클라이언트에게 SYN+ACK 패킷을 보냄으로써 자신도 동기화를 요청함을 나타낸다. ③ 클라이언트는 이에 대한 응답으로 ACK 패킷을 보내 ESTABLISHED 상태가 되며 서버도 ACK 패킷을 받으면 ESTABLISHED 상태가 된다. 이와 같이 패킷이 3번 오가며 서버가 클라이언트와의 연결을 수립하는 3-way handshaking이 완료된다.
하나의 서버가 다중 클라이언트의 메세지를 처리할 수 있는 구조이기 때문에 아래와 같이 2개 이상의 클라이언트가 접속해도 실시간으로 채팅이 가능하다. 귓속말, 아이디 변경, 참여 중인 멤버 목록 확인 기능이 모두 제대로 동작하는지 확인했다.
세 명의 클라이언트들이 실시간 채팅을 하는 모습 
b -> c 귓속말 테스트: a는 b가 보낸 귓속말을 확인할 수 없다 
a -> yoojin 아이디 변경 테스트 
a가 참여 중인 멤버 확인 네트워크에 대한 이해를 위해 asyncio 등 socket I/O 동작을 wrapping한 비동기 라이브러리의 도움을 받지 않고 low level의 I/O interface를 직접 구현해보고 싶었다. 따라서 I/O multiplexing을 위한 system call로 select를 사용했다. 그런데 select 모듈은 I/O 이벤트 처리 시 O(n)의 시간복잡도를 가지므로 이에 대한 개선이 필요하다. 언제가 될지는 모르겠지만 추후 이벤트 처리에 대한 로직과 I/O 모델에 대한 공부를 하고 시간복잡도 개선에 대한 고도화를 진행하려고 한다. 아래 코드는 서버와 클라이언트를 구현한 파이썬 소스이며 첨부한 깃헙 주소로 가면 소스코드를 내려받을 수 있다.
https://github.com/Yoojin-An/chatting-room[server.py]
# -*- coding: utf8 -*- import socket import sys import select import datetime import json class OverlappedError(Exception): def __init__(self): super().__init__("\n※ 이미 사용중인 닉네임입니다. 다시 입력해주세요.\n") class ChatRoom(object): global connectionStatus def find_conn(self, connection_dict, nick_val): return next(conn for conn, nick in connection_dict.items() if nick_val == nick) def find_nick(self, connection_dict, conn_val): return next(nick for conn, nick in connection_dict.items() if conn_val == conn) def find_room_num(self, connection_status, conn_val): return next(room_num for room_num, connection_dict in connection_status.items() if conn_val in connection_dict) def manage_client(self, conn, data): self.nickname = data['client_nick'] self.room_num = data['room_num'] # 클라이언트가 신규 채팅방을 입력한 경우 if self.room_num not in connectionStatus: connectionStatus[self.room_num] = {} connectionStatus[self.room_num][conn] = self.nickname conn.send('Y'.encode()) print(f"[INFO] [room_num_{self.room_num}] {self.nickname}님 접속") for sock in connectionStatus[self.room_num]: if sock != conn: sock.send(f"[INFO] {self.nickname}님 접속".encode()) # 클라이언트가 기존 채팅방을 입력한 경우 else: try: if self.nickname not in connectionStatus[self.room_num].values(): # 같은 채팅방 안에 중복 닉네임이 있는지 여부 판단 connectionStatus[self.room_num][conn] = self.nickname conn.send('Y'.encode()) # 클라이언트에게 닉네임 등록 메시지 전달 print(f"[INFO] [room_num_{self.room_num}] {self.nickname}님 접속") for sock in connectionStatus[self.room_num]: if sock != conn: sock.send(f"[INFO] {self.nickname}님 접속".encode()) else: raise OverlappedError # 중복닉네임 있으면 에러메시지 전달 except OverlappedError as e: conn.send(f'{e}'.encode()) def manage_message(self, data): global connectionStatus room_num = self.find_room_num(connectionStatus, conn) connection_dict = connectionStatus[room_num] if data.split(' ')[1] == '!whisper': sender_conn = conn receiver = data.split(' ')[2] if receiver in connection_dict.values(): receiver_conn = self.find_conn(connection_dict, receiver) msg = data.split(' ')[3:] msg = ' '.join(msg) sender_nick = self.find_nick(connection_dict, sender_conn) # conn으로 nick 찾기 receiver_conn.send(f"(귓속말){sender_nick}{time_str}: {msg}".encode()) else: receiver_conn.send(f"입력하신 닉네임은 존재하지 않습니다.".encode()) elif data.split(' ')[1] == '!change_nick': changed_nick = data.split(' ')[2] original_conn = conn original_nick = self.find_nick(connection_dict, original_conn) # conn으로 nick 찾기 connection_dict[original_conn] = changed_nick # conn의 value에 새로운 nick로 갱신 msg = {'changed_nick': changed_nick} original_conn.send(json.dumps(msg).encode()) for sock in connection_dict.keys(): if sock != original_conn: # 닉네임을 바꾼 클라이언트를 제외한 채팅방 멤버에게 메시지 전달 sock.send(f"[INFO] {original_nick}님이 {changed_nick}로 닉네임 변경".encode()) elif data.split(' ')[1] == '!member': member_list = list(connection_dict.values()) conn.send(f'{member_list}'.encode()) else: msg = data for sock in connection_dict: sock.send(msg.encode()) print(f'[MESSAGE] {data}') HOST = '127.0.0.1' PORT = 9111 ADDR = (HOST, PORT) server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # AF_INET = IPv4, SOCK_STREAM = TCP 통신 server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_sock.bind(ADDR) server_sock.listen() print("==============================================") print(f"채팅 서버를 시작합니다. {PORT} 포트로 접속을 기다립니다.") print("==============================================") connectionStatus = {} # 채팅방별 클라이언트 소켓, 닉네임 저장 ex) {1: {conn1: nick1, conn2: nick2, ...}, 2: {conn1: nick1, conn2:nick2, ...}, ...} chat_room = ChatRoom() connection_list = [server_sock] while True: now = datetime.datetime.now() time_str = now.strftime('[%H:%M]') try: read_sockets, write_sockets, error_sockets = select.select(connection_list, [], [], 30) print("클라이언트 요청 대기...") for sock in read_sockets: if sock == server_sock: # 새로운 클라이언트의 소켓이라면 connection_list에 추가 newsock, addr = server_sock.accept() connection_list.append(newsock) else: # 이미 접속한 클라이언트의 소켓이라면 클라이언트가 보낸 메시지 수신 conn = sock data = conn.recv(1024).decode() if 'room_num' not in data: # 이미 접속한 클라이언트의 메시지 수신 try: chat_room.manage_message(data) # 클라이언트의 메시지에 command가 있으면 해당 내용 수행, 없으면 메시지 자체를 broadcast except Exception as e: connection_list.remove(conn) else: # "최초" 접속한 클라이언트의 정보 수신 login_info = json.loads(data) # json 문자열인 data를 -> json.loads(data) -> 파이썬 객체(dict) try: chat_room.manage_client(conn, login_info) except Exception as e: print(e) except Exception as e: print(e) server_sock.close() sys.exit()[client.py]
# -*- coding: utf8 -*- import socket import sys import datetime import select import json HOST = '127.0.0.1' PORT = 9111 ADDR = (HOST, PORT) # socket 객체 생성 connection_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 서버와의 연결을 시도 try: connection_sock.connect(ADDR) except Exception as e: print(e) else: print(f"채팅서버 {HOST}:{PORT}에 연결되었습니다") # client info try: room_num = int(input("\n◽ 입장할 채팅방 번호를 입력하세요: ")) while True: client_nick = input("◽ 사용하실 닉네임을 입력하세요: ") login_info = {'room_num': room_num, 'client_nick': client_nick} connection_sock.send(json.dumps(login_info).encode()) # 클라이언트 정보 송신 is_possible_nick = connection_sock.recv(1024).decode() # 닉네임 중복 여부 수신 if is_possible_nick != 'Y': # 중복 닉네임인 경우 overlappedError 메시지 수신 print(is_possible_nick) continue # 중복닉네임이 아닌 경우 채팅방 입장 print(f"\n 닉네임[{client_nick}] 생성 완료! :-)") break except Exception as e: print(type(e), e) else: s = "" s += "\n ------------< 추가기능 사용하기 >------------" s += "\n 1. 귓속말 보내기" s += "\n : !whisper [상대방 닉네임] [메시지] 입력" s += "\n 2. 닉네임 변경하기" s += "\n : !change_nick [바꿀 닉네임] 입력" s += "\n 3. 참여 중인 멤버 목록 보기" s += "\n : !member 입력" s += "\n -----------------------------------------" s+="\n" print(s) while True: now = datetime.datetime.now() time_str=now.strftime('[%H:%M]') try: # 클라이언트의 IN 동작을 파악할 수 있도록 read_sockets에 sys.stdin도 포함 connection_list = [sys.stdin, connection_sock] read_sockets, write_sockets, error_sockets = select.select(connection_list, [], [], 3) for sock in read_sockets: # 서버에서 받은 메시지인 경우 if sock == connection_sock: data = sock.recv(4096).decode() if 'changed_nick' in data: # 닉네임이 바뀐 경우 {"changed_nick": changed_nick}의 dictionary가 전달됨 changed_info = json.loads(data) client_nick = changed_info['changed_nick'].replace('\n', "") else: print(data) # 클라이언트가 터미널에서 입력한 메시지인 경우 else: message = sys.stdin.readline() # 클라이언트가 입력한 문자열을 읽어서 message = message.replace('\n', '') # message connection_sock.send(f'{client_nick}{time_str}: {message}'.encode()) # 서버에 전송 except Exception as e: print(e) connection_sock.close() sys.exit()'Side Project' 카테고리의 다른 글
점심 메뉴 추천 프로그램 (0) 2023.04.09 I/O Multiplexing이 가능한 채팅 프로그램 - 개요 (0) 2022.07.05