Работа с данными при инкапсуляции/декапсуляции сообщения

При формировании или интерпретации пакета в рамках определённого протокола есть 2 способа работы с данными (под данными я буду подразумевать полезную нагрузку по отношению к текущей степени вложенности):

• создание отдельного буфера данных для данного протокола;
• на всём пути обработки кадра пользоваться памятью, выделенной для самого низкоуровнего протокола.

Первый путь наиболее прост и минимизирует вероятность ошибок, особенно, при плохо налаженном механизме блокировок при асинхронном характере поступающих данных. Но есть пару минусов: увеличение времени на обработку пакета, связанную с копированием одних и тех же данных из одного буфера в другой и увеличение требований к ОЗУ, которая используется для этих буферов.

Второй способ предпочтительнее для систем с небольшими аппаратными ресурсами и высокими требованиями ко времени обработки сообщений.


Структура raw_msg характеризует более низкоуровненый протокол, а high_msg - высокоуровневый. Для простоты предположим, что у каждого протокола есть только по одному служебному полю - поле адреса.

struct raw_msg {
    RAW_ADDR addr;
    unsigned char *data;
    int length;
}

struct high_msg {
    HIGH_ADDR addr;
    unsigned char *data;
    int length;
}

Тогда, для первого случая:

void high_input(struct raw_msg *rmsg) {
    struct high_msg hmsg;
    unsigned char high_data[MAX_HIGH_DATA_SIZE];
    int i;

    hmsg.data = high_data;

    hmsg.addr = rmsg->data[ADDR_OFF];

    for(i = 0; i < rmsg->length; i++)
        hmsg.data[i] = rmsg->data[i + RAW_DATA_OVERHEAD];

    hmsg.length = rmsg->length - RAW_DATA_OVERHEAD;

    /* Дальнейшая обработка hmsg */
}

void high_output(RAW_ADDR raddr, struct high_msg *hmsg){
    struct raw_msg rmsg;
    unsigned char raw_data[MAX_RAW_DATA_SIZE];
    int i;

    rmsg.data = raw_data;

    rmsg.addr = raddr;
    rmsg.data[ADDR_OFF] = hmsg->addr;

    for(i = 0; i < hmsg->length; i++)
        rmsg.data[i + HIGH_DATA_OVERHEAD] = hmsg->data[i];

    rmsg.length = hmsg->length + RAW_DATA_OVERHEAD;

    /* Передача rmsg */
}

Альтернативно можно выделить 2 буфера (I/O), размер которых равен максимальному размеру самого низкоуровнего пакета в системе (опишу случай только для входящего сообщения).

unsigned char in_raw_data[MAX_RAW_DATA_SIZE];
unsigned char out_raw_data[MAX_RAW_DATA_SIZE];

void high_input(struct raw_msg *rmsg) {
    struct high_msg hmsg;
    unsigned char *high_data;

    hmsg.addr = rmsg->data[ADDR_OFF];

    /* Ничего никуда копировать не нужно. rmsg->data ссылается на in_raw_data */
    hmsg.data = rmsg->data + RAW_DATA_OVERHEAD;

    hmsg.length = rmsg->length - RAW_DATA_OVERHEAD;

    /* Дальнейшая обработка hmsg */
}

Кольцевой буфер

Синхронный режим передачи данных предполагает некоторую гибкость участников. Прежде всего это отображается в реализации буферов. При асинхронной передаче процесс записи будет выглядеть примерно так:

1. Принял (максимальное или заранее известное количество) данных;
2. Обработка данных; 
3. Отправил подтверждение;
4. GOTO 1

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

struct buf {
  unsigned char *data;
  unsigned long lenth;
};

, то появляется некоторая негибкость или потеря производительности при чтении буфера, если мы берём за правило тот факт, что мы хотим читать произвольное количество данных: если читать самые старые данные, то после нужно смещать остаток буфера, длинной (buf.lenth - read_data_size) к его началу, освобождая место в "хвосте" для новых данных. Такая операция с памятью займёт очень много времени, которое зависит от размера имеющихся данных. Либо же мы можем читать последние данные, декрементируя счётчик размера буфера. Понятно, что далеко не всегда такой подход удовлетворяет логику процесса.
Для увеличения производительности буфера можно организовать его в кольцо. Аттрибуты такого кольцевого буфера: собственно, сам буфер, представляющий собой выделенную область памяти определённой длинны; счётчик записи; счётчик чтения.
Логику работы буфера можно описать следующими положениями:

1. Счётчик чтения всегда "догоняет" счётчик записи;
2. Если значения счётчиков достигло размера выделенной памяти, то они зануляются (уходят на следующий круг);
3. Если счётчик записи обогнал счётчик чтения на круг, то запись должна прекратиться (возможно, теряя поступающие данные), либо счётчик чтения должен инкрементироваться вслед за записью (теряя самые старые данные).

Отмечу лишь, что, если размер буфера равен размерности счётчиков (data[256], unsigned char counter), то следить за значением последних не обязательно: они будут зануляться "естественным путём".

Пример реализации процедур записи и чтения:

/* Round-robin buffer */
struct rr_buf {
  unsigned char *data;
  unsigned short wrPos; // счётчик записи
  unsigned short rdPos; // счётчик чтения
  unsigned char rounded; // счётчик записи ушёл на второй круг, а чтения - нет
  unsigned char overflow; // может быть использовано, если допустима потеря данных
};

struct rr_buf buf;

unsigned short
buf_write(unsigned char *data, unsigned long size)
{
  /* Bytes have been written */
  unsigned short bwritten = 0;

  while(bwritten < size) {
    /* Not enough space */
    if(g_buf.wrPos == g_buf.rdPos && g_buf.rounded)
      break;

    /* Writing byte by byte */
    *(g_buf.data + g_buf.wrPos) = *(data + bwritten);
    g_buf.wrPos++;

    if(g_buf.wrPos == SC_BUF_SIZE) {
      g_buf.wrPos = 0;
      g_buf.rounded = 1;
    }

    bwritten++;
  }

  return bwritten;
}

unsigned short
buf_read(unsigned char *data, unsigned long size)
{
  unsigned short bread = 0;

  while(bread %lt size) {
    /* End of data */
    if(buf.wrPos == buf.rdPos && !buf.rounded)
      break;

    *(data + bread) = *(buf.data + buf.rdPos);
    buf.rdPos++;

    if(buf.rdPos == BUF_SIZE) {
      buf.rdPos = 0;
      buf.rounded = 0;
    }

    bread++;
  }

  return bread;
}

/* May be useful */
unsigned short
get_data_size(void)
{
  unsigned short size;

  if(buf.rounded)
    size = BUF_SIZE - buf.rdPos + buf.wrPos;
  else
    size = buf.wrPos - buf.rdPos;

  return size;
}

Своя отказоустойчивая шина ВВ. Реализация очередей.

Если забыть о минимальном времени обработки прерывания микроконтроллером и о размерах ОЗУ (данные показатели у современных МК довольно высоки), то можно было бы сказать, что параллельная обработка сигналов с "самодельных" шин не представляет особой сложности.
Проблемы начинаются, когда мы видим, что на шине появляются паразитные сигналы: это может быть спецификой работы передатчика при его включении-выключении, внешние наводки и т.п. Например, я время от времени регистрировал характерные данные на своём МК и долго не мог понять их просхождение. Виновником оказалась моя привычка дрыгать ногами под столом и задевать штекер питания настольных аудиоколонок, которые очень сильно "звонили".
В этом случае мы должны оградиться от таких помех логическими, программными, методами: если я знаю характер сигнала, а именно, его количественные и временные характеристики, то я могу написать фильтр для сообщений от передатчика.
Количественно-временные показатели я могу получить с помощью таймера.
Алгоритм прост:
- Начальное состояние: счётчик обнулён, таймер остановлен;
- При получении первого сигнала таймер запускается, счётчик инкрементируется (как и при получении последующих сигналов);
- Если показания счётчика достигли длины ожидаемого сообщения, таймер останавливается и проверяется его значение: оно должно быть не меньше заранее заданной константы. Таким образом мы защищаемся от наводок, несущих большое количество данных за короткий период времени;
- Если таймер достиг максимального значения (т.е. времени, за которое сообщение точно должно было быть доставлено), то и таймер, и счётчик обнуляются. Вернулись к исходным значениям. Этим действием нейтралезуем  короткие всплески на шине.

Теперь надо уточнить ещё один оставшийся момент: сколько мне нужно таймеров для предотвращения DoS в своей системе? Столько же, сколько и линий передачи данных? Достаточно лишь завести очередь, к кажной ячейки которой будет привязан свой таймер. Ячейка будет характеризоваться состоянием свободна-занята. Таким образом доступ к такой ячейке производиться свободно, но ни как в FIFO.
А вопрос о случае заполнения всей очереди при поступлении большого количество запросов (т.е. вопрос о величине и готовности "жертвования" системных таймеров) зависит от характеристик протокола передачи данных и от условий эксплуатации.

Кстати, для выходных сигналов так же потребуется своя очередь, поскольку они имеют свои временные характеристики.

/* Очередь входящих сообщений */
volatile struct sPeriphMsg {
    unsigned long ulPeriphMsgBuf[PERIPH_MSG_LEN]; // буфер для пакета
    unsigned short usBitCount; // счётчик
    unsigned short usDevId; // ID устройства. Так я найду свою ячейку при поступлении следующего сигнала.
} g_sPeriphMsg[PERIPH_QUEUE_LEN];

Организация данных регистрации УВВ.

При проектировании системы управления с элементами БД, например, для описания периферийных устройств ввода-вывода, столкнулся с задачей, попервоначалу поставившей меня в замешательство: важно обеспечить быстрый доступ к данным, имея на руках некий ключ (индекс, если хотите), как PrimaryKey в большинстве таблиц реляционной БД.
Взявшись за описание какого-то абстрактного устройства (читай "Си-структуры"), так и держал в голове некий ID, по которому и буду ориентироваться. Но такая схема доступа приемлема лишь для вывода - для входных значений мы должны изначально привязываться к самым "низшим", с точки зрения предыдущей логики, данным - к каким-то аппаратным адресам.
Так или иначе, роль таблицы реляционной БД  (уж простите за столь настойчивое сравнение) будет выполнять массив, а ключами будут его индексы. Нужно выбрать, какое "направление" будет первичным: ввод или вывод. В первом случае в качестве ключа будут использоваться какие-то аппаратные адреса; во втором - присвоенные разработчиком IDs.
Во избежание чрезмерного использования ОЗУ для организации обратного (по отношению к выбранному первичному направлению) доступа к данным об устройстве создаются массивы (таблицы) с указателями на основные записи.

/* Массив структур, доступ к которым осуществляется по "хардверным" координатам */
struct PeriphDev {
    enum DevType eDevType;
    unsigned long ulFlags;
    ...
} sPeriphDev[HWADD1][HWADD2];

/* Для "обратного" доступа - по ID устройства */
struct FooTypeDev {
    struct PeriphDev *sPeriphDev;
} sFooTypeDev[FOO_TYPE_DEV_COUNT];

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