diff --git a/klippy/chelper/serialqueue.c b/klippy/chelper/serialqueue.c index ec6dd014..4c40cc11 100644 --- a/klippy/chelper/serialqueue.c +++ b/klippy/chelper/serialqueue.c @@ -392,6 +392,7 @@ struct serialqueue { #define MIN_RTO 0.025 #define MAX_RTO 5.000 +#define MAX_PENDING_BLOCKS 12 #define MIN_REQTIME_DELTA 0.250 #define MIN_BACKGROUND_DELTA 0.005 #define IDLE_QUERY_TIME 1.0 @@ -502,18 +503,21 @@ update_receive_seq(struct serialqueue *sq, double eventtime, uint64_t rseq) } // Process a well formed input message -static void +static int handle_message(struct serialqueue *sq, double eventtime, int len) { // Calculate receive sequence number uint64_t rseq = ((sq->receive_seq & ~MESSAGE_SEQ_MASK) | (sq->input_buf[MESSAGE_POS_SEQ] & MESSAGE_SEQ_MASK)); - if (rseq < sq->receive_seq) - rseq += MESSAGE_SEQ_MASK+1; - - if (rseq != sq->receive_seq) + if (rseq != sq->receive_seq) { // New sequence number + if (rseq < sq->receive_seq) + rseq += MESSAGE_SEQ_MASK+1; + if (rseq > sq->send_seq && sq->receive_seq != 1) + // An ack for a message not sent? Out of order message? + return -1; update_receive_seq(sq, eventtime, rseq); + } // Check for pending messages on notify_queue int must_wake = 0; @@ -553,6 +557,7 @@ handle_message(struct serialqueue *sq, double eventtime, int len) if (must_wake) check_wake_receive(sq); + return 0; } // Callback for input activity on the serial fd @@ -568,26 +573,29 @@ input_event(struct serialqueue *sq, double eventtime) } sq->input_pos += ret; for (;;) { - ret = check_message(&sq->need_sync, sq->input_buf, sq->input_pos); - if (!ret) + int len = check_message(&sq->need_sync, sq->input_buf, sq->input_pos); + if (!len) // Need more data return; - if (ret > 0) { + if (len > 0) { // Received a valid message pthread_mutex_lock(&sq->lock); - handle_message(sq, eventtime, ret); - sq->bytes_read += ret; + int ret = handle_message(sq, eventtime, len); + if (ret) + sq->bytes_invalid += len; + else + sq->bytes_read += len; pthread_mutex_unlock(&sq->lock); } else { // Skip bad data at beginning of input - ret = -ret; + len = -len; pthread_mutex_lock(&sq->lock); - sq->bytes_invalid += ret; + sq->bytes_invalid += len; pthread_mutex_unlock(&sq->lock); } - sq->input_pos -= ret; + sq->input_pos -= len; if (sq->input_pos) - memmove(sq->input_buf, &sq->input_buf[ret], sq->input_pos); + memmove(sq->input_buf, &sq->input_buf[len], sq->input_pos); } } @@ -726,7 +734,7 @@ build_and_send_command(struct serialqueue *sq, double eventtime) static double check_send_command(struct serialqueue *sq, double eventtime) { - if (sq->send_seq - sq->receive_seq >= MESSAGE_SEQ_MASK + if (sq->send_seq - sq->receive_seq >= MAX_PENDING_BLOCKS && sq->receive_seq != (uint64_t)-1) // Need an ack before more messages can be sent return PR_NEVER;