LLD in modern C++: ownership and variants

session · lift_session.py

from session import run, version

version()
run(
    """
clang++ -std=c++23 -Wall -Wextra -Werror -g -fsanitize=address,undefined lift/lift.cpp -o lift_demo
./lift_demo
"""
)

# Apple clang version 21.0.0 (clang-2100.3.34.2)
$ clang++ -std=c++23 -Wall -Wextra -Werror -g -fsanitize=address,undefined lift/lift.cpp -o lift_demo
$ ./lift_demo
  [display] lift 1: moving 0 -> 4
call 4: accepted
call 12: no such floor
  [display] lift 1: doors open at 4
call 7: doors are open
  [display] lift 1: idle at 4
  [display] lift 1: moving 4 -> 2
  [display] lift 1: moving 4 -> 9
call 2, then 9: moving 4 -> 9

lift/lift.cpp

// A lift in C++23. The building owns its lifts by value, each lift owns its notifier through a
// std::unique_ptr, the lift's state is a std::variant, and a call returns a std::expected.
#include <expected>
#include <memory>
#include <print>
#include <string>
#include <utility>
#include <variant>
#include <vector>

struct Idle { int floor; };
struct Moving { int from, to; };
struct DoorsOpen { int floor; };
using LiftState = std::variant<Idle, Moving, DoorsOpen>;

template <class... Fs>
struct overloaded : Fs... { using Fs::operator()...; };

std::string describe(const LiftState& state) {
    return std::visit(overloaded{
        [](const Idle& s) { return std::format("idle at {}", s.floor); },
        [](const Moving& s) { return std::format("moving {} -> {}", s.from, s.to); },
        [](const DoorsOpen& s) { return std::format("doors open at {}", s.floor); },
    }, state);
}

enum class CallError { NoSuchFloor, DoorsAreOpen, AlreadyThere };

std::string_view text(CallError e) {
    switch (e) {
        case CallError::NoSuchFloor: return "no such floor";
        case CallError::DoorsAreOpen: return "doors are open";
        case CallError::AlreadyThere: return "already there";
    }
    std::unreachable();
}

// An open set of channels: a polymorphic base with a virtual destructor.
struct Notifier {
    virtual ~Notifier() = default;
    virtual void notify(const std::string& message) = 0;
};

struct Display final : Notifier {
    void notify(const std::string& message) override { std::println("  [display] {}", message); }
};

class Lift {
public:
    Lift(int id, std::unique_ptr<Notifier> notifier) : id_(id), notifier_(std::move(notifier)) {}

    [[nodiscard]] std::expected<LiftState, CallError> call(int floor) {
        if (floor < 0 || floor > 9) return std::unexpected(CallError::NoSuchFloor);
        auto next = std::visit(overloaded{
            [&](const Idle& s) -> std::expected<LiftState, CallError> {
                if (s.floor == floor) return std::unexpected(CallError::AlreadyThere);
                return Moving{s.floor, floor};
            },
            [&](const Moving& s) -> std::expected<LiftState, CallError> { return Moving{s.from, floor}; },
            [&](const DoorsOpen&) -> std::expected<LiftState, CallError> { return std::unexpected(CallError::DoorsAreOpen); },
        }, state_);
        if (next) set(*next);
        return next;
    }

    void arrive() {
        if (auto* moving = std::get_if<Moving>(&state_)) set(DoorsOpen{moving->to});
    }

    void close() {
        if (auto* open = std::get_if<DoorsOpen>(&state_)) set(Idle{open->floor});
    }

private:
    void set(LiftState next) {
        state_ = next;
        notifier_->notify(std::format("lift {}: {}", id_, describe(state_)));
    }

    int id_;
    LiftState state_ = Idle{0};
    std::unique_ptr<Notifier> notifier_;
};

int main() {
    std::vector<Lift> lifts;  // the building owns its lifts; each lift owns its notifier
    lifts.emplace_back(1, std::make_unique<Display>());
    Lift& lift = lifts.front();

    for (int floor : {4, 12}) {
        auto result = lift.call(floor);
        std::println("call {}: {}", floor, result ? "accepted" : text(result.error()));
    }
    lift.arrive();
    auto doors_open = lift.call(7);
    std::println("call 7: {}", doors_open ? "accepted" : text(doors_open.error()));
    lift.close();

    auto chained = lift.call(2).and_then([&](const LiftState&) { return lift.call(9); }).transform(describe);
    std::println("call 2, then 9: {}", chained.value_or("refused"));
}

session.py

"""Prints a terminal session: "$ " and each command, then everything it printed (standard output and standard
error together), and its exit status when that is not 0.

The compiler is the system's Apple clang with its libc++ (clang++ -std=c++23), and every program is built with
AddressSanitizer and UndefinedBehaviorSanitizer. The compiler's launcher keeps a cache file in the user's temporary
folder, which learn:verify's sandbox does not allow, so these sessions are recorded with LEARN_VERIFY_NETWORK=allow;
everything else the tools write stays in the folder the session runs in, and nothing uses the network.

A command that fails makes the session exit with status 1, unless the script lists it in may_fail. For a command
that fails, `keep` (a regular expression) shows only the lines of its output that matter, and in those lines a path
into the compiler's own headers is shortened to the header's name. Addresses that a sanitizer prints change on
every run, so they are shown as 0x….
"""

import os
import re
import signal
import subprocess
import sys
from pathlib import Path

SYSTEM_HEADER = re.compile(r"/\S*/include/c\+\+/v1/")
ADDRESS = re.compile(r"0x[0-9a-f]{6,}")
PROCESS_ID = re.compile(r"^==\d+==")


def _env():
    here = Path.cwd()
    for folder in ("home", "tmp"):
        (here / folder).mkdir(exist_ok=True)
    env = dict(os.environ)
    env.update(PATH="/usr/bin:/bin", HOME=str(here / "home"), TMPDIR=str(here / "tmp"), NO_COLOR="1")
    return env


def _shell(command):
    done = subprocess.run(command, shell=True, env=_env(), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
    lines = [PROCESS_ID.sub("", ADDRESS.sub("0x…", line)) for line in done.stdout.splitlines()]
    return done.returncode, lines


def version():
    """Prints the compiler's version line as a comment."""
    status, lines = _shell("clang++ --version")
    if status != 0 or not lines:
        sys.exit("clang++ is not installed on this computer")
    print("#", lines[0])


def run(script, may_fail=(), keep=None):
    """Runs each line of the script as one shell command, in the examples folder."""
    unexpected = 0
    for command in (line.strip() for line in script.strip().splitlines()):
        print(f"$ {command}")
        status, lines = _shell(command)
        if status != 0 and keep:
            lines = [SYSTEM_HEADER.sub("", line) for line in lines if re.search(keep, line)]
        for line in lines:
            print(line)
        if status < 0:
            print(f"[stopped by signal {signal.Signals(-status).name}]")
        elif status != 0:
            print(f"[exit status {status}]")
            if command not in may_fail:
                unexpected += 1
    if unexpected:
        sys.exit(1)