squashfs: xattr simplifications
[linux-2.6-block.git] / fs / dlm / lowcomms.c
CommitLineData
fdda387f
PC
1/******************************************************************************
2*******************************************************************************
3**
4** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5e9ccc37 5** Copyright (C) 2004-2009 Red Hat, Inc. All rights reserved.
fdda387f
PC
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
2cf12c0b 24 * be expanded for the cluster infrastructure then that is its
fdda387f
PC
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 *
2cf12c0b 39 * lowcomms will choose to use either TCP or SCTP as its transport layer
6ed7257b 40 * depending on the configuration variable 'protocol'. This should be set
2cf12c0b 41 * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
6ed7257b
PC
42 * cluster-wide mechanism as it must be the same on all nodes of the cluster
43 * for the DLM to function.
fdda387f
PC
44 *
45 */
46
fdda387f
PC
47#include <asm/ioctls.h>
48#include <net/sock.h>
49#include <net/tcp.h>
50#include <linux/pagemap.h>
6ed7257b 51#include <linux/file.h>
7a936ce7 52#include <linux/mutex.h>
6ed7257b 53#include <linux/sctp.h>
5a0e3ad6 54#include <linux/slab.h>
2f2d76cc 55#include <net/sctp/sctp.h>
44ad532b 56#include <net/ipv6.h>
fdda387f
PC
57
58#include "dlm_internal.h"
59#include "lowcomms.h"
60#include "midcomms.h"
61#include "config.h"
62
6ed7257b 63#define NEEDED_RMEM (4*1024*1024)
5e9ccc37 64#define CONN_HASH_SIZE 32
6ed7257b 65
f92c8dd7
BP
66/* Number of messages to send before rescheduling */
67#define MAX_SEND_MSG_COUNT 25
68
fdda387f 69struct cbuf {
ac33d071
PC
70 unsigned int base;
71 unsigned int len;
72 unsigned int mask;
fdda387f
PC
73};
74
ac33d071
PC
75static void cbuf_add(struct cbuf *cb, int n)
76{
77 cb->len += n;
78}
fdda387f 79
ac33d071
PC
80static int cbuf_data(struct cbuf *cb)
81{
82 return ((cb->base + cb->len) & cb->mask);
83}
84
85static void cbuf_init(struct cbuf *cb, int size)
86{
87 cb->base = cb->len = 0;
88 cb->mask = size-1;
89}
90
91static void cbuf_eat(struct cbuf *cb, int n)
92{
93 cb->len -= n;
94 cb->base += n;
95 cb->base &= cb->mask;
96}
97
98static bool cbuf_empty(struct cbuf *cb)
99{
100 return cb->len == 0;
101}
fdda387f 102
fdda387f
PC
103struct connection {
104 struct socket *sock; /* NULL if not connected */
105 uint32_t nodeid; /* So we know who we are in the list */
f1f1c1cc 106 struct mutex sock_mutex;
6ed7257b 107 unsigned long flags;
fdda387f
PC
108#define CF_READ_PENDING 1
109#define CF_WRITE_PENDING 2
110#define CF_CONNECT_PENDING 3
6ed7257b
PC
111#define CF_INIT_PENDING 4
112#define CF_IS_OTHERCON 5
063c4c99 113#define CF_CLOSE 6
b36930dd 114#define CF_APP_LIMITED 7
ac33d071 115 struct list_head writequeue; /* List of outgoing writequeue_entries */
fdda387f
PC
116 spinlock_t writequeue_lock;
117 int (*rx_action) (struct connection *); /* What to do when active */
6ed7257b 118 void (*connect_action) (struct connection *); /* What to do to connect */
fdda387f
PC
119 struct page *rx_page;
120 struct cbuf cb;
121 int retries;
fdda387f 122#define MAX_CONNECT_RETRIES 3
5e9ccc37 123 struct hlist_node list;
fdda387f 124 struct connection *othercon;
1d6e8131
PC
125 struct work_struct rwork; /* Receive workqueue */
126 struct work_struct swork; /* Send workqueue */
b3a5bbfd 127 void (*orig_error_report)(struct sock *sk);
fdda387f
PC
128};
129#define sock2con(x) ((struct connection *)(x)->sk_user_data)
130
131/* An entry waiting to be sent */
132struct 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
36b71a8b
DT
142struct dlm_node_addr {
143 struct list_head list;
144 int nodeid;
145 int addr_count;
98e1b60e 146 int curr_addr_index;
36b71a8b
DT
147 struct sockaddr_storage *addr[DLM_MAX_ADDR_COUNT];
148};
149
150static LIST_HEAD(dlm_node_addrs);
151static DEFINE_SPINLOCK(dlm_node_addrs_spin);
152
6ed7257b
PC
153static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
154static int dlm_local_count;
513ef596 155static int dlm_allow_conn;
fdda387f 156
1d6e8131
PC
157/* Work queues */
158static struct workqueue_struct *recv_workqueue;
159static struct workqueue_struct *send_workqueue;
fdda387f 160
5e9ccc37 161static struct hlist_head connection_hash[CONN_HASH_SIZE];
7a936ce7 162static DEFINE_MUTEX(connections_lock);
c80e7c83 163static struct kmem_cache *con_cache;
fdda387f 164
1d6e8131
PC
165static void process_recv_sockets(struct work_struct *work);
166static void process_send_sockets(struct work_struct *work);
fdda387f 167
5e9ccc37
CC
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 */
172static inline int nodeid_hash(int nodeid)
173{
174 return nodeid & (CONN_HASH_SIZE-1);
175}
176
177static struct connection *__find_con(int nodeid)
178{
179 int r;
5e9ccc37
CC
180 struct connection *con;
181
182 r = nodeid_hash(nodeid);
183
b67bfe0d 184 hlist_for_each_entry(con, &connection_hash[r], list) {
5e9ccc37
CC
185 if (con->nodeid == nodeid)
186 return con;
187 }
188 return NULL;
189}
190
6ed7257b
PC
191/*
192 * If 'allocation' is zero then we don't attempt to create a new
193 * connection structure for this node.
194 */
195static struct connection *__nodeid2con(int nodeid, gfp_t alloc)
fdda387f
PC
196{
197 struct connection *con = NULL;
6ed7257b 198 int r;
fdda387f 199
5e9ccc37 200 con = __find_con(nodeid);
6ed7257b
PC
201 if (con || !alloc)
202 return con;
fdda387f 203
6ed7257b
PC
204 con = kmem_cache_zalloc(con_cache, alloc);
205 if (!con)
206 return NULL;
fdda387f 207
5e9ccc37
CC
208 r = nodeid_hash(nodeid);
209 hlist_add_head(&con->list, &connection_hash[r]);
fdda387f 210
6ed7257b
PC
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);
fdda387f 217
6ed7257b
PC
218 /* Setup action pointers for child sockets */
219 if (con->nodeid) {
5e9ccc37 220 struct connection *zerocon = __find_con(0);
fdda387f 221
6ed7257b
PC
222 con->connect_action = zerocon->connect_action;
223 if (!con->rx_action)
224 con->rx_action = zerocon->rx_action;
fdda387f
PC
225 }
226
6ed7257b
PC
227 return con;
228}
229
5e9ccc37
CC
230/* Loop round all connections */
231static void foreach_conn(void (*conn_func)(struct connection *c))
232{
233 int i;
b67bfe0d 234 struct hlist_node *n;
5e9ccc37
CC
235 struct connection *con;
236
237 for (i = 0; i < CONN_HASH_SIZE; i++) {
b67bfe0d 238 hlist_for_each_entry_safe(con, n, &connection_hash[i], list)
5e9ccc37 239 conn_func(con);
5e9ccc37
CC
240 }
241}
242
6ed7257b
PC
243static struct connection *nodeid2con(int nodeid, gfp_t allocation)
244{
245 struct connection *con;
246
7a936ce7 247 mutex_lock(&connections_lock);
6ed7257b 248 con = __nodeid2con(nodeid, allocation);
7a936ce7 249 mutex_unlock(&connections_lock);
6ed7257b 250
fdda387f
PC
251 return con;
252}
253
36b71a8b
DT
254static 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
265static int addr_compare(struct sockaddr_storage *x, struct sockaddr_storage *y)
6ed7257b 266{
36b71a8b
DT
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
292static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
98e1b60e 293 struct sockaddr *sa_out, bool try_new_addr)
36b71a8b
DT
294{
295 struct sockaddr_storage sas;
296 struct dlm_node_addr *na;
6ed7257b
PC
297
298 if (!dlm_local_count)
299 return -1;
300
36b71a8b
DT
301 spin_lock(&dlm_node_addrs_spin);
302 na = find_node_addr(nodeid);
98e1b60e 303 if (na && na->addr_count) {
ee44b4bc
MRL
304 memcpy(&sas, na->addr[na->curr_addr_index],
305 sizeof(struct sockaddr_storage));
306
98e1b60e
MC
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 }
98e1b60e 312 }
36b71a8b
DT
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;
6ed7257b
PC
326
327 if (dlm_local_addr[0]->ss_family == AF_INET) {
36b71a8b
DT
328 struct sockaddr_in *in4 = (struct sockaddr_in *) &sas;
329 struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;
6ed7257b
PC
330 ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
331 } else {
36b71a8b
DT
332 struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &sas;
333 struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;
4e3fd7a0 334 ret6->sin6_addr = in6->sin6_addr;
6ed7257b
PC
335 }
336
337 return 0;
338}
339
36b71a8b
DT
340static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid)
341{
342 struct dlm_node_addr *na;
343 int rv = -EEXIST;
98e1b60e 344 int addr_i;
36b71a8b
DT
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
98e1b60e
MC
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 }
36b71a8b 358 }
98e1b60e 359unlock:
36b71a8b
DT
360 spin_unlock(&dlm_node_addrs_spin);
361 return rv;
362}
363
364int 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
fdda387f 405/* Data available on socket or listen socket received a connect */
676d2369 406static void lowcomms_data_ready(struct sock *sk)
fdda387f
PC
407{
408 struct connection *con = sock2con(sk);
afb853fb 409 if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
1d6e8131 410 queue_work(recv_workqueue, &con->rwork);
fdda387f
PC
411}
412
413static void lowcomms_write_space(struct sock *sk)
414{
415 struct connection *con = sock2con(sk);
416
b36930dd
DM
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(SOCK_ASYNC_NOSPACE, &con->sock->flags);
425 }
426
427 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
1d6e8131 428 queue_work(send_workqueue, &con->swork);
fdda387f
PC
429}
430
431static inline void lowcomms_connect_sock(struct connection *con)
432{
063c4c99
LMB
433 if (test_bit(CF_CLOSE, &con->flags))
434 return;
1d6e8131
PC
435 if (!test_and_set_bit(CF_CONNECT_PENDING, &con->flags))
436 queue_work(send_workqueue, &con->swork);
fdda387f
PC
437}
438
439static void lowcomms_state_change(struct sock *sk)
440{
ee44b4bc
MRL
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) {
fdda387f 450 lowcomms_write_space(sk);
ee44b4bc 451 }
fdda387f
PC
452}
453
391fbdc5
CC
454int 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
b3a5bbfd
BP
468static 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
fdda387f 505/* Make a socket active */
4dd40f0c 506static void add_sock(struct socket *sock, struct connection *con)
fdda387f
PC
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;
6ed7257b 514 con->sock->sk->sk_user_data = con;
d6d7b702 515 con->sock->sk->sk_allocation = GFP_NOFS;
b3a5bbfd
BP
516 con->orig_error_report = con->sock->sk->sk_error_report;
517 con->sock->sk->sk_error_report = lowcomms_error_report;
fdda387f
PC
518}
519
6ed7257b 520/* Add the port number to an IPv6 or 4 sockaddr and return the address
fdda387f
PC
521 length */
522static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
523 int *addr_len)
524{
6ed7257b 525 saddr->ss_family = dlm_local_addr[0]->ss_family;
ac33d071 526 if (saddr->ss_family == AF_INET) {
fdda387f
PC
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);
6ed7257b 530 memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
ac33d071 531 } else {
fdda387f
PC
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 }
01c8cab2 536 memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
fdda387f
PC
537}
538
539/* Close a remote connection and tidy up */
0d737a8c
MRL
540static void close_connection(struct connection *con, bool and_other,
541 bool tx, bool rx)
fdda387f 542{
0d737a8c
MRL
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);
fdda387f 549
0d737a8c 550 mutex_lock(&con->sock_mutex);
fdda387f
PC
551 if (con->sock) {
552 sock_release(con->sock);
553 con->sock = NULL;
554 }
555 if (con->othercon && and_other) {
ac33d071 556 /* Will only re-enter once. */
0d737a8c 557 close_connection(con->othercon, false, true, true);
fdda387f
PC
558 }
559 if (con->rx_page) {
560 __free_page(con->rx_page);
561 con->rx_page = NULL;
562 }
9e5f2825 563
61d96be0
PC
564 con->retries = 0;
565 mutex_unlock(&con->sock_mutex);
fdda387f
PC
566}
567
568/* Data received from remote end */
569static int receive_from_sock(struct connection *con)
570{
571 int ret = 0;
58addbff
AV
572 struct msghdr msg = {};
573 struct kvec iov[2];
fdda387f
PC
574 unsigned len;
575 int r;
576 int call_again_soon = 0;
58addbff 577 int nvec;
fdda387f 578
f1f1c1cc 579 mutex_lock(&con->sock_mutex);
fdda387f 580
a34fbc63
PC
581 if (con->sock == NULL) {
582 ret = -EAGAIN;
583 goto out_close;
584 }
acee4e52
MRL
585 if (con->nodeid == 0) {
586 ret = -EINVAL;
587 goto out_close;
588 }
a34fbc63 589
fdda387f
PC
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;
ac33d071 598 cbuf_init(&con->cb, PAGE_CACHE_SIZE);
fdda387f
PC
599 }
600
fdda387f
PC
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 */
ac33d071
PC
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);
89adc934 607 iov[1].iov_len = 0;
58addbff 608 nvec = 1;
fdda387f
PC
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 */
ac33d071
PC
614 if (cbuf_data(&con->cb) >= con->cb.base) {
615 iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);
fdda387f
PC
616 iov[1].iov_len = con->cb.base;
617 iov[1].iov_base = page_address(con->rx_page);
58addbff 618 nvec = 2;
fdda387f
PC
619 }
620 len = iov[0].iov_len + iov[1].iov_len;
621
58addbff 622 r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,
fdda387f 623 MSG_DONTWAIT | MSG_NOSIGNAL);
fdda387f
PC
624 if (ret <= 0)
625 goto out_close;
ee44b4bc
MRL
626 else if (ret == len)
627 call_again_soon = 1;
bd44e2b0 628
ac33d071 629 cbuf_add(&con->cb, ret);
fdda387f
PC
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) {
ee44b4bc
MRL
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);
fdda387f
PC
638 }
639 if (ret < 0)
640 goto out_close;
ac33d071 641 cbuf_eat(&con->cb, ret);
fdda387f 642
ac33d071 643 if (cbuf_empty(&con->cb) && !call_again_soon) {
fdda387f
PC
644 __free_page(con->rx_page);
645 con->rx_page = NULL;
646 }
647
fdda387f
PC
648 if (call_again_soon)
649 goto out_resched;
f1f1c1cc 650 mutex_unlock(&con->sock_mutex);
ac33d071 651 return 0;
fdda387f 652
ac33d071 653out_resched:
1d6e8131
PC
654 if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
655 queue_work(recv_workqueue, &con->rwork);
f1f1c1cc 656 mutex_unlock(&con->sock_mutex);
bd44e2b0 657 return -EAGAIN;
fdda387f 658
ac33d071 659out_close:
f1f1c1cc 660 mutex_unlock(&con->sock_mutex);
9e5f2825 661 if (ret != -EAGAIN) {
0d737a8c 662 close_connection(con, false, true, false);
fdda387f
PC
663 /* Reconnect when there is something to send */
664 }
a34fbc63
PC
665 /* Don't return success if we really got EOF */
666 if (ret == 0)
667 ret = -EAGAIN;
fdda387f 668
fdda387f
PC
669 return ret;
670}
671
672/* Listening socket is busy, accept a connection */
6ed7257b 673static int tcp_accept_from_sock(struct connection *con)
fdda387f
PC
674{
675 int result;
676 struct sockaddr_storage peeraddr;
677 struct socket *newsock;
678 int len;
679 int nodeid;
680 struct connection *newcon;
bd44e2b0 681 struct connection *addcon;
fdda387f 682
513ef596
DT
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
fdda387f 690 memset(&peeraddr, 0, sizeof(peeraddr));
eeb1bd5c
EB
691 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
692 SOCK_STREAM, IPPROTO_TCP, &newsock);
fdda387f
PC
693 if (result < 0)
694 return -ENOMEM;
695
f1f1c1cc 696 mutex_lock_nested(&con->sock_mutex, 0);
fdda387f
PC
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);
36b71a8b 719 if (addr_to_nodeid(&peeraddr, &nodeid)) {
bcaadf5c 720 unsigned char *b=(unsigned char *)&peeraddr;
617e82e1 721 log_print("connect from non cluster node");
bcaadf5c
MY
722 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
723 b, sizeof(struct sockaddr_storage));
fdda387f 724 sock_release(newsock);
f1f1c1cc 725 mutex_unlock(&con->sock_mutex);
fdda387f
PC
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.
fdda387f
PC
734 * In this case we store the incoming one in "othercon"
735 */
748285cc 736 newcon = nodeid2con(nodeid, GFP_NOFS);
fdda387f
PC
737 if (!newcon) {
738 result = -ENOMEM;
739 goto accept_err;
740 }
f1f1c1cc 741 mutex_lock_nested(&newcon->sock_mutex, 1);
fdda387f 742 if (newcon->sock) {
ac33d071 743 struct connection *othercon = newcon->othercon;
fdda387f
PC
744
745 if (!othercon) {
748285cc 746 othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
fdda387f 747 if (!othercon) {
617e82e1 748 log_print("failed to allocate incoming socket");
f1f1c1cc 749 mutex_unlock(&newcon->sock_mutex);
fdda387f
PC
750 result = -ENOMEM;
751 goto accept_err;
752 }
fdda387f
PC
753 othercon->nodeid = nodeid;
754 othercon->rx_action = receive_from_sock;
f1f1c1cc 755 mutex_init(&othercon->sock_mutex);
1d6e8131
PC
756 INIT_WORK(&othercon->swork, process_send_sockets);
757 INIT_WORK(&othercon->rwork, process_recv_sockets);
fdda387f 758 set_bit(CF_IS_OTHERCON, &othercon->flags);
61d96be0
PC
759 }
760 if (!othercon->sock) {
fdda387f 761 newcon->othercon = othercon;
97d84836
PC
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;
f4fadb23 770 mutex_unlock(&newcon->sock_mutex);
97d84836 771 goto accept_err;
fdda387f 772 }
fdda387f
PC
773 }
774 else {
775 newsock->sk->sk_user_data = newcon;
776 newcon->rx_action = receive_from_sock;
777 add_sock(newsock, newcon);
bd44e2b0 778 addcon = newcon;
fdda387f
PC
779 }
780
f1f1c1cc 781 mutex_unlock(&newcon->sock_mutex);
fdda387f
PC
782
783 /*
784 * Add it to the active queue in case we got data
25985edc 785 * between processing the accept adding the socket
fdda387f
PC
786 * to the read_sockets list
787 */
bd44e2b0
PC
788 if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
789 queue_work(recv_workqueue, &addcon->rwork);
f1f1c1cc 790 mutex_unlock(&con->sock_mutex);
fdda387f
PC
791
792 return 0;
793
ac33d071 794accept_err:
f1f1c1cc 795 mutex_unlock(&con->sock_mutex);
fdda387f
PC
796 sock_release(newsock);
797
798 if (result != -EAGAIN)
617e82e1 799 log_print("error accepting connection from node: %d", result);
fdda387f
PC
800 return result;
801}
802
18df8a87 803static int sctp_accept_from_sock(struct connection *con)
ee44b4bc
MRL
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
907accept_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
6ed7257b
PC
917static void free_entry(struct writequeue_entry *e)
918{
919 __free_page(e->page);
920 kfree(e);
921}
922
5d689871
MC
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 */
930static 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
ee44b4bc
MRL
941/*
942 * sctp_bind_addrs - bind a SCTP socket to all our addresses
943 */
944static 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
6ed7257b
PC
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 */
ee44b4bc 976static void sctp_connect_to_sock(struct connection *con)
6ed7257b 977{
ee44b4bc
MRL
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 }
6ed7257b 988
5d689871 989 mutex_lock(&con->sock_mutex);
6ed7257b 990
ee44b4bc
MRL
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) {
6ed7257b 1003 log_print("no address for nodeid %d", con->nodeid);
ee44b4bc 1004 goto out;
6ed7257b 1005 }
6ed7257b 1006
ee44b4bc
MRL
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;
6ed7257b 1012
ee44b4bc
MRL
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);
6ed7257b 1017
ee44b4bc
MRL
1018 /* Bind to all addresses. */
1019 if (sctp_bind_addrs(con, 0))
1020 goto bind_err;
6ed7257b 1021
ee44b4bc 1022 make_sockaddr(&daddr, dlm_config.ci_tcp_port, &addr_len);
6ed7257b 1023
ee44b4bc 1024 log_print("connecting to %d", con->nodeid);
6ed7257b 1025
ee44b4bc
MRL
1026 /* Turn off Nagle's algorithm */
1027 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1028 sizeof(one));
6ed7257b 1029
ee44b4bc
MRL
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;
98e1b60e 1036
6ed7257b 1037
ee44b4bc
MRL
1038bind_err:
1039 con->sock = NULL;
1040 sock_release(sock);
6ed7257b 1041
ee44b4bc
MRL
1042socket_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);
6ed7257b 1056 clear_bit(CF_CONNECT_PENDING, &con->flags);
ee44b4bc
MRL
1057 lowcomms_connect_sock(con);
1058 return;
6ed7257b 1059 }
5d689871 1060
ee44b4bc 1061out:
5d689871 1062 mutex_unlock(&con->sock_mutex);
00dcffae 1063 set_bit(CF_WRITE_PENDING, &con->flags);
6ed7257b
PC
1064}
1065
fdda387f 1066/* Connect a new socket to its peer */
6ed7257b 1067static void tcp_connect_to_sock(struct connection *con)
fdda387f 1068{
6bd8feda 1069 struct sockaddr_storage saddr, src_addr;
fdda387f 1070 int addr_len;
a89d63a1 1071 struct socket *sock = NULL;
cb2d45da 1072 int one = 1;
36b71a8b 1073 int result;
fdda387f
PC
1074
1075 if (con->nodeid == 0) {
1076 log_print("attempt to connect sock 0 foiled");
ac33d071 1077 return;
fdda387f
PC
1078 }
1079
f1f1c1cc 1080 mutex_lock(&con->sock_mutex);
fdda387f
PC
1081 if (con->retries++ > MAX_CONNECT_RETRIES)
1082 goto out;
1083
1084 /* Some odd races can cause double-connects, ignore them */
36b71a8b 1085 if (con->sock)
fdda387f 1086 goto out;
fdda387f
PC
1087
1088 /* Create a socket to communicate with */
eeb1bd5c
EB
1089 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1090 SOCK_STREAM, IPPROTO_TCP, &sock);
fdda387f
PC
1091 if (result < 0)
1092 goto out_err;
1093
1094 memset(&saddr, 0, sizeof(saddr));
98e1b60e 1095 result = nodeid_to_addr(con->nodeid, &saddr, NULL, false);
36b71a8b
DT
1096 if (result < 0) {
1097 log_print("no address for nodeid %d", con->nodeid);
ac33d071 1098 goto out_err;
36b71a8b 1099 }
fdda387f
PC
1100
1101 sock->sk->sk_user_data = con;
1102 con->rx_action = receive_from_sock;
6ed7257b
PC
1103 con->connect_action = tcp_connect_to_sock;
1104 add_sock(sock, con);
fdda387f 1105
6bd8feda
LH
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
68c817a1 1117 make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
fdda387f 1118
fdda387f 1119 log_print("connecting to %d", con->nodeid);
cb2d45da
DT
1120
1121 /* Turn off Nagle's algorithm */
1122 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1123 sizeof(one));
1124
36b71a8b 1125 result = sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
ac33d071 1126 O_NONBLOCK);
fdda387f
PC
1127 if (result == -EINPROGRESS)
1128 result = 0;
ac33d071
PC
1129 if (result == 0)
1130 goto out;
fdda387f 1131
ac33d071 1132out_err:
fdda387f
PC
1133 if (con->sock) {
1134 sock_release(con->sock);
1135 con->sock = NULL;
a89d63a1
CD
1136 } else if (sock) {
1137 sock_release(sock);
fdda387f
PC
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 */
36b71a8b
DT
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);
356344c4 1152 clear_bit(CF_CONNECT_PENDING, &con->flags);
fdda387f 1153 lowcomms_connect_sock(con);
36b71a8b 1154 return;
fdda387f 1155 }
ac33d071 1156out:
f1f1c1cc 1157 mutex_unlock(&con->sock_mutex);
00dcffae 1158 set_bit(CF_WRITE_PENDING, &con->flags);
ac33d071 1159 return;
fdda387f
PC
1160}
1161
6ed7257b
PC
1162static struct socket *tcp_create_listen_sock(struct connection *con,
1163 struct sockaddr_storage *saddr)
fdda387f 1164{
ac33d071 1165 struct socket *sock = NULL;
fdda387f
PC
1166 int result = 0;
1167 int one = 1;
1168 int addr_len;
1169
6ed7257b 1170 if (dlm_local_addr[0]->ss_family == AF_INET)
fdda387f
PC
1171 addr_len = sizeof(struct sockaddr_in);
1172 else
1173 addr_len = sizeof(struct sockaddr_in6);
1174
1175 /* Create a socket to communicate with */
eeb1bd5c
EB
1176 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1177 SOCK_STREAM, IPPROTO_TCP, &sock);
fdda387f 1178 if (result < 0) {
617e82e1 1179 log_print("Can't create listening comms socket");
fdda387f
PC
1180 goto create_out;
1181 }
1182
cb2d45da
DT
1183 /* Turn off Nagle's algorithm */
1184 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1185 sizeof(one));
1186
6ed7257b
PC
1187 result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
1188 (char *)&one, sizeof(one));
1189
fdda387f 1190 if (result < 0) {
617e82e1 1191 log_print("Failed to set SO_REUSEADDR on socket: %d", result);
fdda387f 1192 }
6ed7257b
PC
1193 con->rx_action = tcp_accept_from_sock;
1194 con->connect_action = tcp_connect_to_sock;
fdda387f
PC
1195
1196 /* Bind to our port */
68c817a1 1197 make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
fdda387f
PC
1198 result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
1199 if (result < 0) {
617e82e1 1200 log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
fdda387f
PC
1201 sock_release(sock);
1202 sock = NULL;
1203 con->sock = NULL;
1204 goto create_out;
1205 }
6ed7257b 1206 result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
ac33d071 1207 (char *)&one, sizeof(one));
fdda387f 1208 if (result < 0) {
617e82e1 1209 log_print("Set keepalive failed: %d", result);
fdda387f
PC
1210 }
1211
1212 result = sock->ops->listen(sock, 5);
1213 if (result < 0) {
617e82e1 1214 log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
fdda387f
PC
1215 sock_release(sock);
1216 sock = NULL;
1217 goto create_out;
1218 }
1219
ac33d071 1220create_out:
fdda387f
PC
1221 return sock;
1222}
1223
6ed7257b
PC
1224/* Get local addresses */
1225static void init_local(void)
1226{
1227 struct sockaddr_storage sas, *addr;
1228 int i;
1229
30d3a237 1230 dlm_local_count = 0;
1b189b88 1231 for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
6ed7257b
PC
1232 if (dlm_our_addr(&sas, i))
1233 break;
1234
573c24c4 1235 addr = kmalloc(sizeof(*addr), GFP_NOFS);
6ed7257b
PC
1236 if (!addr)
1237 break;
1238 memcpy(addr, &sas, sizeof(*addr));
1239 dlm_local_addr[dlm_local_count++] = addr;
1240 }
1241}
1242
6ed7257b
PC
1243/* Initialise SCTP socket and bind to all interfaces */
1244static int sctp_listen_for_all(void)
1245{
1246 struct socket *sock = NULL;
ee44b4bc 1247 int result = -EINVAL;
573c24c4 1248 struct connection *con = nodeid2con(0, GFP_NOFS);
6ed7257b 1249 int bufsize = NEEDED_RMEM;
86e92ad2 1250 int one = 1;
6ed7257b
PC
1251
1252 if (!con)
1253 return -ENOMEM;
1254
1255 log_print("Using SCTP for communications");
1256
eeb1bd5c 1257 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
ee44b4bc 1258 SOCK_STREAM, IPPROTO_SCTP, &sock);
6ed7257b
PC
1259 if (result < 0) {
1260 log_print("Can't create comms socket, check SCTP is loaded");
1261 goto out;
1262 }
1263
df61c952 1264 result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
6ed7257b
PC
1265 (char *)&bufsize, sizeof(bufsize));
1266 if (result)
617e82e1 1267 log_print("Error increasing buffer space on socket %d", result);
6ed7257b 1268
86e92ad2
MC
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
6ed7257b
PC
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;
ee44b4bc
MRL
1278 con->rx_action = sctp_accept_from_sock;
1279 con->connect_action = sctp_connect_to_sock;
6ed7257b 1280
ee44b4bc
MRL
1281 /* Bind to all addresses. */
1282 if (sctp_bind_addrs(con, dlm_config.ci_tcp_port))
1283 goto create_delsock;
6ed7257b
PC
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
1293create_delsock:
1294 sock_release(sock);
1295 con->sock = NULL;
1296out:
1297 return result;
1298}
1299
1300static int tcp_listen_for_all(void)
fdda387f
PC
1301{
1302 struct socket *sock = NULL;
573c24c4 1303 struct connection *con = nodeid2con(0, GFP_NOFS);
fdda387f
PC
1304 int result = -EINVAL;
1305
6ed7257b
PC
1306 if (!con)
1307 return -ENOMEM;
1308
fdda387f 1309 /* We don't support multi-homed hosts */
6ed7257b 1310 if (dlm_local_addr[1] != NULL) {
617e82e1
DT
1311 log_print("TCP protocol can't handle multi-homed hosts, "
1312 "try SCTP");
6ed7257b
PC
1313 return -EINVAL;
1314 }
1315
1316 log_print("Using TCP for communications");
1317
6ed7257b 1318 sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
fdda387f
PC
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
1332static 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
617e82e1 1356void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
fdda387f
PC
1357{
1358 struct connection *con;
1359 struct writequeue_entry *e;
1360 int offset = 0;
fdda387f 1361
fdda387f
PC
1362 con = nodeid2con(nodeid, allocation);
1363 if (!con)
1364 return NULL;
1365
4edde74e 1366 spin_lock(&con->writequeue_lock);
fdda387f 1367 e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
ac33d071 1368 if ((&e->list == &con->writequeue) ||
fdda387f
PC
1369 (PAGE_CACHE_SIZE - e->end < len)) {
1370 e = NULL;
1371 } else {
1372 offset = e->end;
1373 e->end += len;
eeee2b5f 1374 e->users++;
fdda387f
PC
1375 }
1376 spin_unlock(&con->writequeue_lock);
1377
1378 if (e) {
ac33d071 1379 got_one:
fdda387f
PC
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;
eeee2b5f 1389 e->users++;
fdda387f
PC
1390 list_add_tail(&e->list, &con->writequeue);
1391 spin_unlock(&con->writequeue_lock);
1392 goto got_one;
1393 }
1394 return NULL;
1395}
1396
1397void 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
4edde74e 1403 spin_lock(&con->writequeue_lock);
fdda387f
PC
1404 users = --e->users;
1405 if (users)
1406 goto out;
1407 e->len = e->end - e->offset;
fdda387f
PC
1408 spin_unlock(&con->writequeue_lock);
1409
1d6e8131
PC
1410 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {
1411 queue_work(send_workqueue, &con->swork);
fdda387f
PC
1412 }
1413 return;
1414
ac33d071 1415out:
fdda387f
PC
1416 spin_unlock(&con->writequeue_lock);
1417 return;
1418}
1419
fdda387f 1420/* Send a message */
ac33d071 1421static void send_to_sock(struct connection *con)
fdda387f
PC
1422{
1423 int ret = 0;
fdda387f
PC
1424 const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
1425 struct writequeue_entry *e;
1426 int len, offset;
f92c8dd7 1427 int count = 0;
fdda387f 1428
f1f1c1cc 1429 mutex_lock(&con->sock_mutex);
fdda387f
PC
1430 if (con->sock == NULL)
1431 goto out_connect;
1432
fdda387f
PC
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) {
1329e3f2
PB
1447 ret = kernel_sendpage(con->sock, e->page, offset, len,
1448 msg_flags);
d66f8277 1449 if (ret == -EAGAIN || ret == 0) {
b36930dd
DM
1450 if (ret == -EAGAIN &&
1451 test_bit(SOCK_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 }
d66f8277 1459 cond_resched();
fdda387f 1460 goto out;
9c5bef58 1461 } else if (ret < 0)
fdda387f 1462 goto send_error;
d66f8277 1463 }
f92c8dd7
BP
1464
1465 /* Don't starve people filling buffers */
1466 if (++count >= MAX_SEND_MSG_COUNT) {
ac33d071 1467 cond_resched();
f92c8dd7
BP
1468 count = 0;
1469 }
fdda387f
PC
1470
1471 spin_lock(&con->writequeue_lock);
5d689871 1472 writequeue_entry_complete(e, ret);
fdda387f
PC
1473 }
1474 spin_unlock(&con->writequeue_lock);
ac33d071 1475out:
f1f1c1cc 1476 mutex_unlock(&con->sock_mutex);
ac33d071 1477 return;
fdda387f 1478
ac33d071 1479send_error:
f1f1c1cc 1480 mutex_unlock(&con->sock_mutex);
0d737a8c 1481 close_connection(con, false, false, true);
fdda387f 1482 lowcomms_connect_sock(con);
ac33d071 1483 return;
fdda387f 1484
ac33d071 1485out_connect:
f1f1c1cc 1486 mutex_unlock(&con->sock_mutex);
ee44b4bc 1487 lowcomms_connect_sock(con);
fdda387f
PC
1488}
1489
1490static void clean_one_writequeue(struct connection *con)
1491{
5e9ccc37 1492 struct writequeue_entry *e, *safe;
fdda387f
PC
1493
1494 spin_lock(&con->writequeue_lock);
5e9ccc37 1495 list_for_each_entry_safe(e, safe, &con->writequeue, list) {
fdda387f
PC
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 */
1504int dlm_lowcomms_close(int nodeid)
1505{
1506 struct connection *con;
36b71a8b 1507 struct dlm_node_addr *na;
fdda387f 1508
fdda387f
PC
1509 log_print("closing connection to node %d", nodeid);
1510 con = nodeid2con(nodeid, 0);
1511 if (con) {
063c4c99 1512 set_bit(CF_CLOSE, &con->flags);
0d737a8c 1513 close_connection(con, true, true, true);
fdda387f 1514 clean_one_writequeue(con);
fdda387f 1515 }
36b71a8b
DT
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
fdda387f 1527 return 0;
fdda387f
PC
1528}
1529
6ed7257b 1530/* Receive workqueue function */
1d6e8131 1531static void process_recv_sockets(struct work_struct *work)
fdda387f 1532{
1d6e8131
PC
1533 struct connection *con = container_of(work, struct connection, rwork);
1534 int err;
fdda387f 1535
1d6e8131
PC
1536 clear_bit(CF_READ_PENDING, &con->flags);
1537 do {
1538 err = con->rx_action(con);
1539 } while (!err);
fdda387f
PC
1540}
1541
6ed7257b 1542/* Send workqueue function */
1d6e8131 1543static void process_send_sockets(struct work_struct *work)
fdda387f 1544{
1d6e8131 1545 struct connection *con = container_of(work, struct connection, swork);
fdda387f 1546
00dcffae 1547 if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags))
6ed7257b 1548 con->connect_action(con);
063c4c99
LMB
1549 if (test_and_clear_bit(CF_WRITE_PENDING, &con->flags))
1550 send_to_sock(con);
fdda387f
PC
1551}
1552
1553
1554/* Discard all entries on the write queues */
1555static void clean_writequeues(void)
1556{
5e9ccc37 1557 foreach_conn(clean_one_writequeue);
fdda387f
PC
1558}
1559
1d6e8131 1560static void work_stop(void)
fdda387f 1561{
1d6e8131
PC
1562 destroy_workqueue(recv_workqueue);
1563 destroy_workqueue(send_workqueue);
fdda387f
PC
1564}
1565
1d6e8131 1566static int work_start(void)
fdda387f 1567{
e43f055a
DT
1568 recv_workqueue = alloc_workqueue("dlm_recv",
1569 WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
b9d41052
NK
1570 if (!recv_workqueue) {
1571 log_print("can't start dlm_recv");
1572 return -ENOMEM;
fdda387f 1573 }
fdda387f 1574
e43f055a
DT
1575 send_workqueue = alloc_workqueue("dlm_send",
1576 WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
b9d41052
NK
1577 if (!send_workqueue) {
1578 log_print("can't start dlm_send");
1d6e8131 1579 destroy_workqueue(recv_workqueue);
b9d41052 1580 return -ENOMEM;
fdda387f 1581 }
fdda387f
PC
1582
1583 return 0;
1584}
1585
5e9ccc37 1586static void stop_conn(struct connection *con)
fdda387f 1587{
5e9ccc37 1588 con->flags |= 0x0F;
391fbdc5 1589 if (con->sock && con->sock->sk)
5e9ccc37
CC
1590 con->sock->sk->sk_user_data = NULL;
1591}
fdda387f 1592
5e9ccc37
CC
1593static void free_conn(struct connection *con)
1594{
0d737a8c 1595 close_connection(con, true, true, true);
5e9ccc37
CC
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
1602void dlm_lowcomms_stop(void)
1603{
ac33d071 1604 /* Set all the flags to prevent any
fdda387f
PC
1605 socket activity.
1606 */
7a936ce7 1607 mutex_lock(&connections_lock);
513ef596 1608 dlm_allow_conn = 0;
5e9ccc37 1609 foreach_conn(stop_conn);
7a936ce7 1610 mutex_unlock(&connections_lock);
ac33d071 1611
1d6e8131 1612 work_stop();
6ed7257b 1613
7a936ce7 1614 mutex_lock(&connections_lock);
fdda387f
PC
1615 clean_writequeues();
1616
5e9ccc37
CC
1617 foreach_conn(free_conn);
1618
7a936ce7 1619 mutex_unlock(&connections_lock);
fdda387f
PC
1620 kmem_cache_destroy(con_cache);
1621}
1622
fdda387f
PC
1623int dlm_lowcomms_start(void)
1624{
6ed7257b
PC
1625 int error = -EINVAL;
1626 struct connection *con;
5e9ccc37
CC
1627 int i;
1628
1629 for (i = 0; i < CONN_HASH_SIZE; i++)
1630 INIT_HLIST_HEAD(&connection_hash[i]);
fdda387f 1631
6ed7257b
PC
1632 init_local();
1633 if (!dlm_local_count) {
617e82e1 1634 error = -ENOTCONN;
fdda387f 1635 log_print("no local IP address has been set");
513ef596 1636 goto fail;
fdda387f
PC
1637 }
1638
6ed7257b 1639 error = -ENOMEM;
fdda387f 1640 con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
ac33d071 1641 __alignof__(struct connection), 0,
20c2df83 1642 NULL);
fdda387f 1643 if (!con_cache)
513ef596
DT
1644 goto fail;
1645
1646 error = work_start();
1647 if (error)
1648 goto fail_destroy;
1649
1650 dlm_allow_conn = 1;
fdda387f 1651
fdda387f 1652 /* Start listening */
6ed7257b
PC
1653 if (dlm_config.ci_protocol == 0)
1654 error = tcp_listen_for_all();
1655 else
1656 error = sctp_listen_for_all();
fdda387f
PC
1657 if (error)
1658 goto fail_unlisten;
1659
fdda387f
PC
1660 return 0;
1661
ac33d071 1662fail_unlisten:
513ef596 1663 dlm_allow_conn = 0;
6ed7257b
PC
1664 con = nodeid2con(0,0);
1665 if (con) {
0d737a8c 1666 close_connection(con, false, true, true);
6ed7257b
PC
1667 kmem_cache_free(con_cache, con);
1668 }
513ef596 1669fail_destroy:
fdda387f 1670 kmem_cache_destroy(con_cache);
513ef596 1671fail:
fdda387f
PC
1672 return error;
1673}
36b71a8b
DT
1674
1675void 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}