Skip to content

Commit

Permalink
rxrpc: Delay terminal ACK transmission on a client call
Browse files Browse the repository at this point in the history 
Delay terminal ACK transmission on a client call by deferring it to the
connection processor.  This allows it to be skipped if we can send the next
call instead, the first DATA packet of which will implicitly ack this call.

Signed-off-by: David Howells <dhowells@redhat.com>
  • Loading branch information
David Howells committed Nov 24, 2017
1 parent 9faaff5 commit 3136ef4
Show file tree
Hide file tree
Showing 5 changed files with 108 additions and 13 deletions.
17 changes: 17 additions & 0 deletions net/rxrpc/ar-internal.h
View file
Original file line number Diff line number Diff line change
Expand Up @@ -338,8 +338,17 @@ enum rxrpc_conn_flag {
RXRPC_CONN_DONT_REUSE, /* Don't reuse this connection */
RXRPC_CONN_COUNTED, /* Counted by rxrpc_nr_client_conns */
RXRPC_CONN_PROBING_FOR_UPGRADE, /* Probing for service upgrade */
RXRPC_CONN_FINAL_ACK_0, /* Need final ACK for channel 0 */
RXRPC_CONN_FINAL_ACK_1, /* Need final ACK for channel 1 */
RXRPC_CONN_FINAL_ACK_2, /* Need final ACK for channel 2 */
RXRPC_CONN_FINAL_ACK_3, /* Need final ACK for channel 3 */
};

#define RXRPC_CONN_FINAL_ACK_MASK ((1UL << RXRPC_CONN_FINAL_ACK_0) | \
(1UL << RXRPC_CONN_FINAL_ACK_1) | \
(1UL << RXRPC_CONN_FINAL_ACK_2) | \
(1UL << RXRPC_CONN_FINAL_ACK_3))

/*
* Events that can be raised upon a connection.
*/
Expand Down Expand Up @@ -393,6 +402,7 @@ struct rxrpc_connection {
#define RXRPC_ACTIVE_CHANS_MASK ((1 << RXRPC_MAXCALLS) - 1)
struct list_head waiting_calls; /* Calls waiting for channels */
struct rxrpc_channel {
unsigned long final_ack_at; /* Time at which to issue final ACK */
struct rxrpc_call __rcu *call; /* Active call */
u32 call_id; /* ID of current call */
u32 call_counter; /* Call ID counter */
Expand All @@ -404,6 +414,7 @@ struct rxrpc_connection {
};
} channels[RXRPC_MAXCALLS];

struct timer_list timer; /* Conn event timer */
struct work_struct processor; /* connection event processor */
union {
struct rb_node client_node; /* Node in local->client_conns */
Expand Down Expand Up @@ -861,6 +872,12 @@ static inline void rxrpc_put_connection(struct rxrpc_connection *conn)
rxrpc_put_service_conn(conn);
}

static inline void rxrpc_reduce_conn_timer(struct rxrpc_connection *conn,
unsigned long expire_at)
{
timer_reduce(&conn->timer, expire_at);
}

/*
* conn_service.c
*/
Expand Down
18 changes: 18 additions & 0 deletions net/rxrpc/conn_client.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -554,6 +554,11 @@ static void rxrpc_activate_one_channel(struct rxrpc_connection *conn,

trace_rxrpc_client(conn, channel, rxrpc_client_chan_activate);

/* Cancel the final ACK on the previous call if it hasn't been sent yet
* as the DATA packet will implicitly ACK it.
*/
clear_bit(RXRPC_CONN_FINAL_ACK_0 + channel, &conn->flags);

write_lock_bh(&call->state_lock);
if (!test_bit(RXRPC_CALL_TX_LASTQ, &call->flags))
call->state = RXRPC_CALL_CLIENT_SEND_REQUEST;
Expand Down Expand Up @@ -813,6 +818,19 @@ void rxrpc_disconnect_client_call(struct rxrpc_call *call)
goto out_2;
}

/* Schedule the final ACK to be transmitted in a short while so that it
* can be skipped if we find a follow-on call. The first DATA packet
* of the follow on call will implicitly ACK this call.
*/
if (test_bit(RXRPC_CALL_EXPOSED, &call->flags)) {
unsigned long final_ack_at = jiffies + 2;

WRITE_ONCE(chan->final_ack_at, final_ack_at);
smp_wmb(); /* vs rxrpc_process_delayed_final_acks() */
set_bit(RXRPC_CONN_FINAL_ACK_0 + channel, &conn->flags);
rxrpc_reduce_conn_timer(conn, final_ack_at);
}

/* Things are more complex and we need the cache lock. We might be
* able to simply idle the conn or it might now be lurking on the wait
* list. It might even get moved back to the active list whilst we're
Expand Down
74 changes: 61 additions & 13 deletions net/rxrpc/conn_event.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,10 @@
* Retransmit terminal ACK or ABORT of the previous call.
*/
static void rxrpc_conn_retransmit_call(struct rxrpc_connection *conn,
struct sk_buff *skb)
struct sk_buff *skb,
unsigned int channel)
{
struct rxrpc_skb_priv *sp = rxrpc_skb(skb);
struct rxrpc_skb_priv *sp = skb ? rxrpc_skb(skb) : NULL;
struct rxrpc_channel *chan;
struct msghdr msg;
struct kvec iov;
Expand All @@ -48,15 +49,15 @@ static void rxrpc_conn_retransmit_call(struct rxrpc_connection *conn,

_enter("%d", conn->debug_id);

chan = &conn->channels[sp->hdr.cid & RXRPC_CHANNELMASK];
chan = &conn->channels[channel];

/* If the last call got moved on whilst we were waiting to run, just
* ignore this packet.
*/
call_id = READ_ONCE(chan->last_call);
/* Sync with __rxrpc_disconnect_call() */
smp_rmb();
if (call_id != sp->hdr.callNumber)
if (skb && call_id != sp->hdr.callNumber)
return;

msg.msg_name = &conn->params.peer->srx.transport;
Expand All @@ -65,9 +66,9 @@ static void rxrpc_conn_retransmit_call(struct rxrpc_connection *conn,
msg.msg_controllen = 0;
msg.msg_flags = 0;

pkt.whdr.epoch = htonl(sp->hdr.epoch);
pkt.whdr.cid = htonl(sp->hdr.cid);
pkt.whdr.callNumber = htonl(sp->hdr.callNumber);
pkt.whdr.epoch = htonl(conn->proto.epoch);
pkt.whdr.cid = htonl(conn->proto.cid);
pkt.whdr.callNumber = htonl(call_id);
pkt.whdr.seq = 0;
pkt.whdr.type = chan->last_type;
pkt.whdr.flags = conn->out_clientflag;
Expand All @@ -87,11 +88,11 @@ static void rxrpc_conn_retransmit_call(struct rxrpc_connection *conn,
mtu = conn->params.peer->if_mtu;
mtu -= conn->params.peer->hdrsize;
pkt.ack.bufferSpace = 0;
pkt.ack.maxSkew = htons(skb->priority);
pkt.ack.firstPacket = htonl(chan->last_seq);
pkt.ack.previousPacket = htonl(chan->last_seq - 1);
pkt.ack.serial = htonl(sp->hdr.serial);
pkt.ack.reason = RXRPC_ACK_DUPLICATE;
pkt.ack.maxSkew = htons(skb ? skb->priority : 0);
pkt.ack.firstPacket = htonl(chan->last_seq + 1);
pkt.ack.previousPacket = htonl(chan->last_seq);
pkt.ack.serial = htonl(skb ? sp->hdr.serial : 0);
pkt.ack.reason = skb ? RXRPC_ACK_DUPLICATE : RXRPC_ACK_IDLE;
pkt.ack.nAcks = 0;
pkt.info.rxMTU = htonl(rxrpc_rx_mtu);
pkt.info.maxMTU = htonl(mtu);
Expand Down Expand Up @@ -272,7 +273,8 @@ static int rxrpc_process_event(struct rxrpc_connection *conn,
switch (sp->hdr.type) {
case RXRPC_PACKET_TYPE_DATA:
case RXRPC_PACKET_TYPE_ACK:
rxrpc_conn_retransmit_call(conn, skb);
rxrpc_conn_retransmit_call(conn, skb,
sp->hdr.cid & RXRPC_CHANNELMASK);
return 0;

case RXRPC_PACKET_TYPE_BUSY:
Expand Down Expand Up @@ -378,6 +380,48 @@ static void rxrpc_secure_connection(struct rxrpc_connection *conn)
_leave(" [aborted]");
}

/*
* Process delayed final ACKs that we haven't subsumed into a subsequent call.
*/
static void rxrpc_process_delayed_final_acks(struct rxrpc_connection *conn)
{
unsigned long j = jiffies, next_j;
unsigned int channel;
bool set;

again:
next_j = j + LONG_MAX;
set = false;
for (channel = 0; channel < RXRPC_MAXCALLS; channel++) {
struct rxrpc_channel *chan = &conn->channels[channel];
unsigned long ack_at;

if (!test_bit(RXRPC_CONN_FINAL_ACK_0 + channel, &conn->flags))
continue;

smp_rmb(); /* vs rxrpc_disconnect_client_call */
ack_at = READ_ONCE(chan->final_ack_at);

if (time_before(j, ack_at)) {
if (time_before(ack_at, next_j)) {
next_j = ack_at;
set = true;
}
continue;
}

if (test_and_clear_bit(RXRPC_CONN_FINAL_ACK_0 + channel,
&conn->flags))
rxrpc_conn_retransmit_call(conn, NULL, channel);
}

j = jiffies;
if (time_before_eq(next_j, j))
goto again;
if (set)
rxrpc_reduce_conn_timer(conn, next_j);
}

/*
* connection-level event processor
*/
Expand All @@ -394,6 +438,10 @@ void rxrpc_process_connection(struct work_struct *work)
if (test_and_clear_bit(RXRPC_CONN_EV_CHALLENGE, &conn->events))
rxrpc_secure_connection(conn);

/* Process delayed ACKs whose time has come. */
if (conn->flags & RXRPC_CONN_FINAL_ACK_MASK)
rxrpc_process_delayed_final_acks(conn);

/* go through the conn-level event packets, releasing the ref on this
* connection that each one has when we've finished with it */
while ((skb = skb_dequeue(&conn->rx_queue))) {
Expand Down
10 changes: 10 additions & 0 deletions net/rxrpc/conn_object.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,14 @@ unsigned int rxrpc_connection_expiry = 10 * 60;

static void rxrpc_destroy_connection(struct rcu_head *);

static void rxrpc_connection_timer(struct timer_list *timer)
{
struct rxrpc_connection *conn =
container_of(timer, struct rxrpc_connection, timer);

rxrpc_queue_conn(conn);
}

/*
* allocate a new connection
*/
Expand All @@ -38,6 +46,7 @@ struct rxrpc_connection *rxrpc_alloc_connection(gfp_t gfp)
INIT_LIST_HEAD(&conn->cache_link);
spin_lock_init(&conn->channel_lock);
INIT_LIST_HEAD(&conn->waiting_calls);
timer_setup(&conn->timer, &rxrpc_connection_timer, 0);
INIT_WORK(&conn->processor, &rxrpc_process_connection);
INIT_LIST_HEAD(&conn->proc_link);
INIT_LIST_HEAD(&conn->link);
Expand Down Expand Up @@ -332,6 +341,7 @@ static void rxrpc_destroy_connection(struct rcu_head *rcu)

_net("DESTROY CONN %d", conn->debug_id);

del_timer_sync(&conn->timer);
rxrpc_purge_queue(&conn->rx_queue);

conn->security->clear(conn);
Expand Down
2 changes: 2 additions & 0 deletions net/rxrpc/recvmsg.c
View file
Original file line number Diff line number Diff line change
Expand Up @@ -144,11 +144,13 @@ static void rxrpc_end_rx_phase(struct rxrpc_call *call, rxrpc_serial_t serial)
trace_rxrpc_receive(call, rxrpc_receive_end, 0, call->rx_top);
ASSERTCMP(call->rx_hard_ack, ==, call->rx_top);

#if 0 // TODO: May want to transmit final ACK under some circumstances anyway
if (call->state == RXRPC_CALL_CLIENT_RECV_REPLY) {
rxrpc_propose_ACK(call, RXRPC_ACK_IDLE, 0, serial, true, false,
rxrpc_propose_ack_terminal_ack);
rxrpc_send_ack_packet(call, false);
}
#endif

write_lock_bh(&call->state_lock);

Expand Down

0 comments on commit 3136ef4

Please sign in to comment.