Три способа ввода-вывода -- опрос, прерывания, DMA

В предыдущей лекции мы узнали, как ЦП и память взаимодействуют через шину. Но компьютер — это не только ЦП плюс память: ему также нужно обмениваться данными с различными внешними устройствами (устройствами ввода-вывода), такими как клавиатуры, мыши, жёсткие диски и сетевые карты.

Существует три основных способа обмена данными между процессором и внешними устройствами:Опрос (Polling)、Прерывание (Interrupt)、DMA (прямой доступ к памяти)Эффективность этих трёх методов различается огромно — от того, какой из них вы выберете, часто зависит производительность всей системы.

В этой лекции мы используем три бытовых сценария, чтобы помочь вам сформировать интуицию, а затем с помощью кода на Python наглядно сравним различия в эффективности этих трёх методов.


Три аналогии из реальной жизни

Сценарий 1: Опрос — проверяйте каждые 5 секунд, закипела ли вода

Вы кипятите воду на кухне, но у чайника нет свистка. Каждые 5 секунд вы бегаете из гостиной на кухню, чтобы проверить: «Вода кипит? Вода кипит? Вода кипит?»

В течение этого времени выСовершенно не можете спокойно смотреть телевизорПотому что вам приходится бегать каждые несколько секунд, вы тратите почти всё время на «проверку состояния».

ЭтоопросОпрос: процессор непрерывно зацикливается, проверяя, готово ли устройство, тратя время на бессмысленные запросы.

Сценарий 2: Прерывание — выключите плиту, когда засвистит чайник

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

выПодавляющую часть времени вы занимаетесь осмысленными делами(смотрите телевизор), прерываясь лишь ненадолго в критический момент.

ЭтопрерываниеПрерывание: процессор нормально выполняет свою программу; когда устройство готово, оно активно посылает сигнал, чтобы уведомить процессор. Процессор приостанавливает текущую работу, обрабатывает данные устройства и затем возобновляет предыдущую работу.

Сценарий 3: DMA — просто вызовите компанию по переезду, и вам не придётся палец о палец ударить

Вы переезжаете, и целый грузовик мебели нужно перевезти из старого дома в новый. Если вы сами переносите вещи по одной (способ с прерываниями), вам придётся бегать туда и обратно десятки раз.

Вы нанимаете фирму по переезду (контроллер DMA) и говорите ей: «Адрес старого дома — X, адрес нового дома — Y, всего Z предметов мебели». Затем вы продолжаете заниматься своими делами. Когда фирма всё перевезёт, она звонит вам и говорит: «Переезд завершён».

ЭтоDMADMA: выделенный контроллер DMA напрямую управляет передачей данных между памятью и периферийными устройствами, а ЦП вообще не нужно участвовать. После завершения передачи контроллер DMA отправляет ЦП уведомление в виде прерывания.


Принципиальные различия между тремя методами

ОсобенностиОпросПрерватьDMA
Участие ЦППостоянно занят, всё время проверяетНенадолго участвует во время передачиПочти не участвует, только получает уведомление о завершении
Кто перемещает данныеЦП сам переносит каждый байтЦП сам переносит каждый байтПеремещением занимается контроллер DMA; CPU в этом не участвует
Время простоя CPUПрактически отсутствуетПростаивает большую часть времениПростаивает почти всё время
Сложность реализацииСамый простойТребуется аппаратный контроллер прерыванийТребуется аппаратный контроллер DMA
Задержка откликаЗависит от интервала опросаНа уровне микросекундЗависит от общей длительности передачи DMA
Применимый размер данныхКрайне малый объёмМалые и средние объёмы (уровень KB)Большие объёмы (уровень MB/GB)
Типичные сценарииПростые встраиваемые системы, крайне короткие ожиданияВвод с клавиатуры, движение мыши, поступление сетевого пакетаЧтение/запись на диск, буфер кадра видеокарты, большие сетевые пакеты

Каждый раз при нажатии клавиши генерируется прерывание; при копировании большого файла используется DMA. Современные операционные системы автоматически выбирают лучшую схему из трёх методов в зависимости от объёма данных и сценария.


Визуализация временной шкалы: сравнение состояний CPU «занят» и «простаивает»

Предположим, нам нужно считать 4 блока данных с жёсткого диска в память. Мы используем ASCII-временную шкалу, чтобы наглядно сравнить состояние CPU при трёх методах:

任务: 从硬盘读取 4 个数据块到内存

【轮询 - CPU 几乎全程忙碌】
时间: 0----1----2----3----4----5----6----7----8----9----10--->
CPU:  [查][查][搬][查][查][搬][查][查][搬][查][查][搬]
       忙  忙  忙  忙  忙  忙  忙  忙  忙  忙  忙  忙
                                   ↑ 几乎没有任何空闲

【中断 - CPU 大部分时间空闲】
时间: 0----1----2----3----4----5----6----7----8----9----10--->
CPU:  [工作][等待...][搬][工作]  [等待...][搬][工作]
       忙    闲       忙  忙      闲       忙  忙
                    ↑ 中断触发    ↑ 中断触发

【DMA - CPU 几乎全程空闲】
时间: 0----1----2----3----4----5----6----7----8----9----10--->
CPU:  [设置DMA][工作][工作][工作]      [被通知完成]
       忙       忙    忙    忙              忙
DMA:         [搬][搬][搬][搬]                  ↑ DMA 完成后中断
             忙  忙  忙  忙
                                  ↑ CPU 完全不管传输过程

Сразу видно: при опросе CPU занят; при прерываниях CPU прерывается лишь ненадолго; при DMA CPU почти полностью освобождается.


Демонстрация кода: имитация трёх методов ввода-вывода на Python и вывод временной шкалы

Приведённый ниже код имитирует передачу одного и того же набора данных с использованием соответственно опроса, прерываний и DMA, вычисляет долю времени занятости/простоя CPU и с помощью ASCII-временной шкалы наглядно показывает различия.

Пример

"""
Сравнение трёх методов ввода-вывода: программный опрос (polling) против прерываний (interrupt) против DMA (демонстрационный пример)

Эта программа моделирует сценарий, в котором CPU передаёт 10 блоков данных из «внешнего устройства» в память,
Реализуйте соответственно с использованием программного опроса (polling), прерываний (interrupt) и DMA и выведите временную шкалу для наглядного сравнения состояний CPU.

Основная метрика: коэффициент занятости CPU. Чем выше коэффициент, тем сильнее CPU загружается операциями ввода-вывода.
"""


import time
import random


class IODevice:
    """Моделирует внешние устройства ввода-вывода (такие как жёсткие диски, клавиатуры и т. д.)"""

    def __init__(self, name="Жёсткий диск", response_time=4):
        """
name: имя устройства
response_time: количество 'единиц времени', необходимое для ответа устройства
        """

        self.name = name
        self.response_time = response_time  # Время подготовки устройства на один блок данных
        self.data_buffer = []               # Внутренний буфер данных устройства
        self.ready = False                  # Готово ли устройство?

    def prepare_data(self, block_id):
        """
Устройство подготавливает данные — имитация внутренней обработки устройства.
В реальном оборудовании этот шаг включает такие операции, как поиск на диске, приём пакетов сетевой картой и т. д.
        """

        self.ready = False
        data = f"DATA_BLOCK_{block_id:02d}"
        self.data_buffer.append(data)
        # Имитировать, что устройства готовы через разные единицы времени
        return data

    def is_ready(self):
        """CPU проверяет этот статус при опросе"""
        return self.ready

    def mark_ready(self):
        """Устройство завершило подготовку, отправляет сигнал прерывания"""
        self.ready = True

    def get_data(self):
        """Взять блок данных из буфера устройства"""
        if self.data_buffer:
            return self.data_buffer.pop(0)
        return None


class Timeline:
    """
Логгер Timeline — записывает состояние CPU и DMA для каждой единицы времени
Используется для финального рендеринга ASCII-таймлайна
    """


    def __init__(self):
        self.cpu_states = []     # состояние CPU на единицу времени: 'BUSY' или 'IDLE'
        self.dma_states = []     # состояние DMA на единицу времени
        self.events = []         # журнал событий: (time, description)

    def record_cpu(self, state):
        self.cpu_states.append(state)

    def record_dma(self, state):
        self.dma_states.append(state)

    def log_event(self, time_unit, desc):
        self.events.append((time_unit, desc))

    def draw(self, title="Временная шкала"):
        """Нарисовать временную шкалу в ASCII"""
        total = len(self.cpu_states)
        if total == 0:
            print(" (нет данных о времени)")
            return

        print(f"\n  ╔══ {title} ══╗")
        print(f" ║ Всего единиц времени: {total}")

        # Строка состояния CPU
        cpu_line = "CPU: "
        for s in self.cpu_states:
            cpu_line += "[" + ("Занят" if s == "BUSY" else "Простой") + "]"
        print(f" ║ {cpu_line}")

        # Строка состояния DMA (если есть активность DMA)
        if any(s == "BUSY" for s in self.dma_states):
            dma_line = "DMA: "
            for s in self.dma_states:
                dma_line += "[" + ("Занят" if s == "BUSY" else "Простой") + "]"
            print(f" ║ {dma_line}")

        # Журнал событий
        if self.events:
            print(f" ╠══ Журнал событий ══╣")
            for t, desc in self.events:
                print(f" ║ T={t}: {desc}")

        # Статистическая сводка
        busy_count = sum(1 for s in self.cpu_states if s == "BUSY")
        idle_count = total - busy_count
        busy_pct = busy_count / total * 100 if total > 0 else 0
        idle_pct = idle_count / total * 100 if total > 0 else 0
        print(f" ╠══ Статистика ══╣")
        print(f║ CPU занят: {busy_count}/{total} ({busy_pct:.0f}%))
        print(f║ CPU простаивает: {idle_count}/{total} ({idle_pct:.0f}%))
        print(f"  ╚{'═' * 20}╝")

        return busy_pct


class Memory:
    """Имитация основной памяти"""
    def __init__(self):
        self.data = {}

    def write(self, addr, data):
        self.data[addr] = data

    def __repr__(self):
        return fMemory({len(self.data)} элементов данных)


class Controller:
    """Контроллер CPU — имитация выполнения различных режимов ввода-вывода"""

    def __init__(self):
        self.memory = Memory()
        self.total_transferred = 0

    def simulate_polling(self, device, block_count):
        """
Имитация опроса:
CPU проверяет, готово ли устройство, на каждом временном шаге.
Если готово, переместить блок данных; иначе продолжать проверку.
        """

        print(f"\n{'#'*55}")
        print(f"# Метод 1: Опрос")
        print(f# Устройство: {device.name} | Количество блоков: {block_count})
        print(f"# Стратегия: CPU проверяет состояние устройства каждую единицу времени")
        print(f"{'#'*55}")

        timeline = Временная шкала()
        time_unit = 0
        blocks_done = 0
        device_ready_at = 0  # В какую единицу времени устройство готово к следующему блоку

        # Инициализация: устройство начинает подготовку первого блока данных
        device.prepare_data(0)
        device_ready_at = time_unit + device.response_time

        while blocks_done < block_count:
            # CPU проверяет состояние устройства (опрос)
            timeline.record_cpu("BUSY")
            timeline.record_dma("IDLE")

            if time_unit >= device_ready_at and device.is_ready():
                # Устройство готово -> ЦП передаёт данные
                data = device.get_data()
                self.memory.write(blocks_done, data)
                timeline.log_event(time_unit, f"ЦП переместил блок данных {blocks_done}: {data}")
                blocks_done += 1
                self.total_transferred += 1

                # Подготовить следующий блок данных (если есть)
                if blocks_done < block_count:
                    device.prepare_data(blocks_done)
                    device_ready_at = time_unit + device.response_time
            else:
                # Устройство не готово -> ЦП сделал напрасную попытку
                timeline.log_event(time_unit,
                    fCPU проверяет устройство... {'Готово' if device.is_ready() else 'Не готово'})

            # Фоновая работа устройства (если в фазе подготовки)
            if time_unit == device_ready_at:
                device.mark_ready()
                timeline.log_event(time_unit, Устройство готово, данные готовы)

            time_unit += 1

        print(f"\nПередача завершена: {self.total_transferred} блоков данных)
        busy_pct = timeline.draw("Временная шкала режима опроса")
        return busy_pct

    def simulate_interrupt(self, device, block_count):
        """
Симуляция прерывания:
CPU большую часть времени выполняет другие задачи (обозначается как IDLE),
Только после получения сигнала прерывания он приостанавливает текущую работу, обрабатывает передачу данных и затем продолжает.
        """

        print(f"\n{'#'*55}")
        print(f# Способ 2: Прерывание)
        print(f# Устройство: {device.name} | Количество блоков: {block_count})
        print(f# Стратегия: Процессор выполняет задачи в обычном режиме и отправляет уведомление о прерывании, когда устройство готово)
        print(f"{'#'*55}")

        timeline = Временная шкала()
        time_unit = 0
        blocks_done = 0
        transferred_in_interrupt = 0
        device_ready_at = 0

        # Инициализация
        device.prepare_data(0)
        device_ready_at = time_unit + device.response_time

        while blocks_done < block_count:
            # По умолчанию: Процессор выполняет другие задачи (простой)
            if time_unit == device_ready_at:
                # Устройство готово, отправляем сигнал прерывания
                device.mark_ready()
                timeline.log_event(time_unit, Сработало прерывание! Данные устройства готовы)

                # Процессор реагирует на прерывание и передаёт данные (в это время процессор занят)
                data = device.get_data()
                for substep in range(2):  # Обработка прерывания требует 2 единицы времени
                    if substep == 0:
                        timeline.record_cpu("BUSY")  # Сохранить контекст, войти в обработку прерывания
                        timeline.record_dma("IDLE")
                        timeline.log_event(time_unit + substep, f"CPU входит в обработку прерывания")
                    else:
                        timeline.record_cpu("BUSY")  # Переместить данные
                        timeline.record_dma("IDLE")
                        self.memory.write(blocks_done, data)
                        transferred_in_interrupt += 1
                        timeline.log_event(time_unit + substep, f"CPU переместил блок данных {blocks_done}: {data}")
                        blocks_done += 1

                # Обработка прерывания завершена, CPU возобновляет свою предыдущую работу.
                timeline.record_cpu("BUSY")
                timeline.record_dma("IDLE")
                timeline.log_event(time_unit + 2, CPU возобновляет предыдущую работу)
                time_unit += 3  # Пропустить время, затраченное на обработку прерывания

                # Подготовить следующий блок
                if blocks_done < block_count:
                    device.prepare_data(blocks_done)
                    device_ready_at = time_unit + device.response_time
                continue

            # Фаза нормального выполнения — CPU выполняет свои задачи
            timeline.record_cpu("IDLE")
            timeline.record_dma("IDLE")
            timeline.log_event(time_unit, CPU выполняет другие задачи (простой/пользовательская программа))

            time_unit += 1

        print(f"\nПередача завершена: {self.total_transferred} блоков данных)
        busy_pct = timeline.draw("Временная шкала режима прерывания")
        return busy_pct

    def simulate_dma(self, device, block_count):
        """
Симуляция режима DMA:
ЦП только настраивает контроллер DMA в начале, а затем полностью не вмешивается в передачу данных.
Контроллер DMA автономно перемещает все блоки данных.
После завершения передачи DMA отправляет прерывание, чтобы уведомить ЦП.
        """

        print(f"\n{'#'*55}")
        print(f# Метод 3: DMA (прямой доступ к памяти))
        print(f# Устройство: {device.name} | Количество блоков: {block_count})
        print(f# Стратегия: ЦП настраивает DMA и продолжает работу, контроллер DMA автоматически передаёт данные)
        print(f"{'#'*55}")

        timeline = Временная шкала()
        time_unit = 0
        dma_busy_until = 0
        setup_done = False
        blocks_to_transfer = list(range(block_count))
        dma_buffer = []

        # Подготовить все блоки данных
        for i in range(block_count):
            device.prepare_data(i)

        while blocks_to_transfer or dma_buffer:
            # Этап 1: CPU настраивает контроллер DMA (только один раз, занимает 2 единицы времени)
            if not setup_done:
                for substep in range(2):
                    timeline.record_cpu("BUSY")
                    timeline.record_dma("IDLE")
                    if substep == 0:
                        timeline.log_event(time_unit,
                            CPU настраивает DMA: задаёт адрес источника, адрес назначения и количество передаваемых байтов)
                    else:
                        timeline.log_event(time_unit,
                            CPU инициирует передачу DMA, затем продолжает свою работу)
                    time_unit += 1
                setup_done = True
                dma_busy_until = time_unit + block_count * device.response_time
                # Пометить устройство как готовое
                device.mark_ready()
                # Предварительно заполнить буфер DMA
                while device.data_buffer:
                    dma_buffer.append(device.data_buffer.pop(0))
                continue

            # Вторая фаза: контроллер DMA передает данные автономно, CPU простаивает
            if time_unit < dma_busy_until:
                timeline.record_cpu("IDLE")
                timeline.record_dma("BUSY")
                if dma_buffer:
                    data = dma_buffer.pop(0)
                    blocks_to_transfer.pop(0)
                    mem_addr = block_count - len(blocks_to_transfer) - 1
                    self.memory.write(mem_addr, data)
                    self.total_transferred += 1
                    timeline.log_event(time_unit, fDMA передает блок данных {mem_addr}: {data})
                else:
                    timeline.log_event(time_unit, "контроллер DMA работает...")
            else:
                # Этап 3: DMA завершено, отправляет прерывание для уведомления CPU
                timeline.record_cpu("BUSY")
                timeline.record_dma("BUSY")
                timeline.log_event(time_unit, "Передача DMA завершена, прерывание уведомляет ЦП")
                timeline.record_cpu("BUSY")
                timeline.record_dma("IDLE")
                timeline.log_event(time_unit + 1, "ЦП подтверждает завершение передачи")
                time_unit += 2
                break

            time_unit += 1

        print(f"\nПередача завершена: {self.total_transferred} блоков данных)
        busy_pct = timeline.draw("Временная диаграмма режима DMA")
        return busy_pct


# ===== Основная программа: сравнение трёх подходов =====
if __name__ == "__main__":
    print("=" * 60)
    print(Сравнение эффективности трёх режимов ввода-вывода — ПРИМЕР демонстрации принципов организации ЭВМ)
    print("=" * 60)
    print()
    print(Сценарий: передача 5 блоков данных из внешнего устройства в память)
    print(Время отклика устройства (временных единиц, необходимых для подготовки каждого блока данных): 4)
    print()

    BLOCK_COUNT = 5
    RESPONSE_TIME = 4

    results = {}

    # ---- Метод 1: опрос ----
    dev1 = IODevice("Жёсткий диск", RESPONSE_TIME)
    ctrl1 = Controller()
    pct1 = ctrl1.simulate_polling(dev1, BLOCK_COUNT)
    results["Опрос"] = pct1

    # ---- Метод 2: Прерывание ----
    dev2 = IODevice("Жесткий диск", RESPONSE_TIME)
    ctrl2 = Controller()
    pct2 = ctrl2.simulate_interrupt(dev2, BLOCK_COUNT)
    results["Прерывание"] = pct2

    # ---- Метод 3: DMA ----
    dev3 = IODevice("Жесткий диск", RESPONSE_TIME)
    ctrl3 = Controller()
    pct3 = ctrl3.simulate_dma(dev3, BLOCK_COUNT)
    results["DMA"] = pct3

    # ---- Финальное сравнение ----
    print("\n" + "=" * 60)
    print(Сравнение долей времени занятости CPU для трёх методов)
    print("=" * 60)
    print(fОпрос: CPU занят {results['polling']:.0f}% времени)
    print(fПрерывания: CPU занят {results['interrupt']:.0f}% времени)
    print(fDMA: CPU занят {results['DMA']:.0f}% времени)
    print()

    best = min(results, key=results.get)
    worst = max(results, key=results.get)
    print(f" Лучший метод: {best} (наименьшая загрузка CPU)")
    print(f" Худший метод: {worst} (наибольшая загрузка CPU)")
    print()
    print(fВывод: при передаче больших объёмов данных DMA снижает загрузку CPU по сравнению с опросом.)
    print(fпримерно на {results['polling'] - results['DMA']:.0f} процентных пунктов.)
    print(f" Вот почему современные компьютеры используют DMA при передаче больших файлов.")
    print("=" * 60)

Реальные практические сценарии применения

Ввод с клавиатуры — типичный пример прерываний

Каждый раз, когда вы нажимаете клавишу, контроллер клавиатуры отправляет в CPU сигнал прерывания. CPU приостанавливает текущую программу, считывает код клавиши и затем возобновляет работу. Этот процесс настолько быстр, что вы не замечаете никакой задержки.

Если бы клавиатура использовала опрос — CPU проверял бы 1,000 раз в секунду: «Пользователь нажал клавишу?» — ваш компьютер завис бы, и вы не смогли бы ничего делать.

Копирование больших файлов — типичный пример DMA

Вы копируете фильм размером 2GB с диска C на диск D. Дисковый контроллер использует DMA для перемещения данных напрямую в память; CPU почти не участвует во всём процессе. Вы также можете одновременно просматривать веб-страницы и писать код, и это совершенно не отражается на работе.

Без DMA — CPU пришлось бы самому переносить 2GB данных, — и в течение этого времени CPU практически не может выполнять другие задачи.

Почему опрос до сих пор не исключили?

Хотя опрос имеет самую низкую эффективность, он всё ещё полезен вчрезвычайно простой встраиваемой системе:

  • Самая низкая стоимость аппаратного обеспечения — не нужен ни контроллер прерываний, ни контроллер DMA.
  • В сценариях, где «устройство готово почти мгновенно», накладные расходы при опросе сопоставимы с прерываниями, но его гораздо проще реализовать.
  • В некоторых системах реального времени детерминированность опроса (точное знание, когда проверять) имеет преимущество перед случайностью прерываний.

Итоги и проверка

Краткая сводка одним предложением:опрос — это когда CPU постоянно спрашивает «Уже готово?», прерывание — когда устройство само сообщает «I/O готов», DMA находит помощников, чтобы делать работу за вас, — чем больше объём данных, тем очевиднее преимущества DMA.

Вопросы для самопроверки

  1. Вы используете компьютер для написания документа, пока в фоне копируется большой файл. Какой метод I/O используется для копирования файла? Какой используется для ввода с клавиатуры?
  2. Если время подготовки данных устройством чрезвычайно мало (например, всего 1 microsecond), будет ли опрос всё ещё впустую расходовать CPU? Почему?
  3. После завершения передачи данных контроллером DMA, как он уведомляет CPU? Какой механизм для этого использует?

Эталонный ответ: 1. Для копирования файлов используется DMA, для ввода с клавиатуры — прерывания. 2. Всё равно тратит впустую — потому что во время опроса CPU не может делать другие вещи; даже если каждый опрос быстр, занятый временной срез нельзя использовать для выполнения пользовательских программ. Но если устройство действительно очень быстрое, простота реализации опроса может оказаться более выгодной. 3. После завершения DMA он уведомляет CPU через прерывание — это единственный момент, когда DMA требует участия CPU.

Другие расширения