lowcomms.c 43.2 KB
Newer Older
1 2 3 4
/******************************************************************************
*******************************************************************************
**
**  Copyright (C) Sistina Software, Inc.  1997-2003  All rights reserved.
5
**  Copyright (C) 2004-2009 Red Hat, Inc.  All rights reserved.
6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
**
**  This copyrighted material is made available to anyone wishing to use,
**  modify, copy, or redistribute it subject to the terms and conditions
**  of the GNU General Public License v.2.
**
*******************************************************************************
******************************************************************************/

/*
 * 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
Joe Perches's avatar
Joe Perches committed
24
 * be expanded for the cluster infrastructure then that is its
25 26 27 28 29 30 31 32 33 34 35 36 37 38
 * 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.
 *
Joe Perches's avatar
Joe Perches committed
39
 * lowcomms will choose to use either TCP or SCTP as its transport layer
40
 * depending on the configuration variable 'protocol'. This should be set
Joe Perches's avatar
Joe Perches committed
41
 * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
42 43
 * cluster-wide mechanism as it must be the same on all nodes of the cluster
 * for the DLM to function.
44 45 46 47 48 49 50
 *
 */

#include <asm/ioctls.h>
#include <net/sock.h>
#include <net/tcp.h>
#include <linux/pagemap.h>
51
#include <linux/file.h>
52
#include <linux/mutex.h>
53
#include <linux/sctp.h>
54
#include <linux/slab.h>
55
#include <net/sctp/sctp.h>
Joe Perches's avatar
Joe Perches committed
56
#include <net/ipv6.h>
57 58 59 60 61 62

#include "dlm_internal.h"
#include "lowcomms.h"
#include "midcomms.h"
#include "config.h"

63
#define NEEDED_RMEM (4*1024*1024)
64
#define CONN_HASH_SIZE 32
65

66 67 68
/* Number of messages to send before rescheduling */
#define MAX_SEND_MSG_COUNT 25

69
struct cbuf {
70 71 72
	unsigned int base;
	unsigned int len;
	unsigned int mask;
73 74
};

75 76 77 78
static void cbuf_add(struct cbuf *cb, int n)
{
	cb->len += n;
}
79

80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101
static int cbuf_data(struct cbuf *cb)
{
	return ((cb->base + cb->len) & cb->mask);
}

static void cbuf_init(struct cbuf *cb, int size)
{
	cb->base = cb->len = 0;
	cb->mask = size-1;
}

static void cbuf_eat(struct cbuf *cb, int n)
{
	cb->len  -= n;
	cb->base += n;
	cb->base &= cb->mask;
}

static bool cbuf_empty(struct cbuf *cb)
{
	return cb->len == 0;
}
102 103 104 105

struct connection {
	struct socket *sock;	/* NULL if not connected */
	uint32_t nodeid;	/* So we know who we are in the list */
106
	struct mutex sock_mutex;
107
	unsigned long flags;
108
#define CF_READ_PENDING 1
109
#define CF_WRITE_PENDING 2
110 111
#define CF_INIT_PENDING 4
#define CF_IS_OTHERCON 5
112
#define CF_CLOSE 6
113
#define CF_APP_LIMITED 7
114
#define CF_CLOSING 8
115
	struct list_head writequeue;  /* List of outgoing writequeue_entries */
116 117
	spinlock_t writequeue_lock;
	int (*rx_action) (struct connection *);	/* What to do when active */
118
	void (*connect_action) (struct connection *);	/* What to do to connect */
119 120 121 122
	struct page *rx_page;
	struct cbuf cb;
	int retries;
#define MAX_CONNECT_RETRIES 3
123
	struct hlist_node list;
124
	struct connection *othercon;
125 126
	struct work_struct rwork; /* Receive workqueue */
	struct work_struct swork; /* Send workqueue */
127 128 129 130 131 132 133 134 135 136 137 138 139 140
};
#define sock2con(x) ((struct connection *)(x)->sk_user_data)

/* An entry waiting to be sent */
struct writequeue_entry {
	struct list_head list;
	struct page *page;
	int offset;
	int len;
	int end;
	int users;
	struct connection *con;
};

141 142 143 144
struct dlm_node_addr {
	struct list_head list;
	int nodeid;
	int addr_count;
145
	int curr_addr_index;
146 147 148
	struct sockaddr_storage *addr[DLM_MAX_ADDR_COUNT];
};

149 150 151 152 153 154 155
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;

156 157 158
static LIST_HEAD(dlm_node_addrs);
static DEFINE_SPINLOCK(dlm_node_addrs_spin);

159 160
static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
static int dlm_local_count;
161
static int dlm_allow_conn;
162

163 164 165
/* Work queues */
static struct workqueue_struct *recv_workqueue;
static struct workqueue_struct *send_workqueue;
166

167
static struct hlist_head connection_hash[CONN_HASH_SIZE];
168
static DEFINE_MUTEX(connections_lock);
169
static struct kmem_cache *con_cache;
170

171 172
static void process_recv_sockets(struct work_struct *work);
static void process_send_sockets(struct work_struct *work);
173

174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189

/* 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;

	r = nodeid_hash(nodeid);

190
	hlist_for_each_entry(con, &connection_hash[r], list) {
191 192 193 194 195 196
		if (con->nodeid == nodeid)
			return con;
	}
	return NULL;
}

197 198 199 200 201
/*
 * 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)
202 203
{
	struct connection *con = NULL;
204
	int r;
205

206
	con = __find_con(nodeid);
207 208
	if (con || !alloc)
		return con;
209

210 211 212
	con = kmem_cache_zalloc(con_cache, alloc);
	if (!con)
		return NULL;
213

214 215
	r = nodeid_hash(nodeid);
	hlist_add_head(&con->list, &connection_hash[r]);
216

217 218 219 220 221 222
	con->nodeid = nodeid;
	mutex_init(&con->sock_mutex);
	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);
223

224 225
	/* Setup action pointers for child sockets */
	if (con->nodeid) {
226
		struct connection *zerocon = __find_con(0);
227

228 229 230
		con->connect_action = zerocon->connect_action;
		if (!con->rx_action)
			con->rx_action = zerocon->rx_action;
231 232
	}

233 234 235
	return con;
}

236 237 238 239
/* Loop round all connections */
static void foreach_conn(void (*conn_func)(struct connection *c))
{
	int i;
240
	struct hlist_node *n;
241 242 243
	struct connection *con;

	for (i = 0; i < CONN_HASH_SIZE; i++) {
244
		hlist_for_each_entry_safe(con, n, &connection_hash[i], list)
245 246 247 248
			conn_func(con);
	}
}

249 250 251 252
static struct connection *nodeid2con(int nodeid, gfp_t allocation)
{
	struct connection *con;

253
	mutex_lock(&connections_lock);
254
	con = __nodeid2con(nodeid, allocation);
255
	mutex_unlock(&connections_lock);
256

257 258 259
	return con;
}

260 261 262 263 264 265 266 267 268 269 270 271
static struct dlm_node_addr *find_node_addr(int nodeid)
{
	struct dlm_node_addr *na;

	list_for_each_entry(na, &dlm_node_addrs, list) {
		if (na->nodeid == nodeid)
			return na;
	}
	return NULL;
}

static int addr_compare(struct sockaddr_storage *x, struct sockaddr_storage *y)
272
{
273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298
	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,
299
			  struct sockaddr *sa_out, bool try_new_addr)
300 301 302
{
	struct sockaddr_storage sas;
	struct dlm_node_addr *na;
303 304 305 306

	if (!dlm_local_count)
		return -1;

307 308
	spin_lock(&dlm_node_addrs_spin);
	na = find_node_addr(nodeid);
309
	if (na && na->addr_count) {
310 311 312
		memcpy(&sas, na->addr[na->curr_addr_index],
		       sizeof(struct sockaddr_storage));

313 314 315 316 317 318
		if (try_new_addr) {
			na->curr_addr_index++;
			if (na->curr_addr_index == na->addr_count)
				na->curr_addr_index = 0;
		}
	}
319 320 321 322 323 324 325 326 327 328 329 330 331
	spin_unlock(&dlm_node_addrs_spin);

	if (!na)
		return -EEXIST;

	if (!na->addr_count)
		return -ENOENT;

	if (sas_out)
		memcpy(sas_out, &sas, sizeof(struct sockaddr_storage));

	if (!sa_out)
		return 0;
332 333

	if (dlm_local_addr[0]->ss_family == AF_INET) {
334 335
		struct sockaddr_in *in4  = (struct sockaddr_in *) &sas;
		struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;
336 337
		ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
	} else {
338 339
		struct sockaddr_in6 *in6  = (struct sockaddr_in6 *) &sas;
		struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;
340
		ret6->sin6_addr = in6->sin6_addr;
341 342 343 344 345
	}

	return 0;
}

346 347 348 349
static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid)
{
	struct dlm_node_addr *na;
	int rv = -EEXIST;
350
	int addr_i;
351 352 353 354 355 356

	spin_lock(&dlm_node_addrs_spin);
	list_for_each_entry(na, &dlm_node_addrs, list) {
		if (!na->addr_count)
			continue;

357 358 359 360 361 362 363
		for (addr_i = 0; addr_i < na->addr_count; addr_i++) {
			if (addr_compare(na->addr[addr_i], addr)) {
				*nodeid = na->nodeid;
				rv = 0;
				goto unlock;
			}
		}
364
	}
365
unlock:
366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410
	spin_unlock(&dlm_node_addrs_spin);
	return rv;
}

int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len)
{
	struct sockaddr_storage *new_addr;
	struct dlm_node_addr *new_node, *na;

	new_node = kzalloc(sizeof(struct dlm_node_addr), GFP_NOFS);
	if (!new_node)
		return -ENOMEM;

	new_addr = kzalloc(sizeof(struct sockaddr_storage), GFP_NOFS);
	if (!new_addr) {
		kfree(new_node);
		return -ENOMEM;
	}

	memcpy(new_addr, addr, len);

	spin_lock(&dlm_node_addrs_spin);
	na = find_node_addr(nodeid);
	if (!na) {
		new_node->nodeid = nodeid;
		new_node->addr[0] = new_addr;
		new_node->addr_count = 1;
		list_add(&new_node->list, &dlm_node_addrs);
		spin_unlock(&dlm_node_addrs_spin);
		return 0;
	}

	if (na->addr_count >= DLM_MAX_ADDR_COUNT) {
		spin_unlock(&dlm_node_addrs_spin);
		kfree(new_addr);
		kfree(new_node);
		return -ENOSPC;
	}

	na->addr[na->addr_count++] = new_addr;
	spin_unlock(&dlm_node_addrs_spin);
	kfree(new_node);
	return 0;
}

411
/* Data available on socket or listen socket received a connect */
412
static void lowcomms_data_ready(struct sock *sk)
413
{
414 415 416 417
	struct connection *con;

	read_lock_bh(&sk->sk_callback_lock);
	con = sock2con(sk);
418
	if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
419
		queue_work(recv_workqueue, &con->rwork);
420
	read_unlock_bh(&sk->sk_callback_lock);
421 422 423 424
}

static void lowcomms_write_space(struct sock *sk)
{
425
	struct connection *con;
426

427 428
	read_lock_bh(&sk->sk_callback_lock);
	con = sock2con(sk);
429
	if (!con)
430
		goto out;
431 432 433 434 435

	clear_bit(SOCK_NOSPACE, &con->sock->flags);

	if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
		con->sock->sk->sk_write_pending--;
436
		clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags);
437 438
	}

439
	queue_work(send_workqueue, &con->swork);
440 441
out:
	read_unlock_bh(&sk->sk_callback_lock);
442 443 444 445
}

static inline void lowcomms_connect_sock(struct connection *con)
{
446 447
	if (test_bit(CF_CLOSE, &con->flags))
		return;
448 449
	queue_work(send_workqueue, &con->swork);
	cond_resched();
450 451 452 453
}

static void lowcomms_state_change(struct sock *sk)
{
454 455 456 457 458 459 460 461 462
	/* SCTP layer is not calling sk_data_ready when the connection
	 * is done, so we catch the signal through here. Also, it
	 * doesn't switch socket state when entering shutdown, so we
	 * skip the write in that case.
	 */
	if (sk->sk_shutdown) {
		if (sk->sk_shutdown == RCV_SHUTDOWN)
			lowcomms_data_ready(sk);
	} else if (sk->sk_state == TCP_ESTABLISHED) {
463
		lowcomms_write_space(sk);
464
	}
465 466
}

467 468 469 470 471 472 473 474 475 476 477 478 479 480
int dlm_lowcomms_connect_node(int nodeid)
{
	struct connection *con;

	if (nodeid == dlm_our_nodeid())
		return 0;

	con = nodeid2con(nodeid, GFP_NOFS);
	if (!con)
		return -ENOMEM;
	lowcomms_connect_sock(con);
	return 0;
}

481 482
static void lowcomms_error_report(struct sock *sk)
{
483
	struct connection *con;
484
	struct sockaddr_storage saddr;
485
	int buflen;
486
	void (*orig_report)(struct sock *) = NULL;
487

488 489 490 491 492
	read_lock_bh(&sk->sk_callback_lock);
	con = sock2con(sk);
	if (con == NULL)
		goto out;

493
	orig_report = listen_sock.sk_error_report;
494 495
	if (con->sock == NULL ||
	    kernel_getpeername(con->sock, (struct sockaddr *)&saddr, &buflen)) {
496 497 498 499 500 501 502 503 504 505 506 507 508 509 510 511 512 513 514 515 516 517 518 519 520 521 522
		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
				   "sending to node %d, port %d, "
				   "sk_err=%d/%d\n", dlm_our_nodeid(),
				   con->nodeid, dlm_config.ci_tcp_port,
				   sk->sk_err, sk->sk_err_soft);
	} else if (saddr.ss_family == AF_INET) {
		struct sockaddr_in *sin4 = (struct sockaddr_in *)&saddr;

		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
				   "sending to node %d at %pI4, port %d, "
				   "sk_err=%d/%d\n", dlm_our_nodeid(),
				   con->nodeid, &sin4->sin_addr.s_addr,
				   dlm_config.ci_tcp_port, sk->sk_err,
				   sk->sk_err_soft);
	} else {
		struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&saddr;

		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
				   "sending to node %d at %u.%u.%u.%u, "
				   "port %d, sk_err=%d/%d\n", dlm_our_nodeid(),
				   con->nodeid, sin6->sin6_addr.s6_addr32[0],
				   sin6->sin6_addr.s6_addr32[1],
				   sin6->sin6_addr.s6_addr32[2],
				   sin6->sin6_addr.s6_addr32[3],
				   dlm_config.ci_tcp_port, sk->sk_err,
				   sk->sk_err_soft);
	}
523 524 525 526 527 528 529
out:
	read_unlock_bh(&sk->sk_callback_lock);
	if (orig_report)
		orig_report(sk);
}

/* Note: sk_callback_lock must be locked before calling this function. */
530
static void save_listen_callbacks(struct socket *sock)
531
{
532 533 534 535 536 537
	struct sock *sk = sock->sk;

	listen_sock.sk_data_ready = sk->sk_data_ready;
	listen_sock.sk_state_change = sk->sk_state_change;
	listen_sock.sk_write_space = sk->sk_write_space;
	listen_sock.sk_error_report = sk->sk_error_report;
538 539
}

540
static void restore_callbacks(struct socket *sock)
541
{
542 543
	struct sock *sk = sock->sk;

544 545
	write_lock_bh(&sk->sk_callback_lock);
	sk->sk_user_data = NULL;
546 547 548 549
	sk->sk_data_ready = listen_sock.sk_data_ready;
	sk->sk_state_change = listen_sock.sk_state_change;
	sk->sk_write_space = listen_sock.sk_write_space;
	sk->sk_error_report = listen_sock.sk_error_report;
550
	write_unlock_bh(&sk->sk_callback_lock);
551 552
}

553
/* Make a socket active */
554
static void add_sock(struct socket *sock, struct connection *con)
555
{
556 557 558
	struct sock *sk = sock->sk;

	write_lock_bh(&sk->sk_callback_lock);
559 560
	con->sock = sock;

561
	sk->sk_user_data = con;
562
	/* Install a data_ready callback */
563 564 565 566 567 568
	sk->sk_data_ready = lowcomms_data_ready;
	sk->sk_write_space = lowcomms_write_space;
	sk->sk_state_change = lowcomms_state_change;
	sk->sk_allocation = GFP_NOFS;
	sk->sk_error_report = lowcomms_error_report;
	write_unlock_bh(&sk->sk_callback_lock);
569 570
}

571
/* Add the port number to an IPv6 or 4 sockaddr and return the address
572 573 574 575
   length */
static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
			  int *addr_len)
{
576
	saddr->ss_family =  dlm_local_addr[0]->ss_family;
577
	if (saddr->ss_family == AF_INET) {
578 579 580
		struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
		in4_addr->sin_port = cpu_to_be16(port);
		*addr_len = sizeof(struct sockaddr_in);
581
		memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
582
	} else {
583 584 585 586
		struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
		in6_addr->sin6_port = cpu_to_be16(port);
		*addr_len = sizeof(struct sockaddr_in6);
	}
587
	memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
588 589 590
}

/* Close a remote connection and tidy up */
591 592
static void close_connection(struct connection *con, bool and_other,
			     bool tx, bool rx)
593
{
594 595
	bool closing = test_and_set_bit(CF_CLOSING, &con->flags);

596
	if (tx && !closing && cancel_work_sync(&con->swork)) {
597
		log_print("canceled swork for node %d", con->nodeid);
598 599 600
		clear_bit(CF_WRITE_PENDING, &con->flags);
	}
	if (rx && !closing && cancel_work_sync(&con->rwork)) {
601
		log_print("canceled rwork for node %d", con->nodeid);
602 603
		clear_bit(CF_READ_PENDING, &con->flags);
	}
604

605
	mutex_lock(&con->sock_mutex);
606
	if (con->sock) {
607
		restore_callbacks(con->sock);
608 609 610 611
		sock_release(con->sock);
		con->sock = NULL;
	}
	if (con->othercon && and_other) {
612
		/* Will only re-enter once. */
613
		close_connection(con->othercon, false, true, true);
614 615 616 617 618
	}
	if (con->rx_page) {
		__free_page(con->rx_page);
		con->rx_page = NULL;
	}
619

620 621
	con->retries = 0;
	mutex_unlock(&con->sock_mutex);
622
	clear_bit(CF_CLOSING, &con->flags);
623 624 625 626 627 628
}

/* Data received from remote end */
static int receive_from_sock(struct connection *con)
{
	int ret = 0;
629 630
	struct msghdr msg = {};
	struct kvec iov[2];
631 632 633
	unsigned len;
	int r;
	int call_again_soon = 0;
634
	int nvec;
635

636
	mutex_lock(&con->sock_mutex);
637

638 639 640 641
	if (con->sock == NULL) {
		ret = -EAGAIN;
		goto out_close;
	}
642 643 644 645
	if (con->nodeid == 0) {
		ret = -EINVAL;
		goto out_close;
	}
646

647 648 649 650 651 652 653 654
	if (con->rx_page == NULL) {
		/*
		 * This doesn't need to be atomic, but I think it should
		 * improve performance if it is.
		 */
		con->rx_page = alloc_page(GFP_ATOMIC);
		if (con->rx_page == NULL)
			goto out_resched;
655
		cbuf_init(&con->cb, PAGE_SIZE);
656 657 658 659 660 661
	}

	/*
	 * iov[0] is the bit of the circular buffer between the current end
	 * point (cb.base + cb.len) and the end of the buffer.
	 */
662 663
	iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
	iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
664
	iov[1].iov_len = 0;
665
	nvec = 1;
666 667 668 669 670

	/*
	 * iov[1] is the bit of the circular buffer between the start of the
	 * buffer and the start of the currently used section (cb.base)
	 */
671
	if (cbuf_data(&con->cb) >= con->cb.base) {
672
		iov[0].iov_len = PAGE_SIZE - cbuf_data(&con->cb);
673 674
		iov[1].iov_len = con->cb.base;
		iov[1].iov_base = page_address(con->rx_page);
675
		nvec = 2;
676 677 678
	}
	len = iov[0].iov_len + iov[1].iov_len;

679
	r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,
680 681 682
			       MSG_DONTWAIT | MSG_NOSIGNAL);
	if (ret <= 0)
		goto out_close;
683 684
	else if (ret == len)
		call_again_soon = 1;
685

686
	cbuf_add(&con->cb, ret);
687 688 689
	ret = dlm_process_incoming_buffer(con->nodeid,
					  page_address(con->rx_page),
					  con->cb.base, con->cb.len,
690
					  PAGE_SIZE);
691
	if (ret == -EBADMSG) {
692 693 694
		log_print("lowcomms: addr=%p, base=%u, len=%u, read=%d",
			  page_address(con->rx_page), con->cb.base,
			  con->cb.len, r);
695 696 697
	}
	if (ret < 0)
		goto out_close;
698
	cbuf_eat(&con->cb, ret);
699

700
	if (cbuf_empty(&con->cb) && !call_again_soon) {
701 702 703 704 705 706
		__free_page(con->rx_page);
		con->rx_page = NULL;
	}

	if (call_again_soon)
		goto out_resched;
707
	mutex_unlock(&con->sock_mutex);
708
	return 0;
709

710
out_resched:
711 712
	if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
		queue_work(recv_workqueue, &con->rwork);
713
	mutex_unlock(&con->sock_mutex);
714
	return -EAGAIN;
715

716
out_close:
717
	mutex_unlock(&con->sock_mutex);
718
	if (ret != -EAGAIN) {
719
		close_connection(con, true, true, false);
720 721
		/* Reconnect when there is something to send */
	}
722 723 724
	/* Don't return success if we really got EOF */
	if (ret == 0)
		ret = -EAGAIN;
725 726 727 728 729

	return ret;
}

/* Listening socket is busy, accept a connection */
730
static int tcp_accept_from_sock(struct connection *con)
731 732 733 734 735 736 737
{
	int result;
	struct sockaddr_storage peeraddr;
	struct socket *newsock;
	int len;
	int nodeid;
	struct connection *newcon;
738
	struct connection *addcon;
739

740 741 742 743 744 745 746
	mutex_lock(&connections_lock);
	if (!dlm_allow_conn) {
		mutex_unlock(&connections_lock);
		return -1;
	}
	mutex_unlock(&connections_lock);

747
	mutex_lock_nested(&con->sock_mutex, 0);
748

749 750 751 752
	if (!con->sock) {
		mutex_unlock(&con->sock_mutex);
		return -ENOTCONN;
	}
753

754
	result = kernel_accept(con->sock, &newsock, O_NONBLOCK);
755 756 757 758 759 760 761 762 763 764 765 766 767
	if (result < 0)
		goto accept_err;

	/* Get the connected socket's peer */
	memset(&peeraddr, 0, sizeof(peeraddr));
	if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr,
				  &len, 2)) {
		result = -ECONNABORTED;
		goto accept_err;
	}

	/* Get the new node's NODEID */
	make_sockaddr(&peeraddr, 0, &len);
768
	if (addr_to_nodeid(&peeraddr, &nodeid)) {
769
		unsigned char *b=(unsigned char *)&peeraddr;
David Teigland's avatar
David Teigland committed
770
		log_print("connect from non cluster node");
771 772
		print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 
				     b, sizeof(struct sockaddr_storage));
773
		sock_release(newsock);
774
		mutex_unlock(&con->sock_mutex);
775 776 777 778 779 780 781 782 783 784
		return -1;
	}

	log_print("got connection from %d", nodeid);

	/*  Check to see if we already have a connection to this node. This
	 *  could happen if the two nodes initiate a connection at roughly
	 *  the same time and the connections cross on the wire.
	 *  In this case we store the incoming one in "othercon"
	 */
785
	newcon = nodeid2con(nodeid, GFP_NOFS);
786 787 788 789
	if (!newcon) {
		result = -ENOMEM;
		goto accept_err;
	}
790
	mutex_lock_nested(&newcon->sock_mutex, 1);
791
	if (newcon->sock) {
792
		struct connection *othercon = newcon->othercon;
793 794

		if (!othercon) {
795
			othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
796
			if (!othercon) {
David Teigland's avatar
David Teigland committed
797
				log_print("failed to allocate incoming socket");
798
				mutex_unlock(&newcon->sock_mutex);
799 800 801 802 803
				result = -ENOMEM;
				goto accept_err;
			}
			othercon->nodeid = nodeid;
			othercon->rx_action = receive_from_sock;
804
			mutex_init(&othercon->sock_mutex);
805 806
			INIT_LIST_HEAD(&othercon->writequeue);
			spin_lock_init(&othercon->writequeue_lock);
807 808
			INIT_WORK(&othercon->swork, process_send_sockets);
			INIT_WORK(&othercon->rwork, process_recv_sockets);
809
			set_bit(CF_IS_OTHERCON, &othercon->flags);
810
		}
811
		mutex_lock_nested(&othercon->sock_mutex, 2);
812
		if (!othercon->sock) {
813
			newcon->othercon = othercon;
814
			add_sock(newsock, othercon);
815
			addcon = othercon;
816
			mutex_unlock(&othercon->sock_mutex);
817 818 819 820
		}
		else {
			printk("Extra connection from node %d attempted\n", nodeid);
			result = -EAGAIN;
821
			mutex_unlock(&othercon->sock_mutex);
822
			mutex_unlock(&newcon->sock_mutex);
823
			goto accept_err;
824 825 826 827
		}
	}
	else {
		newcon->rx_action = receive_from_sock;
828 829 830
		/* accept copies the sk after we've saved the callbacks, so we
		   don't want to save them a second time or comm errors will
		   result in calling sk_error_report recursively. */
831
		add_sock(newsock, newcon);
832
		addcon = newcon;
833 834
	}

835
	mutex_unlock(&newcon->sock_mutex);
836 837 838

	/*
	 * Add it to the active queue in case we got data
839
	 * between processing the accept adding the socket
840 841
	 * to the read_sockets list
	 */
842 843
	if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
		queue_work(recv_workqueue, &addcon->rwork);
844
	mutex_unlock(&con->sock_mutex);
845 846 847

	return 0;

848
accept_err:
849
	mutex_unlock(&con->sock_mutex);
850 851
	if (newsock)
		sock_release(newsock);
852 853

	if (result != -EAGAIN)
David Teigland's avatar
David Teigland committed
854
		log_print("error accepting connection from node: %d", result);
855 856 857
	return result;
}

858
static int sctp_accept_from_sock(struct connection *con)
859 860 861 862 863 864 865 866 867 868 869 870 871 872 873 874 875 876 877 878 879 880 881 882 883 884 885 886 887 888 889 890 891 892
{
	/* Check that the new node is in the lockspace */
	struct sctp_prim prim;
	int nodeid;
	int prim_len, ret;
	int addr_len;
	struct connection *newcon;
	struct connection *addcon;
	struct socket *newsock;

	mutex_lock(&connections_lock);
	if (!dlm_allow_conn) {
		mutex_unlock(&connections_lock);
		return -1;
	}
	mutex_unlock(&connections_lock);

	mutex_lock_nested(&con->sock_mutex, 0);

	ret = kernel_accept(con->sock, &newsock, O_NONBLOCK);
	if (ret < 0)
		goto accept_err;

	memset(&prim, 0, sizeof(struct sctp_prim));
	prim_len = sizeof(struct sctp_prim);

	ret = kernel_getsockopt(newsock, IPPROTO_SCTP, SCTP_PRIMARY_ADDR,
				(char *)&prim, &prim_len);
	if (ret < 0) {
		log_print("getsockopt/sctp_primary_addr failed: %d", ret);
		goto accept_err;
	}

	make_sockaddr(&prim.ssp_addr, 0, &addr_len);
893 894
	ret = addr_to_nodeid(&prim.ssp_addr, &nodeid);
	if (ret) {
895 896 897 898 899 900 901 902 903 904 905 906 907 908 909 910 911 912 913 914 915 916 917 918 919 920 921 922 923 924
		unsigned char *b = (unsigned char *)&prim.ssp_addr;

		log_print("reject connect from unknown addr");
		print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
				     b, sizeof(struct sockaddr_storage));
		goto accept_err;
	}

	newcon = nodeid2con(nodeid, GFP_NOFS);
	if (!newcon) {
		ret = -ENOMEM;
		goto accept_err;
	}

	mutex_lock_nested(&newcon->sock_mutex, 1);

	if (newcon->sock) {
		struct connection *othercon = newcon->othercon;

		if (!othercon) {
			othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
			if (!othercon) {
				log_print("failed to allocate incoming socket");
				mutex_unlock(&newcon->sock_mutex);
				ret = -ENOMEM;
				goto accept_err;
			}
			othercon->nodeid = nodeid;
			othercon->rx_action = receive_from_sock;
			mutex_init(&othercon->sock_mutex);
925 926
			INIT_LIST_HEAD(&othercon->writequeue);
			spin_lock_init(&othercon->writequeue_lock);
927 928 929 930
			INIT_WORK(&othercon->swork, process_send_sockets);
			INIT_WORK(&othercon->rwork, process_recv_sockets);
			set_bit(CF_IS_OTHERCON, &othercon->flags);
		}
931
		mutex_lock_nested(&othercon->sock_mutex, 2);
932 933
		if (!othercon->sock) {
			newcon->othercon = othercon;
934
			add_sock(newsock, othercon);
935
			addcon = othercon;
936
			mutex_unlock(&othercon->sock_mutex);
937 938 939
		} else {
			printk("Extra connection from node %d attempted\n", nodeid);
			ret = -EAGAIN;
940
			mutex_unlock(&othercon->sock_mutex);
941 942 943 944 945
			mutex_unlock(&newcon->sock_mutex);
			goto accept_err;
		}
	} else {
		newcon->rx_action = receive_from_sock;
946
		add_sock(newsock, newcon);
947 948 949 950 951 952 953 954 955 956 957 958 959 960 961 962 963 964 965 966 967 968 969 970 971 972 973 974
		addcon = newcon;
	}

	log_print("connected to %d", nodeid);

	mutex_unlock(&newcon->sock_mutex);

	/*
	 * Add it to the active queue in case we got data
	 * between processing the accept adding the socket
	 * to the read_sockets list
	 */
	if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
		queue_work(recv_workqueue, &addcon->rwork);
	mutex_unlock(&con->sock_mutex);

	return 0;

accept_err:
	mutex_unlock(&con->sock_mutex);
	if (newsock)
		sock_release(newsock);
	if (ret != -EAGAIN)
		log_print("error accepting connection from node: %d", ret);

	return ret;
}

975 976 977 978 979 980
static void free_entry(struct writequeue_entry *e)
{
	__free_page(e->page);
	kfree(e);
}

981 982 983 984 985 986 987 988 989 990 991 992 993 994 995 996 997 998
/*
 * writequeue_entry_complete - try to delete and free write queue entry
 * @e: write queue entry to try to delete
 * @completed: bytes completed
 *
 * writequeue_lock must be held.
 */
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);
		free_entry(e);
	}
}

999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012 1013 1014 1015 1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028
/*
 * sctp_bind_addrs - bind a SCTP socket to all our addresses
 */
static int sctp_bind_addrs(struct connection *con, uint16_t port)
{
	struct sockaddr_storage localaddr;
	int i, addr_len, result = 0;

	for (i = 0; i < dlm_local_count; i++) {
		memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr));
		make_sockaddr(&localaddr, port, &addr_len);

		if (!i)
			result = kernel_bind(con->sock,
					     (struct sockaddr *)&localaddr,
					     addr_len);
		else
			result = kernel_setsockopt(con->sock, SOL_SCTP,
						   SCTP_SOCKOPT_BINDX_ADD,
						   (char *)&localaddr, addr_len);

		if (result < 0) {
			log_print("Can't bind to %d addr number %d, %d.\n",
				  port, i + 1, result);
			break;
		}
	}
	return result;
}

1029 1030 1031 1032 1033
/* Initiate an SCTP association.
   This is a special case of send_to_sock() in that we don't yet have a
   peeled-off socket for this association, so we use the listening socket
   and add the primary IP address of the remote node.
 */
1034
static void sctp_connect_to_sock(struct connection *con)
1035
{
1036 1037 1038 1039 1040 1041 1042 1043 1044 1045
	struct sockaddr_storage daddr;
	int one = 1;
	int result;
	int addr_len;
	struct socket *sock;

	if (con->nodeid == 0) {
		log_print("attempt to connect sock 0 foiled");
		return;
	}
1046

1047
	mutex_lock(&con->sock_mutex);
1048

1049 1050 1051 1052 1053 1054 1055 1056 1057 1058 1059 1060
	/* Some odd races can cause double-connects, ignore them */
	if (con->retries++ > MAX_CONNECT_RETRIES)
		goto out;

	if (con->sock) {
		log_print("node %d already connected.", con->nodeid);
		goto out;
	}

	memset(&daddr, 0, sizeof(daddr));
	result = nodeid_to_addr(con->nodeid, &daddr, NULL, true);
	if (result < 0) {
1061
		log_print("no address for nodeid %d", con->nodeid);
1062
		goto out;
1063 1064
	}

1065 1066 1067 1068 1069
	/* Create a socket to communicate with */
	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
				  SOCK_STREAM, IPPROTO_SCTP, &sock);
	if (result < 0)
		goto socket_err;
1070

1071 1072
	con->rx_action = receive_from_sock;
	con->connect_action = sctp_connect_to_sock;
1073
	add_sock(sock, con);
1074

1075 1076 1077
	/* Bind to all addresses. */
	if (sctp_bind_addrs(con, 0))
		goto bind_err;
1078

1079
	make_sockaddr(&daddr, dlm_config.ci_tcp_port, &addr_len);
1080

1081
	log_print("connecting to %d", con->nodeid);
1082

1083 1084 1085
	/* Turn off Nagle's algorithm */
	kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
			  sizeof(one));
1086

1087 1088 1089 1090 1091 1092
	result = sock->ops->connect(sock, (struct sockaddr *)&daddr, addr_len,
				   O_NONBLOCK);
	if (result == -EINPROGRESS)
		result = 0;
	if (result == 0)
		goto out;
1093

1094 1095 1096
bind_err:
	con->sock = NULL;
	sock_release(sock);
1097

1098 1099 1100 1101 1102 1103 1104 1105 1106 1107 1108 1109 1110 1111 1112 1113
socket_err:
	/*
	 * Some errors are fatal and this list might need adjusting. For other
	 * errors we try again until the max number of retries is reached.
	 */
	if (result != -EHOSTUNREACH &&
	    result != -ENETUNREACH &&
	    result != -ENETDOWN &&
	    result != -EINVAL &&
	    result != -EPROTONOSUPPORT) {
		log_print("connect %d try %d error %d", con->nodeid,
			  con->retries, result);
		mutex_unlock(&con->sock_mutex);
		msleep(1000);
		lowcomms_connect_sock(con);
		return;
1114
	}
1115

1116
out:
1117
	mutex_unlock(&con->sock_mutex);
1118 1119
}

1120
/* Connect a new socket to its peer */
1121
static void tcp_connect_to_sock(struct connection *con)
1122
{
1123
	struct sockaddr_storage saddr, src_addr;
1124
	int addr_len;
1125
	struct socket *sock = NULL;
David Teigland's avatar
David Teigland committed
1126
	int one = 1;
1127
	int result;
1128 1129 1130

	if (con->nodeid == 0) {
		log_print("attempt to connect sock 0 foiled");
1131
		return;
1132 1133
	}

1134
	mutex_lock(&con->sock_mutex);
1135 1136 1137 1138
	if (con->retries++ > MAX_CONNECT_RETRIES)
		goto out;

	/* Some odd races can cause double-connects, ignore them */
1139
	if (con->sock)
1140 1141 1142
		goto out;

	/* Create a socket to communicate with */
1143 1144
	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
				  SOCK_STREAM, IPPROTO_TCP, &sock);
1145 1146 1147 1148
	if (result < 0)
		goto out_err;

	memset(&saddr, 0, sizeof(saddr));
1149
	result = nodeid_to_addr(con->nodeid, &saddr, NULL, false);
1150 1151
	if (result < 0) {
		log_print("no address for nodeid %d", con->nodeid);
1152
		goto out_err;
1153
	}
1154 1155

	con->rx_action = receive_from_sock;
1156
	con->connect_action = tcp_connect_to_sock;
1157
	add_sock(sock, con);
1158

1159 1160 1161 1162 1163 1164 1165 1166 1167 1168 1169
	/* Bind to our cluster-known address connecting to avoid
	   routing problems */
	memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
	make_sockaddr(&src_addr, 0, &addr_len);
	result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
				 addr_len);
	if (result < 0) {
		log_print("could not bind for connect: %d", result);
		/* This *may* not indicate a critical error */
	}

1170
	make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
1171 1172

	log_print("connecting to %d", con->nodeid);
David Teigland's avatar
David Teigland committed
1173 1174 1175 1176 1177

	/* Turn off Nagle's algorithm */
	kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
			  sizeof(one));

1178
	result = sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
1179
				   O_NONBLOCK);
1180 1181
	if (result == -EINPROGRESS)
		result = 0;
1182 1183
	if (result == 0)
		goto out;
1184

1185
out_err:
1186 1187 1188
	if (con->sock) {
		sock_release(con->sock);
		con->sock = NULL;
1189 1190
	} else if (sock) {
		sock_release(sock);
1191 1192 1193 1194 1195
	}
	/*
	 * Some errors are fatal and this list might need adjusting. For other
	 * errors we try again until the max number of retries is reached.
	 */
1196 1197 1198 1199 1200 1201 1202 1203 1204
	if (result != -EHOSTUNREACH &&
	    result != -ENETUNREACH &&
	    result != -ENETDOWN && 
	    result != -EINVAL &&
	    result != -EPROTONOSUPPORT) {
		log_print("connect %d try %d error %d", con->nodeid,
			  con->retries, result);
		mutex_unlock(&con->sock_mutex);
		msleep(1000);
1205
		lowcomms_connect_sock(con);
1206
		return;
1207
	}
1208
out:
1209
	mutex_unlock(&con->sock_mutex);
1210
	return;
1211 1212
}

1213 1214
static struct socket *tcp_create_listen_sock(struct connection *con,
					     struct sockaddr_storage *saddr)
1215
{
1216
	struct socket *sock = NULL;
1217 1218 1219 1220
	int result = 0;
	int one = 1;
	int addr_len;

1221
	if (dlm_local_addr[0]->ss_family == AF_INET)
1222 1223 1224 1225 1226
		addr_len = sizeof(struct sockaddr_in);
	else
		addr_len = sizeof(struct sockaddr_in6);

	/* Create a socket to communicate with */
1227 1228
	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
				  SOCK_STREAM, IPPROTO_TCP, &sock);
1229
	if (result < 0) {
David Teigland's avatar
David Teigland committed
1230
		log_print("Can't create listening comms socket");
1231 1232 1233
		goto create_out;
	}

David Teigland's avatar
David Teigland committed
1234 1235 1236 1237
	/* Turn off Nagle's algorithm */
	kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
			  sizeof(one));

1238 1239 1240
	result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
				   (char *)&one, sizeof(one));

1241
	if (result < 0) {
David Teigland's avatar
David Teigland committed
1242
		log_print("Failed to set SO_REUSEADDR on socket: %d", result);
1243
	}
1244
	write_lock_bh(&sock->sk->sk_callback_lock);
1245
	sock->sk->sk_user_data = con;
1246
	save_listen_callbacks(sock);
1247 1248
	con->rx_action = tcp_accept_from_sock;
	con->connect_action = tcp_connect_to_sock;
1249
	write_unlock_bh(&sock->sk->sk_callback_lock);
1250 1251

	/* Bind to our port */
1252
	make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
1253 1254
	result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
	if (result < 0) {
David Teigland's avatar
David Teigland committed
1255
		log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
1256 1257 1258 1259 1260
		sock_release(sock);
		sock = NULL;
		con->sock = NULL;
		goto create_out;
	}
1261
	result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
1262
				 (char *)&one, sizeof(one));
1263
	if (result < 0) {
David Teigland's avatar
David Teigland committed
1264
		log_print("Set keepalive failed: %d", result);
1265 1266 1267 1268
	}

	result = sock->ops->listen(sock, 5);
	if (result < 0) {
David Teigland's avatar
David Teigland committed
1269
		log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
1270 1271 1272 1273 1274
		sock_release(sock);
		sock = NULL;
		goto create_out;
	}

1275
create_out:
1276 1277 1278
	return sock;
}

1279 1280 1281 1282 1283 1284
/* Get local addresses */
static void init_local(void)
{
	struct sockaddr_storage sas, *addr;
	int i;

1285
	dlm_local_count = 0;
1286
	for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
1287 1288 1289
		if (dlm_our_addr(&sas, i))
			break;

1290
		addr = kmemdup(&sas, sizeof(*addr), GFP_NOFS);
1291 1292 1293 1294 1295 1296 1297 1298 1299 1300
		if (!addr)
			break;
		dlm_local_addr[dlm_local_count++] = addr;
	}
}

/* Initialise SCTP socket and bind to all interfaces */
static int sctp_listen_for_all(void)
{
	struct socket *sock = NULL;
1301
	int result = -EINVAL;
1302
	struct connection *con = nodeid2con(0, GFP_NOFS);
1303
	int bufsize = NEEDED_RMEM;
1304
	int one = 1;
1305 1306 1307 1308 1309 1310

	if (!con)
		return -ENOMEM;

	log_print("Using SCTP for communications");

1311
	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1312
				  SOCK_STREAM, IPPROTO_SCTP, &sock);
1313 1314 1315 1316 1317
	if (result < 0) {
		log_print("Can't create comms socket, check SCTP is loaded");
		goto out;
	}

1318
	result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
1319 1320
				 (char *)&bufsize, sizeof(bufsize));
	if (result)
David Teigland's avatar
David Teigland committed
1321
		log_print("Error increasing buffer space on socket %d", result);
1322

1323 1324 1325 1326 1327
	result = kernel_setsockopt(sock, SOL_SCTP, SCTP_NODELAY, (char *)&one,
				   sizeof(one));
	if (result < 0)
		log_print("Could not set SCTP NODELAY error %d\n", result);

1328
	write_lock_bh(&sock->sk->sk_callback_lock);
1329 1330
	/* Init con struct */
	sock->sk->sk_user_data = con;
1331
	save_listen_callbacks(sock);
1332 1333
	con->sock = sock;
	con->sock->sk->sk_data_ready = lowcomms_data_ready;
1334 1335
	con->rx_action = sctp_accept_from_sock;
	con->connect_action = sctp_connect_to_sock;
1336

1337 1338
	write_unlock_bh(&sock->sk->sk_callback_lock);

1339 1340 1341
	/* Bind to all addresses. */
	if (sctp_bind_addrs(con, dlm_config.ci_tcp_port))
		goto create_delsock;
1342 1343 1344 1345 1346 1347 1348 1349 1350 1351 1352 1353 1354 1355 1356 1357 1358

	result = sock->ops->listen(sock, 5);
	if (result < 0) {
		log_print("Can't set socket listening");
		goto create_delsock;
	}

	return 0;

create_delsock:
	sock_release(sock);
	con->sock = NULL;
out:
	return result;
}

static int tcp_listen_for_all(void)
1359 1360
{
	struct socket *sock = NULL;
1361
	struct connection *con = nodeid2con(0, GFP_NOFS);
1362 1363
	int result = -EINVAL;

1364 1365 1366
	if (!con)
		return -ENOMEM;

1367
	/* We don't support multi-homed hosts */
1368
	if (dlm_local_addr[1] != NULL) {
David Teigland's avatar
David Teigland committed
1369 1370
		log_print("TCP protocol can't handle multi-homed hosts, "
			  "try SCTP");
1371 1372 1373 1374 1375 1376
		return -EINVAL;
	}

	log_print("Using TCP for communications");

	sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
1377
	if (sock) {
1378
		add_sock(sock, con);
1379 1380 1381 1382 1383 1384 1385 1386 1387 1388 1389 1390 1391 1392 1393 1394 1395 1396 1397 1398 1399 1400 1401 1402 1403 1404 1405 1406 1407 1408 1409 1410 1411 1412 1413
		result = 0;
	}
	else {
		result = -EADDRINUSE;
	}

	return result;
}



static struct writequeue_entry *new_writequeue_entry(struct connection *con,
						     gfp_t allocation)
{
	struct writequeue_entry *entry;

	entry = kmalloc(sizeof(struct writequeue_entry), allocation);
	if (!entry)
		return NULL;

	entry->page = alloc_page(allocation);
	if (!entry->page) {
		kfree(entry);
		return NULL;
	}

	entry->offset = 0;
	entry->len = 0;
	entry->end = 0;
	entry->users = 0;
	entry->con = con;

	return entry;
}

David Teigland's avatar
David Teigland committed
1414
void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
1415 1416 1417 1418 1419 1420 1421 1422 1423
{
	struct connection *con;
	struct writequeue_entry *e;
	int offset = 0;

	con = nodeid2con(nodeid, allocation);
	if (!con)
		return NULL;

1424
	spin_lock(&con->writequeue_lock);
1425
	e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
1426
	if ((&e->list == &con->writequeue) ||