C로 작성된 함수를 파이썬에서 사용하기

표준 파이썬 해석기는 C로 만들어져 있다. 그리고 C에는 동적 라이브러리를 링크하여 사용할 수 있는 기능이 마련되어 있다. 따라서 표준 파이썬 해석기를 만드는 사람들도 실행시간에 C로 만들어진 동적 라이브러리를 링크하여 사용할 수 있을 것이라는 생각을 당연히 했을 것이고, 그 결과 ctypes 라는 라이브러리가 탑재되어 있다. 

C로 작성된 함수를 파이썬에서 사용하기 더보기

파이썬의 숫자판별함수

파이썬에서 주어진 문자열이 숫자로 되어 있는지를 검사하는데에는 흔히 문자열의 메소드인 isdigit()을 사용한다. 그런데 파이썬의 문자열 타입을 조사해보면 같은 기능을 하는 거 같은 메소드가 무려 3종 세트로 알차게 구성되어 있는 것을 알 수 있다. 바로 isdigit(), isnumeric(), isdecimal() 이다. 그렇다면 이 세 가지 함수는 왜 각각 존재하는 것일까?

급한 사람을 위한 빠른 결론 – int 타입으로 안전하게 변환하고 싶다면 isdigit() 대신에 isdecimal()을 사용하여 검사하라

이 세 종류의 메소드는 문자값 중에서 ‘숫자’를 찾는데 사용된다. 우리 말에서 ‘숫자’라는 개념은 사실 모호한데 (응? 이게 왜 모호하다는 거지?) 우리는 일상생활에서 흔히 0, 1, 2, 3, .. , 9의 10개의 숫자를 사용하고 있다. 즉 우리는 10개의 ‘숫자’만을 갖고 있는 셈이다. 그런데 사실은 이게 좀 간단한 문제가 아닌거라는게 문제다.

흔히 사용하는 isdigit() 메소드를 보자. 말 그대로 문자열에 사용된 글자들이 ‘digit’인지를 확인하는 것이다. 아 그러니까 우리는 ‘이게 숫자로 된 문자열이네’라는 것을 알 수 있을 거라 추측한다. 그리고 digit, numeric, decimal은 다같이 ‘숫자’로 번역되기 때문에 이 사이의 차이가 뭔지를 정확하게 모르게 된다.

다시 이번에는 decimal 을 보자. 이것이 우리가 흔히 생각하는 ‘숫자’와 같은 개념이다. 즉 '0123456789'의 구성으로 이루어진 10개의 글자. 말 그대로 int 타입으로 즉시 변환이 가능한 리터럴을 구성하는 문자를 말한다.

아니 그렇다면 digit은 뭐길래? 특수 기호 중에서 어깨위에 제곱이나 세제곱을 표시하는 문자가 있다. digit은 이렇게 ‘숫자처럼 생긴’ 모든 글자를 다 숫자로 치는 것이다. 다음 예를 보면 digit과 decimal의 차이를 알 수 있을 것이다.

x = '3²'
x.isdigit()
# True
x.isdecimal()
# False
int(x)
# ERROR!!!!

위 예에서 볼 수 있는  에서 두 번째 ² 는 특수문자이며, 우리가 흔히 키보드로 입력하는 2와는 다른 코드 값이다. 이는 decimal 값을 표기하는데 사용하는 문자는 아니지만, 사람이 읽었을 때 숫자로 인지하기 때문에 isdecimal()에서는 False가 리턴되고, isdigit()에서는 True가 리턴된다.

따라서 만약 어떤 텍스트가 int 값으로 변환이 가능한지를 검사하고자 한다면 isdigit()을 사용해서는 안되며, isdecimal()을 써야 할 것이다.

그럼 isnumeric()은 어떤 것일까?

numeric 하다는 것은 보다 넓은 의미인데, isdigit()은 단일 글자가 ‘숫자’ 모양으로 생겼으면 True를 리턴한다고 했다. isnumeric()은 숫자값 표현에 해당하는 텍스트까지 인정해준다. 예를 들어 “½” 이런 특수문자도 isnumeric()에서는 True로 판정된다.

파이썬으로 구현하는 스트림리더

파이썬에서 파일의 내용을 읽어와서 처리할 때 가장 기본적인 방법은 opne() 함수를 이용해서 파일 객체를 만들고, read() 메소드를 이용해서 파일의 전체 내용을 한 번에 읽어오는 것이다. 그런데 많은 경우에 실제로 다루는 파일은 텍스트포맷인 경우가 많다. 텍스트 포맷을 다룰 때에는 다음과 같은 몇 가지 전략이 존재한다.

  • read()를 이용해서 파일의 전체 내용을 읽어와 하나의 문자열로 사용한다.
  • readlines()를 이용하면 파일의 전체 내용을 읽어와서 라인 단위로 잘라 문자열의 리스트로 만들 수 있다.
  • readline()을 이용해서 파일의 내용을 한 줄씩 읽어와서 처리한다.

실제로는 텍스트 파일을 한줄 단위로 읽어와서 파싱해서 사용하는 경우가 많기 때문에 위 방법 중에서는 readline()이 가장 많이 쓰이며, 아예 텍스트 파일을 연 파일 객체는 문자열의 제너레이터처럼 동작하기 때문에 for ... in 구문으로 한줄씩 데이터를 읽어서 사용한다. 이 방식의 가장 좋은 점은 파일을 한 줄 단위로만 읽어와서 처리하기 때문에 불필요한 메모리 낭비를 하지 않는다는 것이다. 심지어 수 기가짜리 텍스트 파일이 있더라도, 한 줄씩 처리하는 경우에는 메모리에 한줄 만큼의 분량을 읽어와서 처리하기 때문에 프로그램이 메모리 부족으로 죽을 일이 없다.

그런데 어떤 경우에는 텍스트로 구성된 대용량 파일이 있고, 이 파일 내의 데이터가 컴마나 스페이스 같은 문자만 구분되어 있고 개행문자가 포함되지 않았다면 어떻게 처리하면 좋을까? 극단적인 예이기는 하지만 몇 기가 짜리 텍스트 파일이 있고, 여기에는 컴마로 구분된 숫자값들이 있다고 하자. 이를 읽어서 split()으로 나눠서 써야겠지만, split()이라는 동작 자체가 일단 잘라낼 문자열을 메모리에 올린 다음, 한꺼번에 잘라서 리스트로 만드는 방식으로 처리되기 때문에 실제 문자열의 크기의 두 배 이상의 메모리를 요구하게 된다.

이 문제를 해결하기 위해서 조금 느긋하게 파일의 내용을 읽어들이는 방법을 구현해보자. 기본적인 아이디어는 다음과 같다.

  1. 파일의 내용을 일부 읽어올 버퍼를 준비한다.
  2. 미리 정해둔 사이즈만큼 파일의 내용을 읽어와서 버퍼에 담는다.
  3. 버퍼의 처음부터 구분자가 있는 곳까지를 탐색한 후, 해당 영역을 잘라내어 사용한다.
  4. 구분자를 버린다.
  5. 3의 과정을 반복한다. 만약 구분자가 발견되지 않으면, 파일로부터 다시 정해진 사이즈만큼의 데이터를 읽어와서 버퍼에 덧붙인다.
  6. 파일에서 더 이상 읽을 내용이 없다면 버퍼에 남아있는 데이터가 마지막 조각이 된다.
  7. 파일을 닫는다.

이처럼 파일로부터 특정한 구분자까지의 데이터를 읽어서 순차적으로 잘라내어 리턴해주는 제너레이터를 만들 수 있을 것이다.

def read_strem(filename, sep=","): 
    ## filename : 읽을 파일
    ## sep : 구분자
    buffer = bytearray()
    chunk_size = 1024 # 한 번에 1024바이트씩 읽어온다. 
    token = sep.encode()

    with open(filename, 'rb') as f: ## 파일을 이진 파일로 읽어온다. 
      while True:
        data = f.read(chunk_size)
        if not data: ## 파일에서 더이상 읽을 내용이 없으면 루프 종료
          break
        buffer.extend(data)
        ## 버퍼내에서 구분자까지 자르는 작업을 반복
        while True:
          i = buffer.find(token)
          if i < 0 :  ## 구분자가 발견되지 않으면 한 번 더 읽어온다.
            break
          found = buffer[:i].decode()
          yield found
          ## 구분자앞까지 자른 내용을 내놓은 후에는 버퍼의 앞쪽을 정리한다.
          buffer[:i+len(token)] = []
       ## 파일에서 읽는 내용을 모두 처리했다. 버퍼에 남은 내용이 있으면 내놓는다.
       if buffer:
         yield buffer.decode()

만약 아주 커다란 텍스트파일로부터 컴마로 나누어진 데이터를 읽어와서 처리하는 동작을 수행한다고 해보자. 심지어 어떤 경우에는 데이터 전체가 필요한 것도 아니고 앞 부분에서 1000개 정도만 필요할 수도 있다. 위에서 만든 함수는 제너레이터 함수이기 때문에 for 문에서 사용할 수 있고, 만약 앞 1000개까지만 사용하려면 다음과 같이 쓸 수 있다.

## 파일이름이 'verybigfile.txt'라 할 때,
g = read_stream('verybigfile.txt')
for (i, w) in enumerate(g):
    if not i < 1000:
       break
    print(w)

약간 다른 접근방법도 있다. 파이썬보다는 Objective-C나 Swift에 어울릴 것 같은 방법인데, “필요할 때마다 꺼내 쓰는” 제너레이터가 아니라 튀어나오는 데이터를 처리할 핸들러를 넘겨주고 함수 내에서 다 처리해버리는 방법이다. 핸들러의 디자인은 대략 다음과 같은 식으로 처리하면 되겠다.

  • 처리할 문자열 데이터와, 몇 번째 데이터인지를 나타내는 정수 값을 인자로 받고
  • 문자열을 처리한 후, 더 처리할 것인지 그렇지 않은지를 리턴한다. None이나 False로 평가되는 값을 리턴하면 계속하고, 그렇지 않으면 더 이상 실행하지 않도록 할 것이다.

예를 들어 다음과 같은 식으로 핸들러를 만들 수 있는데,

def process_word(word, idx):
  if idx < 10:
     print(word)
     return
  return True

위 핸들러는 10번째까지는 출력하고, 그 이후로는 더 이상 실행하지 않겠다는 의미가 된다. 이런 핸들러를 받아서 처리해주는 형태로 위 제너레이터 함수를 다시 쓰면 다음과 같은 모양이 될 것이다.

def split_file(filename, handler, sep=","):
  buffer = bytearray()
  chunk_size = 1024
  token = sep.encode()
  idx = 0
  with open(filename, 'rb') as f:
    while True:
      data = f.read(chunk_size)
      if not data:
        break
      buffer.extend(data)
      while True:
        i = buffer.find(token)
        if i < 0:
          break
        found = buffer[:i].decode()
        r = handler(found, idx)
        if r:
          return
        buffer[:i+len(token)], idx = [], idx+1
    if buffer:
      handler(buffer.decode(), idx)

## 10개 데이터만 출력하고 끝낸다.
split_file('varybidfile.txt', process_word)

이 알고리듬은 제법 간단하면서도 대용량 데이터 파일을 안전하게 처리할 수 있도록 해주는 제법 괜찮은  방법이다. 간단하기 때문에 Swift 등의 다른 언어로도 충분히 컨버팅할 수 있다.

 

 

ZMQ – Poller 사용하기

PUB-SUB 패턴이나 PUSH-PULL 패턴을 사용하면 데이터를 발생시키는 1개 (혹은 여러 개의) 노드로부터 데이터를 전달받는 N개의 노드들이 돌아가는 방식의 네트워크를 구성할 수 있음을 지난 글에서 보았다. 이 때 각각의 클라이언트 노드들은 루프를 돌면서 큐에 들어온 메시지를 순서대로 처리하게 된다.

만약 개별 클라이언트가 한 번에 처리해야 할 메시지의 최대 개수가 정해져 있다면 while 무한 루프가 아닌 for 루프를 통해서 유한한 루프를 돌 수 있을 것이다. 그 외에도 메시지의 내용으로부터 동작이나 중지를 결정할 수 있는 경우에도 그 스스로 동작을 멈출 시점을 결정할 수 있을 것이다.

여러 개의 소켓을 듣는 노드

하지만 클라이언트들의 중지를 다른 제 3의 노드가 결정해야 한다면? ‘커맨드 노드’라는 노드가 있고 이 노드가 PUB 패턴으로 모든 클라이언트들에게 중지 명령을 내리는 구조를 만든다고 가정해보자.

그렇다면 클라이언트 입장에서는 한 번에 두 개의 소켓(하나는 SUB 패턴으로 데이터를 지속적으로 수신받고, 다른 하나는 PULL 패턴으로 종료 명령을 수신받는다.) 을 사용하게 된다. 이것은 이전의 패턴들과는 사뭇다르다. 이전 예제들에서 보여왔던 다중 접속은 하나의 노드가 단일 소켓을 사용하면서, 소켓 자체가 여러 포트에 바인딩되거나, 여러 서버에 커넥트되는 것을 보였다. (그리고 ZMQ는 이 구조 내에서 메시지 전달의 교통정리를 훌륭하게 수행해준다.)

소켓으로 데이터를 반복적으로 읽어야 하는 노드의 특성상, 두 개 이상의 소켓을 하나의 노드가 갖는 것이 문제될 일은 없지만 두 노드를 한꺼번에 대기할 수는 없다. 두 개 소켓으로부터 각각 메시지를 수신하기 위해서는 각각 별도의 스레드를 통해서 처리해야 할 것이다.

POLLER

ZMQ에서는 이러한 상황을 좀 더 쉽게 해결할 수 있도록 POLLER라는 것을 제공한다. Poller는 두 개 이상의 소켓을 등록하여두면 소켓들로부터의 입력을 감지하여 (소켓, 이벤트)의 리스트를 리턴해준다. 이를 이용해서 각각의 소켓을 동시에 수신하면서 소켓별로 구분된 메시지를 얻을 수 있다.

def client(data_port=5556, com_port=5558):
  ctx = zmq.Context()

  # 1. SUB 소켓 연결
  sock_sub = ctx.socket(zmq.SUB)
  sock_sub.setsockopts_string(zmq.SUBSCRIBE, '9')
  sock_sub.connect(f'tcp://localhost:{data_port}')

  # 2. PULL 소켓 연결
  sock_pull = ctx.socket(zmq.PULL)
  sock_pull.connect(f'tcp://localhost:{com_port}')
  
  # 3. 폴러 생성 및 설정
  poller = ctx.Poller()
  poller.register(sock_sub, zmq.POLLIN)
  poller.register(sock_pull, zmq.POLLIN)

  # 폴러로부터 이벤트 수신
  should_continue = True
  while should_continue:
    # 4. 각각의 이벤트는 (소켓, 이벤트), (소켓, 이벤트) ... 의 형태로 전달
    # 이를 사전 타입으로 변환
    socks = dict(poller.poll())
    if socks.get(sock_pull, None) == zmq.POLLIN:
      command = sock_pull.recv_string()
      if command == 'EXIT':
         print('This client will be terminated...')
         should_continue = False
    if socks.get(sock_sub, None) == zmq.POLLIN:
      topic, msg = sock_sub.recv_string().split(' ', 1)
      print(f'Processing: {msg}')

  
  

poller는 복수 개의 소켓을 번갈아 수신하기 보다는 특정한 주기로 각 소켓의 메시지 큐를 검사해서 신규 메시지들을 소켓별로 하나씩 가져와서 이벤트의 리스트로 리턴해준다. 폴러에 등록되는 소켓의 타입은 서로 같은 것들이어도 무방하다. 지난 글에서 살펴본 PUSH-PULL 패턴을 사용한 분산처리는 다음과 같이 개선해볼 수 있다.

  1. Ventilator는 PUSH 패턴으로 복수의 Worker에게 데이터를 순차적으로 뿌린다.
  2. Worker는 PULL 소켓으로 데이터를 수신하여 처리한 후, 다시 PUSH 소켓으로 Sink에게 그 결과를 전달한다.
  3. Worker는 또한 SUB 소켓으로 중지명령을 기다리고 있다. 따라서 Worker에는 PULL, SUB 소켓이 Poller로 묶여있게 된다.
  4. Sink는 PULL 소켓으로 여러 Worker로부터 결과를 수신받아 취합한다.
  5. Sink는 모든 결과가 수신되면 PUB 소켓을 통해 모든 Worker들에게 kill 시그널을 전송한다.

이처럼 단순 소켓이 아니라 소켓을 결합해주는 Poller를 사용하면 복잡한 네트워크 그래프를 간단한 방법으로 구축할 수 있다. Poller외에도 ZMQ에서는 양 끝단에 소켓을 부착하여 메시지를 포워딩해주는 device라는 개념이 있는데, 다음 번에는 이 부분에 대해서 한 번 알아보는 기회를 갖도록 하자.