이산 사건 시스템 명세 DEVS(Discrete Event System Specification) 알고리즘

 

DEVS는 복잡한 시스템을 모델링하고 시뮬레이션하기 위한 강력하고 체계적인 프레임워크를 일컫는데, 의 가장 큰 특징은 계층적이고 모듈화(hierarchical and modular) 된 접근 방식으로 대규모 시스템의 개발 및 검증에 용이합니다.

 

핵심 구성 요소

Atomic Model - 시스템의 가장 기본적인 단위, 더 이상 나눌 수 없는 최소 단위의 동작을 정의

• 상태 집합
• 입력 사건 집합
• 출력 사건 집합
• 내부 상태 변환 함수
• 외부 상태 변환 함수
• 동시 상태 변환 함수
• 출력 함수
• 시간 진행 함수

Coupled Model - 여러 개의 아토믹 모델 또는 다른 커플드 모델들을 연결하여 더 큰 시스템을 구성하는 역할

• 구성 요소 집합 (D)
• 외부 입력 연결 (EIC)
• 내부 연결 (IC)
• 외부 출력 연결 (EOC)

DEVS 작동원리

1. 초기화: 모든 모델과 내부 사건 발생 시간 초기 세팅 및 Queue에 따른 초기 시작 시간 세팅
2. 시간 동기화
   1. 자신의 자식 모델들로부터 다음 사건 발생 시간을 보고 받음.
   2. 보고 받은 시간 중 가장 이른 시간을 찾아 이 시간을 자신의 다음 사건 발생 시간으로 코디네이터(Coordinator)에게 보고.
   3. 위의 과정을 반복하여 최상위 Coordinator는 전체 시스템에서 가장 먼저 발생할 사건의 시간, 즉 임박시간을 결졍
3. 사건 처리
   1. 최상위 코디네이터는 결정된 임박 시간을 모든 하위 모델에 전파.
   2. 자신의 다음 사건 발생 시간이 임박 시간과 일치하는 모델들을 임박 모델(Imminent model)로 간주.
   3. 출력 생성: 임박 모델들은 먼저 출력 함수를 실행하여 출력 생성.
   4. 출력 라우팅: 코디네이터는 생성된 출력을 커플드 모델에 정의된 연결에 따라 다른 모델들의 입력으로 전달.
   5. 상태 변환:
      1. 출력을 생성한 임박 모델 중 외부 입력을 받지 않은 모델 - 내부 상태 변환을 수행. 내부 상태 변환은 외부로부터 간섭이 없는 내부 스케줄에 따라 시간이 지나갔을 때의 상태를 말합니다.
      2. 외부 입력을 받은 모델은 외부 상태 변환을 수행. 외부 변환 상태는 외부에 이벤트에 영향을 받아 시스템의 상태가 강제로 바뀌는 상태를 말합니다.
      3. 출력을 생성하면서 동시에 외부 입력도 받은 임박 모델은 동시 상태 변환 수행. 내부 상태 변환이 일어나야 할 정확한 시각이나 끝나기 전에 외부 입력 사건이 동시에 도착 했을 때 입니다.
4. 다음 사건 시간 계산: 상태 변환을 수향한 모델들은 시간 진행 함수를 통해 자신의 다음 내부 사건 발생 시간을 다시 계산하여 상위 코디네이터에게 보고.
5. 반복: 시뮬레이션 종료 조건이 만족 될 때까지 2~4 단계 과정 반복

 

여기서 중요한것은 각 모델들의 coupling을 최소화 하고 각 모델의 사건 발생을 동일한 사건이라도 독립적으로 수행이 가능해야 한다.

 

의존성 분리에 효율적인 방법론은 Port 기반 통신을 활용하는 것 입니다. 인스턴스, 혹은 reference pointer로 서로 직접 호출하거나 참조하지 않는다는 방법론을 활용합니다. 이를 준수하기 위해서는 모든 모델은 정해진 입력 포트로만 데이터를 받고 출력 포트로만 데이터를 내보냅니다. 만약 모델간의 연결과 상호적 연결은 Coupled Model을 따로 만들어 전적으로 결정하고 관리(Mediator Pattern)합니다.

 

추가로 인터페이스 기반으로 표준화와 추상화를 적극 활용합니다. 이렇게 코드를 작성하면 유지보수 비용을 줄이면서 기능확장이 가능 해집니다.

 

이벤트 버스 또는 브로커 모델을 도입합니다. '발행'과 '구독'방식으로 의존성을 원천 차단하는 것 입니다.

 

마지막으로 모든 이벤트를 팩토리 패턴을 적용하여 생성되게 만들면 모델마다 독립된 이벤트를 사용할 수 있게 됩니다.

import time

# 1. 이벤트 클래스 정의 (실제로는 여러 종류가 될 수 있음)
# ----------------------------------------------------------------------
class Event:
    """모든 이벤트의 기반이 되는 간단한 클래스"""
    def __init__(self, name, timestamp, payload=None):
        self.name = name
        self.timestamp = timestamp
        self.payload = payload if payload is not None else {}

    def __repr__(self):
        return f"Event(name='{self.name}', payload={self.payload})"

# 2. 이벤트 팩토리 정의
# ----------------------------------------------------------------------
class EventFactory:
    """
    이벤트 생성을 전담하는 팩토리.
    모델은 이 팩토리에만 의존하며, 구체적인 Event 클래스를 알 필요가 없다.
    """
    def create_event(self, event_type: str, **kwargs):
        now = time.time()
        if event_type == "DATA_PRODUCED":
            # 데이터 생성 이벤트 로직
            if 'data' not in kwargs:
                raise ValueError("DATA_PRODUCED 이벤트는 'data'가 필요합니다.")
            return Event("DATA_PRODUCED", now, payload={'data': kwargs['data']})
            
        elif event_type == "PROCESSING_COMPLETE":
            # 처리 완료 이벤트 로직
            if 'result' not in kwargs:
                raise ValueError("PROCESSING_COMPLETE 이벤트는 'result'가 필요합니다.")
            return Event("PROCESSING_COMPLETE", now, payload={'result': kwargs['result']})
            
        else:
            # 기본 이벤트 또는 에러 처리
            return Event("UNKNOWN", now, payload={'error': f"알 수 없는 이벤트 타입: {event_type}"})

# 3. DEVS의 아토믹 모델을 흉내 낸 클래스들
# ----------------------------------------------------------------------
class Producer:
    """데이터를 생성하는 모델"""
    def __init__(self, event_factory: EventFactory, item_id: int):
        self.factory = event_factory # 팩토리를 외부에서 주입받음 (Dependency Injection)
        self.item_id = item_id
        print(f"Producer 생성됨. 구체적인 Event 클래스를 모름.")

    def generate_data(self):
        print(f"\n[Producer] ID {self.item_id} 데이터 생성 중...")
        # 모델은 'DATA_PRODUCED' 라는 문자열만 알면 될 뿐,
        # Event 클래스의 구조나 생성 방법을 전혀 알 필요가 없다.
        new_event = self.factory.create_event("DATA_PRODUCED", data=f"Item-{self.item_id}")
        self.item_id += 1
        return new_event

class Consumer:
    """데이터를 소비(처리)하는 모델"""
    def __init__(self, event_factory: EventFactory):
        self.factory = event_factory # 팩토리를 외부에서 주입받음
        print(f"Consumer 생성됨. 구체적인 Event 클래스를 모름.")
        
    def process_data(self, input_event: Event):
        print(f"[Consumer] 입력 받음: {input_event}")
        if input_event.name == "DATA_PRODUCED":
            data = input_event.payload.get('data', '없음')
            result = f"'{data}' 처리 완료"
            print(f"[Consumer] 처리 결과: {result}")
            # 처리 완료 후, 결과를 담은 새로운 이벤트를 팩토리를 통해 생성
            return self.factory.create_event("PROCESSING_COMPLETE", result=result)

# 4. 간단한 시뮬레이션 실행
# ----------------------------------------------------------------------
if __name__ == "__main__":
    # 1. 팩토리를 먼저 생성
    event_factory = EventFactory()

    # 2. 모델을 생성할 때, 팩토리 인스턴스를 '주입'해 줌
    producer = Producer(event_factory, item_id=101)
    consumer = Consumer(event_factory)

    print("\n--- 시뮬레이션 시작 ---")

    # 3. 생산자가 이벤트를 생성 (팩토리 사용)
    produced_event = producer.generate_data()
    print(f"Producer가 생성한 이벤트: {produced_event}")

    # 4. 소비자가 이벤트를 받아 처리하고, 새로운 이벤트를 생성 (팩토리 사용)
    processed_event = consumer.process_data(produced_event)
    print(f"Consumer가 생성한 이벤트: {processed_event}")
    
    print("--- 시뮬레이션 종료 ---")