1// SPDX-License-Identifier: GPL-2.0-only
   2/******************************************************************************
   3*******************************************************************************
   4**
   5**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
   6**  Copyright (C) 2004-2009 Red Hat, Inc.  All rights reserved.
   7**
   8**
   9*******************************************************************************
  10******************************************************************************/
  11
  12/*
  13 * lowcomms.c
  14 *
  15 * This is the "low-level" comms layer.
  16 *
  17 * It is responsible for sending/receiving messages
  18 * from other nodes in the cluster.
  19 *
  20 * Cluster nodes are referred to by their nodeids. nodeids are
  21 * simply 32 bit numbers to the locking module - if they need to
  22 * be expanded for the cluster infrastructure then that is its
  23 * responsibility. It is this layer's
  24 * responsibility to resolve these into IP address or
  25 * whatever it needs for inter-node communication.
  26 *
  27 * The comms level is two kernel threads that deal mainly with
  28 * the receiving of messages from other nodes and passing them
  29 * up to the mid-level comms layer (which understands the
  30 * message format) for execution by the locking core, and
  31 * a send thread which does all the setting up of connections
  32 * to remote nodes and the sending of data. Threads are not allowed
  33 * to send their own data because it may cause them to wait in times
  34 * of high load. Also, this way, the sending thread can collect together
  35 * messages bound for one node and send them in one block.
  36 *
  37 * lowcomms will choose to use either TCP or SCTP as its transport layer
  38 * depending on the configuration variable 'protocol'. This should be set
  39 * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
  40 * cluster-wide mechanism as it must be the same on all nodes of the cluster
  41 * for the DLM to function.
  42 *
  43 */
  44
  45#include <asm/ioctls.h>
  46#include <net/sock.h>
  47#include <net/tcp.h>
  48#include <linux/pagemap.h>
  49#include <linux/file.h>
  50#include <linux/mutex.h>
  51#include <linux/sctp.h>
  52#include <linux/slab.h>
  53#include <net/sctp/sctp.h>
  54#include <net/ipv6.h>
  55
  56#include <trace/events/dlm.h>
  57#include <trace/events/sock.h>
  58
  59#include "dlm_internal.h"
  60#include "lowcomms.h"
  61#include "midcomms.h"
  62#include "memory.h"
  63#include "config.h"
  64
  65#define DLM_SHUTDOWN_WAIT_TIMEOUT msecs_to_jiffies(5000)
  66#define DLM_MAX_PROCESS_BUFFERS 24
  67#define NEEDED_RMEM (4*1024*1024)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
  68
  69struct connection {
  70	struct socket *sock;	/* NULL if not connected */
  71	uint32_t nodeid;	/* So we know who we are in the list */
  72	/* this semaphore is used to allow parallel recv/send in read
  73	 * lock mode. When we release a sock we need to held the write lock.
  74	 *
  75	 * However this is locking code and not nice. When we remove the
  76	 * othercon handling we can look into other mechanism to synchronize
  77	 * io handling to call sock_release() at the right time.
  78	 */
  79	struct rw_semaphore sock_lock;
  80	unsigned long flags;
  81#define CF_APP_LIMITED 0
  82#define CF_RECV_PENDING 1
  83#define CF_SEND_PENDING 2
  84#define CF_RECV_INTR 3
  85#define CF_IO_STOP 4
  86#define CF_IS_OTHERCON 5
 
 
 
 
  87	struct list_head writequeue;  /* List of outgoing writequeue_entries */
  88	spinlock_t writequeue_lock;
 
 
 
 
 
  89	int retries;
 
  90	struct hlist_node list;
  91	/* due some connect()/accept() races we currently have this cross over
  92	 * connection attempt second connection for one node.
  93	 *
  94	 * There is a solution to avoid the race by introducing a connect
  95	 * rule as e.g. our_nodeid > nodeid_to_connect who is allowed to
  96	 * connect. Otherside can connect but will only be considered that
  97	 * the other side wants to have a reconnect.
  98	 *
  99	 * However changing to this behaviour will break backwards compatible.
 100	 * In a DLM protocol major version upgrade we should remove this!
 101	 */
 102	struct connection *othercon;
 103	struct work_struct rwork; /* receive worker */
 104	struct work_struct swork; /* send worker */
 105	wait_queue_head_t shutdown_wait;
 106	unsigned char rx_leftover_buf[DLM_MAX_SOCKET_BUFSIZE];
 107	int rx_leftover;
 108	int mark;
 109	int addr_count;
 110	int curr_addr_index;
 111	struct sockaddr_storage addr[DLM_MAX_ADDR_COUNT];
 112	spinlock_t addrs_lock;
 113	struct rcu_head rcu;
 114};
 115#define sock2con(x) ((struct connection *)(x)->sk_user_data)
 116
 117struct listen_connection {
 118	struct socket *sock;
 119	struct work_struct rwork;
 120};
 121
 122#define DLM_WQ_REMAIN_BYTES(e) (PAGE_SIZE - e->end)
 123#define DLM_WQ_LENGTH_BYTES(e) (e->end - e->offset)
 124
 125/* An entry waiting to be sent */
 126struct writequeue_entry {
 127	struct list_head list;
 128	struct page *page;
 129	int offset;
 130	int len;
 131	int end;
 132	int users;
 133	bool dirty;
 134	struct connection *con;
 135	struct list_head msgs;
 136	struct kref ref;
 137};
 138
 139struct dlm_msg {
 140	struct writequeue_entry *entry;
 141	struct dlm_msg *orig_msg;
 142	bool retransmit;
 143	void *ppc;
 144	int len;
 145	int idx; /* new()/commit() idx exchange */
 146
 147	struct list_head list;
 148	struct kref ref;
 149};
 150
 151struct processqueue_entry {
 152	unsigned char *buf;
 153	int nodeid;
 154	int buflen;
 155
 156	struct list_head list;
 157};
 158
 159struct dlm_proto_ops {
 160	bool try_new_addr;
 161	const char *name;
 162	int proto;
 163
 164	void (*sockopts)(struct socket *sock);
 165	int (*bind)(struct socket *sock);
 166	int (*listen_validate)(void);
 167	void (*listen_sockopts)(struct socket *sock);
 168	int (*listen_bind)(struct socket *sock);
 169};
 170
 171static struct listen_sock_callbacks {
 172	void (*sk_error_report)(struct sock *);
 173	void (*sk_data_ready)(struct sock *);
 174	void (*sk_state_change)(struct sock *);
 175	void (*sk_write_space)(struct sock *);
 176} listen_sock;
 177
 178static struct listen_connection listen_con;
 179static struct sockaddr_storage dlm_local_addr[DLM_MAX_ADDR_COUNT];
 
 
 180static int dlm_local_count;
 
 181
 182/* Work queues */
 183static struct workqueue_struct *io_workqueue;
 184static struct workqueue_struct *process_workqueue;
 185
 186static struct hlist_head connection_hash[CONN_HASH_SIZE];
 187static DEFINE_SPINLOCK(connections_lock);
 188DEFINE_STATIC_SRCU(connections_srcu);
 189
 190static const struct dlm_proto_ops *dlm_proto_ops;
 191
 192#define DLM_IO_SUCCESS 0
 193#define DLM_IO_END 1
 194#define DLM_IO_EOF 2
 195#define DLM_IO_RESCHED 3
 196#define DLM_IO_FLUSH 4
 197
 198static void process_recv_sockets(struct work_struct *work);
 199static void process_send_sockets(struct work_struct *work);
 200static void process_dlm_messages(struct work_struct *work);
 201
 202static DECLARE_WORK(process_work, process_dlm_messages);
 203static DEFINE_SPINLOCK(processqueue_lock);
 204static bool process_dlm_messages_pending;
 205static DECLARE_WAIT_QUEUE_HEAD(processqueue_wq);
 206static atomic_t processqueue_count;
 207static LIST_HEAD(processqueue);
 208
 209bool dlm_lowcomms_is_running(void)
 210{
 211	return !!listen_con.sock;
 212}
 213
 214static void lowcomms_queue_swork(struct connection *con)
 215{
 216	assert_spin_locked(&con->writequeue_lock);
 217
 218	if (!test_bit(CF_IO_STOP, &con->flags) &&
 219	    !test_bit(CF_APP_LIMITED, &con->flags) &&
 220	    !test_and_set_bit(CF_SEND_PENDING, &con->flags))
 221		queue_work(io_workqueue, &con->swork);
 222}
 223
 224static void lowcomms_queue_rwork(struct connection *con)
 225{
 226#ifdef CONFIG_LOCKDEP
 227	WARN_ON_ONCE(!lockdep_sock_is_held(con->sock->sk));
 228#endif
 229
 230	if (!test_bit(CF_IO_STOP, &con->flags) &&
 231	    !test_and_set_bit(CF_RECV_PENDING, &con->flags))
 232		queue_work(io_workqueue, &con->rwork);
 233}
 234
 235static void writequeue_entry_ctor(void *data)
 236{
 237	struct writequeue_entry *entry = data;
 238
 239	INIT_LIST_HEAD(&entry->msgs);
 240}
 241
 242struct kmem_cache *dlm_lowcomms_writequeue_cache_create(void)
 
 
 
 243{
 244	return kmem_cache_create("dlm_writequeue", sizeof(struct writequeue_entry),
 245				 0, 0, writequeue_entry_ctor);
 246}
 247
 248struct kmem_cache *dlm_lowcomms_msg_cache_create(void)
 249{
 250	return KMEM_CACHE(dlm_msg, 0);
 251}
 252
 253/* need to held writequeue_lock */
 254static struct writequeue_entry *con_next_wq(struct connection *con)
 255{
 256	struct writequeue_entry *e;
 257
 258	e = list_first_entry_or_null(&con->writequeue, struct writequeue_entry,
 259				     list);
 260	/* if len is zero nothing is to send, if there are users filling
 261	 * buffers we wait until the users are done so we can send more.
 262	 */
 263	if (!e || e->users || e->len == 0)
 264		return NULL;
 265
 266	return e;
 267}
 268
 269static struct connection *__find_con(int nodeid, int r)
 270{
 
 271	struct connection *con;
 272
 273	hlist_for_each_entry_rcu(con, &connection_hash[r], list) {
 
 
 274		if (con->nodeid == nodeid)
 275			return con;
 276	}
 277
 278	return NULL;
 279}
 280
 281static void dlm_con_init(struct connection *con, int nodeid)
 282{
 283	con->nodeid = nodeid;
 284	init_rwsem(&con->sock_lock);
 285	INIT_LIST_HEAD(&con->writequeue);
 286	spin_lock_init(&con->writequeue_lock);
 287	INIT_WORK(&con->swork, process_send_sockets);
 288	INIT_WORK(&con->rwork, process_recv_sockets);
 289	spin_lock_init(&con->addrs_lock);
 290	init_waitqueue_head(&con->shutdown_wait);
 291}
 292
 293/*
 294 * If 'allocation' is zero then we don't attempt to create a new
 295 * connection structure for this node.
 296 */
 297static struct connection *nodeid2con(int nodeid, gfp_t alloc)
 298{
 299	struct connection *con, *tmp;
 300	int r;
 301
 302	r = nodeid_hash(nodeid);
 303	con = __find_con(nodeid, r);
 304	if (con || !alloc)
 305		return con;
 306
 307	con = kzalloc(sizeof(*con), alloc);
 308	if (!con)
 309		return NULL;
 310
 311	dlm_con_init(con, nodeid);
 
 312
 313	spin_lock(&connections_lock);
 314	/* Because multiple workqueues/threads calls this function it can
 315	 * race on multiple cpu's. Instead of locking hot path __find_con()
 316	 * we just check in rare cases of recently added nodes again
 317	 * under protection of connections_lock. If this is the case we
 318	 * abort our connection creation and return the existing connection.
 319	 */
 320	tmp = __find_con(nodeid, r);
 321	if (tmp) {
 322		spin_unlock(&connections_lock);
 323		kfree(con);
 324		return tmp;
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 325	}
 
 326
 327	hlist_add_head_rcu(&con->list, &connection_hash[r]);
 328	spin_unlock(&connections_lock);
 
 
 
 
 
 329
 330	return con;
 331}
 332
 333static int addr_compare(const struct sockaddr_storage *x,
 334			const struct sockaddr_storage *y)
 
 
 
 
 
 
 
 
 
 
 335{
 336	switch (x->ss_family) {
 337	case AF_INET: {
 338		struct sockaddr_in *sinx = (struct sockaddr_in *)x;
 339		struct sockaddr_in *siny = (struct sockaddr_in *)y;
 340		if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr)
 341			return 0;
 342		if (sinx->sin_port != siny->sin_port)
 343			return 0;
 344		break;
 345	}
 346	case AF_INET6: {
 347		struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x;
 348		struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y;
 349		if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr))
 350			return 0;
 351		if (sinx->sin6_port != siny->sin6_port)
 352			return 0;
 353		break;
 354	}
 355	default:
 356		return 0;
 357	}
 358	return 1;
 359}
 360
 361static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
 362			  struct sockaddr *sa_out, bool try_new_addr,
 363			  unsigned int *mark)
 364{
 365	struct sockaddr_storage sas;
 366	struct connection *con;
 367	int idx;
 368
 369	if (!dlm_local_count)
 370		return -1;
 371
 372	idx = srcu_read_lock(&connections_srcu);
 373	con = nodeid2con(nodeid, 0);
 374	if (!con) {
 375		srcu_read_unlock(&connections_srcu, idx);
 376		return -ENOENT;
 377	}
 378
 379	spin_lock(&con->addrs_lock);
 380	if (!con->addr_count) {
 381		spin_unlock(&con->addrs_lock);
 382		srcu_read_unlock(&connections_srcu, idx);
 383		return -ENOENT;
 384	}
 
 385
 386	memcpy(&sas, &con->addr[con->curr_addr_index],
 387	       sizeof(struct sockaddr_storage));
 388
 389	if (try_new_addr) {
 390		con->curr_addr_index++;
 391		if (con->curr_addr_index == con->addr_count)
 392			con->curr_addr_index = 0;
 393	}
 394
 395	*mark = con->mark;
 396	spin_unlock(&con->addrs_lock);
 397
 398	if (sas_out)
 399		memcpy(sas_out, &sas, sizeof(struct sockaddr_storage));
 400
 401	if (!sa_out) {
 402		srcu_read_unlock(&connections_srcu, idx);
 403		return 0;
 404	}
 405
 406	if (dlm_local_addr[0].ss_family == AF_INET) {
 407		struct sockaddr_in *in4  = (struct sockaddr_in *) &sas;
 408		struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;
 409		ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
 410	} else {
 411		struct sockaddr_in6 *in6  = (struct sockaddr_in6 *) &sas;
 412		struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;
 413		ret6->sin6_addr = in6->sin6_addr;
 414	}
 415
 416	srcu_read_unlock(&connections_srcu, idx);
 417	return 0;
 418}
 419
 420static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid,
 421			  unsigned int *mark)
 422{
 423	struct connection *con;
 424	int i, idx, addr_i;
 425
 426	idx = srcu_read_lock(&connections_srcu);
 427	for (i = 0; i < CONN_HASH_SIZE; i++) {
 428		hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
 429			WARN_ON_ONCE(!con->addr_count);
 430
 431			spin_lock(&con->addrs_lock);
 432			for (addr_i = 0; addr_i < con->addr_count; addr_i++) {
 433				if (addr_compare(&con->addr[addr_i], addr)) {
 434					*nodeid = con->nodeid;
 435					*mark = con->mark;
 436					spin_unlock(&con->addrs_lock);
 437					srcu_read_unlock(&connections_srcu, idx);
 438					return 0;
 439				}
 440			}
 441			spin_unlock(&con->addrs_lock);
 442		}
 443	}
 444	srcu_read_unlock(&connections_srcu, idx);
 445
 446	return -ENOENT;
 447}
 448
 449static bool dlm_lowcomms_con_has_addr(const struct connection *con,
 450				      const struct sockaddr_storage *addr)
 451{
 452	int i;
 453
 454	for (i = 0; i < con->addr_count; i++) {
 455		if (addr_compare(&con->addr[i], addr))
 456			return true;
 457	}
 458
 459	return false;
 460}
 461
 462int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr)
 463{
 464	struct connection *con;
 465	bool ret;
 466	int idx;
 467
 468	idx = srcu_read_lock(&connections_srcu);
 469	con = nodeid2con(nodeid, GFP_NOFS);
 470	if (!con) {
 471		srcu_read_unlock(&connections_srcu, idx);
 472		return -ENOMEM;
 473	}
 474
 475	spin_lock(&con->addrs_lock);
 476	if (!con->addr_count) {
 477		memcpy(&con->addr[0], addr, sizeof(*addr));
 478		con->addr_count = 1;
 479		con->mark = dlm_config.ci_mark;
 480		spin_unlock(&con->addrs_lock);
 481		srcu_read_unlock(&connections_srcu, idx);
 482		return 0;
 483	}
 484
 485	ret = dlm_lowcomms_con_has_addr(con, addr);
 486	if (ret) {
 487		spin_unlock(&con->addrs_lock);
 488		srcu_read_unlock(&connections_srcu, idx);
 489		return -EEXIST;
 
 
 
 
 490	}
 491
 492	if (con->addr_count >= DLM_MAX_ADDR_COUNT) {
 493		spin_unlock(&con->addrs_lock);
 494		srcu_read_unlock(&connections_srcu, idx);
 
 495		return -ENOSPC;
 496	}
 497
 498	memcpy(&con->addr[con->addr_count++], addr, sizeof(*addr));
 499	srcu_read_unlock(&connections_srcu, idx);
 500	spin_unlock(&con->addrs_lock);
 501	return 0;
 502}
 503
 504/* Data available on socket or listen socket received a connect */
 505static void lowcomms_data_ready(struct sock *sk)
 506{
 507	struct connection *con = sock2con(sk);
 508
 509	trace_sk_data_ready(sk);
 510
 511	set_bit(CF_RECV_INTR, &con->flags);
 512	lowcomms_queue_rwork(con);
 
 
 
 513}
 514
 515static void lowcomms_write_space(struct sock *sk)
 516{
 517	struct connection *con = sock2con(sk);
 
 
 
 
 
 518
 519	clear_bit(SOCK_NOSPACE, &con->sock->flags);
 520
 521	spin_lock_bh(&con->writequeue_lock);
 522	if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
 523		con->sock->sk->sk_write_pending--;
 524		clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags);
 525	}
 526
 527	lowcomms_queue_swork(con);
 528	spin_unlock_bh(&con->writequeue_lock);
 
 529}
 530
 531static void lowcomms_state_change(struct sock *sk)
 532{
 533	/* SCTP layer is not calling sk_data_ready when the connection
 534	 * is done, so we catch the signal through here.
 535	 */
 536	if (sk->sk_shutdown == RCV_SHUTDOWN)
 537		lowcomms_data_ready(sk);
 538}
 539
 540static void lowcomms_listen_data_ready(struct sock *sk)
 541{
 542	trace_sk_data_ready(sk);
 543
 544	queue_work(io_workqueue, &listen_con.rwork);
 
 
 
 
 
 
 
 
 545}
 546
 547int dlm_lowcomms_connect_node(int nodeid)
 548{
 549	struct connection *con;
 550	int idx;
 551
 552	idx = srcu_read_lock(&connections_srcu);
 553	con = nodeid2con(nodeid, 0);
 554	if (WARN_ON_ONCE(!con)) {
 555		srcu_read_unlock(&connections_srcu, idx);
 556		return -ENOENT;
 557	}
 558
 559	down_read(&con->sock_lock);
 560	if (!con->sock) {
 561		spin_lock_bh(&con->writequeue_lock);
 562		lowcomms_queue_swork(con);
 563		spin_unlock_bh(&con->writequeue_lock);
 564	}
 565	up_read(&con->sock_lock);
 566	srcu_read_unlock(&connections_srcu, idx);
 567
 568	cond_resched();
 
 
 
 569	return 0;
 570}
 571
 572int dlm_lowcomms_nodes_set_mark(int nodeid, unsigned int mark)
 573{
 574	struct connection *con;
 575	int idx;
 576
 577	idx = srcu_read_lock(&connections_srcu);
 578	con = nodeid2con(nodeid, 0);
 579	if (!con) {
 580		srcu_read_unlock(&connections_srcu, idx);
 581		return -ENOENT;
 582	}
 583
 584	spin_lock(&con->addrs_lock);
 585	con->mark = mark;
 586	spin_unlock(&con->addrs_lock);
 587	srcu_read_unlock(&connections_srcu, idx);
 588	return 0;
 589}
 590
 591static void lowcomms_error_report(struct sock *sk)
 592{
 593	struct connection *con = sock2con(sk);
 594	struct inet_sock *inet;
 595
 596	inet = inet_sk(sk);
 597	switch (sk->sk_family) {
 598	case AF_INET:
 599		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
 600				   "sending to node %d at %pI4, dport %d, "
 601				   "sk_err=%d/%d\n", dlm_our_nodeid(),
 602				   con->nodeid, &inet->inet_daddr,
 603				   ntohs(inet->inet_dport), sk->sk_err,
 604				   READ_ONCE(sk->sk_err_soft));
 605		break;
 606#if IS_ENABLED(CONFIG_IPV6)
 607	case AF_INET6:
 608		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
 609				   "sending to node %d at %pI6c, "
 610				   "dport %d, sk_err=%d/%d\n", dlm_our_nodeid(),
 611				   con->nodeid, &sk->sk_v6_daddr,
 612				   ntohs(inet->inet_dport), sk->sk_err,
 613				   READ_ONCE(sk->sk_err_soft));
 614		break;
 615#endif
 616	default:
 617		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
 618				   "invalid socket family %d set, "
 619				   "sk_err=%d/%d\n", dlm_our_nodeid(),
 620				   sk->sk_family, sk->sk_err,
 621				   READ_ONCE(sk->sk_err_soft));
 622		break;
 
 
 
 
 
 
 
 
 
 
 
 
 623	}
 
 
 
 
 
 624
 625	dlm_midcomms_unack_msg_resend(con->nodeid);
 
 
 
 626
 627	listen_sock.sk_error_report(sk);
 
 
 
 628}
 629
 630static void restore_callbacks(struct sock *sk)
 631{
 632#ifdef CONFIG_LOCKDEP
 633	WARN_ON_ONCE(!lockdep_sock_is_held(sk));
 634#endif
 635
 
 636	sk->sk_user_data = NULL;
 637	sk->sk_data_ready = listen_sock.sk_data_ready;
 638	sk->sk_state_change = listen_sock.sk_state_change;
 639	sk->sk_write_space = listen_sock.sk_write_space;
 640	sk->sk_error_report = listen_sock.sk_error_report;
 
 641}
 642
 643/* Make a socket active */
 644static void add_sock(struct socket *sock, struct connection *con)
 645{
 646	struct sock *sk = sock->sk;
 647
 648	lock_sock(sk);
 649	con->sock = sock;
 650
 651	sk->sk_user_data = con;
 
 652	sk->sk_data_ready = lowcomms_data_ready;
 653	sk->sk_write_space = lowcomms_write_space;
 654	if (dlm_config.ci_protocol == DLM_PROTO_SCTP)
 655		sk->sk_state_change = lowcomms_state_change;
 656	sk->sk_allocation = GFP_NOFS;
 657	sk->sk_use_task_frag = false;
 658	sk->sk_error_report = lowcomms_error_report;
 659	release_sock(sk);
 660}
 661
 662/* Add the port number to an IPv6 or 4 sockaddr and return the address
 663   length */
 664static void make_sockaddr(struct sockaddr_storage *saddr, __be16 port,
 665			  int *addr_len)
 666{
 667	saddr->ss_family =  dlm_local_addr[0].ss_family;
 668	if (saddr->ss_family == AF_INET) {
 669		struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
 670		in4_addr->sin_port = port;
 671		*addr_len = sizeof(struct sockaddr_in);
 672		memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
 673	} else {
 674		struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
 675		in6_addr->sin6_port = port;
 676		*addr_len = sizeof(struct sockaddr_in6);
 677	}
 678	memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
 679}
 680
 681static void dlm_page_release(struct kref *kref)
 682{
 683	struct writequeue_entry *e = container_of(kref, struct writequeue_entry,
 684						  ref);
 685
 686	__free_page(e->page);
 687	dlm_free_writequeue(e);
 688}
 689
 690static void dlm_msg_release(struct kref *kref)
 691{
 692	struct dlm_msg *msg = container_of(kref, struct dlm_msg, ref);
 693
 694	kref_put(&msg->entry->ref, dlm_page_release);
 695	dlm_free_msg(msg);
 696}
 697
 698static void free_entry(struct writequeue_entry *e)
 699{
 700	struct dlm_msg *msg, *tmp;
 701
 702	list_for_each_entry_safe(msg, tmp, &e->msgs, list) {
 703		if (msg->orig_msg) {
 704			msg->orig_msg->retransmit = false;
 705			kref_put(&msg->orig_msg->ref, dlm_msg_release);
 706		}
 707
 708		list_del(&msg->list);
 709		kref_put(&msg->ref, dlm_msg_release);
 710	}
 711
 712	list_del(&e->list);
 713	kref_put(&e->ref, dlm_page_release);
 714}
 715
 716static void dlm_close_sock(struct socket **sock)
 717{
 718	lock_sock((*sock)->sk);
 719	restore_callbacks((*sock)->sk);
 720	release_sock((*sock)->sk);
 721
 722	sock_release(*sock);
 723	*sock = NULL;
 724}
 725
 726static void allow_connection_io(struct connection *con)
 727{
 728	if (con->othercon)
 729		clear_bit(CF_IO_STOP, &con->othercon->flags);
 730	clear_bit(CF_IO_STOP, &con->flags);
 731}
 732
 733static void stop_connection_io(struct connection *con)
 734{
 735	if (con->othercon)
 736		stop_connection_io(con->othercon);
 737
 738	spin_lock_bh(&con->writequeue_lock);
 739	set_bit(CF_IO_STOP, &con->flags);
 740	spin_unlock_bh(&con->writequeue_lock);
 741
 742	down_write(&con->sock_lock);
 743	if (con->sock) {
 744		lock_sock(con->sock->sk);
 745		restore_callbacks(con->sock->sk);
 746		release_sock(con->sock->sk);
 747	}
 748	up_write(&con->sock_lock);
 749
 750	cancel_work_sync(&con->swork);
 751	cancel_work_sync(&con->rwork);
 752}
 753
 754/* Close a remote connection and tidy up */
 755static void close_connection(struct connection *con, bool and_other)
 
 756{
 757	struct writequeue_entry *e;
 758
 759	if (con->othercon && and_other)
 760		close_connection(con->othercon, false);
 761
 762	down_write(&con->sock_lock);
 763	if (!con->sock) {
 764		up_write(&con->sock_lock);
 765		return;
 
 
 
 766	}
 767
 768	dlm_close_sock(&con->sock);
 769
 770	/* if we send a writequeue entry only a half way, we drop the
 771	 * whole entry because reconnection and that we not start of the
 772	 * middle of a msg which will confuse the other end.
 773	 *
 774	 * we can always drop messages because retransmits, but what we
 775	 * cannot allow is to transmit half messages which may be processed
 776	 * at the other side.
 777	 *
 778	 * our policy is to start on a clean state when disconnects, we don't
 779	 * know what's send/received on transport layer in this case.
 780	 */
 781	spin_lock_bh(&con->writequeue_lock);
 782	if (!list_empty(&con->writequeue)) {
 783		e = list_first_entry(&con->writequeue, struct writequeue_entry,
 784				     list);
 785		if (e->dirty)
 786			free_entry(e);
 787	}
 788	spin_unlock_bh(&con->writequeue_lock);
 789
 790	con->rx_leftover = 0;
 791	con->retries = 0;
 792	clear_bit(CF_APP_LIMITED, &con->flags);
 793	clear_bit(CF_RECV_PENDING, &con->flags);
 794	clear_bit(CF_SEND_PENDING, &con->flags);
 795	up_write(&con->sock_lock);
 796}
 797
 798static void shutdown_connection(struct connection *con, bool and_other)
 799{
 800	int ret;
 801
 802	if (con->othercon && and_other)
 803		shutdown_connection(con->othercon, false);
 
 
 804
 805	flush_workqueue(io_workqueue);
 806	down_read(&con->sock_lock);
 807	/* nothing to shutdown */
 808	if (!con->sock) {
 809		up_read(&con->sock_lock);
 810		return;
 811	}
 812
 
 813	ret = kernel_sock_shutdown(con->sock, SHUT_WR);
 814	up_read(&con->sock_lock);
 815	if (ret) {
 816		log_print("Connection %p failed to shutdown: %d will force close",
 817			  con, ret);
 818		goto force_close;
 819	} else {
 820		ret = wait_event_timeout(con->shutdown_wait, !con->sock,
 
 821					 DLM_SHUTDOWN_WAIT_TIMEOUT);
 822		if (ret == 0) {
 823			log_print("Connection %p shutdown timed out, will force close",
 824				  con);
 825			goto force_close;
 826		}
 827	}
 828
 829	return;
 830
 831force_close:
 832	close_connection(con, false);
 
 833}
 834
 835static struct processqueue_entry *new_processqueue_entry(int nodeid,
 836							 int buflen)
 837{
 838	struct processqueue_entry *pentry;
 839
 840	pentry = kmalloc(sizeof(*pentry), GFP_NOFS);
 841	if (!pentry)
 842		return NULL;
 843
 844	pentry->buf = kmalloc(buflen, GFP_NOFS);
 845	if (!pentry->buf) {
 846		kfree(pentry);
 847		return NULL;
 848	}
 849
 850	pentry->nodeid = nodeid;
 851	return pentry;
 852}
 853
 854static void free_processqueue_entry(struct processqueue_entry *pentry)
 
 855{
 856	kfree(pentry->buf);
 857	kfree(pentry);
 858}
 
 
 
 
 859
 860static void process_dlm_messages(struct work_struct *work)
 861{
 862	struct processqueue_entry *pentry;
 863
 864	spin_lock_bh(&processqueue_lock);
 865	pentry = list_first_entry_or_null(&processqueue,
 866					  struct processqueue_entry, list);
 867	if (WARN_ON_ONCE(!pentry)) {
 868		process_dlm_messages_pending = false;
 869		spin_unlock_bh(&processqueue_lock);
 870		return;
 871	}
 872
 873	list_del(&pentry->list);
 874	if (atomic_dec_and_test(&processqueue_count))
 875		wake_up(&processqueue_wq);
 876	spin_unlock_bh(&processqueue_lock);
 877
 878	for (;;) {
 879		dlm_process_incoming_buffer(pentry->nodeid, pentry->buf,
 880					    pentry->buflen);
 881		free_processqueue_entry(pentry);
 882
 883		spin_lock_bh(&processqueue_lock);
 884		pentry = list_first_entry_or_null(&processqueue,
 885						  struct processqueue_entry, list);
 886		if (!pentry) {
 887			process_dlm_messages_pending = false;
 888			spin_unlock_bh(&processqueue_lock);
 889			break;
 890		}
 891
 892		list_del(&pentry->list);
 893		if (atomic_dec_and_test(&processqueue_count))
 894			wake_up(&processqueue_wq);
 895		spin_unlock_bh(&processqueue_lock);
 896	}
 897}
 898
 899/* Data received from remote end */
 900static int receive_from_sock(struct connection *con, int buflen)
 901{
 902	struct processqueue_entry *pentry;
 903	int ret, buflen_real;
 904	struct msghdr msg;
 905	struct kvec iov;
 906
 907	pentry = new_processqueue_entry(con->nodeid, buflen);
 908	if (!pentry)
 909		return DLM_IO_RESCHED;
 910
 911	memcpy(pentry->buf, con->rx_leftover_buf, con->rx_leftover);
 912
 913	/* calculate new buffer parameter regarding last receive and
 914	 * possible leftover bytes
 
 915	 */
 916	iov.iov_base = pentry->buf + con->rx_leftover;
 917	iov.iov_len = buflen - con->rx_leftover;
 918
 919	memset(&msg, 0, sizeof(msg));
 920	msg.msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
 921	clear_bit(CF_RECV_INTR, &con->flags);
 922again:
 923	ret = kernel_recvmsg(con->sock, &msg, &iov, 1, iov.iov_len,
 924			     msg.msg_flags);
 925	trace_dlm_recv(con->nodeid, ret);
 926	if (ret == -EAGAIN) {
 927		lock_sock(con->sock->sk);
 928		if (test_and_clear_bit(CF_RECV_INTR, &con->flags)) {
 929			release_sock(con->sock->sk);
 930			goto again;
 931		}
 932
 933		clear_bit(CF_RECV_PENDING, &con->flags);
 934		release_sock(con->sock->sk);
 935		free_processqueue_entry(pentry);
 936		return DLM_IO_END;
 937	} else if (ret == 0) {
 938		/* close will clear CF_RECV_PENDING */
 939		free_processqueue_entry(pentry);
 940		return DLM_IO_EOF;
 941	} else if (ret < 0) {
 942		free_processqueue_entry(pentry);
 943		return ret;
 944	}
 945
 946	/* new buflen according readed bytes and leftover from last receive */
 947	buflen_real = ret + con->rx_leftover;
 948	ret = dlm_validate_incoming_buffer(con->nodeid, pentry->buf,
 949					   buflen_real);
 950	if (ret < 0) {
 951		free_processqueue_entry(pentry);
 952		return ret;
 
 
 
 
 953	}
 954
 955	pentry->buflen = ret;
 956
 957	/* calculate leftover bytes from process and put it into begin of
 958	 * the receive buffer, so next receive we have the full message
 959	 * at the start address of the receive buffer.
 960	 */
 961	con->rx_leftover = buflen_real - ret;
 962	memmove(con->rx_leftover_buf, pentry->buf + ret,
 963		con->rx_leftover);
 964
 965	spin_lock_bh(&processqueue_lock);
 966	ret = atomic_inc_return(&processqueue_count);
 967	list_add_tail(&pentry->list, &processqueue);
 968	if (!process_dlm_messages_pending) {
 969		process_dlm_messages_pending = true;
 970		queue_work(process_workqueue, &process_work);
 971	}
 972	spin_unlock_bh(&processqueue_lock);
 973
 974	if (ret > DLM_MAX_PROCESS_BUFFERS)
 975		return DLM_IO_FLUSH;
 
 
 976
 977	return DLM_IO_SUCCESS;
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 978}
 979
 980/* Listening socket is busy, accept a connection */
 981static int accept_from_sock(void)
 982{
 
 983	struct sockaddr_storage peeraddr;
 984	int len, idx, result, nodeid;
 985	struct connection *newcon;
 986	struct socket *newsock;
 987	unsigned int mark;
 
 
 
 988
 989	result = kernel_accept(listen_con.sock, &newsock, O_NONBLOCK);
 990	if (result == -EAGAIN)
 991		return DLM_IO_END;
 992	else if (result < 0)
 
 
 
 
 
 
 
 
 
 
 
 
 993		goto accept_err;
 994
 995	/* Get the connected socket's peer */
 996	memset(&peeraddr, 0, sizeof(peeraddr));
 997	len = newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 2);
 998	if (len < 0) {
 999		result = -ECONNABORTED;
1000		goto accept_err;
1001	}
1002
1003	/* Get the new node's NODEID */
1004	make_sockaddr(&peeraddr, 0, &len);
1005	if (addr_to_nodeid(&peeraddr, &nodeid, &mark)) {
1006		switch (peeraddr.ss_family) {
1007		case AF_INET: {
1008			struct sockaddr_in *sin = (struct sockaddr_in *)&peeraddr;
1009
1010			log_print("connect from non cluster IPv4 node %pI4",
1011				  &sin->sin_addr);
1012			break;
1013		}
1014#if IS_ENABLED(CONFIG_IPV6)
1015		case AF_INET6: {
1016			struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&peeraddr;
1017
1018			log_print("connect from non cluster IPv6 node %pI6c",
1019				  &sin6->sin6_addr);
1020			break;
1021		}
1022#endif
1023		default:
1024			log_print("invalid family from non cluster node");
1025			break;
1026		}
1027
1028		sock_release(newsock);
 
1029		return -1;
1030	}
1031
1032	log_print("got connection from %d", nodeid);
1033
1034	/*  Check to see if we already have a connection to this node. This
1035	 *  could happen if the two nodes initiate a connection at roughly
1036	 *  the same time and the connections cross on the wire.
1037	 *  In this case we store the incoming one in "othercon"
1038	 */
1039	idx = srcu_read_lock(&connections_srcu);
1040	newcon = nodeid2con(nodeid, 0);
1041	if (WARN_ON_ONCE(!newcon)) {
1042		srcu_read_unlock(&connections_srcu, idx);
1043		result = -ENOENT;
1044		goto accept_err;
1045	}
1046
1047	sock_set_mark(newsock->sk, mark);
1048
1049	down_write(&newcon->sock_lock);
1050	if (newcon->sock) {
1051		struct connection *othercon = newcon->othercon;
1052
1053		if (!othercon) {
1054			othercon = kzalloc(sizeof(*othercon), GFP_NOFS);
1055			if (!othercon) {
1056				log_print("failed to allocate incoming socket");
1057				up_write(&newcon->sock_lock);
1058				srcu_read_unlock(&connections_srcu, idx);
1059				result = -ENOMEM;
1060				goto accept_err;
1061			}
1062
1063			dlm_con_init(othercon, nodeid);
1064			lockdep_set_subclass(&othercon->sock_lock, 1);
1065			newcon->othercon = othercon;
 
 
 
 
1066			set_bit(CF_IS_OTHERCON, &othercon->flags);
1067		} else {
1068			/* close other sock con if we have something new */
1069			close_connection(othercon, false);
1070		}
1071
1072		down_write(&othercon->sock_lock);
 
1073		add_sock(newsock, othercon);
1074
1075		/* check if we receved something while adding */
1076		lock_sock(othercon->sock->sk);
1077		lowcomms_queue_rwork(othercon);
1078		release_sock(othercon->sock->sk);
1079		up_write(&othercon->sock_lock);
1080	}
1081	else {
 
1082		/* accept copies the sk after we've saved the callbacks, so we
1083		   don't want to save them a second time or comm errors will
1084		   result in calling sk_error_report recursively. */
1085		add_sock(newsock, newcon);
1086
1087		/* check if we receved something while adding */
1088		lock_sock(newcon->sock->sk);
1089		lowcomms_queue_rwork(newcon);
1090		release_sock(newcon->sock->sk);
1091	}
1092	up_write(&newcon->sock_lock);
1093	srcu_read_unlock(&connections_srcu, idx);
1094
1095	return DLM_IO_SUCCESS;
 
 
 
 
 
 
 
 
 
 
 
1096
1097accept_err:
 
1098	if (newsock)
1099		sock_release(newsock);
1100
 
 
1101	return result;
1102}
1103
 
 
 
 
 
 
1104/*
1105 * writequeue_entry_complete - try to delete and free write queue entry
1106 * @e: write queue entry to try to delete
1107 * @completed: bytes completed
1108 *
1109 * writequeue_lock must be held.
1110 */
1111static void writequeue_entry_complete(struct writequeue_entry *e, int completed)
1112{
1113	e->offset += completed;
1114	e->len -= completed;
1115	/* signal that page was half way transmitted */
1116	e->dirty = true;
1117
1118	if (e->len == 0 && e->users == 0)
 
1119		free_entry(e);
 
1120}
1121
1122/*
1123 * sctp_bind_addrs - bind a SCTP socket to all our addresses
1124 */
1125static int sctp_bind_addrs(struct socket *sock, __be16 port)
1126{
1127	struct sockaddr_storage localaddr;
1128	struct sockaddr *addr = (struct sockaddr *)&localaddr;
1129	int i, addr_len, result = 0;
1130
1131	for (i = 0; i < dlm_local_count; i++) {
1132		memcpy(&localaddr, &dlm_local_addr[i], sizeof(localaddr));
1133		make_sockaddr(&localaddr, port, &addr_len);
1134
1135		if (!i)
1136			result = kernel_bind(sock, addr, addr_len);
1137		else
1138			result = sock_bind_add(sock->sk, addr, addr_len);
1139
1140		if (result < 0) {
1141			log_print("Can't bind to %d addr number %d, %d.\n",
1142				  port, i + 1, result);
1143			break;
1144		}
1145	}
1146	return result;
1147}
1148
1149/* Get local addresses */
1150static void init_local(void)
1151{
1152	struct sockaddr_storage sas;
1153	int i;
1154
1155	dlm_local_count = 0;
1156	for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
1157		if (dlm_our_addr(&sas, i))
1158			break;
1159
1160		memcpy(&dlm_local_addr[dlm_local_count++], &sas, sizeof(sas));
1161	}
1162}
1163
1164static struct writequeue_entry *new_writequeue_entry(struct connection *con)
1165{
1166	struct writequeue_entry *entry;
1167
1168	entry = dlm_allocate_writequeue();
1169	if (!entry)
1170		return NULL;
1171
1172	entry->page = alloc_page(GFP_ATOMIC | __GFP_ZERO);
1173	if (!entry->page) {
1174		dlm_free_writequeue(entry);
1175		return NULL;
1176	}
1177
1178	entry->offset = 0;
1179	entry->len = 0;
1180	entry->end = 0;
1181	entry->dirty = false;
1182	entry->con = con;
1183	entry->users = 1;
1184	kref_init(&entry->ref);
1185	return entry;
1186}
1187
1188static struct writequeue_entry *new_wq_entry(struct connection *con, int len,
1189					     char **ppc, void (*cb)(void *data),
1190					     void *data)
1191{
1192	struct writequeue_entry *e;
1193
1194	spin_lock_bh(&con->writequeue_lock);
1195	if (!list_empty(&con->writequeue)) {
1196		e = list_last_entry(&con->writequeue, struct writequeue_entry, list);
1197		if (DLM_WQ_REMAIN_BYTES(e) >= len) {
1198			kref_get(&e->ref);
1199
1200			*ppc = page_address(e->page) + e->end;
1201			if (cb)
1202				cb(data);
1203
1204			e->end += len;
1205			e->users++;
1206			goto out;
1207		}
1208	}
1209
1210	e = new_writequeue_entry(con);
1211	if (!e)
 
 
1212		goto out;
 
1213
1214	kref_get(&e->ref);
1215	*ppc = page_address(e->page);
1216	e->end += len;
1217	if (cb)
1218		cb(data);
1219
1220	list_add_tail(&e->list, &con->writequeue);
1221
1222out:
1223	spin_unlock_bh(&con->writequeue_lock);
1224	return e;
1225};
1226
1227static struct dlm_msg *dlm_lowcomms_new_msg_con(struct connection *con, int len,
1228						char **ppc, void (*cb)(void *data),
1229						void *data)
1230{
1231	struct writequeue_entry *e;
1232	struct dlm_msg *msg;
1233
1234	msg = dlm_allocate_msg();
1235	if (!msg)
1236		return NULL;
1237
1238	kref_init(&msg->ref);
 
 
1239
1240	e = new_wq_entry(con, len, ppc, cb, data);
1241	if (!e) {
1242		dlm_free_msg(msg);
1243		return NULL;
1244	}
1245
1246	msg->retransmit = false;
1247	msg->orig_msg = NULL;
1248	msg->ppc = *ppc;
1249	msg->len = len;
1250	msg->entry = e;
1251
1252	return msg;
1253}
1254
1255/* avoid false positive for nodes_srcu, unlock happens in
1256 * dlm_lowcomms_commit_msg which is a must call if success
1257 */
1258#ifndef __CHECKER__
1259struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, char **ppc,
1260				     void (*cb)(void *data), void *data)
1261{
1262	struct connection *con;
1263	struct dlm_msg *msg;
1264	int idx;
1265
1266	if (len > DLM_MAX_SOCKET_BUFSIZE ||
1267	    len < sizeof(struct dlm_header)) {
1268		BUILD_BUG_ON(PAGE_SIZE < DLM_MAX_SOCKET_BUFSIZE);
1269		log_print("failed to allocate a buffer of size %d", len);
1270		WARN_ON_ONCE(1);
1271		return NULL;
1272	}
 
 
 
 
 
 
 
1273
1274	idx = srcu_read_lock(&connections_srcu);
1275	con = nodeid2con(nodeid, 0);
1276	if (WARN_ON_ONCE(!con)) {
1277		srcu_read_unlock(&connections_srcu, idx);
1278		return NULL;
1279	}
1280
1281	msg = dlm_lowcomms_new_msg_con(con, len, ppc, cb, data);
1282	if (!msg) {
1283		srcu_read_unlock(&connections_srcu, idx);
1284		return NULL;
 
 
 
 
 
 
 
 
 
 
 
 
1285	}
1286
1287	/* for dlm_lowcomms_commit_msg() */
1288	kref_get(&msg->ref);
1289	/* we assume if successful commit must called */
1290	msg->idx = idx;
1291	return msg;
1292}
1293#endif
1294
1295static void _dlm_lowcomms_commit_msg(struct dlm_msg *msg)
 
1296{
1297	struct writequeue_entry *e = msg->entry;
1298	struct connection *con = e->con;
1299	int users;
 
 
1300
1301	spin_lock_bh(&con->writequeue_lock);
1302	kref_get(&msg->ref);
1303	list_add(&msg->list, &e->msgs);
 
1304
1305	users = --e->users;
1306	if (users)
1307		goto out;
1308
1309	e->len = DLM_WQ_LENGTH_BYTES(e);
1310
1311	lowcomms_queue_swork(con);
1312
1313out:
1314	spin_unlock_bh(&con->writequeue_lock);
1315	return;
1316}
 
1317
1318/* avoid false positive for nodes_srcu, lock was happen in
1319 * dlm_lowcomms_new_msg
1320 */
1321#ifndef __CHECKER__
1322void dlm_lowcomms_commit_msg(struct dlm_msg *msg)
1323{
1324	_dlm_lowcomms_commit_msg(msg);
1325	srcu_read_unlock(&connections_srcu, msg->idx);
1326	/* because dlm_lowcomms_new_msg() */
1327	kref_put(&msg->ref, dlm_msg_release);
1328}
1329#endif
1330
1331void dlm_lowcomms_put_msg(struct dlm_msg *msg)
1332{
1333	kref_put(&msg->ref, dlm_msg_release);
1334}
1335
1336/* does not held connections_srcu, usage lowcomms_error_report only */
1337int dlm_lowcomms_resend_msg(struct dlm_msg *msg)
1338{
1339	struct dlm_msg *msg_resend;
1340	char *ppc;
 
1341
1342	if (msg->retransmit)
1343		return 1;
 
 
1344
1345	msg_resend = dlm_lowcomms_new_msg_con(msg->entry->con, msg->len, &ppc,
1346					      NULL, NULL);
1347	if (!msg_resend)
1348		return -ENOMEM;
 
 
 
 
 
 
1349
1350	msg->retransmit = true;
1351	kref_get(&msg->ref);
1352	msg_resend->orig_msg = msg;
1353
1354	memcpy(ppc, msg->ppc, msg->len);
1355	_dlm_lowcomms_commit_msg(msg_resend);
1356	dlm_lowcomms_put_msg(msg_resend);
1357
1358	return 0;
1359}
1360
1361/* Send a message */
1362static int send_to_sock(struct connection *con)
1363{
1364	struct writequeue_entry *e;
1365	struct bio_vec bvec;
1366	struct msghdr msg = {
1367		.msg_flags = MSG_SPLICE_PAGES | MSG_DONTWAIT | MSG_NOSIGNAL,
1368	};
1369	int len, offset, ret;
1370
1371	spin_lock_bh(&con->writequeue_lock);
1372	e = con_next_wq(con);
1373	if (!e) {
1374		clear_bit(CF_SEND_PENDING, &con->flags);
1375		spin_unlock_bh(&con->writequeue_lock);
1376		return DLM_IO_END;
1377	}
1378
1379	len = e->len;
1380	offset = e->offset;
1381	WARN_ON_ONCE(len == 0 && e->users == 0);
1382	spin_unlock_bh(&con->writequeue_lock);
1383
1384	bvec_set_page(&bvec, e->page, len, offset);
1385	iov_iter_bvec(&msg.msg_iter, ITER_SOURCE, &bvec, 1, len);
1386	ret = sock_sendmsg(con->sock, &msg);
1387	trace_dlm_send(con->nodeid, ret);
1388	if (ret == -EAGAIN || ret == 0) {
1389		lock_sock(con->sock->sk);
1390		spin_lock_bh(&con->writequeue_lock);
1391		if (test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) &&
1392		    !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1393			/* Notify TCP that we're limited by the
1394			 * application window size.
1395			 */
1396			set_bit(SOCK_NOSPACE, &con->sock->sk->sk_socket->flags);
1397			con->sock->sk->sk_write_pending++;
1398
1399			clear_bit(CF_SEND_PENDING, &con->flags);
1400			spin_unlock_bh(&con->writequeue_lock);
1401			release_sock(con->sock->sk);
1402
1403			/* wait for write_space() event */
1404			return DLM_IO_END;
1405		}
1406		spin_unlock_bh(&con->writequeue_lock);
1407		release_sock(con->sock->sk);
 
1408
1409		return DLM_IO_RESCHED;
1410	} else if (ret < 0) {
1411		return ret;
 
 
 
1412	}
1413
1414	spin_lock_bh(&con->writequeue_lock);
1415	writequeue_entry_complete(e, ret);
1416	spin_unlock_bh(&con->writequeue_lock);
1417
1418	return DLM_IO_SUCCESS;
 
 
 
 
 
 
 
 
 
 
 
 
 
1419}
1420
1421static void clean_one_writequeue(struct connection *con)
 
1422{
1423	struct writequeue_entry *e, *safe;
 
 
1424
1425	spin_lock_bh(&con->writequeue_lock);
1426	list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1427		free_entry(e);
 
 
 
 
 
 
 
 
1428	}
1429	spin_unlock_bh(&con->writequeue_lock);
1430}
1431
1432static void connection_release(struct rcu_head *rcu)
1433{
1434	struct connection *con = container_of(rcu, struct connection, rcu);
1435
1436	WARN_ON_ONCE(!list_empty(&con->writequeue));
1437	WARN_ON_ONCE(con->sock);
1438	kfree(con);
1439}
1440
1441/* Called from recovery when it knows that a node has
1442   left the cluster */
1443int dlm_lowcomms_close(int nodeid)
1444{
1445	struct connection *con;
1446	int idx;
1447
1448	log_print("closing connection to node %d", nodeid);
 
 
 
 
 
1449
1450	idx = srcu_read_lock(&connections_srcu);
1451	con = nodeid2con(nodeid, 0);
1452	if (WARN_ON_ONCE(!con)) {
1453		srcu_read_unlock(&connections_srcu, idx);
1454		return -ENOENT;
 
 
 
 
1455	}
 
1456
1457	stop_connection_io(con);
1458	log_print("io handling for node: %d stopped", nodeid);
1459	close_connection(con, true);
1460
1461	spin_lock(&connections_lock);
1462	hlist_del_rcu(&con->list);
1463	spin_unlock(&connections_lock);
1464
1465	clean_one_writequeue(con);
1466	call_srcu(&connections_srcu, &con->rcu, connection_release);
1467	if (con->othercon) {
1468		clean_one_writequeue(con->othercon);
1469		call_srcu(&connections_srcu, &con->othercon->rcu, connection_release);
1470	}
1471	srcu_read_unlock(&connections_srcu, idx);
1472
1473	/* for debugging we print when we are done to compare with other
1474	 * messages in between. This function need to be correctly synchronized
1475	 * with io handling
1476	 */
1477	log_print("closing connection to node %d done", nodeid);
1478
1479	return 0;
 
1480}
1481
1482/* Receive worker function */
1483static void process_recv_sockets(struct work_struct *work)
1484{
1485	struct connection *con = container_of(work, struct connection, rwork);
1486	int ret, buflen;
1487
1488	down_read(&con->sock_lock);
1489	if (!con->sock) {
1490		up_read(&con->sock_lock);
1491		return;
1492	}
1493
1494	buflen = READ_ONCE(dlm_config.ci_buffer_size);
1495	do {
1496		ret = receive_from_sock(con, buflen);
1497	} while (ret == DLM_IO_SUCCESS);
1498	up_read(&con->sock_lock);
1499
1500	switch (ret) {
1501	case DLM_IO_END:
1502		/* CF_RECV_PENDING cleared */
1503		break;
1504	case DLM_IO_EOF:
1505		close_connection(con, false);
1506		wake_up(&con->shutdown_wait);
1507		/* CF_RECV_PENDING cleared */
1508		break;
1509	case DLM_IO_FLUSH:
1510		/* we can't flush the process_workqueue here because a
1511		 * WQ_MEM_RECLAIM workequeue can occurr a deadlock for a non
1512		 * WQ_MEM_RECLAIM workqueue such as process_workqueue. Instead
1513		 * we have a waitqueue to wait until all messages are
1514		 * processed.
1515		 *
1516		 * This handling is only necessary to backoff the sender and
1517		 * not queue all messages from the socket layer into DLM
1518		 * processqueue. When DLM is capable to parse multiple messages
1519		 * on an e.g. per socket basis this handling can might be
1520		 * removed. Especially in a message burst we are too slow to
1521		 * process messages and the queue will fill up memory.
1522		 */
1523		wait_event(processqueue_wq, !atomic_read(&processqueue_count));
1524		fallthrough;
1525	case DLM_IO_RESCHED:
1526		cond_resched();
1527		queue_work(io_workqueue, &con->rwork);
1528		/* CF_RECV_PENDING not cleared */
1529		break;
1530	default:
1531		if (ret < 0) {
1532			if (test_bit(CF_IS_OTHERCON, &con->flags)) {
1533				close_connection(con, false);
1534			} else {
1535				spin_lock_bh(&con->writequeue_lock);
1536				lowcomms_queue_swork(con);
1537				spin_unlock_bh(&con->writequeue_lock);
1538			}
1539
1540			/* CF_RECV_PENDING cleared for othercon
1541			 * we trigger send queue if not already done
1542			 * and process_send_sockets will handle it
1543			 */
1544			break;
1545		}
1546
1547		WARN_ON_ONCE(1);
1548		break;
 
 
1549	}
1550}
1551
1552static void process_listen_recv_socket(struct work_struct *work)
 
1553{
1554	int ret;
1555
1556	if (WARN_ON_ONCE(!listen_con.sock))
1557		return;
1558
1559	do {
1560		ret = accept_from_sock();
1561	} while (ret == DLM_IO_SUCCESS);
1562
1563	if (ret < 0)
1564		log_print("critical error accepting connection: %d", ret);
1565}
1566
1567static int dlm_connect(struct connection *con)
1568{
1569	struct sockaddr_storage addr;
1570	int result, addr_len;
1571	struct socket *sock;
1572	unsigned int mark;
1573
1574	memset(&addr, 0, sizeof(addr));
1575	result = nodeid_to_addr(con->nodeid, &addr, NULL,
1576				dlm_proto_ops->try_new_addr, &mark);
1577	if (result < 0) {
1578		log_print("no address for nodeid %d", con->nodeid);
1579		return result;
1580	}
1581
1582	/* Create a socket to communicate with */
1583	result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family,
1584				  SOCK_STREAM, dlm_proto_ops->proto, &sock);
1585	if (result < 0)
1586		return result;
1587
1588	sock_set_mark(sock->sk, mark);
1589	dlm_proto_ops->sockopts(sock);
1590
1591	result = dlm_proto_ops->bind(sock);
1592	if (result < 0) {
1593		sock_release(sock);
1594		return result;
1595	}
 
 
 
1596
1597	add_sock(sock, con);
1598
1599	log_print_ratelimited("connecting to %d", con->nodeid);
1600	make_sockaddr(&addr, dlm_config.ci_tcp_port, &addr_len);
1601	result = kernel_connect(sock, (struct sockaddr *)&addr, addr_len, 0);
1602	switch (result) {
1603	case -EINPROGRESS:
1604		/* not an error */
1605		fallthrough;
1606	case 0:
1607		break;
1608	default:
1609		if (result < 0)
1610			dlm_close_sock(&con->sock);
1611
1612		break;
 
 
 
1613	}
1614
 
 
 
 
 
 
1615	return result;
1616}
1617
1618/* Send worker function */
1619static void process_send_sockets(struct work_struct *work)
1620{
1621	struct connection *con = container_of(work, struct connection, swork);
1622	int ret;
 
1623
1624	WARN_ON_ONCE(test_bit(CF_IS_OTHERCON, &con->flags));
 
1625
1626	down_read(&con->sock_lock);
1627	if (!con->sock) {
1628		up_read(&con->sock_lock);
1629		down_write(&con->sock_lock);
1630		if (!con->sock) {
1631			ret = dlm_connect(con);
1632			switch (ret) {
1633			case 0:
1634				break;
1635			default:
1636				/* CF_SEND_PENDING not cleared */
1637				up_write(&con->sock_lock);
1638				log_print("connect to node %d try %d error %d",
1639					  con->nodeid, con->retries++, ret);
1640				msleep(1000);
1641				/* For now we try forever to reconnect. In
1642				 * future we should send a event to cluster
1643				 * manager to fence itself after certain amount
1644				 * of retries.
1645				 */
1646				queue_work(io_workqueue, &con->swork);
1647				return;
1648			}
1649		}
1650		downgrade_write(&con->sock_lock);
1651	}
1652
1653	do {
1654		ret = send_to_sock(con);
1655	} while (ret == DLM_IO_SUCCESS);
1656	up_read(&con->sock_lock);
1657
1658	switch (ret) {
1659	case DLM_IO_END:
1660		/* CF_SEND_PENDING cleared */
1661		break;
1662	case DLM_IO_RESCHED:
1663		/* CF_SEND_PENDING not cleared */
1664		cond_resched();
1665		queue_work(io_workqueue, &con->swork);
1666		break;
1667	default:
1668		if (ret < 0) {
1669			close_connection(con, false);
1670
1671			/* CF_SEND_PENDING cleared */
1672			spin_lock_bh(&con->writequeue_lock);
1673			lowcomms_queue_swork(con);
1674			spin_unlock_bh(&con->writequeue_lock);
1675			break;
1676		}
1677
1678		WARN_ON_ONCE(1);
1679		break;
 
 
1680	}
1681}
1682
1683static void work_stop(void)
1684{
1685	if (io_workqueue) {
1686		destroy_workqueue(io_workqueue);
1687		io_workqueue = NULL;
1688	}
1689
1690	if (process_workqueue) {
1691		destroy_workqueue(process_workqueue);
1692		process_workqueue = NULL;
1693	}
1694}
1695
1696static int work_start(void)
 
 
 
1697{
1698	io_workqueue = alloc_workqueue("dlm_io", WQ_HIGHPRI | WQ_MEM_RECLAIM |
1699				       WQ_UNBOUND, 0);
1700	if (!io_workqueue) {
1701		log_print("can't start dlm_io");
1702		return -ENOMEM;
1703	}
1704
1705	process_workqueue = alloc_workqueue("dlm_process", WQ_HIGHPRI | WQ_BH, 0);
1706	if (!process_workqueue) {
1707		log_print("can't start dlm_process");
1708		destroy_workqueue(io_workqueue);
1709		io_workqueue = NULL;
1710		return -ENOMEM;
 
 
1711	}
1712
1713	return 0;
 
 
 
 
 
 
1714}
1715
1716void dlm_lowcomms_shutdown(void)
1717{
1718	struct connection *con;
1719	int i, idx;
1720
1721	/* stop lowcomms_listen_data_ready calls */
1722	lock_sock(listen_con.sock->sk);
1723	listen_con.sock->sk->sk_data_ready = listen_sock.sk_data_ready;
1724	release_sock(listen_con.sock->sk);
1725
1726	cancel_work_sync(&listen_con.rwork);
1727	dlm_close_sock(&listen_con.sock);
 
1728
1729	idx = srcu_read_lock(&connections_srcu);
1730	for (i = 0; i < CONN_HASH_SIZE; i++) {
1731		hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
1732			shutdown_connection(con, true);
1733			stop_connection_io(con);
1734			flush_workqueue(process_workqueue);
1735			close_connection(con, true);
1736
1737			clean_one_writequeue(con);
1738			if (con->othercon)
1739				clean_one_writequeue(con->othercon);
1740			allow_connection_io(con);
1741		}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1742	}
1743	srcu_read_unlock(&connections_srcu, idx);
1744}
1745
1746void dlm_lowcomms_stop(void)
1747{
1748	work_stop();
1749	dlm_proto_ops = NULL;
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1750}
1751
1752static int dlm_listen_for_all(void)
 
1753{
1754	struct socket *sock;
1755	int result;
 
 
 
1756
1757	log_print("Using %s for communications",
1758		  dlm_proto_ops->name);
 
1759
1760	result = dlm_proto_ops->listen_validate();
1761	if (result < 0)
1762		return result;
 
 
 
1763
1764	result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family,
1765				  SOCK_STREAM, dlm_proto_ops->proto, &sock);
1766	if (result < 0) {
1767		log_print("Can't create comms socket: %d", result);
1768		return result;
1769	}
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1770
1771	sock_set_mark(sock->sk, dlm_config.ci_mark);
1772	dlm_proto_ops->listen_sockopts(sock);
 
 
 
1773
1774	result = dlm_proto_ops->listen_bind(sock);
1775	if (result < 0)
1776		goto out;
 
 
 
 
1777
1778	lock_sock(sock->sk);
1779	listen_sock.sk_data_ready = sock->sk->sk_data_ready;
1780	listen_sock.sk_write_space = sock->sk->sk_write_space;
1781	listen_sock.sk_error_report = sock->sk->sk_error_report;
1782	listen_sock.sk_state_change = sock->sk->sk_state_change;
1783
1784	listen_con.sock = sock;
1785
1786	sock->sk->sk_allocation = GFP_NOFS;
1787	sock->sk->sk_use_task_frag = false;
1788	sock->sk->sk_data_ready = lowcomms_listen_data_ready;
1789	release_sock(sock->sk);
1790
1791	result = sock->ops->listen(sock, 128);
1792	if (result < 0) {
1793		dlm_close_sock(&listen_con.sock);
1794		return result;
1795	}
1796
1797	return 0;
 
 
1798
1799out:
1800	sock_release(sock);
1801	return result;
 
 
 
1802}
1803
1804static int dlm_tcp_bind(struct socket *sock)
 
 
1805{
1806	struct sockaddr_storage src_addr;
1807	int result, addr_len;
1808
1809	/* Bind to our cluster-known address connecting to avoid
1810	 * routing problems.
1811	 */
1812	memcpy(&src_addr, &dlm_local_addr[0], sizeof(src_addr));
1813	make_sockaddr(&src_addr, 0, &addr_len);
 
 
1814
1815	result = kernel_bind(sock, (struct sockaddr *)&src_addr,
1816			     addr_len);
1817	if (result < 0) {
1818		/* This *may* not indicate a critical error */
1819		log_print("could not bind for connect: %d", result);
 
 
1820	}
 
1821
1822	return 0;
1823}
1824
1825static int dlm_tcp_listen_validate(void)
 
1826{
1827	/* We don't support multi-homed hosts */
1828	if (dlm_local_count > 1) {
1829		log_print("TCP protocol can't handle multi-homed hosts, try SCTP");
1830		return -EINVAL;
1831	}
1832
1833	return 0;
 
 
 
1834}
1835
1836static void dlm_tcp_sockopts(struct socket *sock)
 
1837{
1838	/* Turn off Nagle's algorithm */
1839	tcp_sock_set_nodelay(sock->sk);
 
 
 
 
 
1840}
1841
1842static void dlm_tcp_listen_sockopts(struct socket *sock)
 
 
1843{
1844	dlm_tcp_sockopts(sock);
1845	sock_set_reuseaddr(sock->sk);
1846}
1847
1848static int dlm_tcp_listen_bind(struct socket *sock)
1849{
1850	int addr_len;
1851
1852	/* Bind to our port */
1853	make_sockaddr(&dlm_local_addr[0], dlm_config.ci_tcp_port, &addr_len);
1854	return kernel_bind(sock, (struct sockaddr *)&dlm_local_addr[0],
1855			   addr_len);
1856}
1857
1858static const struct dlm_proto_ops dlm_tcp_ops = {
1859	.name = "TCP",
1860	.proto = IPPROTO_TCP,
1861	.sockopts = dlm_tcp_sockopts,
1862	.bind = dlm_tcp_bind,
1863	.listen_validate = dlm_tcp_listen_validate,
1864	.listen_sockopts = dlm_tcp_listen_sockopts,
1865	.listen_bind = dlm_tcp_listen_bind,
1866};
1867
1868static int dlm_sctp_bind(struct socket *sock)
1869{
1870	return sctp_bind_addrs(sock, 0);
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1871}
1872
1873static int dlm_sctp_listen_validate(void)
1874{
1875	if (!IS_ENABLED(CONFIG_IP_SCTP)) {
1876		log_print("SCTP is not enabled by this kernel");
1877		return -EOPNOTSUPP;
 
 
 
 
 
1878	}
 
 
 
 
1879
1880	request_module("sctp");
1881	return 0;
 
1882}
1883
1884static int dlm_sctp_bind_listen(struct socket *sock)
1885{
1886	return sctp_bind_addrs(sock, dlm_config.ci_tcp_port);
 
1887}
1888
1889static void dlm_sctp_sockopts(struct socket *sock)
1890{
1891	/* Turn off Nagle's algorithm */
1892	sctp_sock_set_nodelay(sock->sk);
1893	sock_set_rcvbuf(sock->sk, NEEDED_RMEM);
 
 
1894}
1895
1896static const struct dlm_proto_ops dlm_sctp_ops = {
1897	.name = "SCTP",
1898	.proto = IPPROTO_SCTP,
1899	.try_new_addr = true,
1900	.sockopts = dlm_sctp_sockopts,
1901	.bind = dlm_sctp_bind,
1902	.listen_validate = dlm_sctp_listen_validate,
1903	.listen_sockopts = dlm_sctp_sockopts,
1904	.listen_bind = dlm_sctp_bind_listen,
1905};
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1906
1907int dlm_lowcomms_start(void)
1908{
1909	int error;
 
 
 
 
 
1910
1911	init_local();
1912	if (!dlm_local_count) {
1913		error = -ENOTCONN;
1914		log_print("no local IP address has been set");
1915		goto fail;
1916	}
1917
 
 
 
 
 
 
 
1918	error = work_start();
1919	if (error)
1920		goto fail;
1921
1922	/* Start listening */
1923	switch (dlm_config.ci_protocol) {
1924	case DLM_PROTO_TCP:
1925		dlm_proto_ops = &dlm_tcp_ops;
1926		break;
1927	case DLM_PROTO_SCTP:
1928		dlm_proto_ops = &dlm_sctp_ops;
1929		break;
1930	default:
1931		log_print("Invalid protocol identifier %d set",
1932			  dlm_config.ci_protocol);
1933		error = -EINVAL;
1934		goto fail_proto_ops;
1935	}
1936
1937	error = dlm_listen_for_all();
 
 
 
 
1938	if (error)
1939		goto fail_listen;
1940
1941	return 0;
1942
1943fail_listen:
1944	dlm_proto_ops = NULL;
1945fail_proto_ops:
1946	work_stop();
 
 
 
 
 
1947fail:
1948	return error;
1949}
1950
1951void dlm_lowcomms_init(void)
1952{
1953	int i;
1954
1955	for (i = 0; i < CONN_HASH_SIZE; i++)
1956		INIT_HLIST_HEAD(&connection_hash[i]);
1957
1958	INIT_WORK(&listen_con.rwork, process_listen_recv_socket);
1959}
1960
1961void dlm_lowcomms_exit(void)
1962{
1963	struct connection *con;
1964	int i, idx;
1965
1966	idx = srcu_read_lock(&connections_srcu);
1967	for (i = 0; i < CONN_HASH_SIZE; i++) {
1968		hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
1969			spin_lock(&connections_lock);
1970			hlist_del_rcu(&con->list);
1971			spin_unlock(&connections_lock);
1972
1973			if (con->othercon)
1974				call_srcu(&connections_srcu, &con->othercon->rcu,
1975					  connection_release);
1976			call_srcu(&connections_srcu, &con->rcu, connection_release);
1977		}
1978	}
1979	srcu_read_unlock(&connections_srcu, idx);
1980}