Блог

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

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

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

Интересно, что я смог получать больше или меньше сообщений после изменения MAX_MSG_SIZE значения. Кроме того, на справочной странице говорится только о том, что размер буфера приема больше, чем mq_msgsize .

Означает ли это, что последовательный порт каким-то образом засоряется, особенно если приемный конец последовательного порта не считывает данные?

Я не вижу никаких ошибок ни в одной из функций.

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

 // common.h

#define MAX_MESSAGES                            10
#define MAX_MSG_SIZE                            72
#define MSG_BUFFER_SIZE                         MAX_MSG_SIZE   10
#define QUEUE_PERMISSIONS                       0660

#define COMMON_QUEUE                            "/MSG_QUEUE"

typedef enum
{
    LEFT,
    RIGHT
} CmdType;

typedef struct
{
    CmdType cmd;
    uint16_t length;
    uint8_t payload[64];
} Pkt;

 

Потребитель

 #include "common.h"

int fd;  // file descriptor for serial port

static struct mq_attr attr = {.mq_flags = 0, .mq_maxmsg = MAX_MESSAGES, .mq_msgsize = MAX_MSG_SIZE, .mq_curmsgs = 0};

void *Consumer(void *args)
{   
    uint8_t rcvBuffer[MSG_BUFFER_SIZE];
    mqd_t *mqd = (mqd_t *) args;
    int ret;

    while(1)
    {   
        ret = mq_receive(*mqd, rcvBuffer, sizeof(rcvBuffer), 0);
        if (ret == -1)
        {
            perror ("mq_receive failed");
        }
        printf ("Received data: %dn", *rcvBuffer);

        // serial write 
        int sizeWritten = write(fd, rcvBuffer, sizeof(rcvBuffer));
        if (sizeWritten < 0)
        { 
            perror ("Write to serial failed");
        }
    }
}

int UartInit()
{
    fd = open("/dev/ttyUSB4", O_RDWR); 

    if (fd == -1)
    {
        perror ("Serial port failed to open");
        return -1;
    }

    if(tcgetattr(fd, amp;tty) != 0)    
    {
      perror("Error from tcgetattr");
      return -1;
    }

    cfsetspeed(tty, B115200);     // baud rate to 115200

    tty->c_cflag |= CS8;

    tty->c_cflag amp;= ~CSTOPB;

    tty->c_cflag |= CRTSCTS;       // HW flow control

    tty->c_cflag amp;= ~PARENB;

    tty.c_iflag |= ICRNL;                // enable translating carriage return to newline on input
    tty.c_iflag |= IGNBRK;               // ignore break condition
    tty.c_iflag amp;= ~BRKINT;
    tty.c_iflag amp;= ~INPCK;               // disable input parity check
    tty.c_iflag amp;= ~PARMRK;             // ignore input bytes with parity or framing errors are marked when passed to the program
    tty.c_iflag amp;= ~ISTRIP;             // ignore stripping off the 8th bit
    
    tty.c_oflag |= OCRNL;              // Map CR to NL on output.
    tty.c_oflag |= ONLCR;             // Map NL to CR-NL on output.
    tty.c_oflag amp;= ~OPOST;             // Disable implementation-defined output processing
    
    tty.c_lflag amp;= ~ICANON;            // Disable canonical mode
    tty.c_lflag amp;= ~ISIG;
    tty.c_lflag amp;= ~IEXTEN;            // Disable implementation-defined input processing
    tty.c_lflag amp;= ~(ECHO|ECHOE|ECHOK|ECHONL|ECHOCTL|ECHOPRT|ECHOKE);
    
    // settimg serial read blocking behavior
    tty.c_cc[VTIME] = 0; 
   
    if (tcsetattr(fd, TCSANOW, amp;tty) != 0)
    {
        printf("Error %i from tcsetattr: %sn", errno, strerror(errno));
        return -1;
    }
    return 1
}

int main() 
{
    mqd_t mqd;
    
    int ret = UartInit(); 
    
    if ((mqd = mq_open(COMMON_QUEUE, O_RDWR | O_CREAT, QUEUE_PERMISSIONS, amp;attr)) == -1) 
    {
        perror ("Message queue failed to instantiate");
        return -1;
    }
        
    if (pthread_create(amp;consumerTd, NULL, Consumer, (void *) amp;mqd) != 0)
    {
        return -1;
    }
    
    pthread_join(consumerTd, 0);
    return 0;
}
 

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

Я рекомендую проверить, чтобы убедиться, что обе стороны соединения, используя тот же поток управления настройками, что ваш серийный кабель на самом деле есть поток управления штыри подключены (много недорогих лишь подключить передатчик/приемник/заземление линий), и что ваш последовательный порт поддерживает аппаратное управление потоком данных (USB-последовательный адаптеры часто не надо). Отключение управления потоком позволит избежать зависания, но вы можете передавать данные, когда другая сторона не готова их принимать.