redisgate Gent helio helio_raw_echo_server

Helio raw_echo_server

---

소개

raw_echo_server는 입력받은 내용을 그대로 내보낸다.
즉, "Hello"를 받으면 "Hello"를 그대로 리턴한다.

목차

• 실행 옵션
• 테스트 프로그램
• 서버 로그
• 소스 설명

실행 옵션

실행 옵션 보기: build-dbg$ raw_echo_server --helpfull

• Flags from helio/examples/raw_echo_server.cc:
  • --port (Echo server port); default: 8081;
  • --size (Message size); default: 512;
  • --threads (Number of connection threads. 0 to run as single threaded. Otherwise, the main thread accepts connections and worker threads are handling TCP connections); default: 0;
• Flags from helio/build-dbg/_deps/abseil_cpp-src/absl/flags/parse.cc:
  • --flagfile (comma-separated list of files to load flags from); default: ;
  • --fromenv (comma-separated list of flags to set from the environment [use 'export FLAGS_flag1=value']); default: ;
  • --tryfromenv (comma-separated list of flags to try to set from the environment if present); default: ;
  • --undefok (comma-separated list of flag names that it is okay to specify on the command line even if the program does not define a flag with that name); default: ;
• Flags from helio/build-dbg/_deps/glog-src/src/logging.cc:
  • --alsologtoemail (log messages go to these email addresses in addition to logfiles); default: "";
  • --alsologtostderr (log messages go to stderr in addition to logfiles); default: false;
  • --colorlogtostderr (color messages logged to stderr (if supported by terminal)); default: false;
  • --colorlogtostdout (color messages logged to stdout (if supported by terminal)); default: false;
  • --drop_log_memory (Drop in-memory buffers of log contents. Logs can grow very quickly and they are rarely read before they need to be evicted from memory. Instead, drop them from memory as soon as they are flushed to disk.); default: true;
  • --log_backtrace_at (Emit a backtrace when logging at file:linenum.); default: "";
  • --log_dir (If specified, logfiles are written into this directory instead of the default logging directory.); default: "";
  • --log_link (Put additional links to the log files in this directory); default: "";
  • --log_prefix (Prepend the log prefix to the start of each log line); default: true;
  • --log_utc_time (Use UTC time for logging.); default: false;
  • --log_year_in_prefix (Include the year in the log prefix); default: true;
  • --logbuflevel (Buffer log messages logged at this level or lower (-1 means don't buffer; 0 means buffer INFO only; ...)); default: 0;
  • --logbufsecs (Buffer log messages for at most this many seconds); default: 30;
  • --logcleansecs (Clean overdue logs every this many seconds); default: 300;
  • --logemaillevel (Email log messages logged at this level or higher (0 means email all; 3 means email FATAL only; ...)); default: 999;
  • --logfile_mode (Log file mode/permissions.); default: 436;
  • --logmailer (Mailer used to send logging email); default: "";
  • --logtostderr (log messages go to stderr instead of logfiles); default: false;
  • --logtostdout (log messages go to stdout instead of logfiles); default: false;
  • --max_log_size (approx. maximum log file size (in MB). A value of 0 will be silently overridden to 1.); default: 1800;
  • --minloglevel (Messages logged at a lower level than this don't actually get logged anywhere); default: 0;
  • --stderrthreshold (log messages at or above this level are copied to stderr in addition to logfiles. This flag obsoletes --alsologtostderr.); default: 2;
  • --stop_logging_if_full_disk (Stop attempting to log to disk if the disk is full.); default: false;
  • --timestamp_in_logfile_name (put a timestamp at the end of the log file name); default: true;
• Flags from helio/build-dbg/_deps/glog-src/src/utilities.cc:
  • --symbolize_stacktrace (Symbolize the stack trace in the tombstone); default: true;
• Flags from helio/build-dbg/_deps/glog-src/src/vlog_is_on.cc:
  • --v (Show all VLOG(m) messages for m <= this. Overridable by --vmodule.); default: 0;
  • --vmodule (per-module verbose level. Argument is a comma-separated list of =. is a glob pattern, matched against the filename base (that is, name ignoring .cc/.h./-inl.h). overrides any value given by --v.); default: "";

테스트 프로그램

테스트 프로그램은 python으로 작성했다.

• t_raw_echo1.py: 8081 포트에 접속해서 'Hello, raw_echo_server!' 메시지를 보내고 그대로 받는다. import socket def test_raw_echo_server(host, port): # 서버에 연결 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect((host, port)) try: # 서버에 메시지 전송 message = 'Hello, raw_echo_server!' print(f'Sending: {message}') client_socket.sendall(message.encode()) # 서버로부터 에코 메시지 수신 response = client_socket.recv(1024) print(f'Received: {response.decode()}') finally: # 연결 종료 client_socket.close() if __name__ == "__main__": test_raw_echo_server('127.0.0.1', 8081) # 서버가 실행 중인 호스트와 포트
• t_raw_echo2.py: 메시지를 입력할 수 있다. 메시지(문자열) 길이는 23바이트이다. 23바이트 이상 입력하면 다음 응답 시 받는다. import socket def test_raw_echo_server(host, port): # 서버에 연결 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect((host, port)) try: while True: # 사용자로부터 메시지 입력 받기 message = input('Enter message to send (type "exit" to quit): ') if message.lower() == 'exit': print('Exiting...') break # 서버에 메시지 전송 print(f'Sending: {message}') client_socket.sendall(message.encode()) # 서버로부터 에코 메시지 수신 response = client_socket.recv(1024) print(f'Received: {response.decode()}') finally: # 연결 종료 client_socket.close() if __name__ == "__main__": test_raw_echo_server('127.0.0.1', 8081) # 서버가 실행 중인 호스트와 포트
• t_raw_echo3.py: 횟수(count)를 입력하면 메시지를 반복해서 보낸다. import socket def test_raw_echo_server(host, port, message, count): # 서버에 연결 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect((host, port)) try: for i in range(count): # 서버에 메시지 전송 print(f'Sending : {message} (Message {i+1}/{count})') client_socket.sendall(message.encode()) # 서버로부터 에코 메시지 수신 response = client_socket.recv(1024) print(f'Received: {response.decode()}') finally: # 연결 종료 client_socket.close() if __name__ == "__main__": # 사용자로부터 반복 횟수 입력 받기 count = int(input('Enter the number of times to send the message: ')) message = 'Hello, raw_echo_server!' test_raw_echo_server('127.0.0.1', 8081, message, count) # 서버가 실행 중인 호스트와 포트
• t_raw_echo4.py: 횟수(count)를 반복해서 입력할 수 있도록 수정했다. 종료는 'exit'. import socket def send_fixed_messages(client_socket, message, count): for i in range(count): # 서버에 메시지 전송 print(f'Sending: {message} (Message {i+1}/{count})') client_socket.sendall(message.encode()) # 서버로부터 에코 메시지 수신 response = client_socket.recv(1024) print(f'Received: {response.decode()}') def main(): host = '127.0.0.1' port = 8081 message = 'Hello, raw_echo_server!' # 서버에 연결 client_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_socket.connect((host, port)) try: while True: # 사용자로부터 반복 횟수 입력 받기 count = input('Enter the number of times to send the message (or type "exit" to quit): ') if count.lower() == 'exit': print('Exiting...') break if count.isdigit(): count = int(count) send_fixed_messages(client_socket, message, count) else: print('Please enter a valid number.') finally: # 연결 종료 client_socket.close() if __name__ == "__main__": main()
• t_raw_echo5.py: 스레드 개수를 입력받아 여러 개 스레드로 메시지를 보내도록 수정했다. import socket import threading def connect_to_server(host, port, message): try: with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s: s.connect((host, port)) s.sendall(message.encode()) response = s.recv(1024) print(f"Received: {response.decode()}") except Exception as e: print(f"Connection failed: {e}") def main(): host = '127.0.0.1' # 서버 주소 port = 8081 # 서버 포트 num_connections = int(input("Enter number of connections: ")) message = "Hello, raw_echo_server!" threads = [] for _ in range(num_connections): thread = threading.Thread(target=connect_to_server, args=(host, port, message)) threads.append(thread) thread.start() for thread in threads: thread.join() if __name__ == "__main__": main()

서버 로그

서버 실행: [build-dbg]$ raw_echo_server --logtostderr --v 1 --threads 2 --size 23

서버 실행 직후 서버 로그 2669 init.cc:112] raw_echo_server running in debug mode. 2669 raw_echo_server.cc:801] main() Start 2669 raw_echo_server.cc:807] main() new MyPolicy(-1) 2669 raw_echo_server.cc:698] MyPolicy class constructor. call SetupIORing(&ring_) 2669 raw_echo_server.cc:610] SetupIORing() Start 2669 raw_echo_server.cc:809] main() fb2::SetCustomDispatcher(policy) 2669 fiber_interface.cc:117] TL_FiberInitializer TSC Frequency : 2496000/ms 2669 raw_echo_server.cc:815] main() Num threads 2 2669 raw_echo_server.cc:646] SetupListener() socket(port:8081) listen_fd:4 2669 raw_echo_server.cc:671] SetupListener() bind(4) res:0 2669 raw_echo_server.cc:822] main() SetupListener(8081), listen_fd:4 2669 raw_echo_server.cc:827] main() worker0 2669 pthread_utils.cc:97] Starting thread worker0 2670 raw_echo_server.cc:743] WorkerThread(0) 2670 raw_echo_server.cc:698] MyPolicy class constructor. call SetupIORing(&ring_) 2670 raw_echo_server.cc:610] SetupIORing() Start 2669 raw_echo_server.cc:827] main() worker1 2670 raw_echo_server.cc:711] MyPolicy.Run() call RunEventLoop() index_:0 2669 pthread_utils.cc:97] Starting thread worker1 2670 raw_echo_server.cc:515] RunEventLoop() Start. worker:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:423] AcceptFiber(listen_fd:4) Start 2670 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2670 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:1 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:2 2670 raw_echo_server.cc:586] RunEventLoop() worker:0, io_uring_wait_cqes() waiting... 2671 raw_echo_server.cc:743] WorkerThread(1) 2671 raw_echo_server.cc:698] MyPolicy class constructor. call SetupIORing(&ring_) 2671 raw_echo_server.cc:610] SetupIORing() Start 2671 raw_echo_server.cc:711] MyPolicy.Run() call RunEventLoop() index_:1 2671 raw_echo_server.cc:515] RunEventLoop() Start. worker:1 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:423] AcceptFiber(listen_fd:4) Start 2671 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2671 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:586] RunEventLoop() worker:1, io_uring_wait_cqes() waiting...

테스트 프로그램 t_raw_echo1.py 실행
[examples]$ python t_raw_echo1.py
Sending: Hello, raw_echo_server!
Received: Hello, raw_echo_server!

서버 로그 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:432] AcceptFiber() accept4():7, errno:11 2671 raw_echo_server.cc:437] AcceptFiber() workers current_worker:1 2670 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2671 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2671 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2670 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2671 raw_echo_server.cc:535] RunEventLoop() call HandleSocket() 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:1 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:2 2670 raw_echo_server.cc:586] RunEventLoop() worker:0, io_uring_wait_cqes() waiting... 2671 raw_echo_server.cc:378] HandleSocket() Start fd:7 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:23 2671 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2671 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2671 raw_echo_server.cc:347] Send() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:353] Send() cqe_result.res:23 2671 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:586] RunEventLoop() worker:1, io_uring_wait_cqes() waiting... 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:0 2671 raw_echo_server.cc:313] Recv() cqe_result.res:0 return 'ECONNABORTED' 2671 raw_echo_server.cc:388] HandleSocket() Recv res:103 Fail 2671 raw_echo_server.cc:391] HandleSocket() Recv Error break 2671 raw_echo_server.cc:410] HandleSocket() close(fd:7) 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:586] RunEventLoop() worker:1, io_uring_wait_cqes() waiting...

테스트 프로그램 t_raw_echo2.py 실행
[examples]$ python t_raw_echo2.py
Enter message to send (type "exit" to quit): Hello, raw_echo_server!AAAA -> 'AAAA'를 추가로 입력
Sending: Hello, raw_echo_server!AAAA -> 27바이트 입력
Received: Hello, raw_echo_server! -> 23바이트만 도착, 'AAAA'는 다음에 옵니다.

서버 로그 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:432] AcceptFiber() accept4():7, errno:11 2670 raw_echo_server.cc:437] AcceptFiber() workers current_worker:0 2671 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2670 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2671 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2670 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2670 raw_echo_server.cc:535] RunEventLoop() call HandleSocket() 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:586] RunEventLoop() worker:1, io_uring_wait_cqes() waiting... 2670 raw_echo_server.cc:378] HandleSocket() Start fd:7 2670 raw_echo_server.cc:294] Recv() len:23 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:1 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:2 2670 raw_echo_server.cc:586] RunEventLoop() worker:0, io_uring_wait_cqes() waiting... 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:301] Recv() cqe_result.res:23 2670 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2670 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2670 raw_echo_server.cc:347] Send() len:23 -> "Hello, raw_echo_server!" 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:353] Send() cqe_result.res:23 2670 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2670 raw_echo_server.cc:294] Recv() len:23 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:301] Recv() cqe_result.res:4 2670 raw_echo_server.cc:294] Recv() len:19 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:1 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:2 2670 raw_echo_server.cc:586] RunEventLoop() worker:0, io_uring_wait_cqes() waiting...

t_raw_echo2.py 계속
Enter message to send (type "exit" to quit): Hello, raw_echo_server!BBBBB -> 'BBBBB'를 추가로 입력
Sending: Hello, raw_echo_server!BBBBB
Received: AAAAHello, raw_echo_ser -> 이전에 입력한 'AAAA'가 앞에 붙어서 23바이트 도착

서버 로그 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:301] Recv() cqe_result.res:19 2670 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2670 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2670 raw_echo_server.cc:347] Send() len:23 -> "AAAAHello, raw_echo_ser" 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:353] Send() cqe_result.res:23 2670 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2670 raw_echo_server.cc:294] Recv() len:23 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:301] Recv() cqe_result.res:9 2670 raw_echo_server.cc:294] Recv() len:14 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:1 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:2 2670 raw_echo_server.cc:586] RunEventLoop() worker:0, io_uring_wait_cqes() waiting...

t_raw_echo2.py 종료
Enter message to send (type "exit" to quit): exit -> 테스트 프로그램 종료
Exiting...

서버 로그 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:301] Recv() cqe_result.res:0 2670 raw_echo_server.cc:313] Recv() cqe_result.res:0 return 'ECONNABORTED' 2670 raw_echo_server.cc:388] HandleSocket() Recv res:103 Fail 2670 raw_echo_server.cc:391] HandleSocket() Recv Error break 2670 raw_echo_server.cc:410] HandleSocket() close(fd:7) 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:1 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:2 2670 raw_echo_server.cc:586] RunEventLoop() worker:0, io_uring_wait_cqes() waiting...

테스트 프로그램 t_raw_echo3.py 실행
[examples]$ python t_raw_echo3.py
Enter the number of times to send the message: 5 -> 횟수(count) 입력
Sending : Hello, raw_echo_server! (Message 1/5)
Received: Hello, raw_echo_server!
Sending : Hello, raw_echo_server! (Message 2/5)
Received: Hello, raw_echo_server!
Sending : Hello, raw_echo_server! (Message 3/5)
Received: Hello, raw_echo_server!
Sending : Hello, raw_echo_server! (Message 4/5)
Received: Hello, raw_echo_server!
Sending : Hello, raw_echo_server! (Message 5/5)
Received: Hello, raw_echo_server!

서버 로그 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:432] AcceptFiber() accept4():7, errno:11 2670 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2671 raw_echo_server.cc:437] AcceptFiber() workers current_worker:1 2670 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2671 raw_echo_server.cc:432] AcceptFiber() accept4():-1, errno:11 2671 raw_echo_server.cc:449] AcceptFiber() Error:EAGAIN 'Resource temporarily unavailable' 2670 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:1 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:535] RunEventLoop() call HandleSocket() 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:2 2670 raw_echo_server.cc:586] RunEventLoop() worker:0, io_uring_wait_cqes() waiting... 2671 raw_echo_server.cc:378] HandleSocket() Start fd:7 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:23 -> 1회 입력 2671 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2671 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2671 raw_echo_server.cc:347] Send() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:353] Send() cqe_result.res:23 2671 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:23 -> 2회 입력 2671 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2671 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2671 raw_echo_server.cc:347] Send() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:353] Send() cqe_result.res:23 2671 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:586] RunEventLoop() worker:1, io_uring_wait_cqes() waiting... 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:23 -> 3회 입력 2671 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2671 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2671 raw_echo_server.cc:347] Send() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:353] Send() cqe_result.res:23 2671 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:23 -> 4회 입력 2671 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2671 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2671 raw_echo_server.cc:347] Send() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:353] Send() cqe_result.res:23 2671 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:23 -> 5회 입력 2671 raw_echo_server.cc:388] HandleSocket() Recv res:0 Success 2671 raw_echo_server.cc:397] HandleSocket() Send to socket fd:7 2671 raw_echo_server.cc:347] Send() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:353] Send() cqe_result.res:23 2671 raw_echo_server.cc:401] HandleSocket() Send res:0 Success 2671 raw_echo_server.cc:294] Recv() len:23 2671 raw_echo_server.cc:264] SuspendMyself() Before Suspend 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:301] Recv() cqe_result.res:0 2671 raw_echo_server.cc:313] Recv() cqe_result.res:0 return 'ECONNABORTED' 2671 raw_echo_server.cc:388] HandleSocket() Recv res:103 Fail 2671 raw_echo_server.cc:391] HandleSocket() Recv Error break 2671 raw_echo_server.cc:410] HandleSocket() close(fd:7) 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:1 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:2 2671 raw_echo_server.cc:586] RunEventLoop() worker:1, io_uring_wait_cqes() waiting...

Ctrl+C(^C) 서버 종료

서버 로그 2669 raw_echo_server.cc:804] Exiting on signal Interrupt 2670 raw_echo_server.cc:528] RunEventLoop() work:0, spins:0 2671 raw_echo_server.cc:528] RunEventLoop() work:1, spins:0 2670 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:266] SuspendMyself() After Suspend 2671 raw_echo_server.cc:467] aborted 8 2671 raw_echo_server.cc:468] AcceptFiber() aborted 8 2670 raw_echo_server.cc:467] aborted 8 2670 raw_echo_server.cc:468] AcceptFiber() aborted 8 2670 raw_echo_server.cc:595] Exit RunEventLoop() total/sumitted/stalled 34/11/7 2671 raw_echo_server.cc:595] Exit RunEventLoop() total/sumitted/stalled 39/18/6 2670 raw_echo_server.cc:598] RunEventLoop() CQE histogram Count:11 Average:1.0 StdDev:0.0 Min: 1.0000 Median: 1.0000 Max: 1.0000 ------------------------------------------------------ [ 1, 2 ) 11 100.000% 100.000% #################### 2671 raw_echo_server.cc:598] RunEventLoop() CQE histogram Count:18 Average:1.0 StdDev:0.0 Min: 1.0000 Median: 1.0000 Max: 1.0000 ------------------------------------------------------ [ 1, 2 ) 18 100.000% 100.000% #################### 2670 raw_echo_server.cc:703] ~MyPolicy 2671 raw_echo_server.cc:703] ~MyPolicy 2669 raw_echo_server.cc:854] Send histogram Count: 8 Average: 566.3750 StdDev: 132.69 Min: 413.0000 Median: 533.3333 Max: 830.0000 ------------------------------------------------------ [ 400, 450 ) 2 25.000% 25.000% ##### [ 450, 500 ) 1 12.500% 37.500% ### [ 500, 600 ) 2 25.000% 62.500% ##### [ 600, 700 ) 2 25.000% 87.500% ##### [ 800, 900 ) 1 12.500% 100.000% ### WARNING: Logging before InitGoogleLogging() is written to STDERR 2669 raw_echo_server.cc:711] MyPolicy.Run() call RunEventLoop() index_:-1 2669 raw_echo_server.cc:515] RunEventLoop() Start. worker:-1 2669 raw_echo_server.cc:528] RunEventLoop() work:-1, spins:0 2669 raw_echo_server.cc:595] Exit RunEventLoop() total/sumitted/stalled 1/0/0 2669 raw_echo_server.cc:598] RunEventLoop() CQE histogram Count: 0 Average:0.0 StdDev:0.0 Min: 0.0000 Median: 0.0000 Max: 0.0000 ------------------------------------------------------ 2669 raw_echo_server.cc:703] ~MyPolicy

소스 설명

raw_echo_server.cc 소스 설명입니다.

◼️ include, FLAG(option)

// Copyright 2022, Roman Gershman. All rights reserved. // See LICENSE for licensing terms. // helio/examples/raw_echo_server.cc - 501줄 // 2024.07.24(Wed) Charlie // https://chatgpt.com/share/2a340cc7-a275-402e-8a5a-eb763309daa8 // 예) string nm = absl::StrCat("worker", i); main() #include <absl/strings/str_cat.h> // https://chatgpt.com/share/f4fe6ac4-5632-4a72-84e1-cbde9fe2619f // C++ 프로그램에서 CPU 프로파일링을 수행하기 위해 사용되는 헤더 파일입니다. // 이를 통해 개발자는 코드의 어느 부분이 CPU 시간을 많이 소모하는지 파악할 수 있습니다. // 예) ProfilerEnable() #include <gperftools/profiler.h> #include <liburing.h> // https://chatgpt.com/share/2c0f49f0-5c72-4562-8022-9a4be724a066 // UNIX 계열 운영체제(예: Linux, macOS)에서 네트워크 소켓 프로그래밍을 할 때 필수적으로 사용됩니다. // 이 헤더 파일은 인터넷 주소 패밀리와 관련된 다양한 구조체와 상수, 함수 프로토타입을 정의합니다. // struct sockaddr_in, struct in_addr, socket(), bind(), connect(), listen(), etc #include <netinet/in.h> // https://chatgpt.com/share/c0e6c452-5a4a-4c9d-b470-286a30dd2b91 // poll(), POLLIN, POLLOUT, POLLERR, POLLHUP, POLLNVAL #include <poll.h> // https://chatgpt.com/share/9b458eff-8d3a-417a-bd5e-f98939a6423c // std::queue, push(), pop(), front(), back(), empty(), size(), etc #include <queue> // https://chatgpt.com/share/8c88cc6e-20dd-4e61-8754-374df1b29e45 // std::thread, join(), detach() #include <thread> #include "base/function2.hpp" #include "base/histogram.h" #include "base/init.h" #include "base/logging.h" #include "base/mpmc_bounded_queue.h" #include "base/pthread_utils.h" #include "util/fibers/fibers.h" #include "util/fibers/detail/utils.h" ABSL_FLAG(int16_t, port, 8081, "Echo server port"); ABSL_FLAG(uint32_t, size, 512, "Message size"); // 연결 스레드 수. 단일 스레드로 실행하려면 0을 지정합니다. // 그렇지 않으면 메인 스레드가 연결을 수락하고 작업자 스레드가 TCP 연결을 처리합니다. ABSL_FLAG(uint16_t, threads, 0, "Number of connection threads. 0 to run as single threaded. " "Otherwise, the main thread accepts connections and worker threads are handling TCP " "connections");

include 파일 설명(chatgpt)

• absl/strings/str_cat.h   -> string nm = absl::StrCat("worker", i); main()
• gperftools/profiler.h   -> ProfilerEnable()
  CPU 프로파일링을 수행하기 위해 사용되는 헤더 파일입니다.
  이를 통해 개발자는 코드의 어느 부분이 CPU 시간을 많이 소모하는지 파악할 수 있습니다.
• netinet/in.h   -> struct sockaddr_in, struct in_addr, socket(), bind(), connect(), listen(), etc
  UNIX 계열 운영체제(예: Linux, macOS)에서 네트워크 소켓 프로그래밍을 할 때 필수적으로 사용됩니다.
  이 헤더 파일은 인터넷 주소 패밀리와 관련된 다양한 구조체와 상수, 함수 프로토타입을 정의합니다.
  
• poll.h   -> poll(), POLLIN, POLLOUT, POLLERR, POLLHUP, POLLNVAL
• queue   -> std::queue, push(), pop(), front(), back(), empty(), size(), etc
• thread   -> std::thread, join(), detach()
  스레드는 --threads 옵션을 받아 main()에서 base::StartThread() 함수를 사용해서 직접 생성합니다.
  ping_ioring_server에서 사용했던 --proactor_threads(proactor_pool.cc)를 사용하지 않습니다.

◼️ 시그널(signal) 처리 관련 함수들

namespace ctx = boost::context; using namespace std; using namespace util; using util::fb2::Fiber; using util::fb2::detail::FiberInterface; namespace { // util/fibers/proactor_base.h 에도 거의 비슷한 signal_state가 있다. struct signal_state { struct Item { // 신호가 발생했을 때 호출되는 콜백 함수입니다. // 이 함수는 신호 번호를 인자로 받습니다. std::function<void(int)> cb; }; // _NSIG(65:0~64)' 크기의 배열로, 각 신호에 대해 하나의 'Item'을 보유합니다. Item signal_map[_NSIG]; }; // 이 함수는 'signal_state' 구조체의 정적 인스턴스를 반환합니다. // 'static' 키워드로 선언된 'signal_state state'는 프로그램 실행 동안 한 번만 초기화되고, // 이후에는 동일한 인스턴스를 반환합니다. 이는 신호 처리 상태를 전역적으로 유지하기 위함입니다. signal_state* get_signal_state() { static signal_state state; return &state; } // 이 함수는 실제 신호 처리기를 구현합니다. 특정 신호가 발생했을 때 이 함수가 호출됩니다. // 'int signal': 발생한 신호의 번호입니다. // 'siginfo_t*': 신호에 대한 추가 정보를 담고 있는 포인터입니다. 여기서는 사용되지 않습니다. // 'void*': 추가적인 컨텍스트 정보를 담고 있는 포인터입니다. 여기서는 사용되지 않습니다. void SigAction(int signal, siginfo_t*, void*) { // 1) 'get_signal_state'를 호출하여 'signal_state' 인스턴스를 가져옵니다. signal_state* state = get_signal_state(); // 2) 'signal' 값이 '_NSIG'보다 작은지 확인합니다. DCHECK_LT(signal, _NSIG); // 3) 현재 신호에 해당하는 'Item'을 가져옵니다. auto& item = state->signal_map[signal]; // item.cb가 있으면 실행한다. 없으면 에러 로그. if (item.cb) { item.cb(signal); } else { LOG(ERROR) << "Tangling signal handler " << signal; } } // 이 함수는 시그널에 대한 콜백 함수를 등록하거나 등록된 콜백을 해제합니다. // 'l': 시그널 번호들의 리스트. // 'cb': 시그널이 발생했을 때 호출될 콜백 함수. 콜백 함수가 nullptr이면 시그널 등록을 해제합니다. void RegisterSignal(std::initializer_list<uint16_t> l, std::function<void(int)> cb) { auto* state = get_signal_state(); // 'sigaction' 구조체로 시그널 핸들링 정보를 담고 있습니다. struct sigaction sa; memset(&sa, 0, sizeof(sa)); // 콜백 함수가 있는 경우 (시그널 등록) if (cb) { sa.sa_flags = SA_SIGINFO; // 시그널 핸들러가 추가 정보를 받을 수 있도록 설정. sa.sa_sigaction = &SigAction; // 시그널이 발생했을 때 호출될 함수 포인터를 설정. // 시그널 리스트 'l'을 순회하며 각 시그널 'val'에 대해 작업을 수행합니다. for (uint16_t val : l) { // 해당 시그널에 이미 콜백이 등록되어 있지 않은지 확인합니다. 등록되어 있다면 오류를 출력합니다. CHECK(!state->signal_map[val].cb) << "Signal " << val << " was already registered"; state->signal_map[val].cb = cb; // 콜백 함수를 시그널 맵에 저장합니다. CHECK_EQ(0, sigaction(val, &sa, NULL)); // 'sigaction' 함수를 호출하여 시그널 핸들러를 등록합니다. } } else { // 콜백 함수가 없는 경우 (시그널 해제) sa.sa_handler = SIG_DFL; // 시그널 핸들러를 기본 핸들러로 설정합니다. for (uint16_t val : l) { // 해당 시그널에 콜백이 등록되어 있는지 확인합니다. 등록되어 있지 않다면 오류를 출력합니다. CHECK(state->signal_map[val].cb) << "Signal " << val << " was already registered"; state->signal_map[val].cb = nullptr; // 콜백 함수를 시그널 맵에서 제거합니다. CHECK_EQ(0, sigaction(val, &sa, NULL)); // 성공하면 0을 반환합니다. } } }

◼️ CbNotify, struct ResumablePoint

'CbNotify'는 특정한 조건을 가진 함수 객체를 정의하는 타입 별칭입니다.
이 함수 객체는 내부에 함수를 저장하고, 해당 함수를 호출할 때 'fb2::detail::FiberInterface*'와 'CqeResult'를 인자로 받으며, 반환값은 없습니다. using IoResult = int; struct CqeResult { int32_t res; uint32_t flags; }; // using CbNotify = std::function<void(fb2::detail::FiberInterface* current, CqeResult)>; // // https://chatgpt.com/share/c91e7b41-c30f-4f7e-9920-1afac54e02d9 // 2024.07.26(Fri) Charlie // 이 코드에서 'CbNotify'는 'fu2::function_base'라는 템플릿을 기반으로 하는 타입 별칭입니다. // 각 템플릿 매개변수는 'CbNotify'가 어떻게 작동하는지를 정의합니다. // 각 매개변수의 의미는 다음과 같습니다: // // 1. 'true owns': 이 매개변수는 함수 객체가 자신이 소유하는 호출 객체를 소유함을 의미합니다. // 즉, 'CbNotify'는 자신이 호출할 함수의 복사본을 내부에 저장합니다. // 2. 'false non-copyable': 이 매개변수는 함수 객체가 복사 불가능하다는 것을 의미합니다. // 따라서 'CbNotify' 객체는 복사될 수 없고, 오직 이동만 가능합니다. // 3. 'fu2::capacity_fixed<16, 8>': 이 매개변수는 함수 객체의 고정된 크기를 정의합니다. // 여기서 '16'은 내부 저장소의 크기(바이트 단위)를, '8'은 내부 저장소의 정렬(alignment)을 나타냅니다. // 이는 함수 객체가 일정한 크기와 정렬을 가지도록 합니다. // 4. 'false non-throwing': 이 매개변수는 함수 객체의 연산이 예외를 던질 수 있음을 의미합니다. // 따라서 'CbNotify'의 연산은 예외를 발생시킬 수 있습니다. // 5. 'false strong exceptions guarantees': 이 매개변수는 함수 객체가 강력한 예외 안전성을 // 제공하지 않음을 의미합니다. // 강력한 예외 안전성은 예외 발생 시 상태가 변하지 않음을 보장하지만, 여기서는 이러한 보장이 없습니다. // 6. 'void(fb2::detail::FiberInterface*, CqeResult)': 이 매개변수는 호출 객체의 함수 시그니처를 정의합니다. // 'CbNotify' 객체는 'fb2::detail::FiberInterface*'와 'CqeResult' 두 가지 인자를 받아들이고, // 반환 타입은 'void'입니다. // // 즉, 'CbNotify'는 소유하는 호출 객체를 저장하고, 복사 불가능하며, 고정된 크기를 가지고, // 예외를 던질 수 있고, 강력한 예외 안전성을 제공하지 않으며, // 두 개의 인자를 받아들이고 반환값이 없는 함수 객체를 정의합니다. // // 정리하면, 'CbNotify'는 특정한 조건을 가진 함수 객체를 정의하는 타입 별칭입니다. // 이 함수 객체는 내부에 함수를 저장하고, 해당 함수를 호출할 때 'fb2::detail::FiberInterface*'와 // 'CqeResult'를 인자로 받으며, 반환값은 없습니다. /*----------------------------------------------------------------------------- base/function2.hpp template <bool IsOwning, bool IsCopyable, typename Capacity, bool IsThrowing, bool HasStrongExceptGuarantee, typename... Signatures> using function_base = detail::function<detail::config<IsOwning, IsCopyable, Capacity>, detail::property<IsThrowing, HasStrongExceptGuarantee, Signatures...>>; -----------------------------------------------------------------------------*/ using CbNotify = fu2::function_base<true /*owns*/, false /*non-copyable*/, fu2::capacity_fixed<16, 8>, false /* non-throwing*/, false /* strong exceptions guarantees*/, void(fb2::detail::FiberInterface*, CqeResult)>; // 재개 가능한 지점 // ResumeablePoint: 스펠링 오류 'e'가 빠져야한다. struct ResumablePoint { CbNotify cb; // void(fb2::detail::FiberInterface*, CqeResult) uint32_t next; ResumablePoint() : next(0) { } };

• CbNotify 설명(chatgpt)

◼️ struct Worker

멀티스레딩 환경에서 여러 작업자(Worker)들을 관리하기 위한 기초를 제공합니다.
각 작업자(Worker)는 자체 큐를 가지고 있어, 작업을 독립적으로 처리할 수 있습니다.
스레드는 'pthread_t'를 사용하여 식별되고, 작업자의 큐는 MPMC 큐를 통해 생산자와 소비자 간의 데이터를 안전하게 전달할 수 있게 합니다.
'workers' 벡터를 통해 여러 작업자를 손쉽게 관리할 수 있습니다.
// 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/9e5dc430-f7f9-42b0-8057-8562759f015d // 이 함수는 멀티스레딩 환경에서 여러 작업자들을 관리하기 위한 기초를 제공합니다. // 각 작업자는 자체 큐를 가지고 있어, 작업을 독립적으로 처리할 수 있습니다. // 스레드는 'pthread_t'를 사용하여 식별되고, 작업자의 큐는 MPMC 큐를 통해 생산자와 소비자 간의 // 데이터를 안전하게 전달할 수 있게 합니다. // 'workers' 벡터를 통해 여러 작업자를 손쉽게 관리할 수 있습니다. struct Worker { // 여기서 'base::mpmc_bounded_queue<int>'는 멀티 프로듀서 멀티 컨슈머(여러 생산자 및 여러 소비자)를 // 지원하는 고정 크기 큐로 보입니다. 큐의 크기는 256으로 고정되어 있습니다. // 스마트 포인터를 사용하여 'mpmc_bounded_queue' 객체의 수명 관리를 자동으로 처리합니다. unique_ptr<base::mpmc_bounded_queue<int>> queue; // POSIX 스레드를 식별하는 데 사용되는 데이터 타입입니다. pthread_t pid; // 'Worker' 생성자 // 'Worker'의 생성자는 'queue'를 초기화합니다. // 이 초기화는 큐를 생성하고 크기를 256으로 설정합니다: // 생성자가 호출될 때마다 새 'mpmc_bounded_queue<int>' 객체가 동적으로 할당되고, // 이는 'unique_ptr'인 'queue'에 의해 관리됩니다. Worker() : queue(new base::mpmc_bounded_queue<int>(256)){}; }; // 'workers'는 'Worker' 구조체의 인스턴스를 저장하는 벡터입니다. // 이 벡터는 여러 'Worker' 객체를 보관하여 작업자들을 관리하는 데 사용됩니다. // 각각의 작업자는 큐와 스레드를 가지고 있습니다. // 아래 main()에서 // 1) parameter로 받은 thread 수 만큼 workers를 할당한다. workers.resize(num_workers); // 2) num_workers 수 만큼 돌면서 pthread를 생성한다. // pthread_t tid = base::StartThread(nm.c_str(), [i] { WorkerThread(i); }); // 3) tid를 workers.pid에 넣는다. workers[i].pid = tid; std::vector<Worker> workers; // SetupIORing() 함수에서 suspended_list.resize(1024); thread_local std::vector<ResumablePoint> suspended_list; // SetupIORing() 함수에서 next_rp_id = 0; thread_local uint32_t next_rp_id = UINT32_MAX; bool loop_exit = false; int listen_fd = -1; // socket_fd // 2024.07.26(Fri) Charlie // SuspendMyself()에서만 호출한다. // 현재 next_rp_id를 리턴하고, next_rp_id를 증가시킨다. uint32_t NextRp() { ResumablePoint* rp = &suspended_list[next_rp_id]; CHECK_LT(rp->next, suspended_list.size()); uint32_t res = next_rp_id; next_rp_id = rp->next; return res; }

• struct Worker 설명(chatgpt)

◼️ SuspendMyself() 함수

이 함수는 io_uring을 사용하여 비동기 작업을 처리하는 중에 현재 Fiber를 일시 정지시키고, 특정 조건이 충족되면 동일한 스레드에서 다시 활성화되도록 설정하는 역할을 합니다. // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/ae6f3ba2-c4a4-4d22-9b7f-521ef14e87bb // 이 함수는 io_uring을 사용하여 비동기 작업을 처리하는 중에 현재 Fiber를 일시 정지시키고, // 특정 조건이 충족되면 동일한 스레드에서 다시 활성화되도록 설정하는 역할을 합니다. // Recv(), Send(), AcceptFiber()에서 호출합니다. CqeResult SuspendMyself(io_uring_sqe* sqe) { DVLOG(2) << "SuspendMyself() sqe->fd:"<< sqe->fd << ", sqe->len:" << sqe->len; // myindex 설정: 'NextRp()' 함수를 호출하여 'myindex'라는 값을 가져옵니다. // 이 값은 아마도 현재 실행 중인 작업을 식별하는 인덱스일 것입니다. uint32_t myindex = NextRp(); sqe->user_data = myindex; CqeResult cqe_result; // CQE(completion queue event) auto& rp = suspended_list[myindex]; // rp: 'myindex'에 해당하는 항목을 참조하여 'rp'에 저장합니다. auto* fi = fb2::detail::FiberActive(); // fi: 현재 활성화된 Fiber 객체를 'fi'에 저장합니다. // 깨어날 때 이 cb가 실행된다. rp.cb = [&cqe_result, fi](fb2::detail::FiberInterface* current, CqeResult res) { cqe_result = res; DVLOG(2) << "rp.cb() current=active.name:'" << current->name() << "', fi=other.name:'" << fi->name() << "'"; DVLOG(2) << "rp.cb() call ActivateSameThread(current, fi)"; ActivateSameThread(current, fi); // fi를 active로 만든다. }; // 현재 실행 중인 Fiber를 중단하고 나중에 재개할 수 있도록 만듭니다. DVLOG(1) << "SuspendMyself() Before Suspend"; fi->Suspend(); DVLOG(1) << "SuspendMyself() After Suspend"; return cqe_result; }

• SuspendMyself() 함수 설명(chatgpt)

◼️ Recv() 함수

이 함수는 주어진 파일 디스크립터(fd)에서 데이터를 읽어오는 함수입니다.
io_uring을 사용하여 비동기적으로 데이터를 수신하는 기능을 제공합니다. // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/a2edcce9-d8e5-4317-8fca-eb87ffe9e96c // 이 함수는 주어진 파일 디스크립터(fd)에서 데이터를 읽어오는 함수입니다. // io_uring을 사용하여 비동기적으로 데이터를 수신하는 기능을 제공합니다. // return 0 if succeeded. int Recv(int fd, uint8_t* buf, size_t len, io_uring* ring) { // 1. 파일 디스크립터가 유효한지 확인합니다. DCHECK_GE(fd, 0); // 2. 변수 초기화 CqeResult cqe_result; // SuspendMyself 함수의 결과를 저장합니다. size_t offs = 0; msghdr msg; // 메시지 헤더를 나타내는 구조체입니다. iovec vec[1]; // 데이터 버퍼를 나타내는 iovec 배열입니다. memset(&msg, 0, sizeof(msg)); msg.msg_iov = const_cast<iovec*>(vec); msg.msg_iovlen = 1; // 3. 루프를 통해 데이터 수신 while (len > 0) { // len이 0보다 큰 동안 계속 반복합니다. DVLOG(1) << "Recv() len:"<< len; io_uring_sqe* sqe = io_uring_get_sqe(ring); // io_uring_sqe 객체를 가져와 설정합니다. vec->iov_base = buf + offs; vec->iov_len = len; io_uring_prep_recv(sqe, fd, buf + offs, len, 0); // 수신 작업을 준비합니다. cqe_result = SuspendMyself(sqe); // 작업이 완료될 때까지 대기합니다. DVLOG(1) << "Recv() cqe_result.res:" << cqe_result.res; // len:받은 바이트수 // 4. 결과 처리 // cqe_result.res가 0보다 크면, 수신된 데이터의 길이를 len에서 빼고, // 오프셋을 증가시켜 다음 수신 준비를 합니다. if (cqe_result.res > 0) { CHECK_LE(uint32_t(cqe_result.res), len); len -= cqe_result.res; offs += cqe_result.res; continue; } // cqe_result.res가 0이면 연결이 중단된 것이므로 'ECONNABORTED'를 반환합니다. if (cqe_result.res == 0) { DVLOG(1) << "Recv() cqe_result.res:0 return 'ECONNABORTED'"; return ECONNABORTED; // 103 } // cqe_result.res가 -EAGAIN이면, 다시 시도합니다. if (-cqe_result.res == EAGAIN) { DVLOG(1) << "Recv() -cqe_result.res == EAGAIN"; continue; } // 그 외의 경우에는 오류 코드를 반환합니다. return -cqe_result.res; } // 5. 성공적으로 수신이 완료되면 0 반환 return 0; } // 요약 // 이 함수는 비동기 I/O 프레임워크인 io_uring을 사용하여 파일 디스크립터에서 // 데이터를 수신하는 작업을 수행합니다. // 데이터를 성공적으로 수신할 때까지 반복하며, 오류가 발생하면 적절한 오류 코드를 반환합니다.

• Recv() 함수 설명(chatgpt)

◼️ Send() 함수

이 함수는 특정 파일 디스크립터(fd)에 데이터를 비동기적으로 전송하는 함수입니다.
io_uring 인터페이스를 사용하여 효율적인 입출력 작업을 수행합니다. static base::Histogram send_hist; // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/827548fe-f40e-4536-81a1-ac2158b708b0 // 이 함수는 특정 파일 디스크립터(fd)에 데이터를 비동기적으로 전송하는 함수입니다. // io_uring 인터페이스를 사용하여 효율적인 입출력 작업을 수행합니다. int Send(int fd, const uint8_t* buf, size_t len, io_uring* ring) { CqeResult cqe_result; // completion queue event size_t offs = 0; // while 루프는 len이 0보다 큰 동안 반복됩니다. // 즉, 전송할 데이터가 남아있는 동안 계속 실행됩니다. while (len > 0) { DVLOG(1) << "Send() len:"<< len; io_uring_sqe* sqe = io_uring_get_sqe(ring); // // submission queue entry io_uring_prep_send(sqe, fd, buf + offs, len, MSG_NOSIGNAL); // 현재 SQE에 대해 완료 큐 이벤트(CQE)를 기다립니다. // 그 결과를 cqe_result에 저장합니다. cqe_result = SuspendMyself(sqe); DVLOG(1) << "Send() cqe_result.res:" << cqe_result.res; // len:보낸 바이트수 // cqe_result.res가 0 이상이면, 정상적으로 전송된 바이트 수를 나타냅니다. if (cqe_result.res >= 0) { CHECK_LE(uint32_t(cqe_result.res), len); len -= cqe_result.res; // len에서 전송된 바이트 수(cqe_result.res)를 빼고, // offs에 전송된 바이트 수를 더해, 다음 전송할 데이터의 위치와 남은 길이를 조정합니다. offs += cqe_result.res; continue; } // cqe_result.res가 0보다 작으면 전송 중 오류가 발생한 것입니다. // 이 경우 -cqe_result.res를 반환하여 오류를 나타냅니다. return -cqe_result.res; // +값이 반환된다. } // 모든 데이터가 성공적으로 전송되면 0을 반환합니다. return 0; }

• Send() 함수 설명(chatgpt)

◼️ HandleSocket() 함수

이 함수는 데이터를 소켓으로부터 받아 다시 소켓으로 보내는 에코 서버의 주요 기능을 수행합니다.
데이터 수신 및 전송 시 발생하는 시간을 기록하여 성능을 모니터링할 수 있습니다.
AcceptFiber(), RunEventLoop()에서 호출합니다. // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/24ae2ed5-eb6d-468b-b8b7-e85a590f6ca6 // 이 함수는 데이터를 소켓으로부터 받아 다시 소켓으로 보내는 에코 서버의 주요 기능을 수행합니다. // 데이터 수신 및 전송 시 발생하는 시간을 기록하여 성능을 모니터링할 수 있습니다. // AcceptFiber(), RunEventLoop()에서 호출합니다. // fd == accepted_fd = accept4() void HandleSocket(io_uring* ring, int fd) { VLOG(1) << "HandleSocket() Start fd:" << fd; uint32_t size = absl::GetFlag(FLAGS_size); // 버퍼의 크기를 가져옵니다. // 크기가 size인 버퍼를 동적으로 할당하고 이를 고유 포인터로 관리합니다. std::unique_ptr<uint8_t[]> buf(new uint8_t[size]); while (true) { DVLOG(2) << "HandleSocket() Trying to Recv() from socket " << fd; // Recv 함수를 호출하여 소켓 fd로부터 데이터를 받아 버퍼에 저장합니다. // 성공하면 0을 리턴합니다. int res = Recv(fd, buf.get(), size, ring); DVLOG(1) << "HandleSocket() Recv res:" << res << (res==0?" Success":" Fail"); if (res > 0) { // error LOG(INFO) << "HandleSocket() Recv Error break"; break; } //DVLOG(1) << "After recv from socket " << fd; auto prev = absl::GetCurrentTimeNanos(); // 현재 시간을 나노초 단위로 기록합니다. DVLOG(1) << "HandleSocket() Send to socket fd:" << fd; // Send 함수를 호출하여 소켓 fd로 데이터를 보냅니다. // 성공하면 0을 리턴합니다. res = Send(fd, buf.get(), size, ring); DVLOG(1) << "HandleSocket() Send res:" << res << (res==0?" Success":" Fail"); auto now = absl::GetCurrentTimeNanos(); // 데이터 전송에 걸린 시간을 마이크로초 단위로 계산하여 send_hist에 추가합니다. send_hist.Add((now - prev) / 1000); if (res > 0) { break; } } LOG(INFO) << "HandleSocket() close(fd:" << fd <<")"; close(fd); // 소켓 파일 디스크립터를 닫습니다. accepted_fd }

• HandleSocket() 함수 설명(chatgpt)

◼️ AcceptFiber() 함수

주어진 리스닝 파일 디스크립터(listen_fd)로 들어오는 네트워크 연결을 수락하고, 이를 처리하기 위해 io_uring 인터페이스를 사용하는 함수입니다.
함수의 주요 목적은 새로운 연결을 지속적으로 수락하고, 이러한 연결을 워커(worker)에 분배하거나 직접 처리하는 것입니다. // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/1b2a8cd3-77fd-47f5-9596-b7bafae9abb9 // 주어진 리스닝 파일 디스크립터(listen_fd)로 들어오는 네트워크 연결을 수락하고, // 이를 처리하기 위해 io_uring 인터페이스를 사용하는 함수입니다. // 함수의 주요 목적은 새로운 연결을 지속적으로 수락하고, // 이러한 연결을 워커(worker))에 분배하거나 직접 처리하는 것입니다. // listen_fd == socket_fd void AcceptFiber(io_uring* ring, int listen_fd, unsigned worker_id) { // AcceptFiber 함수가 시작됨을 로그로 남깁니다. DVLOG(1) << "AcceptFiber(listen_fd:" << listen_fd <<") Start"; CqeResult cqe_result; // 'cqe_result'를 초기화하여 완료 큐 이벤트 결과를 저장합니다. // 새로운 연결을 지속적으로 수락하기 위해 무한 루프에 들어갑니다. while (true) { // accept4 함수를 호출하여 리스닝 소켓(listen_fd)에서 새로운 연결을 수락합니다. // accept4()에서 대기하는 것은 아니다. int accepted_fd = accept4(listen_fd, NULL, NULL, SOCK_NONBLOCK | SOCK_CLOEXEC); DVLOG(1) << "AcceptFiber() accept4():" << accepted_fd << ", errno:" << errno; // accept4가 성공적으로 실행되어 음수가 아닌 파일 디스크립터를 반환하면, // 해당 파일 디스크립터를 로그로 기록합니다. if (accepted_fd>= 0) { if (!workers.empty()) { // 사용 가능한 워커가 있다면. num_workers>0 DVLOG(1) << "AcceptFiber() workers current_worker:" << worker_id; // 수락한 파일 디스크립터(accepted_fd)를 다음 워커의 큐에 추가합니다. // 성공하면 true, 실패하면 false // Worker() : queue(new base::mpmc_bounded_queue<int>(256)){}; CHECK(workers[worker_id].queue->try_enqueue(accepted_fd)); } else { // num_workers==0 DVLOG(1) << "AcceptFiber() No workers -> Create Fiber('connX',HandleSocket())"; // 워커가 없으면, "conn"이라는 이름의 새로운 파이버(fiber)를 생성하여 직접 연결을 처리합니다. // 이때 HandleSocket 함수를 호출합니다. while(true) Fiber("connX", [ring, accepted_fd] { HandleSocket(ring, accepted_fd); }).Detach(); } } else if (errno == EAGAIN) { DVLOG(1) << "AcceptFiber() Error:EAGAIN '" << strerror(errno) << "'"; // accept4()가 EAGAIN을 반환하면, 이는 대기 중인 연결이 없음을 의미합니다. // 새로운 io_uring 제출 큐 엔트리(sqe)를 준비하여 리스닝 소켓에서 입력 이벤트(POLLIN)를 폴링합니다. io_uring_sqe* sqe = io_uring_get_sqe(ring); io_uring_prep_poll_add(sqe, listen_fd, POLLIN); // switch back to the main loop. // 현재 파이버를 중단하고 폴링 이벤트가 완료될 때까지 대기합니다. //DVLOG(1) << "Resuming to loop from AcceptFiber"; // "AcceptFiber에서 루프 재개" DVLOG(2) << "AcceptFiber() Before SuspendMyself()"; cqe_result = SuspendMyself(sqe); // 폴링 이벤트가 완료되면 파이버를 재개하고 로그를 남깁니다. DVLOG(2) << "AcceptFiber() After SuspendMyself()"; //DVLOG(1) << "Resuming AcceptFiber after loop"; // "루프 후 AcceptFiber 재개" if ((cqe_result.res & (POLLERR | POLLHUP)) != 0) { // 오류가 발생하면 오류를 로그로 기록하고 loop_exit를 true로 설정하여 루프를 종료합니다. // Ctril+C 여기로 온다. LOG(ERROR) << "aborted " << cqe_result.res; LOG(INFO) << "AcceptFiber() aborted " << cqe_result.res; loop_exit = true; break; } } else { // accept4가 EAGAIN 이외의 오류를 반환하면, 오류를 로그로 기록하고 // loop_exit를 true로 설정하여 루프를 종료합니다. LOG(ERROR) << "Error " << strerror(errno); LOG(INFO) << "AcceptFiber() Error: " << strerror(errno); loop_exit = true; break; } } }; // 요약 // 'AcceptFiber' 함수는 논블로킹 I/O와 io_uring을 사용하여 새로운 연결을 효율적으로 수락하고, // 이를 워커 파이버에 분배하거나 직접 처리하는 역할을 합니다. // 대기 중인 연결이 없을 경우 io_uring을 사용하여 이벤트를 대기하고, 오류가 발생하면 적절히 처리합니다.

• AcceptFiber() 함수 설명(chatgpt)

◼️ DispatchCqe() 함수

이 함수는 IO 작업이 완료된 후 적절한 콜백을 호출하여 결과를 처리하는 역할을 합니다.
'io_uring_cqe'는 IO 작업 완료 이벤트를 나타내며, 'suspended_list'는 대기 중인 작업의 콜백 함수를 관리합니다. // 2024.07.26(Fri) Charlie // 소스 설명: https://chatgpt.com/share/15680860-72dd-4041-9da5-83f618ee14aa // 이 함수는 IO 작업이 완료된 후 적절한 콜백을 호출하여 결과를 처리하는 역할을 합니다. // 'io_uring_cqe'는 IO 작업 완료 이벤트를 나타내며, // 'suspended_list'는 대기 중인 작업의 콜백 함수를 관리합니다. void DispatchCqe(FiberInterface* current, const io_uring_cqe& cqe) { uint32_t index = cqe.user_data; CHECK_LT(index, suspended_list.size()); // 'index'가 'suspended_list'의 크기보다 작은지 확인합니다. CqeResult res; res.res = cqe.res; res.flags = cqe.flags; suspended_list[index].next = next_rp_id; next_rp_id = index; DVLOG(2) << "DispatchCqe() call cb"; // 'std::move'는 소유권을 이전하여 성능을 최적화하는 데 사용됩니다. auto func = std::move(suspended_list[index].cb); func(current, res); // using CbNotify }

• DispatchCqe() 함수 설명(chatgpt)

◼️ RunEventLoop() 함수

이 이벤트 루프 함수는 주어진 'io_uring' 인스턴스를 사용하여 이벤트를 처리하고, Fiber를 통해 비동기 작업을 관리합니다. MyPolicy.Run() call RunEventLoop() // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/048be66d-fa3f-46f5-8688-e42db0d3caa9 // 이 이벤트 루프 함수는 주어진 'io_uring' 인스턴스를 사용하여 이벤트를 처리하고, // Fiber를 통해 비동기 작업을 관리합니다. MyPolicy.Run() call RunEventLoop() void RunEventLoop(int worker_id, io_uring* ring, fb2::detail::Scheduler* sched) { VLOG(1) << "RunEventLoop() Start. worker:" << worker_id; // debug 모드로 컴파일해서 실행하니 아래 DCHECK()에서 서버 멈춤. // 이 코들 주석처리해도 아래 if(sched->HasReady())로 분기해서 처리함. // DCHECK(!sched->HasReady()); 2024.09.27 comment io_uring_cqe* cqe = nullptr; // 완료 큐 이벤트(CQE)를 가리키는 포인터. unsigned spins = 0; // 반복 횟수를 세는 변수. // 총 처리된 작업, 제출된 작업, 지연된 작업의 수를 추적하는 변수. size_t total = 0, submitted = 0, stalled = 0; FiberInterface* dispatcher = fb2::detail::FiberActive(); // 현재 활성화된 Fiber. base::Histogram cqe_hist; // CQE 히스토그램을 기록하는 변수. do { ++total; // 총 반복 횟수를 증가시킵니다. DVLOG(1) << "RunEventLoop() work:" << worker_id << ", spins:" << spins; // 'worker_id'가 유효한 경우:main()에서 생성한 MyPolicy(-1)은 이 부분을 실행하지 않는다. if (worker_id >= 0) { int fd = -1; // accepted_fd // queue에서 파일 디스크립터(fd)를 꺼내어 HandleSocket 함수를 통해 처리하는 // Fiber를 생성하고 분리합니다. while(true) while (workers[worker_id].queue->try_dequeue(fd)) { // fd를 받아옴. DVLOG(1) << "RunEventLoop() call HandleSocket()"; Fiber("connY", [ring, fd] { HandleSocket(ring, fd); }).Detach(); DVLOG(2) << "RunEventLoop() After HandleSocket() fd:" << fd; } } int num_submitted = io_uring_submit(ring); // io_uring_submit을 호출하여 작업을 제출하고, CHECK_GE(num_submitted, 0); // 제출된 작업의 수가 0 이상인지 확인합니다. uint32_t head = 0; unsigned i = 0; submitted += (num_submitted > 0); // 제출된 작업의 수를 'submitted'에 누적합니다. // io_uring_for_each_cqe를 통해 완료된 이벤트를 처리하고, // 각 이벤트에 대해 DispatchCqe 함수를 호출합니다. io_uring_for_each_cqe(ring, head, cqe) { DispatchCqe(dispatcher, *cqe); ++i; } io_uring_cq_advance(ring, i); // 완료된 이벤트 수만큼 완료 큐를 전진시킵니다. if (i) { cqe_hist.Add(i); // 완료된 이벤트 수를 히스토그램에 추가합니다. } // 스케줄러에 잠자고 있는 Fiber가 있는 경우 이를 처리합니다. if (sched->HasSleepingFibers()) { sched->ProcessSleep(); } // 스케줄러에 준비된 Fiber가 있는 경우: ready_queue_ if (sched->HasReady()) { do { // 준비된 Fiber를 꺼내어 현재 Fiber(dispatcher)를 준비 상태로 돌려놓고, // 꺼낸 Fiber로 전환합니다. FiberInterface* fi = sched->PopReady(); // ready_queue_.pop_front(); DVLOG(2) << "RunEventLoop() sched->AddReady(dispatcher:'" << dispatcher->name() << "')"; sched->AddReady(dispatcher); // ready_queue_.push_back(*fibi); DVLOG(2) << "RunEventLoop() fi->SwitchTo() name:'" << fi->name() << "'"; auto fc = fi->SwitchTo(); DCHECK(!fc); } while (sched->HasReady()); // 더 이상 준비된 Fiber가 없을 때까지 반복합니다. continue; } // 스핀 횟수가 20 미만인 경우 반복을 계속합니다. if (spins++ < 2) continue; // to ensure we exit upon signal. /*struct __kernel_timespec ts { .tv_sec = 0, .tv_nsec = 10000000 }; sigset_t ss; sigemptyset(&ss); sigaddset(&ss, SIGINT); sigaddset(&ss, SIGTERM);*/ DVLOG(1) << "RunEventLoop() worker:" << worker_id << ", io_uring_wait_cqes() waiting..."; // 그렇지 않은 경우, 'io_uring_wait_cqes'를 호출하여 새로운 이벤트가 발생할 때까지 대기합니다. io_uring_wait_cqes(ring, &cqe, 1, NULL, NULL); spins = 0; // 스핀 횟수를 초기화하고 ++stalled; // 지연된 작업 수를 증가시킵니다. } while (!loop_exit); // 함수 종료 시, 총 처리된 작업 수, 제출된 작업 수, 지연된 작업 수를 로그로 출력합니다. // Ctrl+C를 했을때 위 while loop를 빠져나온다. VLOG(1) << "Exit RunEventLoop() total/sumitted/stalled " << total << "/" << submitted << "/" << stalled; // CQE 히스토그램을 로그로 출력합니다. VLOG(1) << "RunEventLoop() CQE histogram " << cqe_hist.ToString(); } // 요약 // 이 함수는 주로 I/O 작업을 비동기적으로 처리하고, 스케줄러를 통해 Fiber를 관리하여 // 효율적인 이벤트 처리를 구현하는 데 사용됩니다.

• RunEventLoop() 함수 설명(chatgpt)

◼️ SetupIORing() 함수

이 함수는 'io_uring'을 설정하는 역할을 합니다. // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/9a2dc883-e620-44a1-b297-e59917baf715 // 이 함수는 'io_uring'을 설정하는 역할을 합니다. void SetupIORing(io_uring* ring) { DVLOG(1) << "SetupIORing() Start"; io_uring_params params; // 'io_uring' 파라미터를 저장할 구조체 'params'를 선언합니다. memset(¶ms, 0, sizeof(params)); // 'params' 구조체를 0으로 초기화합니다. // 큐 깊이(queue depth)가 512인 'io_uring'을 초기화합니다. // 초기화가 성공했는지 확인하며, 실패하면 오류를 발생시킵니다. CHECK_EQ(0, io_uring_queue_init_params(512, ring, ¶ms)); // 이 작업은 'io_uring' 인터페이스와 커널 간의 상호작용을 위해 필요합니다. int res = io_uring_register_ring_fd(ring); // 파일 디스크립터 등록에 실패하면, 에러 메시지를 출력합니다. VLOG_IF(1, res < 0) << "io_uring_register_ring_fd failed: " << -res; suspended_list.resize(1024); for (size_t i = 0; i < suspended_list.size(); ++i) { // 'suspended_list'의 각 항목에 대해 다음 인덱스를 가리키도록 설정합니다. suspended_list[i].next = i + 1; } next_rp_id = 0; // thread_local uint32_t next_rp_id = UINT32_MAX; } // 요약 // 이 함수는 'io_uring' 인터페이스를 설정하고, // 작업 관리에 필요한 리스트를 초기화하는 작업을 수행합니다.

• SetupIORing() 함수 설명(chatgpt)

◼️ SetupListener() 함수

이 함수는 주어진 포트 번호를 사용하여 네트워크 소켓을 설정하고, 이 소켓을 통해 클라이언트 연결을 수신할 준비를 합니다. // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/ed019481-fffa-4d6a-bc52-274a0683eb57 // 이 함수는 주어진 포트 번호를 사용하여 네트워크 소켓을 설정하고, // 이 소켓을 통해 클라이언트 연결을 수신할 준비를 합니다. int SetupListener(uint16_t port) { // 1. 소켓 생성 // 'socket()' 함수는 새로운 소켓을 생성합니다. // 'AF_INET'은 IPv4 인터넷 프로토콜을 사용하겠다는 뜻입니다. // 'SOCK_STREAM'은 TCP 소켓을 의미합니다. // 'SOCK_NONBLOCK'는 소켓을 논블록 모드로 설정하여, 입출력 작업이 즉시 반환되도록 합니다. // 'SOCK_CLOEXEC'는 자식 프로세스가 'exec'를 호출할 때 소켓을 자동으로 닫도록 합니다. int listen_fd = socket(AF_INET, SOCK_STREAM | SOCK_NONBLOCK | SOCK_CLOEXEC, 0); DVLOG(1) << "SetupListener() socket(port:" << port << ") listen_fd:" << listen_fd; // 소켓 생성이 성공했는지 확인합니다. // 성공하면 'listen_fd'는 0 이상의 값이어야 합니다. CHECK_GE(listen_fd, 0); // 2. 소켓 옵션 설정 const int val = 1; // 'setsockopt()' 함수는 소켓 옵션을 설정합니다. // 'SO_REUSEADDR' 옵션은 소켓이 바인딩되었던 주소를 즉시 재사용할 수 있도록 합니다. // 'CHECK_EQ(0, ...)'는 'setsockopt()' 함수가 성공적으로 완료되었는지 확인합니다. CHECK_EQ(0, setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof(val))); // 3. 서버 주소 구조체 설정 sockaddr_in server_addr; memset(&server_addr, 0, sizeof(server_addr)); server_addr.sin_family = AF_INET; // 'sin_port'는 포트 번호를 설정하며, 'htons(port)'를 사용하여 // 호스트 바이트 순서에서 네트워크 바이트 순서로 변환합니다. server_addr.sin_port = htons(port); // 'INADDR_ANY'로 설정하여 모든 인터페이스에서 들어오는 연결을 허용합니다. server_addr.sin_addr.s_addr = INADDR_ANY; // 4. 소켓 바인딩 // 'bind()' 함수는 소켓을 특정 주소와 포트에 바인딩합니다. int res = bind(listen_fd, (struct sockaddr*)&server_addr, sizeof(server_addr)); DVLOG(1) << "SetupListener() bind(" << listen_fd << ") res:" << res; // 'CHECK_EQ(0, res)'는 'bind()' 함수가 성공적으로 완료되었는지 확인합니다. CHECK_EQ(0, res); // 5. 연결 대기 // 'listen()' 함수는 소켓을 수신 대기 상태로 설정합니다. res = listen(listen_fd, 32); // 'listen()' 함수가 성공적으로 완료되었는지 확인합니다. CHECK_EQ(0, res); // 6. 소켓 파일 디스크립터 반환 return listen_fd; // socket_fd } // 요약 // 이 함수는 성공적으로 실행되면 클라이언트 연결을 수신할 준비가 된 // 소켓 파일 디스크립터를 반환합니다.

• SetupListener() 함수 설명(chatgpt)

◼️ MyPolicy() 클래스

MyPolicy는 fb2::DispatchPolicy를 상속받은 클래스입니다.
MyPolicy 클래스는 특정 인덱스를 기반으로 I/O 링을 설정하고, 이를 통해 이벤트 루프를 실행하는 역할을 합니다. // 2024.07.25(Thu) Charlie // 소스 설명: https://chatgpt.com/share/218615a9-7f5d-4cac-8313-ea423dd35b30 // MyPolicy는 fb2::DispatchPolicy를 상속받은 클래스입니다. // MyPolicy 클래스는 특정 인덱스를 기반으로 I/O 링을 설정하고, // 이를 통해 이벤트 루프를 실행하는 역할을 합니다. class MyPolicy : public fb2::DispatchPolicy { public: MyPolicy(int index) : index_(index) { // 생성자 DVLOG(1) << "MyPolicy class constructor. call SetupIORing(&ring_)"; SetupIORing(&ring_); // io_uring 'ring_'을 초기화하는 함수 } ~MyPolicy() { // 소멸자 VLOG(1) << "~MyPolicy"; // 객체가 소멸될 때 호출되며, 'ring_'을 정리합니다. io_uring_queue_exit(&ring_); } // util/fibers/detail/scheduler.cc DispatcherImpl::Run()에서 이 Run()함수를 실행한다. void Run(fb2::detail::Scheduler* sched) final { // index_ = worker_id DVLOG(1) << "MyPolicy.Run() call RunEventLoop() index_:" << index_; RunEventLoop(index_, &ring_, sched); // RunEventLoop 함수를 호출 } void Notify() final { } io_uring* ring() { return &ring_; // ring_ 멤버 변수의 포인터를 반환합니다. } private: int index_; // MyPolicy 객체의 인덱스를 저장하는 정수형 멤버 변수입니다. io_uring ring_; // io_uring 구조체 타입의 멤버 변수입니다. }; // 요약 // MyPolicy 클래스는 특정 인덱스와 함께 초기화되며, // I/O 링을 설정하여 이벤트 루프를 실행하는 역할을 합니다. // 소멸 시에는 I/O 링을 정리합니다. // 주요 기능은 Run 함수에서 실행되는 이벤트 루프입니다.

• MyPolicy() 클래스 설명(chatgpt)

◼️ WorkerThread() 함수

네트워크 연결을 처리하는 워커 스레드를 초기화하고 관리하는 역할을 합니다. // 2024.07.25(Tue) Charlie // 소스 설명: https://chatgpt.com/share/aa91ec4c-59da-47f4-8d2c-3bcec34a7edf // 네트워크 연결을 처리하는 워커 스레드를 초기화하고 관리하는 역할을 합니다. void WorkerThread(unsigned worker_id) { // 'unshare(CLONE_FS)'는 시스템 호출로, 이 스레드의 파일 시스템 네임스페이스 // (작업 디렉토리 및 루트 디렉토리 등)를 다른 스레드와 분리합니다. // 이를 통해 이 스레드에서 파일 시스템에 대한 변경이 다른 스레드에 영향을 주지 않도록 합니다. unshare(CLONE_FS); DVLOG(1) << "WorkerThread(" << worker_id << ")"; // 'index'를 사용하여 새로운 'MyPolicy' 인스턴스를 생성합니다. MyPolicy* policy = new MyPolicy(worker_id); // 'fb2::SetCustomDispatcher(policy)'를 통해 이 정책을 스레드의 커스텀 디스패처로 설정합니다. fb2::SetCustomDispatcher(policy); // 'accept_fb'라는 이름의 파이버를 생성하여 'AcceptFiber(policy->ring(), listen_fd)'를 // 호출하여 들어오는 연결을 처리합니다. while(true) Fiber accept_fb("acceptY", [&] { AcceptFiber(policy->ring(), listen_fd, worker_id); }); accept_fb.Join(); // 'accept_fb' 파이버가 실행을 완료할 때까지 기다립니다. /* 주석 처리된 이 부분이 원래 소스 코드이다. if (index == 0) { // 'accept_fb'라는 이름의 파이버를 생성하여 'AcceptFiber(policy->ring(), listen_fd)'를 // 호출하여 들어오는 연결을 처리합니다. while(true) DVLOG(1) << "WorkerThread(" << index << ") call AcceptFiber()"; Fiber accept_fb("acceptY", [&] { AcceptFiber(policy->ring(), listen_fd); }); accept_fb.Join(); // 'accept_fb' 파이버가 실행을 완료할 때까지 기다립니다. } else { // 'index'가 0이 아닌 워커 스레드의 경우: DVLOG(1) << "WorkerThread(" << index << ") Before FiberActive()"; // 현재 파이버를 선점하고 파이버 컨텍스트를 가져옵니다. ctx::fiber_context fc = fb2::detail::FiberActive()->scheduler()->Preempt(); DVLOG(1) << "WorkerThread(" << index << ") After FiberActive()"; // 실행이 여기로 오지 않는다. // 이는 선점이 활성 파이버 컨텍스트를 생성하지 않았을 경우 참이 됩니다. DCHECK(!fc); // 워커 스레드의 종료를 'index'와 함께 기록합니다. VLOG(1) << "Exiting worker " << index << " " << bool(fc); } */ } // 요약 // - 'WorkerThread' 함수는 네트워크 연결을 처리하기 위해 워커 스레드를 초기화합니다. // - 메인 워커 스레드('index'가 0)는 리스너 소켓을 설정하고 파이버를 사용하여 들어오는 연결을 처리합니다. // - 다른 워커 스레드는 선점 작업을 수행하고 종료합니다. // - 각 스레드에 대해 커스텀 디스패처 정책을 설정하고 파일 시스템 정보를 분리하여 // 스레드의 파일 작업을 격리합니다. } // namespace

• WorkerThread() 함수 설명(chatgpt)

◼️ main() 함수

raw_echo_server의 main() 함수입니다. // raw_echo_server can be 40% faster than echo_server in single threaded mode. // why? it took me months to figure this out (blamed boost.fibers on this) but // apparently it's due to how OS assigns CPU for raw_echo_server thread. // OS figures out that it's better move it to one of the CPUs handling NIC queues. // This is not the case for echo_server as it assigns its threads to CPUs 0-N // and with single thread it's always CPU 0. It is possible to tune this by // using taskset and this is how I verified that the difference is due to thread affinity. // // raw_echo_server는 단일 스레드 모드에서 echo_server보다 40% 더 빠를 수 있습니다. // 왜? 이것을 알아내는 데 몇 달이 걸렸지만(이에 대해 Boost.섬유를 비난함) 분명히 OS가 // raw_echo_server 스레드에 CPU를 할당하는 방식 때문입니다. // OS는 이를 NIC 대기열을 처리하는 CPU 중 하나로 옮기는 것이 더 낫다는 것을 알아냅니다. // echo_server는 스레드를 CPU 0-N에 할당하고 단일 스레드의 경우 항상 CPU 0이므로 그렇지 않습니다. // 'taskset'을 사용하여 이를 조정할 수 있으며 이것이 스레드 선호도에 따른 차이임을 확인한 방법입니다. int main(int argc, char* argv[]) { MainInitGuard guard(&argc, &argv); ProfilerEnable(); DVLOG(1) << "main() Start"; RegisterSignal({SIGINT, SIGTERM}, [&](int signal) { LOG(INFO) << "Exiting on signal " << strsignal(signal); shutdown(listen_fd, SHUT_RDWR); // listen_fd == socket_fd }); DVLOG(1) << "main() new MyPolicy(-1)"; MyPolicy* policy = new MyPolicy(-1); DVLOG(1) << "main() fb2::SetCustomDispatcher(policy)"; fb2::SetCustomDispatcher(policy); // parameter로 받은 thread 수 만큼 workers를 할당한다. uint16_t num_workers = absl::GetFlag(FLAGS_threads); workers.resize(num_workers); LOG(INFO) << "main() Num threads " << num_workers; cpu_set_t cps; CPU_ZERO(&cps); uint16_t port = absl::GetFlag(FLAGS_port); listen_fd = SetupListener(port); // socket_fd DVLOG(1) << "main() SetupListener(" << port << "), listen_fd:" << listen_fd; // num_workers가 1 이상일 경우 num_workers 수 만큼 돌면서 pthread를 생성한다. for (size_t i = 0; i < num_workers; ++i) { string nm = absl::StrCat("worker", i); DVLOG(1) << "main() " << nm; // 스레드를 생성합니다. base/pthread_utils.h,cc base::StartThread() // 실행 함수는 WorkerThread()입니다. // 'i'가 0일 경우, 이 스레드는 연결 수락을 담당하는 메인 워커로 지정됩니다. // 'i'가 1 이상일 경우, 일반 워커 스레드 pthread_t tid = base::StartThread(nm.c_str(), [i] { WorkerThread(i); }); CPU_SET(i, &cps); // pthread_setaffinity_np(tid, sizeof(cpu_set_t), &cps); CPU_CLR(i, &cps); workers[i].pid = tid; // tid를 workers.pid에 넣는다. } // num_workers가 0일 경우 if (num_workers == 0) { DVLOG(1) << "main() Fiber accept_fb('acceptX')"; // util/fibers/fibers.h Fiber(name, fn); Fiber accept_fb("acceptX", [&] { AcceptFiber(policy->ring(), listen_fd, 0); }); accept_fb.Join(); } // detail::FiberActive()->scheduler()->DestroyTerminated(); for (size_t i = 0; i < num_workers; ++i) { pthread_join(workers[i].pid, nullptr); // 스레드가 종료되기를 기다린다. } LOG(INFO) << "Send histogram " << send_hist.ToString(); // terminated_queue.clear(); // TODO: there is a resource leak here. return 0; }

◼️ 프로그램 수정 내용

1. 원 소스 코드의 오류: 첫번째 스레드에서 accept4()해서 데이터를 받아서 worker 스레드에게 분배하는 방식인데, 스레드가 2개 이상일 때는 두번째 스레드가 깨어나지 않으므로 작동하지 않는다. 그래서 각 스레드마다 accept4()를 실행해서 클라이언트의 접속을 받아서 처리하도록 수정했다.
   위에 설명한 소스는 변경된 소스입니다.
2. 테스트: t_raw_echo1.py ~ t_raw_echo5.py -> 성공.
3. 소스 코드 설명
   Server - Client 통신 순서: socket()->bind()->listen()->accept() SetupListener(port) -> 프로그램 main()에서 한번만 실행한다. 서버가 준비하는 과정. int listen_fd = socket(); bind(listen_fd, server_addr); listen(listen_fd); AcceptFiber(listen_fd) -> 스레드마다 실행한다. int accepted_fd = accept4(listen_fd); // 클라이언트가 connect()를 실행하면 여기서 받는다. accept4()를 실행해서 대기하고 있는 스레드가 여러 개일 때 여러 개의 스레드가 accept4()를 실행해서 대기하고 있을 때 클라이언트 connect()가 들어오면 리눅스(OS)에서 여러 스레드 중 하나에 분배(할당)한다. 이후 데이터 보내고 받기는 해당 스레드와 클라이언트 실행한다. 다른 스레드가 간섭하지 않는다. 서버가 클라이언트 요청을 받으면: workers[worker_id].queue->try_enqueue(accepted_fd); while (workers[worker_id].queue->try_dequeue(fd)) { // fd를 받아옴. Fiber("connY", [ring, fd] { HandleSocket(ring, fd); }).Detach(); } HandleSocket(fd) Recv(fd); io_uring_prep_recv(sqe, fd); Send(fd); io_uring_prep_send(sqe, fd);
4. 변경된 코드 main() listen_fd = SetupListener(port); // 첫번째 스레드에 있던것을 main()으로 옮겼다. WorkerThread() index 구분없이 아래 코드르 실행한다. Fiber accept_fb("acceptY", [&] { AcceptFiber(policy->ring(), listen_fd, worker_id); });

---