From c937aabbd7f46fa3b283744169a6b48dafbd6cc7 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:33 -0400 Subject: fs: dlm: always run complete for possible waiters This patch changes the ping_members() result that we always run complete() for possible waiters. We handle the -EINTR error code as successful. This error code is returned if the recovery is stopped which is likely that a new recovery is triggered with a new members configuration and ping_members() runs again. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/member.c | 18 ++++++++++++------ 1 file changed, 12 insertions(+), 6 deletions(-) diff --git a/fs/dlm/member.c b/fs/dlm/member.c index ceef3f2074ff..48245ba3c640 100644 --- a/fs/dlm/member.c +++ b/fs/dlm/member.c @@ -576,12 +576,18 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out) *neg_out = neg; error = ping_members(ls); - if (!error || error == -EPROTO) { - /* new_lockspace() may be waiting to know if the config - is good or bad */ - ls->ls_members_result = error; - complete(&ls->ls_members_done); - } + /* error -EINTR means that a new recovery action is triggered. + * We ignore this recovery action and let run the new one which might + * have new member configuration. + */ + if (error == -EINTR) + error = 0; + + /* new_lockspace() may be waiting to know if the config + * is good or bad + */ + ls->ls_members_result = error; + complete(&ls->ls_members_done); log_rinfo(ls, "dlm_recover_members %d nodes", ls->ls_num_nodes); return error; -- cgit From 2df6b7627a81b1407378125246eacdb1d8c90036 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:34 -0400 Subject: fs: dlm: add dlm macros for ratelimit log This patch add ratelimit macro to dlm subsystem and will set the connecting log message to ratelimit. In non blocking connecting cases it will print out this message a lot. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/dlm_internal.h | 2 ++ fs/dlm/lowcomms.c | 4 ++-- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h index 04fe9f525ac7..ae3fdf6d9cda 100644 --- a/fs/dlm/dlm_internal.h +++ b/fs/dlm/dlm_internal.h @@ -60,6 +60,8 @@ struct dlm_mhandle; #define log_print(fmt, args...) \ printk(KERN_ERR "dlm: "fmt"\n" , ##args) +#define log_print_ratelimited(fmt, args...) \ + printk_ratelimited(KERN_ERR "dlm: "fmt"\n", ##args) #define log_error(ls, fmt, args...) \ printk(KERN_ERR "dlm: %s: " fmt "\n", (ls)->ls_name , ##args) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 166e36fcf3e4..15810701b13e 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -1075,7 +1075,7 @@ static void sctp_connect_to_sock(struct connection *con) make_sockaddr(&daddr, dlm_config.ci_tcp_port, &addr_len); - log_print("connecting to %d", con->nodeid); + log_print_ratelimited("connecting to %d", con->nodeid); /* Turn off Nagle's algorithm */ sctp_sock_set_nodelay(sock->sk); @@ -1171,7 +1171,7 @@ static void tcp_connect_to_sock(struct connection *con) make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len); - log_print("connecting to %d", con->nodeid); + log_print_ratelimited("connecting to %d", con->nodeid); /* Turn off Nagle's algorithm */ tcp_sock_set_nodelay(sock->sk); -- cgit From b38bc9c2b3171f4411d80015ecb876bc6f9bcd26 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:35 -0400 Subject: fs: dlm: fix srcu read lock usage This patch holds the srcu connection read lock in cases where we lookup the connections and accessing it. We don't hold the srcu lock in workers function where the scheduled worker is part of the connection itself. The connection should not be freed if any worker is scheduled or pending. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 75 ++++++++++++++++++++++++++++++++++++++----------------- 1 file changed, 52 insertions(+), 23 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 15810701b13e..7c7a31377f27 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -113,6 +113,7 @@ struct writequeue_entry { int len; int end; int users; + int idx; /* get()/commit() idx exchange */ struct connection *con; }; @@ -163,21 +164,14 @@ static inline int nodeid_hash(int nodeid) return nodeid & (CONN_HASH_SIZE-1); } -static struct connection *__find_con(int nodeid) +static struct connection *__find_con(int nodeid, int r) { - int r, idx; struct connection *con; - r = nodeid_hash(nodeid); - - idx = srcu_read_lock(&connections_srcu); hlist_for_each_entry_rcu(con, &connection_hash[r], list) { - if (con->nodeid == nodeid) { - srcu_read_unlock(&connections_srcu, idx); + if (con->nodeid == nodeid) return con; - } } - srcu_read_unlock(&connections_srcu, idx); return NULL; } @@ -216,7 +210,8 @@ static struct connection *nodeid2con(int nodeid, gfp_t alloc) struct connection *con, *tmp; int r, ret; - con = __find_con(nodeid); + r = nodeid_hash(nodeid); + con = __find_con(nodeid, r); if (con || !alloc) return con; @@ -230,8 +225,6 @@ static struct connection *nodeid2con(int nodeid, gfp_t alloc) return NULL; } - r = nodeid_hash(nodeid); - spin_lock(&connections_lock); /* Because multiple workqueues/threads calls this function it can * race on multiple cpu's. Instead of locking hot path __find_con() @@ -239,7 +232,7 @@ static struct connection *nodeid2con(int nodeid, gfp_t alloc) * under protection of connections_lock. If this is the case we * abort our connection creation and return the existing connection. */ - tmp = __find_con(nodeid); + tmp = __find_con(nodeid, r); if (tmp) { spin_unlock(&connections_lock); kfree(con->rx_buf); @@ -256,15 +249,13 @@ static struct connection *nodeid2con(int nodeid, gfp_t alloc) /* Loop round all connections */ static void foreach_conn(void (*conn_func)(struct connection *c)) { - int i, idx; + int i; struct connection *con; - idx = srcu_read_lock(&connections_srcu); for (i = 0; i < CONN_HASH_SIZE; i++) { hlist_for_each_entry_rcu(con, &connection_hash[i], list) conn_func(con); } - srcu_read_unlock(&connections_srcu, idx); } static struct dlm_node_addr *find_node_addr(int nodeid) @@ -518,14 +509,21 @@ static void lowcomms_state_change(struct sock *sk) int dlm_lowcomms_connect_node(int nodeid) { struct connection *con; + int idx; if (nodeid == dlm_our_nodeid()) return 0; + idx = srcu_read_lock(&connections_srcu); con = nodeid2con(nodeid, GFP_NOFS); - if (!con) + if (!con) { + srcu_read_unlock(&connections_srcu, idx); return -ENOMEM; + } + lowcomms_connect_sock(con); + srcu_read_unlock(&connections_srcu, idx); + return 0; } @@ -864,7 +862,7 @@ static int accept_from_sock(struct listen_connection *con) int result; struct sockaddr_storage peeraddr; struct socket *newsock; - int len; + int len, idx; int nodeid; struct connection *newcon; struct connection *addcon; @@ -907,8 +905,10 @@ static int accept_from_sock(struct listen_connection *con) * the same time and the connections cross on the wire. * In this case we store the incoming one in "othercon" */ + idx = srcu_read_lock(&connections_srcu); newcon = nodeid2con(nodeid, GFP_NOFS); if (!newcon) { + srcu_read_unlock(&connections_srcu, idx); result = -ENOMEM; goto accept_err; } @@ -924,6 +924,7 @@ static int accept_from_sock(struct listen_connection *con) if (!othercon) { log_print("failed to allocate incoming socket"); mutex_unlock(&newcon->sock_mutex); + srcu_read_unlock(&connections_srcu, idx); result = -ENOMEM; goto accept_err; } @@ -932,6 +933,7 @@ static int accept_from_sock(struct listen_connection *con) if (result < 0) { kfree(othercon); mutex_unlock(&newcon->sock_mutex); + srcu_read_unlock(&connections_srcu, idx); goto accept_err; } @@ -966,6 +968,8 @@ static int accept_from_sock(struct listen_connection *con) if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags)) queue_work(recv_workqueue, &addcon->rwork); + srcu_read_unlock(&connections_srcu, idx); + return 0; accept_err: @@ -1403,7 +1407,9 @@ static struct writequeue_entry *new_wq_entry(struct connection *con, int len, void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc) { + struct writequeue_entry *e; struct connection *con; + int idx; if (len > DEFAULT_BUFFER_SIZE || len < sizeof(struct dlm_header)) { @@ -1413,11 +1419,23 @@ void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc) return NULL; } + idx = srcu_read_lock(&connections_srcu); con = nodeid2con(nodeid, allocation); - if (!con) + if (!con) { + srcu_read_unlock(&connections_srcu, idx); return NULL; + } + + e = new_wq_entry(con, len, allocation, ppc); + if (!e) { + srcu_read_unlock(&connections_srcu, idx); + return NULL; + } + + /* we assume if successful commit must called */ + e->idx = idx; - return new_wq_entry(con, len, allocation, ppc); + return e; } void dlm_lowcomms_commit_buffer(void *mh) @@ -1435,10 +1453,12 @@ void dlm_lowcomms_commit_buffer(void *mh) spin_unlock(&con->writequeue_lock); queue_work(send_workqueue, &con->swork); + srcu_read_unlock(&connections_srcu, e->idx); return; out: spin_unlock(&con->writequeue_lock); + srcu_read_unlock(&connections_srcu, e->idx); return; } @@ -1532,8 +1552,10 @@ int dlm_lowcomms_close(int nodeid) { struct connection *con; struct dlm_node_addr *na; + int idx; log_print("closing connection to node %d", nodeid); + idx = srcu_read_lock(&connections_srcu); con = nodeid2con(nodeid, 0); if (con) { set_bit(CF_CLOSE, &con->flags); @@ -1542,6 +1564,7 @@ int dlm_lowcomms_close(int nodeid) if (con->othercon) clean_one_writequeue(con->othercon); } + srcu_read_unlock(&connections_srcu, idx); spin_lock(&dlm_node_addrs_spin); na = find_node_addr(nodeid); @@ -1621,6 +1644,8 @@ static void shutdown_conn(struct connection *con) void dlm_lowcomms_shutdown(void) { + int idx; + /* Set all the flags to prevent any * socket activity. */ @@ -1633,7 +1658,9 @@ void dlm_lowcomms_shutdown(void) dlm_close_sock(&listen_con.sock); + idx = srcu_read_lock(&connections_srcu); foreach_conn(shutdown_conn); + srcu_read_unlock(&connections_srcu, idx); } static void _stop_conn(struct connection *con, bool and_other) @@ -1682,7 +1709,7 @@ static void free_conn(struct connection *con) static void work_flush(void) { - int ok, idx; + int ok; int i; struct connection *con; @@ -1693,7 +1720,6 @@ static void work_flush(void) flush_workqueue(recv_workqueue); if (send_workqueue) flush_workqueue(send_workqueue); - idx = srcu_read_lock(&connections_srcu); for (i = 0; i < CONN_HASH_SIZE && ok; i++) { hlist_for_each_entry_rcu(con, &connection_hash[i], list) { @@ -1707,14 +1733,17 @@ static void work_flush(void) } } } - srcu_read_unlock(&connections_srcu, idx); } while (!ok); } void dlm_lowcomms_stop(void) { + int idx; + + idx = srcu_read_lock(&connections_srcu); work_flush(); foreach_conn(free_conn); + srcu_read_unlock(&connections_srcu, idx); work_stop(); deinit_local(); } -- cgit From 7443bc962509912c70c587db71449daff26b9678 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:36 -0400 Subject: fs: dlm: set is othercon flag There is a is othercon flag which is never used, this patch will set it and printout a warning if the othercon ever sends a dlm message which should never be the case. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 3 +++ 1 file changed, 3 insertions(+) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 7c7a31377f27..4944aef24aa5 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -938,6 +938,7 @@ static int accept_from_sock(struct listen_connection *con) } lockdep_set_subclass(&othercon->sock_mutex, 1); + set_bit(CF_IS_OTHERCON, &othercon->flags); newcon->othercon = othercon; } else { /* close other sock con if we have something new */ @@ -1601,6 +1602,8 @@ static void process_send_sockets(struct work_struct *work) { struct connection *con = container_of(work, struct connection, swork); + WARN_ON(test_bit(CF_IS_OTHERCON, &con->flags)); + clear_bit(CF_WRITE_PENDING, &con->flags); if (con->sock == NULL) /* not mutex protected so check it inside too */ con->connect_action(con); -- cgit From ba868d9deaab2bb1c09e50650127823925154802 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:37 -0400 Subject: fs: dlm: reconnect if socket error report occurs This patch will change the reconnect handling that if an error occurs if a socket error callback is occurred. This will also handle reconnects in a non blocking connecting case which is currently missing. If error ECONNREFUSED is reported we delay the reconnect by one second. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 60 ++++++++++++++++++++++++++++++++++++------------------- 1 file changed, 39 insertions(+), 21 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 4944aef24aa5..051f22dbb83a 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -79,6 +79,8 @@ struct connection { #define CF_CLOSING 8 #define CF_SHUTDOWN 9 #define CF_CONNECTED 10 +#define CF_RECONNECT 11 +#define CF_DELAY_CONNECT 12 struct list_head writequeue; /* List of outgoing writequeue_entries */ spinlock_t writequeue_lock; void (*connect_action) (struct connection *); /* What to do to connect */ @@ -87,6 +89,7 @@ struct connection { #define MAX_CONNECT_RETRIES 3 struct hlist_node list; struct connection *othercon; + struct connection *sendcon; struct work_struct rwork; /* Receive workqueue */ struct work_struct swork; /* Send workqueue */ wait_queue_head_t shutdown_wait; /* wait for graceful shutdown */ @@ -585,6 +588,22 @@ static void lowcomms_error_report(struct sock *sk) dlm_config.ci_tcp_port, sk->sk_err, sk->sk_err_soft); } + + /* below sendcon only handling */ + if (test_bit(CF_IS_OTHERCON, &con->flags)) + con = con->sendcon; + + switch (sk->sk_err) { + case ECONNREFUSED: + set_bit(CF_DELAY_CONNECT, &con->flags); + break; + default: + break; + } + + if (!test_and_set_bit(CF_RECONNECT, &con->flags)) + queue_work(send_workqueue, &con->swork); + out: read_unlock_bh(&sk->sk_callback_lock); if (orig_report) @@ -702,6 +721,8 @@ static void close_connection(struct connection *con, bool and_other, con->rx_leftover = 0; con->retries = 0; clear_bit(CF_CONNECTED, &con->flags); + clear_bit(CF_DELAY_CONNECT, &con->flags); + clear_bit(CF_RECONNECT, &con->flags); mutex_unlock(&con->sock_mutex); clear_bit(CF_CLOSING, &con->flags); } @@ -840,18 +861,15 @@ out_resched: out_close: mutex_unlock(&con->sock_mutex); - if (ret != -EAGAIN) { - /* Reconnect when there is something to send */ + if (ret == 0) { close_connection(con, false, true, false); - if (ret == 0) { - log_print("connection %p got EOF from %d", - con, con->nodeid); - /* handling for tcp shutdown */ - clear_bit(CF_SHUTDOWN, &con->flags); - wake_up(&con->shutdown_wait); - /* signal to breaking receive worker */ - ret = -1; - } + log_print("connection %p got EOF from %d", + con, con->nodeid); + /* handling for tcp shutdown */ + clear_bit(CF_SHUTDOWN, &con->flags); + wake_up(&con->shutdown_wait); + /* signal to breaking receive worker */ + ret = -1; } return ret; } @@ -940,6 +958,7 @@ static int accept_from_sock(struct listen_connection *con) lockdep_set_subclass(&othercon->sock_mutex, 1); set_bit(CF_IS_OTHERCON, &othercon->flags); newcon->othercon = othercon; + othercon->sendcon = newcon; } else { /* close other sock con if we have something new */ close_connection(othercon, false, true, false); @@ -1504,7 +1523,7 @@ static void send_to_sock(struct connection *con) cond_resched(); goto out; } else if (ret < 0) - goto send_error; + goto out; } /* Don't starve people filling buffers */ @@ -1521,14 +1540,6 @@ out: mutex_unlock(&con->sock_mutex); return; -send_error: - mutex_unlock(&con->sock_mutex); - close_connection(con, false, false, true); - /* Requeue the send work. When the work daemon runs again, it will try - a new connection, then call this function again. */ - queue_work(send_workqueue, &con->swork); - return; - out_connect: mutex_unlock(&con->sock_mutex); queue_work(send_workqueue, &con->swork); @@ -1605,8 +1616,15 @@ static void process_send_sockets(struct work_struct *work) WARN_ON(test_bit(CF_IS_OTHERCON, &con->flags)); clear_bit(CF_WRITE_PENDING, &con->flags); - if (con->sock == NULL) /* not mutex protected so check it inside too */ + + if (test_and_clear_bit(CF_RECONNECT, &con->flags)) + close_connection(con, false, false, true); + + if (con->sock == NULL) { /* not mutex protected so check it inside too */ + if (test_and_clear_bit(CF_DELAY_CONNECT, &con->flags)) + msleep(1000); con->connect_action(con); + } if (!list_empty(&con->writequeue)) send_to_sock(con); } -- cgit From c6aa00e3d20c2767ba3f57b64eb862572b9744b3 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:38 -0400 Subject: fs: dlm: cancel work sync othercon These rx tx flags arguments are for signaling close_connection() from which worker they are called. Obviously the receive worker cannot cancel itself and vice versa for swork. For the othercon the receive worker should only be used, however to avoid deadlocks we should pass the same flags as the original close_connection() was called. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 051f22dbb83a..0a4851b3cd4b 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -715,7 +715,7 @@ static void close_connection(struct connection *con, bool and_other, if (con->othercon && and_other) { /* Will only re-enter once. */ - close_connection(con->othercon, false, true, true); + close_connection(con->othercon, false, tx, rx); } con->rx_leftover = 0; -- cgit From 8aa31cbf20ad168c35dd83476629402aacbf5a44 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:39 -0400 Subject: fs: dlm: fix connection tcp EOF handling This patch fixes the EOF handling for TCP that if and EOF is received we will close the socket next time the writequeue runs empty. This is a half-closed socket functionality which doesn't exists in SCTP. The midcomms layer will do a half closed socket functionality on DLM side to solve this problem for the SCTP case. However there is still the last ack flying around but other reset functionality will take care of it if it got lost. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 48 +++++++++++++++++++++++++++++++++++++++++++----- 1 file changed, 43 insertions(+), 5 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 0a4851b3cd4b..14ca3eda6a83 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -81,10 +81,13 @@ struct connection { #define CF_CONNECTED 10 #define CF_RECONNECT 11 #define CF_DELAY_CONNECT 12 +#define CF_EOF 13 struct list_head writequeue; /* List of outgoing writequeue_entries */ spinlock_t writequeue_lock; + atomic_t writequeue_cnt; void (*connect_action) (struct connection *); /* What to do to connect */ void (*shutdown_action)(struct connection *con); /* What to do to shutdown */ + bool (*eof_condition)(struct connection *con); /* What to do to eof check */ int retries; #define MAX_CONNECT_RETRIES 3 struct hlist_node list; @@ -179,6 +182,11 @@ static struct connection *__find_con(int nodeid, int r) return NULL; } +static bool tcp_eof_condition(struct connection *con) +{ + return atomic_read(&con->writequeue_cnt); +} + static int dlm_con_init(struct connection *con, int nodeid) { con->rx_buflen = dlm_config.ci_buffer_size; @@ -190,6 +198,7 @@ static int dlm_con_init(struct connection *con, int nodeid) mutex_init(&con->sock_mutex); INIT_LIST_HEAD(&con->writequeue); spin_lock_init(&con->writequeue_lock); + atomic_set(&con->writequeue_cnt, 0); INIT_WORK(&con->swork, process_send_sockets); INIT_WORK(&con->rwork, process_recv_sockets); init_waitqueue_head(&con->shutdown_wait); @@ -197,6 +206,7 @@ static int dlm_con_init(struct connection *con, int nodeid) if (dlm_config.ci_protocol == 0) { con->connect_action = tcp_connect_to_sock; con->shutdown_action = dlm_tcp_shutdown; + con->eof_condition = tcp_eof_condition; } else { con->connect_action = sctp_connect_to_sock; } @@ -723,6 +733,7 @@ static void close_connection(struct connection *con, bool and_other, clear_bit(CF_CONNECTED, &con->flags); clear_bit(CF_DELAY_CONNECT, &con->flags); clear_bit(CF_RECONNECT, &con->flags); + clear_bit(CF_EOF, &con->flags); mutex_unlock(&con->sock_mutex); clear_bit(CF_CLOSING, &con->flags); } @@ -860,16 +871,26 @@ out_resched: return -EAGAIN; out_close: - mutex_unlock(&con->sock_mutex); if (ret == 0) { - close_connection(con, false, true, false); log_print("connection %p got EOF from %d", con, con->nodeid); - /* handling for tcp shutdown */ - clear_bit(CF_SHUTDOWN, &con->flags); - wake_up(&con->shutdown_wait); + + if (con->eof_condition && con->eof_condition(con)) { + set_bit(CF_EOF, &con->flags); + mutex_unlock(&con->sock_mutex); + } else { + mutex_unlock(&con->sock_mutex); + close_connection(con, false, true, false); + + /* handling for tcp shutdown */ + clear_bit(CF_SHUTDOWN, &con->flags); + wake_up(&con->shutdown_wait); + } + /* signal to breaking receive worker */ ret = -1; + } else { + mutex_unlock(&con->sock_mutex); } return ret; } @@ -1021,6 +1042,7 @@ static void writequeue_entry_complete(struct writequeue_entry *e, int completed) if (e->len == 0 && e->users == 0) { list_del(&e->list); + atomic_dec(&e->con->writequeue_cnt); free_entry(e); } } @@ -1417,6 +1439,7 @@ static struct writequeue_entry *new_wq_entry(struct connection *con, int len, *ppc = page_address(e->page); e->end += len; + atomic_inc(&con->writequeue_cnt); spin_lock(&con->writequeue_lock); list_add_tail(&e->list, &con->writequeue); @@ -1536,6 +1559,21 @@ static void send_to_sock(struct connection *con) writequeue_entry_complete(e, ret); } spin_unlock(&con->writequeue_lock); + + /* close if we got EOF */ + if (test_and_clear_bit(CF_EOF, &con->flags)) { + mutex_unlock(&con->sock_mutex); + close_connection(con, false, false, true); + + /* handling for tcp shutdown */ + clear_bit(CF_SHUTDOWN, &con->flags); + wake_up(&con->shutdown_wait); + } else { + mutex_unlock(&con->sock_mutex); + } + + return; + out: mutex_unlock(&con->sock_mutex); return; -- cgit From 6fb5cf9d4206f2cdccb05be1bf2307dab4e5babe Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:40 -0400 Subject: fs: dlm: public header in out utility This patch allows to use header_out() and header_in() outside of dlm util functionality. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/util.c | 4 ++-- fs/dlm/util.h | 2 ++ 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/fs/dlm/util.c b/fs/dlm/util.c index cfd0d00b19ae..74a8c5bfe9b5 100644 --- a/fs/dlm/util.c +++ b/fs/dlm/util.c @@ -20,7 +20,7 @@ #define DLM_ERRNO_ETIMEDOUT 110 #define DLM_ERRNO_EINPROGRESS 115 -static void header_out(struct dlm_header *hd) +void header_out(struct dlm_header *hd) { hd->h_version = cpu_to_le32(hd->h_version); hd->h_lockspace = cpu_to_le32(hd->h_lockspace); @@ -28,7 +28,7 @@ static void header_out(struct dlm_header *hd) hd->h_length = cpu_to_le16(hd->h_length); } -static void header_in(struct dlm_header *hd) +void header_in(struct dlm_header *hd) { hd->h_version = le32_to_cpu(hd->h_version); hd->h_lockspace = le32_to_cpu(hd->h_lockspace); diff --git a/fs/dlm/util.h b/fs/dlm/util.h index cc719ca9397e..d46f23c7a6a0 100644 --- a/fs/dlm/util.h +++ b/fs/dlm/util.h @@ -15,6 +15,8 @@ void dlm_message_out(struct dlm_message *ms); void dlm_message_in(struct dlm_message *ms); void dlm_rcom_out(struct dlm_rcom *rc); void dlm_rcom_in(struct dlm_rcom *rc); +void header_out(struct dlm_header *hd); +void header_in(struct dlm_header *hd); #endif -- cgit From a070a91cf1402b5328d3517d1fccbdeec58d3f2d Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:41 -0400 Subject: fs: dlm: add more midcomms hooks This patch prepares hooks to redirect to the midcomms layer which will be used by the midcomms re-transmit handling. There exists the new concept of stateless buffers allocation and commits. This can be used to bypass the midcomms re-transmit handling. It is used by RCOM_STATUS and RCOM_NAMES messages, because they have their own ping-like re-transmit handling. As well these two messages will be used to determine the DLM version per node, because these two messages are per observation the first messages which are exchanged. Cluster manager events for node membership are added to add support for half-closed connections in cases that the peer connection get to an end of file but DLM still holds membership of the node. In this time DLM can still trigger new message which we should allow. After the cluster manager node removal event occurs it safe to close the connection. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/config.c | 3 +- fs/dlm/lock.c | 8 ++--- fs/dlm/lockspace.c | 7 ++-- fs/dlm/member.c | 17 +++++++-- fs/dlm/midcomms.c | 31 +++++++++++++++- fs/dlm/midcomms.h | 8 +++++ fs/dlm/rcom.c | 101 +++++++++++++++++++++++++++++++++++++---------------- 7 files changed, 133 insertions(+), 42 deletions(-) diff --git a/fs/dlm/config.c b/fs/dlm/config.c index 88d95d96e36c..01ae294743e9 100644 --- a/fs/dlm/config.c +++ b/fs/dlm/config.c @@ -20,6 +20,7 @@ #include #include "config.h" +#include "midcomms.h" #include "lowcomms.h" /* @@ -532,7 +533,7 @@ static void drop_comm(struct config_group *g, struct config_item *i) struct dlm_comm *cm = config_item_to_comm(i); if (local_comm == cm) local_comm = NULL; - dlm_lowcomms_close(cm->nodeid); + dlm_midcomms_close(cm->nodeid); while (cm->addr_count--) kfree(cm->addr[cm->addr_count]); config_item_put(i); diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c index b93df39d0915..b625ce92464a 100644 --- a/fs/dlm/lock.c +++ b/fs/dlm/lock.c @@ -59,7 +59,7 @@ #include "dlm_internal.h" #include #include "memory.h" -#include "lowcomms.h" +#include "midcomms.h" #include "requestqueue.h" #include "util.h" #include "dir.h" @@ -3534,10 +3534,10 @@ static int _create_message(struct dlm_ls *ls, int mb_len, char *mb; /* get_buffer gives us a message handle (mh) that we need to - pass into lowcomms_commit and a message buffer (mb) that we + pass into midcomms_commit and a message buffer (mb) that we write our data into */ - mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb); + mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb); if (!mh) return -ENOBUFS; @@ -3589,7 +3589,7 @@ static int create_message(struct dlm_rsb *r, struct dlm_lkb *lkb, static int send_message(struct dlm_mhandle *mh, struct dlm_message *ms) { dlm_message_out(ms); - dlm_lowcomms_commit_buffer(mh); + dlm_midcomms_commit_mhandle(mh); return 0; } diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c index c14cf2b7faab..bf5c55ef9d0d 100644 --- a/fs/dlm/lockspace.c +++ b/fs/dlm/lockspace.c @@ -16,6 +16,7 @@ #include "member.h" #include "recoverd.h" #include "dir.h" +#include "midcomms.h" #include "lowcomms.h" #include "config.h" #include "memory.h" @@ -390,7 +391,7 @@ static int threads_start(void) } /* Thread for sending/receiving messages for all lockspace's */ - error = dlm_lowcomms_start(); + error = dlm_midcomms_start(); if (error) { log_print("cannot start dlm lowcomms %d", error); goto scand_fail; @@ -698,7 +699,7 @@ int dlm_new_lockspace(const char *name, const char *cluster, error = 0; if (!ls_count) { dlm_scand_stop(); - dlm_lowcomms_shutdown(); + dlm_midcomms_shutdown(); dlm_lowcomms_stop(); } out: @@ -787,7 +788,7 @@ static int release_lockspace(struct dlm_ls *ls, int force) if (ls_count == 1) { dlm_scand_stop(); - dlm_lowcomms_shutdown(); + dlm_midcomms_shutdown(); } dlm_callback_stop(ls); diff --git a/fs/dlm/member.c b/fs/dlm/member.c index 48245ba3c640..63971c594bdc 100644 --- a/fs/dlm/member.c +++ b/fs/dlm/member.c @@ -15,6 +15,7 @@ #include "recover.h" #include "rcom.h" #include "config.h" +#include "midcomms.h" #include "lowcomms.h" int dlm_slots_version(struct dlm_header *h) @@ -329,6 +330,7 @@ static int dlm_add_member(struct dlm_ls *ls, struct dlm_config_node *node) memb->nodeid = node->nodeid; memb->weight = node->weight; memb->comm_seq = node->comm_seq; + dlm_midcomms_add_member(node->nodeid); add_ordered_member(ls, memb); ls->ls_num_nodes++; return 0; @@ -359,26 +361,34 @@ int dlm_is_removed(struct dlm_ls *ls, int nodeid) return 0; } -static void clear_memb_list(struct list_head *head) +static void clear_memb_list(struct list_head *head, + void (*after_del)(int nodeid)) { struct dlm_member *memb; while (!list_empty(head)) { memb = list_entry(head->next, struct dlm_member, list); list_del(&memb->list); + if (after_del) + after_del(memb->nodeid); kfree(memb); } } +static void clear_members_cb(int nodeid) +{ + dlm_midcomms_remove_member(nodeid); +} + void dlm_clear_members(struct dlm_ls *ls) { - clear_memb_list(&ls->ls_nodes); + clear_memb_list(&ls->ls_nodes, clear_members_cb); ls->ls_num_nodes = 0; } void dlm_clear_members_gone(struct dlm_ls *ls) { - clear_memb_list(&ls->ls_nodes_gone); + clear_memb_list(&ls->ls_nodes_gone, NULL); } static void make_member_array(struct dlm_ls *ls) @@ -552,6 +562,7 @@ int dlm_recover_members(struct dlm_ls *ls, struct dlm_recover *rv, int *neg_out) neg++; list_move(&memb->list, &ls->ls_nodes_gone); + dlm_midcomms_remove_member(memb->nodeid); ls->ls_num_nodes--; dlm_lsop_recover_slot(ls, memb); } diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index 1c6654a21ec4..1a280dda99d3 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -28,6 +28,36 @@ #include "lock.h" #include "midcomms.h" +struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, + gfp_t allocation, char **ppc) +{ + return dlm_lowcomms_get_buffer(nodeid, len, allocation, ppc); +} + +void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh) +{ + dlm_lowcomms_commit_buffer(mh); +} + +void dlm_midcomms_add_member(int nodeid) { } + +void dlm_midcomms_remove_member(int nodeid) { } + +int dlm_midcomms_start(void) +{ + return dlm_lowcomms_start(); +} + +void dlm_midcomms_shutdown(void) +{ + dlm_lowcomms_shutdown(); +} + +int dlm_midcomms_close(int nodeid) +{ + return dlm_lowcomms_close(nodeid); +} + /* * Called from the low-level comms layer to process a buffer of * commands. @@ -101,4 +131,3 @@ skip: return ret; } - diff --git a/fs/dlm/midcomms.h b/fs/dlm/midcomms.h index 61e90a921849..9ac1190ce277 100644 --- a/fs/dlm/midcomms.h +++ b/fs/dlm/midcomms.h @@ -13,6 +13,14 @@ #define __MIDCOMMS_DOT_H__ int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int buflen); +struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, + gfp_t allocation, char **ppc); +void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh); +int dlm_midcomms_close(int nodeid); +int dlm_midcomms_start(void); +void dlm_midcomms_shutdown(void); +void dlm_midcomms_add_member(int nodeid); +void dlm_midcomms_remove_member(int nodeid); #endif /* __MIDCOMMS_DOT_H__ */ diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c index f5b1bd65728d..2661674364af 100644 --- a/fs/dlm/rcom.c +++ b/fs/dlm/rcom.c @@ -27,20 +27,10 @@ static int rcom_response(struct dlm_ls *ls) return test_bit(LSFL_RCOM_READY, &ls->ls_flags); } -static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len, - struct dlm_rcom **rc_ret, struct dlm_mhandle **mh_ret) +static void _create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len, + struct dlm_rcom **rc_ret, char *mb, int mb_len) { struct dlm_rcom *rc; - struct dlm_mhandle *mh; - char *mb; - int mb_len = sizeof(struct dlm_rcom) + len; - - mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb); - if (!mh) { - log_print("create_rcom to %d type %d len %d ENOBUFS", - to_nodeid, type, len); - return -ENOBUFS; - } rc = (struct dlm_rcom *) mb; @@ -56,15 +46,64 @@ static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len, rc->rc_seq = ls->ls_recover_seq; spin_unlock(&ls->ls_recover_lock); - *mh_ret = mh; *rc_ret = rc; +} + +static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len, + struct dlm_rcom **rc_ret, struct dlm_mhandle **mh_ret) +{ + int mb_len = sizeof(struct dlm_rcom) + len; + struct dlm_mhandle *mh; + char *mb; + + mh = dlm_midcomms_get_mhandle(to_nodeid, mb_len, GFP_NOFS, &mb); + if (!mh) { + log_print("%s to %d type %d len %d ENOBUFS", + __func__, to_nodeid, type, len); + return -ENOBUFS; + } + + _create_rcom(ls, to_nodeid, type, len, rc_ret, mb, mb_len); + *mh_ret = mh; return 0; } +static int create_rcom_stateless(struct dlm_ls *ls, int to_nodeid, int type, + int len, struct dlm_rcom **rc_ret, + void **mh_ret) +{ + int mb_len = sizeof(struct dlm_rcom) + len; + void *mh; + char *mb; + + mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb); + if (!mh) { + log_print("create_rcom to %d type %d len %d ENOBUFS", + to_nodeid, type, len); + return -ENOBUFS; + } + + _create_rcom(ls, to_nodeid, type, len, rc_ret, mb, mb_len); + *mh_ret = mh; + return 0; +} + +static void _send_rcom(struct dlm_ls *ls, struct dlm_rcom *rc) +{ + dlm_rcom_out(rc); +} + static void send_rcom(struct dlm_ls *ls, struct dlm_mhandle *mh, struct dlm_rcom *rc) { - dlm_rcom_out(rc); + _send_rcom(ls, rc); + dlm_midcomms_commit_mhandle(mh); +} + +static void send_rcom_stateless(struct dlm_ls *ls, void *mh, + struct dlm_rcom *rc) +{ + _send_rcom(ls, rc); dlm_lowcomms_commit_buffer(mh); } @@ -141,8 +180,8 @@ static void disallow_sync_reply(struct dlm_ls *ls) int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags) { struct dlm_rcom *rc; - struct dlm_mhandle *mh; int error = 0; + void *mh; ls->ls_recover_nodeid = nodeid; @@ -153,8 +192,8 @@ int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags) } retry: - error = create_rcom(ls, nodeid, DLM_RCOM_STATUS, - sizeof(struct rcom_status), &rc, &mh); + error = create_rcom_stateless(ls, nodeid, DLM_RCOM_STATUS, + sizeof(struct rcom_status), &rc, &mh); if (error) goto out; @@ -163,7 +202,7 @@ retry: allow_sync_reply(ls, &rc->rc_id); memset(ls->ls_recover_buf, 0, LOWCOMMS_MAX_TX_BUFFER_LEN); - send_rcom(ls, mh, rc); + send_rcom_stateless(ls, mh, rc); error = dlm_wait_function(ls, &rcom_response); disallow_sync_reply(ls); @@ -191,13 +230,13 @@ retry: static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) { struct dlm_rcom *rc; - struct dlm_mhandle *mh; struct rcom_status *rs; uint32_t status; int nodeid = rc_in->rc_header.h_nodeid; int len = sizeof(struct rcom_config); int num_slots = 0; int error; + void *mh; if (!dlm_slots_version(&rc_in->rc_header)) { status = dlm_recover_status(ls); @@ -218,8 +257,8 @@ static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) len += num_slots * sizeof(struct rcom_slot); do_create: - error = create_rcom(ls, nodeid, DLM_RCOM_STATUS_REPLY, - len, &rc, &mh); + error = create_rcom_stateless(ls, nodeid, DLM_RCOM_STATUS_REPLY, + len, &rc, &mh); if (error) return; @@ -246,7 +285,7 @@ static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) spin_unlock(&ls->ls_recover_lock); do_send: - send_rcom(ls, mh, rc); + send_rcom_stateless(ls, mh, rc); } static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in) @@ -271,13 +310,14 @@ static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in) int dlm_rcom_names(struct dlm_ls *ls, int nodeid, char *last_name, int last_len) { struct dlm_rcom *rc; - struct dlm_mhandle *mh; int error = 0; + void *mh; ls->ls_recover_nodeid = nodeid; retry: - error = create_rcom(ls, nodeid, DLM_RCOM_NAMES, last_len, &rc, &mh); + error = create_rcom_stateless(ls, nodeid, DLM_RCOM_NAMES, last_len, + &rc, &mh); if (error) goto out; memcpy(rc->rc_buf, last_name, last_len); @@ -285,7 +325,7 @@ retry: allow_sync_reply(ls, &rc->rc_id); memset(ls->ls_recover_buf, 0, LOWCOMMS_MAX_TX_BUFFER_LEN); - send_rcom(ls, mh, rc); + send_rcom_stateless(ls, mh, rc); error = dlm_wait_function(ls, &rcom_response); disallow_sync_reply(ls); @@ -298,14 +338,15 @@ retry: static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in) { struct dlm_rcom *rc; - struct dlm_mhandle *mh; int error, inlen, outlen, nodeid; + void *mh; nodeid = rc_in->rc_header.h_nodeid; inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom); outlen = LOWCOMMS_MAX_TX_BUFFER_LEN - sizeof(struct dlm_rcom); - error = create_rcom(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &mh); + error = create_rcom_stateless(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, + &rc, &mh); if (error) return; rc->rc_id = rc_in->rc_id; @@ -313,7 +354,7 @@ static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in) dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen, nodeid); - send_rcom(ls, mh, rc); + send_rcom_stateless(ls, mh, rc); } int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid) @@ -458,7 +499,7 @@ int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in) char *mb; int mb_len = sizeof(struct dlm_rcom) + sizeof(struct rcom_config); - mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_NOFS, &mb); + mh = dlm_midcomms_get_mhandle(nodeid, mb_len, GFP_NOFS, &mb); if (!mh) return -ENOBUFS; @@ -479,7 +520,7 @@ int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in) rf->rf_lvblen = cpu_to_le32(~0U); dlm_rcom_out(rc); - dlm_lowcomms_commit_buffer(mh); + dlm_midcomms_commit_mhandle(mh); return 0; } -- cgit From 8f2dc78dbc2010b497bb58e0460cb44c678a3c5b Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:42 -0400 Subject: fs: dlm: make buffer handling per msg This patch makes the void pointer handle for lowcomms functionality per message and not per page allocation entry. A refcount handling for the handle was added to keep the message alive until the user doesn't need it anymore. There exists now a per message callback which will be called when allocating a new buffer. This callback will be guaranteed to be called according the order of the sending buffer, which can be used that the caller increments a sequence number for the dlm message handle. For transition process we cast the dlm_mhandle to dlm_msg and vice versa until the midcomms layer will implement a specific dlm_mhandle structure. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/dlm_internal.h | 1 + fs/dlm/lowcomms.c | 101 ++++++++++++++++++++++++++++++++++++++++++-------- fs/dlm/lowcomms.h | 7 +++- fs/dlm/midcomms.c | 7 +++- fs/dlm/rcom.c | 40 ++++++++++---------- 5 files changed, 117 insertions(+), 39 deletions(-) diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h index ae3fdf6d9cda..e8dc5f4f1f9e 100644 --- a/fs/dlm/dlm_internal.h +++ b/fs/dlm/dlm_internal.h @@ -57,6 +57,7 @@ struct dlm_header; struct dlm_message; struct dlm_rcom; struct dlm_mhandle; +struct dlm_msg; #define log_print(fmt, args...) \ printk(KERN_ERR "dlm: "fmt"\n" , ##args) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 14ca3eda6a83..d222e6088ab2 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -119,8 +119,19 @@ struct writequeue_entry { int len; int end; int users; - int idx; /* get()/commit() idx exchange */ struct connection *con; + struct list_head msgs; + struct kref ref; +}; + +struct dlm_msg { + struct writequeue_entry *entry; + void *ppc; + int len; + int idx; /* new()/commit() idx exchange */ + + struct list_head list; + struct kref ref; }; struct dlm_node_addr { @@ -1022,12 +1033,37 @@ accept_err: return result; } -static void free_entry(struct writequeue_entry *e) +static void dlm_page_release(struct kref *kref) { + struct writequeue_entry *e = container_of(kref, struct writequeue_entry, + ref); + __free_page(e->page); kfree(e); } +static void dlm_msg_release(struct kref *kref) +{ + struct dlm_msg *msg = container_of(kref, struct dlm_msg, ref); + + kref_put(&msg->entry->ref, dlm_page_release); + kfree(msg); +} + +static void free_entry(struct writequeue_entry *e) +{ + struct dlm_msg *msg, *tmp; + + list_for_each_entry_safe(msg, tmp, &e->msgs, list) { + list_del(&msg->list); + kref_put(&msg->ref, dlm_msg_release); + } + + list_del(&e->list); + atomic_dec(&e->con->writequeue_cnt); + kref_put(&e->ref, dlm_page_release); +} + /* * writequeue_entry_complete - try to delete and free write queue entry * @e: write queue entry to try to delete @@ -1040,11 +1076,8 @@ static void writequeue_entry_complete(struct writequeue_entry *e, int completed) e->offset += completed; e->len -= completed; - if (e->len == 0 && e->users == 0) { - list_del(&e->list); - atomic_dec(&e->con->writequeue_cnt); + if (e->len == 0 && e->users == 0) free_entry(e); - } } /* @@ -1410,12 +1443,16 @@ static struct writequeue_entry *new_writequeue_entry(struct connection *con, entry->con = con; entry->users = 1; + kref_init(&entry->ref); + INIT_LIST_HEAD(&entry->msgs); return entry; } static struct writequeue_entry *new_wq_entry(struct connection *con, int len, - gfp_t allocation, char **ppc) + gfp_t allocation, char **ppc, + void (*cb)(struct dlm_mhandle *mh), + struct dlm_mhandle *mh) { struct writequeue_entry *e; @@ -1423,7 +1460,12 @@ static struct writequeue_entry *new_wq_entry(struct connection *con, int len, if (!list_empty(&con->writequeue)) { e = list_last_entry(&con->writequeue, struct writequeue_entry, list); if (DLM_WQ_REMAIN_BYTES(e) >= len) { + kref_get(&e->ref); + *ppc = page_address(e->page) + e->end; + if (cb) + cb(mh); + e->end += len; e->users++; spin_unlock(&con->writequeue_lock); @@ -1437,21 +1479,28 @@ static struct writequeue_entry *new_wq_entry(struct connection *con, int len, if (!e) return NULL; + kref_get(&e->ref); *ppc = page_address(e->page); e->end += len; atomic_inc(&con->writequeue_cnt); spin_lock(&con->writequeue_lock); + if (cb) + cb(mh); + list_add_tail(&e->list, &con->writequeue); spin_unlock(&con->writequeue_lock); return e; }; -void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc) +struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, + char **ppc, void (*cb)(struct dlm_mhandle *mh), + struct dlm_mhandle *mh) { struct writequeue_entry *e; struct connection *con; + struct dlm_msg *msg; int idx; if (len > DEFAULT_BUFFER_SIZE || @@ -1469,25 +1518,41 @@ void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc) return NULL; } - e = new_wq_entry(con, len, allocation, ppc); + msg = kzalloc(sizeof(*msg), allocation); + if (!msg) { + srcu_read_unlock(&connections_srcu, idx); + return NULL; + } + + kref_init(&msg->ref); + + e = new_wq_entry(con, len, allocation, ppc, cb, mh); if (!e) { srcu_read_unlock(&connections_srcu, idx); + kfree(msg); return NULL; } + msg->ppc = *ppc; + msg->len = len; + msg->entry = e; + /* we assume if successful commit must called */ - e->idx = idx; + msg->idx = idx; - return e; + return msg; } -void dlm_lowcomms_commit_buffer(void *mh) +void dlm_lowcomms_commit_msg(struct dlm_msg *msg) { - struct writequeue_entry *e = (struct writequeue_entry *)mh; + struct writequeue_entry *e = msg->entry; struct connection *con = e->con; int users; spin_lock(&con->writequeue_lock); + kref_get(&msg->ref); + list_add(&msg->list, &e->msgs); + users = --e->users; if (users) goto out; @@ -1496,15 +1561,20 @@ void dlm_lowcomms_commit_buffer(void *mh) spin_unlock(&con->writequeue_lock); queue_work(send_workqueue, &con->swork); - srcu_read_unlock(&connections_srcu, e->idx); + srcu_read_unlock(&connections_srcu, msg->idx); return; out: spin_unlock(&con->writequeue_lock); - srcu_read_unlock(&connections_srcu, e->idx); + srcu_read_unlock(&connections_srcu, msg->idx); return; } +void dlm_lowcomms_put_msg(struct dlm_msg *msg) +{ + kref_put(&msg->ref, dlm_msg_release); +} + /* Send a message */ static void send_to_sock(struct connection *con) { @@ -1590,7 +1660,6 @@ static void clean_one_writequeue(struct connection *con) spin_lock(&con->writequeue_lock); list_for_each_entry_safe(e, safe, &con->writequeue, list) { - list_del(&e->list); free_entry(e); } spin_unlock(&con->writequeue_lock); diff --git a/fs/dlm/lowcomms.h b/fs/dlm/lowcomms.h index 48bbc4e18761..cdb8f066f0d8 100644 --- a/fs/dlm/lowcomms.h +++ b/fs/dlm/lowcomms.h @@ -22,8 +22,11 @@ void dlm_lowcomms_shutdown(void); void dlm_lowcomms_stop(void); void dlm_lowcomms_exit(void); int dlm_lowcomms_close(int nodeid); -void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc); -void dlm_lowcomms_commit_buffer(void *mh); +struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, + char **ppc, void (*cb)(struct dlm_mhandle *mh), + struct dlm_mhandle *mh); +void dlm_lowcomms_commit_msg(struct dlm_msg *msg); +void dlm_lowcomms_put_msg(struct dlm_msg *msg); int dlm_lowcomms_connect_node(int nodeid); int dlm_lowcomms_nodes_set_mark(int nodeid, unsigned int mark); int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len); diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index 1a280dda99d3..aadb3781bebe 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -31,12 +31,15 @@ struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, gfp_t allocation, char **ppc) { - return dlm_lowcomms_get_buffer(nodeid, len, allocation, ppc); + return (struct dlm_mhandle *)dlm_lowcomms_new_msg(nodeid, len, + allocation, ppc, + NULL, NULL); } void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh) { - dlm_lowcomms_commit_buffer(mh); + dlm_lowcomms_commit_msg((struct dlm_msg *)mh); + dlm_lowcomms_put_msg((struct dlm_msg *)mh); } void dlm_midcomms_add_member(int nodeid) { } diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c index 2661674364af..6f653a339bea 100644 --- a/fs/dlm/rcom.c +++ b/fs/dlm/rcom.c @@ -70,21 +70,22 @@ static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len, static int create_rcom_stateless(struct dlm_ls *ls, int to_nodeid, int type, int len, struct dlm_rcom **rc_ret, - void **mh_ret) + struct dlm_msg **msg_ret) { int mb_len = sizeof(struct dlm_rcom) + len; - void *mh; + struct dlm_msg *msg; char *mb; - mh = dlm_lowcomms_get_buffer(to_nodeid, mb_len, GFP_NOFS, &mb); - if (!mh) { + msg = dlm_lowcomms_new_msg(to_nodeid, mb_len, GFP_NOFS, &mb, + NULL, NULL); + if (!msg) { log_print("create_rcom to %d type %d len %d ENOBUFS", to_nodeid, type, len); return -ENOBUFS; } _create_rcom(ls, to_nodeid, type, len, rc_ret, mb, mb_len); - *mh_ret = mh; + *msg_ret = msg; return 0; } @@ -100,11 +101,12 @@ static void send_rcom(struct dlm_ls *ls, struct dlm_mhandle *mh, dlm_midcomms_commit_mhandle(mh); } -static void send_rcom_stateless(struct dlm_ls *ls, void *mh, +static void send_rcom_stateless(struct dlm_ls *ls, struct dlm_msg *msg, struct dlm_rcom *rc) { _send_rcom(ls, rc); - dlm_lowcomms_commit_buffer(mh); + dlm_lowcomms_commit_msg(msg); + dlm_lowcomms_put_msg(msg); } static void set_rcom_status(struct dlm_ls *ls, struct rcom_status *rs, @@ -180,8 +182,8 @@ static void disallow_sync_reply(struct dlm_ls *ls) int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags) { struct dlm_rcom *rc; + struct dlm_msg *msg; int error = 0; - void *mh; ls->ls_recover_nodeid = nodeid; @@ -193,7 +195,7 @@ int dlm_rcom_status(struct dlm_ls *ls, int nodeid, uint32_t status_flags) retry: error = create_rcom_stateless(ls, nodeid, DLM_RCOM_STATUS, - sizeof(struct rcom_status), &rc, &mh); + sizeof(struct rcom_status), &rc, &msg); if (error) goto out; @@ -202,7 +204,7 @@ retry: allow_sync_reply(ls, &rc->rc_id); memset(ls->ls_recover_buf, 0, LOWCOMMS_MAX_TX_BUFFER_LEN); - send_rcom_stateless(ls, mh, rc); + send_rcom_stateless(ls, msg, rc); error = dlm_wait_function(ls, &rcom_response); disallow_sync_reply(ls); @@ -234,9 +236,9 @@ static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) uint32_t status; int nodeid = rc_in->rc_header.h_nodeid; int len = sizeof(struct rcom_config); + struct dlm_msg *msg; int num_slots = 0; int error; - void *mh; if (!dlm_slots_version(&rc_in->rc_header)) { status = dlm_recover_status(ls); @@ -258,7 +260,7 @@ static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) do_create: error = create_rcom_stateless(ls, nodeid, DLM_RCOM_STATUS_REPLY, - len, &rc, &mh); + len, &rc, &msg); if (error) return; @@ -285,7 +287,7 @@ static void receive_rcom_status(struct dlm_ls *ls, struct dlm_rcom *rc_in) spin_unlock(&ls->ls_recover_lock); do_send: - send_rcom_stateless(ls, mh, rc); + send_rcom_stateless(ls, msg, rc); } static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in) @@ -310,14 +312,14 @@ static void receive_sync_reply(struct dlm_ls *ls, struct dlm_rcom *rc_in) int dlm_rcom_names(struct dlm_ls *ls, int nodeid, char *last_name, int last_len) { struct dlm_rcom *rc; + struct dlm_msg *msg; int error = 0; - void *mh; ls->ls_recover_nodeid = nodeid; retry: error = create_rcom_stateless(ls, nodeid, DLM_RCOM_NAMES, last_len, - &rc, &mh); + &rc, &msg); if (error) goto out; memcpy(rc->rc_buf, last_name, last_len); @@ -325,7 +327,7 @@ retry: allow_sync_reply(ls, &rc->rc_id); memset(ls->ls_recover_buf, 0, LOWCOMMS_MAX_TX_BUFFER_LEN); - send_rcom_stateless(ls, mh, rc); + send_rcom_stateless(ls, msg, rc); error = dlm_wait_function(ls, &rcom_response); disallow_sync_reply(ls); @@ -339,14 +341,14 @@ static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in) { struct dlm_rcom *rc; int error, inlen, outlen, nodeid; - void *mh; + struct dlm_msg *msg; nodeid = rc_in->rc_header.h_nodeid; inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom); outlen = LOWCOMMS_MAX_TX_BUFFER_LEN - sizeof(struct dlm_rcom); error = create_rcom_stateless(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, - &rc, &mh); + &rc, &msg); if (error) return; rc->rc_id = rc_in->rc_id; @@ -354,7 +356,7 @@ static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in) dlm_copy_master_names(ls, rc_in->rc_buf, inlen, rc->rc_buf, outlen, nodeid); - send_rcom_stateless(ls, mh, rc); + send_rcom_stateless(ls, msg, rc); } int dlm_send_rcom_lookup(struct dlm_rsb *r, int dir_nodeid) -- cgit From 2874d1a68c4ec5623a05c8118f5dbaefb30b37ff Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:43 -0400 Subject: fs: dlm: add functionality to re-transmit a message This patch introduces a retransmit functionality for a lowcomms message handle. It's just allocates a new buffer and transmit it again, no special handling about prioritize it because keeping bytestream in order. To avoid another connection look some refactor was done to make a new buffer allocation with a preexisting connection pointer. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 85 ++++++++++++++++++++++++++++++++++++++++++------------- fs/dlm/lowcomms.h | 1 + 2 files changed, 67 insertions(+), 19 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index d222e6088ab2..df9827ec12f3 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -126,6 +126,8 @@ struct writequeue_entry { struct dlm_msg { struct writequeue_entry *entry; + struct dlm_msg *orig_msg; + bool retransmit; void *ppc; int len; int idx; /* new()/commit() idx exchange */ @@ -1055,6 +1057,10 @@ static void free_entry(struct writequeue_entry *e) struct dlm_msg *msg, *tmp; list_for_each_entry_safe(msg, tmp, &e->msgs, list) { + if (msg->orig_msg) { + msg->orig_msg->retransmit = false; + kref_put(&msg->orig_msg->ref, dlm_msg_release); + } list_del(&msg->list); kref_put(&msg->ref, dlm_msg_release); } @@ -1494,11 +1500,37 @@ static struct writequeue_entry *new_wq_entry(struct connection *con, int len, return e; }; +static struct dlm_msg *dlm_lowcomms_new_msg_con(struct connection *con, int len, + gfp_t allocation, char **ppc, + void (*cb)(struct dlm_mhandle *mh), + struct dlm_mhandle *mh) +{ + struct writequeue_entry *e; + struct dlm_msg *msg; + + msg = kzalloc(sizeof(*msg), allocation); + if (!msg) + return NULL; + + kref_init(&msg->ref); + + e = new_wq_entry(con, len, allocation, ppc, cb, mh); + if (!e) { + kfree(msg); + return NULL; + } + + msg->ppc = *ppc; + msg->len = len; + msg->entry = e; + + return msg; +} + struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, char **ppc, void (*cb)(struct dlm_mhandle *mh), struct dlm_mhandle *mh) { - struct writequeue_entry *e; struct connection *con; struct dlm_msg *msg; int idx; @@ -1518,32 +1550,18 @@ struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, return NULL; } - msg = kzalloc(sizeof(*msg), allocation); + msg = dlm_lowcomms_new_msg_con(con, len, allocation, ppc, cb, mh); if (!msg) { srcu_read_unlock(&connections_srcu, idx); return NULL; } - kref_init(&msg->ref); - - e = new_wq_entry(con, len, allocation, ppc, cb, mh); - if (!e) { - srcu_read_unlock(&connections_srcu, idx); - kfree(msg); - return NULL; - } - - msg->ppc = *ppc; - msg->len = len; - msg->entry = e; - /* we assume if successful commit must called */ msg->idx = idx; - return msg; } -void dlm_lowcomms_commit_msg(struct dlm_msg *msg) +static void _dlm_lowcomms_commit_msg(struct dlm_msg *msg) { struct writequeue_entry *e = msg->entry; struct connection *con = e->con; @@ -1561,20 +1579,49 @@ void dlm_lowcomms_commit_msg(struct dlm_msg *msg) spin_unlock(&con->writequeue_lock); queue_work(send_workqueue, &con->swork); - srcu_read_unlock(&connections_srcu, msg->idx); return; out: spin_unlock(&con->writequeue_lock); - srcu_read_unlock(&connections_srcu, msg->idx); return; } +void dlm_lowcomms_commit_msg(struct dlm_msg *msg) +{ + _dlm_lowcomms_commit_msg(msg); + srcu_read_unlock(&connections_srcu, msg->idx); +} + void dlm_lowcomms_put_msg(struct dlm_msg *msg) { kref_put(&msg->ref, dlm_msg_release); } +/* does not held connections_srcu, usage workqueue only */ +int dlm_lowcomms_resend_msg(struct dlm_msg *msg) +{ + struct dlm_msg *msg_resend; + char *ppc; + + if (msg->retransmit) + return 1; + + msg_resend = dlm_lowcomms_new_msg_con(msg->entry->con, msg->len, + GFP_ATOMIC, &ppc, NULL, NULL); + if (!msg_resend) + return -ENOMEM; + + msg->retransmit = true; + kref_get(&msg->ref); + msg_resend->orig_msg = msg; + + memcpy(ppc, msg->ppc, msg->len); + _dlm_lowcomms_commit_msg(msg_resend); + dlm_lowcomms_put_msg(msg_resend); + + return 0; +} + /* Send a message */ static void send_to_sock(struct connection *con) { diff --git a/fs/dlm/lowcomms.h b/fs/dlm/lowcomms.h index cdb8f066f0d8..a4384826442c 100644 --- a/fs/dlm/lowcomms.h +++ b/fs/dlm/lowcomms.h @@ -27,6 +27,7 @@ struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, struct dlm_mhandle *mh); void dlm_lowcomms_commit_msg(struct dlm_msg *msg); void dlm_lowcomms_put_msg(struct dlm_msg *msg); +int dlm_lowcomms_resend_msg(struct dlm_msg *msg); int dlm_lowcomms_connect_node(int nodeid); int dlm_lowcomms_nodes_set_mark(int nodeid, unsigned int mark); int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len); -- cgit From 37a247da517f4315eed21585be8aa516e0b9cec9 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:44 -0400 Subject: fs: dlm: move out some hash functionality This patch moves out some lowcomms hash functionality into lowcomms header to provide them to other layers like midcomms as well. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 9 --------- fs/dlm/lowcomms.h | 10 ++++++++++ 2 files changed, 10 insertions(+), 9 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index df9827ec12f3..1f2759cfda09 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -59,7 +59,6 @@ #include "config.h" #define NEEDED_RMEM (4*1024*1024) -#define CONN_HASH_SIZE 32 /* Number of messages to send before rescheduling */ #define MAX_SEND_MSG_COUNT 25 @@ -175,14 +174,6 @@ static void sctp_connect_to_sock(struct connection *con); static void tcp_connect_to_sock(struct connection *con); static void dlm_tcp_shutdown(struct connection *con); -/* This is deliberately very simple because most clusters have simple - sequential nodeids, so we should be able to go straight to a connection - struct in the array */ -static inline int nodeid_hash(int nodeid) -{ - return nodeid & (CONN_HASH_SIZE-1); -} - static struct connection *__find_con(int nodeid, int r) { struct connection *con; diff --git a/fs/dlm/lowcomms.h b/fs/dlm/lowcomms.h index a4384826442c..66dc1bb3de7f 100644 --- a/fs/dlm/lowcomms.h +++ b/fs/dlm/lowcomms.h @@ -13,6 +13,16 @@ #define __LOWCOMMS_DOT_H__ #define LOWCOMMS_MAX_TX_BUFFER_LEN 4096 +#define CONN_HASH_SIZE 32 + +/* This is deliberately very simple because most clusters have simple + * sequential nodeids, so we should be able to go straight to a connection + * struct in the array + */ +static inline int nodeid_hash(int nodeid) +{ + return nodeid & (CONN_HASH_SIZE-1); +} /* switch to check if dlm is running */ extern int dlm_allow_conn; -- cgit From 8e2e40860c7f67c0b19b13d92cfea03a19232ce2 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:45 -0400 Subject: fs: dlm: add union in dlm header for lockspace id This patch adds union inside the lockspace id to handle it also for another use case for a different dlm command. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/dlm_internal.h | 5 ++++- fs/dlm/lock.c | 8 ++++---- fs/dlm/rcom.c | 4 ++-- fs/dlm/util.c | 6 ++++-- 4 files changed, 14 insertions(+), 9 deletions(-) diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h index e8dc5f4f1f9e..8f5980909d80 100644 --- a/fs/dlm/dlm_internal.h +++ b/fs/dlm/dlm_internal.h @@ -380,7 +380,10 @@ static inline int rsb_flag(struct dlm_rsb *r, enum rsb_flags flag) struct dlm_header { uint32_t h_version; - uint32_t h_lockspace; + union { + /* for DLM_MSG and DLM_RCOM */ + uint32_t h_lockspace; + } u; uint32_t h_nodeid; /* nodeid of sender */ uint16_t h_length; uint8_t h_cmd; /* DLM_MSG, DLM_RCOM */ diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c index b625ce92464a..c502c065d007 100644 --- a/fs/dlm/lock.c +++ b/fs/dlm/lock.c @@ -3544,7 +3544,7 @@ static int _create_message(struct dlm_ls *ls, int mb_len, ms = (struct dlm_message *) mb; ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - ms->m_header.h_lockspace = ls->ls_global_id; + ms->m_header.u.h_lockspace = ls->ls_global_id; ms->m_header.h_nodeid = dlm_our_nodeid(); ms->m_header.h_length = mb_len; ms->m_header.h_cmd = DLM_MSG; @@ -5038,16 +5038,16 @@ void dlm_receive_buffer(union dlm_packet *p, int nodeid) if (hd->h_nodeid != nodeid) { log_print("invalid h_nodeid %d from %d lockspace %x", - hd->h_nodeid, nodeid, hd->h_lockspace); + hd->h_nodeid, nodeid, hd->u.h_lockspace); return; } - ls = dlm_find_lockspace_global(hd->h_lockspace); + ls = dlm_find_lockspace_global(hd->u.h_lockspace); if (!ls) { if (dlm_config.ci_log_debug) { printk_ratelimited(KERN_DEBUG "dlm: invalid lockspace " "%u from %d cmd %d type %d\n", - hd->h_lockspace, nodeid, hd->h_cmd, type); + hd->u.h_lockspace, nodeid, hd->h_cmd, type); } if (hd->h_cmd == DLM_RCOM && type == DLM_RCOM_STATUS) diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c index 6f653a339bea..7c1a06bcd418 100644 --- a/fs/dlm/rcom.c +++ b/fs/dlm/rcom.c @@ -35,7 +35,7 @@ static void _create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len, rc = (struct dlm_rcom *) mb; rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - rc->rc_header.h_lockspace = ls->ls_global_id; + rc->rc_header.u.h_lockspace = ls->ls_global_id; rc->rc_header.h_nodeid = dlm_our_nodeid(); rc->rc_header.h_length = mb_len; rc->rc_header.h_cmd = DLM_RCOM; @@ -508,7 +508,7 @@ int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in) rc = (struct dlm_rcom *) mb; rc->rc_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); - rc->rc_header.h_lockspace = rc_in->rc_header.h_lockspace; + rc->rc_header.u.h_lockspace = rc_in->rc_header.u.h_lockspace; rc->rc_header.h_nodeid = dlm_our_nodeid(); rc->rc_header.h_length = mb_len; rc->rc_header.h_cmd = DLM_RCOM; diff --git a/fs/dlm/util.c b/fs/dlm/util.c index 74a8c5bfe9b5..58acbcc2081a 100644 --- a/fs/dlm/util.c +++ b/fs/dlm/util.c @@ -23,7 +23,8 @@ void header_out(struct dlm_header *hd) { hd->h_version = cpu_to_le32(hd->h_version); - hd->h_lockspace = cpu_to_le32(hd->h_lockspace); + /* does it for others u32 in union as well */ + hd->u.h_lockspace = cpu_to_le32(hd->u.h_lockspace); hd->h_nodeid = cpu_to_le32(hd->h_nodeid); hd->h_length = cpu_to_le16(hd->h_length); } @@ -31,7 +32,8 @@ void header_out(struct dlm_header *hd) void header_in(struct dlm_header *hd) { hd->h_version = le32_to_cpu(hd->h_version); - hd->h_lockspace = le32_to_cpu(hd->h_lockspace); + /* does it for others u32 in union as well */ + hd->u.h_lockspace = le32_to_cpu(hd->u.h_lockspace); hd->h_nodeid = le32_to_cpu(hd->h_nodeid); hd->h_length = le16_to_cpu(hd->h_length); } -- cgit From 489d8e559c6596eb08e16447d9830bc39afbe54e Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:46 -0400 Subject: fs: dlm: add reliable connection if reconnect This patch introduce to make a tcp lowcomms connection reliable even if reconnects occurs. This is done by an application layer re-transmission handling and sequence numbers in dlm protocols. There are three new dlm commands: DLM_OPTS: This will encapsulate an existing dlm message (and rcom message if they don't have an own application side re-transmission handling). As optional handling additional tlv's (type length fields) can be appended. This can be for example a sequence number field. However because in DLM_OPTS the lockspace field is unused and a sequence number is a mandatory field it isn't made as a tlv and we put the sequence number inside the lockspace id. The possibility to add optional options are still there for future purposes. DLM_ACK: Just a dlm header to acknowledge the receive of a DLM_OPTS message to it's sender. DLM_FIN: This provides a 4 way handshake for connection termination inclusive support for half-closed connections. It's provided on application layer because SCTP doesn't support half-closed sockets, the shutdown() call can interrupted by e.g. TCP resets itself and a hard logic to implement it because the othercon paradigm in lowcomms. The 4-way termination handshake also solve problems to synchronize peer EOF arrival and that the cluster manager removes the peer in the node membership handling of DLM. In some cases messages can be still transmitted in this time and we need to wait for the node membership event. To provide a reliable connection the node will retransmit all unacknowledges message to it's peer on reconnect. The receiver will then filtering out the next received message and drop all messages which are duplicates. As RCOM_STATUS and RCOM_NAMES messages are the first messages which are exchanged and they have they own re-transmission handling, there exists logic that these messages must be first. If these messages arrives we store the dlm version field. This handling is on DLM 3.1 and after this patch 3.2 the same. A backwards compatibility handling has been added which seems to work on tests without tcpkill, however it's not recommended to use DLM 3.1 and 3.2 at the same time, because DLM 3.2 tries to fix long term bugs in the DLM protocol. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/dlm_internal.h | 30 +- fs/dlm/lockspace.c | 7 +- fs/dlm/lowcomms.c | 4 +- fs/dlm/lowcomms.h | 7 +- fs/dlm/midcomms.c | 1290 +++++++++++++++++++++++++++++++++++++++++++++++-- fs/dlm/midcomms.h | 1 + fs/dlm/rcom.c | 4 +- 7 files changed, 1292 insertions(+), 51 deletions(-) diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h index 8f5980909d80..40917c878370 100644 --- a/fs/dlm/dlm_internal.h +++ b/fs/dlm/dlm_internal.h @@ -371,18 +371,26 @@ static inline int rsb_flag(struct dlm_rsb *r, enum rsb_flags flag) /* dlm_header is first element of all structs sent between nodes */ #define DLM_HEADER_MAJOR 0x00030000 -#define DLM_HEADER_MINOR 0x00000001 +#define DLM_HEADER_MINOR 0x00000002 + +#define DLM_VERSION_3_1 0x00030001 +#define DLM_VERSION_3_2 0x00030002 #define DLM_HEADER_SLOTS 0x00000001 #define DLM_MSG 1 #define DLM_RCOM 2 +#define DLM_OPTS 3 +#define DLM_ACK 4 +#define DLM_FIN 5 struct dlm_header { uint32_t h_version; union { /* for DLM_MSG and DLM_RCOM */ uint32_t h_lockspace; + /* for DLM_ACK and DLM_OPTS */ + uint32_t h_seq; } u; uint32_t h_nodeid; /* nodeid of sender */ uint16_t h_length; @@ -390,7 +398,6 @@ struct dlm_header { uint8_t h_pad; }; - #define DLM_MSG_REQUEST 1 #define DLM_MSG_CONVERT 2 #define DLM_MSG_UNLOCK 3 @@ -458,10 +465,29 @@ struct dlm_rcom { char rc_buf[]; }; +struct dlm_opt_header { + uint16_t t_type; + uint16_t t_length; + uint32_t o_pad; + /* need to be 8 byte aligned */ + char t_value[]; +}; + +/* encapsulation header */ +struct dlm_opts { + struct dlm_header o_header; + uint8_t o_nextcmd; + uint8_t o_pad; + uint16_t o_optlen; + uint32_t o_pad2; + char o_opts[]; +}; + union dlm_packet { struct dlm_header header; /* common to other two */ struct dlm_message message; struct dlm_rcom rcom; + struct dlm_opts opts; }; #define DLM_RSF_NEED_SLOTS 0x00000001 diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c index bf5c55ef9d0d..2b738be8d7e4 100644 --- a/fs/dlm/lockspace.c +++ b/fs/dlm/lockspace.c @@ -567,7 +567,12 @@ static int new_lockspace(const char *name, const char *cluster, mutex_init(&ls->ls_requestqueue_mutex); mutex_init(&ls->ls_clear_proc_locks); - ls->ls_recover_buf = kmalloc(LOWCOMMS_MAX_TX_BUFFER_LEN, GFP_NOFS); + /* Due backwards compatibility with 3.1 we need to use maximum + * possible dlm message size to be sure the message will fit and + * not having out of bounds issues. However on sending side 3.2 + * might send less. + */ + ls->ls_recover_buf = kmalloc(DEFAULT_BUFFER_SIZE, GFP_NOFS); if (!ls->ls_recover_buf) goto out_lkbidr; diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 1f2759cfda09..fe9113bd5ba0 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -1762,8 +1762,10 @@ static void process_send_sockets(struct work_struct *work) clear_bit(CF_WRITE_PENDING, &con->flags); - if (test_and_clear_bit(CF_RECONNECT, &con->flags)) + if (test_and_clear_bit(CF_RECONNECT, &con->flags)) { close_connection(con, false, false, true); + dlm_midcomms_unack_msg_resend(con->nodeid); + } if (con->sock == NULL) { /* not mutex protected so check it inside too */ if (test_and_clear_bit(CF_DELAY_CONNECT, &con->flags)) diff --git a/fs/dlm/lowcomms.h b/fs/dlm/lowcomms.h index 66dc1bb3de7f..730c34317183 100644 --- a/fs/dlm/lowcomms.h +++ b/fs/dlm/lowcomms.h @@ -12,7 +12,12 @@ #ifndef __LOWCOMMS_DOT_H__ #define __LOWCOMMS_DOT_H__ -#define LOWCOMMS_MAX_TX_BUFFER_LEN 4096 +#include "dlm_internal.h" + +#define DLM_MIDCOMMS_OPT_LEN sizeof(struct dlm_opts) +#define LOWCOMMS_MAX_TX_BUFFER_LEN (DEFAULT_BUFFER_SIZE - \ + DLM_MIDCOMMS_OPT_LEN) + #define CONN_HASH_SIZE 32 /* This is deliberately very simple because most clusters have simple diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index aadb3781bebe..eef3938a363e 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -3,7 +3,7 @@ ******************************************************************************* ** ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved. -** Copyright (C) 2004-2008 Red Hat, Inc. All rights reserved. +** Copyright (C) 2004-2021 Red Hat, Inc. All rights reserved. ** ** ******************************************************************************* @@ -12,53 +12,817 @@ /* * midcomms.c * - * This is the appallingly named "mid-level" comms layer. + * This is the appallingly named "mid-level" comms layer. It takes care about + * deliver an on application layer "reliable" communication above the used + * lowcomms transport layer. * - * Its purpose is to take packets from the "real" comms layer, - * split them up into packets and pass them to the interested - * part of the locking mechanism. + * How it works: * - * It also takes messages from the locking layer, formats them - * into packets and sends them to the comms layer. + * Each nodes keeps track of all send DLM messages in send_queue with a sequence + * number. The receive will send an DLM_ACK message back for every DLM message + * received at the other side. If a reconnect happens in lowcomms we will send + * all unacknowledged dlm messages again. The receiving side might drop any already + * received message by comparing sequence numbers. + * + * How version detection works: + * + * Due the fact that dlm has pre-configured node addresses on every side + * it is in it's nature that every side connects at starts to transmit + * dlm messages which ends in a race. However DLM_RCOM_NAMES, DLM_RCOM_STATUS + * and their replies are the first messages which are exchanges. Due backwards + * compatibility these messages are not covered by the midcomms re-transmission + * layer. These messages have their own re-transmission handling in the dlm + * application layer. The version field of every node will be set on these RCOM + * messages as soon as they arrived and the node isn't yet part of the nodes + * hash. There exists also logic to detect version mismatched if something weird + * going on or the first messages isn't an expected one. + * + * Termination: + * + * The midcomms layer does a 4 way handshake for termination on DLM protocol + * like TCP supports it with half-closed socket support. SCTP doesn't support + * half-closed socket, so we do it on DLM layer. Also socket shutdown() can be + * interrupted by .e.g. tcp reset itself. Additional there exists the othercon + * paradigm in lowcomms which cannot be easily without breaking backwards + * compatibility. A node cannot send anything to another node when a DLM_FIN + * message was send. There exists additional logic to print a warning if + * DLM wants to do it. There exists a state handling like RFC 793 but reduced + * to termination only. The event "member removal event" describes the cluster + * manager removed the node from internal lists, at this point DLM does not + * send any message to the other node. There exists two cases: + * + * 1. The cluster member was removed and we received a FIN + * OR + * 2. We received a FIN but the member was not removed yet + * + * One of these cases will do the CLOSE_WAIT to LAST_ACK change. + * + * + * +---------+ + * | CLOSED | + * +---------+ + * | add member/receive RCOM version + * | detection msg + * V + * +---------+ + * | ESTAB | + * +---------+ + * CLOSE | | rcv FIN + * ------- | | ------- + * +---------+ snd FIN / \ snd ACK +---------+ + * | FIN |<----------------- ------------------>| CLOSE | + * | WAIT-1 |------------------ | WAIT | + * +---------+ rcv FIN \ +---------+ + * | rcv ACK of FIN ------- | CLOSE | member + * | -------------- snd ACK | ------- | removal + * V x V snd FIN V event + * +---------+ +---------+ +---------+ + * |FINWAIT-2| | CLOSING | | LAST-ACK| + * +---------+ +---------+ +---------+ + * | rcv ACK of FIN | rcv ACK of FIN | + * | rcv FIN -------------- | -------------- | + * | ------- x V x V + * \ snd ACK +---------+ +---------+ + * ------------------------>| CLOSED | | CLOSED | + * +---------+ +---------+ + * + * NOTE: any state can interrupted by midcomms_close() and state will be + * switched to CLOSED in case of fencing. There exists also some timeout + * handling when we receive the version detection RCOM messages which is + * made by observation. + * + * Future improvements: + * + * There exists some known issues/improvements of the dlm handling. Some + * of them should be done in a next major dlm version bump which makes + * it incompatible with previous versions. + * + * Unaligned memory access: + * + * There exists cases when the dlm message buffer length is not aligned + * to 8 byte. However seems nobody detected any problem with it. This + * can be fixed in the next major version bump of dlm. + * + * Version detection: + * + * The version detection and how it's done is related to backwards + * compatibility. There exists better ways to make a better handling. + * However this should be changed in the next major version bump of dlm. + * + * Ack handling: + * + * Currently we send an ack message for every dlm message. However we + * can ack multiple dlm messages with one ack by just delaying the ack + * message. Will reduce some traffic but makes the drop detection slower. + * + * Tail Size checking: + * + * There exists a message tail payload in e.g. DLM_MSG however we don't + * check it against the message length yet regarding to the receive buffer + * length. That need to be validated. + * + * Fencing bad nodes: + * + * At timeout places or weird sequence number behaviours we should send + * a fencing request to the cluster manager. */ +/* Debug switch to enable a 5 seconds sleep waiting of a termination. + * This can be useful to test fencing while termination is running. + * This requires a setup with only gfs2 as dlm user, so that the + * last umount will terminate the connection. + * + * However it became useful to test, while the 5 seconds block in umount + * just press the reset button. In a lot of dropping the termination + * process can could take several seconds. + */ +#define DLM_DEBUG_FENCE_TERMINATION 0 + +#include + #include "dlm_internal.h" #include "lowcomms.h" #include "config.h" #include "lock.h" +#include "util.h" #include "midcomms.h" -struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, - gfp_t allocation, char **ppc) +/* init value for sequence numbers for testing purpose only e.g. overflows */ +#define DLM_SEQ_INIT 0 +/* 3 minutes wait to sync ending of dlm */ +#define DLM_SHUTDOWN_TIMEOUT msecs_to_jiffies(3 * 60 * 1000) +#define DLM_VERSION_NOT_SET 0 + +struct midcomms_node { + int nodeid; + uint32_t version; + uint32_t seq_send; + uint32_t seq_next; + /* These queues are unbound because we cannot drop any message in dlm. + * We could send a fence signal for a specific node to the cluster + * manager if queues hits some maximum value, however this handling + * not supported yet. + */ + struct list_head send_queue; + spinlock_t send_queue_lock; + atomic_t send_queue_cnt; +#define DLM_NODE_FLAG_CLOSE 1 +#define DLM_NODE_FLAG_STOP_TX 2 +#define DLM_NODE_FLAG_STOP_RX 3 + unsigned long flags; + wait_queue_head_t shutdown_wait; + + /* dlm tcp termination state */ +#define DLM_CLOSED 1 +#define DLM_ESTABLISHED 2 +#define DLM_FIN_WAIT1 3 +#define DLM_FIN_WAIT2 4 +#define DLM_CLOSE_WAIT 5 +#define DLM_LAST_ACK 6 +#define DLM_CLOSING 7 + int state; + spinlock_t state_lock; + + /* counts how many lockspaces are using this node + * this refcount is necessary to determine if the + * node wants to disconnect. + */ + int users; + + struct hlist_node hlist; + struct rcu_head rcu; +}; + +struct dlm_mhandle { + const struct dlm_header *inner_hd; + struct midcomms_node *node; + struct dlm_opts *opts; + struct dlm_msg *msg; + bool committed; + uint32_t seq; + + void (*ack_rcv)(struct midcomms_node *node); + + /* get_mhandle/commit srcu idx exchange */ + int idx; + + struct list_head list; + struct rcu_head rcu; +}; + +static struct hlist_head node_hash[CONN_HASH_SIZE]; +static DEFINE_SPINLOCK(nodes_lock); +DEFINE_STATIC_SRCU(nodes_srcu); + +/* This mutex prevents that midcomms_close() is running while + * stop() or remove(). As I experienced invalid memory access + * behaviours when DLM_DEBUG_FENCE_TERMINATION is enabled and + * resetting machines. I will end in some double deletion in nodes + * datastructure. + */ +static DEFINE_MUTEX(close_lock); + +static inline const char *dlm_state_str(int state) { - return (struct dlm_mhandle *)dlm_lowcomms_new_msg(nodeid, len, - allocation, ppc, - NULL, NULL); + switch (state) { + case DLM_CLOSED: + return "CLOSED"; + case DLM_ESTABLISHED: + return "ESTABLISHED"; + case DLM_FIN_WAIT1: + return "FIN_WAIT1"; + case DLM_FIN_WAIT2: + return "FIN_WAIT2"; + case DLM_CLOSE_WAIT: + return "CLOSE_WAIT"; + case DLM_LAST_ACK: + return "LAST_ACK"; + case DLM_CLOSING: + return "CLOSING"; + default: + return "UNKNOWN"; + } } -void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh) +static struct midcomms_node *__find_node(int nodeid, int r) { - dlm_lowcomms_commit_msg((struct dlm_msg *)mh); - dlm_lowcomms_put_msg((struct dlm_msg *)mh); + struct midcomms_node *node; + + hlist_for_each_entry_rcu(node, &node_hash[r], hlist) { + if (node->nodeid == nodeid) + return node; + } + + return NULL; } -void dlm_midcomms_add_member(int nodeid) { } +static void dlm_mhandle_release(struct rcu_head *rcu) +{ + struct dlm_mhandle *mh = container_of(rcu, struct dlm_mhandle, rcu); -void dlm_midcomms_remove_member(int nodeid) { } + dlm_lowcomms_put_msg(mh->msg); + kfree(mh); +} -int dlm_midcomms_start(void) +static void dlm_send_queue_flush(struct midcomms_node *node) { - return dlm_lowcomms_start(); + struct dlm_mhandle *mh; + + pr_debug("flush midcomms send queue of node %d\n", node->nodeid); + + rcu_read_lock(); + list_for_each_entry_rcu(mh, &node->send_queue, list) { + spin_lock(&node->send_queue_lock); + list_del_rcu(&mh->list); + spin_unlock(&node->send_queue_lock); + + atomic_dec(&node->send_queue_cnt); + + call_rcu(&mh->rcu, dlm_mhandle_release); + } + rcu_read_unlock(); } -void dlm_midcomms_shutdown(void) +static void midcomms_node_reset(struct midcomms_node *node) { - dlm_lowcomms_shutdown(); + pr_debug("reset node %d\n", node->nodeid); + + node->seq_next = DLM_SEQ_INIT; + node->seq_send = DLM_SEQ_INIT; + node->version = DLM_VERSION_NOT_SET; + node->flags = 0; + + dlm_send_queue_flush(node); + node->state = DLM_CLOSED; + wake_up(&node->shutdown_wait); } -int dlm_midcomms_close(int nodeid) +static struct midcomms_node *nodeid2node(int nodeid, gfp_t alloc) +{ + struct midcomms_node *node, *tmp; + int r = nodeid_hash(nodeid); + + node = __find_node(nodeid, r); + if (node || !alloc) + return node; + + node = kmalloc(sizeof(*node), alloc); + if (!node) + return NULL; + + node->nodeid = nodeid; + spin_lock_init(&node->state_lock); + spin_lock_init(&node->send_queue_lock); + atomic_set(&node->send_queue_cnt, 0); + INIT_LIST_HEAD(&node->send_queue); + init_waitqueue_head(&node->shutdown_wait); + node->users = 0; + midcomms_node_reset(node); + + spin_lock(&nodes_lock); + /* check again if there was somebody else + * earlier here to add the node + */ + tmp = __find_node(nodeid, r); + if (tmp) { + spin_unlock(&nodes_lock); + kfree(node); + return tmp; + } + + hlist_add_head_rcu(&node->hlist, &node_hash[r]); + spin_unlock(&nodes_lock); + return node; +} + +static int dlm_send_ack(int nodeid, uint32_t seq) +{ + int mb_len = sizeof(struct dlm_header); + struct dlm_header *m_header; + struct dlm_msg *msg; + char *ppc; + + msg = dlm_lowcomms_new_msg(nodeid, mb_len, GFP_NOFS, &ppc, + NULL, NULL); + if (!msg) + return -ENOMEM; + + m_header = (struct dlm_header *)ppc; + + m_header->h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + m_header->h_nodeid = dlm_our_nodeid(); + m_header->h_length = mb_len; + m_header->h_cmd = DLM_ACK; + m_header->u.h_seq = seq; + + header_out(m_header); + dlm_lowcomms_commit_msg(msg); + dlm_lowcomms_put_msg(msg); + + return 0; +} + +static int dlm_send_fin(struct midcomms_node *node, + void (*ack_rcv)(struct midcomms_node *node)) +{ + int mb_len = sizeof(struct dlm_header); + struct dlm_header *m_header; + struct dlm_mhandle *mh; + char *ppc; + + mh = dlm_midcomms_get_mhandle(node->nodeid, mb_len, GFP_NOFS, &ppc); + if (!mh) + return -ENOMEM; + + mh->ack_rcv = ack_rcv; + + m_header = (struct dlm_header *)ppc; + + m_header->h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + m_header->h_nodeid = dlm_our_nodeid(); + m_header->h_length = mb_len; + m_header->h_cmd = DLM_FIN; + + header_out(m_header); + + pr_debug("sending fin msg to node %d\n", node->nodeid); + dlm_midcomms_commit_mhandle(mh); + set_bit(DLM_NODE_FLAG_STOP_TX, &node->flags); + + return 0; +} + +static void dlm_receive_ack(struct midcomms_node *node, uint32_t seq) +{ + struct dlm_mhandle *mh; + + rcu_read_lock(); + list_for_each_entry_rcu(mh, &node->send_queue, list) { + if (before(mh->seq, seq)) { + spin_lock(&node->send_queue_lock); + list_del_rcu(&mh->list); + spin_unlock(&node->send_queue_lock); + + atomic_dec(&node->send_queue_cnt); + + if (mh->ack_rcv) + mh->ack_rcv(node); + + call_rcu(&mh->rcu, dlm_mhandle_release); + } else { + /* send queue should be ordered */ + break; + } + } + rcu_read_unlock(); +} + +static void dlm_pas_fin_ack_rcv(struct midcomms_node *node) +{ + spin_lock(&node->state_lock); + pr_debug("receive passive fin ack from node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + + switch (node->state) { + case DLM_LAST_ACK: + /* DLM_CLOSED */ + midcomms_node_reset(node); + break; + case DLM_CLOSED: + /* not valid but somehow we got what we want */ + wake_up(&node->shutdown_wait); + break; + default: + spin_unlock(&node->state_lock); + log_print("%s: unexpected state: %d\n", + __func__, node->state); + WARN_ON(1); + return; + } + spin_unlock(&node->state_lock); +} + +static void dlm_midcomms_receive_buffer(union dlm_packet *p, + struct midcomms_node *node, + uint32_t seq) +{ + if (seq == node->seq_next) { + node->seq_next++; + /* send ack before fin */ + dlm_send_ack(node->nodeid, node->seq_next); + + switch (p->header.h_cmd) { + case DLM_FIN: + spin_lock(&node->state_lock); + pr_debug("receive fin msg from node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + + switch (node->state) { + case DLM_ESTABLISHED: + node->state = DLM_CLOSE_WAIT; + pr_debug("switch node %d to state %s\n", + node->nodeid, dlm_state_str(node->state)); + /* passive shutdown DLM_LAST_ACK case 1 + * additional we check if the node is used by + * cluster manager events at all. + */ + if (node->users == 0) { + node->state = DLM_LAST_ACK; + pr_debug("switch node %d to state %s case 1\n", + node->nodeid, dlm_state_str(node->state)); + spin_unlock(&node->state_lock); + goto send_fin; + } + break; + case DLM_FIN_WAIT1: + node->state = DLM_CLOSING; + pr_debug("switch node %d to state %s\n", + node->nodeid, dlm_state_str(node->state)); + break; + case DLM_FIN_WAIT2: + midcomms_node_reset(node); + pr_debug("switch node %d to state %s\n", + node->nodeid, dlm_state_str(node->state)); + wake_up(&node->shutdown_wait); + break; + case DLM_LAST_ACK: + /* probably remove_member caught it, do nothing */ + break; + default: + spin_unlock(&node->state_lock); + log_print("%s: unexpected state: %d\n", + __func__, node->state); + WARN_ON(1); + return; + } + spin_unlock(&node->state_lock); + + set_bit(DLM_NODE_FLAG_STOP_RX, &node->flags); + break; + default: + WARN_ON(test_bit(DLM_NODE_FLAG_STOP_RX, &node->flags)); + dlm_receive_buffer(p, node->nodeid); + break; + } + } else { + /* retry to ack message which we already have by sending back + * current node->seq_next number as ack. + */ + if (seq < node->seq_next) + dlm_send_ack(node->nodeid, node->seq_next); + + log_print_ratelimited("ignore dlm msg because seq mismatch, seq: %u, expected: %u, nodeid: %d", + seq, node->seq_next, node->nodeid); + } + + return; + +send_fin: + set_bit(DLM_NODE_FLAG_STOP_RX, &node->flags); + dlm_send_fin(node, dlm_pas_fin_ack_rcv); +} + +static struct midcomms_node * +dlm_midcomms_recv_node_lookup(int nodeid, const union dlm_packet *p, + uint16_t msglen, int (*cb)(struct midcomms_node *node)) +{ + struct midcomms_node *node = NULL; + gfp_t allocation = 0; + int ret; + + switch (p->header.h_cmd) { + case DLM_RCOM: + if (msglen < sizeof(struct dlm_rcom)) { + log_print("rcom msg too small: %u, will skip this message from node %d", + msglen, nodeid); + return NULL; + } + + switch (le32_to_cpu(p->rcom.rc_type)) { + case DLM_RCOM_NAMES: + fallthrough; + case DLM_RCOM_NAMES_REPLY: + fallthrough; + case DLM_RCOM_STATUS: + fallthrough; + case DLM_RCOM_STATUS_REPLY: + node = nodeid2node(nodeid, 0); + if (node) { + spin_lock(&node->state_lock); + if (node->state != DLM_ESTABLISHED) + pr_debug("receive begin RCOM msg from node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + + switch (node->state) { + case DLM_CLOSED: + node->state = DLM_ESTABLISHED; + pr_debug("switch node %d to state %s\n", + node->nodeid, dlm_state_str(node->state)); + break; + case DLM_ESTABLISHED: + break; + default: + /* some invalid state passive shutdown + * was failed, we try to reset and + * hope it will go on. + */ + log_print("reset node %d because shutdown stucked", + node->nodeid); + + midcomms_node_reset(node); + node->state = DLM_ESTABLISHED; + break; + } + spin_unlock(&node->state_lock); + } + + allocation = GFP_NOFS; + break; + default: + break; + } + + break; + default: + break; + } + + node = nodeid2node(nodeid, allocation); + if (!node) { + log_print_ratelimited("received dlm message cmd %d nextcmd %d from node %d in an invalid sequence", + p->header.h_cmd, p->opts.o_nextcmd, nodeid); + return NULL; + } + + ret = cb(node); + if (ret < 0) + return NULL; + + return node; +} + +static int dlm_midcomms_version_check_3_2(struct midcomms_node *node) +{ + switch (node->version) { + case DLM_VERSION_NOT_SET: + node->version = DLM_VERSION_3_2; + log_print("version 0x%08x for node %d detected", DLM_VERSION_3_2, + node->nodeid); + break; + case DLM_VERSION_3_2: + break; + default: + log_print_ratelimited("version mismatch detected, assumed 0x%08x but node %d has 0x%08x", + DLM_VERSION_3_2, node->nodeid, node->version); + return -1; + } + + return 0; +} + +static int dlm_opts_check_msglen(union dlm_packet *p, uint16_t msglen, int nodeid) +{ + int len = msglen; + + /* we only trust outer header msglen because + * it's checked against receive buffer length. + */ + if (len < sizeof(struct dlm_opts)) + return -1; + len -= sizeof(struct dlm_opts); + + if (len < le16_to_cpu(p->opts.o_optlen)) + return -1; + len -= le16_to_cpu(p->opts.o_optlen); + + switch (p->opts.o_nextcmd) { + case DLM_FIN: + if (len < sizeof(struct dlm_header)) { + log_print("fin too small: %d, will skip this message from node %d", + len, nodeid); + return -1; + } + + break; + case DLM_MSG: + if (len < sizeof(struct dlm_message)) { + log_print("msg too small: %d, will skip this message from node %d", + msglen, nodeid); + return -1; + } + + break; + case DLM_RCOM: + if (len < sizeof(struct dlm_rcom)) { + log_print("rcom msg too small: %d, will skip this message from node %d", + len, nodeid); + return -1; + } + + break; + default: + log_print("unsupported o_nextcmd received: %u, will skip this message from node %d", + p->opts.o_nextcmd, nodeid); + return -1; + } + + return 0; +} + +static void dlm_midcomms_receive_buffer_3_2(union dlm_packet *p, int nodeid) +{ + uint16_t msglen = le16_to_cpu(p->header.h_length); + struct midcomms_node *node; + uint32_t seq; + int ret, idx; + + idx = srcu_read_lock(&nodes_srcu); + node = dlm_midcomms_recv_node_lookup(nodeid, p, msglen, + dlm_midcomms_version_check_3_2); + if (!node) + goto out; + + switch (p->header.h_cmd) { + case DLM_RCOM: + /* these rcom message we use to determine version. + * they have their own retransmission handling and + * are the first messages of dlm. + * + * length already checked. + */ + switch (le32_to_cpu(p->rcom.rc_type)) { + case DLM_RCOM_NAMES: + fallthrough; + case DLM_RCOM_NAMES_REPLY: + fallthrough; + case DLM_RCOM_STATUS: + fallthrough; + case DLM_RCOM_STATUS_REPLY: + break; + default: + log_print("unsupported rcom type received: %u, will skip this message from node %d", + le32_to_cpu(p->rcom.rc_type), nodeid); + goto out; + } + + WARN_ON(test_bit(DLM_NODE_FLAG_STOP_RX, &node->flags)); + dlm_receive_buffer(p, nodeid); + break; + case DLM_OPTS: + seq = le32_to_cpu(p->header.u.h_seq); + + ret = dlm_opts_check_msglen(p, msglen, nodeid); + if (ret < 0) { + log_print("opts msg too small: %u, will skip this message from node %d", + msglen, nodeid); + goto out; + } + + p = (union dlm_packet *)((unsigned char *)p->opts.o_opts + + le16_to_cpu(p->opts.o_optlen)); + + /* recheck inner msglen just if it's not garbage */ + msglen = le16_to_cpu(p->header.h_length); + switch (p->header.h_cmd) { + case DLM_RCOM: + if (msglen < sizeof(struct dlm_rcom)) { + log_print("inner rcom msg too small: %u, will skip this message from node %d", + msglen, nodeid); + goto out; + } + + break; + case DLM_MSG: + if (msglen < sizeof(struct dlm_message)) { + log_print("inner msg too small: %u, will skip this message from node %d", + msglen, nodeid); + goto out; + } + + break; + case DLM_FIN: + if (msglen < sizeof(struct dlm_header)) { + log_print("inner fin too small: %u, will skip this message from node %d", + msglen, nodeid); + goto out; + } + + break; + default: + log_print("unsupported inner h_cmd received: %u, will skip this message from node %d", + msglen, nodeid); + goto out; + } + + dlm_midcomms_receive_buffer(p, node, seq); + break; + case DLM_ACK: + seq = le32_to_cpu(p->header.u.h_seq); + dlm_receive_ack(node, seq); + break; + default: + log_print("unsupported h_cmd received: %u, will skip this message from node %d", + p->header.h_cmd, nodeid); + break; + } + +out: + srcu_read_unlock(&nodes_srcu, idx); +} + +static int dlm_midcomms_version_check_3_1(struct midcomms_node *node) { - return dlm_lowcomms_close(nodeid); + switch (node->version) { + case DLM_VERSION_NOT_SET: + node->version = DLM_VERSION_3_1; + log_print("version 0x%08x for node %d detected", DLM_VERSION_3_1, + node->nodeid); + break; + case DLM_VERSION_3_1: + break; + default: + log_print_ratelimited("version mismatch detected, assumed 0x%08x but node %d has 0x%08x", + DLM_VERSION_3_1, node->nodeid, node->version); + return -1; + } + + return 0; +} + +static void dlm_midcomms_receive_buffer_3_1(union dlm_packet *p, int nodeid) +{ + uint16_t msglen = le16_to_cpu(p->header.h_length); + struct midcomms_node *node; + int idx; + + idx = srcu_read_lock(&nodes_srcu); + node = dlm_midcomms_recv_node_lookup(nodeid, p, msglen, + dlm_midcomms_version_check_3_1); + if (!node) { + srcu_read_unlock(&nodes_srcu, idx); + return; + } + srcu_read_unlock(&nodes_srcu, idx); + + switch (p->header.h_cmd) { + case DLM_RCOM: + /* length already checked */ + break; + case DLM_MSG: + if (msglen < sizeof(struct dlm_message)) { + log_print("msg too small: %u, will skip this message from node %d", + msglen, nodeid); + return; + } + + break; + default: + log_print("unsupported h_cmd received: %u, will skip this message from node %d", + p->header.h_cmd, nodeid); + return; + } + + dlm_receive_buffer(p, nodeid); } /* @@ -101,32 +865,19 @@ int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int len) if (msglen > len) break; - switch (hd->h_cmd) { - case DLM_MSG: - if (msglen < sizeof(struct dlm_message)) { - log_print("dlm msg too small: %u, will skip this message", - msglen); - goto skip; - } - + switch (le32_to_cpu(hd->h_version)) { + case DLM_VERSION_3_1: + dlm_midcomms_receive_buffer_3_1((union dlm_packet *)ptr, nodeid); break; - case DLM_RCOM: - if (msglen < sizeof(struct dlm_rcom)) { - log_print("dlm rcom msg too small: %u, will skip this message", - msglen); - goto skip; - } - + case DLM_VERSION_3_2: + dlm_midcomms_receive_buffer_3_2((union dlm_packet *)ptr, nodeid); break; default: - log_print("unsupported h_cmd received: %u, will skip this message", - hd->h_cmd); - goto skip; + log_print("received invalid version header: %u from node %d, will skip this message", + le32_to_cpu(hd->h_version), nodeid); + break; } - dlm_receive_buffer((union dlm_packet *)ptr, nodeid); - -skip: ret += msglen; len -= msglen; ptr += msglen; @@ -134,3 +885,454 @@ skip: return ret; } + +void dlm_midcomms_unack_msg_resend(int nodeid) +{ + struct midcomms_node *node; + struct dlm_mhandle *mh; + int idx, ret; + + idx = srcu_read_lock(&nodes_srcu); + node = nodeid2node(nodeid, 0); + if (!node) { + srcu_read_unlock(&nodes_srcu, idx); + return; + } + + /* old protocol, we don't support to retransmit on failure */ + switch (node->version) { + case DLM_VERSION_3_2: + break; + default: + srcu_read_unlock(&nodes_srcu, idx); + return; + } + + rcu_read_lock(); + list_for_each_entry_rcu(mh, &node->send_queue, list) { + if (!mh->committed) + continue; + + ret = dlm_lowcomms_resend_msg(mh->msg); + if (!ret) + log_print_ratelimited("retransmit dlm msg, seq %u, nodeid %d", + mh->seq, node->nodeid); + } + rcu_read_unlock(); + srcu_read_unlock(&nodes_srcu, idx); +} + +static void dlm_fill_opts_header(struct dlm_opts *opts, uint16_t inner_len, + uint32_t seq) +{ + opts->o_header.h_cmd = DLM_OPTS; + opts->o_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR); + opts->o_header.h_nodeid = dlm_our_nodeid(); + opts->o_header.h_length = DLM_MIDCOMMS_OPT_LEN + inner_len; + opts->o_header.u.h_seq = seq; + header_out(&opts->o_header); +} + +static void midcomms_new_msg_cb(struct dlm_mhandle *mh) +{ + atomic_inc(&mh->node->send_queue_cnt); + + spin_lock(&mh->node->send_queue_lock); + list_add_tail_rcu(&mh->list, &mh->node->send_queue); + spin_unlock(&mh->node->send_queue_lock); + + mh->seq = mh->node->seq_send++; +} + +static struct dlm_msg *dlm_midcomms_get_msg_3_2(struct dlm_mhandle *mh, int nodeid, + int len, gfp_t allocation, char **ppc) +{ + struct dlm_opts *opts; + struct dlm_msg *msg; + + msg = dlm_lowcomms_new_msg(nodeid, len + DLM_MIDCOMMS_OPT_LEN, + allocation, ppc, midcomms_new_msg_cb, mh); + if (!msg) + return NULL; + + opts = (struct dlm_opts *)*ppc; + mh->opts = opts; + + /* add possible options here */ + dlm_fill_opts_header(opts, len, mh->seq); + + *ppc += sizeof(*opts); + mh->inner_hd = (const struct dlm_header *)*ppc; + return msg; +} + +struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, + gfp_t allocation, char **ppc) +{ + struct midcomms_node *node; + struct dlm_mhandle *mh; + struct dlm_msg *msg; + int idx; + + idx = srcu_read_lock(&nodes_srcu); + node = nodeid2node(nodeid, 0); + if (!node) { + WARN_ON_ONCE(1); + goto err; + } + + /* this is a bug, however we going on and hope it will be resolved */ + WARN_ON(test_bit(DLM_NODE_FLAG_STOP_TX, &node->flags)); + + mh = kzalloc(sizeof(*mh), GFP_NOFS); + if (!mh) + goto err; + + mh->idx = idx; + mh->node = node; + + switch (node->version) { + case DLM_VERSION_3_1: + msg = dlm_lowcomms_new_msg(nodeid, len, allocation, ppc, + NULL, NULL); + if (!msg) { + kfree(mh); + goto err; + } + + break; + case DLM_VERSION_3_2: + msg = dlm_midcomms_get_msg_3_2(mh, nodeid, len, allocation, + ppc); + if (!msg) { + kfree(mh); + goto err; + } + + break; + default: + kfree(mh); + WARN_ON(1); + goto err; + } + + mh->msg = msg; + + /* keep in mind that is a must to call + * dlm_midcomms_commit_msg() which releases + * nodes_srcu using mh->idx which is assumed + * here that the application will call it. + */ + return mh; + +err: + srcu_read_unlock(&nodes_srcu, idx); + return NULL; +} + +static void dlm_midcomms_commit_msg_3_2(struct dlm_mhandle *mh) +{ + /* nexthdr chain for fast lookup */ + mh->opts->o_nextcmd = mh->inner_hd->h_cmd; + mh->committed = true; + dlm_lowcomms_commit_msg(mh->msg); +} + +void dlm_midcomms_commit_mhandle(struct dlm_mhandle *mh) +{ + switch (mh->node->version) { + case DLM_VERSION_3_1: + srcu_read_unlock(&nodes_srcu, mh->idx); + + dlm_lowcomms_commit_msg(mh->msg); + dlm_lowcomms_put_msg(mh->msg); + /* mh is not part of rcu list in this case */ + kfree(mh); + break; + case DLM_VERSION_3_2: + dlm_midcomms_commit_msg_3_2(mh); + srcu_read_unlock(&nodes_srcu, mh->idx); + break; + default: + srcu_read_unlock(&nodes_srcu, mh->idx); + WARN_ON(1); + break; + } +} + +int dlm_midcomms_start(void) +{ + int i; + + for (i = 0; i < CONN_HASH_SIZE; i++) + INIT_HLIST_HEAD(&node_hash[i]); + + return dlm_lowcomms_start(); +} + +static void dlm_act_fin_ack_rcv(struct midcomms_node *node) +{ + spin_lock(&node->state_lock); + pr_debug("receive active fin ack from node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + + switch (node->state) { + case DLM_FIN_WAIT1: + node->state = DLM_FIN_WAIT2; + pr_debug("switch node %d to state %s\n", + node->nodeid, dlm_state_str(node->state)); + break; + case DLM_CLOSING: + midcomms_node_reset(node); + pr_debug("switch node %d to state %s\n", + node->nodeid, dlm_state_str(node->state)); + wake_up(&node->shutdown_wait); + break; + case DLM_CLOSED: + /* not valid but somehow we got what we want */ + wake_up(&node->shutdown_wait); + break; + default: + spin_unlock(&node->state_lock); + log_print("%s: unexpected state: %d\n", + __func__, node->state); + WARN_ON(1); + return; + } + spin_unlock(&node->state_lock); +} + +void dlm_midcomms_add_member(int nodeid) +{ + struct midcomms_node *node; + int idx; + + if (nodeid == dlm_our_nodeid()) + return; + + idx = srcu_read_lock(&nodes_srcu); + node = nodeid2node(nodeid, GFP_NOFS); + if (!node) { + srcu_read_unlock(&nodes_srcu, idx); + return; + } + + spin_lock(&node->state_lock); + if (!node->users) { + pr_debug("receive add member from node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + switch (node->state) { + case DLM_ESTABLISHED: + break; + case DLM_CLOSED: + node->state = DLM_ESTABLISHED; + pr_debug("switch node %d to state %s\n", + node->nodeid, dlm_state_str(node->state)); + break; + default: + /* some invalid state passive shutdown + * was failed, we try to reset and + * hope it will go on. + */ + log_print("reset node %d because shutdown stucked", + node->nodeid); + + midcomms_node_reset(node); + node->state = DLM_ESTABLISHED; + break; + } + } + + node->users++; + pr_debug("users inc count %d\n", node->users); + spin_unlock(&node->state_lock); + + srcu_read_unlock(&nodes_srcu, idx); +} + +void dlm_midcomms_remove_member(int nodeid) +{ + struct midcomms_node *node; + int idx; + + if (nodeid == dlm_our_nodeid()) + return; + + idx = srcu_read_lock(&nodes_srcu); + node = nodeid2node(nodeid, 0); + if (!node) { + srcu_read_unlock(&nodes_srcu, idx); + return; + } + + spin_lock(&node->state_lock); + node->users--; + pr_debug("users dec count %d\n", node->users); + + /* hitting users count to zero means the + * other side is running dlm_midcomms_stop() + * we meet us to have a clean disconnect. + */ + if (node->users == 0) { + pr_debug("receive remove member from node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + switch (node->state) { + case DLM_ESTABLISHED: + break; + case DLM_CLOSE_WAIT: + /* passive shutdown DLM_LAST_ACK case 2 */ + node->state = DLM_LAST_ACK; + spin_unlock(&node->state_lock); + + pr_debug("switch node %d to state %s case 2\n", + node->nodeid, dlm_state_str(node->state)); + goto send_fin; + case DLM_LAST_ACK: + /* probably receive fin caught it, do nothing */ + break; + case DLM_CLOSED: + /* already gone, do nothing */ + break; + default: + log_print("%s: unexpected state: %d\n", + __func__, node->state); + break; + } + } + spin_unlock(&node->state_lock); + + srcu_read_unlock(&nodes_srcu, idx); + return; + +send_fin: + set_bit(DLM_NODE_FLAG_STOP_RX, &node->flags); + dlm_send_fin(node, dlm_pas_fin_ack_rcv); + srcu_read_unlock(&nodes_srcu, idx); +} + +static void midcomms_node_release(struct rcu_head *rcu) +{ + struct midcomms_node *node = container_of(rcu, struct midcomms_node, rcu); + + WARN_ON(atomic_read(&node->send_queue_cnt)); + kfree(node); +} + +static void midcomms_shutdown(struct midcomms_node *node) +{ + int ret; + + /* old protocol, we don't wait for pending operations */ + switch (node->version) { + case DLM_VERSION_3_2: + break; + default: + return; + } + + spin_lock(&node->state_lock); + pr_debug("receive active shutdown for node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + switch (node->state) { + case DLM_ESTABLISHED: + node->state = DLM_FIN_WAIT1; + pr_debug("switch node %d to state %s case 2\n", + node->nodeid, dlm_state_str(node->state)); + break; + case DLM_CLOSED: + /* we have what we want */ + spin_unlock(&node->state_lock); + return; + default: + /* busy to enter DLM_FIN_WAIT1, wait until passive + * done in shutdown_wait to enter DLM_CLOSED. + */ + break; + } + spin_unlock(&node->state_lock); + + if (node->state == DLM_FIN_WAIT1) { + dlm_send_fin(node, dlm_act_fin_ack_rcv); + + if (DLM_DEBUG_FENCE_TERMINATION) + msleep(5000); + } + + /* wait for other side dlm + fin */ + ret = wait_event_timeout(node->shutdown_wait, + node->state == DLM_CLOSED || + test_bit(DLM_NODE_FLAG_CLOSE, &node->flags), + DLM_SHUTDOWN_TIMEOUT); + if (!ret || test_bit(DLM_NODE_FLAG_CLOSE, &node->flags)) { + pr_debug("active shutdown timed out for node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); + midcomms_node_reset(node); + return; + } + + pr_debug("active shutdown done for node %d with state %s\n", + node->nodeid, dlm_state_str(node->state)); +} + +void dlm_midcomms_shutdown(void) +{ + struct midcomms_node *node; + int i, idx; + + mutex_lock(&close_lock); + idx = srcu_read_lock(&nodes_srcu); + for (i = 0; i < CONN_HASH_SIZE; i++) { + hlist_for_each_entry_rcu(node, &node_hash[i], hlist) { + midcomms_shutdown(node); + + spin_lock(&nodes_lock); + hlist_del_rcu(&node->hlist); + spin_unlock(&nodes_lock); + + call_srcu(&nodes_srcu, &node->rcu, midcomms_node_release); + } + } + srcu_read_unlock(&nodes_srcu, idx); + mutex_unlock(&close_lock); + + dlm_lowcomms_shutdown(); +} + +int dlm_midcomms_close(int nodeid) +{ + struct midcomms_node *node; + int idx, ret; + + if (nodeid == dlm_our_nodeid()) + return 0; + + idx = srcu_read_lock(&nodes_srcu); + /* Abort pending close/remove operation */ + node = nodeid2node(nodeid, 0); + if (node) { + /* let shutdown waiters leave */ + set_bit(DLM_NODE_FLAG_CLOSE, &node->flags); + wake_up(&node->shutdown_wait); + } + srcu_read_unlock(&nodes_srcu, idx); + + synchronize_srcu(&nodes_srcu); + + idx = srcu_read_lock(&nodes_srcu); + mutex_lock(&close_lock); + node = nodeid2node(nodeid, 0); + if (!node) { + mutex_unlock(&close_lock); + srcu_read_unlock(&nodes_srcu, idx); + return dlm_lowcomms_close(nodeid); + } + + ret = dlm_lowcomms_close(nodeid); + spin_lock(&node->state_lock); + midcomms_node_reset(node); + spin_unlock(&node->state_lock); + srcu_read_unlock(&nodes_srcu, idx); + mutex_unlock(&close_lock); + + return ret; +} diff --git a/fs/dlm/midcomms.h b/fs/dlm/midcomms.h index 9ac1190ce277..1178b836315b 100644 --- a/fs/dlm/midcomms.h +++ b/fs/dlm/midcomms.h @@ -21,6 +21,7 @@ int dlm_midcomms_start(void); void dlm_midcomms_shutdown(void); void dlm_midcomms_add_member(int nodeid); void dlm_midcomms_remove_member(int nodeid); +void dlm_midcomms_unack_msg_resend(int nodeid); #endif /* __MIDCOMMS_DOT_H__ */ diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c index 7c1a06bcd418..085f21966c72 100644 --- a/fs/dlm/rcom.c +++ b/fs/dlm/rcom.c @@ -202,7 +202,7 @@ retry: set_rcom_status(ls, (struct rcom_status *)rc->rc_buf, status_flags); allow_sync_reply(ls, &rc->rc_id); - memset(ls->ls_recover_buf, 0, LOWCOMMS_MAX_TX_BUFFER_LEN); + memset(ls->ls_recover_buf, 0, DEFAULT_BUFFER_SIZE); send_rcom_stateless(ls, msg, rc); @@ -325,7 +325,7 @@ retry: memcpy(rc->rc_buf, last_name, last_len); allow_sync_reply(ls, &rc->rc_id); - memset(ls->ls_recover_buf, 0, LOWCOMMS_MAX_TX_BUFFER_LEN); + memset(ls->ls_recover_buf, 0, DEFAULT_BUFFER_SIZE); send_rcom_stateless(ls, msg, rc); -- cgit From 5b2f981fde8b0dbf0bfa117bb4322342fcfb7174 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:47 -0400 Subject: fs: dlm: add midcomms debugfs functionality This patch adds functionality to debug midcomms per connection state inside a comms directory which is similar like dlm configfs. Currently there exists the possibility to read out two attributes which is the send queue counter and the version of each midcomms node state. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/debug_fs.c | 54 +++++++++++++++++++++++++++++++++++++++++++++++++++ fs/dlm/dlm_internal.h | 4 ++++ fs/dlm/midcomms.c | 27 ++++++++++++++++++++++++++ fs/dlm/midcomms.h | 6 ++++++ 4 files changed, 91 insertions(+) diff --git a/fs/dlm/debug_fs.c b/fs/dlm/debug_fs.c index d5bd990bcab8..47e9d57e4cae 100644 --- a/fs/dlm/debug_fs.c +++ b/fs/dlm/debug_fs.c @@ -16,6 +16,7 @@ #include #include "dlm_internal.h" +#include "midcomms.h" #include "lock.h" #define DLM_DEBUG_BUF_LEN 4096 @@ -23,6 +24,7 @@ static char debug_buf[DLM_DEBUG_BUF_LEN]; static struct mutex debug_buf_lock; static struct dentry *dlm_root; +static struct dentry *dlm_comms; static char *print_lockmode(int mode) { @@ -738,6 +740,57 @@ void dlm_delete_debug_file(struct dlm_ls *ls) debugfs_remove(ls->ls_debug_toss_dentry); } +static int dlm_state_show(struct seq_file *file, void *offset) +{ + seq_printf(file, "%s\n", dlm_midcomms_state(file->private)); + return 0; +} +DEFINE_SHOW_ATTRIBUTE(dlm_state); + +static int dlm_flags_show(struct seq_file *file, void *offset) +{ + seq_printf(file, "%lu\n", dlm_midcomms_flags(file->private)); + return 0; +} +DEFINE_SHOW_ATTRIBUTE(dlm_flags); + +static int dlm_send_queue_cnt_show(struct seq_file *file, void *offset) +{ + seq_printf(file, "%d\n", dlm_midcomms_send_queue_cnt(file->private)); + return 0; +} +DEFINE_SHOW_ATTRIBUTE(dlm_send_queue_cnt); + +static int dlm_version_show(struct seq_file *file, void *offset) +{ + seq_printf(file, "0x%08x\n", dlm_midcomms_version(file->private)); + return 0; +} +DEFINE_SHOW_ATTRIBUTE(dlm_version); + +void *dlm_create_debug_comms_file(int nodeid, void *data) +{ + struct dentry *d_node; + char name[256]; + + memset(name, 0, sizeof(name)); + snprintf(name, 256, "%d", nodeid); + + d_node = debugfs_create_dir(name, dlm_comms); + debugfs_create_file("state", 0444, d_node, data, &dlm_state_fops); + debugfs_create_file("flags", 0444, d_node, data, &dlm_flags_fops); + debugfs_create_file("send_queue_count", 0444, d_node, data, + &dlm_send_queue_cnt_fops); + debugfs_create_file("version", 0444, d_node, data, &dlm_version_fops); + + return d_node; +} + +void dlm_delete_debug_comms_file(void *ctx) +{ + debugfs_remove(ctx); +} + void dlm_create_debug_file(struct dlm_ls *ls) { char name[DLM_LOCKSPACE_LEN + 8]; @@ -797,6 +850,7 @@ void __init dlm_register_debugfs(void) { mutex_init(&debug_buf_lock); dlm_root = debugfs_create_dir("dlm", NULL); + dlm_comms = debugfs_create_dir("comms", dlm_root); } void dlm_unregister_debugfs(void) diff --git a/fs/dlm/dlm_internal.h b/fs/dlm/dlm_internal.h index 40917c878370..91d1ca3a121a 100644 --- a/fs/dlm/dlm_internal.h +++ b/fs/dlm/dlm_internal.h @@ -754,11 +754,15 @@ void dlm_register_debugfs(void); void dlm_unregister_debugfs(void); void dlm_create_debug_file(struct dlm_ls *ls); void dlm_delete_debug_file(struct dlm_ls *ls); +void *dlm_create_debug_comms_file(int nodeid, void *data); +void dlm_delete_debug_comms_file(void *ctx); #else static inline void dlm_register_debugfs(void) { } static inline void dlm_unregister_debugfs(void) { } static inline void dlm_create_debug_file(struct dlm_ls *ls) { } static inline void dlm_delete_debug_file(struct dlm_ls *ls) { } +static inline void *dlm_create_debug_comms_file(int nodeid, void *data) { return NULL; } +static inline void dlm_delete_debug_comms_file(void *ctx) { } #endif #endif /* __DLM_INTERNAL_DOT_H__ */ diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index eef3938a363e..35664950f6b7 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -189,6 +189,9 @@ struct midcomms_node { */ int users; + /* not protected by srcu, node_hash lifetime */ + void *debugfs; + struct hlist_node hlist; struct rcu_head rcu; }; @@ -244,6 +247,26 @@ static inline const char *dlm_state_str(int state) } } +const char *dlm_midcomms_state(struct midcomms_node *node) +{ + return dlm_state_str(node->state); +} + +unsigned long dlm_midcomms_flags(struct midcomms_node *node) +{ + return node->flags; +} + +int dlm_midcomms_send_queue_cnt(struct midcomms_node *node) +{ + return atomic_read(&node->send_queue_cnt); +} + +uint32_t dlm_midcomms_version(struct midcomms_node *node) +{ + return node->version; +} + static struct midcomms_node *__find_node(int nodeid, int r) { struct midcomms_node *node; @@ -332,6 +355,8 @@ static struct midcomms_node *nodeid2node(int nodeid, gfp_t alloc) hlist_add_head_rcu(&node->hlist, &node_hash[r]); spin_unlock(&nodes_lock); + + node->debugfs = dlm_create_debug_comms_file(nodeid, node); return node; } @@ -1285,6 +1310,8 @@ void dlm_midcomms_shutdown(void) hlist_for_each_entry_rcu(node, &node_hash[i], hlist) { midcomms_shutdown(node); + dlm_delete_debug_comms_file(node->debugfs); + spin_lock(&nodes_lock); hlist_del_rcu(&node->hlist); spin_unlock(&nodes_lock); diff --git a/fs/dlm/midcomms.h b/fs/dlm/midcomms.h index 1178b836315b..579abc6929be 100644 --- a/fs/dlm/midcomms.h +++ b/fs/dlm/midcomms.h @@ -12,6 +12,8 @@ #ifndef __MIDCOMMS_DOT_H__ #define __MIDCOMMS_DOT_H__ +struct midcomms_node; + int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int buflen); struct dlm_mhandle *dlm_midcomms_get_mhandle(int nodeid, int len, gfp_t allocation, char **ppc); @@ -22,6 +24,10 @@ void dlm_midcomms_shutdown(void); void dlm_midcomms_add_member(int nodeid); void dlm_midcomms_remove_member(int nodeid); void dlm_midcomms_unack_msg_resend(int nodeid); +const char *dlm_midcomms_state(struct midcomms_node *node); +unsigned long dlm_midcomms_flags(struct midcomms_node *node); +int dlm_midcomms_send_queue_cnt(struct midcomms_node *node); +uint32_t dlm_midcomms_version(struct midcomms_node *node); #endif /* __MIDCOMMS_DOT_H__ */ -- cgit From 706474fbc5fedd7799b488962aad3541b235165b Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 21 May 2021 15:08:48 -0400 Subject: fs: dlm: don't allow half transmitted messages This patch will clean a dirty page buffer if a reconnect occurs. If a page buffer was half transmitted we cannot start inside the middle of a dlm message if a node connects again. I observed invalid length receptions errors and was guessing that this behaviour occurs, after this patch I never saw an invalid message length again. This patch might drops more messages for dlm version 3.1 but 3.1 can't deal with half messages as well, for 3.2 it might trigger more re-transmissions but will not leave dlm in a broken state. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 95 +++++++++++++++++++++++++++++++++++-------------------- 1 file changed, 60 insertions(+), 35 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index fe9113bd5ba0..36adccc4f849 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -118,6 +118,7 @@ struct writequeue_entry { int len; int end; int users; + bool dirty; struct connection *con; struct list_head msgs; struct kref ref; @@ -700,6 +701,42 @@ static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port, memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len); } +static void dlm_page_release(struct kref *kref) +{ + struct writequeue_entry *e = container_of(kref, struct writequeue_entry, + ref); + + __free_page(e->page); + kfree(e); +} + +static void dlm_msg_release(struct kref *kref) +{ + struct dlm_msg *msg = container_of(kref, struct dlm_msg, ref); + + kref_put(&msg->entry->ref, dlm_page_release); + kfree(msg); +} + +static void free_entry(struct writequeue_entry *e) +{ + struct dlm_msg *msg, *tmp; + + list_for_each_entry_safe(msg, tmp, &e->msgs, list) { + if (msg->orig_msg) { + msg->orig_msg->retransmit = false; + kref_put(&msg->orig_msg->ref, dlm_msg_release); + } + + list_del(&msg->list); + kref_put(&msg->ref, dlm_msg_release); + } + + list_del(&e->list); + atomic_dec(&e->con->writequeue_cnt); + kref_put(&e->ref, dlm_page_release); +} + static void dlm_close_sock(struct socket **sock) { if (*sock) { @@ -714,6 +751,7 @@ static void close_connection(struct connection *con, bool and_other, bool tx, bool rx) { bool closing = test_and_set_bit(CF_CLOSING, &con->flags); + struct writequeue_entry *e; if (tx && !closing && cancel_work_sync(&con->swork)) { log_print("canceled swork for node %d", con->nodeid); @@ -732,6 +770,26 @@ static void close_connection(struct connection *con, bool and_other, close_connection(con->othercon, false, tx, rx); } + /* if we send a writequeue entry only a half way, we drop the + * whole entry because reconnection and that we not start of the + * middle of a msg which will confuse the other end. + * + * we can always drop messages because retransmits, but what we + * cannot allow is to transmit half messages which may be processed + * at the other side. + * + * our policy is to start on a clean state when disconnects, we don't + * know what's send/received on transport layer in this case. + */ + spin_lock(&con->writequeue_lock); + if (!list_empty(&con->writequeue)) { + e = list_first_entry(&con->writequeue, struct writequeue_entry, + list); + if (e->dirty) + free_entry(e); + } + spin_unlock(&con->writequeue_lock); + con->rx_leftover = 0; con->retries = 0; clear_bit(CF_CONNECTED, &con->flags); @@ -1026,41 +1084,6 @@ accept_err: return result; } -static void dlm_page_release(struct kref *kref) -{ - struct writequeue_entry *e = container_of(kref, struct writequeue_entry, - ref); - - __free_page(e->page); - kfree(e); -} - -static void dlm_msg_release(struct kref *kref) -{ - struct dlm_msg *msg = container_of(kref, struct dlm_msg, ref); - - kref_put(&msg->entry->ref, dlm_page_release); - kfree(msg); -} - -static void free_entry(struct writequeue_entry *e) -{ - struct dlm_msg *msg, *tmp; - - list_for_each_entry_safe(msg, tmp, &e->msgs, list) { - if (msg->orig_msg) { - msg->orig_msg->retransmit = false; - kref_put(&msg->orig_msg->ref, dlm_msg_release); - } - list_del(&msg->list); - kref_put(&msg->ref, dlm_msg_release); - } - - list_del(&e->list); - atomic_dec(&e->con->writequeue_cnt); - kref_put(&e->ref, dlm_page_release); -} - /* * writequeue_entry_complete - try to delete and free write queue entry * @e: write queue entry to try to delete @@ -1072,6 +1095,8 @@ static void writequeue_entry_complete(struct writequeue_entry *e, int completed) { e->offset += completed; e->len -= completed; + /* signal that page was half way transmitted */ + e->dirty = true; if (e->len == 0 && e->users == 0) free_entry(e); -- cgit From f6089981d07e6e1cc053f4c239e458eed122c092 Mon Sep 17 00:00:00 2001 From: Colin Ian King Date: Wed, 26 May 2021 15:46:05 +0100 Subject: fs: dlm: Fix memory leak of object mh There is an error return path that is not kfree'ing mh after it has been successfully allocates. Fix this by moving the call to create_rcom to after the check on rc_in->rc_id check to avoid this. Thanks to Alexander Ahring Oder Aring for suggesting the correct way to fix this. Addresses-Coverity: ("Resource leak") Fixes: a070a91cf140 ("fs: dlm: add more midcomms hooks") Signed-off-by: Colin Ian King Signed-off-by: David Teigland --- fs/dlm/rcom.c | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c index 085f21966c72..a7727b9e5e83 100644 --- a/fs/dlm/rcom.c +++ b/fs/dlm/rcom.c @@ -385,10 +385,6 @@ static void receive_rcom_lookup(struct dlm_ls *ls, struct dlm_rcom *rc_in) int error, ret_nodeid, nodeid = rc_in->rc_header.h_nodeid; int len = rc_in->rc_header.h_length - sizeof(struct dlm_rcom); - error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh); - if (error) - return; - /* Old code would send this special id to trigger a debug dump. */ if (rc_in->rc_id == 0xFFFFFFFF) { log_error(ls, "receive_rcom_lookup dump from %d", nodeid); @@ -396,6 +392,10 @@ static void receive_rcom_lookup(struct dlm_ls *ls, struct dlm_rcom *rc_in) return; } + error = create_rcom(ls, nodeid, DLM_RCOM_LOOKUP_REPLY, 0, &rc, &mh); + if (error) + return; + error = dlm_master_lookup(ls, nodeid, rc_in->rc_buf, len, DLM_LU_RECOVER_MASTER, &ret_nodeid, NULL); if (error) -- cgit From 7d3848c03e09ea9cdfde8bb2b82282d252943ee6 Mon Sep 17 00:00:00 2001 From: Colin Ian King Date: Wed, 26 May 2021 09:53:39 +0100 Subject: fs: dlm: Fix spelling mistake "stucked" -> "stuck" There are spelling mistake in log messages. Fix these. Signed-off-by: Colin Ian King Signed-off-by: David Teigland --- fs/dlm/midcomms.c | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index 35664950f6b7..4e36e418b6bf 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -591,7 +591,7 @@ dlm_midcomms_recv_node_lookup(int nodeid, const union dlm_packet *p, * was failed, we try to reset and * hope it will go on. */ - log_print("reset node %d because shutdown stucked", + log_print("reset node %d because shutdown stuck", node->nodeid); midcomms_node_reset(node); @@ -1159,7 +1159,7 @@ void dlm_midcomms_add_member(int nodeid) * was failed, we try to reset and * hope it will go on. */ - log_print("reset node %d because shutdown stucked", + log_print("reset node %d because shutdown stuck", node->nodeid); midcomms_node_reset(node); -- cgit From fcef0e6c27ce109d2c617aa12f0bfd9f7ff47d38 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Wed, 2 Jun 2021 09:45:15 -0400 Subject: fs: dlm: fix lowcomms_start error case This patch fixes the error path handling in lowcomms_start(). We need to cleanup some static allocated data structure and cleanup possible workqueue if these have started. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 15 ++++++++++++--- 1 file changed, 12 insertions(+), 3 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 36adccc4f849..b71f7eafb808 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -1803,10 +1803,15 @@ static void process_send_sockets(struct work_struct *work) static void work_stop(void) { - if (recv_workqueue) + if (recv_workqueue) { destroy_workqueue(recv_workqueue); - if (send_workqueue) + recv_workqueue = NULL; + } + + if (send_workqueue) { destroy_workqueue(send_workqueue); + send_workqueue = NULL; + } } static int work_start(void) @@ -1823,6 +1828,7 @@ static int work_start(void) if (!send_workqueue) { log_print("can't start dlm_send"); destroy_workqueue(recv_workqueue); + recv_workqueue = NULL; return -ENOMEM; } @@ -1960,7 +1966,7 @@ int dlm_lowcomms_start(void) error = work_start(); if (error) - goto fail; + goto fail_local; dlm_allow_conn = 1; @@ -1977,6 +1983,9 @@ int dlm_lowcomms_start(void) fail_unlisten: dlm_allow_conn = 0; dlm_close_sock(&listen_con.sock); + work_stop(); +fail_local: + deinit_local(); fail: return error; } -- cgit From 700ab1c363c7b54c9ea3222379b33fc00ab02f7b Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Wed, 2 Jun 2021 09:45:16 -0400 Subject: fs: dlm: fix memory leak when fenced I got some kmemleak report when a node was fenced. The user space tool dlm_controld will therefore run some rmdir() in dlm configfs which was triggering some memleaks. This patch stores the sps and cms attributes which stores some handling for subdirectories of the configfs cluster entry and free them if they get released as the parent directory gets freed. unreferenced object 0xffff88810d9e3e00 (size 192): comm "dlm_controld", pid 342, jiffies 4294698126 (age 55438.801s) hex dump (first 32 bytes): 00 00 00 00 00 00 00 00 73 70 61 63 65 73 00 00 ........spaces.. 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................ backtrace: [<00000000db8b640b>] make_cluster+0x5d/0x360 [<000000006a571db4>] configfs_mkdir+0x274/0x730 [<00000000b094501c>] vfs_mkdir+0x27e/0x340 [<0000000058b0adaf>] do_mkdirat+0xff/0x1b0 [<00000000d1ffd156>] do_syscall_64+0x40/0x80 [<00000000ab1408c8>] entry_SYSCALL_64_after_hwframe+0x44/0xae unreferenced object 0xffff88810d9e3a00 (size 192): comm "dlm_controld", pid 342, jiffies 4294698126 (age 55438.801s) hex dump (first 32 bytes): 00 00 00 00 00 00 00 00 63 6f 6d 6d 73 00 00 00 ........comms... 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 00 ................ backtrace: [<00000000a7ef6ad2>] make_cluster+0x82/0x360 [<000000006a571db4>] configfs_mkdir+0x274/0x730 [<00000000b094501c>] vfs_mkdir+0x27e/0x340 [<0000000058b0adaf>] do_mkdirat+0xff/0x1b0 [<00000000d1ffd156>] do_syscall_64+0x40/0x80 [<00000000ab1408c8>] entry_SYSCALL_64_after_hwframe+0x44/0xae Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/config.c | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/fs/dlm/config.c b/fs/dlm/config.c index 01ae294743e9..db717a879537 100644 --- a/fs/dlm/config.c +++ b/fs/dlm/config.c @@ -80,6 +80,9 @@ struct dlm_cluster { unsigned int cl_new_rsb_count; unsigned int cl_recover_callbacks; char cl_cluster_name[DLM_LOCKSPACE_LEN]; + + struct dlm_spaces *sps; + struct dlm_comms *cms; }; static struct dlm_cluster *config_item_to_cluster(struct config_item *i) @@ -410,6 +413,9 @@ static struct config_group *make_cluster(struct config_group *g, if (!cl || !sps || !cms) goto fail; + cl->sps = sps; + cl->cms = cms; + config_group_init_type_name(&cl->group, name, &cluster_type); config_group_init_type_name(&sps->ss_group, "spaces", &spaces_type); config_group_init_type_name(&cms->cs_group, "comms", &comms_type); @@ -459,6 +465,9 @@ static void drop_cluster(struct config_group *g, struct config_item *i) static void release_cluster(struct config_item *i) { struct dlm_cluster *cl = config_item_to_cluster(i); + + kfree(cl->sps); + kfree(cl->cms); kfree(cl); } -- cgit From 6c6a1cc666956cbb3ac6db79ed401ee027e6f950 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Wed, 2 Jun 2021 09:45:17 -0400 Subject: fs: dlm: use alloc_ordered_workqueue The proper way to allocate ordered workqueues is to use alloc_ordered_workqueue() function. The current way implies an ordered workqueue which is also required by dlm. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index b71f7eafb808..02b636d113fb 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -1816,15 +1816,13 @@ static void work_stop(void) static int work_start(void) { - recv_workqueue = alloc_workqueue("dlm_recv", - WQ_UNBOUND | WQ_MEM_RECLAIM, 1); + recv_workqueue = alloc_ordered_workqueue("dlm_recv", WQ_MEM_RECLAIM); if (!recv_workqueue) { log_print("can't start dlm_recv"); return -ENOMEM; } - send_workqueue = alloc_workqueue("dlm_send", - WQ_UNBOUND | WQ_MEM_RECLAIM, 1); + send_workqueue = alloc_ordered_workqueue("dlm_send", WQ_MEM_RECLAIM); if (!send_workqueue) { log_print("can't start dlm_send"); destroy_workqueue(recv_workqueue); -- cgit From 9a4139a79403161f190cf30be7d89ac877ae3b12 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Wed, 2 Jun 2021 09:45:18 -0400 Subject: fs: dlm: move dlm allow conn This patch checks if possible allowing new connections is allowed before queueing the listen socket to accept new connections. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/lowcomms.c | 7 +++---- 1 file changed, 3 insertions(+), 4 deletions(-) diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 02b636d113fb..6b150e3aa30c 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -471,6 +471,9 @@ static void lowcomms_data_ready(struct sock *sk) static void lowcomms_listen_data_ready(struct sock *sk) { + if (!dlm_allow_conn) + return; + queue_work(recv_workqueue, &listen_con.rwork); } @@ -969,10 +972,6 @@ static int accept_from_sock(struct listen_connection *con) struct connection *addcon; unsigned int mark; - if (!dlm_allow_conn) { - return -1; - } - if (!con->sock) return -ENOTCONN; -- cgit From ac7d5d036dc93710971f532ed57f9a6858a2b262 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Wed, 2 Jun 2021 09:45:19 -0400 Subject: fs: dlm: introduce proto values Currently the dlm protocol values are that TCP is 0 and everything else is SCTP. This makes it difficult to introduce possible other transport layers. The only one user space tool dlm_controld, which I am aware of, handles the protocol value 1 for SCTP. We change it now to handle SCTP as 1, this will break user space API but it will fix it so we can add possible other transport layers. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/config.c | 2 +- fs/dlm/config.h | 3 +++ fs/dlm/lowcomms.c | 23 +++++++++++++++++++---- 3 files changed, 23 insertions(+), 5 deletions(-) diff --git a/fs/dlm/config.c b/fs/dlm/config.c index db717a879537..c91c1c73ed9d 100644 --- a/fs/dlm/config.c +++ b/fs/dlm/config.c @@ -952,7 +952,7 @@ int dlm_our_addr(struct sockaddr_storage *addr, int num) #define DEFAULT_SCAN_SECS 5 #define DEFAULT_LOG_DEBUG 0 #define DEFAULT_LOG_INFO 1 -#define DEFAULT_PROTOCOL 0 +#define DEFAULT_PROTOCOL DLM_PROTO_TCP #define DEFAULT_MARK 0 #define DEFAULT_TIMEWARN_CS 500 /* 5 sec = 500 centiseconds */ #define DEFAULT_WAITWARN_US 0 diff --git a/fs/dlm/config.h b/fs/dlm/config.h index d2cd4bd20313..00374b45c748 100644 --- a/fs/dlm/config.h +++ b/fs/dlm/config.h @@ -23,6 +23,9 @@ struct dlm_config_node { #define DLM_MAX_ADDR_COUNT 3 +#define DLM_PROTO_TCP 0 +#define DLM_PROTO_SCTP 1 + struct dlm_config_info { int ci_tcp_port; int ci_buffer_size; diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index 6b150e3aa30c..f2a3b0401b9c 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -208,12 +208,18 @@ static int dlm_con_init(struct connection *con, int nodeid) INIT_WORK(&con->rwork, process_recv_sockets); init_waitqueue_head(&con->shutdown_wait); - if (dlm_config.ci_protocol == 0) { + switch (dlm_config.ci_protocol) { + case DLM_PROTO_TCP: con->connect_action = tcp_connect_to_sock; con->shutdown_action = dlm_tcp_shutdown; con->eof_condition = tcp_eof_condition; - } else { + break; + case DLM_PROTO_SCTP: con->connect_action = sctp_connect_to_sock; + break; + default: + kfree(con->rx_buf); + return -EINVAL; } return 0; @@ -1968,10 +1974,19 @@ int dlm_lowcomms_start(void) dlm_allow_conn = 1; /* Start listening */ - if (dlm_config.ci_protocol == 0) + switch (dlm_config.ci_protocol) { + case DLM_PROTO_TCP: error = tcp_listen_for_all(); - else + break; + case DLM_PROTO_SCTP: error = sctp_listen_for_all(&listen_con); + break; + default: + log_print("Invalid protocol identifier %d set", + dlm_config.ci_protocol); + error = -EINVAL; + break; + } if (error) goto fail_unlisten; -- cgit From d10a0b88751a0954c14e11fd988da00d3b0d5445 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Wed, 2 Jun 2021 09:45:20 -0400 Subject: fs: dlm: rename socket and app buffer defines This patch renames DEFAULT_BUFFER_SIZE to DLM_MAX_SOCKET_BUFSIZE and LOWCOMMS_MAX_TX_BUFFER_LEN to DLM_MAX_APP_BUFSIZE as they are proper names to define what's behind those values. The DLM_MAX_SOCKET_BUFSIZE defines the maximum size of buffer which can be handled on socket layer, the DLM_MAX_APP_BUFSIZE defines the maximum size of buffer which can be handled by the DLM application layer. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/config.c | 4 ++-- fs/dlm/config.h | 2 +- fs/dlm/lockspace.c | 2 +- fs/dlm/lowcomms.c | 4 ++-- fs/dlm/lowcomms.h | 2 +- fs/dlm/member.c | 2 +- fs/dlm/midcomms.c | 4 ++-- fs/dlm/rcom.c | 6 +++--- 8 files changed, 13 insertions(+), 13 deletions(-) diff --git a/fs/dlm/config.c b/fs/dlm/config.c index c91c1c73ed9d..42eee2783756 100644 --- a/fs/dlm/config.c +++ b/fs/dlm/config.c @@ -208,7 +208,7 @@ static int dlm_check_zero(unsigned int x) static int dlm_check_buffer_size(unsigned int x) { - if (x < DEFAULT_BUFFER_SIZE) + if (x < DLM_MAX_SOCKET_BUFSIZE) return -EINVAL; return 0; @@ -962,7 +962,7 @@ int dlm_our_addr(struct sockaddr_storage *addr, int num) struct dlm_config_info dlm_config = { .ci_tcp_port = DEFAULT_TCP_PORT, - .ci_buffer_size = DEFAULT_BUFFER_SIZE, + .ci_buffer_size = DLM_MAX_SOCKET_BUFSIZE, .ci_rsbtbl_size = DEFAULT_RSBTBL_SIZE, .ci_recover_timer = DEFAULT_RECOVER_TIMER, .ci_toss_secs = DEFAULT_TOSS_SECS, diff --git a/fs/dlm/config.h b/fs/dlm/config.h index 00374b45c748..df92b0a07fc6 100644 --- a/fs/dlm/config.h +++ b/fs/dlm/config.h @@ -12,7 +12,7 @@ #ifndef __CONFIG_DOT_H__ #define __CONFIG_DOT_H__ -#define DEFAULT_BUFFER_SIZE 4096 +#define DLM_MAX_SOCKET_BUFSIZE 4096 struct dlm_config_node { int nodeid; diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c index 2b738be8d7e4..d71aba8c3e64 100644 --- a/fs/dlm/lockspace.c +++ b/fs/dlm/lockspace.c @@ -572,7 +572,7 @@ static int new_lockspace(const char *name, const char *cluster, * not having out of bounds issues. However on sending side 3.2 * might send less. */ - ls->ls_recover_buf = kmalloc(DEFAULT_BUFFER_SIZE, GFP_NOFS); + ls->ls_recover_buf = kmalloc(DLM_MAX_SOCKET_BUFSIZE, GFP_NOFS); if (!ls->ls_recover_buf) goto out_lkbidr; diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c index f2a3b0401b9c..0ea9ae35da0b 100644 --- a/fs/dlm/lowcomms.c +++ b/fs/dlm/lowcomms.c @@ -1556,9 +1556,9 @@ struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, struct dlm_msg *msg; int idx; - if (len > DEFAULT_BUFFER_SIZE || + if (len > DLM_MAX_SOCKET_BUFSIZE || len < sizeof(struct dlm_header)) { - BUILD_BUG_ON(PAGE_SIZE < DEFAULT_BUFFER_SIZE); + BUILD_BUG_ON(PAGE_SIZE < DLM_MAX_SOCKET_BUFSIZE); log_print("failed to allocate a buffer of size %d", len); WARN_ON(1); return NULL; diff --git a/fs/dlm/lowcomms.h b/fs/dlm/lowcomms.h index 730c34317183..aaae7115c00d 100644 --- a/fs/dlm/lowcomms.h +++ b/fs/dlm/lowcomms.h @@ -15,7 +15,7 @@ #include "dlm_internal.h" #define DLM_MIDCOMMS_OPT_LEN sizeof(struct dlm_opts) -#define LOWCOMMS_MAX_TX_BUFFER_LEN (DEFAULT_BUFFER_SIZE - \ +#define DLM_MAX_APP_BUFSIZE (DLM_MAX_SOCKET_BUFSIZE - \ DLM_MIDCOMMS_OPT_LEN) #define CONN_HASH_SIZE 32 diff --git a/fs/dlm/member.c b/fs/dlm/member.c index 63971c594bdc..d9e1e4170eb1 100644 --- a/fs/dlm/member.c +++ b/fs/dlm/member.c @@ -271,7 +271,7 @@ int dlm_slots_assign(struct dlm_ls *ls, int *num_slots, int *slots_size, log_slots(ls, gen, num, NULL, array, array_size); - max_slots = (LOWCOMMS_MAX_TX_BUFFER_LEN - sizeof(struct dlm_rcom) - + max_slots = (DLM_MAX_APP_BUFSIZE - sizeof(struct dlm_rcom) - sizeof(struct rcom_config)) / sizeof(struct rcom_slot); if (num > max_slots) { diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index 4e36e418b6bf..7d217234b697 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -865,7 +865,7 @@ int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int len) while (len >= sizeof(struct dlm_header)) { hd = (struct dlm_header *)ptr; - /* no message should be more than DEFAULT_BUFFER_SIZE or + /* no message should be more than DLM_MAX_SOCKET_BUFSIZE or * less than dlm_header size. * * Some messages does not have a 8 byte length boundary yet @@ -877,7 +877,7 @@ int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int len) * the next major version bump. */ msglen = le16_to_cpu(hd->h_length); - if (msglen > DEFAULT_BUFFER_SIZE || + if (msglen > DLM_MAX_SOCKET_BUFSIZE || msglen < sizeof(struct dlm_header)) { log_print("received invalid length header: %u from node %d, will abort message parsing", msglen, nodeid); diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c index a7727b9e5e83..5651933f54a4 100644 --- a/fs/dlm/rcom.c +++ b/fs/dlm/rcom.c @@ -202,7 +202,7 @@ retry: set_rcom_status(ls, (struct rcom_status *)rc->rc_buf, status_flags); allow_sync_reply(ls, &rc->rc_id); - memset(ls->ls_recover_buf, 0, DEFAULT_BUFFER_SIZE); + memset(ls->ls_recover_buf, 0, DLM_MAX_SOCKET_BUFSIZE); send_rcom_stateless(ls, msg, rc); @@ -325,7 +325,7 @@ retry: memcpy(rc->rc_buf, last_name, last_len); allow_sync_reply(ls, &rc->rc_id); - memset(ls->ls_recover_buf, 0, DEFAULT_BUFFER_SIZE); + memset(ls->ls_recover_buf, 0, DLM_MAX_SOCKET_BUFSIZE); send_rcom_stateless(ls, msg, rc); @@ -345,7 +345,7 @@ static void receive_rcom_names(struct dlm_ls *ls, struct dlm_rcom *rc_in) nodeid = rc_in->rc_header.h_nodeid; inlen = rc_in->rc_header.h_length - sizeof(struct dlm_rcom); - outlen = LOWCOMMS_MAX_TX_BUFFER_LEN - sizeof(struct dlm_rcom); + outlen = DLM_MAX_APP_BUFSIZE - sizeof(struct dlm_rcom); error = create_rcom_stateless(ls, nodeid, DLM_RCOM_NAMES_REPLY, outlen, &rc, &msg); -- cgit From f5fe8d5107ad68279528f39ceae64ab0d68deb3c Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 11 Jun 2021 12:55:40 -0400 Subject: fs: dlm: fix race in mhandle deletion This patch fixes a race between mhandle deletion in case of receiving an acknowledge and flush of all pending mhandle in cases of an timeout or resetting node states. Fixes: 489d8e559c65 ("fs: dlm: add reliable connection if reconnect") Reported-by: Guillaume Nault Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/midcomms.c | 35 +++++++++++++++++++++-------------- 1 file changed, 21 insertions(+), 14 deletions(-) diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index 7d217234b697..92f95ee7003a 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -287,6 +287,14 @@ static void dlm_mhandle_release(struct rcu_head *rcu) kfree(mh); } +static void dlm_mhandle_delete(struct midcomms_node *node, + struct dlm_mhandle *mh) +{ + list_del_rcu(&mh->list); + atomic_dec(&node->send_queue_cnt); + call_rcu(&mh->rcu, dlm_mhandle_release); +} + static void dlm_send_queue_flush(struct midcomms_node *node) { struct dlm_mhandle *mh; @@ -294,15 +302,11 @@ static void dlm_send_queue_flush(struct midcomms_node *node) pr_debug("flush midcomms send queue of node %d\n", node->nodeid); rcu_read_lock(); + spin_lock(&node->send_queue_lock); list_for_each_entry_rcu(mh, &node->send_queue, list) { - spin_lock(&node->send_queue_lock); - list_del_rcu(&mh->list); - spin_unlock(&node->send_queue_lock); - - atomic_dec(&node->send_queue_cnt); - - call_rcu(&mh->rcu, dlm_mhandle_release); + dlm_mhandle_delete(node, mh); } + spin_unlock(&node->send_queue_lock); rcu_read_unlock(); } @@ -424,21 +428,24 @@ static void dlm_receive_ack(struct midcomms_node *node, uint32_t seq) rcu_read_lock(); list_for_each_entry_rcu(mh, &node->send_queue, list) { if (before(mh->seq, seq)) { - spin_lock(&node->send_queue_lock); - list_del_rcu(&mh->list); - spin_unlock(&node->send_queue_lock); - - atomic_dec(&node->send_queue_cnt); - if (mh->ack_rcv) mh->ack_rcv(node); + } else { + /* send queue should be ordered */ + break; + } + } - call_rcu(&mh->rcu, dlm_mhandle_release); + spin_lock(&node->send_queue_lock); + list_for_each_entry_rcu(mh, &node->send_queue, list) { + if (before(mh->seq, seq)) { + dlm_mhandle_delete(node, mh); } else { /* send queue should be ordered */ break; } } + spin_unlock(&node->send_queue_lock); rcu_read_unlock(); } -- cgit From 957adb68b3f7df8421a05f1647d3027f2acad310 Mon Sep 17 00:00:00 2001 From: Alexander Aring Date: Fri, 11 Jun 2021 12:55:41 -0400 Subject: fs: dlm: invalid buffer access in lookup error This patch will evaluate the message length if a dlm opts header can fit in before accessing it if a node lookup fails. The invalid sequence error means that the version detection failed and an unexpected message arrived. For debugging such situation the type of arrived message is important to know. Signed-off-by: Alexander Aring Signed-off-by: David Teigland --- fs/dlm/midcomms.c | 19 +++++++++++++++++-- 1 file changed, 17 insertions(+), 2 deletions(-) diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c index 92f95ee7003a..e3de268898ed 100644 --- a/fs/dlm/midcomms.c +++ b/fs/dlm/midcomms.c @@ -621,8 +621,23 @@ dlm_midcomms_recv_node_lookup(int nodeid, const union dlm_packet *p, node = nodeid2node(nodeid, allocation); if (!node) { - log_print_ratelimited("received dlm message cmd %d nextcmd %d from node %d in an invalid sequence", - p->header.h_cmd, p->opts.o_nextcmd, nodeid); + switch (p->header.h_cmd) { + case DLM_OPTS: + if (msglen < sizeof(struct dlm_opts)) { + log_print("opts msg too small: %u, will skip this message from node %d", + msglen, nodeid); + return NULL; + } + + log_print_ratelimited("received dlm opts message nextcmd %d from node %d in an invalid sequence", + p->opts.o_nextcmd, nodeid); + break; + default: + log_print_ratelimited("received dlm message cmd %d from node %d in an invalid sequence", + p->header.h_cmd, nodeid); + break; + } + return NULL; } -- cgit