summaryrefslogtreecommitdiff
path: root/fs
diff options
context:
space:
mode:
authorLinus Torvalds <torvalds@linux-foundation.org>2021-04-30 15:28:25 -0700
committerLinus Torvalds <torvalds@linux-foundation.org>2021-04-30 15:28:25 -0700
commit6bab076a3d2ff4298483c1a3b77162811b451863 (patch)
tree142cdef533b761343e36b3f263d069d6187ce293 /fs
parent9ec1efbf9ded6cf38fd910c6fe943972d970f384 (diff)
parent2fd8db2dd05d895961c7c7b9fa02d72f385560e4 (diff)
Merge tag 'dlm-5.13' of git://git.kernel.org/pub/scm/linux/kernel/git/teigland/linux-dlm
Pull dlm updates from David Teigland: "This includes more dlm networking cleanups and improvements for making dlm shutdowns more robust" * tag 'dlm-5.13' of git://git.kernel.org/pub/scm/linux/kernel/git/teigland/linux-dlm: fs: dlm: fix missing unlock on error in accept_from_sock() fs: dlm: add shutdown hook fs: dlm: flush swork on shutdown fs: dlm: remove unaligned memory access handling fs: dlm: check on minimum msglen size fs: dlm: simplify writequeue handling fs: dlm: use GFP_ZERO for page buffer fs: dlm: change allocation limits fs: dlm: add check if dlm is currently running fs: dlm: add errno handling to check callback fs: dlm: set subclass for othercon sock_mutex fs: dlm: set connected bit after accept fs: dlm: fix mark setting deadlock fs: dlm: fix debugfs dump
Diffstat (limited to 'fs')
-rw-r--r--fs/dlm/config.c86
-rw-r--r--fs/dlm/config.h1
-rw-r--r--fs/dlm/debug_fs.c1
-rw-r--r--fs/dlm/lock.c2
-rw-r--r--fs/dlm/lockspace.c20
-rw-r--r--fs/dlm/lowcomms.c194
-rw-r--r--fs/dlm/lowcomms.h5
-rw-r--r--fs/dlm/midcomms.c33
-rw-r--r--fs/dlm/rcom.c2
9 files changed, 202 insertions, 142 deletions
diff --git a/fs/dlm/config.c b/fs/dlm/config.c
index 49c5f9407098..88d95d96e36c 100644
--- a/fs/dlm/config.c
+++ b/fs/dlm/config.c
@@ -125,7 +125,7 @@ static ssize_t cluster_cluster_name_store(struct config_item *item,
CONFIGFS_ATTR(cluster_, cluster_name);
static ssize_t cluster_set(struct dlm_cluster *cl, unsigned int *cl_field,
- int *info_field, bool (*check_cb)(unsigned int x),
+ int *info_field, int (*check_cb)(unsigned int x),
const char *buf, size_t len)
{
unsigned int x;
@@ -137,8 +137,11 @@ static ssize_t cluster_set(struct dlm_cluster *cl, unsigned int *cl_field,
if (rc)
return rc;
- if (check_cb && check_cb(x))
- return -EINVAL;
+ if (check_cb) {
+ rc = check_cb(x);
+ if (rc)
+ return rc;
+ }
*cl_field = x;
*info_field = x;
@@ -161,17 +164,53 @@ static ssize_t cluster_##name##_show(struct config_item *item, char *buf) \
} \
CONFIGFS_ATTR(cluster_, name);
-static bool dlm_check_zero(unsigned int x)
+static int dlm_check_protocol_and_dlm_running(unsigned int x)
+{
+ switch (x) {
+ case 0:
+ /* TCP */
+ break;
+ case 1:
+ /* SCTP */
+ break;
+ default:
+ return -EINVAL;
+ }
+
+ if (dlm_allow_conn)
+ return -EBUSY;
+
+ return 0;
+}
+
+static int dlm_check_zero_and_dlm_running(unsigned int x)
+{
+ if (!x)
+ return -EINVAL;
+
+ if (dlm_allow_conn)
+ return -EBUSY;
+
+ return 0;
+}
+
+static int dlm_check_zero(unsigned int x)
{
- return !x;
+ if (!x)
+ return -EINVAL;
+
+ return 0;
}
-static bool dlm_check_buffer_size(unsigned int x)
+static int dlm_check_buffer_size(unsigned int x)
{
- return (x < DEFAULT_BUFFER_SIZE);
+ if (x < DEFAULT_BUFFER_SIZE)
+ return -EINVAL;
+
+ return 0;
}
-CLUSTER_ATTR(tcp_port, dlm_check_zero);
+CLUSTER_ATTR(tcp_port, dlm_check_zero_and_dlm_running);
CLUSTER_ATTR(buffer_size, dlm_check_buffer_size);
CLUSTER_ATTR(rsbtbl_size, dlm_check_zero);
CLUSTER_ATTR(recover_timer, dlm_check_zero);
@@ -179,7 +218,7 @@ CLUSTER_ATTR(toss_secs, dlm_check_zero);
CLUSTER_ATTR(scan_secs, dlm_check_zero);
CLUSTER_ATTR(log_debug, NULL);
CLUSTER_ATTR(log_info, NULL);
-CLUSTER_ATTR(protocol, NULL);
+CLUSTER_ATTR(protocol, dlm_check_protocol_and_dlm_running);
CLUSTER_ATTR(mark, NULL);
CLUSTER_ATTR(timewarn_cs, dlm_check_zero);
CLUSTER_ATTR(waitwarn_us, NULL);
@@ -688,6 +727,7 @@ static ssize_t comm_mark_show(struct config_item *item, char *buf)
static ssize_t comm_mark_store(struct config_item *item, const char *buf,
size_t len)
{
+ struct dlm_comm *comm;
unsigned int mark;
int rc;
@@ -695,7 +735,15 @@ static ssize_t comm_mark_store(struct config_item *item, const char *buf,
if (rc)
return rc;
- config_item_to_comm(item)->mark = mark;
+ if (mark == 0)
+ mark = dlm_config.ci_mark;
+
+ comm = config_item_to_comm(item);
+ rc = dlm_lowcomms_nodes_set_mark(comm->nodeid, mark);
+ if (rc)
+ return rc;
+
+ comm->mark = mark;
return len;
}
@@ -870,24 +918,6 @@ int dlm_comm_seq(int nodeid, uint32_t *seq)
return 0;
}
-void dlm_comm_mark(int nodeid, unsigned int *mark)
-{
- struct dlm_comm *cm;
-
- cm = get_comm(nodeid);
- if (!cm) {
- *mark = dlm_config.ci_mark;
- return;
- }
-
- if (cm->mark)
- *mark = cm->mark;
- else
- *mark = dlm_config.ci_mark;
-
- put_comm(cm);
-}
-
int dlm_our_nodeid(void)
{
return local_comm ? local_comm->nodeid : 0;
diff --git a/fs/dlm/config.h b/fs/dlm/config.h
index c210250a2581..d2cd4bd20313 100644
--- a/fs/dlm/config.h
+++ b/fs/dlm/config.h
@@ -48,7 +48,6 @@ void dlm_config_exit(void);
int dlm_config_nodes(char *lsname, struct dlm_config_node **nodes_out,
int *count_out);
int dlm_comm_seq(int nodeid, uint32_t *seq);
-void dlm_comm_mark(int nodeid, unsigned int *mark);
int dlm_our_nodeid(void);
int dlm_our_addr(struct sockaddr_storage *addr, int num);
diff --git a/fs/dlm/debug_fs.c b/fs/dlm/debug_fs.c
index d6bbccb0ed15..d5bd990bcab8 100644
--- a/fs/dlm/debug_fs.c
+++ b/fs/dlm/debug_fs.c
@@ -542,6 +542,7 @@ static void *table_seq_next(struct seq_file *seq, void *iter_ptr, loff_t *pos)
if (bucket >= ls->ls_rsbtbl_size) {
kfree(ri);
+ ++*pos;
return NULL;
}
tree = toss ? &ls->ls_rsbtbl[bucket].toss : &ls->ls_rsbtbl[bucket].keep;
diff --git a/fs/dlm/lock.c b/fs/dlm/lock.c
index 002123efc6b0..b93df39d0915 100644
--- a/fs/dlm/lock.c
+++ b/fs/dlm/lock.c
@@ -3541,8 +3541,6 @@ static int _create_message(struct dlm_ls *ls, int mb_len,
if (!mh)
return -ENOBUFS;
- memset(mb, 0, mb_len);
-
ms = (struct dlm_message *) mb;
ms->m_header.h_version = (DLM_HEADER_MAJOR | DLM_HEADER_MINOR);
diff --git a/fs/dlm/lockspace.c b/fs/dlm/lockspace.c
index 561dcad08ad6..c14cf2b7faab 100644
--- a/fs/dlm/lockspace.c
+++ b/fs/dlm/lockspace.c
@@ -404,12 +404,6 @@ static int threads_start(void)
return error;
}
-static void threads_stop(void)
-{
- dlm_scand_stop();
- dlm_lowcomms_stop();
-}
-
static int new_lockspace(const char *name, const char *cluster,
uint32_t flags, int lvblen,
const struct dlm_lockspace_ops *ops, void *ops_arg,
@@ -702,8 +696,11 @@ int dlm_new_lockspace(const char *name, const char *cluster,
ls_count++;
if (error > 0)
error = 0;
- if (!ls_count)
- threads_stop();
+ if (!ls_count) {
+ dlm_scand_stop();
+ dlm_lowcomms_shutdown();
+ dlm_lowcomms_stop();
+ }
out:
mutex_unlock(&ls_lock);
return error;
@@ -788,6 +785,11 @@ static int release_lockspace(struct dlm_ls *ls, int force)
dlm_recoverd_stop(ls);
+ if (ls_count == 1) {
+ dlm_scand_stop();
+ dlm_lowcomms_shutdown();
+ }
+
dlm_callback_stop(ls);
remove_lockspace(ls);
@@ -880,7 +882,7 @@ int dlm_release_lockspace(void *lockspace, int force)
if (!error)
ls_count--;
if (!ls_count)
- threads_stop();
+ dlm_lowcomms_stop();
mutex_unlock(&ls_lock);
return error;
diff --git a/fs/dlm/lowcomms.c b/fs/dlm/lowcomms.c
index 372c34ff8594..166e36fcf3e4 100644
--- a/fs/dlm/lowcomms.c
+++ b/fs/dlm/lowcomms.c
@@ -102,6 +102,9 @@ struct listen_connection {
struct work_struct rwork;
};
+#define DLM_WQ_REMAIN_BYTES(e) (PAGE_SIZE - e->end)
+#define DLM_WQ_LENGTH_BYTES(e) (e->end - e->offset)
+
/* An entry waiting to be sent */
struct writequeue_entry {
struct list_head list;
@@ -116,6 +119,7 @@ struct writequeue_entry {
struct dlm_node_addr {
struct list_head list;
int nodeid;
+ int mark;
int addr_count;
int curr_addr_index;
struct sockaddr_storage *addr[DLM_MAX_ADDR_COUNT];
@@ -134,7 +138,7 @@ static DEFINE_SPINLOCK(dlm_node_addrs_spin);
static struct listen_connection listen_con;
static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
static int dlm_local_count;
-static int dlm_allow_conn;
+int dlm_allow_conn;
/* Work queues */
static struct workqueue_struct *recv_workqueue;
@@ -303,7 +307,8 @@ static int addr_compare(const struct sockaddr_storage *x,
}
static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
- struct sockaddr *sa_out, bool try_new_addr)
+ struct sockaddr *sa_out, bool try_new_addr,
+ unsigned int *mark)
{
struct sockaddr_storage sas;
struct dlm_node_addr *na;
@@ -331,6 +336,8 @@ static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
if (!na->addr_count)
return -ENOENT;
+ *mark = na->mark;
+
if (sas_out)
memcpy(sas_out, &sas, sizeof(struct sockaddr_storage));
@@ -350,7 +357,8 @@ static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
return 0;
}
-static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid)
+static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid,
+ unsigned int *mark)
{
struct dlm_node_addr *na;
int rv = -EEXIST;
@@ -364,6 +372,7 @@ static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid)
for (addr_i = 0; addr_i < na->addr_count; addr_i++) {
if (addr_compare(na->addr[addr_i], addr)) {
*nodeid = na->nodeid;
+ *mark = na->mark;
rv = 0;
goto unlock;
}
@@ -412,6 +421,7 @@ int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len)
new_node->nodeid = nodeid;
new_node->addr[0] = new_addr;
new_node->addr_count = 1;
+ new_node->mark = dlm_config.ci_mark;
list_add(&new_node->list, &dlm_node_addrs);
spin_unlock(&dlm_node_addrs_spin);
return 0;
@@ -519,6 +529,23 @@ int dlm_lowcomms_connect_node(int nodeid)
return 0;
}
+int dlm_lowcomms_nodes_set_mark(int nodeid, unsigned int mark)
+{
+ struct dlm_node_addr *na;
+
+ spin_lock(&dlm_node_addrs_spin);
+ na = find_node_addr(nodeid);
+ if (!na) {
+ spin_unlock(&dlm_node_addrs_spin);
+ return -ENOENT;
+ }
+
+ na->mark = mark;
+ spin_unlock(&dlm_node_addrs_spin);
+
+ return 0;
+}
+
static void lowcomms_error_report(struct sock *sk)
{
struct connection *con;
@@ -685,10 +712,7 @@ static void shutdown_connection(struct connection *con)
{
int ret;
- if (cancel_work_sync(&con->swork)) {
- log_print("canceled swork for node %d", con->nodeid);
- clear_bit(CF_WRITE_PENDING, &con->flags);
- }
+ flush_work(&con->swork);
mutex_lock(&con->sock_mutex);
/* nothing to shutdown */
@@ -867,7 +891,7 @@ static int accept_from_sock(struct listen_connection *con)
/* Get the new node's NODEID */
make_sockaddr(&peeraddr, 0, &len);
- if (addr_to_nodeid(&peeraddr, &nodeid)) {
+ if (addr_to_nodeid(&peeraddr, &nodeid, &mark)) {
unsigned char *b=(unsigned char *)&peeraddr;
log_print("connect from non cluster node");
print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
@@ -876,9 +900,6 @@ static int accept_from_sock(struct listen_connection *con)
return -1;
}
- dlm_comm_mark(nodeid, &mark);
- sock_set_mark(newsock->sk, mark);
-
log_print("got connection from %d", nodeid);
/* Check to see if we already have a connection to this node. This
@@ -892,6 +913,8 @@ static int accept_from_sock(struct listen_connection *con)
goto accept_err;
}
+ sock_set_mark(newsock->sk, mark);
+
mutex_lock(&newcon->sock_mutex);
if (newcon->sock) {
struct connection *othercon = newcon->othercon;
@@ -908,16 +931,18 @@ static int accept_from_sock(struct listen_connection *con)
result = dlm_con_init(othercon, nodeid);
if (result < 0) {
kfree(othercon);
+ mutex_unlock(&newcon->sock_mutex);
goto accept_err;
}
+ lockdep_set_subclass(&othercon->sock_mutex, 1);
newcon->othercon = othercon;
} else {
/* close other sock con if we have something new */
close_connection(othercon, false, true, false);
}
- mutex_lock_nested(&othercon->sock_mutex, 1);
+ mutex_lock(&othercon->sock_mutex);
add_sock(newsock, othercon);
addcon = othercon;
mutex_unlock(&othercon->sock_mutex);
@@ -930,6 +955,7 @@ static int accept_from_sock(struct listen_connection *con)
addcon = newcon;
}
+ set_bit(CF_CONNECTED, &addcon->flags);
mutex_unlock(&newcon->sock_mutex);
/*
@@ -1015,8 +1041,6 @@ static void sctp_connect_to_sock(struct connection *con)
struct socket *sock;
unsigned int mark;
- dlm_comm_mark(con->nodeid, &mark);
-
mutex_lock(&con->sock_mutex);
/* Some odd races can cause double-connects, ignore them */
@@ -1029,7 +1053,7 @@ static void sctp_connect_to_sock(struct connection *con)
}
memset(&daddr, 0, sizeof(daddr));
- result = nodeid_to_addr(con->nodeid, &daddr, NULL, true);
+ result = nodeid_to_addr(con->nodeid, &daddr, NULL, true, &mark);
if (result < 0) {
log_print("no address for nodeid %d", con->nodeid);
goto out;
@@ -1104,13 +1128,11 @@ out:
static void tcp_connect_to_sock(struct connection *con)
{
struct sockaddr_storage saddr, src_addr;
+ unsigned int mark;
int addr_len;
struct socket *sock = NULL;
- unsigned int mark;
int result;
- dlm_comm_mark(con->nodeid, &mark);
-
mutex_lock(&con->sock_mutex);
if (con->retries++ > MAX_CONNECT_RETRIES)
goto out;
@@ -1125,15 +1147,15 @@ static void tcp_connect_to_sock(struct connection *con)
if (result < 0)
goto out_err;
- sock_set_mark(sock->sk, mark);
-
memset(&saddr, 0, sizeof(saddr));
- result = nodeid_to_addr(con->nodeid, &saddr, NULL, false);
+ result = nodeid_to_addr(con->nodeid, &saddr, NULL, false, &mark);
if (result < 0) {
log_print("no address for nodeid %d", con->nodeid);
goto out_err;
}
+ sock_set_mark(sock->sk, mark);
+
add_sock(sock, con);
/* Bind to our cluster-known address connecting to avoid
@@ -1330,70 +1352,72 @@ static struct writequeue_entry *new_writequeue_entry(struct connection *con,
{
struct writequeue_entry *entry;
- entry = kmalloc(sizeof(struct writequeue_entry), allocation);
+ entry = kzalloc(sizeof(*entry), allocation);
if (!entry)
return NULL;
- entry->page = alloc_page(allocation);
+ entry->page = alloc_page(allocation | __GFP_ZERO);
if (!entry->page) {
kfree(entry);
return NULL;
}
- entry->offset = 0;
- entry->len = 0;
- entry->end = 0;
- entry->users = 0;
entry->con = con;
+ entry->users = 1;
return entry;
}
-void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
+static struct writequeue_entry *new_wq_entry(struct connection *con, int len,
+ gfp_t allocation, char **ppc)
{
- struct connection *con;
struct writequeue_entry *e;
- int offset = 0;
- if (len > LOWCOMMS_MAX_TX_BUFFER_LEN) {
- BUILD_BUG_ON(PAGE_SIZE < LOWCOMMS_MAX_TX_BUFFER_LEN);
- log_print("failed to allocate a buffer of size %d", len);
- return NULL;
+ spin_lock(&con->writequeue_lock);
+ if (!list_empty(&con->writequeue)) {
+ e = list_last_entry(&con->writequeue, struct writequeue_entry, list);
+ if (DLM_WQ_REMAIN_BYTES(e) >= len) {
+ *ppc = page_address(e->page) + e->end;
+ e->end += len;
+ e->users++;
+ spin_unlock(&con->writequeue_lock);
+
+ return e;
+ }
}
+ spin_unlock(&con->writequeue_lock);
- con = nodeid2con(nodeid, allocation);
- if (!con)
+ e = new_writequeue_entry(con, allocation);
+ if (!e)
return NULL;
+ *ppc = page_address(e->page);
+ e->end += len;
+
spin_lock(&con->writequeue_lock);
- e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
- if ((&e->list == &con->writequeue) ||
- (PAGE_SIZE - e->end < len)) {
- e = NULL;
- } else {
- offset = e->end;
- e->end += len;
- e->users++;
- }
+ list_add_tail(&e->list, &con->writequeue);
spin_unlock(&con->writequeue_lock);
- if (e) {
- got_one:
- *ppc = page_address(e->page) + offset;
- return e;
- }
+ return e;
+};
- e = new_writequeue_entry(con, allocation);
- if (e) {
- spin_lock(&con->writequeue_lock);
- offset = e->end;
- e->end += len;
- e->users++;
- list_add_tail(&e->list, &con->writequeue);
- spin_unlock(&con->writequeue_lock);
- goto got_one;
+void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
+{
+ struct connection *con;
+
+ if (len > DEFAULT_BUFFER_SIZE ||
+ len < sizeof(struct dlm_header)) {
+ BUILD_BUG_ON(PAGE_SIZE < DEFAULT_BUFFER_SIZE);
+ log_print("failed to allocate a buffer of size %d", len);
+ WARN_ON(1);
+ return NULL;
}
- return NULL;
+
+ con = nodeid2con(nodeid, allocation);
+ if (!con)
+ return NULL;
+
+ return new_wq_entry(con, len, allocation, ppc);
}
void dlm_lowcomms_commit_buffer(void *mh)
@@ -1406,7 +1430,8 @@ void dlm_lowcomms_commit_buffer(void *mh)
users = --e->users;
if (users)
goto out;
- e->len = e->end - e->offset;
+
+ e->len = DLM_WQ_LENGTH_BYTES(e);
spin_unlock(&con->writequeue_lock);
queue_work(send_workqueue, &con->swork);
@@ -1432,11 +1457,10 @@ static void send_to_sock(struct connection *con)
spin_lock(&con->writequeue_lock);
for (;;) {
- e = list_entry(con->writequeue.next, struct writequeue_entry,
- list);
- if ((struct list_head *) e == &con->writequeue)
+ if (list_empty(&con->writequeue))
break;
+ e = list_first_entry(&con->writequeue, struct writequeue_entry, list);
len = e->len;
offset = e->offset;
BUG_ON(len == 0 && e->users == 0);
@@ -1589,6 +1613,29 @@ static int work_start(void)
return 0;
}
+static void shutdown_conn(struct connection *con)
+{
+ if (con->shutdown_action)
+ con->shutdown_action(con);
+}
+
+void dlm_lowcomms_shutdown(void)
+{
+ /* Set all the flags to prevent any
+ * socket activity.
+ */
+ dlm_allow_conn = 0;
+
+ if (recv_workqueue)
+ flush_workqueue(recv_workqueue);
+ if (send_workqueue)
+ flush_workqueue(send_workqueue);
+
+ dlm_close_sock(&listen_con.sock);
+
+ foreach_conn(shutdown_conn);
+}
+
static void _stop_conn(struct connection *con, bool and_other)
{
mutex_lock(&con->sock_mutex);
@@ -1610,12 +1657,6 @@ static void stop_conn(struct connection *con)
_stop_conn(con, true);
}
-static void shutdown_conn(struct connection *con)
-{
- if (con->shutdown_action)
- con->shutdown_action(con);
-}
-
static void connection_release(struct rcu_head *rcu)
{
struct connection *con = container_of(rcu, struct connection, rcu);
@@ -1672,19 +1713,6 @@ static void work_flush(void)
void dlm_lowcomms_stop(void)
{
- /* Set all the flags to prevent any
- socket activity.
- */
- dlm_allow_conn = 0;
-
- if (recv_workqueue)
- flush_workqueue(recv_workqueue);
- if (send_workqueue)
- flush_workqueue(send_workqueue);
-
- dlm_close_sock(&listen_con.sock);
-
- foreach_conn(shutdown_conn);
work_flush();
foreach_conn(free_conn);
work_stop();
diff --git a/fs/dlm/lowcomms.h b/fs/dlm/lowcomms.h
index 0918f9376489..48bbc4e18761 100644
--- a/fs/dlm/lowcomms.h
+++ b/fs/dlm/lowcomms.h
@@ -14,13 +14,18 @@
#define LOWCOMMS_MAX_TX_BUFFER_LEN 4096
+/* switch to check if dlm is running */
+extern int dlm_allow_conn;
+
int dlm_lowcomms_start(void);
+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);
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);
#endif /* __LOWCOMMS_DOT_H__ */
diff --git a/fs/dlm/midcomms.c b/fs/dlm/midcomms.c
index fde3a6afe4be..1c6654a21ec4 100644
--- a/fs/dlm/midcomms.c
+++ b/fs/dlm/midcomms.c
@@ -22,8 +22,6 @@
* into packets and sends them to the comms layer.
*/
-#include <asm/unaligned.h>
-
#include "dlm_internal.h"
#include "lowcomms.h"
#include "config.h"
@@ -45,13 +43,22 @@ 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 this otherwise we
- * cannot deliver this message to upper layers
+ /* no message should be more than DEFAULT_BUFFER_SIZE or
+ * less than dlm_header size.
+ *
+ * Some messages does not have a 8 byte length boundary yet
+ * which can occur in a unaligned memory access of some dlm
+ * messages. However this problem need to be fixed at the
+ * sending side, for now it seems nobody run into architecture
+ * related issues yet but it slows down some processing.
+ * Fixing this issue should be scheduled in future by doing
+ * the next major version bump.
*/
- msglen = get_unaligned_le16(&hd->h_length);
- if (msglen > DEFAULT_BUFFER_SIZE) {
- log_print("received invalid length header: %u, will abort message parsing",
- msglen);
+ msglen = le16_to_cpu(hd->h_length);
+ if (msglen > DEFAULT_BUFFER_SIZE ||
+ msglen < sizeof(struct dlm_header)) {
+ log_print("received invalid length header: %u from node %d, will abort message parsing",
+ msglen, nodeid);
return -EBADMSG;
}
@@ -84,15 +91,7 @@ int dlm_process_incoming_buffer(int nodeid, unsigned char *buf, int len)
goto skip;
}
- /* for aligned memory access, we just copy current message
- * to begin of the buffer which contains already parsed buffer
- * data and should provide align access for upper layers
- * because the start address of the buffer has a aligned
- * address. This memmove can be removed when the upperlayer
- * is capable of unaligned memory access.
- */
- memmove(buf, ptr, msglen);
- dlm_receive_buffer((union dlm_packet *)buf, nodeid);
+ dlm_receive_buffer((union dlm_packet *)ptr, nodeid);
skip:
ret += msglen;
diff --git a/fs/dlm/rcom.c b/fs/dlm/rcom.c
index 73ddee5159d7..f5b1bd65728d 100644
--- a/fs/dlm/rcom.c
+++ b/fs/dlm/rcom.c
@@ -41,7 +41,6 @@ static int create_rcom(struct dlm_ls *ls, int to_nodeid, int type, int len,
to_nodeid, type, len);
return -ENOBUFS;
}
- memset(mb, 0, mb_len);
rc = (struct dlm_rcom *) mb;
@@ -462,7 +461,6 @@ int dlm_send_ls_not_ready(int nodeid, struct dlm_rcom *rc_in)
mh = dlm_lowcomms_get_buffer(nodeid, mb_len, GFP_NOFS, &mb);
if (!mh)
return -ENOBUFS;
- memset(mb, 0, mb_len);
rc = (struct dlm_rcom *) mb;