The RX window indicates how many reception buffers the peer has available for that cport. The other peer must not send more messages than that window, or the chances of those messages being lost is very high, leading to retransmissions and poor performance.
The RX window is associated with the ack number, and indicates the valid range of sequence number the other peer can use:
Ack RX window Valid Sequence Range
X 0 None X 1 X X 2 X -> X+1
So everytime an ack is received, the driver evaluates if the valid sequence range has changed and if so pops message from its holding queue.
As the skb is moved to another queue, it cannot be passed directly to the lower layer anymore, instead a clone is passed.
Signed-off-by: Damien Riégel damien.riegel@silabs.com --- drivers/greybus/cpc/cpc.h | 9 ++++ drivers/greybus/cpc/cport.c | 5 ++ drivers/greybus/cpc/header.c | 88 ++++++++++++++++++++++++++++++++++ drivers/greybus/cpc/header.h | 6 +++ drivers/greybus/cpc/host.c | 9 ---- drivers/greybus/cpc/host.h | 1 - drivers/greybus/cpc/protocol.c | 72 +++++++++++++++++++++++++--- 7 files changed, 173 insertions(+), 17 deletions(-)
diff --git a/drivers/greybus/cpc/cpc.h b/drivers/greybus/cpc/cpc.h index f1669585c45..e8cbe916630 100644 --- a/drivers/greybus/cpc/cpc.h +++ b/drivers/greybus/cpc/cpc.h @@ -18,6 +18,7 @@ * @cpc_hd: pointer to the CPC host device this cport belongs to * @lock: mutex to synchronize accesses to tcb and other attributes * @holding_queue: list of frames queued to be sent + * @retx_queue: list of frames sent and waiting for acknowledgment * @tcb: Transmission Control Block */ struct cpc_cport { @@ -27,12 +28,20 @@ struct cpc_cport { struct mutex lock; /* Synchronize access to state variables */
struct sk_buff_head holding_queue; + struct sk_buff_head retx_queue;
/* + * @send_wnd: send window, maximum number of frames that the remote can accept. + * TX frames should have a sequence in the range [send_una; send_una + send_wnd] + * @send_nxt: send next, the next sequence number that will be used for transmission + * @send_una: send unacknowledged, the oldest unacknowledged sequence number * @ack: current acknowledge number * @seq: current sequence number */ struct { + u8 send_wnd; + u8 send_nxt; + u8 send_una; u8 ack; u8 seq; } tcb; diff --git a/drivers/greybus/cpc/cport.c b/drivers/greybus/cpc/cport.c index f850da7acfb..2f37bd035b6 100644 --- a/drivers/greybus/cpc/cport.c +++ b/drivers/greybus/cpc/cport.c @@ -16,6 +16,9 @@ static void cpc_cport_tcb_reset(struct cpc_cport *cport) { cport->tcb.ack = 0; cport->tcb.seq = 0; + cport->tcb.send_nxt = 0; + cport->tcb.send_una = 0; + cport->tcb.send_wnd = 1; }
/** @@ -38,12 +41,14 @@ struct cpc_cport *cpc_cport_alloc(u16 cport_id, gfp_t gfp_mask)
mutex_init(&cport->lock); skb_queue_head_init(&cport->holding_queue); + skb_queue_head_init(&cport->retx_queue);
return cport; }
void cpc_cport_release(struct cpc_cport *cport) { + skb_queue_purge(&cport->retx_queue); skb_queue_purge(&cport->holding_queue); kfree(cport); } diff --git a/drivers/greybus/cpc/header.c b/drivers/greybus/cpc/header.c index 8875a6fed26..43038f103b5 100644 --- a/drivers/greybus/cpc/header.c +++ b/drivers/greybus/cpc/header.c @@ -22,6 +22,17 @@ bool cpc_header_is_control(const struct cpc_header *hdr) return hdr->ctrl_flags & CPC_HEADER_CONTROL_IS_CONTROL_MASK; }
+/** + * cpc_header_get_recv_wnd() - Get the receive window. + * @hdr: CPC header. + * + * Return: Receive window. + */ +u8 cpc_header_get_recv_wnd(const struct cpc_header *hdr) +{ + return hdr->recv_wnd; +} + /** * cpc_header_get_seq() - Get the sequence number. * @hdr: CPC header. @@ -33,6 +44,17 @@ u8 cpc_header_get_seq(const struct cpc_header *hdr) return hdr->seq; }
+/** + * cpc_header_get_ack() - Get the acknowledge number. + * @hdr: CPC header. + * + * Return: Acknowledge number. + */ +u8 cpc_header_get_ack(const struct cpc_header *hdr) +{ + return hdr->ack; +} + /** * cpc_header_get_req_ack() - Get the request acknowledge frame flag. * @hdr: CPC header. @@ -56,3 +78,69 @@ u8 cpc_header_encode_ctrl_flags(bool control, bool req_ack) return FIELD_PREP(CPC_HEADER_CONTROL_IS_CONTROL_MASK, control) | FIELD_PREP(CPC_HEADER_CONTROL_REQ_ACK_MASK, req_ack); } + +/** + * cpc_header_get_frames_acked_count() - Get frames to be acknowledged. + * @seq: Current sequence number of the endpoint. + * @ack: Acknowledge number of the received frame. + * + * Return: Frames to be acknowledged. + */ +u8 cpc_header_get_frames_acked_count(u8 seq, u8 ack) +{ + u8 frames_acked_count; + + /* Find number of frames acknowledged with ACK number. */ + if (ack > seq) { + frames_acked_count = ack - seq; + } else { + frames_acked_count = 256 - seq; + frames_acked_count += ack; + } + + return frames_acked_count; +} + +/** + * cpc_header_number_in_window() - Test if a number is within a window. + * @start: Start of the window. + * @end: Window size. + * @n: Number to be tested. + * + * Given the start of the window and its size, test if the number is + * in the range [start; start + wnd). + * + * @return True if start <= n <= start + wnd - 1 (modulo 256), otherwise false. + */ +bool cpc_header_number_in_window(u8 start, u8 wnd, u8 n) +{ + u8 end; + + if (wnd == 0) + return false; + + end = start + wnd - 1; + + return cpc_header_number_in_range(start, end, n); +} + +/** + * cpc_header_number_in_range() - Test if a number is between start and end (included). + * @start: Lowest limit. + * @end: Highest limit inclusively. + * @n: Number to be tested. + * + * @return True if start <= n <= end (modulo 256), otherwise false. + */ +bool cpc_header_number_in_range(u8 start, u8 end, u8 n) +{ + if (end >= start) { + if (n < start || n > end) + return false; + } else { + if (n > end && n < start) + return false; + } + + return true; +} diff --git a/drivers/greybus/cpc/header.h b/drivers/greybus/cpc/header.h index 0c9f6e56524..053dc3707d0 100644 --- a/drivers/greybus/cpc/header.h +++ b/drivers/greybus/cpc/header.h @@ -42,8 +42,14 @@ struct cpc_header { #define GREYBUS_HEADER_SIZE (sizeof(struct gb_operation_msg_hdr))
bool cpc_header_is_control(const struct cpc_header *hdr); +u8 cpc_header_get_recv_wnd(const struct cpc_header *hdr); u8 cpc_header_get_seq(const struct cpc_header *hdr); +u8 cpc_header_get_ack(const struct cpc_header *hdr); bool cpc_header_get_req_ack(const struct cpc_header *hdr); u8 cpc_header_encode_ctrl_flags(bool control, bool req_ack);
+u8 cpc_header_get_frames_acked_count(u8 seq, u8 ack); +bool cpc_header_number_in_window(u8 start, u8 wnd, u8 n); +bool cpc_header_number_in_range(u8 start, u8 end, u8 n); + #endif diff --git a/drivers/greybus/cpc/host.c b/drivers/greybus/cpc/host.c index ec43d33dfc6..a7715c0a960 100644 --- a/drivers/greybus/cpc/host.c +++ b/drivers/greybus/cpc/host.c @@ -199,15 +199,6 @@ void cpc_hd_del(struct cpc_host_device *cpc_hd) } EXPORT_SYMBOL_GPL(cpc_hd_del);
-void cpc_hd_message_sent(struct sk_buff *skb, int status) -{ - struct cpc_host_device *cpc_hd = CPC_SKB_CB(skb)->cport->cpc_hd; - struct gb_host_device *hd = cpc_hd->gb_hd; - - greybus_message_sent(hd, CPC_SKB_CB(skb)->gb_message, status); -} -EXPORT_SYMBOL_GPL(cpc_hd_message_sent); - void cpc_hd_rcvd(struct cpc_host_device *cpc_hd, struct sk_buff *skb) { struct gb_operation_msg_hdr *gb_hdr; diff --git a/drivers/greybus/cpc/host.h b/drivers/greybus/cpc/host.h index cc835f5298b..8f05877b2be 100644 --- a/drivers/greybus/cpc/host.h +++ b/drivers/greybus/cpc/host.h @@ -46,7 +46,6 @@ int cpc_hd_add(struct cpc_host_device *cpc_hd); void cpc_hd_put(struct cpc_host_device *cpc_hd); void cpc_hd_del(struct cpc_host_device *cpc_hd); void cpc_hd_rcvd(struct cpc_host_device *cpc_hd, struct sk_buff *skb); -void cpc_hd_message_sent(struct sk_buff *skb, int status);
int cpc_hd_send_skb(struct cpc_host_device *cpc_hd, struct sk_buff *skb);
diff --git a/drivers/greybus/cpc/protocol.c b/drivers/greybus/cpc/protocol.c index ef8ff0cac24..b5de63d0be8 100644 --- a/drivers/greybus/cpc/protocol.c +++ b/drivers/greybus/cpc/protocol.c @@ -14,7 +14,7 @@ static bool cpc_skb_is_sequenced(struct sk_buff *skb) return CPC_SKB_CB(skb)->cpc_flags & CPC_SKB_FLAG_REQ_ACK; }
-static void cpc_protocol_prepare_header(struct sk_buff *skb, u8 ack) +static void cpc_protocol_prepare_header(struct sk_buff *skb, u8 ack, u8 recv_window) { struct cpc_header *hdr;
@@ -22,7 +22,7 @@ static void cpc_protocol_prepare_header(struct sk_buff *skb, u8 ack)
hdr = (struct cpc_header *)skb->data; hdr->ack = ack; - hdr->recv_wnd = 0; + hdr->recv_wnd = recv_window; hdr->seq = CPC_SKB_CB(skb)->seq; hdr->ctrl_flags = cpc_header_encode_ctrl_flags(!CPC_SKB_CB(skb)->gb_message, cpc_skb_is_sequenced(skb)); @@ -46,11 +46,47 @@ static void cpc_protocol_queue_ack(struct cpc_cport *cport, u8 ack) gb_hdr->size = sizeof(*gb_hdr); cpc_cport_pack(gb_hdr, cport->id);
- cpc_protocol_prepare_header(skb, ack); + cpc_protocol_prepare_header(skb, ack, CPC_HEADER_MAX_RX_WINDOW);
cpc_hd_send_skb(cport->cpc_hd, skb); }
+static void __cpc_protocol_receive_ack(struct cpc_cport *cport, u8 recv_wnd, u8 ack) +{ + struct gb_host_device *gb_hd = cport->cpc_hd->gb_hd; + struct sk_buff *skb; + u8 acked_frames; + + cport->tcb.send_wnd = recv_wnd; + + skb = skb_peek(&cport->retx_queue); + if (!skb) + return; + + /* Return if no frame to ACK. */ + if (!cpc_header_number_in_range(cport->tcb.send_una, cport->tcb.send_nxt, ack)) + return; + + /* Calculate how many frames will be ACK'd. */ + acked_frames = cpc_header_get_frames_acked_count(CPC_SKB_CB(skb)->seq, ack); + + for (u8 i = 0; i < acked_frames; i++) { + skb = skb_dequeue(&cport->retx_queue); + if (!skb) { + dev_err_ratelimited(cpc_hd_dev(cport->cpc_hd), + "pending ack queue shorter than expected"); + break; + } + + if (CPC_SKB_CB(skb)->gb_message) + greybus_message_sent(gb_hd, CPC_SKB_CB(skb)->gb_message, 0); + + kfree_skb(skb); + + cport->tcb.send_una++; + } +} + void cpc_protocol_on_data(struct cpc_cport *cport, struct sk_buff *skb) { struct cpc_header *cpc_hdr = (struct cpc_header *)skb->data; @@ -61,6 +97,9 @@ void cpc_protocol_on_data(struct cpc_cport *cport, struct sk_buff *skb)
mutex_lock(&cport->lock);
+ __cpc_protocol_receive_ack(cport, cpc_header_get_recv_wnd(cpc_hdr), + cpc_header_get_ack(cpc_hdr)); + if (require_ack) { expected_seq = seq == cport->tcb.ack; if (expected_seq) @@ -72,6 +111,8 @@ void cpc_protocol_on_data(struct cpc_cport *cport, struct sk_buff *skb)
ack = cport->tcb.ack;
+ __cpc_protocol_write_head(cport); + mutex_unlock(&cport->lock);
/* Ack no matter if the sequence was valid or not, to resync with remote */ @@ -85,9 +126,10 @@ void cpc_protocol_on_data(struct cpc_cport *cport, struct sk_buff *skb) } }
-static void __cpc_protocol_write_skb(struct cpc_cport *cport, struct sk_buff *skb, u8 ack) +static void __cpc_protocol_write_skb(struct cpc_cport *cport, struct sk_buff *skb, u8 ack, + u8 recv_window) { - cpc_protocol_prepare_header(skb, ack); + cpc_protocol_prepare_header(skb, ack, recv_window);
cpc_hd_send_skb(cport->cpc_hd, skb); } @@ -96,14 +138,30 @@ static void __cpc_protocol_write_skb(struct cpc_cport *cport, struct sk_buff *sk void __cpc_protocol_write_head(struct cpc_cport *cport) { struct sk_buff *skb; - u8 ack; + u8 ack, send_una, send_wnd;
ack = cport->tcb.ack; + send_una = cport->tcb.send_una; + send_wnd = cport->tcb.send_wnd;
/* For each SKB in the holding queue, clone it and pass it to lower layer */ while ((skb = skb_peek(&cport->holding_queue))) { + struct sk_buff *out_skb; + + /* Skip this skb if it must be acked but the remote has no room for it. */ + if (!cpc_header_number_in_window(send_una, send_wnd, CPC_SKB_CB(skb)->seq)) + break; + + /* Clone and send out the skb */ + out_skb = skb_clone(skb, GFP_KERNEL); + if (!out_skb) + return; + skb_unlink(skb, &cport->holding_queue);
- __cpc_protocol_write_skb(cport, skb, ack); + __cpc_protocol_write_skb(cport, out_skb, ack, CPC_HEADER_MAX_RX_WINDOW); + + cport->tcb.send_nxt++; + skb_queue_tail(&cport->retx_queue, skb); } }