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_INIT_PENDING 4
111 #define CF_IS_OTHERCON 5
113 #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 */
128 #define sock2con(x) ((struct connection *)(x)->sk_user_data)
130 /* An entry waiting to be sent */
131 struct writequeue_entry
{
132 struct list_head list
;
138 struct connection
*con
;
141 struct dlm_node_addr
{
142 struct list_head list
;
146 struct sockaddr_storage
*addr
[DLM_MAX_ADDR_COUNT
];
149 static struct listen_sock_callbacks
{
150 void (*sk_error_report
)(struct sock
*);
151 void (*sk_data_ready
)(struct sock
*);
152 void (*sk_state_change
)(struct sock
*);
153 void (*sk_write_space
)(struct sock
*);
156 static LIST_HEAD(dlm_node_addrs
);
157 static DEFINE_SPINLOCK(dlm_node_addrs_spin
);
159 static struct sockaddr_storage
*dlm_local_addr
[DLM_MAX_ADDR_COUNT
];
160 static int dlm_local_count
;
161 static int dlm_allow_conn
;
164 static struct workqueue_struct
*recv_workqueue
;
165 static struct workqueue_struct
*send_workqueue
;
167 static struct hlist_head connection_hash
[CONN_HASH_SIZE
];
168 static DEFINE_MUTEX(connections_lock
);
169 static struct kmem_cache
*con_cache
;
171 static void process_recv_sockets(struct work_struct
*work
);
172 static void process_send_sockets(struct work_struct
*work
);
175 /* This is deliberately very simple because most clusters have simple
176 sequential nodeids, so we should be able to go straight to a connection
177 struct in the array */
178 static inline int nodeid_hash(int nodeid
)
180 return nodeid
& (CONN_HASH_SIZE
-1);
183 static struct connection
*__find_con(int nodeid
)
186 struct connection
*con
;
188 r
= nodeid_hash(nodeid
);
190 hlist_for_each_entry(con
, &connection_hash
[r
], list
) {
191 if (con
->nodeid
== nodeid
)
198 * If 'allocation' is zero then we don't attempt to create a new
199 * connection structure for this node.
201 static struct connection
*__nodeid2con(int nodeid
, gfp_t alloc
)
203 struct connection
*con
= NULL
;
206 con
= __find_con(nodeid
);
210 con
= kmem_cache_zalloc(con_cache
, alloc
);
214 r
= nodeid_hash(nodeid
);
215 hlist_add_head(&con
->list
, &connection_hash
[r
]);
217 con
->nodeid
= nodeid
;
218 mutex_init(&con
->sock_mutex
);
219 INIT_LIST_HEAD(&con
->writequeue
);
220 spin_lock_init(&con
->writequeue_lock
);
221 INIT_WORK(&con
->swork
, process_send_sockets
);
222 INIT_WORK(&con
->rwork
, process_recv_sockets
);
224 /* Setup action pointers for child sockets */
226 struct connection
*zerocon
= __find_con(0);
228 con
->connect_action
= zerocon
->connect_action
;
230 con
->rx_action
= zerocon
->rx_action
;
236 /* Loop round all connections */
237 static void foreach_conn(void (*conn_func
)(struct connection
*c
))
240 struct hlist_node
*n
;
241 struct connection
*con
;
243 for (i
= 0; i
< CONN_HASH_SIZE
; i
++) {
244 hlist_for_each_entry_safe(con
, n
, &connection_hash
[i
], list
)
249 static struct connection
*nodeid2con(int nodeid
, gfp_t allocation
)
251 struct connection
*con
;
253 mutex_lock(&connections_lock
);
254 con
= __nodeid2con(nodeid
, allocation
);
255 mutex_unlock(&connections_lock
);
260 static struct dlm_node_addr
*find_node_addr(int nodeid
)
262 struct dlm_node_addr
*na
;
264 list_for_each_entry(na
, &dlm_node_addrs
, list
) {
265 if (na
->nodeid
== nodeid
)
271 static int addr_compare(struct sockaddr_storage
*x
, struct sockaddr_storage
*y
)
273 switch (x
->ss_family
) {
275 struct sockaddr_in
*sinx
= (struct sockaddr_in
*)x
;
276 struct sockaddr_in
*siny
= (struct sockaddr_in
*)y
;
277 if (sinx
->sin_addr
.s_addr
!= siny
->sin_addr
.s_addr
)
279 if (sinx
->sin_port
!= siny
->sin_port
)
284 struct sockaddr_in6
*sinx
= (struct sockaddr_in6
*)x
;
285 struct sockaddr_in6
*siny
= (struct sockaddr_in6
*)y
;
286 if (!ipv6_addr_equal(&sinx
->sin6_addr
, &siny
->sin6_addr
))
288 if (sinx
->sin6_port
!= siny
->sin6_port
)
298 static int nodeid_to_addr(int nodeid
, struct sockaddr_storage
*sas_out
,
299 struct sockaddr
*sa_out
, bool try_new_addr
)
301 struct sockaddr_storage sas
;
302 struct dlm_node_addr
*na
;
304 if (!dlm_local_count
)
307 spin_lock(&dlm_node_addrs_spin
);
308 na
= find_node_addr(nodeid
);
309 if (na
&& na
->addr_count
) {
310 memcpy(&sas
, na
->addr
[na
->curr_addr_index
],
311 sizeof(struct sockaddr_storage
));
314 na
->curr_addr_index
++;
315 if (na
->curr_addr_index
== na
->addr_count
)
316 na
->curr_addr_index
= 0;
319 spin_unlock(&dlm_node_addrs_spin
);
328 memcpy(sas_out
, &sas
, sizeof(struct sockaddr_storage
));
333 if (dlm_local_addr
[0]->ss_family
== AF_INET
) {
334 struct sockaddr_in
*in4
= (struct sockaddr_in
*) &sas
;
335 struct sockaddr_in
*ret4
= (struct sockaddr_in
*) sa_out
;
336 ret4
->sin_addr
.s_addr
= in4
->sin_addr
.s_addr
;
338 struct sockaddr_in6
*in6
= (struct sockaddr_in6
*) &sas
;
339 struct sockaddr_in6
*ret6
= (struct sockaddr_in6
*) sa_out
;
340 ret6
->sin6_addr
= in6
->sin6_addr
;
346 static int addr_to_nodeid(struct sockaddr_storage
*addr
, int *nodeid
)
348 struct dlm_node_addr
*na
;
352 spin_lock(&dlm_node_addrs_spin
);
353 list_for_each_entry(na
, &dlm_node_addrs
, list
) {
357 for (addr_i
= 0; addr_i
< na
->addr_count
; addr_i
++) {
358 if (addr_compare(na
->addr
[addr_i
], addr
)) {
359 *nodeid
= na
->nodeid
;
366 spin_unlock(&dlm_node_addrs_spin
);
370 int dlm_lowcomms_addr(int nodeid
, struct sockaddr_storage
*addr
, int len
)
372 struct sockaddr_storage
*new_addr
;
373 struct dlm_node_addr
*new_node
, *na
;
375 new_node
= kzalloc(sizeof(struct dlm_node_addr
), GFP_NOFS
);
379 new_addr
= kzalloc(sizeof(struct sockaddr_storage
), GFP_NOFS
);
385 memcpy(new_addr
, addr
, len
);
387 spin_lock(&dlm_node_addrs_spin
);
388 na
= find_node_addr(nodeid
);
390 new_node
->nodeid
= nodeid
;
391 new_node
->addr
[0] = new_addr
;
392 new_node
->addr_count
= 1;
393 list_add(&new_node
->list
, &dlm_node_addrs
);
394 spin_unlock(&dlm_node_addrs_spin
);
398 if (na
->addr_count
>= DLM_MAX_ADDR_COUNT
) {
399 spin_unlock(&dlm_node_addrs_spin
);
405 na
->addr
[na
->addr_count
++] = new_addr
;
406 spin_unlock(&dlm_node_addrs_spin
);
411 /* Data available on socket or listen socket received a connect */
412 static void lowcomms_data_ready(struct sock
*sk
)
414 struct connection
*con
;
416 read_lock_bh(&sk
->sk_callback_lock
);
418 if (con
&& !test_and_set_bit(CF_READ_PENDING
, &con
->flags
))
419 queue_work(recv_workqueue
, &con
->rwork
);
420 read_unlock_bh(&sk
->sk_callback_lock
);
423 static void lowcomms_write_space(struct sock
*sk
)
425 struct connection
*con
;
427 read_lock_bh(&sk
->sk_callback_lock
);
432 clear_bit(SOCK_NOSPACE
, &con
->sock
->flags
);
434 if (test_and_clear_bit(CF_APP_LIMITED
, &con
->flags
)) {
435 con
->sock
->sk
->sk_write_pending
--;
436 clear_bit(SOCKWQ_ASYNC_NOSPACE
, &con
->sock
->flags
);
439 queue_work(send_workqueue
, &con
->swork
);
441 read_unlock_bh(&sk
->sk_callback_lock
);
444 static inline void lowcomms_connect_sock(struct connection
*con
)
446 if (test_bit(CF_CLOSE
, &con
->flags
))
448 queue_work(send_workqueue
, &con
->swork
);
452 static void lowcomms_state_change(struct sock
*sk
)
454 /* SCTP layer is not calling sk_data_ready when the connection
455 * is done, so we catch the signal through here. Also, it
456 * doesn't switch socket state when entering shutdown, so we
457 * skip the write in that case.
459 if (sk
->sk_shutdown
) {
460 if (sk
->sk_shutdown
== RCV_SHUTDOWN
)
461 lowcomms_data_ready(sk
);
462 } else if (sk
->sk_state
== TCP_ESTABLISHED
) {
463 lowcomms_write_space(sk
);
467 int dlm_lowcomms_connect_node(int nodeid
)
469 struct connection
*con
;
471 if (nodeid
== dlm_our_nodeid())
474 con
= nodeid2con(nodeid
, GFP_NOFS
);
477 lowcomms_connect_sock(con
);
481 static void lowcomms_error_report(struct sock
*sk
)
483 struct connection
*con
;
484 struct sockaddr_storage saddr
;
485 void (*orig_report
)(struct sock
*) = NULL
;
487 read_lock_bh(&sk
->sk_callback_lock
);
492 orig_report
= listen_sock
.sk_error_report
;
493 if (con
->sock
== NULL
||
494 kernel_getpeername(con
->sock
, (struct sockaddr
*)&saddr
) < 0) {
495 printk_ratelimited(KERN_ERR
"dlm: node %d: socket error "
496 "sending to node %d, port %d, "
497 "sk_err=%d/%d\n", dlm_our_nodeid(),
498 con
->nodeid
, dlm_config
.ci_tcp_port
,
499 sk
->sk_err
, sk
->sk_err_soft
);
500 } else if (saddr
.ss_family
== AF_INET
) {
501 struct sockaddr_in
*sin4
= (struct sockaddr_in
*)&saddr
;
503 printk_ratelimited(KERN_ERR
"dlm: node %d: socket error "
504 "sending to node %d at %pI4, port %d, "
505 "sk_err=%d/%d\n", dlm_our_nodeid(),
506 con
->nodeid
, &sin4
->sin_addr
.s_addr
,
507 dlm_config
.ci_tcp_port
, sk
->sk_err
,
510 struct sockaddr_in6
*sin6
= (struct sockaddr_in6
*)&saddr
;
512 printk_ratelimited(KERN_ERR
"dlm: node %d: socket error "
513 "sending to node %d at %u.%u.%u.%u, "
514 "port %d, sk_err=%d/%d\n", dlm_our_nodeid(),
515 con
->nodeid
, sin6
->sin6_addr
.s6_addr32
[0],
516 sin6
->sin6_addr
.s6_addr32
[1],
517 sin6
->sin6_addr
.s6_addr32
[2],
518 sin6
->sin6_addr
.s6_addr32
[3],
519 dlm_config
.ci_tcp_port
, sk
->sk_err
,
523 read_unlock_bh(&sk
->sk_callback_lock
);
528 /* Note: sk_callback_lock must be locked before calling this function. */
529 static void save_listen_callbacks(struct socket
*sock
)
531 struct sock
*sk
= sock
->sk
;
533 listen_sock
.sk_data_ready
= sk
->sk_data_ready
;
534 listen_sock
.sk_state_change
= sk
->sk_state_change
;
535 listen_sock
.sk_write_space
= sk
->sk_write_space
;
536 listen_sock
.sk_error_report
= sk
->sk_error_report
;
539 static void restore_callbacks(struct socket
*sock
)
541 struct sock
*sk
= sock
->sk
;
543 write_lock_bh(&sk
->sk_callback_lock
);
544 sk
->sk_user_data
= NULL
;
545 sk
->sk_data_ready
= listen_sock
.sk_data_ready
;
546 sk
->sk_state_change
= listen_sock
.sk_state_change
;
547 sk
->sk_write_space
= listen_sock
.sk_write_space
;
548 sk
->sk_error_report
= listen_sock
.sk_error_report
;
549 write_unlock_bh(&sk
->sk_callback_lock
);
552 /* Make a socket active */
553 static void add_sock(struct socket
*sock
, struct connection
*con
)
555 struct sock
*sk
= sock
->sk
;
557 write_lock_bh(&sk
->sk_callback_lock
);
560 sk
->sk_user_data
= con
;
561 /* Install a data_ready callback */
562 sk
->sk_data_ready
= lowcomms_data_ready
;
563 sk
->sk_write_space
= lowcomms_write_space
;
564 sk
->sk_state_change
= lowcomms_state_change
;
565 sk
->sk_allocation
= GFP_NOFS
;
566 sk
->sk_error_report
= lowcomms_error_report
;
567 write_unlock_bh(&sk
->sk_callback_lock
);
570 /* Add the port number to an IPv6 or 4 sockaddr and return the address
572 static void make_sockaddr(struct sockaddr_storage
*saddr
, uint16_t port
,
575 saddr
->ss_family
= dlm_local_addr
[0]->ss_family
;
576 if (saddr
->ss_family
== AF_INET
) {
577 struct sockaddr_in
*in4_addr
= (struct sockaddr_in
*)saddr
;
578 in4_addr
->sin_port
= cpu_to_be16(port
);
579 *addr_len
= sizeof(struct sockaddr_in
);
580 memset(&in4_addr
->sin_zero
, 0, sizeof(in4_addr
->sin_zero
));
582 struct sockaddr_in6
*in6_addr
= (struct sockaddr_in6
*)saddr
;
583 in6_addr
->sin6_port
= cpu_to_be16(port
);
584 *addr_len
= sizeof(struct sockaddr_in6
);
586 memset((char *)saddr
+ *addr_len
, 0, sizeof(struct sockaddr_storage
) - *addr_len
);
589 /* Close a remote connection and tidy up */
590 static void close_connection(struct connection
*con
, bool and_other
,
593 bool closing
= test_and_set_bit(CF_CLOSING
, &con
->flags
);
595 if (tx
&& !closing
&& cancel_work_sync(&con
->swork
)) {
596 log_print("canceled swork for node %d", con
->nodeid
);
597 clear_bit(CF_WRITE_PENDING
, &con
->flags
);
599 if (rx
&& !closing
&& cancel_work_sync(&con
->rwork
)) {
600 log_print("canceled rwork for node %d", con
->nodeid
);
601 clear_bit(CF_READ_PENDING
, &con
->flags
);
604 mutex_lock(&con
->sock_mutex
);
606 restore_callbacks(con
->sock
);
607 sock_release(con
->sock
);
610 if (con
->othercon
&& and_other
) {
611 /* Will only re-enter once. */
612 close_connection(con
->othercon
, false, true, true);
615 __free_page(con
->rx_page
);
620 mutex_unlock(&con
->sock_mutex
);
621 clear_bit(CF_CLOSING
, &con
->flags
);
624 /* Data received from remote end */
625 static int receive_from_sock(struct connection
*con
)
628 struct msghdr msg
= {};
632 int call_again_soon
= 0;
635 mutex_lock(&con
->sock_mutex
);
637 if (con
->sock
== NULL
) {
641 if (con
->nodeid
== 0) {
646 if (con
->rx_page
== NULL
) {
648 * This doesn't need to be atomic, but I think it should
649 * improve performance if it is.
651 con
->rx_page
= alloc_page(GFP_ATOMIC
);
652 if (con
->rx_page
== NULL
)
654 cbuf_init(&con
->cb
, PAGE_SIZE
);
658 * iov[0] is the bit of the circular buffer between the current end
659 * point (cb.base + cb.len) and the end of the buffer.
661 iov
[0].iov_len
= con
->cb
.base
- cbuf_data(&con
->cb
);
662 iov
[0].iov_base
= page_address(con
->rx_page
) + cbuf_data(&con
->cb
);
667 * iov[1] is the bit of the circular buffer between the start of the
668 * buffer and the start of the currently used section (cb.base)
670 if (cbuf_data(&con
->cb
) >= con
->cb
.base
) {
671 iov
[0].iov_len
= PAGE_SIZE
- cbuf_data(&con
->cb
);
672 iov
[1].iov_len
= con
->cb
.base
;
673 iov
[1].iov_base
= page_address(con
->rx_page
);
676 len
= iov
[0].iov_len
+ iov
[1].iov_len
;
677 iov_iter_kvec(&msg
.msg_iter
, READ
, iov
, nvec
, len
);
679 r
= ret
= sock_recvmsg(con
->sock
, &msg
, MSG_DONTWAIT
| MSG_NOSIGNAL
);
685 cbuf_add(&con
->cb
, ret
);
686 ret
= dlm_process_incoming_buffer(con
->nodeid
,
687 page_address(con
->rx_page
),
688 con
->cb
.base
, con
->cb
.len
,
690 if (ret
== -EBADMSG
) {
691 log_print("lowcomms: addr=%p, base=%u, len=%u, read=%d",
692 page_address(con
->rx_page
), con
->cb
.base
,
697 cbuf_eat(&con
->cb
, ret
);
699 if (cbuf_empty(&con
->cb
) && !call_again_soon
) {
700 __free_page(con
->rx_page
);
706 mutex_unlock(&con
->sock_mutex
);
710 if (!test_and_set_bit(CF_READ_PENDING
, &con
->flags
))
711 queue_work(recv_workqueue
, &con
->rwork
);
712 mutex_unlock(&con
->sock_mutex
);
716 mutex_unlock(&con
->sock_mutex
);
717 if (ret
!= -EAGAIN
) {
718 close_connection(con
, true, true, false);
719 /* Reconnect when there is something to send */
721 /* Don't return success if we really got EOF */
728 /* Listening socket is busy, accept a connection */
729 static int tcp_accept_from_sock(struct connection
*con
)
732 struct sockaddr_storage peeraddr
;
733 struct socket
*newsock
;
736 struct connection
*newcon
;
737 struct connection
*addcon
;
739 mutex_lock(&connections_lock
);
740 if (!dlm_allow_conn
) {
741 mutex_unlock(&connections_lock
);
744 mutex_unlock(&connections_lock
);
746 mutex_lock_nested(&con
->sock_mutex
, 0);
749 mutex_unlock(&con
->sock_mutex
);
753 result
= kernel_accept(con
->sock
, &newsock
, O_NONBLOCK
);
757 /* Get the connected socket's peer */
758 memset(&peeraddr
, 0, sizeof(peeraddr
));
759 len
= newsock
->ops
->getname(newsock
, (struct sockaddr
*)&peeraddr
, 2);
761 result
= -ECONNABORTED
;
765 /* Get the new node's NODEID */
766 make_sockaddr(&peeraddr
, 0, &len
);
767 if (addr_to_nodeid(&peeraddr
, &nodeid
)) {
768 unsigned char *b
=(unsigned char *)&peeraddr
;
769 log_print("connect from non cluster node");
770 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE
,
771 b
, sizeof(struct sockaddr_storage
));
772 sock_release(newsock
);
773 mutex_unlock(&con
->sock_mutex
);
777 log_print("got connection from %d", nodeid
);
779 /* Check to see if we already have a connection to this node. This
780 * could happen if the two nodes initiate a connection at roughly
781 * the same time and the connections cross on the wire.
782 * In this case we store the incoming one in "othercon"
784 newcon
= nodeid2con(nodeid
, GFP_NOFS
);
789 mutex_lock_nested(&newcon
->sock_mutex
, 1);
791 struct connection
*othercon
= newcon
->othercon
;
794 othercon
= kmem_cache_zalloc(con_cache
, GFP_NOFS
);
796 log_print("failed to allocate incoming socket");
797 mutex_unlock(&newcon
->sock_mutex
);
801 othercon
->nodeid
= nodeid
;
802 othercon
->rx_action
= receive_from_sock
;
803 mutex_init(&othercon
->sock_mutex
);
804 INIT_LIST_HEAD(&othercon
->writequeue
);
805 spin_lock_init(&othercon
->writequeue_lock
);
806 INIT_WORK(&othercon
->swork
, process_send_sockets
);
807 INIT_WORK(&othercon
->rwork
, process_recv_sockets
);
808 set_bit(CF_IS_OTHERCON
, &othercon
->flags
);
810 mutex_lock_nested(&othercon
->sock_mutex
, 2);
811 if (!othercon
->sock
) {
812 newcon
->othercon
= othercon
;
813 add_sock(newsock
, othercon
);
815 mutex_unlock(&othercon
->sock_mutex
);
818 printk("Extra connection from node %d attempted\n", nodeid
);
820 mutex_unlock(&othercon
->sock_mutex
);
821 mutex_unlock(&newcon
->sock_mutex
);
826 newcon
->rx_action
= receive_from_sock
;
827 /* accept copies the sk after we've saved the callbacks, so we
828 don't want to save them a second time or comm errors will
829 result in calling sk_error_report recursively. */
830 add_sock(newsock
, newcon
);
834 mutex_unlock(&newcon
->sock_mutex
);
837 * Add it to the active queue in case we got data
838 * between processing the accept adding the socket
839 * to the read_sockets list
841 if (!test_and_set_bit(CF_READ_PENDING
, &addcon
->flags
))
842 queue_work(recv_workqueue
, &addcon
->rwork
);
843 mutex_unlock(&con
->sock_mutex
);
848 mutex_unlock(&con
->sock_mutex
);
850 sock_release(newsock
);
852 if (result
!= -EAGAIN
)
853 log_print("error accepting connection from node: %d", result
);
857 static int sctp_accept_from_sock(struct connection
*con
)
859 /* Check that the new node is in the lockspace */
860 struct sctp_prim prim
;
864 struct connection
*newcon
;
865 struct connection
*addcon
;
866 struct socket
*newsock
;
868 mutex_lock(&connections_lock
);
869 if (!dlm_allow_conn
) {
870 mutex_unlock(&connections_lock
);
873 mutex_unlock(&connections_lock
);
875 mutex_lock_nested(&con
->sock_mutex
, 0);
877 ret
= kernel_accept(con
->sock
, &newsock
, O_NONBLOCK
);
881 memset(&prim
, 0, sizeof(struct sctp_prim
));
882 prim_len
= sizeof(struct sctp_prim
);
884 ret
= kernel_getsockopt(newsock
, IPPROTO_SCTP
, SCTP_PRIMARY_ADDR
,
885 (char *)&prim
, &prim_len
);
887 log_print("getsockopt/sctp_primary_addr failed: %d", ret
);
891 make_sockaddr(&prim
.ssp_addr
, 0, &addr_len
);
892 ret
= addr_to_nodeid(&prim
.ssp_addr
, &nodeid
);
894 unsigned char *b
= (unsigned char *)&prim
.ssp_addr
;
896 log_print("reject connect from unknown addr");
897 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE
,
898 b
, sizeof(struct sockaddr_storage
));
902 newcon
= nodeid2con(nodeid
, GFP_NOFS
);
908 mutex_lock_nested(&newcon
->sock_mutex
, 1);
911 struct connection
*othercon
= newcon
->othercon
;
914 othercon
= kmem_cache_zalloc(con_cache
, GFP_NOFS
);
916 log_print("failed to allocate incoming socket");
917 mutex_unlock(&newcon
->sock_mutex
);
921 othercon
->nodeid
= nodeid
;
922 othercon
->rx_action
= receive_from_sock
;
923 mutex_init(&othercon
->sock_mutex
);
924 INIT_LIST_HEAD(&othercon
->writequeue
);
925 spin_lock_init(&othercon
->writequeue_lock
);
926 INIT_WORK(&othercon
->swork
, process_send_sockets
);
927 INIT_WORK(&othercon
->rwork
, process_recv_sockets
);
928 set_bit(CF_IS_OTHERCON
, &othercon
->flags
);
930 mutex_lock_nested(&othercon
->sock_mutex
, 2);
931 if (!othercon
->sock
) {
932 newcon
->othercon
= othercon
;
933 add_sock(newsock
, othercon
);
935 mutex_unlock(&othercon
->sock_mutex
);
937 printk("Extra connection from node %d attempted\n", nodeid
);
939 mutex_unlock(&othercon
->sock_mutex
);
940 mutex_unlock(&newcon
->sock_mutex
);
944 newcon
->rx_action
= receive_from_sock
;
945 add_sock(newsock
, newcon
);
949 log_print("connected to %d", nodeid
);
951 mutex_unlock(&newcon
->sock_mutex
);
954 * Add it to the active queue in case we got data
955 * between processing the accept adding the socket
956 * to the read_sockets list
958 if (!test_and_set_bit(CF_READ_PENDING
, &addcon
->flags
))
959 queue_work(recv_workqueue
, &addcon
->rwork
);
960 mutex_unlock(&con
->sock_mutex
);
965 mutex_unlock(&con
->sock_mutex
);
967 sock_release(newsock
);
969 log_print("error accepting connection from node: %d", ret
);
974 static void free_entry(struct writequeue_entry
*e
)
976 __free_page(e
->page
);
981 * writequeue_entry_complete - try to delete and free write queue entry
982 * @e: write queue entry to try to delete
983 * @completed: bytes completed
985 * writequeue_lock must be held.
987 static void writequeue_entry_complete(struct writequeue_entry
*e
, int completed
)
989 e
->offset
+= completed
;
992 if (e
->len
== 0 && e
->users
== 0) {
999 * sctp_bind_addrs - bind a SCTP socket to all our addresses
1001 static int sctp_bind_addrs(struct connection
*con
, uint16_t port
)
1003 struct sockaddr_storage localaddr
;
1004 int i
, addr_len
, result
= 0;
1006 for (i
= 0; i
< dlm_local_count
; i
++) {
1007 memcpy(&localaddr
, dlm_local_addr
[i
], sizeof(localaddr
));
1008 make_sockaddr(&localaddr
, port
, &addr_len
);
1011 result
= kernel_bind(con
->sock
,
1012 (struct sockaddr
*)&localaddr
,
1015 result
= kernel_setsockopt(con
->sock
, SOL_SCTP
,
1016 SCTP_SOCKOPT_BINDX_ADD
,
1017 (char *)&localaddr
, addr_len
);
1020 log_print("Can't bind to %d addr number %d, %d.\n",
1021 port
, i
+ 1, result
);
1028 /* Initiate an SCTP association.
1029 This is a special case of send_to_sock() in that we don't yet have a
1030 peeled-off socket for this association, so we use the listening socket
1031 and add the primary IP address of the remote node.
1033 static void sctp_connect_to_sock(struct connection
*con
)
1035 struct sockaddr_storage daddr
;
1039 struct socket
*sock
;
1040 struct timeval tv
= { .tv_sec
= 5, .tv_usec
= 0 };
1042 if (con
->nodeid
== 0) {
1043 log_print("attempt to connect sock 0 foiled");
1047 mutex_lock(&con
->sock_mutex
);
1049 /* Some odd races can cause double-connects, ignore them */
1050 if (con
->retries
++ > MAX_CONNECT_RETRIES
)
1054 log_print("node %d already connected.", con
->nodeid
);
1058 memset(&daddr
, 0, sizeof(daddr
));
1059 result
= nodeid_to_addr(con
->nodeid
, &daddr
, NULL
, true);
1061 log_print("no address for nodeid %d", con
->nodeid
);
1065 /* Create a socket to communicate with */
1066 result
= sock_create_kern(&init_net
, dlm_local_addr
[0]->ss_family
,
1067 SOCK_STREAM
, IPPROTO_SCTP
, &sock
);
1071 con
->rx_action
= receive_from_sock
;
1072 con
->connect_action
= sctp_connect_to_sock
;
1073 add_sock(sock
, con
);
1075 /* Bind to all addresses. */
1076 if (sctp_bind_addrs(con
, 0))
1079 make_sockaddr(&daddr
, dlm_config
.ci_tcp_port
, &addr_len
);
1081 log_print("connecting to %d", con
->nodeid
);
1083 /* Turn off Nagle's algorithm */
1084 kernel_setsockopt(sock
, SOL_SCTP
, SCTP_NODELAY
, (char *)&one
,
1088 * Make sock->ops->connect() function return in specified time,
1089 * since O_NONBLOCK argument in connect() function does not work here,
1090 * then, we should restore the default value of this attribute.
1092 kernel_setsockopt(sock
, SOL_SOCKET
, SO_SNDTIMEO_OLD
, (char *)&tv
,
1094 result
= sock
->ops
->connect(sock
, (struct sockaddr
*)&daddr
, addr_len
,
1096 memset(&tv
, 0, sizeof(tv
));
1097 kernel_setsockopt(sock
, SOL_SOCKET
, SO_SNDTIMEO_OLD
, (char *)&tv
,
1100 if (result
== -EINPROGRESS
)
1111 * Some errors are fatal and this list might need adjusting. For other
1112 * errors we try again until the max number of retries is reached.
1114 if (result
!= -EHOSTUNREACH
&&
1115 result
!= -ENETUNREACH
&&
1116 result
!= -ENETDOWN
&&
1117 result
!= -EINVAL
&&
1118 result
!= -EPROTONOSUPPORT
) {
1119 log_print("connect %d try %d error %d", con
->nodeid
,
1120 con
->retries
, result
);
1121 mutex_unlock(&con
->sock_mutex
);
1123 lowcomms_connect_sock(con
);
1128 mutex_unlock(&con
->sock_mutex
);
1131 /* Connect a new socket to its peer */
1132 static void tcp_connect_to_sock(struct connection
*con
)
1134 struct sockaddr_storage saddr
, src_addr
;
1136 struct socket
*sock
= NULL
;
1140 if (con
->nodeid
== 0) {
1141 log_print("attempt to connect sock 0 foiled");
1145 mutex_lock(&con
->sock_mutex
);
1146 if (con
->retries
++ > MAX_CONNECT_RETRIES
)
1149 /* Some odd races can cause double-connects, ignore them */
1153 /* Create a socket to communicate with */
1154 result
= sock_create_kern(&init_net
, dlm_local_addr
[0]->ss_family
,
1155 SOCK_STREAM
, IPPROTO_TCP
, &sock
);
1159 memset(&saddr
, 0, sizeof(saddr
));
1160 result
= nodeid_to_addr(con
->nodeid
, &saddr
, NULL
, false);
1162 log_print("no address for nodeid %d", con
->nodeid
);
1166 con
->rx_action
= receive_from_sock
;
1167 con
->connect_action
= tcp_connect_to_sock
;
1168 add_sock(sock
, con
);
1170 /* Bind to our cluster-known address connecting to avoid
1172 memcpy(&src_addr
, dlm_local_addr
[0], sizeof(src_addr
));
1173 make_sockaddr(&src_addr
, 0, &addr_len
);
1174 result
= sock
->ops
->bind(sock
, (struct sockaddr
*) &src_addr
,
1177 log_print("could not bind for connect: %d", result
);
1178 /* This *may* not indicate a critical error */
1181 make_sockaddr(&saddr
, dlm_config
.ci_tcp_port
, &addr_len
);
1183 log_print("connecting to %d", con
->nodeid
);
1185 /* Turn off Nagle's algorithm */
1186 kernel_setsockopt(sock
, SOL_TCP
, TCP_NODELAY
, (char *)&one
,
1189 result
= sock
->ops
->connect(sock
, (struct sockaddr
*)&saddr
, addr_len
,
1191 if (result
== -EINPROGRESS
)
1198 sock_release(con
->sock
);
1204 * Some errors are fatal and this list might need adjusting. For other
1205 * errors we try again until the max number of retries is reached.
1207 if (result
!= -EHOSTUNREACH
&&
1208 result
!= -ENETUNREACH
&&
1209 result
!= -ENETDOWN
&&
1210 result
!= -EINVAL
&&
1211 result
!= -EPROTONOSUPPORT
) {
1212 log_print("connect %d try %d error %d", con
->nodeid
,
1213 con
->retries
, result
);
1214 mutex_unlock(&con
->sock_mutex
);
1216 lowcomms_connect_sock(con
);
1220 mutex_unlock(&con
->sock_mutex
);
1224 static struct socket
*tcp_create_listen_sock(struct connection
*con
,
1225 struct sockaddr_storage
*saddr
)
1227 struct socket
*sock
= NULL
;
1232 if (dlm_local_addr
[0]->ss_family
== AF_INET
)
1233 addr_len
= sizeof(struct sockaddr_in
);
1235 addr_len
= sizeof(struct sockaddr_in6
);
1237 /* Create a socket to communicate with */
1238 result
= sock_create_kern(&init_net
, dlm_local_addr
[0]->ss_family
,
1239 SOCK_STREAM
, IPPROTO_TCP
, &sock
);
1241 log_print("Can't create listening comms socket");
1245 /* Turn off Nagle's algorithm */
1246 kernel_setsockopt(sock
, SOL_TCP
, TCP_NODELAY
, (char *)&one
,
1249 result
= kernel_setsockopt(sock
, SOL_SOCKET
, SO_REUSEADDR
,
1250 (char *)&one
, sizeof(one
));
1253 log_print("Failed to set SO_REUSEADDR on socket: %d", result
);
1255 write_lock_bh(&sock
->sk
->sk_callback_lock
);
1256 sock
->sk
->sk_user_data
= con
;
1257 save_listen_callbacks(sock
);
1258 con
->rx_action
= tcp_accept_from_sock
;
1259 con
->connect_action
= tcp_connect_to_sock
;
1260 write_unlock_bh(&sock
->sk
->sk_callback_lock
);
1262 /* Bind to our port */
1263 make_sockaddr(saddr
, dlm_config
.ci_tcp_port
, &addr_len
);
1264 result
= sock
->ops
->bind(sock
, (struct sockaddr
*) saddr
, addr_len
);
1266 log_print("Can't bind to port %d", dlm_config
.ci_tcp_port
);
1272 result
= kernel_setsockopt(sock
, SOL_SOCKET
, SO_KEEPALIVE
,
1273 (char *)&one
, sizeof(one
));
1275 log_print("Set keepalive failed: %d", result
);
1278 result
= sock
->ops
->listen(sock
, 5);
1280 log_print("Can't listen on port %d", dlm_config
.ci_tcp_port
);
1290 /* Get local addresses */
1291 static void init_local(void)
1293 struct sockaddr_storage sas
, *addr
;
1296 dlm_local_count
= 0;
1297 for (i
= 0; i
< DLM_MAX_ADDR_COUNT
; i
++) {
1298 if (dlm_our_addr(&sas
, i
))
1301 addr
= kmemdup(&sas
, sizeof(*addr
), GFP_NOFS
);
1304 dlm_local_addr
[dlm_local_count
++] = addr
;
1308 /* Initialise SCTP socket and bind to all interfaces */
1309 static int sctp_listen_for_all(void)
1311 struct socket
*sock
= NULL
;
1312 int result
= -EINVAL
;
1313 struct connection
*con
= nodeid2con(0, GFP_NOFS
);
1314 int bufsize
= NEEDED_RMEM
;
1320 log_print("Using SCTP for communications");
1322 result
= sock_create_kern(&init_net
, dlm_local_addr
[0]->ss_family
,
1323 SOCK_STREAM
, IPPROTO_SCTP
, &sock
);
1325 log_print("Can't create comms socket, check SCTP is loaded");
1329 result
= kernel_setsockopt(sock
, SOL_SOCKET
, SO_RCVBUFFORCE
,
1330 (char *)&bufsize
, sizeof(bufsize
));
1332 log_print("Error increasing buffer space on socket %d", result
);
1334 result
= kernel_setsockopt(sock
, SOL_SCTP
, SCTP_NODELAY
, (char *)&one
,
1337 log_print("Could not set SCTP NODELAY error %d\n", result
);
1339 write_lock_bh(&sock
->sk
->sk_callback_lock
);
1340 /* Init con struct */
1341 sock
->sk
->sk_user_data
= con
;
1342 save_listen_callbacks(sock
);
1344 con
->sock
->sk
->sk_data_ready
= lowcomms_data_ready
;
1345 con
->rx_action
= sctp_accept_from_sock
;
1346 con
->connect_action
= sctp_connect_to_sock
;
1348 write_unlock_bh(&sock
->sk
->sk_callback_lock
);
1350 /* Bind to all addresses. */
1351 if (sctp_bind_addrs(con
, dlm_config
.ci_tcp_port
))
1352 goto create_delsock
;
1354 result
= sock
->ops
->listen(sock
, 5);
1356 log_print("Can't set socket listening");
1357 goto create_delsock
;
1369 static int tcp_listen_for_all(void)
1371 struct socket
*sock
= NULL
;
1372 struct connection
*con
= nodeid2con(0, GFP_NOFS
);
1373 int result
= -EINVAL
;
1378 /* We don't support multi-homed hosts */
1379 if (dlm_local_addr
[1] != NULL
) {
1380 log_print("TCP protocol can't handle multi-homed hosts, "
1385 log_print("Using TCP for communications");
1387 sock
= tcp_create_listen_sock(con
, dlm_local_addr
[0]);
1389 add_sock(sock
, con
);
1393 result
= -EADDRINUSE
;
1401 static struct writequeue_entry
*new_writequeue_entry(struct connection
*con
,
1404 struct writequeue_entry
*entry
;
1406 entry
= kmalloc(sizeof(struct writequeue_entry
), allocation
);
1410 entry
->page
= alloc_page(allocation
);
1425 void *dlm_lowcomms_get_buffer(int nodeid
, int len
, gfp_t allocation
, char **ppc
)
1427 struct connection
*con
;
1428 struct writequeue_entry
*e
;
1431 con
= nodeid2con(nodeid
, allocation
);
1435 spin_lock(&con
->writequeue_lock
);
1436 e
= list_entry(con
->writequeue
.prev
, struct writequeue_entry
, list
);
1437 if ((&e
->list
== &con
->writequeue
) ||
1438 (PAGE_SIZE
- e
->end
< len
)) {
1445 spin_unlock(&con
->writequeue_lock
);
1449 *ppc
= page_address(e
->page
) + offset
;
1453 e
= new_writequeue_entry(con
, allocation
);
1455 spin_lock(&con
->writequeue_lock
);
1459 list_add_tail(&e
->list
, &con
->writequeue
);
1460 spin_unlock(&con
->writequeue_lock
);
1466 void dlm_lowcomms_commit_buffer(void *mh
)
1468 struct writequeue_entry
*e
= (struct writequeue_entry
*)mh
;
1469 struct connection
*con
= e
->con
;
1472 spin_lock(&con
->writequeue_lock
);
1476 e
->len
= e
->end
- e
->offset
;
1477 spin_unlock(&con
->writequeue_lock
);
1479 queue_work(send_workqueue
, &con
->swork
);
1483 spin_unlock(&con
->writequeue_lock
);
1487 /* Send a message */
1488 static void send_to_sock(struct connection
*con
)
1491 const int msg_flags
= MSG_DONTWAIT
| MSG_NOSIGNAL
;
1492 struct writequeue_entry
*e
;
1496 mutex_lock(&con
->sock_mutex
);
1497 if (con
->sock
== NULL
)
1500 spin_lock(&con
->writequeue_lock
);
1502 e
= list_entry(con
->writequeue
.next
, struct writequeue_entry
,
1504 if ((struct list_head
*) e
== &con
->writequeue
)
1509 BUG_ON(len
== 0 && e
->users
== 0);
1510 spin_unlock(&con
->writequeue_lock
);
1514 ret
= kernel_sendpage(con
->sock
, e
->page
, offset
, len
,
1516 if (ret
== -EAGAIN
|| ret
== 0) {
1517 if (ret
== -EAGAIN
&&
1518 test_bit(SOCKWQ_ASYNC_NOSPACE
, &con
->sock
->flags
) &&
1519 !test_and_set_bit(CF_APP_LIMITED
, &con
->flags
)) {
1520 /* Notify TCP that we're limited by the
1521 * application window size.
1523 set_bit(SOCK_NOSPACE
, &con
->sock
->flags
);
1524 con
->sock
->sk
->sk_write_pending
++;
1532 /* Don't starve people filling buffers */
1533 if (++count
>= MAX_SEND_MSG_COUNT
) {
1538 spin_lock(&con
->writequeue_lock
);
1539 writequeue_entry_complete(e
, ret
);
1541 spin_unlock(&con
->writequeue_lock
);
1543 mutex_unlock(&con
->sock_mutex
);
1547 mutex_unlock(&con
->sock_mutex
);
1548 close_connection(con
, true, false, true);
1549 /* Requeue the send work. When the work daemon runs again, it will try
1550 a new connection, then call this function again. */
1551 queue_work(send_workqueue
, &con
->swork
);
1555 mutex_unlock(&con
->sock_mutex
);
1556 queue_work(send_workqueue
, &con
->swork
);
1560 static void clean_one_writequeue(struct connection
*con
)
1562 struct writequeue_entry
*e
, *safe
;
1564 spin_lock(&con
->writequeue_lock
);
1565 list_for_each_entry_safe(e
, safe
, &con
->writequeue
, list
) {
1569 spin_unlock(&con
->writequeue_lock
);
1572 /* Called from recovery when it knows that a node has
1574 int dlm_lowcomms_close(int nodeid
)
1576 struct connection
*con
;
1577 struct dlm_node_addr
*na
;
1579 log_print("closing connection to node %d", nodeid
);
1580 con
= nodeid2con(nodeid
, 0);
1582 set_bit(CF_CLOSE
, &con
->flags
);
1583 close_connection(con
, true, true, true);
1584 clean_one_writequeue(con
);
1587 spin_lock(&dlm_node_addrs_spin
);
1588 na
= find_node_addr(nodeid
);
1590 list_del(&na
->list
);
1591 while (na
->addr_count
--)
1592 kfree(na
->addr
[na
->addr_count
]);
1595 spin_unlock(&dlm_node_addrs_spin
);
1600 /* Receive workqueue function */
1601 static void process_recv_sockets(struct work_struct
*work
)
1603 struct connection
*con
= container_of(work
, struct connection
, rwork
);
1606 clear_bit(CF_READ_PENDING
, &con
->flags
);
1608 err
= con
->rx_action(con
);
1612 /* Send workqueue function */
1613 static void process_send_sockets(struct work_struct
*work
)
1615 struct connection
*con
= container_of(work
, struct connection
, swork
);
1617 clear_bit(CF_WRITE_PENDING
, &con
->flags
);
1618 if (con
->sock
== NULL
) /* not mutex protected so check it inside too */
1619 con
->connect_action(con
);
1620 if (!list_empty(&con
->writequeue
))
1625 /* Discard all entries on the write queues */
1626 static void clean_writequeues(void)
1628 foreach_conn(clean_one_writequeue
);
1631 static void work_stop(void)
1633 destroy_workqueue(recv_workqueue
);
1634 destroy_workqueue(send_workqueue
);
1637 static int work_start(void)
1639 recv_workqueue
= alloc_workqueue("dlm_recv",
1640 WQ_UNBOUND
| WQ_MEM_RECLAIM
, 1);
1641 if (!recv_workqueue
) {
1642 log_print("can't start dlm_recv");
1646 send_workqueue
= alloc_workqueue("dlm_send",
1647 WQ_UNBOUND
| WQ_MEM_RECLAIM
, 1);
1648 if (!send_workqueue
) {
1649 log_print("can't start dlm_send");
1650 destroy_workqueue(recv_workqueue
);
1657 static void _stop_conn(struct connection
*con
, bool and_other
)
1659 mutex_lock(&con
->sock_mutex
);
1660 set_bit(CF_CLOSE
, &con
->flags
);
1661 set_bit(CF_READ_PENDING
, &con
->flags
);
1662 set_bit(CF_WRITE_PENDING
, &con
->flags
);
1663 if (con
->sock
&& con
->sock
->sk
) {
1664 write_lock_bh(&con
->sock
->sk
->sk_callback_lock
);
1665 con
->sock
->sk
->sk_user_data
= NULL
;
1666 write_unlock_bh(&con
->sock
->sk
->sk_callback_lock
);
1668 if (con
->othercon
&& and_other
)
1669 _stop_conn(con
->othercon
, false);
1670 mutex_unlock(&con
->sock_mutex
);
1673 static void stop_conn(struct connection
*con
)
1675 _stop_conn(con
, true);
1678 static void free_conn(struct connection
*con
)
1680 close_connection(con
, true, true, true);
1682 kmem_cache_free(con_cache
, con
->othercon
);
1683 hlist_del(&con
->list
);
1684 kmem_cache_free(con_cache
, con
);
1687 static void work_flush(void)
1691 struct hlist_node
*n
;
1692 struct connection
*con
;
1694 flush_workqueue(recv_workqueue
);
1695 flush_workqueue(send_workqueue
);
1698 foreach_conn(stop_conn
);
1699 flush_workqueue(recv_workqueue
);
1700 flush_workqueue(send_workqueue
);
1701 for (i
= 0; i
< CONN_HASH_SIZE
&& ok
; i
++) {
1702 hlist_for_each_entry_safe(con
, n
,
1703 &connection_hash
[i
], list
) {
1704 ok
&= test_bit(CF_READ_PENDING
, &con
->flags
);
1705 ok
&= test_bit(CF_WRITE_PENDING
, &con
->flags
);
1706 if (con
->othercon
) {
1707 ok
&= test_bit(CF_READ_PENDING
,
1708 &con
->othercon
->flags
);
1709 ok
&= test_bit(CF_WRITE_PENDING
,
1710 &con
->othercon
->flags
);
1717 void dlm_lowcomms_stop(void)
1719 /* Set all the flags to prevent any
1722 mutex_lock(&connections_lock
);
1724 mutex_unlock(&connections_lock
);
1726 clean_writequeues();
1727 foreach_conn(free_conn
);
1730 kmem_cache_destroy(con_cache
);
1733 int dlm_lowcomms_start(void)
1735 int error
= -EINVAL
;
1736 struct connection
*con
;
1739 for (i
= 0; i
< CONN_HASH_SIZE
; i
++)
1740 INIT_HLIST_HEAD(&connection_hash
[i
]);
1743 if (!dlm_local_count
) {
1745 log_print("no local IP address has been set");
1750 con_cache
= kmem_cache_create("dlm_conn", sizeof(struct connection
),
1751 __alignof__(struct connection
), 0,
1756 error
= work_start();
1762 /* Start listening */
1763 if (dlm_config
.ci_protocol
== 0)
1764 error
= tcp_listen_for_all();
1766 error
= sctp_listen_for_all();
1774 con
= nodeid2con(0,0);
1776 close_connection(con
, false, true, true);
1777 kmem_cache_free(con_cache
, con
);
1780 kmem_cache_destroy(con_cache
);
1785 void dlm_lowcomms_exit(void)
1787 struct dlm_node_addr
*na
, *safe
;
1789 spin_lock(&dlm_node_addrs_spin
);
1790 list_for_each_entry_safe(na
, safe
, &dlm_node_addrs
, list
) {
1791 list_del(&na
->list
);
1792 while (na
->addr_count
--)
1793 kfree(na
->addr
[na
->addr_count
]);
1796 spin_unlock(&dlm_node_addrs_spin
);