// SPDX-License-Identifier: GPL-2.0-only/*****************************************************************************************************************************************************************  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.**  Copyright (C) 2004-2009 Red Hat, Inc.  All rights reserved.*****************************************************************************************************************************************************************/ /* * lowcomms.c * * This is the "low-level" comms layer. * * It is responsible for sending/receiving messages * from other nodes in the cluster. * * Cluster nodes are referred to by their nodeids. nodeids are * simply 32 bit numbers to the locking module - if they need to * be expanded for the cluster infrastructure then that is its * responsibility. It is this layer's * responsibility to resolve these into IP address or * whatever it needs for inter-node communication. * * The comms level is two kernel threads that deal mainly with * the receiving of messages from other nodes and passing them * up to the mid-level comms layer (which understands the * message format) for execution by the locking core, and * a send thread which does all the setting up of connections * to remote nodes and the sending of data. Threads are not allowed * to send their own data because it may cause them to wait in times * of high load. Also, this way, the sending thread can collect together * messages bound for one node and send them in one block. * * lowcomms will choose to use either TCP or SCTP as its transport layer * depending on the configuration variable 'protocol'. This should be set * to 0 (default) for TCP or 1 for SCTP. It should be configured using a * cluster-wide mechanism as it must be the same on all nodes of the cluster * for the DLM to function. * */ #include <asm/ioctls.h>#include <net/sock.h>#include <net/tcp.h>#include <linux/pagemap.h>#include <linux/file.h>#include <linux/mutex.h>#include <linux/sctp.h>#include <linux/slab.h>#include <net/sctp/sctp.h>#include <net/ipv6.h> #include <trace/events/dlm.h>#include <trace/events/sock.h> #include "dlm_internal.h"#include "lowcomms.h"#include "midcomms.h"#include "memory.h"#include "config.h" #define DLM_SHUTDOWN_WAIT_TIMEOUT msecs_to_jiffies(5000)#define DLM_MAX_PROCESS_BUFFERS 24#define NEEDED_RMEM (4*1024*1024) struct connection {	struct socket *sock;	/* NULL if not connected */	uint32_t nodeid;	/* So we know who we are in the list */	/* this semaphore is used to allow parallel recv/send in read	 * lock mode. When we release a sock we need to held the write lock.	 *	 * However this is locking code and not nice. When we remove the	 * othercon handling we can look into other mechanism to synchronize	 * io handling to call sock_release() at the right time.	 */	struct rw_semaphore sock_lock;	unsigned long flags;#define CF_APP_LIMITED 0#define CF_RECV_PENDING 1#define CF_SEND_PENDING 2#define CF_RECV_INTR 3#define CF_IO_STOP 4#define CF_IS_OTHERCON 5	struct list_head writequeue;  /* List of outgoing writequeue_entries */	spinlock_t writequeue_lock;	int retries;	struct hlist_node list;	/* due some connect()/accept() races we currently have this cross over	 * connection attempt second connection for one node.	 *	 * There is a solution to avoid the race by introducing a connect	 * rule as e.g. our_nodeid > nodeid_to_connect who is allowed to	 * connect. Otherside can connect but will only be considered that	 * the other side wants to have a reconnect.	 *	 * However changing to this behaviour will break backwards compatible.	 * In a DLM protocol major version upgrade we should remove this!	 */	struct connection *othercon;	struct work_struct rwork; /* receive worker */	struct work_struct swork; /* send worker */	wait_queue_head_t shutdown_wait;	unsigned char rx_leftover_buf[DLM_MAX_SOCKET_BUFSIZE];	int rx_leftover;	int mark;	int addr_count;	int curr_addr_index;	struct sockaddr_storage addr[DLM_MAX_ADDR_COUNT];	spinlock_t addrs_lock;	struct rcu_head rcu;};#define sock2con(x) ((struct connection *)(x)->sk_user_data) struct listen_connection {	struct socket *sock;	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;	struct page *page;	int offset;	int len;	int end;	int users;	bool dirty;	struct connection *con;	struct list_head msgs;	struct kref ref;}; struct dlm_msg {	struct writequeue_entry *entry;	struct dlm_msg *orig_msg;	bool retransmit;	void *ppc;	int len;	int idx; /* new()/commit() idx exchange */ 	struct list_head list;	struct kref ref;}; struct processqueue_entry {	unsigned char *buf;	int nodeid;	int buflen; 	struct list_head list;}; struct dlm_proto_ops {	bool try_new_addr;	const char *name;	int proto;	int how; 	void (*sockopts)(struct socket *sock);	int (*bind)(struct socket *sock);	int (*listen_validate)(void);	void (*listen_sockopts)(struct socket *sock);	int (*listen_bind)(struct socket *sock);}; static struct listen_sock_callbacks {	void (*sk_error_report)(struct sock *);	void (*sk_data_ready)(struct sock *);	void (*sk_state_change)(struct sock *);	void (*sk_write_space)(struct sock *);} listen_sock; static struct listen_connection listen_con;static struct sockaddr_storage dlm_local_addr[DLM_MAX_ADDR_COUNT];static int dlm_local_count; /* Work queues */static struct workqueue_struct *io_workqueue;static struct workqueue_struct *process_workqueue; static struct hlist_head connection_hash[CONN_HASH_SIZE];static DEFINE_SPINLOCK(connections_lock);DEFINE_STATIC_SRCU(connections_srcu); static const struct dlm_proto_ops *dlm_proto_ops; #define DLM_IO_SUCCESS 0#define DLM_IO_END 1#define DLM_IO_EOF 2#define DLM_IO_RESCHED 3#define DLM_IO_FLUSH 4 static void process_recv_sockets(struct work_struct *work);static void process_send_sockets(struct work_struct *work);static void process_dlm_messages(struct work_struct *work); static DECLARE_WORK(process_work, process_dlm_messages);static DEFINE_SPINLOCK(processqueue_lock);static bool process_dlm_messages_pending;static DECLARE_WAIT_QUEUE_HEAD(processqueue_wq);static atomic_t processqueue_count;static LIST_HEAD(processqueue); bool dlm_lowcomms_is_running(void){	return !!listen_con.sock;} static void lowcomms_queue_swork(struct connection *con){	assert_spin_locked(&con->writequeue_lock); 	if (!test_bit(CF_IO_STOP, &con->flags) &&	    !test_bit(CF_APP_LIMITED, &con->flags) &&	    !test_and_set_bit(CF_SEND_PENDING, &con->flags))		queue_work(io_workqueue, &con->swork);} static void lowcomms_queue_rwork(struct connection *con){#ifdef CONFIG_LOCKDEP	WARN_ON_ONCE(!lockdep_sock_is_held(con->sock->sk));#endif 	if (!test_bit(CF_IO_STOP, &con->flags) &&	    !test_and_set_bit(CF_RECV_PENDING, &con->flags))		queue_work(io_workqueue, &con->rwork);} static void writequeue_entry_ctor(void *data){	struct writequeue_entry *entry = data; 	INIT_LIST_HEAD(&entry->msgs);} struct kmem_cache *dlm_lowcomms_writequeue_cache_create(void){	return kmem_cache_create("dlm_writequeue", sizeof(struct writequeue_entry),				 0, 0, writequeue_entry_ctor);} struct kmem_cache *dlm_lowcomms_msg_cache_create(void){	return KMEM_CACHE(dlm_msg, 0);} /* need to held writequeue_lock */static struct writequeue_entry *con_next_wq(struct connection *con){	struct writequeue_entry *e; 	e = list_first_entry_or_null(&con->writequeue, struct writequeue_entry,				     list);	/* if len is zero nothing is to send, if there are users filling	 * buffers we wait until the users are done so we can send more.	 */	if (!e || e->users || e->len == 0)		return NULL; 	return e;} static struct connection *__find_con(int nodeid, int r){	struct connection *con; 	hlist_for_each_entry_srcu(con, &connection_hash[r], list,			srcu_read_lock_held(&connections_srcu)) {		if (con->nodeid == nodeid)			return con;	} 	return NULL;} static void dlm_con_init(struct connection *con, int nodeid){	con->nodeid = nodeid;	init_rwsem(&con->sock_lock);	INIT_LIST_HEAD(&con->writequeue);	spin_lock_init(&con->writequeue_lock);	INIT_WORK(&con->swork, process_send_sockets);	INIT_WORK(&con->rwork, process_recv_sockets);	spin_lock_init(&con->addrs_lock);	init_waitqueue_head(&con->shutdown_wait);} /* * If 'allocation' is zero then we don't attempt to create a new * connection structure for this node. */static struct connection *nodeid2con(int nodeid, gfp_t alloc){	struct connection *con, *tmp;	int r; 	r = nodeid_hash(nodeid);	con = __find_con(nodeid, r);	if (con || !alloc)		return con; 	con = kzalloc_obj(*con, alloc);	if (!con)		return NULL; 	dlm_con_init(con, 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()	 * we just check in rare cases of recently added nodes again	 * under protection of connections_lock. If this is the case we	 * abort our connection creation and return the existing connection.	 */	tmp = __find_con(nodeid, r);	if (tmp) {		spin_unlock(&connections_lock);		kfree(con);		return tmp;	} 	hlist_add_head_rcu(&con->list, &connection_hash[r]);	spin_unlock(&connections_lock); 	return con;} static int addr_compare(const struct sockaddr_storage *x,			const struct sockaddr_storage *y){	switch (x->ss_family) {	case AF_INET: {		struct sockaddr_in *sinx = (struct sockaddr_in *)x;		struct sockaddr_in *siny = (struct sockaddr_in *)y;		if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr)			return 0;		if (sinx->sin_port != siny->sin_port)			return 0;		break;	}	case AF_INET6: {		struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x;		struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y;		if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr))			return 0;		if (sinx->sin6_port != siny->sin6_port)			return 0;		break;	}	default:		return 0;	}	return 1;} static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,			  struct sockaddr *sa_out, bool try_new_addr,			  unsigned int *mark){	struct sockaddr_storage sas;	struct connection *con;	int idx; 	if (!dlm_local_count)		return -1; 	idx = srcu_read_lock(&connections_srcu);	con = nodeid2con(nodeid, 0);	if (!con) {		srcu_read_unlock(&connections_srcu, idx);		return -ENOENT;	} 	spin_lock(&con->addrs_lock);	if (!con->addr_count) {		spin_unlock(&con->addrs_lock);		srcu_read_unlock(&connections_srcu, idx);		return -ENOENT;	} 	memcpy(&sas, &con->addr[con->curr_addr_index],	       sizeof(struct sockaddr_storage)); 	if (try_new_addr) {		con->curr_addr_index++;		if (con->curr_addr_index == con->addr_count)			con->curr_addr_index = 0;	} 	*mark = con->mark;	spin_unlock(&con->addrs_lock); 	if (sas_out)		memcpy(sas_out, &sas, sizeof(struct sockaddr_storage)); 	if (!sa_out) {		srcu_read_unlock(&connections_srcu, idx);		return 0;	} 	if (dlm_local_addr[0].ss_family == AF_INET) {		struct sockaddr_in *in4  = (struct sockaddr_in *) &sas;		struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;		ret4->sin_addr.s_addr = in4->sin_addr.s_addr;	} else {		struct sockaddr_in6 *in6  = (struct sockaddr_in6 *) &sas;		struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;		ret6->sin6_addr = in6->sin6_addr;	} 	srcu_read_unlock(&connections_srcu, idx);	return 0;} static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid,			  unsigned int *mark){	struct connection *con;	int i, idx, addr_i; 	idx = srcu_read_lock(&connections_srcu);	for (i = 0; i < CONN_HASH_SIZE; i++) {		hlist_for_each_entry_srcu(con, &connection_hash[i], list,				srcu_read_lock_held(&connections_srcu)) {			WARN_ON_ONCE(!con->addr_count); 			spin_lock(&con->addrs_lock);			for (addr_i = 0; addr_i < con->addr_count; addr_i++) {				if (addr_compare(&con->addr[addr_i], addr)) {					*nodeid = con->nodeid;					*mark = con->mark;					spin_unlock(&con->addrs_lock);					srcu_read_unlock(&connections_srcu, idx);					return 0;				}			}			spin_unlock(&con->addrs_lock);		}	}	srcu_read_unlock(&connections_srcu, idx); 	return -ENOENT;} static bool dlm_lowcomms_con_has_addr(const struct connection *con,				      const struct sockaddr_storage *addr){	int i; 	for (i = 0; i < con->addr_count; i++) {		if (addr_compare(&con->addr[i], addr))			return true;	} 	return false;} int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr){	struct connection *con;	bool ret;	int idx; 	idx = srcu_read_lock(&connections_srcu);	con = nodeid2con(nodeid, GFP_NOFS);	if (!con) {		srcu_read_unlock(&connections_srcu, idx);		return -ENOMEM;	} 	spin_lock(&con->addrs_lock);	if (!con->addr_count) {		memcpy(&con->addr[0], addr, sizeof(*addr));		con->addr_count = 1;		con->mark = dlm_config.ci_mark;		spin_unlock(&con->addrs_lock);		srcu_read_unlock(&connections_srcu, idx);		return 0;	} 	ret = dlm_lowcomms_con_has_addr(con, addr);	if (ret) {		spin_unlock(&con->addrs_lock);		srcu_read_unlock(&connections_srcu, idx);		return -EEXIST;	} 	if (con->addr_count >= DLM_MAX_ADDR_COUNT) {		spin_unlock(&con->addrs_lock);		srcu_read_unlock(&connections_srcu, idx);		return -ENOSPC;	} 

Lines 1–500 of 1987. Load more