镜像站点 · 本页由第三方 GitHub 只读镜像提供,非 GitHub 官方站点,不接受任何登录或凭据输入。前往 github.com
Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions homa_incoming.c
Original file line number Diff line number Diff line change
Expand Up @@ -1085,9 +1085,8 @@ void homa_need_ack_pkt(struct sk_buff *skb, struct homa_sock *hsk,
ack.common.sport = h->dport;
ack.common.dport = h->sport;
ack.common.sender_id = cpu_to_be64(id);
ack.num_acks = htons(homa_peer_get_acks(route->peer,
HOMA_MAX_ACKS_PER_PKT,
ack.acks));
ack.num_acks = htons(homa_route_get_acks(route, HOMA_MAX_ACKS_PER_PKT,
ack.acks));
if (rpc)
homa_rpc_unlock(rpc);
__homa_xmit_control(&ack, sizeof(ack), route, hsk);
Expand Down
10 changes: 5 additions & 5 deletions homa_metrics.c
Original file line number Diff line number Diff line change
Expand Up @@ -376,10 +376,10 @@ char *homa_metrics_print(void)
"Socket lock misses\n");
M("socket_lock_miss_cycles", m->socket_lock_miss_cycles,
"Time lost waiting for socket locks\n");
M("peer_ack_lock_misses", m->peer_ack_lock_misses,
"Misses on peer ack locks\n");
M("peer_ack_lock_miss_cycles", m->peer_ack_lock_miss_cycles,
"Time lost waiting for peer ack locks\n");
M("route_ack_lock_misses", m->route_ack_lock_misses,
"Misses on route ack locks\n");
M("route_ack_lock_miss_cycles", m->route_ack_lock_miss_cycles,
"Time lost waiting for route ack locks\n");
M("grant_locks", m->grant_locks,
"Grant lock acquisitions\n");
M("grant_lock_misses", m->grant_lock_misses,
Expand All @@ -401,7 +401,7 @@ char *homa_metrics_print(void)
M("reaper_calls", m->reaper_calls,
"Reaper invocations that were not disabled\n");
M("ack_overflows", m->ack_overflows,
"Explicit ACKs sent because peer->acks was full\n");
"Explicit ACKs sent because route->acks was full\n");
M("ignored_need_acks", m->ignored_need_acks,
"NEED_ACKs ignored because RPC result not yet received\n");
M("bpool_lock_misses", m->bpool_lock_misses,
Expand Down
13 changes: 7 additions & 6 deletions homa_metrics.h
Original file line number Diff line number Diff line change
Expand Up @@ -650,15 +650,16 @@ struct homa_metrics {
u64 socket_lock_misses;

/**
* @peer_ack_lock_miss_cycles: total time spent waiting for peer lock misses.
* @route_ack_lock_miss_cycles: total time spent waiting for route ack
* lock misses.
*/
u64 peer_ack_lock_miss_cycles;
u64 route_ack_lock_miss_cycles;

/**
* @peer_ack_lock_misses: total number of times that Homa had to wait
* to acquire the lock used for managing acks for a peer.
* @route_ack_lock_misses: total number of times that Homa had to wait
* to acquire the lock used for managing acks for a route.
*/
u64 peer_ack_lock_misses;
u64 route_ack_lock_misses;

/**
* @grant_lock_miss_cycles: total time spent waiting for grant lock
Expand Down Expand Up @@ -722,7 +723,7 @@ struct homa_metrics {
u64 reaper_calls;

/**
* @ack_overflows: total number of times that homa_peer_add_ack
* @ack_overflows: total number of times that homa_route_add_ack
* found insufficient space for the new id and hence had to send an
* ACK message.
*/
Expand Down
2 changes: 1 addition & 1 deletion homa_outgoing.c
Original file line number Diff line number Diff line change
Expand Up @@ -267,7 +267,7 @@ struct sk_buff *homa_tx_skb_alloc(struct homa_rpc *rpc, u32 offset, u32 *end)
sizeof(struct homa_seg_hdr));
h->common.sender_id = cpu_to_be64(rpc->id);
h->msg_length = htonl(rpc->msgout.length);
homa_peer_get_acks(rpc->route->peer, 1, &h->ack);
homa_route_get_acks(rpc->route, 1, &h->ack);
IF_NO_STRIP(h->cutoff_version = rpc->route->peer->cutoff_version);
if (offset < rpc->msgout.next_xmit_offset)
h->retransmit = 1;
Expand Down
78 changes: 39 additions & 39 deletions homa_peer.c
Original file line number Diff line number Diff line change
Expand Up @@ -243,7 +243,6 @@ struct homa_peer *homa_peer_alloc(struct homa_sock *hsk,
peer->ht_key.addr = *addr;
peer->ht_key.hnet = hsk->hnet;
refcount_set(&peer->refs, 1);
spin_lock_init(&peer->lock);
#ifndef __STRIP__ /* See strip.py */
peer->unsched_cutoffs[HOMA_MAX_PRIORITIES - 1] = 0;
peer->unsched_cutoffs[HOMA_MAX_PRIORITIES - 2] = INT_MAX;
Expand Down Expand Up @@ -769,81 +768,82 @@ void homa_peer_set_cutoffs(struct homa_peer *peer, int c0, int c1, int c2,
}

/**
* homa_peer_lock_slow() - This function implements the slow path for
* acquiring a peer's @lock. It is invoked when the lock isn't
* homa_route_lock_slow() - This function implements the slow path for
* acquiring a route's @lock. It is invoked when the lock isn't
* immediately available. It waits for the lock, but also records statistics
* about the waiting time.
* @peer: Peer to lock.
* @route: Route to lock.
*/
void homa_peer_lock_slow(struct homa_peer *peer)
__acquires(peer->lock)
void homa_route_lock_slow(struct homa_route *route)
__acquires(route->lock)
{
u64 start = homa_clock();

tt_record("beginning wait for peer lock");
spin_lock_bh(&peer->lock);
tt_record("ending wait for peer lock");
INC_METRIC(peer_ack_lock_misses, 1);
INC_METRIC(peer_ack_lock_miss_cycles, homa_clock() - start);
tt_record("beginning wait for route lock");
spin_lock_bh(&route->lock);
tt_record("ending wait for route lock");
INC_METRIC(route_ack_lock_misses, 1);
INC_METRIC(route_ack_lock_miss_cycles, homa_clock() - start);
}
#endif /* See strip.py */

/**
* homa_peer_add_ack() - Add a given RPC to the list of unacked
* RPCs for its server. Once this method has been invoked, it's safe
* homa_route_add_ack() - Add a given RPC to the list of unacked
* RPCs for its route. Once this method has been invoked, it's safe
* to delete the RPC, since it will eventually be acked to the server.
* @rpc: Client RPC that has now completed. Must be locked by caller.
*/
void homa_peer_add_ack(struct homa_rpc *rpc)
void homa_route_add_ack(struct homa_rpc *rpc)
__must_hold(rpc->bucket->lock)
{
struct homa_peer *peer = rpc->route->peer;
struct homa_route *route = rpc->route;
struct homa_ack_hdr ack;

homa_peer_lock(peer);
if (peer->num_acks < HOMA_MAX_ACKS_PER_PKT) {
peer->acks[peer->num_acks].client_id = cpu_to_be64(rpc->id);
peer->acks[peer->num_acks].server_port = htons(rpc->dport);
peer->num_acks++;
homa_peer_unlock(peer);
homa_route_lock(route);
if (route->num_acks < HOMA_MAX_ACKS_PER_PKT) {
route->acks[route->num_acks].client_id = cpu_to_be64(rpc->id);
route->acks[route->num_acks].server_port = htons(rpc->dport);
route->num_acks++;
homa_route_unlock(route);
return;
}

/* The peer has filled up; send an ACK message to empty it. The
/* The route has filled up; send an ACK message to empty it. The
* RPC in the message header will also be considered ACKed.
*/
INC_METRIC(ack_overflows, 1);
memcpy(ack.acks, peer->acks, sizeof(peer->acks));
ack.num_acks = htons(peer->num_acks);
peer->num_acks = 0;
homa_peer_unlock(peer);
memcpy(ack.acks, route->acks, sizeof(route->acks));
ack.num_acks = htons(route->num_acks);
route->num_acks = 0;
homa_route_unlock(route);
homa_xmit_control(ACK, &ack, sizeof(ack), rpc);
}

/**
* homa_peer_get_acks() - Copy acks out of a peer, and remove them from the
* peer.
* @peer: Peer to check for possible unacked RPCs.
* homa_route_get_acks() - Copy acks out of a route, and remove them from the
* route.
* @route: Route to check for possible unacked RPCs.
* @count: Maximum number of acks to return.
* @dst: The acks are copied to this location.
*
* Return: The number of acks extracted from the peer (<= count).
* Return: The number of acks extracted from the route (<= count).
*/
int homa_peer_get_acks(struct homa_peer *peer, int count, struct homa_ack *dst)
int homa_route_get_acks(struct homa_route *route, int count,
struct homa_ack *dst)
{
/* Don't waste time acquiring the lock if there are no ids available. */
if (peer->num_acks == 0)
if (route->num_acks == 0)
return 0;

homa_peer_lock(peer);
homa_route_lock(route);

if (count > peer->num_acks)
count = peer->num_acks;
memcpy(dst, &peer->acks[peer->num_acks - count],
count * sizeof(peer->acks[0]));
peer->num_acks -= count;
if (count > route->num_acks)
count = route->num_acks;
memcpy(dst, &route->acks[route->num_acks - count],
count * sizeof(route->acks[0]));
route->num_acks -= count;

homa_peer_unlock(peer);
homa_route_unlock(route);
return count;
}

Expand Down
89 changes: 46 additions & 43 deletions homa_peer.h
Original file line number Diff line number Diff line change
Expand Up @@ -156,24 +156,6 @@ struct homa_peer {
*/
struct rhash_head ht_linkage;

/**
* @lock: used to synchronize access to fields in this struct, such
* as @num_acks, @acks, @dst, and @dst_cookie.
*/
spinlock_t lock ____cacheline_aligned_in_smp;

/**
* @num_acks: the number of (initial) entries in @acks that
* currently hold valid information.
*/
int num_acks;

/**
* @acks: info about client RPCs whose results have been completely
* received.
*/
struct homa_ack acks[HOMA_MAX_ACKS_PER_PKT];

#ifndef __STRIP__ /* See strip.py */
/**
* @unsched_cutoffs: priorities to use for unscheduled packets
Expand Down Expand Up @@ -270,7 +252,8 @@ struct homa_route_key {
/**
* struct homa_route - Holds a dst_entry for communicating with a
* particular peer, as well as information needed to cache the dst_entry
* and share it for multiple RPCs. There can be more than one of these
* and share it for multiple RPCs. Also holds acks waiting to be piggybacked
* on packets sent over this route. There can be more than one of these
* objects per peer, which have different homa_route_keys.
*/
struct homa_route {
Expand Down Expand Up @@ -328,18 +311,37 @@ struct homa_route {
*/
unsigned long access_jiffies;

/** @rcu_head: Holds state of a pending call_rcu invocation. */
struct rcu_head rcu_head;

/**
* @lock: used to synchronize updates to this struct.
* @lock: used to synchronize access to @num_acks and @acks. These
* are updated whenever a client RPC completes and drained by
* outgoing packets, so they get their own cache line, away from the
* fields read on every route lookup.
*/
spinlock_t lock;
spinlock_t lock ____cacheline_aligned_in_smp;

/** @rcu_head: Holds state of a pending call_rcu invocation. */
struct rcu_head rcu_head;
/**
* @num_acks: the number of (initial) entries in @acks that
* currently hold valid information. homa_route_get_acks reads it
* without the lock as a fast-path check, which is racy, but the
* worst that can happen is a wasted lock or an ack that waits for
* the next packet.
*/
int num_acks;

/**
* @acks: info about client RPCs sent on this route whose results
* have been completely received. Kept per route, not per peer: a
* server matches an ack by the source address of the packet
* carrying it, so an ack must leave from the address its RPC used.
*/
struct homa_ack acks[HOMA_MAX_ACKS_PER_PKT];
};

void homa_dst_refresh(struct homa_peertab *peertab,
struct homa_peer *peer, struct homa_sock *hsk);
void homa_peer_add_ack(struct homa_rpc *rpc);
struct homa_peer
*homa_peer_alloc(struct homa_sock *hsk, const struct in6_addr *addr);
struct homa_peertab
Expand All @@ -351,13 +353,11 @@ void homa_peer_free_net(struct homa_net *hnet);
void homa_peer_free_peertab(struct homa_peertab *peertab);
struct homa_peer
*homa_peer_get(struct homa_sock *hsk, const struct in6_addr *addr);
int homa_peer_get_acks(struct homa_peer *peer, int count,
struct homa_ack *dst);
void homa_peer_lock_slow(struct homa_peer *peer);
void homa_peer_release_fn(void *object, void *dummy);
void homa_peer_update_sysctl_deps(struct homa_peertab *peertab);
void homa_peer_set_cutoffs(struct homa_peer *peer, int c0, int c1,
int c2, int c3, int c4, int c5, int c6, int c7);
void homa_route_add_ack(struct homa_rpc *rpc);
struct homa_route
*homa_route_alloc(struct homa_sock *hsk,
const struct homa_route_key *key);
Expand All @@ -367,7 +367,10 @@ void homa_route_gc(struct homa_peertab *peertab);
struct homa_route
*homa_route_get(struct homa_sock *hsk,
const struct in6_addr *addr);
int homa_route_get_acks(struct homa_route *route, int count,
struct homa_ack *dst);
u32 homa_route_hash(const void *data, u32 len, u32 seed);
void homa_route_lock_slow(struct homa_route *route);
int homa_route_pick_victims(struct homa_peertab *peertab,
struct homa_route *victims[], int max_victims);
int homa_route_prefer_evict(struct homa_peertab *peertab,
Expand All @@ -382,36 +385,36 @@ extern const struct rhashtable_params route_ht_params;

#ifndef __STRIP__ /* See strip.py */
/**
* homa_peer_lock() - Acquire the lock for a peer. If the lock isn't
* homa_route_lock() - Acquire the lock for a route. If the lock isn't
* immediately available, record stats on the waiting time.
* @peer: Peer to lock.
* @route: Route to lock.
*/
static inline void homa_peer_lock(struct homa_peer *peer)
__acquires(peer->lock)
static inline void homa_route_lock(struct homa_route *route)
__acquires(route->lock)
{
if (!spin_trylock_bh(&peer->lock))
homa_peer_lock_slow(peer);
if (!spin_trylock_bh(&route->lock))
homa_route_lock_slow(route);
}
#else /* See strip.py */
/**
* homa_peer_lock() - Acquire the lock for a peer.
* @peer: Peer to lock.
* homa_route_lock() - Acquire the lock for a route.
* @route: Route to lock.
*/
static inline void homa_peer_lock(struct homa_peer *peer)
__acquires(peer->lock)
static inline void homa_route_lock(struct homa_route *route)
__acquires(route->lock)
{
spin_lock_bh(&peer->lock);
spin_lock_bh(&route->lock);
}
#endif /* See strip.py */

/**
* homa_peer_unlock() - Release the lock for a peer.
* @peer: Peer to lock.
* homa_route_unlock() - Release the lock for a route.
* @route: Route to unlock.
*/
static inline void homa_peer_unlock(struct homa_peer *peer)
__releases(peer->lock)
static inline void homa_route_unlock(struct homa_route *route)
__releases(route->lock)
{
spin_unlock_bh(&peer->lock);
spin_unlock_bh(&route->lock);
}

/**
Expand Down
2 changes: 1 addition & 1 deletion homa_plumbing.c
Original file line number Diff line number Diff line change
Expand Up @@ -1523,7 +1523,7 @@ int homa_recvmsg(struct sock *sk, struct msghdr *msg, size_t len, int flags,
*/
rpc->msgin.num_bpages = 0;
if (homa_is_client(rpc->id)) {
homa_peer_add_ack(rpc);
homa_route_add_ack(rpc);
homa_rpc_end(rpc);
} else {
rpc->state = RPC_IN_SERVICE;
Expand Down
5 changes: 3 additions & 2 deletions protocol.md
Original file line number Diff line number Diff line change
Expand Up @@ -129,8 +129,9 @@ the RPC. Acks can get sent in two ways. First, each DATA packet
has room for one ack, so if a client is having an ongoing conversation
with a server, it can use future RPCs to ack older ones. Second, clients
can send explicit ACK packets, each of which can carry multiple acks.
A client has limited storage for acks for each peer, so it will send
an ACK packet if its storage for a peer overflows. In addition, the server
A client has limited storage for pending acks for each route (a
source and destination address pair), so it will send an ACK packet
if that storage overflows. In addition, the server
will use its timeout mechanism to request an explicit ack if all of the
data has been transmitted for a response but no ack has been received.
The server sends a NEED_ACK packet to request the ack, and the client
Expand Down
6 changes: 3 additions & 3 deletions test/unit_homa_incoming.c
Original file line number Diff line number Diff line change
Expand Up @@ -2388,9 +2388,9 @@ TEST_F(homa_incoming, homa_need_ack_pkt__rpc_doesnt_exist)
.sender_id = cpu_to_be64(self->server_id),
.type = NEED_ACK}};

route->peer->acks[0].server_port = htons(self->server_port);
route->peer->acks[0].client_id = cpu_to_be64(self->client_id+2);
route->peer->num_acks = 1;
route->acks[0].server_port = htons(self->server_port);
route->acks[0].client_id = cpu_to_be64(self->client_id+2);
route->num_acks = 1;
mock_xmit_log_verbose = 1;
homa_dispatch_pkts(mock_skb_alloc(self->server_ip, self->client_ip,
&h.common, 0, 0));
Expand Down
Loading