Behavioral Programming 라이브러리를 코틀린에서 구현해보자 feat Channel

Clojureverse 는 지식의 원천

Clojureverse를 돌아다니다가 Behavioral Programming이라는 작은 패러다임에 관한 포스팅을 발견했다. 이 글을 보고 Behavioral Programming에 대해 좀 더 찾아봤는데, 주류는 아니지만 개인적으로 흥미가 생겼다.

BThread라는 독립적인 Behavioral을 작성하고, BThread들이 조합되면서 하나의 BProgram이 완성되는 구조다. 복잡하게 얽히고 설킨 소프트웨어의 레고 블록들을 sync라는 공통된 이벤트 기반 시스템으로 관리하여 완성시킨다는 이론이다.

요런식으로 동작한다

절차 지향부터 함수형에 이르기까지 모든 소프트웨어가 공통적으로 추구하는 방향성은 모듈화와 그로 인한 레고처럼 조립하는 시스템이다. 
Behavioral Programming의 동작 원리가 바로 이러한 이상향을 지향하고 있다고 생각했다.

실질적으로 이런 구조가 대세 프로그래밍이 될지는 모르겠지만, 아무튼 이런 철학의 패러다임은 환영.

 

개발에 이유가 어딨어

sync 시에 Suspend 되는 동작원리와 clojure 의 go macro 로 구현된 예제를 보고 있노나리,

 

엥 이거 완전 Kotlin Coroutine 이랑 Channel 인데?

 

라는 생각이 들어서 개발에 돌입

public class RequestFiveAddColdEvents extends BThread {
  public void runBThread() {
    for (int i = 1; i <= 5; i++) {
      bp.bSync(addCold, none, none);
    }
  }
}

위는 자바라이브러리인 BPJ 예시중 하나로 개념은 다음과 같다.

• BThread 
  어플리케이션 논리, 독립적인 하나의 Behavior 를 다룸
• Sync
  BThread 간에 동기화되는 지점으로, 모든 BThread가 sync를 호출할 때까지 대기. 모든 BThread가 sync를 호출하면 요청사항들(Event들) 중에서 적절히 선택한 다음, 받을 만한 BThread들에게 전송
• Sync.Request
  Event를 요청하고 해당 이벤트를 기다림
• Sync.Wait
  Event를 요청하지 않고 해당 이벤트를 기다림
• Sync.Block
  해당 Event의 요청을 막음

이렇게 BThread 와 Sync 가 동작의 핵심 개념이고, 이를 구성하여 만든 BThread 를 조합하여 하나의 프로그램을 만든다.

요런식으로 동작한다.

 

예제를 보고 대충 어떤 식으로 돌아가는지 확인했으니, 대략적인 DSL부터 설계해보자.

필자는 대략적인 DSL 모양부터 Top-Down 설계하는걸 좋아하는데, 해당 라이브러리의 추상화된 최종 형태부터 설계해 나가면서 구현으로 확장하는 방법을 통하여 실제로 필요한 Part 를 정립하고, 최종적인 코드가 원하는 대로 이쁘게나오는것을 최대한 유지해 나갈수 있기 때문이다. DSL 도 일종의 언어 설계이므로 언어 문법부터 생각한다고 하면 편하다.

내 스타일을 좀더 구체적으로 작성하자면

1. 언어 무관하게 이런식으로 짜고 싶은 DSL 설계
2. 언어 Spec 에 맞게 가능한 DSL 로 수정
3. DSL 을 유지 시킬수 있는 인터프리터 혹은 언어 Spec에 맞게 클래스 설계 -> DSL 구현 여부 확인
4. 커스텀 타입이나 함수도 별로 없이 일단 구현 → 동작 구현 여부 확인
5. DSL 도 구현 가능하고, 동작여부도 확인했으니, DSL 과 구현을 적절히 수정하여 합침
6. 완성!

 

암튼 BPJ의 예시 프로그램을 보고 아래처럼 짜면 Kotlin에 맞게 이쁠 것 같다는 생각을 하면서 먼저 DSL부터 설계했다.

참고한 Java 코드

//Add hot
public class RequestFiveAddHotEvents extends BThread {
  public void runBThread() {
    for (int i = 1; i <= 5; i++) {
      bp.bSync(addHot, none, none);
    }
  }
}
  
//Add cold
public class RequestFiveAddColdEvents extends BThread {
  public void runBThread() {
    for (int i = 1; i <= 5; i++) {
      bp.bSync(addCold, none, none);
    }
  }
}
  
//Interleave
public class Interleave extends BThread {
  public void runBThread() {
    while (true) {
      bp.bSync(none, addHot, addCold);
      bp.bSync(none, addCold, addHot);
    }
  }
}
  
//Display events
public class DisplayEvents extends BThread {
  public void runBThread() throws BPJException {
    while (true) {
      bp.bSync(none, all, none);
      System.out.println("turned water tap: " + bp.lastEvent);
    }
  }
}
  
//Main method
public class Main {
  public static void main(String[] args) {
    BProgram bp = new BProgram();
    bp.add(new AddHotThreeTimes(), 1.0);
    bp.add(new DisplayEvents(), 2.0);
    bp.add(new AddColdThreeTimes(), 3.0);
    bp.add(new Interleave(), 4.0);
    bp.startAll();
  }
}

 

작성한 DSL

// Define the Hot Water BThread
val hotWater = bThread(name = "Hot Water") {
    for (i in 1..3) {
        sync(request = setOf(WaterEvent.ADD_HOT), waitFor = None, blockEvent = None)
    }
}

// Define the Cold Water BThread
val coldWater = bThread(name = "Cold Water") {
    for (i in 1..3) {
        sync(request = setOf(WaterEvent.ADD_COLD))
    }
}

// Define the Interleave BThread
val interleave = bThread(name = "Interleave") {
    for (i in 1..3) { // Limit interleave to 5 times
        sync(waitFor = setOf(WaterEvent.ADD_HOT), blockEvent = setOf(WaterEvent.ADD_COLD))
        sync(waitFor = setOf(WaterEvent.ADD_COLD), blockEvent = setOf(WaterEvent.ADD_HOT))
    }
}


// Define the Display BThread
val display = bThread(name = "Display") {
    while(true) {
        sync(waitFor = All)
        println("[${this.name}] turned water tap: $lastEvent")
    }
}

// Create the BProgram with all BThreads
val program = bProgram(
    hotWater,
    coldWater,
    interleave,
    display
)

program.enableDebug()
program.runAllBThreads()

 

이정도면 개인적으로 만족할정도로 이쁘다. 이제 라이브러리를 구현하자.

요구사항을 보아하니, 대충 이런것들이 필요하다

• BThread 는 동작하다가, Sync 를 만나면 다른 BThread 가 Sync 를 만날때 까지 Suspend 해야한다
  -> Suspend? 코루틴을 이용할수 있을것 같다.
• BProgram 은 BThread 를 관리하면서, Event 를 수집, 적절한 다음 Event 를 BThread 에게 전달 해줘야 한다
  -> Event 를 전달? Channel 이 딱이다.

class RegisteredBThread(
    val name: String,
    val behavior: suspend RegisteredBThread.() -> Unit,
    val priority: Double,
    private val syncChannel: Channel<SyncChannelMessage>
) {
	/* skip */
    
    suspend fun sync(
        request: Set<Event> = None,
        waitFor: Set<Event> = None,
        blockEvent: Set<Event> = None
    ) {
        val syncPoint = SyncChannelMessage.SyncPoint(
            this,
            request,
            waitFor,
            blockEvent
        )
        syncChannel.send(syncPoint)
        val event = eventChannel.receive()
        lastEvent = event
    }
}

 

BThread는 sync를 만나면 Channel을 통해 BProgram에게 넘겨주고, 다른 Channel에서 Event를 receive할 때까지 Block된다. 코루틴의 채널이니까 알아서 Suspend 상태로 변경될 것이다.

(RegisterdBThread인 이유는 priority를 나중에 지정하기 위해서 BThread와 별도로 나눔)

이제 이 BThread를 관리하는 BProgram을 작성하면 끝

BProgram이 해줘야 하는 일들은 다음과 같다:

1. 모든 BThread 가 sync 상태인지 확인한다
2. 모든 BThread 가 sync 상태면, 받은 요구사항을 분석한다
   Block 되지 않은 우선순위가 가장 높은 이벤트를 찾는다 (Next Event)
3. Next Event 를 Wait 하고 있는 BThread 에게 보내준다 (Request 한 BThread 포함)
4. 모든 BThread 가 busy 하지 않은 (즉 Suspend or Terminate 되지 않은) 상태 인데, Event 를 더이상 줄수없으면 프로그램 종료 (근데 이건 내가 다른 코드 몇번 보고 한 추측이라 Offical Spec 은 아닐듯 싶음..)

class BProgram {
    private val bThreads = mutableListOf<RegisteredBThread>()
    private val syncPoints = ConcurrentHashMap<RegisteredBThread, SyncChannelMessage.SyncPoint>()
    private val syncChannel = Channel<SyncChannelMessage>(Channel.UNLIMITED)
    private val activeThreads = AtomicInteger(0)

    /* SKIP */
    
	val executeBProgram: suspend CoroutineScope.() -> Unit = {
        initializeBThreads()

        while (isActive) {
            handleSyncPoint()
        }

        cancel(BCompletionException())
    }

    private val isActive get() = activeThreads.get() > 0
    private val needsSync get() = activeThreads.get() == 0

    private suspend fun handleSyncPoint() {
        when (val syncPoint = syncChannel.receive()) {
            is SyncChannelMessage.SyncPoint -> {
                syncPoints[syncPoint.sender] = syncPoint
                activeThreads.decrementAndGet()

                if (needsSync) {
                    val nextEvent = selectNextEvent()

                    val threadsToNotify = determineThreadsToNotify(nextEvent)
			        notifyBThreads(threadsToNotify, nextEvent)
                }
            }
            is SyncChannelMessage.Terminate -> {
                syncPoints.remove(syncPoint.sender)
                activeThreads.decrementAndGet()
            }
        }
    }
}

 

ssyncChannel에서 Terminate 메시지를 받는 이유는 원래 OnCancelComplete을 통해 Job 종료를 판단하려 했더니, 동시적으로 receive가 일어나 wait가 무한 대기되는 상태가 발생했다. 일관된 상태 관리를 위해 Terminate 메시지를 BThread에서 직접 받도록 변경했다.

처음에는 Channel을 통한 Receive와 OnCancelComplete 두 부분에서 Modify가 일어나 Thread Safe한 자료구조를 사용했는데, 일관된 Channel Receive를 통하므로 일반 변수를 사용해도 될 것 같긴 하지만… 어쨌든 여러 코루틴을 돌리는 부분이므로 Thread Safe 자료구조를 그대로 사용했다. 굳이 이유를 뽑자면 dynamic BThread 추가 삭제를 위해서..?

 

코드를 변경하면서  WaterDrop / SimpleTicTacToe 두 예제를 계속 테스트 하였다.

 

아무튼 완성된 레포지토리는 여기 Lee-WonJun/bpkc: BPKC

GitHub - Lee-WonJun/bpkc: BPKC (Behavioral Programming for Kotlin through Channels) is a lightweight approach that emphasizes Be

 

BP 의 미래는?

많은 크고 작은 패러다임이 범람하고 있는데, 모든 패러다임은 결국 모듈화다. 독립적인 개체를 다른 곳 신경도 안 쓰고 개발해도 이게 이쁘게 만들어진 레고처럼 잘 동작하느냐의 문제다. 그런 측면에서 BP는 아주 우수한 이상향을 추구하고 있다고 생각한다.

다만, 직관적으로 이해되기 힘든 Sync의 구조와 BThread의 동작에 따른 필수 불가결한 보일러플레이트 코드들, 그리고 BThread를 얼마나 잘게 쪼개는지 등이 걸림돌이다.  그리고 내 설계문제인가.. 싶은 대체 올바른 Priority 컨트롤 을 어떻게 해야하는지

현재는 가장 심플한 형태의 BP 구조만 구현했는데, 실제로 BPJ의 코드를 살펴보면 프로그램 동작 도중 수정 가능한 dynamic BThread와 BProgram의 수많은 Run Mode 등이 있다. 
모든 소프트웨어는 시간이 갈수록 추가되는 Edge case로 인해 복잡성이 심해지기 때문에 요구사항이 복잡해질수록 패러다임의 난이도는 더욱 올라갈 것이다.

그렇기 때문에 BP는 일반적인 패러다임에 안착하기는 개인적으로 좀 힘들 것 같지만, 이러한 개념이나 스타일은 좀 더 Simplify된 이후에 사용할 수 있을 것 같다. 
어찌 보면 Actor 모델이나 CSP에 대한 Management System 중 하나에 불과한 이름을 붙인 것이라는 느낌도 살짝 들고.

 

 

참고자료들

• Behavioral Programming – Communications of the ACM 
• The BPJ Library - BP Wiki (weizmann.ac.il)
• Behavioral Programming: Aligning Code with How We Think - Luca Matteis - DDD Europe 2020 - YouTube 
  (BP 에 관심 있다면 해당 영상 추천)
• EricDw/BPK-4-DROID: Behavioral Programming written in Kotlin and optimized for Android (github.com)
• lmatteis/behavioral: Behavioral Programming for JavaScript (github.com)
• Behavioral Programming in Clojure - Thomas Cothran, Software Developer
• Crash Course: Behavioral Programming in Clojure with core.async | by Eugenii Shevchenko | Sep, 2024 | Medium