Skip to content
BranchesTags

Repository files navigation

smallOS

smallOS is a lightweight cooperative runtime for priority-oriented task management.

It is designed around three ideas:

  • write tasks with modern async / await syntax
  • keep scheduling policy owned by smallOS, not asyncio
  • stay portable enough to run on desktop Python today and MicroPython boards later

Status

The project is currently experimental but usable. The runtime core supports:

  • priority-based cooperative scheduling
  • task spawning, join, and join_all
  • signal-based wakeups
  • time-based sleeping
  • readiness-based socket/I/O waiting
  • generic TCP/TLS kernel hooks for higher-level protocols
  • dependency-free thread and asyncio execution adapters for user libraries
  • smallOS-native HTTP, Redis, MQTT, SSE, and WebSocket helper clients

Why smallOS?

Python's asyncio gives great syntax, but it also brings its own scheduler and event-loop policy. smallOS keeps the syntax while swapping in a custom runtime, so tasks can be scheduled with project-specific priority rules and a smaller portability surface.

That makes it a good fit for:

  • robotics or device-control projects
  • embedded experiments on MicroPython boards
  • custom runtimes where task priority matters
  • learning how coroutine scheduling works under the hood

Project Layout

Installation

Desktop development:

python3.10 -m venv .venv
source .venv/bin/activate
python -m pip install -e ".[dev]"

smallOS supports CPython 3.10 and newer. Python 3.6 through 3.9 are no longer supported. MicroPython compatibility is maintained separately because its language and standard-library support do not map directly to a CPython release number; checker-only imports are kept off embedded runtime paths.

Run the test suite:

python3 -m unittest discover -s tests -v

Run the tests with the same branch-coverage gate used by CI:

coverage run -m unittest discover -s tests -v
coverage report

Run the static type checker:

pyright

Build the wheel and source distribution:

python -m build

The GitHub Actions pipeline runs four gates: Pyright, unit tests across Python 3.10–3.13, branch coverage with a 60% floor, and distribution verification. Packaging runs only after the earlier gates pass, installs the built wheel, and smoke-tests it outside the source checkout.

The package ships a py.typed marker. Type coverage is being tightened by subsystem: configuration, awaitables, task lifecycle, scheduling, signals, platform kernels, and core utilities form the current checked boundary, while protocol clients, shells, and demos remain on the incremental typing backlog.

Quick Start

Minimal desktop runtime:

from SmallPackage.Kernel import Unix
from SmallPackage.SmallConfig import SmallOSConfig
from SmallPackage.SmallOS import SmallOS
from SmallPackage.SmallTask import SmallTask


async def hello(task):
    task.OS.print("hello from smallOS\n")
    await task.sleep(0.1)
    return "done"


config = SmallOSConfig.from_json_file("smallos.config.json")
runtime = SmallOS(config=config).setKernel(Unix())
runtime.setErrorHandler(
    lambda event: print(
        "[smallOS] task failure in {} (PID {}): {}".format(
            event["task_name"] or "unnamed task",
            event["task_id"],
            event["exception_repr"],
        )
    )
)
runtime.fork([SmallTask(2, hello, name="hello")])
runtime.startOS()

Runtime Error Handling

smallOS now supports a runtime-level error observer through runtime.setErrorHandler(handler, include_cancelled=False).

Use it when you want:

  • readable debug output for uncaught task failures
  • lightweight cleanup or bookkeeping at the runtime boundary
  • a single place to surface task errors without crashing the scheduler

The handler is synchronous and receives a failure-event dictionary after the task has been finalized. Current event fields include:

  • task_id
  • task_name
  • parent_id
  • exception
  • exception_type
  • exception_repr
  • is_cancelled
  • blocked_reason
  • waiting_signal
  • io_wait_mode
  • join_target_id
  • join_pending_ids
  • adapter_name
  • adapter_job_id
  • traceback_text

By default, TaskCancelledError does not trigger the handler. Pass include_cancelled=True if you want cancellation events too.

Example:

from SmallPackage.Kernel import Unix
from SmallPackage.SmallOS import SmallOS


def log_runtime_error(event):
    print(
        "[smallOS] task failure in {} (PID {}): {}".format(
            event["task_name"] or "unnamed task",
            event["task_id"],
            event["exception_repr"],
        )
    )
    if event["traceback_text"]:
        print(event["traceback_text"], end="")


runtime = SmallOS().setKernel(Unix())
runtime.setErrorHandler(log_runtime_error)

Closed or Invalid File Descriptors

Closed or invalid file descriptors used in wait_readable(...) or wait_writable(...) no longer crash the whole scheduler through the platform poll/select layer.

Instead:

  • the kernel validates the watched object before polling
  • the waiting task receives a normal exception such as ValueError
  • the runtime finalizes that task cleanly
  • your runtime error handler can log or clean up the failure gracefully

If you do not install an error handler, the task still fails cleanly and the runtime keeps its internal state consistent, but adding setErrorHandler(...) is the recommended way to make these failures visible in applications.

Configuration

The runtime now uses a first-class config object backed by smallos.config.json.

Current config fields:

  • task_capacity: maximum tracked tasks / PID slots
  • priority_levels: number of ready-queue categories
  • io_buffer_length: buffered app output length when the terminal view is hidden
  • eternal_watchers: keep the runtime alive when only watcher tasks remain
  • client_defaults: shared defaults for cooperative clients and streams

Example:

{
  "task_capacity": 1024,
  "priority_levels": 10,
  "io_buffer_length": 1024,
  "eternal_watchers": false,
  "client_defaults": {
    "stream": {
      "max_buffer_size": 16777216
    },
    "http": {
      "max_response_size": 16777216
    },
    "redis": {
      "max_response_size": 16777216,
      "max_nesting_depth": 32
    },
    "mqtt": {
      "keepalive": 60,
      "max_packet_size": 262144,
      "max_queued_messages": 1024
    }
  }
}

The config loader also accepts the aliases oslist_length and num_categories so older notes and experiments can map cleanly onto the current runtime.

Client constructors still accept explicit overrides, but when you create them inside a task they now inherit these defaults from task.OS.config unless you pass a value directly.

Kernels and Board Profiles

Desktop kernel:

  • Unix

MicroPython kernels:

  • MicroPythonKernel
  • ESP32
  • PicoW / RaspberryPiPicoW
  • ESP8266 compatibility profile

You can either pick a board profile explicitly or let the runtime choose a built-in profile from the firmware machine string:

from SmallPackage.Kernel import ESP32, PicoW, build_micropython_kernel

kernel = ESP32(hostname="smallos-esp32")
kernel = PicoW(country="US", hostname="smallos-pico")
kernel = build_micropython_kernel()

Demos

The new demos live in demos:

All of the shared demo entry points now install a default runtime error handler through demos/common.py. That means network failures, invalid I/O wait objects, and other uncaught task exceptions are reported as readable task-failure diagnostics instead of looking like abrupt scheduler crashes or silent exits.

The original root demo remains available in demo.py as a compatibility wrapper around demos/runtime_demo.py.

Runtime Model

smallOS is intentionally small in scope:

  • tasks are async def coroutines wrapped in SmallTask
  • task code awaits smallOS-owned awaitables such as task.sleep(...), task.wait_signal(...), task.wait_readable(...), and task.join(...)
  • the scheduler steps coroutines directly and decides when each task becomes runnable again
  • kernels provide timing, output, and readiness-based transport primitives

This means arbitrary asyncio libraries are not drop-in compatible with the runtime, but it also means scheduling policy and portability stay under your control.

Execution Adapters

Execution adapters let a SmallOS task yield while user-supplied code runs under a different execution model. They use only the Python standard library and do not install, import, configure, or wrap database drivers, ORMs, SDKs, or other third-party packages.

Use ThreadAdapter for a synchronous blocking callable:

from SmallPackage.adapters.threads import ThreadAdapter


def load_record(user_library, settings, record_id):
    connection = user_library.connect(**settings)
    try:
        return connection.load(record_id)
    finally:
        connection.close()


with ThreadAdapter(max_workers=4, max_pending=64) as blocking:
    async def load(task):
        return await blocking.call(
            load_record,
            user_selected_library,
            connection_settings,
            42,
        )

    runtime.fork(SmallTask(2, load, name="load"))
    runtime.start()

Use AsyncioAdapter for an async callable that must run on asyncio. The adapter owns one persistent event loop in a dedicated thread, allowing loop-affine clients to be reused when all their operations are routed through the same adapter:

from SmallPackage.adapters.asyncio_loop import AsyncioAdapter


async def fetch_record(user_library, settings, record_id):
    async with user_library.Client(**settings) as client:
        return await client.fetch(record_id)


with AsyncioAdapter(max_pending=64) as foreign_async:
    async def load(task):
        return await foreign_async.call(
            fetch_record,
            user_selected_async_library,
            connection_settings,
            42,
        )

    runtime.fork(SmallTask(2, load, name="load"))
    runtime.start()

Important behavior:

  • adapters bind to the first SmallOS runtime that uses them;
  • adapter shutdown is explicit, so a context manager should wrap runtime.start();
  • max_pending rejects excess work with AdapterCapacityError instead of blocking the scheduler;
  • cancelling a SmallTask can cancel queued thread work, but cannot forcibly stop a running Python thread;
  • asyncio cancellation is requested on the adapter loop, but a user library may delay or suppress it;
  • pass an async callable to AsyncioAdapter.call(), not a Task or Future already owned by another loop;
  • inspect AsyncioAdapter.shutdown_error after shutdown when application diagnostics need to detect an unexpected loop stop or library teardown failure; normal shutdown leaves it as None;
  • use ThreadAdapter(max_workers=1) when user resources require a serialized, thread-affine execution lane; create, use, and close those resources through calls on that same adapter rather than creating them on the SmallOS thread.

Standard-library examples

All adapter demos run without installing anything beyond SmallOS:

python3 demos/adapters_demo.py
python3 demos/adapters_sqlite_demo.py
python3 demos/adapters_asyncio_demo.py
  • demos/adapters_demo.py runs both adapters beside an ordinary cooperative SmallOS task.
  • demos/adapters_sqlite_demo.py creates, uses, and closes an in-memory sqlite3 connection through ThreadAdapter(max_workers=1). This is the ownership pattern to adapt for a thread-affine PostgreSQL driver or ORM session supplied by the user.
  • demos/adapters_asyncio_demo.py creates an asyncio.Queue, Future objects, and a background task, performs multiple operations, and cleans them up on the same persistent adapter event loop.

Running on MicroPython

For MicroPython targets, the intended flow is:

  1. choose ESP32, PicoW, or build_micropython_kernel()
  2. optionally connect Wi-Fi through the kernel helper
  3. build SmallOS(config=...)
  4. fork tasks and start the runtime

The kernel layer is deliberately generic. Protocol clients such as HTTPS, Redis, MQTT, RabbitMQ/AMQP, and Kafka should be built on top of the shared TCP/TLS socket surface rather than requiring protocol-specific kernel methods.

Clients

The current setup now includes first-party smallOS-native helpers for HTTP, Redis, and MQTT, so users can stay inside the smallOS scheduler instead of dropping down to raw sockets or depending on asyncio-owned clients.

Available helpers:

  • SmallHTTPClient
  • SmallRedisClient
  • SmallMQTTClient

Current scope:

  • HTTP: request/response helper with query params, JSON bodies, TLS, and chunked/content-length response parsing
  • Redis: RESP command execution plus helpers like ping, get, set, delete, publish, and subscribe
  • MQTT: MQTT 3.1.1 connect/disconnect, publish at QoS 0/1/2, subscribe at QoS 0/1/2, and inbound message receive with PUBACK/PUBREC/PUBREL/PUBCOMP handling as required by the protocol
  • Both clients: optional username/password auth and TLS transport setup, with Unix support for custom CA and client certificate paths through tls_ca_file, tls_cert_file, and tls_key_file

For detailed examples, constructor options, response helpers, and transport notes, see SmallPackage/clients/README.md.

Contributing

Contributions, issues, experiments, and board-port notes are welcome. Good areas for contribution include:

  • new MicroPython port validation
  • higher-level protocol clients built on the transport layer
  • shell and debugging tools
  • more board demos and deployment examples
  • additional scheduler tests and edge-case coverage

License

This project is licensed under the MIT License. See LICENSE.

About

Lightweight Robotics Concurrent Task Management Prototype

Resources

Readme
MIT license
Activity

Stars

0 stars

Watchers

1 watching

Forks

0 forks
Report repository

Releases

Packages

Used by

Contributors

Languages