1 /******************************************************************************
2 *******************************************************************************
4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5 ** Copyright (C) 2004-2009 Red Hat, Inc. All rights reserved.
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.
11 *******************************************************************************
12 ******************************************************************************/
17 * This is the "low-level" comms layer.
19 * It is responsible for sending/receiving messages
20 * from other nodes in the cluster.
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.
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.
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.
47 #include <asm/ioctls.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>
58 #include "dlm_internal.h"
63 #define NEEDED_RMEM (4*1024*1024)
64 #define CONN_HASH_SIZE 32
66 /* Number of messages to send before rescheduling */
67 #define MAX_SEND_MSG_COUNT 25
75 static void cbuf_add(struct cbuf
*cb
, int n
)
80 static int cbuf_data(struct cbuf
*cb
)
82 return ((cb
->base
+ cb
->len
) & cb
->mask
);
85 static void cbuf_init(struct cbuf
*cb
, int size
)
87 cb
->base
= cb
->len
= 0;
91 static void cbuf_eat(struct cbuf
*cb
, int n
)
98 static bool cbuf_empty(struct cbuf
*cb
)
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
;
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
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
;
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
);
129 #define sock2con(x) ((struct connection *)(x)->sk_user_data)
131 /* An entry waiting to be sent */
132 struct writequeue_entry
{
133 struct list_head list
;
139 struct connection
*con
;
142 struct dlm_node_addr
{
143 struct list_head list
;
147 struct sockaddr_storage
*addr
[DLM_MAX_ADDR_COUNT
];
150 static LIST_HEAD(dlm_node_addrs
);
151 static DEFINE_SPINLOCK(dlm_node_addrs_spin
);
153 static struct sockaddr_storage
*dlm_local_addr
[DLM_MAX_ADDR_COUNT
];
154 static int dlm_local_count
;
155 static int dlm_allow_conn
;
158 static struct workqueue_struct
*recv_workqueue
;
159 static struct workqueue_struct
*send_workqueue
;
161 static struct hlist_head connection_hash
[CONN_HASH_SIZE
];
162 static DEFINE_MUTEX(connections_lock
);
163 static struct kmem_cache
*con_cache
;
165 static void process_recv_sockets(struct work_struct
*work
);
166 static void process_send_sockets(struct work_struct
*work
);
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
)
174 return nodeid
& (CONN_HASH_SIZE
-1);
177 static struct connection
*__find_con(int nodeid
)
180 struct connection
*con
;
182 r
= nodeid_hash(nodeid
);
184 hlist_for_each_entry(con
, &connection_hash
[r
], list
) {
185 if (con
->nodeid
== nodeid
)
192 * If 'allocation' is zero then we don't attempt to create a new
193 * connection structure for this node.
195 static struct connection
*__nodeid2con(int nodeid
, gfp_t alloc
)
197 struct connection
*con
= NULL
;
200 con
= __find_con(nodeid
);
204 con
= kmem_cache_zalloc(con_cache
, alloc
);
208 r
= nodeid_hash(nodeid
);
209 hlist_add_head(&con
->list
, &connection_hash
[r
]);
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
);
218 /* Setup action pointers for child sockets */
220 struct connection
*zerocon
= __find_con(0);
222 con
->connect_action
= zerocon
->connect_action
;
224 con
->rx_action
= zerocon
->rx_action
;
230 /* Loop round all connections */
231 static void foreach_conn(void (*conn_func
)(struct connection
*c
))
234 struct hlist_node
*n
;
235 struct connection
*con
;
237 for (i
= 0; i
< CONN_HASH_SIZE
; i
++) {
238 hlist_for_each_entry_safe(con
, n
, &connection_hash
[i
], list
)
243 static struct connection
*nodeid2con(int nodeid
, gfp_t allocation
)
245 struct connection
*con
;
247 mutex_lock(&connections_lock
);
248 con
= __nodeid2con(nodeid
, allocation
);
249 mutex_unlock(&connections_lock
);
254 static struct dlm_node_addr
*find_node_addr(int nodeid
)
256 struct dlm_node_addr
*na
;
258 list_for_each_entry(na
, &dlm_node_addrs
, list
) {
259 if (na
->nodeid
== nodeid
)
265 static int addr_compare(struct sockaddr_storage
*x
, struct sockaddr_storage
*y
)
267 switch (x
->ss_family
) {
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
)
273 if (sinx
->sin_port
!= siny
->sin_port
)
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
))
282 if (sinx
->sin6_port
!= siny
->sin6_port
)
292 static int nodeid_to_addr(int nodeid
, struct sockaddr_storage
*sas_out
,
293 struct sockaddr
*sa_out
, bool try_new_addr
)
295 struct sockaddr_storage sas
;
296 struct dlm_node_addr
*na
;
298 if (!dlm_local_count
)
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
));
308 na
->curr_addr_index
++;
309 if (na
->curr_addr_index
== na
->addr_count
)
310 na
->curr_addr_index
= 0;
313 spin_unlock(&dlm_node_addrs_spin
);
322 memcpy(sas_out
, &sas
, sizeof(struct sockaddr_storage
));
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
;
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
;
340 static int addr_to_nodeid(struct sockaddr_storage
*addr
, int *nodeid
)
342 struct dlm_node_addr
*na
;
346 spin_lock(&dlm_node_addrs_spin
);
347 list_for_each_entry(na
, &dlm_node_addrs
, list
) {
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
;
360 spin_unlock(&dlm_node_addrs_spin
);
364 int dlm_lowcomms_addr(int nodeid
, struct sockaddr_storage
*addr
, int len
)
366 struct sockaddr_storage
*new_addr
;
367 struct dlm_node_addr
*new_node
, *na
;
369 new_node
= kzalloc(sizeof(struct dlm_node_addr
), GFP_NOFS
);
373 new_addr
= kzalloc(sizeof(struct sockaddr_storage
), GFP_NOFS
);
379 memcpy(new_addr
, addr
, len
);
381 spin_lock(&dlm_node_addrs_spin
);
382 na
= find_node_addr(nodeid
);
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
);
392 if (na
->addr_count
>= DLM_MAX_ADDR_COUNT
) {
393 spin_unlock(&dlm_node_addrs_spin
);
399 na
->addr
[na
->addr_count
++] = new_addr
;
400 spin_unlock(&dlm_node_addrs_spin
);
405 /* Data available on socket or listen socket received a connect */
406 static void lowcomms_data_ready(struct sock
*sk
)
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
);
413 static void lowcomms_write_space(struct sock
*sk
)
415 struct connection
*con
= sock2con(sk
);
420 clear_bit(SOCK_NOSPACE
, &con
->sock
->flags
);
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
);
427 if (!test_and_set_bit(CF_WRITE_PENDING
, &con
->flags
))
428 queue_work(send_workqueue
, &con
->swork
);
431 static inline void lowcomms_connect_sock(struct connection
*con
)
433 if (test_bit(CF_CLOSE
, &con
->flags
))
435 if (!test_and_set_bit(CF_CONNECT_PENDING
, &con
->flags
))
436 queue_work(send_workqueue
, &con
->swork
);
439 static void lowcomms_state_change(struct sock
*sk
)
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.
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
);
454 int dlm_lowcomms_connect_node(int nodeid
)
456 struct connection
*con
;
458 if (nodeid
== dlm_our_nodeid())
461 con
= nodeid2con(nodeid
, GFP_NOFS
);
464 lowcomms_connect_sock(con
);
468 static void lowcomms_error_report(struct sock
*sk
)
470 struct connection
*con
= sock2con(sk
);
471 struct sockaddr_storage saddr
;
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
);
480 } else if (saddr
.ss_family
== AF_INET
) {
481 struct sockaddr_in
*sin4
= (struct sockaddr_in
*)&saddr
;
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
,
490 struct sockaddr_in6
*sin6
= (struct sockaddr_in6
*)&saddr
;
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
,
502 con
->orig_error_report(sk
);
505 /* Make a socket active */
506 static void add_sock(struct socket
*sock
, struct connection
*con
)
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
;
520 /* Add the port number to an IPv6 or 4 sockaddr and return the address
522 static void make_sockaddr(struct sockaddr_storage
*saddr
, uint16_t port
,
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
));
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
);
536 memset((char *)saddr
+ *addr_len
, 0, sizeof(struct sockaddr_storage
) - *addr_len
);
539 /* Close a remote connection and tidy up */
540 static void close_connection(struct connection
*con
, bool and_other
,
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
);
550 mutex_lock(&con
->sock_mutex
);
552 sock_release(con
->sock
);
555 if (con
->othercon
&& and_other
) {
556 /* Will only re-enter once. */
557 close_connection(con
->othercon
, false, true, true);
560 __free_page(con
->rx_page
);
565 mutex_unlock(&con
->sock_mutex
);
568 /* Data received from remote end */
569 static int receive_from_sock(struct connection
*con
)
572 struct msghdr msg
= {};
576 int call_again_soon
= 0;
579 mutex_lock(&con
->sock_mutex
);
581 if (con
->sock
== NULL
) {
585 if (con
->nodeid
== 0) {
590 if (con
->rx_page
== NULL
) {
592 * This doesn't need to be atomic, but I think it should
593 * improve performance if it is.
595 con
->rx_page
= alloc_page(GFP_ATOMIC
);
596 if (con
->rx_page
== NULL
)
598 cbuf_init(&con
->cb
, PAGE_CACHE_SIZE
);
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.
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
);
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)
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
);
620 len
= iov
[0].iov_len
+ iov
[1].iov_len
;
622 r
= ret
= kernel_recvmsg(con
->sock
, &msg
, iov
, nvec
, len
,
623 MSG_DONTWAIT
| MSG_NOSIGNAL
);
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
,
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
,
641 cbuf_eat(&con
->cb
, ret
);
643 if (cbuf_empty(&con
->cb
) && !call_again_soon
) {
644 __free_page(con
->rx_page
);
650 mutex_unlock(&con
->sock_mutex
);
654 if (!test_and_set_bit(CF_READ_PENDING
, &con
->flags
))
655 queue_work(recv_workqueue
, &con
->rwork
);
656 mutex_unlock(&con
->sock_mutex
);
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 */
665 /* Don't return success if we really got EOF */
672 /* Listening socket is busy, accept a connection */
673 static int tcp_accept_from_sock(struct connection
*con
)
676 struct sockaddr_storage peeraddr
;
677 struct socket
*newsock
;
680 struct connection
*newcon
;
681 struct connection
*addcon
;
683 mutex_lock(&connections_lock
);
684 if (!dlm_allow_conn
) {
685 mutex_unlock(&connections_lock
);
688 mutex_unlock(&connections_lock
);
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
);
696 mutex_lock_nested(&con
->sock_mutex
, 0);
699 if (con
->sock
== NULL
)
702 newsock
->type
= con
->sock
->type
;
703 newsock
->ops
= con
->sock
->ops
;
705 result
= con
->sock
->ops
->accept(con
->sock
, newsock
, O_NONBLOCK
);
709 /* Get the connected socket's peer */
710 memset(&peeraddr
, 0, sizeof(peeraddr
));
711 if (newsock
->ops
->getname(newsock
, (struct sockaddr
*)&peeraddr
,
713 result
= -ECONNABORTED
;
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
);
729 log_print("got connection from %d", nodeid
);
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"
736 newcon
= nodeid2con(nodeid
, GFP_NOFS
);
741 mutex_lock_nested(&newcon
->sock_mutex
, 1);
743 struct connection
*othercon
= newcon
->othercon
;
746 othercon
= kmem_cache_zalloc(con_cache
, GFP_NOFS
);
748 log_print("failed to allocate incoming socket");
749 mutex_unlock(&newcon
->sock_mutex
);
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
);
760 if (!othercon
->sock
) {
761 newcon
->othercon
= othercon
;
762 othercon
->sock
= newsock
;
763 newsock
->sk
->sk_user_data
= othercon
;
764 add_sock(newsock
, othercon
);
768 printk("Extra connection from node %d attempted\n", nodeid
);
770 mutex_unlock(&newcon
->sock_mutex
);
775 newsock
->sk
->sk_user_data
= newcon
;
776 newcon
->rx_action
= receive_from_sock
;
777 add_sock(newsock
, newcon
);
781 mutex_unlock(&newcon
->sock_mutex
);
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
788 if (!test_and_set_bit(CF_READ_PENDING
, &addcon
->flags
))
789 queue_work(recv_workqueue
, &addcon
->rwork
);
790 mutex_unlock(&con
->sock_mutex
);
795 mutex_unlock(&con
->sock_mutex
);
796 sock_release(newsock
);
798 if (result
!= -EAGAIN
)
799 log_print("error accepting connection from node: %d", result
);
803 static int sctp_accept_from_sock(struct connection
*con
)
805 /* Check that the new node is in the lockspace */
806 struct sctp_prim prim
;
810 struct connection
*newcon
;
811 struct connection
*addcon
;
812 struct socket
*newsock
;
814 mutex_lock(&connections_lock
);
815 if (!dlm_allow_conn
) {
816 mutex_unlock(&connections_lock
);
819 mutex_unlock(&connections_lock
);
821 mutex_lock_nested(&con
->sock_mutex
, 0);
823 ret
= kernel_accept(con
->sock
, &newsock
, O_NONBLOCK
);
827 memset(&prim
, 0, sizeof(struct sctp_prim
));
828 prim_len
= sizeof(struct sctp_prim
);
830 ret
= kernel_getsockopt(newsock
, IPPROTO_SCTP
, SCTP_PRIMARY_ADDR
,
831 (char *)&prim
, &prim_len
);
833 log_print("getsockopt/sctp_primary_addr failed: %d", ret
);
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
;
841 log_print("reject connect from unknown addr");
842 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE
,
843 b
, sizeof(struct sockaddr_storage
));
847 newcon
= nodeid2con(nodeid
, GFP_NOFS
);
853 mutex_lock_nested(&newcon
->sock_mutex
, 1);
856 struct connection
*othercon
= newcon
->othercon
;
859 othercon
= kmem_cache_zalloc(con_cache
, GFP_NOFS
);
861 log_print("failed to allocate incoming socket");
862 mutex_unlock(&newcon
->sock_mutex
);
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
);
873 if (!othercon
->sock
) {
874 newcon
->othercon
= othercon
;
875 othercon
->sock
= newsock
;
876 newsock
->sk
->sk_user_data
= othercon
;
877 add_sock(newsock
, othercon
);
880 printk("Extra connection from node %d attempted\n", nodeid
);
882 mutex_unlock(&newcon
->sock_mutex
);
886 newsock
->sk
->sk_user_data
= newcon
;
887 newcon
->rx_action
= receive_from_sock
;
888 add_sock(newsock
, newcon
);
892 log_print("connected to %d", nodeid
);
894 mutex_unlock(&newcon
->sock_mutex
);
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
901 if (!test_and_set_bit(CF_READ_PENDING
, &addcon
->flags
))
902 queue_work(recv_workqueue
, &addcon
->rwork
);
903 mutex_unlock(&con
->sock_mutex
);
908 mutex_unlock(&con
->sock_mutex
);
910 sock_release(newsock
);
912 log_print("error accepting connection from node: %d", ret
);
917 static void free_entry(struct writequeue_entry
*e
)
919 __free_page(e
->page
);
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
928 * writequeue_lock must be held.
930 static void writequeue_entry_complete(struct writequeue_entry
*e
, int completed
)
932 e
->offset
+= completed
;
935 if (e
->len
== 0 && e
->users
== 0) {
942 * sctp_bind_addrs - bind a SCTP socket to all our addresses
944 static int sctp_bind_addrs(struct connection
*con
, uint16_t port
)
946 struct sockaddr_storage localaddr
;
947 int i
, addr_len
, result
= 0;
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
);
954 result
= kernel_bind(con
->sock
,
955 (struct sockaddr
*)&localaddr
,
958 result
= kernel_setsockopt(con
->sock
, SOL_SCTP
,
959 SCTP_SOCKOPT_BINDX_ADD
,
960 (char *)&localaddr
, addr_len
);
963 log_print("Can't bind to %d addr number %d, %d.\n",
964 port
, i
+ 1, result
);
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.
976 static void sctp_connect_to_sock(struct connection
*con
)
978 struct sockaddr_storage daddr
;
984 if (con
->nodeid
== 0) {
985 log_print("attempt to connect sock 0 foiled");
989 mutex_lock(&con
->sock_mutex
);
991 /* Some odd races can cause double-connects, ignore them */
992 if (con
->retries
++ > MAX_CONNECT_RETRIES
)
996 log_print("node %d already connected.", con
->nodeid
);
1000 memset(&daddr
, 0, sizeof(daddr
));
1001 result
= nodeid_to_addr(con
->nodeid
, &daddr
, NULL
, true);
1003 log_print("no address for nodeid %d", con
->nodeid
);
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
);
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
);
1018 /* Bind to all addresses. */
1019 if (sctp_bind_addrs(con
, 0))
1022 make_sockaddr(&daddr
, dlm_config
.ci_tcp_port
, &addr_len
);
1024 log_print("connecting to %d", con
->nodeid
);
1026 /* Turn off Nagle's algorithm */
1027 kernel_setsockopt(sock
, SOL_TCP
, TCP_NODELAY
, (char *)&one
,
1030 result
= sock
->ops
->connect(sock
, (struct sockaddr
*)&daddr
, addr_len
,
1032 if (result
== -EINPROGRESS
)
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.
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
);
1056 clear_bit(CF_CONNECT_PENDING
, &con
->flags
);
1057 lowcomms_connect_sock(con
);
1062 mutex_unlock(&con
->sock_mutex
);
1063 set_bit(CF_WRITE_PENDING
, &con
->flags
);
1066 /* Connect a new socket to its peer */
1067 static void tcp_connect_to_sock(struct connection
*con
)
1069 struct sockaddr_storage saddr
, src_addr
;
1071 struct socket
*sock
= NULL
;
1075 if (con
->nodeid
== 0) {
1076 log_print("attempt to connect sock 0 foiled");
1080 mutex_lock(&con
->sock_mutex
);
1081 if (con
->retries
++ > MAX_CONNECT_RETRIES
)
1084 /* Some odd races can cause double-connects, ignore them */
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
);
1094 memset(&saddr
, 0, sizeof(saddr
));
1095 result
= nodeid_to_addr(con
->nodeid
, &saddr
, NULL
, false);
1097 log_print("no address for nodeid %d", con
->nodeid
);
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
);
1106 /* Bind to our cluster-known address connecting to avoid
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
,
1113 log_print("could not bind for connect: %d", result
);
1114 /* This *may* not indicate a critical error */
1117 make_sockaddr(&saddr
, dlm_config
.ci_tcp_port
, &addr_len
);
1119 log_print("connecting to %d", con
->nodeid
);
1121 /* Turn off Nagle's algorithm */
1122 kernel_setsockopt(sock
, SOL_TCP
, TCP_NODELAY
, (char *)&one
,
1125 result
= sock
->ops
->connect(sock
, (struct sockaddr
*)&saddr
, addr_len
,
1127 if (result
== -EINPROGRESS
)
1134 sock_release(con
->sock
);
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.
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
);
1152 clear_bit(CF_CONNECT_PENDING
, &con
->flags
);
1153 lowcomms_connect_sock(con
);
1157 mutex_unlock(&con
->sock_mutex
);
1158 set_bit(CF_WRITE_PENDING
, &con
->flags
);
1162 static struct socket
*tcp_create_listen_sock(struct connection
*con
,
1163 struct sockaddr_storage
*saddr
)
1165 struct socket
*sock
= NULL
;
1170 if (dlm_local_addr
[0]->ss_family
== AF_INET
)
1171 addr_len
= sizeof(struct sockaddr_in
);
1173 addr_len
= sizeof(struct sockaddr_in6
);
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
);
1179 log_print("Can't create listening comms socket");
1183 /* Turn off Nagle's algorithm */
1184 kernel_setsockopt(sock
, SOL_TCP
, TCP_NODELAY
, (char *)&one
,
1187 result
= kernel_setsockopt(sock
, SOL_SOCKET
, SO_REUSEADDR
,
1188 (char *)&one
, sizeof(one
));
1191 log_print("Failed to set SO_REUSEADDR on socket: %d", result
);
1193 con
->rx_action
= tcp_accept_from_sock
;
1194 con
->connect_action
= tcp_connect_to_sock
;
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
);
1200 log_print("Can't bind to port %d", dlm_config
.ci_tcp_port
);
1206 result
= kernel_setsockopt(sock
, SOL_SOCKET
, SO_KEEPALIVE
,
1207 (char *)&one
, sizeof(one
));
1209 log_print("Set keepalive failed: %d", result
);
1212 result
= sock
->ops
->listen(sock
, 5);
1214 log_print("Can't listen on port %d", dlm_config
.ci_tcp_port
);
1224 /* Get local addresses */
1225 static void init_local(void)
1227 struct sockaddr_storage sas
, *addr
;
1230 dlm_local_count
= 0;
1231 for (i
= 0; i
< DLM_MAX_ADDR_COUNT
; i
++) {
1232 if (dlm_our_addr(&sas
, i
))
1235 addr
= kmalloc(sizeof(*addr
), GFP_NOFS
);
1238 memcpy(addr
, &sas
, sizeof(*addr
));
1239 dlm_local_addr
[dlm_local_count
++] = addr
;
1243 /* Initialise SCTP socket and bind to all interfaces */
1244 static int sctp_listen_for_all(void)
1246 struct socket
*sock
= NULL
;
1247 int result
= -EINVAL
;
1248 struct connection
*con
= nodeid2con(0, GFP_NOFS
);
1249 int bufsize
= NEEDED_RMEM
;
1255 log_print("Using SCTP for communications");
1257 result
= sock_create_kern(&init_net
, dlm_local_addr
[0]->ss_family
,
1258 SOCK_STREAM
, IPPROTO_SCTP
, &sock
);
1260 log_print("Can't create comms socket, check SCTP is loaded");
1264 result
= kernel_setsockopt(sock
, SOL_SOCKET
, SO_RCVBUFFORCE
,
1265 (char *)&bufsize
, sizeof(bufsize
));
1267 log_print("Error increasing buffer space on socket %d", result
);
1269 result
= kernel_setsockopt(sock
, SOL_SCTP
, SCTP_NODELAY
, (char *)&one
,
1272 log_print("Could not set SCTP NODELAY error %d\n", result
);
1274 /* Init con struct */
1275 sock
->sk
->sk_user_data
= con
;
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
;
1281 /* Bind to all addresses. */
1282 if (sctp_bind_addrs(con
, dlm_config
.ci_tcp_port
))
1283 goto create_delsock
;
1285 result
= sock
->ops
->listen(sock
, 5);
1287 log_print("Can't set socket listening");
1288 goto create_delsock
;
1300 static int tcp_listen_for_all(void)
1302 struct socket
*sock
= NULL
;
1303 struct connection
*con
= nodeid2con(0, GFP_NOFS
);
1304 int result
= -EINVAL
;
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, "
1316 log_print("Using TCP for communications");
1318 sock
= tcp_create_listen_sock(con
, dlm_local_addr
[0]);
1320 add_sock(sock
, con
);
1324 result
= -EADDRINUSE
;
1332 static struct writequeue_entry
*new_writequeue_entry(struct connection
*con
,
1335 struct writequeue_entry
*entry
;
1337 entry
= kmalloc(sizeof(struct writequeue_entry
), allocation
);
1341 entry
->page
= alloc_page(allocation
);
1356 void *dlm_lowcomms_get_buffer(int nodeid
, int len
, gfp_t allocation
, char **ppc
)
1358 struct connection
*con
;
1359 struct writequeue_entry
*e
;
1362 con
= nodeid2con(nodeid
, allocation
);
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
)) {
1376 spin_unlock(&con
->writequeue_lock
);
1380 *ppc
= page_address(e
->page
) + offset
;
1384 e
= new_writequeue_entry(con
, allocation
);
1386 spin_lock(&con
->writequeue_lock
);
1390 list_add_tail(&e
->list
, &con
->writequeue
);
1391 spin_unlock(&con
->writequeue_lock
);
1397 void dlm_lowcomms_commit_buffer(void *mh
)
1399 struct writequeue_entry
*e
= (struct writequeue_entry
*)mh
;
1400 struct connection
*con
= e
->con
;
1403 spin_lock(&con
->writequeue_lock
);
1407 e
->len
= e
->end
- e
->offset
;
1408 spin_unlock(&con
->writequeue_lock
);
1410 if (!test_and_set_bit(CF_WRITE_PENDING
, &con
->flags
)) {
1411 queue_work(send_workqueue
, &con
->swork
);
1416 spin_unlock(&con
->writequeue_lock
);
1420 /* Send a message */
1421 static void send_to_sock(struct connection
*con
)
1424 const int msg_flags
= MSG_DONTWAIT
| MSG_NOSIGNAL
;
1425 struct writequeue_entry
*e
;
1429 mutex_lock(&con
->sock_mutex
);
1430 if (con
->sock
== NULL
)
1433 spin_lock(&con
->writequeue_lock
);
1435 e
= list_entry(con
->writequeue
.next
, struct writequeue_entry
,
1437 if ((struct list_head
*) e
== &con
->writequeue
)
1442 BUG_ON(len
== 0 && e
->users
== 0);
1443 spin_unlock(&con
->writequeue_lock
);
1447 ret
= kernel_sendpage(con
->sock
, e
->page
, offset
, len
,
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.
1456 set_bit(SOCK_NOSPACE
, &con
->sock
->flags
);
1457 con
->sock
->sk
->sk_write_pending
++;
1465 /* Don't starve people filling buffers */
1466 if (++count
>= MAX_SEND_MSG_COUNT
) {
1471 spin_lock(&con
->writequeue_lock
);
1472 writequeue_entry_complete(e
, ret
);
1474 spin_unlock(&con
->writequeue_lock
);
1476 mutex_unlock(&con
->sock_mutex
);
1480 mutex_unlock(&con
->sock_mutex
);
1481 close_connection(con
, false, false, true);
1482 lowcomms_connect_sock(con
);
1486 mutex_unlock(&con
->sock_mutex
);
1487 lowcomms_connect_sock(con
);
1490 static void clean_one_writequeue(struct connection
*con
)
1492 struct writequeue_entry
*e
, *safe
;
1494 spin_lock(&con
->writequeue_lock
);
1495 list_for_each_entry_safe(e
, safe
, &con
->writequeue
, list
) {
1499 spin_unlock(&con
->writequeue_lock
);
1502 /* Called from recovery when it knows that a node has
1504 int dlm_lowcomms_close(int nodeid
)
1506 struct connection
*con
;
1507 struct dlm_node_addr
*na
;
1509 log_print("closing connection to node %d", nodeid
);
1510 con
= nodeid2con(nodeid
, 0);
1512 set_bit(CF_CLOSE
, &con
->flags
);
1513 close_connection(con
, true, true, true);
1514 clean_one_writequeue(con
);
1517 spin_lock(&dlm_node_addrs_spin
);
1518 na
= find_node_addr(nodeid
);
1520 list_del(&na
->list
);
1521 while (na
->addr_count
--)
1522 kfree(na
->addr
[na
->addr_count
]);
1525 spin_unlock(&dlm_node_addrs_spin
);
1530 /* Receive workqueue function */
1531 static void process_recv_sockets(struct work_struct
*work
)
1533 struct connection
*con
= container_of(work
, struct connection
, rwork
);
1536 clear_bit(CF_READ_PENDING
, &con
->flags
);
1538 err
= con
->rx_action(con
);
1542 /* Send workqueue function */
1543 static void process_send_sockets(struct work_struct
*work
)
1545 struct connection
*con
= container_of(work
, struct connection
, swork
);
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
))
1554 /* Discard all entries on the write queues */
1555 static void clean_writequeues(void)
1557 foreach_conn(clean_one_writequeue
);
1560 static void work_stop(void)
1562 destroy_workqueue(recv_workqueue
);
1563 destroy_workqueue(send_workqueue
);
1566 static int work_start(void)
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");
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
);
1586 static void stop_conn(struct connection
*con
)
1589 if (con
->sock
&& con
->sock
->sk
)
1590 con
->sock
->sk
->sk_user_data
= NULL
;
1593 static void free_conn(struct connection
*con
)
1595 close_connection(con
, true, true, true);
1597 kmem_cache_free(con_cache
, con
->othercon
);
1598 hlist_del(&con
->list
);
1599 kmem_cache_free(con_cache
, con
);
1602 void dlm_lowcomms_stop(void)
1604 /* Set all the flags to prevent any
1607 mutex_lock(&connections_lock
);
1609 foreach_conn(stop_conn
);
1610 clean_writequeues();
1611 foreach_conn(free_conn
);
1612 mutex_unlock(&connections_lock
);
1616 kmem_cache_destroy(con_cache
);
1619 int dlm_lowcomms_start(void)
1621 int error
= -EINVAL
;
1622 struct connection
*con
;
1625 for (i
= 0; i
< CONN_HASH_SIZE
; i
++)
1626 INIT_HLIST_HEAD(&connection_hash
[i
]);
1629 if (!dlm_local_count
) {
1631 log_print("no local IP address has been set");
1636 con_cache
= kmem_cache_create("dlm_conn", sizeof(struct connection
),
1637 __alignof__(struct connection
), 0,
1642 error
= work_start();
1648 /* Start listening */
1649 if (dlm_config
.ci_protocol
== 0)
1650 error
= tcp_listen_for_all();
1652 error
= sctp_listen_for_all();
1660 con
= nodeid2con(0,0);
1662 close_connection(con
, false, true, true);
1663 kmem_cache_free(con_cache
, con
);
1666 kmem_cache_destroy(con_cache
);
1671 void dlm_lowcomms_exit(void)
1673 struct dlm_node_addr
*na
, *safe
;
1675 spin_lock(&dlm_node_addrs_spin
);
1676 list_for_each_entry_safe(na
, safe
, &dlm_node_addrs
, list
) {
1677 list_del(&na
->list
);
1678 while (na
->addr_count
--)
1679 kfree(na
->addr
[na
->addr_count
]);
1682 spin_unlock(&dlm_node_addrs_spin
);