Pech — это Managed Kernel которая следует концептам Mach 3.0. В 2025 году безопасность важнее ручного управления тактами. Я использую высокоуровневый рантайм для создания математически безопасной среды, где баги памяти устранены на уровне архитектуры. Это то, к чему сейчас стремятся проекты вроде Microsoft Singularity или современные ОС на Rust.

В этой статье я попытаюсь как можно больше перечислить ошибок pyRTOS которые были исправлены в Pech.

Вступление

Всем привет, я — парень который делает своё ядро на MicroPython, и у него это получается.

Изначально я хотел написать это ядро похожей на pyRTOS, но, мне не нравится то, как там всё работает.

Для обычного Python'а это нормальное решение но для MicroPython это плохо.

Да, если вы не поняли ещё, то я создатель «Печки».

Механика переключений: await против yield

Давайте посмотрим на требования от задачи у pyRTOS:

All code before the first yield is considered setup code, and any yield value is ignored.

На русском: "Весь код до первого yield считается кодом настройки, и любое возвращаемое значение этого yield игнорируется".

Во первых: почему нам запрещают делать код настройки где мы хотим?

Во вторых: почему после этого кода надо делать yield?

Ответ прост: так планировщик сможет запустить новую задачу.

А у нас, с asyncio можно использовать функцию create_task() и тогда задача просто не запустится сразу.

Дальше:

UFunctions... must yield False if the condition is not met and True if it is.

Перевод: "Пользовательские функции (UFunctions)... должны возвращать (yield) False, если условие не выполнено, и True, если оно выполнено".

Нет же! У нас нечего не надо возвращать: просто пишешь такой отрезок кода:

await asyncio.sleep(SLEEP_TIME)

SLEEP_TIME определять не нужно. Оно по умолчанию стоит 0.020 секунд (20 миллисекунд).

Закрываться (объявлять об этом) тоже не нужно! Просто нечего не делай в конце задачи: не цикла с pass, не ещё чего-то.

Просто ждешь пока ядро само определит что ты закрыт.

И напоследок:

All tasks must be written as infinite generators.

Перевод: "Все задачи должны быть написаны как бесконечные генераторы".

У нас даже не надо свою функцию задачи объявлять. Просто пишешь в строке свой код, добавляешь с помощью create_proc куда только приоритет указать надо (и код задачи) и запускаешь boot и scheduler.

IPC у pyRTOS против IPC у Pech

Сначала скажу ту часть, где pyRTOS лучше а потом перейду к моей части доминирования.

pyRTOS лучше Pech только в том, что сообщение в pyRTOS более структурированы.

Но это всё.

Посмотрим на канал (в pyRTOS это очередь) в pyRTOS:

class MessageQueue(object):
	def __init__(self, capacity=10):
		self.capacity = capacity
		self.buffer = []

	# This is a blocking condition
	def send(self, msg):
		sent = False

		while True:
			if sent:
				yield True
			elif len(self.buffer) < self.capacity:
				self.buffer.append(msg)
				yield True
			else:
				yield False

	def nb_send(self, msg):
		if len(self.buffer) < self.capacity:
			self.buffer.append(msg)
			return True
		else:
			return False


	# This is a blocking condition.
	# out_buffer should be a list
	def recv(self, out_buffer):
		received = False
		while True:
			if received:
				yield True
			elif len(self.buffer) > 0:
				received = True
				out_buffer.append(self.buffer.pop(0))
				yield True
			else:
				yield False

	
	def nb_recv(self):
		if len(self.buffer) > 0:
			return self.buffer.pop(0)
		else:
			return None

Этот код чисто использует busy waiting. То есть не дает процессору хотя бы несколько наносекунд отдохнуть. Самые лучшие функции (только они должны остаться) это nb_recv и nb_send. Потому что если recv ещё как-то прокатывает то вот send нет.

Эти две функции постоянно шлют планировщику то False, то ещё что-то.

В итоге процессор всё греется и греется (и ещё это по памяти затратно).

У нас же очередь реализована так:

import uasyncio as asyncio

class Queue:
    def __init__(self):
        self._queue = []
        self._ev = asyncio.Event()

    async def put(self, val):
        if len(self._queue) >= 6:
            print("[IPC]: Can't write in channel.")
            return 0
        self._queue.append(val)
        self._ev.set()

    async def get(self):
        while not self._queue:
            await self._ev.wait()
            self._ev.clear()
        return self._queue.pop(0)

Да, этот код в какой-то степени вызывает race condition, но зато он безопаснее и быстрее.

И также если в pyRTOS только очередь сообщений, то в Pech есть и канал:

class Pipe:
    def __init__(self):
        self.channels = [Queue(), Queue()]

    async def write(self, side, data):
        await self.channels[1 - side].put(data)

    async def read(self, side):
        return await self.channels[side].get()

Это полнодуплексный канал (2-ух сторонний иными словами).

Конечно связка Queue и Pipe лучше чем MessageQueue в pyRTOS но тут ещё больше треша:

def deliver_messages(messages, tasks):
	for message in messages:
		if type(message.target) == pyRTOS.Task:
			message.target.deliver(message)
		else:
			targets = filter(lambda t: message.target == t.name, tasks)
			try:
				next(targets).deliver(message)
			except StopIteration:
				pass

Зачем мы перебираем столько информации? Не проще ли явно указывать откуда надо взять?

Таким способом процессор вдвойне сжигается так ещё и это медленно и затратно по памяти.

Здесь O(1) явно победил O(n). Пока pyRTOS тратит такты на перебор списка задач, Pech просто доставляет данные по адресу. Это разница между "скриптом для мигания светодиодом" и фундаментом для операционной системы.

Безопасность и стабильность: кто лучше

Почитав документацию pyRTOS можно увидеть такую фразу:

The error handling philosophy of pyRTOS is: Write good code.

Перевод "Философия обработки ошибок в pyRTOS такова: пишите хороший код".

Вернемся к второму концепту нашего ядра:

Безопасность и стабильность - выше всего: ядро должно любым способ не дать сделать процессу что-то плохое.

Так и получается.

Посмотрим на ко... А стоп, его там нет.

Тогда скажу сразу: ошибка процесса - вылет все pyRTOS.

У нас в ядре есть два типа обработки ошибки:

1. Предсказуемая ошибка: она решается через if-elif.

2. Непредсказуемая ошибка: она решается через try-except.

Лишь в единичных случаях можно добиться ошибки ядра.

А как же безопасность?

Так тут даже слова о ней нет.

Процесс захотел импортировать модуль os и написать os.system('rm -rf /')?

"Да пожалуйста!", скажет pyRTOS.

С одной стороны это нормально.

А с другой же стороны это плохо.

Но возвращаясь к концепту 2 в Pech мы решили сделать такую песочницу:

ctx = {
  "pid": pid, "send": send, "recv": recv, "asyncio": asyncio,
  "os": None, "eval": None, "SLEEP_TIME": 0.020,
  "create_pipe": create_pipe,
  "connect": lambda local_id, side: connect(pid, local_id, create_pipe(), side),
  "machine": None, "gc": None, "micropython": None,
  "__import__": None, "importlib": None, "exec": None,
  "timer": Timer
}

Теперь за место __builtins__ в exec будет стоять наш контекст (та же песочница).

Почему os, eval и другие = None? Эти фичи считаются опасными и с помощью их можно угробить ядро (соответствие 3 концепту моего ядра).

А сам концепт таков:

Всё, что может помешать безопасности (к примеру опасные библиотеки) должно быть обнулено.

Планировщик Pech vs планировщик pyRTOS

Планировщик pyRTOS — это просто "глупый" цикл. Он реализует модель активного опроса (Polling): на каждом круге он обязан вызвать каждую задачу через next(), чтобы проверить, не соизволит ли она проснуться. Если у вас 50 задач спят, процессор всё равно переберет их все 50 раз за один цикл ядра.

Что же я буду говорить, я покажу код этого планировщика:

import pyRTOS


def default_scheduler(tasks):
		messages = []
		running_task = None

		for task in tasks:
			if task.state == pyRTOS.READY:
				if running_task == None:
					running_task = task
			elif task.state == pyRTOS.BLOCKED:
				if True in map(lambda x: next(x), task.ready_conditions):
					task.state = pyRTOS.READY
					task.ready_conditions = []
					if running_task == None:
						running_task = task
			elif task.state == pyRTOS.RUNNING:
				if (running_task == None) or \
				   (task.priority <= running_task.priority):
					running_task = task
				else:
					task.state = pyRTOS.READY


		if running_task:
			running_task.state = pyRTOS.RUNNING

			try:
				messages = running_task.run_next()
			except StopIteration:
				tasks.remove(running_task)

		return messages

Такой метод как и IPC у pyRTOS будет медленно греть процессор и в итоге он сгорит если за столпиться на бесконечном цикле busy waiting'а.

А вот наша реализация планировщика задач:

async def scheduler():
    print("[KERNEL]: Scheduler started.")
    
    for pid in sorted(procs.keys(), key=lambda p: procs[p].prio, reverse=True):
        procs[pid].state = RUNNING
        asyncio.create_task(run_proc(pid))
    
    while True:
        for pid, p in procs.items():
            if p.server and p.state == CLOSED:
                p.state = RUNNING
                asyncio.create_task(run_proc(pid))
        
        active_procs = [p for p in procs.values() if p.state == RUNNING]
        if not active_procs:
            print("[KERNEL]: All processes finished. System idle.")
        
        await asyncio.sleep(0.020)

Во первых, как говорилось ранее, этот планировщик не требует передавать контроль ему же от задачи.

Во вторых спящая задача вообще не занимает время планировщика.

В третьих, если сервер вылетит то планировщик попытается запустить его снова.

Ну... И всё вроде.

Ну тут ещё приоритетность получше работает.

Итог

Я составил таблицу по прототипу характеристика-pyrtos-pech-что лучше

Ребята, я сильно постарался.

Жду конструктивной критики и идей!

Удачи!

UPD: Ссылка на проект: SystemSoftware2/Pech