태스크와 우선순위 모델

상태 진행과 실행 기회를 구분합니다

센서 기록기의 상태 머신은 다음 동작을 정하지만, 그 동작이 CPU를 언제 받는지까지 정하지는 않습니다. 전송 작업 하나가 오래 실행되는 동안 수집 시각이 지나갈 수 있습니다. 이번 레슨에서는 수집과 전송을 각각 작업으로 적고, 실행 가능한 작업 중 무엇을 먼저 고를지 계산합니다. 결과로 작업 순서와 마감 누락 수를 얻어 우선순위를 바꾸는 것만으로 문제가 해결되는지 판단합니다.

이 모듈은 Python 3 표준 라이브러리 브라우저 실습과 C11·pthread 로컬 실습을 사용합니다. 외부 패키지·실물 보드·RTOS 설치는 필요하지 않습니다. 로컬 압축은 독립 폴더에 풀고 Makefile 위치에서 make test를 실행합니다. gcc 명령은 맥에서 Apple Clang일 수 있으므로 gcc --version을 기록합니다. Python 실습은 표준 입력 한 건을 받아 표준 출력 한 줄을 반환하며 디버그 출력은 제출 전에 제거합니다.

태스크는 실행 흐름과 자신이 보관하는 상태를 가진 작업 단위입니다. 센서 요청 상태와 최근 필터 창은 수집 담당이 소유하고, UART·SPI 출력은 전송 담당이 소유하도록 생각할 수 있습니다. 여러 태스크를 선언했다고 동시에 여러 CPU에서 실행되는 것은 아닙니다. 여기서는 CPU 하나의 가상 시간을 사용하며, 실제 호스트 스레드 우선순위를 설정하지 않습니다. 태스크를 나누는 목적은 역할·대기·실행 시간을 드러내는 것입니다.

작업의 다섯 값을 기록합니다

각 작업은 name, release, cost, deadline, priority를 가집니다. name은 로그에 적는 식별자, release는 실행 준비가 되는 절대 시각, cost는 필요한 실행 시간, deadline은 완료해야 할 절대 시각, priority는 선택의 우선도입니다. 시각과 비용의 단위는 이 모델의 ms입니다. 예를 들어 COLLECT의 release가 1이고 deadline이 5라면 준비된 뒤 4ms 안에 완료되어야 합니다. deadline을 상대 제한 5ms로 다시 더하지 않습니다.

우선도는 숫자가 클수록 높도록 이번 모델에서 정합니다. 실제 RTOS마다 숫자 방향과 유효 범위가 다를 수 있으므로 이 규칙을 보드 설정에 그대로 옮기지 않습니다. 작업의 마감과 우선도는 서로 다른 값입니다. 마감이 가깝다고 코드가 자동으로 높은 우선도를 부여하지 않습니다. 실습은 고정 우선도 규칙을 검증하며, deadline이 가장 빠른 작업을 고르는 다른 알고리즘과 혼합하지 않습니다.

준비 큐에는 release가 현재 시각 이하인 미완료 작업만 들어갑니다. 아직 release가 오지 않은 작업은 이름이 수집이라고 해도 선택할 수 없습니다. 실행할 준비가 없으면 가장 이른 release로 시간을 이동합니다. 이 이동은 CPU가 빈 시간을 뜻하며 작업 비용이나 마감 누락으로 세지 않습니다. 준비 작업이 여러 개이면 priority 내림차순, release 오름차순, 입력 순서의 차례로 선택합니다. 동률 처리까지 고정해야 재실행 로그가 같습니다.

비선점 모델의 한계를 실험합니다

선택한 작업은 cost 전체를 실행하고 나서야 다음 작업을 고릅니다. 이것이 비선점 정책입니다. SEND가 시각 0에서 8ms 동안 실행하고 COLLECT가 시각 1에 준비되어도 SEND는 중간에 멈추지 않습니다. COLLECT의 우선도가 9이고 SEND가 1이어도 COLLECT는 시각 8부터 실행합니다. COLLECT의 비용이 2면 완료 시각은 10이며 마감 5를 놓칩니다. 높은 우선도는 다음 선택 시점을 바꿀 뿐 이미 실행 중인 작업을 끊지 않습니다.

두 작업 모두 시각 0에 준비시키면 높은 우선도 COLLECT가 먼저 실행됩니다. 같은 비용과 마감이어도 준비 시각의 변화로 결과가 달라집니다. 이 두 입력을 나란히 실행해 원인을 좁힙니다. 전송을 짧은 단위로 나누거나 선점을 도입하는 설계는 후속 선택입니다. 가상 비용을 줄여 통과한 모델이 실제 센서 응답이나 UART 지연을 해결했다고 주장하지 않습니다. 실제 장치에는 인터럽트 지연과 대기 시간도 존재합니다.

완료 시각이 deadline보다 클 때만 누락을 하나 셉니다. 같으면 마감을 지킨 것입니다. 누락 작업도 완료하며 모든 작업이 끝날 때의 now를 finish로 보고합니다. 이 실습은 관찰 종료 시각을 따로 두지 않고 유한한 작업 목록을 끝까지 처리합니다. 실제 주기 태스크를 분석하려면 반복 도착, 대기 시간, 최악 실행 시간과 관찰 구간을 더 정의해야 합니다. 짧은 입력에서 누락 0이라는 결과는 전체 시스템의 실시간 보장이 아닙니다.

입력부터 선택 규칙까지 구현합니다

입력은 jobs 배열을 가진 JSON 객체입니다. 작업 수는 0부터 100개이며 name은 중복 없는 ASCII 영숫자 문자열입니다. release·deadline·priority는 0부터 1000000까지의 정수이고 cost는 1부터 1000000까지의 정수입니다. deadline은 release 이상입니다. 소수·문자열 숫자·불리언은 정수 계약을 만족하지 않습니다. 파이썬에서는 bool이 int의 하위 타입이므로 type(value) is int로 이 실습의 계약을 엄격히 검사합니다.

출력 첫 항목은 쉼표로 연결한 실행 순서이며 이어서 missed=N finish=T를 적습니다. 빈 배열은 EMPTY missed=0 finish=0입니다. 입력 JSON 파싱 오류·필드 누락·범위 위반은 INVALID 한 줄입니다. 예외를 catch한 뒤 작업 일부를 처리한 로그를 출력하지 않습니다. 이름이 중복되면 어떤 작업의 마감인지 증거가 모호해지므로 실행 전에 전체 입력을 검사합니다. 추가 메타데이터 필드는 실행 정책에 사용하지 않습니다.

구현할 때 미완료 목록에 입력 인덱스를 붙여 보관합니다. 준비 작업 필터를 먼저 만들고 비어 있으면 다음 release로 이동합니다. 준비 작업이 있으면 정렬 키를 비교해 하나를 꺼내고 now에 cost를 더한 뒤 마감을 검사합니다. 모든 목록을 단순 반복해도 작업 100개 한도에서는 의도를 읽기 쉽습니다. 최적화보다 선택 규칙의 재현성을 먼저 확보합니다. 비용을 더하기 전에 마감을 검사하면 작업 시작 시각을 완료 시각으로 착각하게 됩니다.

결과를 읽고 설계를 설명합니다

KeyError는 필드 철자가 입력 계약과 다른 경우, JSONDecodeError는 따옴표나 쉼표가 깨진 경우를 먼저 의심합니다. 채점기에서는 이런 입력을 INVALID로 처리해야 하므로 traceback이 출력되는 구현은 완성되지 않은 것입니다. 정상 입력인데 순서가 틀리면 준비 필터와 우선도 부호를, missed만 틀리면 완료 시각과 등호 경계를 봅니다. finish가 준비 시각보다 작게 나오면 idle 구간에서 now를 이동하는 코드가 빠졌는지 확인합니다.

실험 메모에는 입력의 release·cost·deadline·priority와 출력 순서를 함께 적습니다. 우선도만 올린 실험, 수집 준비 시각을 당긴 실험을 비교하고 어떤 선택 시점이 변했는지 설명합니다. 상태 머신의 MEASURE와 TRANSMIT은 논리 상태이고 여기의 COLLECT와 SEND는 CPU 작업 이름이라는 구분도 적습니다. 운영체제 스케줄러와 부하 지표의 더 넓은 해석은 연결한 서재 장에서 읽습니다.

import json, sys

def integer(x, low, high):
    return type(x) is int and low <= x <= high

def schedule(jobs):
    pending=list(enumerate(jobs)); now=0; missed=0; order=[]
    while pending:
        ready=[x for x in pending if x[1]['release']<=now]
        if not ready:
            now=min(j['release'] for _,j in pending)
            continue
        item=min(ready,key=lambda x:(-x[1]['priority'],x[1]['release'],x[0]))
        pending.remove(item)
        job=item[1]; now+=job['cost']
        missed+=now>job['deadline']; order.append(job['name'])
    return order,missed,now
jobs=[{'name': 'SEND', 'release': 0, 'cost': 8, 'deadline': 20, 'priority': 1}, {'name': 'COLLECT', 'release': 1, 'cost': 2, 'deadline': 5, 'priority': 9}]
order,missed,now=schedule(jobs)
print(",".join(order),f"missed={missed}",f"finish={now}")

SEND,COLLECT missed=1 finish=10

import json, sys

def integer(x, low, high):
    return type(x) is int and low <= x <= high

def schedule(jobs):
    pending=list(enumerate(jobs)); now=0; missed=0; order=[]
    while pending:
        ready=[x for x in pending if x[1]['release']<=now]
        if not ready:
            now=min(j['release'] for _,j in pending)
            continue
        item=min(ready,key=lambda x:(-x[1]['priority'],x[1]['release'],x[0]))
        pending.remove(item)
        job=item[1]; now+=job['cost']
        missed+=now>job['deadline']; order.append(job['name'])
    return order,missed,now
jobs=[{'name': 'SEND', 'release': 0, 'cost': 8, 'deadline': 20, 'priority': 1}, {'name': 'COLLECT', 'release': 0, 'cost': 2, 'deadline': 5, 'priority': 9}]
order,missed,now=schedule(jobs)
print(",".join(order),f"missed={missed}",f"finish={now}")

COLLECT,SEND missed=0 finish=10

import json, sys

def integer(x, low, high):
    return type(x) is int and low <= x <= high

def schedule(jobs):
    pending=list(enumerate(jobs)); now=0; missed=0; order=[]
    while pending:
        ready=[x for x in pending if x[1]['release']<=now]
        if not ready:
            now=min(j['release'] for _,j in pending)
            continue
        item=min(ready,key=lambda x:(-x[1]['priority'],x[1]['release'],x[0]))
        pending.remove(item)
        job=item[1]; now+=job['cost']
        missed+=now>job['deadline']; order.append(job['name'])
    return order,missed,now
for jobs in [[{'name': 'A', 'release': 5, 'cost': 2, 'deadline': 7, 'priority': 1}], []]:
    order,missed,now=schedule(jobs)
    print(",".join(order) or "EMPTY",f"missed={missed}",f"finish={now}")

A missed=0 finish=7
EMPTY missed=0 finish=0

import json, sys

def integer(x, low, high):
    return type(x) is int and low <= x <= high

def schedule(jobs):
    pending=list(enumerate(jobs)); now=0; missed=0; order=[]
    while pending:
        ready=[x for x in pending if x[1]['release']<=now]
        if not ready:
            now=min(j['release'] for _,j in pending)
            continue
        item=min(ready,key=lambda x:(-x[1]['priority'],x[1]['release'],x[0]))
        pending.remove(item)
        job=item[1]; now+=job['cost']
        missed+=now>job['deadline']; order.append(job['name'])
    return order,missed,now
try:
    data=json.load(sys.stdin); jobs=data['jobs']
    if type(jobs) is not list or len(jobs)>100: raise ValueError()
    names=set()
    for j in jobs:
        if type(j) is not dict: raise ValueError()
        name=j['name']
        if type(name) is not str or not name.isascii() or not name.isalnum() or name in names: raise ValueError()
        names.add(name)
        for key in ['release','deadline','priority']:
            if not integer(j[key],0,1000000): raise ValueError()
        if not integer(j['cost'],1,1000000) or j['deadline']<j['release']: raise ValueError()
    order,missed,now=schedule(jobs)
    print(','.join(order) or 'EMPTY',f'missed={missed}',f'finish={now}')
except (ValueError,KeyError,TypeError):
    print('INVALID')