xref: /linux/fs/dlm/lowcomms.c (revision 8c749ce93ee69e789e46b3be98de9e0cbfcf8ed8)
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 **  This copyrighted material is made available to anyone wishing to use,
8 **  modify, copy, or redistribute it subject to the terms and conditions
9 **  of the GNU General Public License v.2.
10 **
11 *******************************************************************************
12 ******************************************************************************/
13 
14 /*
15  * lowcomms.c
16  *
17  * This is the "low-level" comms layer.
18  *
19  * It is responsible for sending/receiving messages
20  * from other nodes in the cluster.
21  *
22  * Cluster nodes are referred to by their nodeids. nodeids are
23  * simply 32 bit numbers to the locking module - if they need to
24  * be expanded for the cluster infrastructure then that is its
25  * responsibility. It is this layer's
26  * responsibility to resolve these into IP address or
27  * whatever it needs for inter-node communication.
28  *
29  * The comms level is two kernel threads that deal mainly with
30  * the receiving of messages from other nodes and passing them
31  * up to the mid-level comms layer (which understands the
32  * message format) for execution by the locking core, and
33  * a send thread which does all the setting up of connections
34  * to remote nodes and the sending of data. Threads are not allowed
35  * to send their own data because it may cause them to wait in times
36  * of high load. Also, this way, the sending thread can collect together
37  * messages bound for one node and send them in one block.
38  *
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
41  * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
42  * cluster-wide mechanism as it must be the same on all nodes of the cluster
43  * for the DLM to function.
44  *
45  */
46 
47 #include <asm/ioctls.h>
48 #include <net/sock.h>
49 #include <net/tcp.h>
50 #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>
56 #include <net/ipv6.h>
57 
58 #include "dlm_internal.h"
59 #include "lowcomms.h"
60 #include "midcomms.h"
61 #include "config.h"
62 
63 #define NEEDED_RMEM (4*1024*1024)
64 #define CONN_HASH_SIZE 32
65 
66 /* Number of messages to send before rescheduling */
67 #define MAX_SEND_MSG_COUNT 25
68 
69 struct cbuf {
70 	unsigned int base;
71 	unsigned int len;
72 	unsigned int mask;
73 };
74 
75 static void cbuf_add(struct cbuf *cb, int n)
76 {
77 	cb->len += n;
78 }
79 
80 static int cbuf_data(struct cbuf *cb)
81 {
82 	return ((cb->base + cb->len) & cb->mask);
83 }
84 
85 static void cbuf_init(struct cbuf *cb, int size)
86 {
87 	cb->base = cb->len = 0;
88 	cb->mask = size-1;
89 }
90 
91 static void cbuf_eat(struct cbuf *cb, int n)
92 {
93 	cb->len  -= n;
94 	cb->base += n;
95 	cb->base &= cb->mask;
96 }
97 
98 static bool cbuf_empty(struct cbuf *cb)
99 {
100 	return cb->len == 0;
101 }
102 
103 struct connection {
104 	struct socket *sock;	/* NULL if not connected */
105 	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 #define CF_CONNECT_PENDING 3
111 #define CF_INIT_PENDING 4
112 #define CF_IS_OTHERCON 5
113 #define CF_CLOSE 6
114 #define CF_APP_LIMITED 7
115 	struct list_head writequeue;  /* List of outgoing writequeue_entries */
116 	spinlock_t writequeue_lock;
117 	int (*rx_action) (struct connection *);	/* What to do when active */
118 	void (*connect_action) (struct connection *);	/* What to do to connect */
119 	struct page *rx_page;
120 	struct cbuf cb;
121 	int retries;
122 #define MAX_CONNECT_RETRIES 3
123 	struct hlist_node list;
124 	struct connection *othercon;
125 	struct work_struct rwork; /* Receive workqueue */
126 	struct work_struct swork; /* Send workqueue */
127 	void (*orig_error_report)(struct sock *sk);
128 };
129 #define sock2con(x) ((struct connection *)(x)->sk_user_data)
130 
131 /* An entry waiting to be sent */
132 struct writequeue_entry {
133 	struct list_head list;
134 	struct page *page;
135 	int offset;
136 	int len;
137 	int end;
138 	int users;
139 	struct connection *con;
140 };
141 
142 struct dlm_node_addr {
143 	struct list_head list;
144 	int nodeid;
145 	int addr_count;
146 	int curr_addr_index;
147 	struct sockaddr_storage *addr[DLM_MAX_ADDR_COUNT];
148 };
149 
150 static LIST_HEAD(dlm_node_addrs);
151 static DEFINE_SPINLOCK(dlm_node_addrs_spin);
152 
153 static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
154 static int dlm_local_count;
155 static int dlm_allow_conn;
156 
157 /* Work queues */
158 static struct workqueue_struct *recv_workqueue;
159 static struct workqueue_struct *send_workqueue;
160 
161 static struct hlist_head connection_hash[CONN_HASH_SIZE];
162 static DEFINE_MUTEX(connections_lock);
163 static struct kmem_cache *con_cache;
164 
165 static void process_recv_sockets(struct work_struct *work);
166 static void process_send_sockets(struct work_struct *work);
167 
168 
169 /* This is deliberately very simple because most clusters have simple
170    sequential nodeids, so we should be able to go straight to a connection
171    struct in the array */
172 static inline int nodeid_hash(int nodeid)
173 {
174 	return nodeid & (CONN_HASH_SIZE-1);
175 }
176 
177 static struct connection *__find_con(int nodeid)
178 {
179 	int r;
180 	struct connection *con;
181 
182 	r = nodeid_hash(nodeid);
183 
184 	hlist_for_each_entry(con, &connection_hash[r], list) {
185 		if (con->nodeid == nodeid)
186 			return con;
187 	}
188 	return NULL;
189 }
190 
191 /*
192  * If 'allocation' is zero then we don't attempt to create a new
193  * connection structure for this node.
194  */
195 static struct connection *__nodeid2con(int nodeid, gfp_t alloc)
196 {
197 	struct connection *con = NULL;
198 	int r;
199 
200 	con = __find_con(nodeid);
201 	if (con || !alloc)
202 		return con;
203 
204 	con = kmem_cache_zalloc(con_cache, alloc);
205 	if (!con)
206 		return NULL;
207 
208 	r = nodeid_hash(nodeid);
209 	hlist_add_head(&con->list, &connection_hash[r]);
210 
211 	con->nodeid = nodeid;
212 	mutex_init(&con->sock_mutex);
213 	INIT_LIST_HEAD(&con->writequeue);
214 	spin_lock_init(&con->writequeue_lock);
215 	INIT_WORK(&con->swork, process_send_sockets);
216 	INIT_WORK(&con->rwork, process_recv_sockets);
217 
218 	/* Setup action pointers for child sockets */
219 	if (con->nodeid) {
220 		struct connection *zerocon = __find_con(0);
221 
222 		con->connect_action = zerocon->connect_action;
223 		if (!con->rx_action)
224 			con->rx_action = zerocon->rx_action;
225 	}
226 
227 	return con;
228 }
229 
230 /* Loop round all connections */
231 static void foreach_conn(void (*conn_func)(struct connection *c))
232 {
233 	int i;
234 	struct hlist_node *n;
235 	struct connection *con;
236 
237 	for (i = 0; i < CONN_HASH_SIZE; i++) {
238 		hlist_for_each_entry_safe(con, n, &connection_hash[i], list)
239 			conn_func(con);
240 	}
241 }
242 
243 static struct connection *nodeid2con(int nodeid, gfp_t allocation)
244 {
245 	struct connection *con;
246 
247 	mutex_lock(&connections_lock);
248 	con = __nodeid2con(nodeid, allocation);
249 	mutex_unlock(&connections_lock);
250 
251 	return con;
252 }
253 
254 static struct dlm_node_addr *find_node_addr(int nodeid)
255 {
256 	struct dlm_node_addr *na;
257 
258 	list_for_each_entry(na, &dlm_node_addrs, list) {
259 		if (na->nodeid == nodeid)
260 			return na;
261 	}
262 	return NULL;
263 }
264 
265 static int addr_compare(struct sockaddr_storage *x, struct sockaddr_storage *y)
266 {
267 	switch (x->ss_family) {
268 	case AF_INET: {
269 		struct sockaddr_in *sinx = (struct sockaddr_in *)x;
270 		struct sockaddr_in *siny = (struct sockaddr_in *)y;
271 		if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr)
272 			return 0;
273 		if (sinx->sin_port != siny->sin_port)
274 			return 0;
275 		break;
276 	}
277 	case AF_INET6: {
278 		struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x;
279 		struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y;
280 		if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr))
281 			return 0;
282 		if (sinx->sin6_port != siny->sin6_port)
283 			return 0;
284 		break;
285 	}
286 	default:
287 		return 0;
288 	}
289 	return 1;
290 }
291 
292 static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
293 			  struct sockaddr *sa_out, bool try_new_addr)
294 {
295 	struct sockaddr_storage sas;
296 	struct dlm_node_addr *na;
297 
298 	if (!dlm_local_count)
299 		return -1;
300 
301 	spin_lock(&dlm_node_addrs_spin);
302 	na = find_node_addr(nodeid);
303 	if (na && na->addr_count) {
304 		memcpy(&sas, na->addr[na->curr_addr_index],
305 		       sizeof(struct sockaddr_storage));
306 
307 		if (try_new_addr) {
308 			na->curr_addr_index++;
309 			if (na->curr_addr_index == na->addr_count)
310 				na->curr_addr_index = 0;
311 		}
312 	}
313 	spin_unlock(&dlm_node_addrs_spin);
314 
315 	if (!na)
316 		return -EEXIST;
317 
318 	if (!na->addr_count)
319 		return -ENOENT;
320 
321 	if (sas_out)
322 		memcpy(sas_out, &sas, sizeof(struct sockaddr_storage));
323 
324 	if (!sa_out)
325 		return 0;
326 
327 	if (dlm_local_addr[0]->ss_family == AF_INET) {
328 		struct sockaddr_in *in4  = (struct sockaddr_in *) &sas;
329 		struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;
330 		ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
331 	} else {
332 		struct sockaddr_in6 *in6  = (struct sockaddr_in6 *) &sas;
333 		struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;
334 		ret6->sin6_addr = in6->sin6_addr;
335 	}
336 
337 	return 0;
338 }
339 
340 static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid)
341 {
342 	struct dlm_node_addr *na;
343 	int rv = -EEXIST;
344 	int addr_i;
345 
346 	spin_lock(&dlm_node_addrs_spin);
347 	list_for_each_entry(na, &dlm_node_addrs, list) {
348 		if (!na->addr_count)
349 			continue;
350 
351 		for (addr_i = 0; addr_i < na->addr_count; addr_i++) {
352 			if (addr_compare(na->addr[addr_i], addr)) {
353 				*nodeid = na->nodeid;
354 				rv = 0;
355 				goto unlock;
356 			}
357 		}
358 	}
359 unlock:
360 	spin_unlock(&dlm_node_addrs_spin);
361 	return rv;
362 }
363 
364 int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len)
365 {
366 	struct sockaddr_storage *new_addr;
367 	struct dlm_node_addr *new_node, *na;
368 
369 	new_node = kzalloc(sizeof(struct dlm_node_addr), GFP_NOFS);
370 	if (!new_node)
371 		return -ENOMEM;
372 
373 	new_addr = kzalloc(sizeof(struct sockaddr_storage), GFP_NOFS);
374 	if (!new_addr) {
375 		kfree(new_node);
376 		return -ENOMEM;
377 	}
378 
379 	memcpy(new_addr, addr, len);
380 
381 	spin_lock(&dlm_node_addrs_spin);
382 	na = find_node_addr(nodeid);
383 	if (!na) {
384 		new_node->nodeid = nodeid;
385 		new_node->addr[0] = new_addr;
386 		new_node->addr_count = 1;
387 		list_add(&new_node->list, &dlm_node_addrs);
388 		spin_unlock(&dlm_node_addrs_spin);
389 		return 0;
390 	}
391 
392 	if (na->addr_count >= DLM_MAX_ADDR_COUNT) {
393 		spin_unlock(&dlm_node_addrs_spin);
394 		kfree(new_addr);
395 		kfree(new_node);
396 		return -ENOSPC;
397 	}
398 
399 	na->addr[na->addr_count++] = new_addr;
400 	spin_unlock(&dlm_node_addrs_spin);
401 	kfree(new_node);
402 	return 0;
403 }
404 
405 /* Data available on socket or listen socket received a connect */
406 static void lowcomms_data_ready(struct sock *sk)
407 {
408 	struct connection *con = sock2con(sk);
409 	if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
410 		queue_work(recv_workqueue, &con->rwork);
411 }
412 
413 static void lowcomms_write_space(struct sock *sk)
414 {
415 	struct connection *con = sock2con(sk);
416 
417 	if (!con)
418 		return;
419 
420 	clear_bit(SOCK_NOSPACE, &con->sock->flags);
421 
422 	if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
423 		con->sock->sk->sk_write_pending--;
424 		clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags);
425 	}
426 
427 	if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
428 		queue_work(send_workqueue, &con->swork);
429 }
430 
431 static inline void lowcomms_connect_sock(struct connection *con)
432 {
433 	if (test_bit(CF_CLOSE, &con->flags))
434 		return;
435 	if (!test_and_set_bit(CF_CONNECT_PENDING, &con->flags))
436 		queue_work(send_workqueue, &con->swork);
437 }
438 
439 static void lowcomms_state_change(struct sock *sk)
440 {
441 	/* SCTP layer is not calling sk_data_ready when the connection
442 	 * is done, so we catch the signal through here. Also, it
443 	 * doesn't switch socket state when entering shutdown, so we
444 	 * skip the write in that case.
445 	 */
446 	if (sk->sk_shutdown) {
447 		if (sk->sk_shutdown == RCV_SHUTDOWN)
448 			lowcomms_data_ready(sk);
449 	} else if (sk->sk_state == TCP_ESTABLISHED) {
450 		lowcomms_write_space(sk);
451 	}
452 }
453 
454 int dlm_lowcomms_connect_node(int nodeid)
455 {
456 	struct connection *con;
457 
458 	if (nodeid == dlm_our_nodeid())
459 		return 0;
460 
461 	con = nodeid2con(nodeid, GFP_NOFS);
462 	if (!con)
463 		return -ENOMEM;
464 	lowcomms_connect_sock(con);
465 	return 0;
466 }
467 
468 static void lowcomms_error_report(struct sock *sk)
469 {
470 	struct connection *con = sock2con(sk);
471 	struct sockaddr_storage saddr;
472 
473 	if (nodeid_to_addr(con->nodeid, &saddr, NULL, false)) {
474 		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
475 				   "sending to node %d, port %d, "
476 				   "sk_err=%d/%d\n", dlm_our_nodeid(),
477 				   con->nodeid, dlm_config.ci_tcp_port,
478 				   sk->sk_err, sk->sk_err_soft);
479 		return;
480 	} else if (saddr.ss_family == AF_INET) {
481 		struct sockaddr_in *sin4 = (struct sockaddr_in *)&saddr;
482 
483 		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
484 				   "sending to node %d at %pI4, port %d, "
485 				   "sk_err=%d/%d\n", dlm_our_nodeid(),
486 				   con->nodeid, &sin4->sin_addr.s_addr,
487 				   dlm_config.ci_tcp_port, sk->sk_err,
488 				   sk->sk_err_soft);
489 	} else {
490 		struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&saddr;
491 
492 		printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
493 				   "sending to node %d at %u.%u.%u.%u, "
494 				   "port %d, sk_err=%d/%d\n", dlm_our_nodeid(),
495 				   con->nodeid, sin6->sin6_addr.s6_addr32[0],
496 				   sin6->sin6_addr.s6_addr32[1],
497 				   sin6->sin6_addr.s6_addr32[2],
498 				   sin6->sin6_addr.s6_addr32[3],
499 				   dlm_config.ci_tcp_port, sk->sk_err,
500 				   sk->sk_err_soft);
501 	}
502 	con->orig_error_report(sk);
503 }
504 
505 /* Make a socket active */
506 static void add_sock(struct socket *sock, struct connection *con)
507 {
508 	con->sock = sock;
509 
510 	/* Install a data_ready callback */
511 	con->sock->sk->sk_data_ready = lowcomms_data_ready;
512 	con->sock->sk->sk_write_space = lowcomms_write_space;
513 	con->sock->sk->sk_state_change = lowcomms_state_change;
514 	con->sock->sk->sk_user_data = con;
515 	con->sock->sk->sk_allocation = GFP_NOFS;
516 	con->orig_error_report = con->sock->sk->sk_error_report;
517 	con->sock->sk->sk_error_report = lowcomms_error_report;
518 }
519 
520 /* Add the port number to an IPv6 or 4 sockaddr and return the address
521    length */
522 static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
523 			  int *addr_len)
524 {
525 	saddr->ss_family =  dlm_local_addr[0]->ss_family;
526 	if (saddr->ss_family == AF_INET) {
527 		struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
528 		in4_addr->sin_port = cpu_to_be16(port);
529 		*addr_len = sizeof(struct sockaddr_in);
530 		memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
531 	} else {
532 		struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
533 		in6_addr->sin6_port = cpu_to_be16(port);
534 		*addr_len = sizeof(struct sockaddr_in6);
535 	}
536 	memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
537 }
538 
539 /* Close a remote connection and tidy up */
540 static void close_connection(struct connection *con, bool and_other,
541 			     bool tx, bool rx)
542 {
543 	clear_bit(CF_CONNECT_PENDING, &con->flags);
544 	clear_bit(CF_WRITE_PENDING, &con->flags);
545 	if (tx && cancel_work_sync(&con->swork))
546 		log_print("canceled swork for node %d", con->nodeid);
547 	if (rx && cancel_work_sync(&con->rwork))
548 		log_print("canceled rwork for node %d", con->nodeid);
549 
550 	mutex_lock(&con->sock_mutex);
551 	if (con->sock) {
552 		sock_release(con->sock);
553 		con->sock = NULL;
554 	}
555 	if (con->othercon && and_other) {
556 		/* Will only re-enter once. */
557 		close_connection(con->othercon, false, true, true);
558 	}
559 	if (con->rx_page) {
560 		__free_page(con->rx_page);
561 		con->rx_page = NULL;
562 	}
563 
564 	con->retries = 0;
565 	mutex_unlock(&con->sock_mutex);
566 }
567 
568 /* Data received from remote end */
569 static int receive_from_sock(struct connection *con)
570 {
571 	int ret = 0;
572 	struct msghdr msg = {};
573 	struct kvec iov[2];
574 	unsigned len;
575 	int r;
576 	int call_again_soon = 0;
577 	int nvec;
578 
579 	mutex_lock(&con->sock_mutex);
580 
581 	if (con->sock == NULL) {
582 		ret = -EAGAIN;
583 		goto out_close;
584 	}
585 	if (con->nodeid == 0) {
586 		ret = -EINVAL;
587 		goto out_close;
588 	}
589 
590 	if (con->rx_page == NULL) {
591 		/*
592 		 * This doesn't need to be atomic, but I think it should
593 		 * improve performance if it is.
594 		 */
595 		con->rx_page = alloc_page(GFP_ATOMIC);
596 		if (con->rx_page == NULL)
597 			goto out_resched;
598 		cbuf_init(&con->cb, PAGE_CACHE_SIZE);
599 	}
600 
601 	/*
602 	 * iov[0] is the bit of the circular buffer between the current end
603 	 * point (cb.base + cb.len) and the end of the buffer.
604 	 */
605 	iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
606 	iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
607 	iov[1].iov_len = 0;
608 	nvec = 1;
609 
610 	/*
611 	 * iov[1] is the bit of the circular buffer between the start of the
612 	 * buffer and the start of the currently used section (cb.base)
613 	 */
614 	if (cbuf_data(&con->cb) >= con->cb.base) {
615 		iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);
616 		iov[1].iov_len = con->cb.base;
617 		iov[1].iov_base = page_address(con->rx_page);
618 		nvec = 2;
619 	}
620 	len = iov[0].iov_len + iov[1].iov_len;
621 
622 	r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,
623 			       MSG_DONTWAIT | MSG_NOSIGNAL);
624 	if (ret <= 0)
625 		goto out_close;
626 	else if (ret == len)
627 		call_again_soon = 1;
628 
629 	cbuf_add(&con->cb, ret);
630 	ret = dlm_process_incoming_buffer(con->nodeid,
631 					  page_address(con->rx_page),
632 					  con->cb.base, con->cb.len,
633 					  PAGE_CACHE_SIZE);
634 	if (ret == -EBADMSG) {
635 		log_print("lowcomms: addr=%p, base=%u, len=%u, read=%d",
636 			  page_address(con->rx_page), con->cb.base,
637 			  con->cb.len, r);
638 	}
639 	if (ret < 0)
640 		goto out_close;
641 	cbuf_eat(&con->cb, ret);
642 
643 	if (cbuf_empty(&con->cb) && !call_again_soon) {
644 		__free_page(con->rx_page);
645 		con->rx_page = NULL;
646 	}
647 
648 	if (call_again_soon)
649 		goto out_resched;
650 	mutex_unlock(&con->sock_mutex);
651 	return 0;
652 
653 out_resched:
654 	if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
655 		queue_work(recv_workqueue, &con->rwork);
656 	mutex_unlock(&con->sock_mutex);
657 	return -EAGAIN;
658 
659 out_close:
660 	mutex_unlock(&con->sock_mutex);
661 	if (ret != -EAGAIN) {
662 		close_connection(con, false, true, false);
663 		/* Reconnect when there is something to send */
664 	}
665 	/* Don't return success if we really got EOF */
666 	if (ret == 0)
667 		ret = -EAGAIN;
668 
669 	return ret;
670 }
671 
672 /* Listening socket is busy, accept a connection */
673 static int tcp_accept_from_sock(struct connection *con)
674 {
675 	int result;
676 	struct sockaddr_storage peeraddr;
677 	struct socket *newsock;
678 	int len;
679 	int nodeid;
680 	struct connection *newcon;
681 	struct connection *addcon;
682 
683 	mutex_lock(&connections_lock);
684 	if (!dlm_allow_conn) {
685 		mutex_unlock(&connections_lock);
686 		return -1;
687 	}
688 	mutex_unlock(&connections_lock);
689 
690 	memset(&peeraddr, 0, sizeof(peeraddr));
691 	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
692 				  SOCK_STREAM, IPPROTO_TCP, &newsock);
693 	if (result < 0)
694 		return -ENOMEM;
695 
696 	mutex_lock_nested(&con->sock_mutex, 0);
697 
698 	result = -ENOTCONN;
699 	if (con->sock == NULL)
700 		goto accept_err;
701 
702 	newsock->type = con->sock->type;
703 	newsock->ops = con->sock->ops;
704 
705 	result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK);
706 	if (result < 0)
707 		goto accept_err;
708 
709 	/* Get the connected socket's peer */
710 	memset(&peeraddr, 0, sizeof(peeraddr));
711 	if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr,
712 				  &len, 2)) {
713 		result = -ECONNABORTED;
714 		goto accept_err;
715 	}
716 
717 	/* Get the new node's NODEID */
718 	make_sockaddr(&peeraddr, 0, &len);
719 	if (addr_to_nodeid(&peeraddr, &nodeid)) {
720 		unsigned char *b=(unsigned char *)&peeraddr;
721 		log_print("connect from non cluster node");
722 		print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
723 				     b, sizeof(struct sockaddr_storage));
724 		sock_release(newsock);
725 		mutex_unlock(&con->sock_mutex);
726 		return -1;
727 	}
728 
729 	log_print("got connection from %d", nodeid);
730 
731 	/*  Check to see if we already have a connection to this node. This
732 	 *  could happen if the two nodes initiate a connection at roughly
733 	 *  the same time and the connections cross on the wire.
734 	 *  In this case we store the incoming one in "othercon"
735 	 */
736 	newcon = nodeid2con(nodeid, GFP_NOFS);
737 	if (!newcon) {
738 		result = -ENOMEM;
739 		goto accept_err;
740 	}
741 	mutex_lock_nested(&newcon->sock_mutex, 1);
742 	if (newcon->sock) {
743 		struct connection *othercon = newcon->othercon;
744 
745 		if (!othercon) {
746 			othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
747 			if (!othercon) {
748 				log_print("failed to allocate incoming socket");
749 				mutex_unlock(&newcon->sock_mutex);
750 				result = -ENOMEM;
751 				goto accept_err;
752 			}
753 			othercon->nodeid = nodeid;
754 			othercon->rx_action = receive_from_sock;
755 			mutex_init(&othercon->sock_mutex);
756 			INIT_WORK(&othercon->swork, process_send_sockets);
757 			INIT_WORK(&othercon->rwork, process_recv_sockets);
758 			set_bit(CF_IS_OTHERCON, &othercon->flags);
759 		}
760 		if (!othercon->sock) {
761 			newcon->othercon = othercon;
762 			othercon->sock = newsock;
763 			newsock->sk->sk_user_data = othercon;
764 			add_sock(newsock, othercon);
765 			addcon = othercon;
766 		}
767 		else {
768 			printk("Extra connection from node %d attempted\n", nodeid);
769 			result = -EAGAIN;
770 			mutex_unlock(&newcon->sock_mutex);
771 			goto accept_err;
772 		}
773 	}
774 	else {
775 		newsock->sk->sk_user_data = newcon;
776 		newcon->rx_action = receive_from_sock;
777 		add_sock(newsock, newcon);
778 		addcon = newcon;
779 	}
780 
781 	mutex_unlock(&newcon->sock_mutex);
782 
783 	/*
784 	 * Add it to the active queue in case we got data
785 	 * between processing the accept adding the socket
786 	 * to the read_sockets list
787 	 */
788 	if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
789 		queue_work(recv_workqueue, &addcon->rwork);
790 	mutex_unlock(&con->sock_mutex);
791 
792 	return 0;
793 
794 accept_err:
795 	mutex_unlock(&con->sock_mutex);
796 	sock_release(newsock);
797 
798 	if (result != -EAGAIN)
799 		log_print("error accepting connection from node: %d", result);
800 	return result;
801 }
802 
803 static int sctp_accept_from_sock(struct connection *con)
804 {
805 	/* Check that the new node is in the lockspace */
806 	struct sctp_prim prim;
807 	int nodeid;
808 	int prim_len, ret;
809 	int addr_len;
810 	struct connection *newcon;
811 	struct connection *addcon;
812 	struct socket *newsock;
813 
814 	mutex_lock(&connections_lock);
815 	if (!dlm_allow_conn) {
816 		mutex_unlock(&connections_lock);
817 		return -1;
818 	}
819 	mutex_unlock(&connections_lock);
820 
821 	mutex_lock_nested(&con->sock_mutex, 0);
822 
823 	ret = kernel_accept(con->sock, &newsock, O_NONBLOCK);
824 	if (ret < 0)
825 		goto accept_err;
826 
827 	memset(&prim, 0, sizeof(struct sctp_prim));
828 	prim_len = sizeof(struct sctp_prim);
829 
830 	ret = kernel_getsockopt(newsock, IPPROTO_SCTP, SCTP_PRIMARY_ADDR,
831 				(char *)&prim, &prim_len);
832 	if (ret < 0) {
833 		log_print("getsockopt/sctp_primary_addr failed: %d", ret);
834 		goto accept_err;
835 	}
836 
837 	make_sockaddr(&prim.ssp_addr, 0, &addr_len);
838 	if (addr_to_nodeid(&prim.ssp_addr, &nodeid)) {
839 		unsigned char *b = (unsigned char *)&prim.ssp_addr;
840 
841 		log_print("reject connect from unknown addr");
842 		print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
843 				     b, sizeof(struct sockaddr_storage));
844 		goto accept_err;
845 	}
846 
847 	newcon = nodeid2con(nodeid, GFP_NOFS);
848 	if (!newcon) {
849 		ret = -ENOMEM;
850 		goto accept_err;
851 	}
852 
853 	mutex_lock_nested(&newcon->sock_mutex, 1);
854 
855 	if (newcon->sock) {
856 		struct connection *othercon = newcon->othercon;
857 
858 		if (!othercon) {
859 			othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
860 			if (!othercon) {
861 				log_print("failed to allocate incoming socket");
862 				mutex_unlock(&newcon->sock_mutex);
863 				ret = -ENOMEM;
864 				goto accept_err;
865 			}
866 			othercon->nodeid = nodeid;
867 			othercon->rx_action = receive_from_sock;
868 			mutex_init(&othercon->sock_mutex);
869 			INIT_WORK(&othercon->swork, process_send_sockets);
870 			INIT_WORK(&othercon->rwork, process_recv_sockets);
871 			set_bit(CF_IS_OTHERCON, &othercon->flags);
872 		}
873 		if (!othercon->sock) {
874 			newcon->othercon = othercon;
875 			othercon->sock = newsock;
876 			newsock->sk->sk_user_data = othercon;
877 			add_sock(newsock, othercon);
878 			addcon = othercon;
879 		} else {
880 			printk("Extra connection from node %d attempted\n", nodeid);
881 			ret = -EAGAIN;
882 			mutex_unlock(&newcon->sock_mutex);
883 			goto accept_err;
884 		}
885 	} else {
886 		newsock->sk->sk_user_data = newcon;
887 		newcon->rx_action = receive_from_sock;
888 		add_sock(newsock, newcon);
889 		addcon = newcon;
890 	}
891 
892 	log_print("connected to %d", nodeid);
893 
894 	mutex_unlock(&newcon->sock_mutex);
895 
896 	/*
897 	 * Add it to the active queue in case we got data
898 	 * between processing the accept adding the socket
899 	 * to the read_sockets list
900 	 */
901 	if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
902 		queue_work(recv_workqueue, &addcon->rwork);
903 	mutex_unlock(&con->sock_mutex);
904 
905 	return 0;
906 
907 accept_err:
908 	mutex_unlock(&con->sock_mutex);
909 	if (newsock)
910 		sock_release(newsock);
911 	if (ret != -EAGAIN)
912 		log_print("error accepting connection from node: %d", ret);
913 
914 	return ret;
915 }
916 
917 static void free_entry(struct writequeue_entry *e)
918 {
919 	__free_page(e->page);
920 	kfree(e);
921 }
922 
923 /*
924  * writequeue_entry_complete - try to delete and free write queue entry
925  * @e: write queue entry to try to delete
926  * @completed: bytes completed
927  *
928  * writequeue_lock must be held.
929  */
930 static void writequeue_entry_complete(struct writequeue_entry *e, int completed)
931 {
932 	e->offset += completed;
933 	e->len -= completed;
934 
935 	if (e->len == 0 && e->users == 0) {
936 		list_del(&e->list);
937 		free_entry(e);
938 	}
939 }
940 
941 /*
942  * sctp_bind_addrs - bind a SCTP socket to all our addresses
943  */
944 static int sctp_bind_addrs(struct connection *con, uint16_t port)
945 {
946 	struct sockaddr_storage localaddr;
947 	int i, addr_len, result = 0;
948 
949 	for (i = 0; i < dlm_local_count; i++) {
950 		memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr));
951 		make_sockaddr(&localaddr, port, &addr_len);
952 
953 		if (!i)
954 			result = kernel_bind(con->sock,
955 					     (struct sockaddr *)&localaddr,
956 					     addr_len);
957 		else
958 			result = kernel_setsockopt(con->sock, SOL_SCTP,
959 						   SCTP_SOCKOPT_BINDX_ADD,
960 						   (char *)&localaddr, addr_len);
961 
962 		if (result < 0) {
963 			log_print("Can't bind to %d addr number %d, %d.\n",
964 				  port, i + 1, result);
965 			break;
966 		}
967 	}
968 	return result;
969 }
970 
971 /* Initiate an SCTP association.
972    This is a special case of send_to_sock() in that we don't yet have a
973    peeled-off socket for this association, so we use the listening socket
974    and add the primary IP address of the remote node.
975  */
976 static void sctp_connect_to_sock(struct connection *con)
977 {
978 	struct sockaddr_storage daddr;
979 	int one = 1;
980 	int result;
981 	int addr_len;
982 	struct socket *sock;
983 
984 	if (con->nodeid == 0) {
985 		log_print("attempt to connect sock 0 foiled");
986 		return;
987 	}
988 
989 	mutex_lock(&con->sock_mutex);
990 
991 	/* Some odd races can cause double-connects, ignore them */
992 	if (con->retries++ > MAX_CONNECT_RETRIES)
993 		goto out;
994 
995 	if (con->sock) {
996 		log_print("node %d already connected.", con->nodeid);
997 		goto out;
998 	}
999 
1000 	memset(&daddr, 0, sizeof(daddr));
1001 	result = nodeid_to_addr(con->nodeid, &daddr, NULL, true);
1002 	if (result < 0) {
1003 		log_print("no address for nodeid %d", con->nodeid);
1004 		goto out;
1005 	}
1006 
1007 	/* Create a socket to communicate with */
1008 	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1009 				  SOCK_STREAM, IPPROTO_SCTP, &sock);
1010 	if (result < 0)
1011 		goto socket_err;
1012 
1013 	sock->sk->sk_user_data = con;
1014 	con->rx_action = receive_from_sock;
1015 	con->connect_action = sctp_connect_to_sock;
1016 	add_sock(sock, con);
1017 
1018 	/* Bind to all addresses. */
1019 	if (sctp_bind_addrs(con, 0))
1020 		goto bind_err;
1021 
1022 	make_sockaddr(&daddr, dlm_config.ci_tcp_port, &addr_len);
1023 
1024 	log_print("connecting to %d", con->nodeid);
1025 
1026 	/* Turn off Nagle's algorithm */
1027 	kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1028 			  sizeof(one));
1029 
1030 	result = sock->ops->connect(sock, (struct sockaddr *)&daddr, addr_len,
1031 				   O_NONBLOCK);
1032 	if (result == -EINPROGRESS)
1033 		result = 0;
1034 	if (result == 0)
1035 		goto out;
1036 
1037 
1038 bind_err:
1039 	con->sock = NULL;
1040 	sock_release(sock);
1041 
1042 socket_err:
1043 	/*
1044 	 * Some errors are fatal and this list might need adjusting. For other
1045 	 * errors we try again until the max number of retries is reached.
1046 	 */
1047 	if (result != -EHOSTUNREACH &&
1048 	    result != -ENETUNREACH &&
1049 	    result != -ENETDOWN &&
1050 	    result != -EINVAL &&
1051 	    result != -EPROTONOSUPPORT) {
1052 		log_print("connect %d try %d error %d", con->nodeid,
1053 			  con->retries, result);
1054 		mutex_unlock(&con->sock_mutex);
1055 		msleep(1000);
1056 		clear_bit(CF_CONNECT_PENDING, &con->flags);
1057 		lowcomms_connect_sock(con);
1058 		return;
1059 	}
1060 
1061 out:
1062 	mutex_unlock(&con->sock_mutex);
1063 	set_bit(CF_WRITE_PENDING, &con->flags);
1064 }
1065 
1066 /* Connect a new socket to its peer */
1067 static void tcp_connect_to_sock(struct connection *con)
1068 {
1069 	struct sockaddr_storage saddr, src_addr;
1070 	int addr_len;
1071 	struct socket *sock = NULL;
1072 	int one = 1;
1073 	int result;
1074 
1075 	if (con->nodeid == 0) {
1076 		log_print("attempt to connect sock 0 foiled");
1077 		return;
1078 	}
1079 
1080 	mutex_lock(&con->sock_mutex);
1081 	if (con->retries++ > MAX_CONNECT_RETRIES)
1082 		goto out;
1083 
1084 	/* Some odd races can cause double-connects, ignore them */
1085 	if (con->sock)
1086 		goto out;
1087 
1088 	/* Create a socket to communicate with */
1089 	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1090 				  SOCK_STREAM, IPPROTO_TCP, &sock);
1091 	if (result < 0)
1092 		goto out_err;
1093 
1094 	memset(&saddr, 0, sizeof(saddr));
1095 	result = nodeid_to_addr(con->nodeid, &saddr, NULL, false);
1096 	if (result < 0) {
1097 		log_print("no address for nodeid %d", con->nodeid);
1098 		goto out_err;
1099 	}
1100 
1101 	sock->sk->sk_user_data = con;
1102 	con->rx_action = receive_from_sock;
1103 	con->connect_action = tcp_connect_to_sock;
1104 	add_sock(sock, con);
1105 
1106 	/* Bind to our cluster-known address connecting to avoid
1107 	   routing problems */
1108 	memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
1109 	make_sockaddr(&src_addr, 0, &addr_len);
1110 	result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
1111 				 addr_len);
1112 	if (result < 0) {
1113 		log_print("could not bind for connect: %d", result);
1114 		/* This *may* not indicate a critical error */
1115 	}
1116 
1117 	make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
1118 
1119 	log_print("connecting to %d", con->nodeid);
1120 
1121 	/* Turn off Nagle's algorithm */
1122 	kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1123 			  sizeof(one));
1124 
1125 	result = sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
1126 				   O_NONBLOCK);
1127 	if (result == -EINPROGRESS)
1128 		result = 0;
1129 	if (result == 0)
1130 		goto out;
1131 
1132 out_err:
1133 	if (con->sock) {
1134 		sock_release(con->sock);
1135 		con->sock = NULL;
1136 	} else if (sock) {
1137 		sock_release(sock);
1138 	}
1139 	/*
1140 	 * Some errors are fatal and this list might need adjusting. For other
1141 	 * errors we try again until the max number of retries is reached.
1142 	 */
1143 	if (result != -EHOSTUNREACH &&
1144 	    result != -ENETUNREACH &&
1145 	    result != -ENETDOWN &&
1146 	    result != -EINVAL &&
1147 	    result != -EPROTONOSUPPORT) {
1148 		log_print("connect %d try %d error %d", con->nodeid,
1149 			  con->retries, result);
1150 		mutex_unlock(&con->sock_mutex);
1151 		msleep(1000);
1152 		clear_bit(CF_CONNECT_PENDING, &con->flags);
1153 		lowcomms_connect_sock(con);
1154 		return;
1155 	}
1156 out:
1157 	mutex_unlock(&con->sock_mutex);
1158 	set_bit(CF_WRITE_PENDING, &con->flags);
1159 	return;
1160 }
1161 
1162 static struct socket *tcp_create_listen_sock(struct connection *con,
1163 					     struct sockaddr_storage *saddr)
1164 {
1165 	struct socket *sock = NULL;
1166 	int result = 0;
1167 	int one = 1;
1168 	int addr_len;
1169 
1170 	if (dlm_local_addr[0]->ss_family == AF_INET)
1171 		addr_len = sizeof(struct sockaddr_in);
1172 	else
1173 		addr_len = sizeof(struct sockaddr_in6);
1174 
1175 	/* Create a socket to communicate with */
1176 	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1177 				  SOCK_STREAM, IPPROTO_TCP, &sock);
1178 	if (result < 0) {
1179 		log_print("Can't create listening comms socket");
1180 		goto create_out;
1181 	}
1182 
1183 	/* Turn off Nagle's algorithm */
1184 	kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1185 			  sizeof(one));
1186 
1187 	result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
1188 				   (char *)&one, sizeof(one));
1189 
1190 	if (result < 0) {
1191 		log_print("Failed to set SO_REUSEADDR on socket: %d", result);
1192 	}
1193 	con->rx_action = tcp_accept_from_sock;
1194 	con->connect_action = tcp_connect_to_sock;
1195 
1196 	/* Bind to our port */
1197 	make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
1198 	result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
1199 	if (result < 0) {
1200 		log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
1201 		sock_release(sock);
1202 		sock = NULL;
1203 		con->sock = NULL;
1204 		goto create_out;
1205 	}
1206 	result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
1207 				 (char *)&one, sizeof(one));
1208 	if (result < 0) {
1209 		log_print("Set keepalive failed: %d", result);
1210 	}
1211 
1212 	result = sock->ops->listen(sock, 5);
1213 	if (result < 0) {
1214 		log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
1215 		sock_release(sock);
1216 		sock = NULL;
1217 		goto create_out;
1218 	}
1219 
1220 create_out:
1221 	return sock;
1222 }
1223 
1224 /* Get local addresses */
1225 static void init_local(void)
1226 {
1227 	struct sockaddr_storage sas, *addr;
1228 	int i;
1229 
1230 	dlm_local_count = 0;
1231 	for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
1232 		if (dlm_our_addr(&sas, i))
1233 			break;
1234 
1235 		addr = kmalloc(sizeof(*addr), GFP_NOFS);
1236 		if (!addr)
1237 			break;
1238 		memcpy(addr, &sas, sizeof(*addr));
1239 		dlm_local_addr[dlm_local_count++] = addr;
1240 	}
1241 }
1242 
1243 /* Initialise SCTP socket and bind to all interfaces */
1244 static int sctp_listen_for_all(void)
1245 {
1246 	struct socket *sock = NULL;
1247 	int result = -EINVAL;
1248 	struct connection *con = nodeid2con(0, GFP_NOFS);
1249 	int bufsize = NEEDED_RMEM;
1250 	int one = 1;
1251 
1252 	if (!con)
1253 		return -ENOMEM;
1254 
1255 	log_print("Using SCTP for communications");
1256 
1257 	result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1258 				  SOCK_STREAM, IPPROTO_SCTP, &sock);
1259 	if (result < 0) {
1260 		log_print("Can't create comms socket, check SCTP is loaded");
1261 		goto out;
1262 	}
1263 
1264 	result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
1265 				 (char *)&bufsize, sizeof(bufsize));
1266 	if (result)
1267 		log_print("Error increasing buffer space on socket %d", result);
1268 
1269 	result = kernel_setsockopt(sock, SOL_SCTP, SCTP_NODELAY, (char *)&one,
1270 				   sizeof(one));
1271 	if (result < 0)
1272 		log_print("Could not set SCTP NODELAY error %d\n", result);
1273 
1274 	/* Init con struct */
1275 	sock->sk->sk_user_data = con;
1276 	con->sock = sock;
1277 	con->sock->sk->sk_data_ready = lowcomms_data_ready;
1278 	con->rx_action = sctp_accept_from_sock;
1279 	con->connect_action = sctp_connect_to_sock;
1280 
1281 	/* Bind to all addresses. */
1282 	if (sctp_bind_addrs(con, dlm_config.ci_tcp_port))
1283 		goto create_delsock;
1284 
1285 	result = sock->ops->listen(sock, 5);
1286 	if (result < 0) {
1287 		log_print("Can't set socket listening");
1288 		goto create_delsock;
1289 	}
1290 
1291 	return 0;
1292 
1293 create_delsock:
1294 	sock_release(sock);
1295 	con->sock = NULL;
1296 out:
1297 	return result;
1298 }
1299 
1300 static int tcp_listen_for_all(void)
1301 {
1302 	struct socket *sock = NULL;
1303 	struct connection *con = nodeid2con(0, GFP_NOFS);
1304 	int result = -EINVAL;
1305 
1306 	if (!con)
1307 		return -ENOMEM;
1308 
1309 	/* We don't support multi-homed hosts */
1310 	if (dlm_local_addr[1] != NULL) {
1311 		log_print("TCP protocol can't handle multi-homed hosts, "
1312 			  "try SCTP");
1313 		return -EINVAL;
1314 	}
1315 
1316 	log_print("Using TCP for communications");
1317 
1318 	sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
1319 	if (sock) {
1320 		add_sock(sock, con);
1321 		result = 0;
1322 	}
1323 	else {
1324 		result = -EADDRINUSE;
1325 	}
1326 
1327 	return result;
1328 }
1329 
1330 
1331 
1332 static struct writequeue_entry *new_writequeue_entry(struct connection *con,
1333 						     gfp_t allocation)
1334 {
1335 	struct writequeue_entry *entry;
1336 
1337 	entry = kmalloc(sizeof(struct writequeue_entry), allocation);
1338 	if (!entry)
1339 		return NULL;
1340 
1341 	entry->page = alloc_page(allocation);
1342 	if (!entry->page) {
1343 		kfree(entry);
1344 		return NULL;
1345 	}
1346 
1347 	entry->offset = 0;
1348 	entry->len = 0;
1349 	entry->end = 0;
1350 	entry->users = 0;
1351 	entry->con = con;
1352 
1353 	return entry;
1354 }
1355 
1356 void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
1357 {
1358 	struct connection *con;
1359 	struct writequeue_entry *e;
1360 	int offset = 0;
1361 
1362 	con = nodeid2con(nodeid, allocation);
1363 	if (!con)
1364 		return NULL;
1365 
1366 	spin_lock(&con->writequeue_lock);
1367 	e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
1368 	if ((&e->list == &con->writequeue) ||
1369 	    (PAGE_CACHE_SIZE - e->end < len)) {
1370 		e = NULL;
1371 	} else {
1372 		offset = e->end;
1373 		e->end += len;
1374 		e->users++;
1375 	}
1376 	spin_unlock(&con->writequeue_lock);
1377 
1378 	if (e) {
1379 	got_one:
1380 		*ppc = page_address(e->page) + offset;
1381 		return e;
1382 	}
1383 
1384 	e = new_writequeue_entry(con, allocation);
1385 	if (e) {
1386 		spin_lock(&con->writequeue_lock);
1387 		offset = e->end;
1388 		e->end += len;
1389 		e->users++;
1390 		list_add_tail(&e->list, &con->writequeue);
1391 		spin_unlock(&con->writequeue_lock);
1392 		goto got_one;
1393 	}
1394 	return NULL;
1395 }
1396 
1397 void dlm_lowcomms_commit_buffer(void *mh)
1398 {
1399 	struct writequeue_entry *e = (struct writequeue_entry *)mh;
1400 	struct connection *con = e->con;
1401 	int users;
1402 
1403 	spin_lock(&con->writequeue_lock);
1404 	users = --e->users;
1405 	if (users)
1406 		goto out;
1407 	e->len = e->end - e->offset;
1408 	spin_unlock(&con->writequeue_lock);
1409 
1410 	if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {
1411 		queue_work(send_workqueue, &con->swork);
1412 	}
1413 	return;
1414 
1415 out:
1416 	spin_unlock(&con->writequeue_lock);
1417 	return;
1418 }
1419 
1420 /* Send a message */
1421 static void send_to_sock(struct connection *con)
1422 {
1423 	int ret = 0;
1424 	const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
1425 	struct writequeue_entry *e;
1426 	int len, offset;
1427 	int count = 0;
1428 
1429 	mutex_lock(&con->sock_mutex);
1430 	if (con->sock == NULL)
1431 		goto out_connect;
1432 
1433 	spin_lock(&con->writequeue_lock);
1434 	for (;;) {
1435 		e = list_entry(con->writequeue.next, struct writequeue_entry,
1436 			       list);
1437 		if ((struct list_head *) e == &con->writequeue)
1438 			break;
1439 
1440 		len = e->len;
1441 		offset = e->offset;
1442 		BUG_ON(len == 0 && e->users == 0);
1443 		spin_unlock(&con->writequeue_lock);
1444 
1445 		ret = 0;
1446 		if (len) {
1447 			ret = kernel_sendpage(con->sock, e->page, offset, len,
1448 					      msg_flags);
1449 			if (ret == -EAGAIN || ret == 0) {
1450 				if (ret == -EAGAIN &&
1451 				    test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) &&
1452 				    !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1453 					/* Notify TCP that we're limited by the
1454 					 * application window size.
1455 					 */
1456 					set_bit(SOCK_NOSPACE, &con->sock->flags);
1457 					con->sock->sk->sk_write_pending++;
1458 				}
1459 				cond_resched();
1460 				goto out;
1461 			} else if (ret < 0)
1462 				goto send_error;
1463 		}
1464 
1465 		/* Don't starve people filling buffers */
1466 		if (++count >= MAX_SEND_MSG_COUNT) {
1467 			cond_resched();
1468 			count = 0;
1469 		}
1470 
1471 		spin_lock(&con->writequeue_lock);
1472 		writequeue_entry_complete(e, ret);
1473 	}
1474 	spin_unlock(&con->writequeue_lock);
1475 out:
1476 	mutex_unlock(&con->sock_mutex);
1477 	return;
1478 
1479 send_error:
1480 	mutex_unlock(&con->sock_mutex);
1481 	close_connection(con, false, false, true);
1482 	lowcomms_connect_sock(con);
1483 	return;
1484 
1485 out_connect:
1486 	mutex_unlock(&con->sock_mutex);
1487 	lowcomms_connect_sock(con);
1488 }
1489 
1490 static void clean_one_writequeue(struct connection *con)
1491 {
1492 	struct writequeue_entry *e, *safe;
1493 
1494 	spin_lock(&con->writequeue_lock);
1495 	list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1496 		list_del(&e->list);
1497 		free_entry(e);
1498 	}
1499 	spin_unlock(&con->writequeue_lock);
1500 }
1501 
1502 /* Called from recovery when it knows that a node has
1503    left the cluster */
1504 int dlm_lowcomms_close(int nodeid)
1505 {
1506 	struct connection *con;
1507 	struct dlm_node_addr *na;
1508 
1509 	log_print("closing connection to node %d", nodeid);
1510 	con = nodeid2con(nodeid, 0);
1511 	if (con) {
1512 		set_bit(CF_CLOSE, &con->flags);
1513 		close_connection(con, true, true, true);
1514 		clean_one_writequeue(con);
1515 	}
1516 
1517 	spin_lock(&dlm_node_addrs_spin);
1518 	na = find_node_addr(nodeid);
1519 	if (na) {
1520 		list_del(&na->list);
1521 		while (na->addr_count--)
1522 			kfree(na->addr[na->addr_count]);
1523 		kfree(na);
1524 	}
1525 	spin_unlock(&dlm_node_addrs_spin);
1526 
1527 	return 0;
1528 }
1529 
1530 /* Receive workqueue function */
1531 static void process_recv_sockets(struct work_struct *work)
1532 {
1533 	struct connection *con = container_of(work, struct connection, rwork);
1534 	int err;
1535 
1536 	clear_bit(CF_READ_PENDING, &con->flags);
1537 	do {
1538 		err = con->rx_action(con);
1539 	} while (!err);
1540 }
1541 
1542 /* Send workqueue function */
1543 static void process_send_sockets(struct work_struct *work)
1544 {
1545 	struct connection *con = container_of(work, struct connection, swork);
1546 
1547 	if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags))
1548 		con->connect_action(con);
1549 	if (test_and_clear_bit(CF_WRITE_PENDING, &con->flags))
1550 		send_to_sock(con);
1551 }
1552 
1553 
1554 /* Discard all entries on the write queues */
1555 static void clean_writequeues(void)
1556 {
1557 	foreach_conn(clean_one_writequeue);
1558 }
1559 
1560 static void work_stop(void)
1561 {
1562 	destroy_workqueue(recv_workqueue);
1563 	destroy_workqueue(send_workqueue);
1564 }
1565 
1566 static int work_start(void)
1567 {
1568 	recv_workqueue = alloc_workqueue("dlm_recv",
1569 					 WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1570 	if (!recv_workqueue) {
1571 		log_print("can't start dlm_recv");
1572 		return -ENOMEM;
1573 	}
1574 
1575 	send_workqueue = alloc_workqueue("dlm_send",
1576 					 WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1577 	if (!send_workqueue) {
1578 		log_print("can't start dlm_send");
1579 		destroy_workqueue(recv_workqueue);
1580 		return -ENOMEM;
1581 	}
1582 
1583 	return 0;
1584 }
1585 
1586 static void stop_conn(struct connection *con)
1587 {
1588 	con->flags |= 0x0F;
1589 	if (con->sock && con->sock->sk)
1590 		con->sock->sk->sk_user_data = NULL;
1591 }
1592 
1593 static void free_conn(struct connection *con)
1594 {
1595 	close_connection(con, true, true, true);
1596 	if (con->othercon)
1597 		kmem_cache_free(con_cache, con->othercon);
1598 	hlist_del(&con->list);
1599 	kmem_cache_free(con_cache, con);
1600 }
1601 
1602 void dlm_lowcomms_stop(void)
1603 {
1604 	/* Set all the flags to prevent any
1605 	   socket activity.
1606 	*/
1607 	mutex_lock(&connections_lock);
1608 	dlm_allow_conn = 0;
1609 	foreach_conn(stop_conn);
1610 	mutex_unlock(&connections_lock);
1611 
1612 	work_stop();
1613 
1614 	mutex_lock(&connections_lock);
1615 	clean_writequeues();
1616 
1617 	foreach_conn(free_conn);
1618 
1619 	mutex_unlock(&connections_lock);
1620 	kmem_cache_destroy(con_cache);
1621 }
1622 
1623 int dlm_lowcomms_start(void)
1624 {
1625 	int error = -EINVAL;
1626 	struct connection *con;
1627 	int i;
1628 
1629 	for (i = 0; i < CONN_HASH_SIZE; i++)
1630 		INIT_HLIST_HEAD(&connection_hash[i]);
1631 
1632 	init_local();
1633 	if (!dlm_local_count) {
1634 		error = -ENOTCONN;
1635 		log_print("no local IP address has been set");
1636 		goto fail;
1637 	}
1638 
1639 	error = -ENOMEM;
1640 	con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
1641 				      __alignof__(struct connection), 0,
1642 				      NULL);
1643 	if (!con_cache)
1644 		goto fail;
1645 
1646 	error = work_start();
1647 	if (error)
1648 		goto fail_destroy;
1649 
1650 	dlm_allow_conn = 1;
1651 
1652 	/* Start listening */
1653 	if (dlm_config.ci_protocol == 0)
1654 		error = tcp_listen_for_all();
1655 	else
1656 		error = sctp_listen_for_all();
1657 	if (error)
1658 		goto fail_unlisten;
1659 
1660 	return 0;
1661 
1662 fail_unlisten:
1663 	dlm_allow_conn = 0;
1664 	con = nodeid2con(0,0);
1665 	if (con) {
1666 		close_connection(con, false, true, true);
1667 		kmem_cache_free(con_cache, con);
1668 	}
1669 fail_destroy:
1670 	kmem_cache_destroy(con_cache);
1671 fail:
1672 	return error;
1673 }
1674 
1675 void dlm_lowcomms_exit(void)
1676 {
1677 	struct dlm_node_addr *na, *safe;
1678 
1679 	spin_lock(&dlm_node_addrs_spin);
1680 	list_for_each_entry_safe(na, safe, &dlm_node_addrs, list) {
1681 		list_del(&na->list);
1682 		while (na->addr_count--)
1683 			kfree(na->addr[na->addr_count]);
1684 		kfree(na);
1685 	}
1686 	spin_unlock(&dlm_node_addrs_spin);
1687 }
1688