Skip to content

Blog

  • 1 min read

2024 05 07

오늘의 계획

  • 스케쥴러 프로세스를 이벤트 루프로 변환 할 수 있는지에 대한 간단한 검증 시작
  • state 에 flag 를 넣고 무한 루프를 돌리자. flag 가 변하면 루프를 끝내자.
  • flag 를 fastapi 에서 조회/변조 할 수 있는지 시험해보자.

해봤다!

state 에 flag 를 넣고 무한 루프를 돌리자. flag 가 변하면 루프를 끝내자. lifespan 내에서 처리하니까 되긴 한다. 즉, 적-절한 옵션이 함께 한다면 가능은 하다!

flag 를 fastapi 에서 조회/변조 할 수 있는지 시험해보자. 이건 한번에 달성 할 수 없었다. 값을 변경하려는 시도는 잘 작동하지 않았다. ASGI 스펙을 다시 읽어보니 state 는 전적으로 application 의 제어에 있으며, server는 얕은 복사만 수행한다고 되어 있다. 그렇다면 dict of dict 하면 변조가 된다는 얘기렸다? -> 된다.

오늘의 결론

  • job runner 를 잘 다듬으면 이벤트 루프로 변환 시킬 수 있다.
  • 그렇게 만든 새 루프와 메인 루틴의 통신은 변수로 가능하다. (같은 프로세스의 같은 스레드 이므로)
  • 한번 끄면 다시 켜는건 어려울 것 같다. pause 가 아니라 kill 이라 이름지어야 하겠다.
  • 1 min read

2024 05 06

  • 2024-05-05 부터 시작해, 아이디어를 모두 정리하여 README.md 를 작성.
  • 전부터 생각만 하던, FastAPI에 airflow webserver 를 mount 해서 기동하면 어떨까? 를 실현. 생각보다 매끄럽게 잘 되서 놀랐다.
  • 당장의 목표는 airflow standalone 명령의 재현으로, 각 컴포넌트는 개별 subprocess 로 만들어질 듯 싶다.
  • 가급적 asyncio 만 쓰고 싶었는데, polling 처리가 너무 어렵다. sh 를 쓴다.
  • 그러나 그렇게 되면, sqlite 의 thread 접근을 따져가면서 사용해야 하므로 불편하다. 궁국적으로는 scheduler, worker, triggerer, dagparser 등을 asyncio 안에 녹여내어야 할 것이다.
  • 그러면, uvicorn 워커가 늘어나면 어떻게 됨? 이란 질문이 생길 수 있는데, 늘리는데에 있어서 장점이 없다. 라고 할 수 있겠다.
  • 어떻게 작동시키는데는 성공했고, WSL1 의 제약 때문에 (마치 NFS 에 있는 sqlite 를 쓰는데 문제가 있듯) sqlite 가 동시 접근을 당하자 프로세스가 죽어버렸다.
  • 이 문제 해결을 위해서라도 스케쥴링 로직을 async 하게 고쳐야 한다.
  • triggerer 는 async 를 쓸테니, async 한 db 접근자가 있을 것이라 가정하고 코드를 따라가 보았다.
    • 그러지 않았다.
    • job runner 가 이벤트 루프를 위한 Tread 를 하나 만든다. 그 안에서 asyncio.run() 한다.
    • job runner의 메인 스레드와 asyncio 를 위한 스레드는 프로세스 분기가 아니라 스레드 분기이므로, 값을 주고 받을 수 있다.
    • job runner의 메인 스레드가 루프를 돌아 db에서 값을 만들어 열심히 deque 를 채우면, asyncio 루프에서 열심히 소비한다.
  • 즉, 이 구조는 일단 멀티 스레드이다. WSL의 sqlite 동시성 극복으로는 쓸 수 없다.
  • 네이티브하게 async 하고 싶다면, sqlalchemy 부터 어떻게 해야 한다.
  • sqlalchemy==1.4.52 로, async API 가 존재는 한다. 그러나 이게 쓸 수 있다는 얘기는 아니다...