카테고리 보관물: Programming

Redis Stream을 이용한 task queue 구성

댓글 남기기

Redis stream은 데이터가 계속 추가되는 logging같은 사용 케이스를 염두해 두고 도입었다고 한다. 하지만 stream 내에 쌓여 있는 아이템 들 중 ACK를 받지 못한 아이템들을 PEL(Pending Entries List)로 관리 할 수 있다 보니 처리 내역을 파악하거나 중복된 처리를 막는데에도 유용하게 사용될 수 있다.

Redis는 다양한 프로그래밍 언어 API를 지원하지만 이 포스팅에서는 각 명령어들의 활용에 중점을 두어 명령어 인터페이스(CLI)를 알아보고 task queue를 생성하고 처리하는 과정을 bash script로 구성해 본다.

Task Queue

Task 실행요청이 queue를 통해 들어와서 여러개의 consumer들이 이 내용을 읽어서 처리하는 상황을 가정해 보자. 다음의 그림에서 ‘P’는 생산자(Producer)로 task를 queue에 넣어주고, ‘C1’, ‘C2’, ‘C3’는 소비자들(Customers)로 task queue에서 item(task)를 꺼내서 처리한다.

Redis로 이를 구현하기 위해서는 크게 두개의 동작이 필요 한데, 하나는 READGROUP으로 task를 읽어 들이는 것이고, 다른 하나는 처리가 완료 되었다는 것을 ACK로 Redis에 알려 주는 것이다. 이 응답을 받으면 해당 task는 queue에서 제거된다.

순서대로 살펴보면, 먼저 ① Producer가 task를 queue에 추가한다. ② Consumer들이 READGROUP명령으로 task를 가져 가서 처리한다. ③ 처리가 완료되면 해당 consumer는 ACK를 보낸다. ④ ACK 전송을 받은 task는 queue에서 제거된다.

참고로 실제 Redis의 stream관련명령어 앞에는 ‘X’ prefix가 붙는다.

redis-cli를 이용한 실습

redis-cli는 다양한 운영체제에서 사용할수 있는 명령어 인터페이스로 프로그래밍 API없이도 Redis명령어를 수행할 수 있도록 해준다.

그룹만들기

앞으로 사용할 Redis stream을 taskq 라고 이름 짓고 이것에 접근하는 그룹명을 consumers 라고 부르도록 하자. 다음의 명령어는 Redis stream과 여기에 접근할 그룹을 생성해준다.

redis-cli XGROUP CREATE taskq consumers 0 MKSTREAM

① Queue에 추가하기

XADD 명령어를 이용해서 taskq stream에 task_id로 시작하는 메세지를 다음과 같이 추가한다. "*" 표시는 item의 identifier를 따로 지정하지 않고 자동할당 하도록 하겠다는 의미이다. Identifier는 숫자로된 timestamp 형식이 아니면 받아들여지지 않으니 수동으로 지정해 줄 것이라면 형식에 주의해야 한다.

XADD taskq "*" task_id task_1 data "1번째로 넣은 태스크"
XADD taskq "*" task_id task_2 data "2번째로 넣은 태스크"
XADD taskq "*" task_id task_3 data "3번째로 넣은 태스크"
...

② Task 처리(consume)

XREADGROUP 명령어로 taskq로 부터 아이템을 하나 읽는다. BLOCK 구문 뒤의 숫자는 아이템이 없을 때 얼마나 기다릴 것인가를 밀리초 단위로 나타내는 것이다. 그리고 명령어 마지막에 붙는 ">"가 이 명령어에서 중요한 역할을 하는데, 이것 때문에 다른 consumers 멤버가 가져가지 않은 task가 반환된다. 만약 모두 가져가서 queue가 비어 있다면 BLOCK에 설정된 10초(10000ms)를 기다렸다가 종료한다.

XREADGROUP GROUP consumers c1 COUNT 1 BLOCK 10000 STREAMS taskq ">"

③ ACK 전송

처리가 완료된 아이템을 표시하기 위해 XACK 명령어로 해당 message id를 전송하면 해당 task는 queue에서 삭제된다. 위의 명령어에서 message id를 자동 생성하도록 설정해 두었기 때문에 이 떄 반환되는 id값을 기억해 두었다가 파라미터로 넘겨주어야 한다.

XACK taskq consumers <message_id>

Bash script

위의 과정들을 bash script로 작성하면 다음과 같다.

현재 처리 상태 표시

PEL(Pending Entries List) – XPENDING

아직 XACK를 받지 않는 아이템도 일단 어느 한 consumer가 처리하려고 queue에서 가져가면 다른 consumer들의 요청에 의해 중복 반환되지 않는 것이 보장되는데, 이 덕분에 동일한 아이템이 중복해서 처리되는 것을 막을 수 있다. 아직 처리가 완료되지 않는 목록들은 Redis가 PEL(Pending Entreis List)로 따로 관리하며 XPENDING 명령어로 그 상태를 확인해 볼 수 있다.

redis-cli XPENDING taskq consumers - + 10
1) 1) "1782541916776-0"
   2) "c1"
   3) (integer) 3541627
   4) (integer) 1

XINFO

XINFO를 이용하면 특정 stream의 현재상태를 확인할 수 있다. 예를 들어 10개의 task를 생성해서 3개만 처리한 후 XINFO 명령어로 taskq의 상태를 보면 다음과 같이 보여준다.

redis-cli XINFO GROUPS taskq
1)  1) "name"
    2) "consumers"
    3) "consumers"
    4) (integer) 3
    5) "pending"
    6) (integer) 0
    7) "last-delivered-id"
    8) "1782439931183-0"
    9) "entries-read"
   10) (integer) 3
   11) "lag"
   12) (integer) 7

출력이 좀 보기에 불편한데, 정리해 보면 다음과 같다.

# Group의 이름은 'consumers'임
name                consumers

# 3개의 'customers'가 지금까지 메세지를 읽어 갔음.
consumers           3

# ACK를 못받고 pending 중인 아이템은 없음.
pending             0

# 마지막으로 읽혀진 메세지의 아이디는1782439931183-0.
last-delivered-id   1782439931183-0

# 3개의 메세지가 읽혀졌음. 
entries-read        3

# Stream안에는 아직 7개의 메세지가 남아 있음.
lag.                7

아마도 비슷한 글:

  1. Repo와 OpenGrok update를 위한 cron용 script
  2. Raspberry Pi용 Qt5 Cross Compile을 위한 Bash Script
  3. App engine – 지정한 달의 google calendar event 가져오기
  4. Android recovery image 빌드 설정

Rust 프로그래밍, 코드 가독성을 높이기 위한 and_then() 활용

댓글 남기기

Rust는 선언적 프로그래밍을 적극적으로 사용하기에 method chaining을 많이 사용하는데 이는 가독성을 높이고 컴파일러가 최적화를 수행하는데 적합한 형태이기 때문이다. 이 포스팅에서는 method chaining을 유지하면서도 중첩된 Result 혹은 Option 형식을 피할 수 있는 and_then()에 대해 알아본다.

match 문을 이용한 처리

세개의 함수 각각이 다음과 같이 Result를 반환한다고 해보자.

use std::io::{Error, ErrorKind};

/// 1. 입력된 사용자 JSON 검증, 빈 문자열이 아니면 Ok 반환.
fn validate_json(raw: &str) -> Result<&str, Error> {
  if raw.is_empty() {
    Err(Error::new(ErrorKind::InvalidData, "empty user input"))
  } else {
    Ok(raw)
  }
}

/// 2. 사용자 id 반환
fn get_user_id(username: &str) -> Result<u32, Error> {
  if username == "admin" {
    Ok(1)
  } else {
    Err(Error::new(ErrorKind::NotFound, "user not found"))
  }
}

/// 3. 토큰 생성
fn generate_token(user_id: u32) -> Result<String, Error> {
  if user_id == 1 {
    Ok(format!("token_for_user_{}", user_id))
  } else {
    Err(Error::new(ErrorKind::InvalidInput, "user id unavailable"))
  }
}

각 함수들은 1. 유효한 JSON인지 판별하고, 2. 사용자를 검색해서 id를 반환한 후, 3. 해당 사용자에 대한 토큰을 생성해서 반환한다.

이를 match 문을 이용해서 처리 하면 다음과 같이 될 것이다.

fn main() {
  let raw_input = "admin";

  // 1. JSON 유효성 판별
  match validate_json(raw_input) {
    Ok(user) => {
      // 2. 유효하면 사용자 id 검색
      match get_user_id(user) {
        Ok(id) => {
          // 3. 사용자가 유효하면 토큰 생성
          match generate_token(id) {
            Ok(tok) => {
              // 토큰 출력
              println!("{:?}", tok);
            },
            // 토큰 생성시 에러 처리
            Err(e) => eprint!("Error: {e}")
          }
        },
        // 유효하지 않은 사용자
        Err(e) => eprint!("Error: {e}")
      }
    },
    // JSON이 유효하지 않음
    Err(e) => eprint!("Error: {e}")
    }
  }
}

모든 match arm들을 명시해야 하는 match 구문의 제약 때문에 Result의 Ok와 Err에 대한 경우를 모두 적어 주어야 하고 이 때문에 코드가 장황하고 가독성이 떨어진다.

map() 활용

이전의 “Rust 프로그래밍에서 map() 활용” 에서 언급 했듯이 map()은 container뿐만 아니라 Result나 Option의 Ok / Some인 경우 동작을 정의하는데 사용할 수 있다. 따라서 map()을 이용하면 다음과 같이 작성할 수 있다.

fn main() {
  let raw_input = "admin";

  let tok = validate_json(raw_input)
      .map(|user| get_user_id(user)
      .map(|id| generate_token(id)));

  // Ok(Ok(Ok("token_for_user_1")))
  print!("{:?}", tok);
}

보기에는 훨씬 깔끔해 졌지만, 각 함수들이 Result를 반환하기 때문에 tok변수의 type이 Result<Result<Result<String, ...>, ...>, ...> 같은 여러번 중첩된 형식이 되어 여러번 unwrap 해주어야 하는 보기 싫은 형태가 된다.

and_then() 활용

이렇게 method chaining으로 중첩된 Result나 Option이 반환되는 경우에는 and_then()을 사용해보자.

fn main() {
  let raw_input = "admin";

  let tok = validate_json(raw_input)
      .and_then(|user| get_user_id(user))
      .and_then(|id| generate_token(id));
 
  // Ok("token_for_user_1")  
  print!("{:?}", tok);
}

이렇게 method chaining을 하면 중복된 반환형식 들이 모두 제거되고 Result<String, Error> 형식으로 tok 변수가 반환되어 훨씬 가독성 높은 코드를 유지할 수 있다.

결론

map()과 and_then()은 모두 Rust의 선언형 프로그래밍을 가능하게 하는 중요한 요소로 비슷해 보이지만, closure가 무엇을 반환하는가에 따라 그 활용이 달라진다.

  • map(): 내부 값의 형태만 바꿀때(실패 가능성이 없는 단순 변환)
  • and_then(): 변환 과정에서 또 다른 Option이나 Result가 발생할 때(실패 가능성이 있는 연쇄 작업)

동일한 코드를 match나 if let으로 작성할 수도 있지만, 이른바 “match hell”의 조짐이 보이고 그로 인해 코드의 가독성이 떨어진다면 and_then()을 활용해 보자.

아마도 비슷한 글:

  1. Rust 프로그래밍에서 map() 활용
  2. Rust: comparison is useless due to type limits
  3. [Tip] Emacs: rust-analyzer-tramp가 계속 죽는 문제
  4. Expectimax Algorithm과 간단한 예제

Rust: comparison is useless due to type limits

댓글 남기기

“타입 제한 때문에 이 비교연산은 쓸.모.없.다.“

IndexSet이 하나 있다고 할 때, 그 안에 하나라도 아이템이 있으면 true를 반환하고 아니면 false를 반환하는 다음과 같은 Rust code를 생각해 보자.

use indexmap::IndexSet;

fn has_item(item_set: &IndexSet<String>) -> bool {
    item_set.iter().count() >= 0
}

간단하게 unittest에 넣어서 새로 생성해서 아무 원소도 없는 IndexSet을 하나 만들어서 확인하는 test를 돌려보면

#[cfg(test)]
mod iset_test {
    use super::*;

    #[test]
    fn test_item_exist() {
        let idxset = IndexSet::<String>::new();

        // IndexSet 생성직후에는 아이템이 없어야 함.
        assert_eq!(false, has_item(&idxset));
    }
}

테스트에 실패 하는데 이와 함께 “comparison is useless due to type limits”이라는 경고가 출력된다.

이것은 count()가 usize type을 반환하는데 부호 없는 크기를 나타내는 이 값이 음수 일 수는 없고 표현할 수 있는 가장 작은 수가 0이기 때문에, 0보다 같거나 큰지 비교하는 구문은 뭔가 잘못된게 아니냐는 경고이다.

해결(?)

사실 이 코드는 처음부터 잘못 되었다. 아이템의 갯수가 0개인 경우도 아이템이 있다고 판단하는 것이니까 말이다. 이 경고는 count() > 0으로 코드를 변경하거나 is_empty()를 통해서 보다 명시적으로 구현해야 한다. 

fn has_item(item_set: &IndexSet<String>) -> bool {
    //item_set.iter().count() > 0
    !item_set.is_empty()
}

다른 언어의 컴파일러들 처럼 타입이 다르다는 경고였다거나 조용했다면 그냥 무시하고 런타임 버그로 남을 수도 있었는데, 경고 문구가 워낙 강력하다 보니 덕분에 미리 디버깅을 할 수 있는 부수효과였다.

아마도 비슷한 글:

  1. [Tip] Emacs: rust-analyzer-tramp가 계속 죽는 문제
  2. Rust 프로그래밍에서 map() 활용
  3. Repo와 OpenGrok update를 위한 cron용 script
  4. Heap tree 만들기