[Pavel V. 0]

Как правильно организовать работу с пакетным протоколом (что-то наподобие WAKE в собственной реализации) в условиях использования операционки?

 

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

 

Ответный пакет формируется в функции обработки команды и запускается отправка по прерыванию.

 

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

 

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

 

Как правильно реализовать такую схему при наличии операционки?

 

Пока видится только такая схема работы:

1. Процесс, который занимается отправкой данных в порт, занят только тем, что висит в вечном channel.pop() из кольцевого буфера и в случае появления данных выполняет отправку байта в порт.

2. Процесс, который обрабатывает асинхронные события. Например, нажали кнопку. Процесс в этом случае вызывает некую функцию SendInputAsync() в которой формируется пакет и делается необходимое количество channel.push(). Затем процесс уходит в Sleep() на заданное настройками время.

 

Возможно, необходимо сделать процесс на каждое асинхронное событие?

 

Смущает передача управления для отправки каждого байта. Можно, конечно, вызывать channel.pop() с таймаутом и воспользоваться дополнительным событием, которое будет устанавливаться только по добавлению в канал полного пакета. Но как-то громоздко получается.

 

Прошу помощи у более опытных товарищей :)

[Сергей Борщ 143]

1. Процесс, который занимается отправкой данных в порт, занят только тем, что висит в вечном channel.pop() из кольцевого буфера и в случае появления данных выполняет отправку байта в порт.

А почему бы не повесить кольцевой буфер на UART (и опустошать его в прерывании), а сам процесс отправки пакета не обрамить мутексом? Тогда любой процесс сможет оправлять пакеты асинхронно. Что-то типа такого (не wake, но подобный протокол):

class bus_frame
{
public:
   typedef uint16_t address_t;
protected:
   enum
   {
       STA = 0xC0,
       STO = 0xC1,
       CTL = 0x7F,
       TRANSPARENCY = 0x20,
   };
   typedef crc::ccitt crc_t;
};

class frame : public bus_frame
{
public:
   frame(iodevice & uart);
   frame(iodevice & uart, address_t address);
   ~frame();

   void send(void const * buffer, size_t size);
   void send(uint8_t data);                        // calculate CRC and send transparently

protected:
   void send_transparently(uint8_t data);
   static OS::TMutex Mutex;
   iodevice & Uart;
   crc_t CRC;
};


frame::frame(iodevice & uart, address_t address)
: Uart(uart)
{
   Mutex.Lock();
   Uart.send(STA);
   send(address >> 0);
   send(address >> 8);
}

frame::frame(iodevice & uart)
: Uart(uart)
{
   Mutex.Lock();
   Uart.send(STA);
}

frame::~frame()
{
   send_transparently(~CRC >> 0);
   send_transparently(~CRC >> 8);
   Uart.send(STO);
   Mutex.Unlock();
}

void frame::send(uint8_t data)
{
   CRC.calculate(data);
   send_transparently(data);
}

void frame::send_transparently(uint8_t data)
{
   if((data == CTL) || (data == STA) || (data == STO))
   {
       Uart.send(CTL);
       data ^= TRANSPARENCY;
   }
   Uart.send(data);
}

void frame::send(void const * buffer, size_t size)
{
   uint8_t const * pSrc = (uint8_t const *)buffer;
   while(size--)
       send(*pSrc++);
}

void Test(bus_frame::address_t destination)
{
    frame Frame(Uplink, destination);
    Frame.send("Tipa test", sizeof("Tipa test"));
}

[Pavel V. 0]

А почему бы не повесить кольцевой буфер на UART (и опустошать его в прерывании), а сам процесс отправки пакета не обрамить мутексом? Тогда любой процесс сможет оправлять пакеты асинхронно. Что-то типа такого (не wake, но подобный протокол):

Сергей, огромное спасибо! Вы уже не первый раз мне помогаете дельным советом.

 

Идею Вашу понял, очень интересно! Насколько я понял, константы STA и STO служат признаками начала и конца пакета. Почему их отправку Вы поместили именно в конструктор и деструктор, а не в функцию send()?

 

Функция Uart.send() помещает данные в кольцевой буфер, который опустошается в прерываниях? Не могли бы Вы показать как сделано это место? Интересует сама функция Uart.send() и обработчик прерывания.

 

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

[Сергей Борщ 143]

Идею Вашу понял, очень интересно! Насколько я понял, константы STA и STO служат признаками начала и конца пакета. Почему их отправку Вы поместили именно в конструктор и деструктор, а не в функцию send()?

Потому что функцию send() можно вызывать несколько раз во время передачи пакета.

void raw_message::send(frame & link) const
{
    link.send(Type);
    link.send(&Header, sizeof(Header));
    link.send(this + 1, Size);
}
void Test()
{
    frame Frame(Uplink);
    pMessage->send(Frame);
}

Функция Uart.send() помещает данные в кольцевой буфер, который опустошается в прерываниях? Не могли бы Вы показать как сделано это место? Интересует сама функция Uart.send() и обработчик прерывания.

Под MSP430F2618 у меня нет, вот под AVR:

#define UART_H__
#include    <stdint.h>
#include    <avr/io.h>
#include    <avr/pgmspace.h>
#include    <scmRTOS.h>
#define UART_RX_BUFF_SIZE   64
#define UART_TX_BUFF_SIZE   64

ISR(USART_RXC_vect);
ISR(USART_UDRE_vect);


class uart_t
{
public:
   uart_t() {};
   bool hasinput();
   uint8_t receive();
   template <typename T>
   bool receive(T & item, timeout_t timeout = 0)
   {
       return RxBuffer.read((uint8_t * const)&item, sizeof(T), timeout);
   }
   void    send(uint8_t byte);
   void    sendHEX(uint8_t byte);
   void    send(PGM_P pString);
   bool    tx_complete() { return UCSRA & ( 1 << TXC ); }
private:
   friend void USART_RXC_vect();
   friend void USART_UDRE_vect();
   void RXC_Handler();
   void UDRE_Handler();
private:
   OS::channel<uint8_t, UART_RX_BUFF_SIZE> RxBuffer;
   OS::channel<uint8_t, UART_TX_BUFF_SIZE> TxBuffer;
};
inline bool uart_t::hasinput (void)
{
   return RxBuffer.get_count();
}


extern uart_t UART;
#endif  // UART_H__
// ========= uart.cpp ==============
#include    <avr/interrupt.h>
#include	"Hardware.h"
#include    "UART.h"

uint8_t uart_t::receive ()
{
   uint8_t Data;
   RxBuffer.pop(Data);
   return Data;
}

void uart_t::send(uint8_t byte)
{
   TxBuffer.push(byte);
   UCSRB |= (1<<UDRIE);    // UCSRB in io-space, no critical section required
}
#include    <avr/pgmspace.h>
void uart_t::sendHEX(uint8_t byte)
{
   static char const PROGMEM Table[] = "0123456789ABCDEF";
   send(pgm_read_byte(&Table[byte >> 4]));
   send(pgm_read_byte(&Table[byte & 0x0F]));
}
void uart_t::send(PGM_P pString)
{
   char c;
   while( (c = pgm_read_byte(pString++)) )
   {
       send©;
   }
}

inline void uart_t::RXC_Handler()
{
   char RxData = UDR;
   if(RxBuffer.get_free_size())
   {
       RxBuffer.push(RxData);
   }
}

OS_INTERRUPT void USART_RXC_vect()
{
   OS::TISRW isr;
   UART.RXC_Handler();
}

inline void uart_t::UDRE_Handler()
{
   UCSRB &= ~(1<<UDRIE);
   sei();
   {
       if(TxBuffer.get_count())
       {
           uint8_t Data;
           TxBuffer.pop(Data);
           UDR = Data;
           cli();  // avoid recursive call
           UCSRB |= (1<<UDRIE);
       }
   }
}

OS_INTERRUPT void USART_UDRE_vect()
{
   OS::TISRW isr;
   UART.UDRE_Handler();
}

uart_t   UART;

 

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

В AVR иначе и не сделаешь. Там нет возможности программно сбросить флаг готовности передатчика, и мне кажется это правильно: все передали (буфер пуст), нет необходимости в дальнейших прерываниях - запретили прерывания. Появилась необходимость (положили что-то в буфер) - разрешили прерывания, и если передатчик готов (флаг взведен) - сразу побежали в обработчик. А если прерывание было разрешено, опустошило буфер и отключило само себя до того, как в send() его разрешили снова - в обработчике есть проверка на пустоту буфера. Минимум лишних движений.

[dxp 65]

Добавлю свои "пять копеек" к обсуждению.

 

Поскольку UART - устройство асинхронное, что прием и передача друг на друга не "завязаны" и реализуются независимо.

 

Мой опыт применения UART'ов следующий. Приемная часть состоит из двух частей - одна в прерывании приемника, вторая - в отдельном процессе. В прерывании приемника осуществляется прием символа и его анализ: если это первый символ в пакете, то в нем закодирована длина пакета - эта величина фиксируется, и ожидающему процессу сигналится флаг событий, по которому процесс переход в ожидание [пакета] с таймаутом (таймаут рассчитывается, исходя из длины пакета, скорости передачи и т.д.). В прерывании теперь в при приеме очередного символа ведется подсчет входящих символов, и когда их количество достигает нужной величины (закодированной в первом символе пакета), код прерывания сигналит флаг события еще раз. Таким образом, получаем две активизации процесса приемника - на старте пакета и в конце. Т.е. процесс приемника не дергается на приеме каждого символа. В конце же и производится собственно обработка пакета (контрольные суммы, требуемые действия).

 

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

 

Передатчик собственно на ОС никак не "завязан". У передатчика есть функция send, которая формирует пакет на отправку и инициирует оную (в MSP430 для этого достаточно просто разрешить прерывание UART Tx Data Register Empty). И есть обработчик прерываний, который выпихивает данные из буфера передатчика в UART. Т.к. источников передачи может быть несколько и все они могут работать асинхронно и одновременно, то буфер передатчика реализован в виде кольцевого буфера. Код для MSP430:

#pragma vector = USART1TX_VECTOR
__interrupt void TUART::TxBUF_Empty_ISR()
{
    U1TXBUF = TxBuf.pop();
    if(TxBuf.get_count() == 0) DisableTxInt();
}
//---------------------------------------------------------------------------

Из кода видно, что действия очень простые: извлекли очередной байт из буфера, сунули его в регистр данных передатчика, проверили, есть ли еще данные в буфере. Если их нет, то запретили прерывание от передатчика, чтобы оно не возникало, когда регистр данных TxD опустеет.

[Сергей Борщ 143]

Приемная часть состоит из двух частей

Интересный подход. Попробую.

[ReAl 0]

Добавлю свои "пять копеек" к обсуждению.

...

Приемная часть состоит из двух частей - одна в прерывании приемника, вторая - в отдельном процессе.

Аналогично (и про пять копеек, и про части :-) )

Только у меня в рамке SLIP бегает всё, «исторически сложилось». Поэтому прерывание просто помещает в буфер всё от флага до флага, и только в самом конце, по завершении пакета один раз даёт знать об этом процессу.

Более того — прерывание по отдельному байту даже не дёргает счётчик вложенности и не вызывает перепланировщик. До тех пор, пока ему нечего сообщить «наверх», оно отрабатывает как «внеосевое». Как-то так:

void  uart_t::rx_isr()
{
    enum { err_mask = (1 << FE) | (1 << DOR) | (1 << UPE) };

    uint8_t temp = UCSR0A;
    do {
        bool error = (temp & err_mask) != 0;
        uint8_t data = UDR0;
        // rx_update() разбирает подстановки SLIP и в случае конца пакета (или ошибка, не лезет в буфер)
        // возвращает true, и только в этом случае будет создан isrw и при выходе вызван ISR_Exit() c  Kernel.SchedISR();
        if( rx_update(data, error) ) {
            OS::TISRW isrw;
            rx_complete(); // а эта даёт отмашку Event
            return;
        }
        temp = UCSR0A;
    } while( temp & (1 << RXC));
    // А при просто очередном байте вообще никаких лишних движений - мы ничего не сигнализировали, поэтому
    // «по нашей вине» не нужно проверять, не надо ли всплыть какому-то процессу - выходим тихо, никого не дёргая
}

Подобным образом и передача, только там у AVR сразу есть два разных прерывания

void uart_t::tx_isr()
{
    OS::TISRW isrw;
    tx_complete();
}

void uart_t::udre_isr()
{
    if( tx_available() ) { // аналогично, false после передачи концевого флага SLIP_END
        UDR0 = tx_get_byte();
        UCSR0A |= (1 << TXC);
    } else {
        UCSR0B &= ~(1 << UDRIE);
    }
}

[dxp 65]

Только у меня в рамке SLIP бегает всё, «исторически сложилось». Поэтому прерывание просто помещает в буфер всё от флага до флага, и только в самом конце, по завершении пакета один раз даёт знать об этом процессу.

А за таймаутами кто следит? У меня следит процесс приемника, и сигналом для начала слежения служит первый флаг (когда первый символ прилетел). С этого момента процесс впадает в ожидание флага уже с таймаутом.

 

Более того — прерывание по отдельному байту даже не дёргает счётчик вложенности и не вызывает перепланировщик. До тех пор, пока ему нечего сообщить «наверх», оно отрабатывает как «внеосевое».

Да, интересно сделано - пока до необходимости в передаче управления не дошло, код вообще обычный. Свежо. :)

[Sirko 0]

Дабы не создавать похожий топик, спрошу здесь.

Возможно, топикстартеру это будет тоже на руку.

 

Суть в следующем.

Есть некое устройство, собрано на меге с двумя UARTами. Протокол ModBus. Сетевой адрес к примеру 10.

Первый UART - мастер, управляет другими устройствами с адресами 1 - 9.

Второй UART - слейв, соединен с персоналкой. Если запрос с ПК адресуется основному устройству, то выполняется необходимое действие и возвращается ответ. Когда пакет адресован не основному, то он просто транслируется на первый UART и, соответственно, с первого на второй транслируется ответ. Если адрес не соответствует ни одному устройству, то и ответ, соответственно, не возвращается.

 

В настоящее время это хозявство благополучно функционирует без операционки. Криво ли, прямо ли, но с задачей своей справляется. Но не так давно решил я постичь csmRTOS и перевести этот проект под ООП. Пока что четких мыслей по поводу реализации нет. Не хочется изначально стартовать в неверном направлении.

 

Отсюда просьба - поделитесь своими мыслями, кто и как реализовал бы подобную задачу. Я имею ввиду взаимосвязь процессов и объектов, чего, с чем и чем :)

 

[remote_job 0]

Отсюда просьба - поделитесь своими мыслями, кто и как реализовал бы подобную задачу. Я имею ввиду взаимосвязь процессов и объектов, чего, с чем и чем :)

Один процесс на работу с первым уартом, один процесс на работу со вторым. Ну и в зависимости от сложности формирования ответа, можно еще и третий, самый приоритетный, на управление всем этим делом и формирование команд/ответов. Как-то так.

[ReAl 0]

А за таймаутами кто следит?

У меня длительность именно приёма пакета не контролируется.

Либо запрос-ответ, тогда просто после передачи пакета ожидается определённое время собственно ответ, акогда там какой из байтов пришёл — не важно.

Либо ответ-запрос (в смысле поток отправил ответ и ждёт очередного запроса) — тоже потоку особенно нечего делать и он может ждать до бесконечности.

Пока в такую модель укладывалось.

 

 

[Pavel V. 0]

Спасибо за продуктивное обсуждение!

 

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

 

В догонку хочется задать один чайниковский вопрос - как оценить состояние стека во время работы программы? Сегодня целый день боролся с разными магическими багами, пока методом тыка не дошел до увеличения размера стека для потока, который производит обработку полученных сообщений (причем, баг вылезал совсем в другом месте - в обработчике прерывания по передаче). Наверняка ведь есть способ оценить его мгновенное состояние в какой-то момент времени?

 

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

[AHTOXA 18]

Вот тут про стек.