Socket Programming in Python (Guide)

- the low-level socket API in Python’s socket module to create client-server applications.

- created our own custom class for an application-layer protocol to exchange messages and data between endpoints. 

# 소켓이란?

소켓(Socket)이란 네트워크상에서 동작하는 프로그램 간 통신의 종착점(Endpoint)입니다.
즉, 프로그램이 네트워크에서 데이터를 통신할 수 있도록 연결해주는 연결부.
데이터를 통신할 수 있도록 통신할 두 프로그램(Client, Server) 모두에 소켓이 생성되야 합니다.

Endpoint : IP Address와 port 번호의 조합으로, 최종 목적지를 가리킴
최종목적지는 사용자의 디바이스(PC, 스마트폰 등) 또는 Server가 될 수 있습니다.

# 소켓의 종류가 있나?

네트워크에 연결하기 위한 소켓 또한 정해진 규약, 즉, 통신을 위한 프로토콜(Protocol)에 맞게 만들어져야 합니다. 보통 OSI 7 Layer(Open System Interconnection 7 Layer)의 네 번째 계층인 TCP(Transport Control Protocol) 상에서 동작하는 소켓을 주로 사용하는데, 이를 "TCP 소켓" 또는 "TCP/IP 소켓"이라고 부릅니다. (UDP에서 동작하는 소켓은 "UDP 소켓"이라고 합니다.)

The most common type of socket applications are client-server applications, where one side acts as the server and waits for connections from clients. More specifically, we’ll look at the socket API for Internet sockets (Berkeley or BSD sockets). There are also Unix domain sockets (to communicate between processes on the same host).

# 소켓 프로그래밍이란?

소켓(Socket)을 사용하여 네트워크 통신 기능을 구현하는 과정.

소켓(Socket)으로 네트워크 통신 기능을 구현하기 위해서 필요한 지식
1) 소켓을 만들기
2) 만들어진 소켓을 통해 데이터를 주고 받는 절차,
3) 운영체제 및 프로그래밍 언어에 종속적으로 제공되는 소켓 API 사용법

네트워크 환경에서의 다양한 예외처리도 필요함.
1) 케이블 분리로 인한 네트워크 단절
2) 트래픽 증가에 따른 데이터 전송 지연
3) 시스템 리소스 관리 문제로 인한 에러 

# TCP소켓 통신과정 이해하기 with 소켓 API

- 정의 : 소켓의 연결 요청과 수신이 각각 클라이언트 소켓과 서버 소켓의 역할이다.

서버소켓은 어떤 연결 요청(일반적으로 포트 번호로 식별)을 받아들일 것인지를 미리 시스템에 등록하여, 요청이 수신되었을 때 해당 요청을 처리할 수 있도록 준비해야 합니다. 서버소켓은 클라이언트 소켓의 연결 요청을 받아들이는 역할만 수행할 뿐, 직접적인 데이터 송수신은 서버 소켓의 연결 요청 수락의 결과로 만들어지는 새로운 소켓을 통해 처리됩니다. 

 

- TCP 소켓을 사용하는 이유

• Is reliable: packets dropped in the network are detected and retransmitted by the sender.
• Has in-order data delivery: data is read by your application in the order it was written by the sender.

# localhost loopback 실습

TCP 소켓 함수들은 기본적으로 (a blocking call)이다.

즉, 실행결과값을 즉각적으로 가져오지 않고 app실행을 지연한다. 따라서 블록함수다.- accept(), connect(), send(), and recv() “block.”  

Blocking calls have to wait on system calls (I/O) to complete before they can return a value. So the caller are blocked until they’re done or a timeout or other error occurs.

 

socket() 

- address family : AF_INET is the Internet address family for IPv4.

- socket type : SOCK_STREAM is the socket type for TCP  (to transport messages in the network)

 

bind()

- IPv4 : (host, port)

- IPv6 : (host, port, flowinfo, scopeid)

*host : hostname(ip주소 계속 바뀔 수 있음), IP address(추천), or empty string(anything possible)

*post :  1-65535 (0 is reserved,  superuser privileges if < 1024.) 

 

listen() 

: setting the maximum length of the queue for pending connections

*the number of unaccepted connections

 

accept()

: blocks and waits for an incoming connection and return new socket objects

 

conn.recv() 

: returns an empty bytes object, b'',

- The bufsize argument  : 한번 수신할때 받을 수 있는 데이터의 최대 개수

* 1024 는 1024 byte를 반환하는게 아님

* 데이터 없으면, 루프종료, 연결종료

#!/usr/bin/env python3

import socket

HOST = '127.0.0.1'  # Standard loopback interface address (localhost)
PORT = 65432        # Port to listen on (non-privileged ports are > 1023)

with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
    s.bind((HOST, PORT))
    s.listen()
    conn, addr = s.accept()
    with conn:
        print('Connected by', addr)
        while True:
            data = conn.recv(1024)
            if not data:
                break
            conn.sendall(data)

* cloud-host loopback 과의 차이점 : " Ethernet interface "

# Multi connections

서버와 클라이언트 사이의 연결이 2개 이상일 때는 어떻게 해야하나?

• How do we handle multiple connections concurrently? 여러 connection의 데이터전송작업을 동시에 처리할 수 있나?
• We need to call send() and recv() until all data is sent or received. 각 connection마다 전송데이터를 bufsize만큼 잘라서 전달시, 안전무결하게 전달하려면 어떻게 해야 하나?

해결책

4가지 동시처리방법(동기화,스레드,비동기,프로세서) 중 가장 이해하기 쉬운 동시처리 방법을 사용해서 처리하기.

- 접근을 제한(Bound)하는 방법 : IO바운드, CPU 바운드, Cachy, Memory 

1) network application(I/O bound) 작업시 , waiting on 로컬 네트워크, 디스크 또는 상대 네트워크의 엔드포인트 .

2) client의 CPU bound 작업시 , 프로세스 풀을 비동기적 콜한다. concurrent.futures.ProcessPoolExecutor

3) 다중 프로세스 작업시, 운영체제 레벨에서 GIL없이 다중 프로세서(코어)로 병럴처리하도록 schedule해놓을 수 있음.

 

서버

- 여러개 소켓을 listening하고 접근하면, selector에 리스트로 등록하기
- selector에서 events(소켓 & read/write event의 조합)을 가져오기
- 소켓상태모드(accept or listen)에 따라서 작업실행하기
- 작업끝나면  selector종료, 소켓도 닫힘.

import sys
import socket
import selectors
import types

sel = selectors.DefaultSelector()

if len(sys.argv) != 3:
    print("usage:", sys.argv[0], "<host> <port>")
    sys.exit(1)

host, port = sys.argv[1], int(sys.argv[2])
lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
lsock.bind((host, port))
lsock.listen()
print("listening on", (host, port))

# 소켓 블록해제
lsock.setblocking(False) 
# sel.select와 register로 여러 소켓(이벤트) 관리가능
# listening 소켓, read event : EVENT_READ
sel.register(lsock, selectors.EVENT_READ, data=None)

try:
    while True:
    
    	# select수행시, 소켓통제 내부블록생성 - IO Bound
    	# select가 이벤트 반환하면 data, fileobj, mask있음	
        events = sel.select(timeout=None)
        # event의 구성요소
        # 1) a SelectorKey namedtuple : fileobj = socket object 
        # 2) an event mask : 이벤트가 ready되었는지 여부
        for key, mask in events:
            if key.data is None:
                accept_wrapper(key.fileobj)
            else:
                service_connection(key, mask)
except KeyboardInterrupt:
    print("caught keyboard interrupt, exiting")
finally:
    sel.close()

def accept_wrapper(sock):

    # Should be ready to read
    conn, addr = sock.accept()  
    print("accepted connection from", addr)
    
    # 다른 소켓도 연결하도록 블록해제
    conn.setblocking(False) 
    
    # create an object to hold the data
    data = types.SimpleNamespace(addr=addr, inb=b"", outb=b"")
    
    # connection의 읽기/쓰기모드 설정해놓기 
    events = selectors.EVENT_READ | selectors.EVENT_WRITE
    sel.register(conn, events, data=data)


def service_connection(key, mask):
    sock = key.fileobj
    data = key.data
    # 데이터 주고 받을 때는 블록해제 안함. 데이터 안정성위해서.
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)  # Should be ready to read
        if recv_data:
            data.outb += recv_data
        else:
            print("closing connection to", data.addr)
            sel.unregister(sock)
            sock.close()
    if mask & selectors.EVENT_WRITE:
        if data.outb:
            print("echoing", repr(data.outb), "to", data.addr)
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]

클라이언트

서버의 핵심업무는 리스닝(accept)!  클라이언트는 요청해야 하므로 초기연결 생성 필요.

# num_conns : the number of connections to create to the server
def start_connections(host, port, num_conns):
    server_addr = (host, port)
    for i in range(0, num_conns):
        connid = i + 1
        print("starting connection", connid, "to", server_addr)
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.setblocking(False)
        
        # while the connection is in progress
        # connect_ex() : initially returns an error indicator, errno.EINPROGRESS, 
        # connect() raising an exception 
        
        sock.connect_ex(server_addr)
        events = selectors.EVENT_READ | selectors.EVENT_WRITE
        data = types.SimpleNamespace(
            connid=connid,
            msg_total=sum(len(m) for m in messages),
            recv_total=0,
            messages=list(messages),
            outb=b"",
        )
        
        # 드디어 소켓에 데이터와 이벤트정보를 등록한다.
        sel.register(sock, events, data=data)
        
        
        
if len(sys.argv) != 4:
    print("usage:", sys.argv[0], "<host> <port> <num_connections>")
    sys.exit(1)

host, port, num_conns = sys.argv[1:4]
messages = [b'1',b'2']
start_connections(host, int(port), int(num_conns))

try:
    while True:
        events = sel.select(timeout=1)
        if events:
            for key, mask in events:
                service_connection(key, mask)
        # Check for a socket being monitored to continue.
        if not sel.get_map():
            break
except KeyboardInterrupt:
    print("caught keyboard interrupt, exiting")
finally:
    sel.close()

 

def service_connection(key, mask):
    sock = key.fileobj
    data = key.data
    if mask & selectors.EVENT_READ:
        recv_data = sock.recv(1024)  # Should be ready to read
        
        # 수신 데이터가 있다면
        if recv_data: 
            print('received', repr(recv_data), 'from connection', data.connid)
            data.recv_total += len(recv_data)            
        # 수신데이터가 없고, 총 수신데이터 byte길이 = 총 송신데이터 byte길이
        if not recv_data or data.recv_total == data.msg_total:
            print('closing connection', data.connid)
            sel.unregister(sock)
            sock.close()
            
    if mask & selectors.EVENT_WRITE:
    
        # 작성데이터가 없고, 남은 메세지가 있다면
        if not data.outb and data.messages:
            data.outb = data.messages.pop(0)
        # 작성한 데이터가 있다면            
        if data.outb:
            print('sending', repr(data.outb), 'to connection', data.connid)
            sent = sock.send(data.outb)  # Should be ready to write
            data.outb = data.outb[sent:]

 

서버는 클라이언트가 메세지전송시, 클라이언트단의 연결을 종료해주길 바람. 그렇지 않으면 서버는 수신할 데이터가 있다고 생각해서 연결대기상태가 됨. 이런 식으로 커넥션이 누적되면 자원낭비되므로, 특정시간을 초과하면 자동으로 커넥션을 종료시키도록 서버단에서 관리해줘야 함.

# connection - ip address , port
$ ./multiconn-server.py 127.0.0.1 65432
# connection - ip address , port, num of connection
$ ./multiconn-client.py 127.0.0.1 65432 2

# 애플리케이션 만들기 실습

from Python 3.3, errors related to socket or address semantics raise OSError or one of its subclasse.

전송하는 메세지를 효율적으로 처리하기 위해서 헤더가 필요하다. 또한 전송하는 메세지는 바이트가 아니라 유니코드를 사용한다. 

The client or server on the other end could have a CPU that uses a different byte order ( CPU’s endianness )than your own. 노트북이랑 코랩이랑 데스크탑이랑 사용하는 바이트순서가 다르다. 그래서 호스트의 native byte order에 맞게 변형해서 사용해야 한다. 그래서 메세지 헤더로  Unicode  (the encoding UTF-8)을 사용한다. UTF-8은 8-bit encoding방식으로 byte ordering 문제가 없다. 이를 통해 모든 포멧(text or binary)의 데이터를 전송할수 있다.

1. 헤더란 무엇인가?

헤더는 메세지의 길이, 사용된 바이트수 등 메세지의 전송데이터에 대한 정보를 담고 있다. 이 정보에 맞게 데이터를 디코딩 및 번역할 수 있다. *socket sending/receiving byte data 

 

2. 어플리케이션 프로토콜 헤더는 무엇인가(구성요소)?

• Variable-length text
• Unicode with the encoding UTF-8
• A Python dictionary serialized using JSON - 헤더는 딕셔너리 형식이라 key/value 짝으로 더 많은 정보를 담을 수 있음.
  • 필수정보만 보자.(아래)

byteorder	The byte order of the machine (uses sys.byteorder). This may not be required for your application.
content-length	The length of the content in bytes.
content-type	The type of content in the payload, for example, text/json or binary/my-binary-type.
content-encoding	The encoding used by the content, for example, utf-8 for Unicode text or binary for binary data.

3. 어플리케이션 메세지 클래스 

- 헤어와 관련된 모든 메세지 설정까지 클래스로 만들어보는 실습

• 클라이언트 메세지 클래스 -> 클라리언트 통신

# client
if len(sys.argv) != 5:
    print("usage:", sys.argv[0], "<host> <port> <action> <value>")
    sys.exit(1)

host, port = sys.argv[1], int(sys.argv[2])
action, value = sys.argv[3], sys.argv[4]
request = create_request(action, value)
start_connection(host, port, request)

try:
    while True:
        events = sel.select(timeout=1)
        for key, mask in events:
            message = key.data
            try:
                message.process_events(mask)
            except Exception:
                print(
                    "main: error: exception for",
                    f"{message.addr}:\n{traceback.format_exc()}",
                )
                message.close()
        # Check for a socket being monitored to continue.
        if not sel.get_map():
            break
except KeyboardInterrupt:
    print("caught keyboard interrupt, exiting")
finally:
    sel.close()

• 서버 메세지 클래스 -> 서버 통신

# server
host, port = sys.argv[1], int(sys.argv[2])
lsock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# Avoid bind() exception: OSError: [Errno 48] Address already in use
lsock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
lsock.bind((host, port))
lsock.listen()
print("listening on", (host, port))
lsock.setblocking(False)
sel.register(lsock, selectors.EVENT_READ, data=None)

try:
    while True:
        events = sel.select(timeout=None)
        for key, mask in events:
            if key.data is None:
                accept_wrapper(key.fileobj)
            else:
                message = key.data
                try:
                    message.process_events(mask)
                except Exception:
                    print(
                        "main: error: exception for",
                        f"{message.addr}:\n{traceback.format_exc()}",
                    )
                    message.close()
except KeyboardInterrupt:
    print("caught keyboard interrupt, exiting")
finally:
    sel.close()

__init__ : 소켓,주소,버퍼,헤더, 리퀘스트, 리스펀스,selector
_set_selector_events_mask : 이벤트모드설정(읽기,쓰기)
_read : 소켓 읽기기능(버퍼사용) : data -> 수신버퍼
_write : 소켓 쓰기기능(버퍼사용) : 송신버퍼(msg) -> 송신
_json_encode : 쓰기
_json_decode : 읽기
_create_message : 매개변수 -> 헤더생성 -> 인코딩 -> 메세지전송[패킷생성 + 인코딩 + 타입] 
_create_reponse_json_content : self.request에서 데이터 받기 & response 생성 후 반환
_create_response_binary_content : self.request 로 response 생성후 반환
create_response : 콘텐트 유형에 맞게 reponse 생성 -> 메세지 생성 -> 송신버퍼
process_events : mask 와 이벤트 모드에 맞는 이벤트 실행(읽기/쓰기)
process_protoheader : 고정길이부분 수신버퍼 & 패킷 -> 언팩킷 -> 수신버퍼 채우기
process_jsonheader : json부분 수신버퍼 -> json 디코딩 -> 수신버퍼 채우기
process_request :  나머지 수신버퍼 채우기,  데이터추출 -> json 인코딩 -> request채우기
queue_request : self.request에서  content정보획득 -> 인코딩 -> 메세지 생성 -> 송신버퍼
read : _read -> 헤더가 없는 경우,  process_* 실행
write :
1) 클라이언트 : queue_request -> _write실행
2) 서버 : create_reponse -> _write 실행
close : 소켓 등록취소 -> 소켓닫기

## 실행하기

./app-server.py '' 65432
./app-client.py 10.0.1.1 65432 search morpheus

 

# 출처

1) recipes4dev.tistory.com/153

소켓 프로그래밍. (Socket Programming)

2) realpython.com/python-sockets/

Socket Programming in Python (Guide) – Real Python

3) docs.python.org/3/howto/sockets.html#socket-howto

Socket Programming HOWTO — Python 3.9.1 documentation