Loading...
1/******************************************************************************
2*******************************************************************************
3**
4** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
5** Copyright (C) 2004-2009 Red Hat, Inc. All rights reserved.
6**
7** This copyrighted material is made available to anyone wishing to use,
8** modify, copy, or redistribute it subject to the terms and conditions
9** of the GNU General Public License v.2.
10**
11*******************************************************************************
12******************************************************************************/
13
14/*
15 * lowcomms.c
16 *
17 * This is the "low-level" comms layer.
18 *
19 * It is responsible for sending/receiving messages
20 * from other nodes in the cluster.
21 *
22 * Cluster nodes are referred to by their nodeids. nodeids are
23 * simply 32 bit numbers to the locking module - if they need to
24 * be expanded for the cluster infrastructure then that is its
25 * responsibility. It is this layer's
26 * responsibility to resolve these into IP address or
27 * whatever it needs for inter-node communication.
28 *
29 * The comms level is two kernel threads that deal mainly with
30 * the receiving of messages from other nodes and passing them
31 * up to the mid-level comms layer (which understands the
32 * message format) for execution by the locking core, and
33 * a send thread which does all the setting up of connections
34 * to remote nodes and the sending of data. Threads are not allowed
35 * to send their own data because it may cause them to wait in times
36 * of high load. Also, this way, the sending thread can collect together
37 * messages bound for one node and send them in one block.
38 *
39 * lowcomms will choose to use either TCP or SCTP as its transport layer
40 * depending on the configuration variable 'protocol'. This should be set
41 * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
42 * cluster-wide mechanism as it must be the same on all nodes of the cluster
43 * for the DLM to function.
44 *
45 */
46
47#include <asm/ioctls.h>
48#include <net/sock.h>
49#include <net/tcp.h>
50#include <linux/pagemap.h>
51#include <linux/file.h>
52#include <linux/mutex.h>
53#include <linux/sctp.h>
54#include <linux/slab.h>
55#include <net/sctp/sctp.h>
56#include <net/sctp/user.h>
57#include <net/ipv6.h>
58
59#include "dlm_internal.h"
60#include "lowcomms.h"
61#include "midcomms.h"
62#include "config.h"
63
64#define NEEDED_RMEM (4*1024*1024)
65#define CONN_HASH_SIZE 32
66
67/* Number of messages to send before rescheduling */
68#define MAX_SEND_MSG_COUNT 25
69
70struct cbuf {
71 unsigned int base;
72 unsigned int len;
73 unsigned int mask;
74};
75
76static void cbuf_add(struct cbuf *cb, int n)
77{
78 cb->len += n;
79}
80
81static int cbuf_data(struct cbuf *cb)
82{
83 return ((cb->base + cb->len) & cb->mask);
84}
85
86static void cbuf_init(struct cbuf *cb, int size)
87{
88 cb->base = cb->len = 0;
89 cb->mask = size-1;
90}
91
92static void cbuf_eat(struct cbuf *cb, int n)
93{
94 cb->len -= n;
95 cb->base += n;
96 cb->base &= cb->mask;
97}
98
99static bool cbuf_empty(struct cbuf *cb)
100{
101 return cb->len == 0;
102}
103
104struct connection {
105 struct socket *sock; /* NULL if not connected */
106 uint32_t nodeid; /* So we know who we are in the list */
107 struct mutex sock_mutex;
108 unsigned long flags;
109#define CF_READ_PENDING 1
110#define CF_WRITE_PENDING 2
111#define CF_CONNECT_PENDING 3
112#define CF_INIT_PENDING 4
113#define CF_IS_OTHERCON 5
114#define CF_CLOSE 6
115#define CF_APP_LIMITED 7
116 struct list_head writequeue; /* List of outgoing writequeue_entries */
117 spinlock_t writequeue_lock;
118 int (*rx_action) (struct connection *); /* What to do when active */
119 void (*connect_action) (struct connection *); /* What to do to connect */
120 struct page *rx_page;
121 struct cbuf cb;
122 int retries;
123#define MAX_CONNECT_RETRIES 3
124 int sctp_assoc;
125 struct hlist_node list;
126 struct connection *othercon;
127 struct work_struct rwork; /* Receive workqueue */
128 struct work_struct swork; /* Send workqueue */
129};
130#define sock2con(x) ((struct connection *)(x)->sk_user_data)
131
132/* An entry waiting to be sent */
133struct writequeue_entry {
134 struct list_head list;
135 struct page *page;
136 int offset;
137 int len;
138 int end;
139 int users;
140 struct connection *con;
141};
142
143static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
144static int dlm_local_count;
145static int dlm_allow_conn;
146
147/* Work queues */
148static struct workqueue_struct *recv_workqueue;
149static struct workqueue_struct *send_workqueue;
150
151static struct hlist_head connection_hash[CONN_HASH_SIZE];
152static DEFINE_MUTEX(connections_lock);
153static struct kmem_cache *con_cache;
154
155static void process_recv_sockets(struct work_struct *work);
156static void process_send_sockets(struct work_struct *work);
157
158
159/* This is deliberately very simple because most clusters have simple
160 sequential nodeids, so we should be able to go straight to a connection
161 struct in the array */
162static inline int nodeid_hash(int nodeid)
163{
164 return nodeid & (CONN_HASH_SIZE-1);
165}
166
167static struct connection *__find_con(int nodeid)
168{
169 int r;
170 struct hlist_node *h;
171 struct connection *con;
172
173 r = nodeid_hash(nodeid);
174
175 hlist_for_each_entry(con, h, &connection_hash[r], list) {
176 if (con->nodeid == nodeid)
177 return con;
178 }
179 return NULL;
180}
181
182/*
183 * If 'allocation' is zero then we don't attempt to create a new
184 * connection structure for this node.
185 */
186static struct connection *__nodeid2con(int nodeid, gfp_t alloc)
187{
188 struct connection *con = NULL;
189 int r;
190
191 con = __find_con(nodeid);
192 if (con || !alloc)
193 return con;
194
195 con = kmem_cache_zalloc(con_cache, alloc);
196 if (!con)
197 return NULL;
198
199 r = nodeid_hash(nodeid);
200 hlist_add_head(&con->list, &connection_hash[r]);
201
202 con->nodeid = nodeid;
203 mutex_init(&con->sock_mutex);
204 INIT_LIST_HEAD(&con->writequeue);
205 spin_lock_init(&con->writequeue_lock);
206 INIT_WORK(&con->swork, process_send_sockets);
207 INIT_WORK(&con->rwork, process_recv_sockets);
208
209 /* Setup action pointers for child sockets */
210 if (con->nodeid) {
211 struct connection *zerocon = __find_con(0);
212
213 con->connect_action = zerocon->connect_action;
214 if (!con->rx_action)
215 con->rx_action = zerocon->rx_action;
216 }
217
218 return con;
219}
220
221/* Loop round all connections */
222static void foreach_conn(void (*conn_func)(struct connection *c))
223{
224 int i;
225 struct hlist_node *h, *n;
226 struct connection *con;
227
228 for (i = 0; i < CONN_HASH_SIZE; i++) {
229 hlist_for_each_entry_safe(con, h, n, &connection_hash[i], list){
230 conn_func(con);
231 }
232 }
233}
234
235static struct connection *nodeid2con(int nodeid, gfp_t allocation)
236{
237 struct connection *con;
238
239 mutex_lock(&connections_lock);
240 con = __nodeid2con(nodeid, allocation);
241 mutex_unlock(&connections_lock);
242
243 return con;
244}
245
246/* This is a bit drastic, but only called when things go wrong */
247static struct connection *assoc2con(int assoc_id)
248{
249 int i;
250 struct hlist_node *h;
251 struct connection *con;
252
253 mutex_lock(&connections_lock);
254
255 for (i = 0 ; i < CONN_HASH_SIZE; i++) {
256 hlist_for_each_entry(con, h, &connection_hash[i], list) {
257 if (con->sctp_assoc == assoc_id) {
258 mutex_unlock(&connections_lock);
259 return con;
260 }
261 }
262 }
263 mutex_unlock(&connections_lock);
264 return NULL;
265}
266
267static int nodeid_to_addr(int nodeid, struct sockaddr *retaddr)
268{
269 struct sockaddr_storage addr;
270 int error;
271
272 if (!dlm_local_count)
273 return -1;
274
275 error = dlm_nodeid_to_addr(nodeid, &addr);
276 if (error)
277 return error;
278
279 if (dlm_local_addr[0]->ss_family == AF_INET) {
280 struct sockaddr_in *in4 = (struct sockaddr_in *) &addr;
281 struct sockaddr_in *ret4 = (struct sockaddr_in *) retaddr;
282 ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
283 } else {
284 struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &addr;
285 struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) retaddr;
286 ret6->sin6_addr = in6->sin6_addr;
287 }
288
289 return 0;
290}
291
292/* Data available on socket or listen socket received a connect */
293static void lowcomms_data_ready(struct sock *sk, int count_unused)
294{
295 struct connection *con = sock2con(sk);
296 if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
297 queue_work(recv_workqueue, &con->rwork);
298}
299
300static void lowcomms_write_space(struct sock *sk)
301{
302 struct connection *con = sock2con(sk);
303
304 if (!con)
305 return;
306
307 clear_bit(SOCK_NOSPACE, &con->sock->flags);
308
309 if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
310 con->sock->sk->sk_write_pending--;
311 clear_bit(SOCK_ASYNC_NOSPACE, &con->sock->flags);
312 }
313
314 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
315 queue_work(send_workqueue, &con->swork);
316}
317
318static inline void lowcomms_connect_sock(struct connection *con)
319{
320 if (test_bit(CF_CLOSE, &con->flags))
321 return;
322 if (!test_and_set_bit(CF_CONNECT_PENDING, &con->flags))
323 queue_work(send_workqueue, &con->swork);
324}
325
326static void lowcomms_state_change(struct sock *sk)
327{
328 if (sk->sk_state == TCP_ESTABLISHED)
329 lowcomms_write_space(sk);
330}
331
332int dlm_lowcomms_connect_node(int nodeid)
333{
334 struct connection *con;
335
336 /* with sctp there's no connecting without sending */
337 if (dlm_config.ci_protocol != 0)
338 return 0;
339
340 if (nodeid == dlm_our_nodeid())
341 return 0;
342
343 con = nodeid2con(nodeid, GFP_NOFS);
344 if (!con)
345 return -ENOMEM;
346 lowcomms_connect_sock(con);
347 return 0;
348}
349
350/* Make a socket active */
351static int add_sock(struct socket *sock, struct connection *con)
352{
353 con->sock = sock;
354
355 /* Install a data_ready callback */
356 con->sock->sk->sk_data_ready = lowcomms_data_ready;
357 con->sock->sk->sk_write_space = lowcomms_write_space;
358 con->sock->sk->sk_state_change = lowcomms_state_change;
359 con->sock->sk->sk_user_data = con;
360 con->sock->sk->sk_allocation = GFP_NOFS;
361 return 0;
362}
363
364/* Add the port number to an IPv6 or 4 sockaddr and return the address
365 length */
366static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
367 int *addr_len)
368{
369 saddr->ss_family = dlm_local_addr[0]->ss_family;
370 if (saddr->ss_family == AF_INET) {
371 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
372 in4_addr->sin_port = cpu_to_be16(port);
373 *addr_len = sizeof(struct sockaddr_in);
374 memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
375 } else {
376 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
377 in6_addr->sin6_port = cpu_to_be16(port);
378 *addr_len = sizeof(struct sockaddr_in6);
379 }
380 memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
381}
382
383/* Close a remote connection and tidy up */
384static void close_connection(struct connection *con, bool and_other)
385{
386 mutex_lock(&con->sock_mutex);
387
388 if (con->sock) {
389 sock_release(con->sock);
390 con->sock = NULL;
391 }
392 if (con->othercon && and_other) {
393 /* Will only re-enter once. */
394 close_connection(con->othercon, false);
395 }
396 if (con->rx_page) {
397 __free_page(con->rx_page);
398 con->rx_page = NULL;
399 }
400
401 con->retries = 0;
402 mutex_unlock(&con->sock_mutex);
403}
404
405/* We only send shutdown messages to nodes that are not part of the cluster */
406static void sctp_send_shutdown(sctp_assoc_t associd)
407{
408 static char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
409 struct msghdr outmessage;
410 struct cmsghdr *cmsg;
411 struct sctp_sndrcvinfo *sinfo;
412 int ret;
413 struct connection *con;
414
415 con = nodeid2con(0,0);
416 BUG_ON(con == NULL);
417
418 outmessage.msg_name = NULL;
419 outmessage.msg_namelen = 0;
420 outmessage.msg_control = outcmsg;
421 outmessage.msg_controllen = sizeof(outcmsg);
422 outmessage.msg_flags = MSG_EOR;
423
424 cmsg = CMSG_FIRSTHDR(&outmessage);
425 cmsg->cmsg_level = IPPROTO_SCTP;
426 cmsg->cmsg_type = SCTP_SNDRCV;
427 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
428 outmessage.msg_controllen = cmsg->cmsg_len;
429 sinfo = CMSG_DATA(cmsg);
430 memset(sinfo, 0x00, sizeof(struct sctp_sndrcvinfo));
431
432 sinfo->sinfo_flags |= MSG_EOF;
433 sinfo->sinfo_assoc_id = associd;
434
435 ret = kernel_sendmsg(con->sock, &outmessage, NULL, 0, 0);
436
437 if (ret != 0)
438 log_print("send EOF to node failed: %d", ret);
439}
440
441static void sctp_init_failed_foreach(struct connection *con)
442{
443 con->sctp_assoc = 0;
444 if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
445 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags))
446 queue_work(send_workqueue, &con->swork);
447 }
448}
449
450/* INIT failed but we don't know which node...
451 restart INIT on all pending nodes */
452static void sctp_init_failed(void)
453{
454 mutex_lock(&connections_lock);
455
456 foreach_conn(sctp_init_failed_foreach);
457
458 mutex_unlock(&connections_lock);
459}
460
461/* Something happened to an association */
462static void process_sctp_notification(struct connection *con,
463 struct msghdr *msg, char *buf)
464{
465 union sctp_notification *sn = (union sctp_notification *)buf;
466
467 if (sn->sn_header.sn_type == SCTP_ASSOC_CHANGE) {
468 switch (sn->sn_assoc_change.sac_state) {
469
470 case SCTP_COMM_UP:
471 case SCTP_RESTART:
472 {
473 /* Check that the new node is in the lockspace */
474 struct sctp_prim prim;
475 int nodeid;
476 int prim_len, ret;
477 int addr_len;
478 struct connection *new_con;
479
480 /*
481 * We get this before any data for an association.
482 * We verify that the node is in the cluster and
483 * then peel off a socket for it.
484 */
485 if ((int)sn->sn_assoc_change.sac_assoc_id <= 0) {
486 log_print("COMM_UP for invalid assoc ID %d",
487 (int)sn->sn_assoc_change.sac_assoc_id);
488 sctp_init_failed();
489 return;
490 }
491 memset(&prim, 0, sizeof(struct sctp_prim));
492 prim_len = sizeof(struct sctp_prim);
493 prim.ssp_assoc_id = sn->sn_assoc_change.sac_assoc_id;
494
495 ret = kernel_getsockopt(con->sock,
496 IPPROTO_SCTP,
497 SCTP_PRIMARY_ADDR,
498 (char*)&prim,
499 &prim_len);
500 if (ret < 0) {
501 log_print("getsockopt/sctp_primary_addr on "
502 "new assoc %d failed : %d",
503 (int)sn->sn_assoc_change.sac_assoc_id,
504 ret);
505
506 /* Retry INIT later */
507 new_con = assoc2con(sn->sn_assoc_change.sac_assoc_id);
508 if (new_con)
509 clear_bit(CF_CONNECT_PENDING, &con->flags);
510 return;
511 }
512 make_sockaddr(&prim.ssp_addr, 0, &addr_len);
513 if (dlm_addr_to_nodeid(&prim.ssp_addr, &nodeid)) {
514 unsigned char *b=(unsigned char *)&prim.ssp_addr;
515 log_print("reject connect from unknown addr");
516 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
517 b, sizeof(struct sockaddr_storage));
518 sctp_send_shutdown(prim.ssp_assoc_id);
519 return;
520 }
521
522 new_con = nodeid2con(nodeid, GFP_NOFS);
523 if (!new_con)
524 return;
525
526 /* Peel off a new sock */
527 sctp_lock_sock(con->sock->sk);
528 ret = sctp_do_peeloff(con->sock->sk,
529 sn->sn_assoc_change.sac_assoc_id,
530 &new_con->sock);
531 sctp_release_sock(con->sock->sk);
532 if (ret < 0) {
533 log_print("Can't peel off a socket for "
534 "connection %d to node %d: err=%d",
535 (int)sn->sn_assoc_change.sac_assoc_id,
536 nodeid, ret);
537 return;
538 }
539 add_sock(new_con->sock, new_con);
540
541 log_print("connecting to %d sctp association %d",
542 nodeid, (int)sn->sn_assoc_change.sac_assoc_id);
543
544 /* Send any pending writes */
545 clear_bit(CF_CONNECT_PENDING, &new_con->flags);
546 clear_bit(CF_INIT_PENDING, &con->flags);
547 if (!test_and_set_bit(CF_WRITE_PENDING, &new_con->flags)) {
548 queue_work(send_workqueue, &new_con->swork);
549 }
550 if (!test_and_set_bit(CF_READ_PENDING, &new_con->flags))
551 queue_work(recv_workqueue, &new_con->rwork);
552 }
553 break;
554
555 case SCTP_COMM_LOST:
556 case SCTP_SHUTDOWN_COMP:
557 {
558 con = assoc2con(sn->sn_assoc_change.sac_assoc_id);
559 if (con) {
560 con->sctp_assoc = 0;
561 }
562 }
563 break;
564
565 /* We don't know which INIT failed, so clear the PENDING flags
566 * on them all. if assoc_id is zero then it will then try
567 * again */
568
569 case SCTP_CANT_STR_ASSOC:
570 {
571 log_print("Can't start SCTP association - retrying");
572 sctp_init_failed();
573 }
574 break;
575
576 default:
577 log_print("unexpected SCTP assoc change id=%d state=%d",
578 (int)sn->sn_assoc_change.sac_assoc_id,
579 sn->sn_assoc_change.sac_state);
580 }
581 }
582}
583
584/* Data received from remote end */
585static int receive_from_sock(struct connection *con)
586{
587 int ret = 0;
588 struct msghdr msg = {};
589 struct kvec iov[2];
590 unsigned len;
591 int r;
592 int call_again_soon = 0;
593 int nvec;
594 char incmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
595
596 mutex_lock(&con->sock_mutex);
597
598 if (con->sock == NULL) {
599 ret = -EAGAIN;
600 goto out_close;
601 }
602
603 if (con->rx_page == NULL) {
604 /*
605 * This doesn't need to be atomic, but I think it should
606 * improve performance if it is.
607 */
608 con->rx_page = alloc_page(GFP_ATOMIC);
609 if (con->rx_page == NULL)
610 goto out_resched;
611 cbuf_init(&con->cb, PAGE_CACHE_SIZE);
612 }
613
614 /* Only SCTP needs these really */
615 memset(&incmsg, 0, sizeof(incmsg));
616 msg.msg_control = incmsg;
617 msg.msg_controllen = sizeof(incmsg);
618
619 /*
620 * iov[0] is the bit of the circular buffer between the current end
621 * point (cb.base + cb.len) and the end of the buffer.
622 */
623 iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
624 iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
625 iov[1].iov_len = 0;
626 nvec = 1;
627
628 /*
629 * iov[1] is the bit of the circular buffer between the start of the
630 * buffer and the start of the currently used section (cb.base)
631 */
632 if (cbuf_data(&con->cb) >= con->cb.base) {
633 iov[0].iov_len = PAGE_CACHE_SIZE - cbuf_data(&con->cb);
634 iov[1].iov_len = con->cb.base;
635 iov[1].iov_base = page_address(con->rx_page);
636 nvec = 2;
637 }
638 len = iov[0].iov_len + iov[1].iov_len;
639
640 r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len,
641 MSG_DONTWAIT | MSG_NOSIGNAL);
642 if (ret <= 0)
643 goto out_close;
644
645 /* Process SCTP notifications */
646 if (msg.msg_flags & MSG_NOTIFICATION) {
647 msg.msg_control = incmsg;
648 msg.msg_controllen = sizeof(incmsg);
649
650 process_sctp_notification(con, &msg,
651 page_address(con->rx_page) + con->cb.base);
652 mutex_unlock(&con->sock_mutex);
653 return 0;
654 }
655 BUG_ON(con->nodeid == 0);
656
657 if (ret == len)
658 call_again_soon = 1;
659 cbuf_add(&con->cb, ret);
660 ret = dlm_process_incoming_buffer(con->nodeid,
661 page_address(con->rx_page),
662 con->cb.base, con->cb.len,
663 PAGE_CACHE_SIZE);
664 if (ret == -EBADMSG) {
665 log_print("lowcomms: addr=%p, base=%u, len=%u, "
666 "iov_len=%u, iov_base[0]=%p, read=%d",
667 page_address(con->rx_page), con->cb.base, con->cb.len,
668 len, iov[0].iov_base, r);
669 }
670 if (ret < 0)
671 goto out_close;
672 cbuf_eat(&con->cb, ret);
673
674 if (cbuf_empty(&con->cb) && !call_again_soon) {
675 __free_page(con->rx_page);
676 con->rx_page = NULL;
677 }
678
679 if (call_again_soon)
680 goto out_resched;
681 mutex_unlock(&con->sock_mutex);
682 return 0;
683
684out_resched:
685 if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
686 queue_work(recv_workqueue, &con->rwork);
687 mutex_unlock(&con->sock_mutex);
688 return -EAGAIN;
689
690out_close:
691 mutex_unlock(&con->sock_mutex);
692 if (ret != -EAGAIN) {
693 close_connection(con, false);
694 /* Reconnect when there is something to send */
695 }
696 /* Don't return success if we really got EOF */
697 if (ret == 0)
698 ret = -EAGAIN;
699
700 return ret;
701}
702
703/* Listening socket is busy, accept a connection */
704static int tcp_accept_from_sock(struct connection *con)
705{
706 int result;
707 struct sockaddr_storage peeraddr;
708 struct socket *newsock;
709 int len;
710 int nodeid;
711 struct connection *newcon;
712 struct connection *addcon;
713
714 mutex_lock(&connections_lock);
715 if (!dlm_allow_conn) {
716 mutex_unlock(&connections_lock);
717 return -1;
718 }
719 mutex_unlock(&connections_lock);
720
721 memset(&peeraddr, 0, sizeof(peeraddr));
722 result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
723 IPPROTO_TCP, &newsock);
724 if (result < 0)
725 return -ENOMEM;
726
727 mutex_lock_nested(&con->sock_mutex, 0);
728
729 result = -ENOTCONN;
730 if (con->sock == NULL)
731 goto accept_err;
732
733 newsock->type = con->sock->type;
734 newsock->ops = con->sock->ops;
735
736 result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK);
737 if (result < 0)
738 goto accept_err;
739
740 /* Get the connected socket's peer */
741 memset(&peeraddr, 0, sizeof(peeraddr));
742 if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr,
743 &len, 2)) {
744 result = -ECONNABORTED;
745 goto accept_err;
746 }
747
748 /* Get the new node's NODEID */
749 make_sockaddr(&peeraddr, 0, &len);
750 if (dlm_addr_to_nodeid(&peeraddr, &nodeid)) {
751 unsigned char *b=(unsigned char *)&peeraddr;
752 log_print("connect from non cluster node");
753 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
754 b, sizeof(struct sockaddr_storage));
755 sock_release(newsock);
756 mutex_unlock(&con->sock_mutex);
757 return -1;
758 }
759
760 log_print("got connection from %d", nodeid);
761
762 /* Check to see if we already have a connection to this node. This
763 * could happen if the two nodes initiate a connection at roughly
764 * the same time and the connections cross on the wire.
765 * In this case we store the incoming one in "othercon"
766 */
767 newcon = nodeid2con(nodeid, GFP_NOFS);
768 if (!newcon) {
769 result = -ENOMEM;
770 goto accept_err;
771 }
772 mutex_lock_nested(&newcon->sock_mutex, 1);
773 if (newcon->sock) {
774 struct connection *othercon = newcon->othercon;
775
776 if (!othercon) {
777 othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
778 if (!othercon) {
779 log_print("failed to allocate incoming socket");
780 mutex_unlock(&newcon->sock_mutex);
781 result = -ENOMEM;
782 goto accept_err;
783 }
784 othercon->nodeid = nodeid;
785 othercon->rx_action = receive_from_sock;
786 mutex_init(&othercon->sock_mutex);
787 INIT_WORK(&othercon->swork, process_send_sockets);
788 INIT_WORK(&othercon->rwork, process_recv_sockets);
789 set_bit(CF_IS_OTHERCON, &othercon->flags);
790 }
791 if (!othercon->sock) {
792 newcon->othercon = othercon;
793 othercon->sock = newsock;
794 newsock->sk->sk_user_data = othercon;
795 add_sock(newsock, othercon);
796 addcon = othercon;
797 }
798 else {
799 printk("Extra connection from node %d attempted\n", nodeid);
800 result = -EAGAIN;
801 mutex_unlock(&newcon->sock_mutex);
802 goto accept_err;
803 }
804 }
805 else {
806 newsock->sk->sk_user_data = newcon;
807 newcon->rx_action = receive_from_sock;
808 add_sock(newsock, newcon);
809 addcon = newcon;
810 }
811
812 mutex_unlock(&newcon->sock_mutex);
813
814 /*
815 * Add it to the active queue in case we got data
816 * between processing the accept adding the socket
817 * to the read_sockets list
818 */
819 if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
820 queue_work(recv_workqueue, &addcon->rwork);
821 mutex_unlock(&con->sock_mutex);
822
823 return 0;
824
825accept_err:
826 mutex_unlock(&con->sock_mutex);
827 sock_release(newsock);
828
829 if (result != -EAGAIN)
830 log_print("error accepting connection from node: %d", result);
831 return result;
832}
833
834static void free_entry(struct writequeue_entry *e)
835{
836 __free_page(e->page);
837 kfree(e);
838}
839
840/* Initiate an SCTP association.
841 This is a special case of send_to_sock() in that we don't yet have a
842 peeled-off socket for this association, so we use the listening socket
843 and add the primary IP address of the remote node.
844 */
845static void sctp_init_assoc(struct connection *con)
846{
847 struct sockaddr_storage rem_addr;
848 char outcmsg[CMSG_SPACE(sizeof(struct sctp_sndrcvinfo))];
849 struct msghdr outmessage;
850 struct cmsghdr *cmsg;
851 struct sctp_sndrcvinfo *sinfo;
852 struct connection *base_con;
853 struct writequeue_entry *e;
854 int len, offset;
855 int ret;
856 int addrlen;
857 struct kvec iov[1];
858
859 if (test_and_set_bit(CF_INIT_PENDING, &con->flags))
860 return;
861
862 if (con->retries++ > MAX_CONNECT_RETRIES)
863 return;
864
865 if (nodeid_to_addr(con->nodeid, (struct sockaddr *)&rem_addr)) {
866 log_print("no address for nodeid %d", con->nodeid);
867 return;
868 }
869 base_con = nodeid2con(0, 0);
870 BUG_ON(base_con == NULL);
871
872 make_sockaddr(&rem_addr, dlm_config.ci_tcp_port, &addrlen);
873
874 outmessage.msg_name = &rem_addr;
875 outmessage.msg_namelen = addrlen;
876 outmessage.msg_control = outcmsg;
877 outmessage.msg_controllen = sizeof(outcmsg);
878 outmessage.msg_flags = MSG_EOR;
879
880 spin_lock(&con->writequeue_lock);
881
882 if (list_empty(&con->writequeue)) {
883 spin_unlock(&con->writequeue_lock);
884 log_print("writequeue empty for nodeid %d", con->nodeid);
885 return;
886 }
887
888 e = list_first_entry(&con->writequeue, struct writequeue_entry, list);
889 len = e->len;
890 offset = e->offset;
891 spin_unlock(&con->writequeue_lock);
892
893 /* Send the first block off the write queue */
894 iov[0].iov_base = page_address(e->page)+offset;
895 iov[0].iov_len = len;
896
897 cmsg = CMSG_FIRSTHDR(&outmessage);
898 cmsg->cmsg_level = IPPROTO_SCTP;
899 cmsg->cmsg_type = SCTP_SNDRCV;
900 cmsg->cmsg_len = CMSG_LEN(sizeof(struct sctp_sndrcvinfo));
901 sinfo = CMSG_DATA(cmsg);
902 memset(sinfo, 0x00, sizeof(struct sctp_sndrcvinfo));
903 sinfo->sinfo_ppid = cpu_to_le32(dlm_our_nodeid());
904 outmessage.msg_controllen = cmsg->cmsg_len;
905
906 ret = kernel_sendmsg(base_con->sock, &outmessage, iov, 1, len);
907 if (ret < 0) {
908 log_print("Send first packet to node %d failed: %d",
909 con->nodeid, ret);
910
911 /* Try again later */
912 clear_bit(CF_CONNECT_PENDING, &con->flags);
913 clear_bit(CF_INIT_PENDING, &con->flags);
914 }
915 else {
916 spin_lock(&con->writequeue_lock);
917 e->offset += ret;
918 e->len -= ret;
919
920 if (e->len == 0 && e->users == 0) {
921 list_del(&e->list);
922 free_entry(e);
923 }
924 spin_unlock(&con->writequeue_lock);
925 }
926}
927
928/* Connect a new socket to its peer */
929static void tcp_connect_to_sock(struct connection *con)
930{
931 int result = -EHOSTUNREACH;
932 struct sockaddr_storage saddr, src_addr;
933 int addr_len;
934 struct socket *sock = NULL;
935 int one = 1;
936
937 if (con->nodeid == 0) {
938 log_print("attempt to connect sock 0 foiled");
939 return;
940 }
941
942 mutex_lock(&con->sock_mutex);
943 if (con->retries++ > MAX_CONNECT_RETRIES)
944 goto out;
945
946 /* Some odd races can cause double-connects, ignore them */
947 if (con->sock) {
948 result = 0;
949 goto out;
950 }
951
952 /* Create a socket to communicate with */
953 result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
954 IPPROTO_TCP, &sock);
955 if (result < 0)
956 goto out_err;
957
958 memset(&saddr, 0, sizeof(saddr));
959 if (dlm_nodeid_to_addr(con->nodeid, &saddr))
960 goto out_err;
961
962 sock->sk->sk_user_data = con;
963 con->rx_action = receive_from_sock;
964 con->connect_action = tcp_connect_to_sock;
965 add_sock(sock, con);
966
967 /* Bind to our cluster-known address connecting to avoid
968 routing problems */
969 memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
970 make_sockaddr(&src_addr, 0, &addr_len);
971 result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
972 addr_len);
973 if (result < 0) {
974 log_print("could not bind for connect: %d", result);
975 /* This *may* not indicate a critical error */
976 }
977
978 make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
979
980 log_print("connecting to %d", con->nodeid);
981
982 /* Turn off Nagle's algorithm */
983 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
984 sizeof(one));
985
986 result =
987 sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
988 O_NONBLOCK);
989 if (result == -EINPROGRESS)
990 result = 0;
991 if (result == 0)
992 goto out;
993
994out_err:
995 if (con->sock) {
996 sock_release(con->sock);
997 con->sock = NULL;
998 } else if (sock) {
999 sock_release(sock);
1000 }
1001 /*
1002 * Some errors are fatal and this list might need adjusting. For other
1003 * errors we try again until the max number of retries is reached.
1004 */
1005 if (result != -EHOSTUNREACH && result != -ENETUNREACH &&
1006 result != -ENETDOWN && result != -EINVAL
1007 && result != -EPROTONOSUPPORT) {
1008 lowcomms_connect_sock(con);
1009 result = 0;
1010 }
1011out:
1012 mutex_unlock(&con->sock_mutex);
1013 return;
1014}
1015
1016static struct socket *tcp_create_listen_sock(struct connection *con,
1017 struct sockaddr_storage *saddr)
1018{
1019 struct socket *sock = NULL;
1020 int result = 0;
1021 int one = 1;
1022 int addr_len;
1023
1024 if (dlm_local_addr[0]->ss_family == AF_INET)
1025 addr_len = sizeof(struct sockaddr_in);
1026 else
1027 addr_len = sizeof(struct sockaddr_in6);
1028
1029 /* Create a socket to communicate with */
1030 result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_STREAM,
1031 IPPROTO_TCP, &sock);
1032 if (result < 0) {
1033 log_print("Can't create listening comms socket");
1034 goto create_out;
1035 }
1036
1037 /* Turn off Nagle's algorithm */
1038 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1039 sizeof(one));
1040
1041 result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
1042 (char *)&one, sizeof(one));
1043
1044 if (result < 0) {
1045 log_print("Failed to set SO_REUSEADDR on socket: %d", result);
1046 }
1047 sock->sk->sk_user_data = con;
1048 con->rx_action = tcp_accept_from_sock;
1049 con->connect_action = tcp_connect_to_sock;
1050 con->sock = sock;
1051
1052 /* Bind to our port */
1053 make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
1054 result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
1055 if (result < 0) {
1056 log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
1057 sock_release(sock);
1058 sock = NULL;
1059 con->sock = NULL;
1060 goto create_out;
1061 }
1062 result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
1063 (char *)&one, sizeof(one));
1064 if (result < 0) {
1065 log_print("Set keepalive failed: %d", result);
1066 }
1067
1068 result = sock->ops->listen(sock, 5);
1069 if (result < 0) {
1070 log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
1071 sock_release(sock);
1072 sock = NULL;
1073 goto create_out;
1074 }
1075
1076create_out:
1077 return sock;
1078}
1079
1080/* Get local addresses */
1081static void init_local(void)
1082{
1083 struct sockaddr_storage sas, *addr;
1084 int i;
1085
1086 dlm_local_count = 0;
1087 for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
1088 if (dlm_our_addr(&sas, i))
1089 break;
1090
1091 addr = kmalloc(sizeof(*addr), GFP_NOFS);
1092 if (!addr)
1093 break;
1094 memcpy(addr, &sas, sizeof(*addr));
1095 dlm_local_addr[dlm_local_count++] = addr;
1096 }
1097}
1098
1099/* Bind to an IP address. SCTP allows multiple address so it can do
1100 multi-homing */
1101static int add_sctp_bind_addr(struct connection *sctp_con,
1102 struct sockaddr_storage *addr,
1103 int addr_len, int num)
1104{
1105 int result = 0;
1106
1107 if (num == 1)
1108 result = kernel_bind(sctp_con->sock,
1109 (struct sockaddr *) addr,
1110 addr_len);
1111 else
1112 result = kernel_setsockopt(sctp_con->sock, SOL_SCTP,
1113 SCTP_SOCKOPT_BINDX_ADD,
1114 (char *)addr, addr_len);
1115
1116 if (result < 0)
1117 log_print("Can't bind to port %d addr number %d",
1118 dlm_config.ci_tcp_port, num);
1119
1120 return result;
1121}
1122
1123/* Initialise SCTP socket and bind to all interfaces */
1124static int sctp_listen_for_all(void)
1125{
1126 struct socket *sock = NULL;
1127 struct sockaddr_storage localaddr;
1128 struct sctp_event_subscribe subscribe;
1129 int result = -EINVAL, num = 1, i, addr_len;
1130 struct connection *con = nodeid2con(0, GFP_NOFS);
1131 int bufsize = NEEDED_RMEM;
1132
1133 if (!con)
1134 return -ENOMEM;
1135
1136 log_print("Using SCTP for communications");
1137
1138 result = sock_create_kern(dlm_local_addr[0]->ss_family, SOCK_SEQPACKET,
1139 IPPROTO_SCTP, &sock);
1140 if (result < 0) {
1141 log_print("Can't create comms socket, check SCTP is loaded");
1142 goto out;
1143 }
1144
1145 /* Listen for events */
1146 memset(&subscribe, 0, sizeof(subscribe));
1147 subscribe.sctp_data_io_event = 1;
1148 subscribe.sctp_association_event = 1;
1149 subscribe.sctp_send_failure_event = 1;
1150 subscribe.sctp_shutdown_event = 1;
1151 subscribe.sctp_partial_delivery_event = 1;
1152
1153 result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
1154 (char *)&bufsize, sizeof(bufsize));
1155 if (result)
1156 log_print("Error increasing buffer space on socket %d", result);
1157
1158 result = kernel_setsockopt(sock, SOL_SCTP, SCTP_EVENTS,
1159 (char *)&subscribe, sizeof(subscribe));
1160 if (result < 0) {
1161 log_print("Failed to set SCTP_EVENTS on socket: result=%d",
1162 result);
1163 goto create_delsock;
1164 }
1165
1166 /* Init con struct */
1167 sock->sk->sk_user_data = con;
1168 con->sock = sock;
1169 con->sock->sk->sk_data_ready = lowcomms_data_ready;
1170 con->rx_action = receive_from_sock;
1171 con->connect_action = sctp_init_assoc;
1172
1173 /* Bind to all interfaces. */
1174 for (i = 0; i < dlm_local_count; i++) {
1175 memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr));
1176 make_sockaddr(&localaddr, dlm_config.ci_tcp_port, &addr_len);
1177
1178 result = add_sctp_bind_addr(con, &localaddr, addr_len, num);
1179 if (result)
1180 goto create_delsock;
1181 ++num;
1182 }
1183
1184 result = sock->ops->listen(sock, 5);
1185 if (result < 0) {
1186 log_print("Can't set socket listening");
1187 goto create_delsock;
1188 }
1189
1190 return 0;
1191
1192create_delsock:
1193 sock_release(sock);
1194 con->sock = NULL;
1195out:
1196 return result;
1197}
1198
1199static int tcp_listen_for_all(void)
1200{
1201 struct socket *sock = NULL;
1202 struct connection *con = nodeid2con(0, GFP_NOFS);
1203 int result = -EINVAL;
1204
1205 if (!con)
1206 return -ENOMEM;
1207
1208 /* We don't support multi-homed hosts */
1209 if (dlm_local_addr[1] != NULL) {
1210 log_print("TCP protocol can't handle multi-homed hosts, "
1211 "try SCTP");
1212 return -EINVAL;
1213 }
1214
1215 log_print("Using TCP for communications");
1216
1217 sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
1218 if (sock) {
1219 add_sock(sock, con);
1220 result = 0;
1221 }
1222 else {
1223 result = -EADDRINUSE;
1224 }
1225
1226 return result;
1227}
1228
1229
1230
1231static struct writequeue_entry *new_writequeue_entry(struct connection *con,
1232 gfp_t allocation)
1233{
1234 struct writequeue_entry *entry;
1235
1236 entry = kmalloc(sizeof(struct writequeue_entry), allocation);
1237 if (!entry)
1238 return NULL;
1239
1240 entry->page = alloc_page(allocation);
1241 if (!entry->page) {
1242 kfree(entry);
1243 return NULL;
1244 }
1245
1246 entry->offset = 0;
1247 entry->len = 0;
1248 entry->end = 0;
1249 entry->users = 0;
1250 entry->con = con;
1251
1252 return entry;
1253}
1254
1255void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
1256{
1257 struct connection *con;
1258 struct writequeue_entry *e;
1259 int offset = 0;
1260 int users = 0;
1261
1262 con = nodeid2con(nodeid, allocation);
1263 if (!con)
1264 return NULL;
1265
1266 spin_lock(&con->writequeue_lock);
1267 e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
1268 if ((&e->list == &con->writequeue) ||
1269 (PAGE_CACHE_SIZE - e->end < len)) {
1270 e = NULL;
1271 } else {
1272 offset = e->end;
1273 e->end += len;
1274 users = e->users++;
1275 }
1276 spin_unlock(&con->writequeue_lock);
1277
1278 if (e) {
1279 got_one:
1280 *ppc = page_address(e->page) + offset;
1281 return e;
1282 }
1283
1284 e = new_writequeue_entry(con, allocation);
1285 if (e) {
1286 spin_lock(&con->writequeue_lock);
1287 offset = e->end;
1288 e->end += len;
1289 users = e->users++;
1290 list_add_tail(&e->list, &con->writequeue);
1291 spin_unlock(&con->writequeue_lock);
1292 goto got_one;
1293 }
1294 return NULL;
1295}
1296
1297void dlm_lowcomms_commit_buffer(void *mh)
1298{
1299 struct writequeue_entry *e = (struct writequeue_entry *)mh;
1300 struct connection *con = e->con;
1301 int users;
1302
1303 spin_lock(&con->writequeue_lock);
1304 users = --e->users;
1305 if (users)
1306 goto out;
1307 e->len = e->end - e->offset;
1308 spin_unlock(&con->writequeue_lock);
1309
1310 if (!test_and_set_bit(CF_WRITE_PENDING, &con->flags)) {
1311 queue_work(send_workqueue, &con->swork);
1312 }
1313 return;
1314
1315out:
1316 spin_unlock(&con->writequeue_lock);
1317 return;
1318}
1319
1320/* Send a message */
1321static void send_to_sock(struct connection *con)
1322{
1323 int ret = 0;
1324 const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
1325 struct writequeue_entry *e;
1326 int len, offset;
1327 int count = 0;
1328
1329 mutex_lock(&con->sock_mutex);
1330 if (con->sock == NULL)
1331 goto out_connect;
1332
1333 spin_lock(&con->writequeue_lock);
1334 for (;;) {
1335 e = list_entry(con->writequeue.next, struct writequeue_entry,
1336 list);
1337 if ((struct list_head *) e == &con->writequeue)
1338 break;
1339
1340 len = e->len;
1341 offset = e->offset;
1342 BUG_ON(len == 0 && e->users == 0);
1343 spin_unlock(&con->writequeue_lock);
1344
1345 ret = 0;
1346 if (len) {
1347 ret = kernel_sendpage(con->sock, e->page, offset, len,
1348 msg_flags);
1349 if (ret == -EAGAIN || ret == 0) {
1350 if (ret == -EAGAIN &&
1351 test_bit(SOCK_ASYNC_NOSPACE, &con->sock->flags) &&
1352 !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1353 /* Notify TCP that we're limited by the
1354 * application window size.
1355 */
1356 set_bit(SOCK_NOSPACE, &con->sock->flags);
1357 con->sock->sk->sk_write_pending++;
1358 }
1359 cond_resched();
1360 goto out;
1361 }
1362 if (ret <= 0)
1363 goto send_error;
1364 }
1365
1366 /* Don't starve people filling buffers */
1367 if (++count >= MAX_SEND_MSG_COUNT) {
1368 cond_resched();
1369 count = 0;
1370 }
1371
1372 spin_lock(&con->writequeue_lock);
1373 e->offset += ret;
1374 e->len -= ret;
1375
1376 if (e->len == 0 && e->users == 0) {
1377 list_del(&e->list);
1378 free_entry(e);
1379 continue;
1380 }
1381 }
1382 spin_unlock(&con->writequeue_lock);
1383out:
1384 mutex_unlock(&con->sock_mutex);
1385 return;
1386
1387send_error:
1388 mutex_unlock(&con->sock_mutex);
1389 close_connection(con, false);
1390 lowcomms_connect_sock(con);
1391 return;
1392
1393out_connect:
1394 mutex_unlock(&con->sock_mutex);
1395 if (!test_bit(CF_INIT_PENDING, &con->flags))
1396 lowcomms_connect_sock(con);
1397 return;
1398}
1399
1400static void clean_one_writequeue(struct connection *con)
1401{
1402 struct writequeue_entry *e, *safe;
1403
1404 spin_lock(&con->writequeue_lock);
1405 list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1406 list_del(&e->list);
1407 free_entry(e);
1408 }
1409 spin_unlock(&con->writequeue_lock);
1410}
1411
1412/* Called from recovery when it knows that a node has
1413 left the cluster */
1414int dlm_lowcomms_close(int nodeid)
1415{
1416 struct connection *con;
1417
1418 log_print("closing connection to node %d", nodeid);
1419 con = nodeid2con(nodeid, 0);
1420 if (con) {
1421 clear_bit(CF_CONNECT_PENDING, &con->flags);
1422 clear_bit(CF_WRITE_PENDING, &con->flags);
1423 set_bit(CF_CLOSE, &con->flags);
1424 if (cancel_work_sync(&con->swork))
1425 log_print("canceled swork for node %d", nodeid);
1426 if (cancel_work_sync(&con->rwork))
1427 log_print("canceled rwork for node %d", nodeid);
1428 clean_one_writequeue(con);
1429 close_connection(con, true);
1430 }
1431 return 0;
1432}
1433
1434/* Receive workqueue function */
1435static void process_recv_sockets(struct work_struct *work)
1436{
1437 struct connection *con = container_of(work, struct connection, rwork);
1438 int err;
1439
1440 clear_bit(CF_READ_PENDING, &con->flags);
1441 do {
1442 err = con->rx_action(con);
1443 } while (!err);
1444}
1445
1446/* Send workqueue function */
1447static void process_send_sockets(struct work_struct *work)
1448{
1449 struct connection *con = container_of(work, struct connection, swork);
1450
1451 if (test_and_clear_bit(CF_CONNECT_PENDING, &con->flags)) {
1452 con->connect_action(con);
1453 set_bit(CF_WRITE_PENDING, &con->flags);
1454 }
1455 if (test_and_clear_bit(CF_WRITE_PENDING, &con->flags))
1456 send_to_sock(con);
1457}
1458
1459
1460/* Discard all entries on the write queues */
1461static void clean_writequeues(void)
1462{
1463 foreach_conn(clean_one_writequeue);
1464}
1465
1466static void work_stop(void)
1467{
1468 destroy_workqueue(recv_workqueue);
1469 destroy_workqueue(send_workqueue);
1470}
1471
1472static int work_start(void)
1473{
1474 recv_workqueue = alloc_workqueue("dlm_recv",
1475 WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1476 if (!recv_workqueue) {
1477 log_print("can't start dlm_recv");
1478 return -ENOMEM;
1479 }
1480
1481 send_workqueue = alloc_workqueue("dlm_send",
1482 WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1483 if (!send_workqueue) {
1484 log_print("can't start dlm_send");
1485 destroy_workqueue(recv_workqueue);
1486 return -ENOMEM;
1487 }
1488
1489 return 0;
1490}
1491
1492static void stop_conn(struct connection *con)
1493{
1494 con->flags |= 0x0F;
1495 if (con->sock && con->sock->sk)
1496 con->sock->sk->sk_user_data = NULL;
1497}
1498
1499static void free_conn(struct connection *con)
1500{
1501 close_connection(con, true);
1502 if (con->othercon)
1503 kmem_cache_free(con_cache, con->othercon);
1504 hlist_del(&con->list);
1505 kmem_cache_free(con_cache, con);
1506}
1507
1508void dlm_lowcomms_stop(void)
1509{
1510 /* Set all the flags to prevent any
1511 socket activity.
1512 */
1513 mutex_lock(&connections_lock);
1514 dlm_allow_conn = 0;
1515 foreach_conn(stop_conn);
1516 mutex_unlock(&connections_lock);
1517
1518 work_stop();
1519
1520 mutex_lock(&connections_lock);
1521 clean_writequeues();
1522
1523 foreach_conn(free_conn);
1524
1525 mutex_unlock(&connections_lock);
1526 kmem_cache_destroy(con_cache);
1527}
1528
1529int dlm_lowcomms_start(void)
1530{
1531 int error = -EINVAL;
1532 struct connection *con;
1533 int i;
1534
1535 for (i = 0; i < CONN_HASH_SIZE; i++)
1536 INIT_HLIST_HEAD(&connection_hash[i]);
1537
1538 init_local();
1539 if (!dlm_local_count) {
1540 error = -ENOTCONN;
1541 log_print("no local IP address has been set");
1542 goto fail;
1543 }
1544
1545 error = -ENOMEM;
1546 con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
1547 __alignof__(struct connection), 0,
1548 NULL);
1549 if (!con_cache)
1550 goto fail;
1551
1552 error = work_start();
1553 if (error)
1554 goto fail_destroy;
1555
1556 dlm_allow_conn = 1;
1557
1558 /* Start listening */
1559 if (dlm_config.ci_protocol == 0)
1560 error = tcp_listen_for_all();
1561 else
1562 error = sctp_listen_for_all();
1563 if (error)
1564 goto fail_unlisten;
1565
1566 return 0;
1567
1568fail_unlisten:
1569 dlm_allow_conn = 0;
1570 con = nodeid2con(0,0);
1571 if (con) {
1572 close_connection(con, false);
1573 kmem_cache_free(con_cache, con);
1574 }
1575fail_destroy:
1576 kmem_cache_destroy(con_cache);
1577fail:
1578 return error;
1579}
1// SPDX-License-Identifier: GPL-2.0-only
2/******************************************************************************
3*******************************************************************************
4**
5** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved.
6** Copyright (C) 2004-2009 Red Hat, Inc. All rights reserved.
7**
8**
9*******************************************************************************
10******************************************************************************/
11
12/*
13 * lowcomms.c
14 *
15 * This is the "low-level" comms layer.
16 *
17 * It is responsible for sending/receiving messages
18 * from other nodes in the cluster.
19 *
20 * Cluster nodes are referred to by their nodeids. nodeids are
21 * simply 32 bit numbers to the locking module - if they need to
22 * be expanded for the cluster infrastructure then that is its
23 * responsibility. It is this layer's
24 * responsibility to resolve these into IP address or
25 * whatever it needs for inter-node communication.
26 *
27 * The comms level is two kernel threads that deal mainly with
28 * the receiving of messages from other nodes and passing them
29 * up to the mid-level comms layer (which understands the
30 * message format) for execution by the locking core, and
31 * a send thread which does all the setting up of connections
32 * to remote nodes and the sending of data. Threads are not allowed
33 * to send their own data because it may cause them to wait in times
34 * of high load. Also, this way, the sending thread can collect together
35 * messages bound for one node and send them in one block.
36 *
37 * lowcomms will choose to use either TCP or SCTP as its transport layer
38 * depending on the configuration variable 'protocol'. This should be set
39 * to 0 (default) for TCP or 1 for SCTP. It should be configured using a
40 * cluster-wide mechanism as it must be the same on all nodes of the cluster
41 * for the DLM to function.
42 *
43 */
44
45#include <asm/ioctls.h>
46#include <net/sock.h>
47#include <net/tcp.h>
48#include <linux/pagemap.h>
49#include <linux/file.h>
50#include <linux/mutex.h>
51#include <linux/sctp.h>
52#include <linux/slab.h>
53#include <net/sctp/sctp.h>
54#include <net/ipv6.h>
55
56#include <trace/events/dlm.h>
57#include <trace/events/sock.h>
58
59#include "dlm_internal.h"
60#include "lowcomms.h"
61#include "midcomms.h"
62#include "memory.h"
63#include "config.h"
64
65#define DLM_SHUTDOWN_WAIT_TIMEOUT msecs_to_jiffies(5000)
66#define DLM_MAX_PROCESS_BUFFERS 24
67#define NEEDED_RMEM (4*1024*1024)
68
69struct connection {
70 struct socket *sock; /* NULL if not connected */
71 uint32_t nodeid; /* So we know who we are in the list */
72 /* this semaphore is used to allow parallel recv/send in read
73 * lock mode. When we release a sock we need to held the write lock.
74 *
75 * However this is locking code and not nice. When we remove the
76 * othercon handling we can look into other mechanism to synchronize
77 * io handling to call sock_release() at the right time.
78 */
79 struct rw_semaphore sock_lock;
80 unsigned long flags;
81#define CF_APP_LIMITED 0
82#define CF_RECV_PENDING 1
83#define CF_SEND_PENDING 2
84#define CF_RECV_INTR 3
85#define CF_IO_STOP 4
86#define CF_IS_OTHERCON 5
87 struct list_head writequeue; /* List of outgoing writequeue_entries */
88 spinlock_t writequeue_lock;
89 int retries;
90 struct hlist_node list;
91 /* due some connect()/accept() races we currently have this cross over
92 * connection attempt second connection for one node.
93 *
94 * There is a solution to avoid the race by introducing a connect
95 * rule as e.g. our_nodeid > nodeid_to_connect who is allowed to
96 * connect. Otherside can connect but will only be considered that
97 * the other side wants to have a reconnect.
98 *
99 * However changing to this behaviour will break backwards compatible.
100 * In a DLM protocol major version upgrade we should remove this!
101 */
102 struct connection *othercon;
103 struct work_struct rwork; /* receive worker */
104 struct work_struct swork; /* send worker */
105 wait_queue_head_t shutdown_wait;
106 unsigned char rx_leftover_buf[DLM_MAX_SOCKET_BUFSIZE];
107 int rx_leftover;
108 int mark;
109 int addr_count;
110 int curr_addr_index;
111 struct sockaddr_storage addr[DLM_MAX_ADDR_COUNT];
112 spinlock_t addrs_lock;
113 struct rcu_head rcu;
114};
115#define sock2con(x) ((struct connection *)(x)->sk_user_data)
116
117struct listen_connection {
118 struct socket *sock;
119 struct work_struct rwork;
120};
121
122#define DLM_WQ_REMAIN_BYTES(e) (PAGE_SIZE - e->end)
123#define DLM_WQ_LENGTH_BYTES(e) (e->end - e->offset)
124
125/* An entry waiting to be sent */
126struct writequeue_entry {
127 struct list_head list;
128 struct page *page;
129 int offset;
130 int len;
131 int end;
132 int users;
133 bool dirty;
134 struct connection *con;
135 struct list_head msgs;
136 struct kref ref;
137};
138
139struct dlm_msg {
140 struct writequeue_entry *entry;
141 struct dlm_msg *orig_msg;
142 bool retransmit;
143 void *ppc;
144 int len;
145 int idx; /* new()/commit() idx exchange */
146
147 struct list_head list;
148 struct kref ref;
149};
150
151struct processqueue_entry {
152 unsigned char *buf;
153 int nodeid;
154 int buflen;
155
156 struct list_head list;
157};
158
159struct dlm_proto_ops {
160 bool try_new_addr;
161 const char *name;
162 int proto;
163
164 int (*connect)(struct connection *con, struct socket *sock,
165 struct sockaddr *addr, int addr_len);
166 void (*sockopts)(struct socket *sock);
167 int (*bind)(struct socket *sock);
168 int (*listen_validate)(void);
169 void (*listen_sockopts)(struct socket *sock);
170 int (*listen_bind)(struct socket *sock);
171};
172
173static struct listen_sock_callbacks {
174 void (*sk_error_report)(struct sock *);
175 void (*sk_data_ready)(struct sock *);
176 void (*sk_state_change)(struct sock *);
177 void (*sk_write_space)(struct sock *);
178} listen_sock;
179
180static struct listen_connection listen_con;
181static struct sockaddr_storage dlm_local_addr[DLM_MAX_ADDR_COUNT];
182static int dlm_local_count;
183
184/* Work queues */
185static struct workqueue_struct *io_workqueue;
186static struct workqueue_struct *process_workqueue;
187
188static struct hlist_head connection_hash[CONN_HASH_SIZE];
189static DEFINE_SPINLOCK(connections_lock);
190DEFINE_STATIC_SRCU(connections_srcu);
191
192static const struct dlm_proto_ops *dlm_proto_ops;
193
194#define DLM_IO_SUCCESS 0
195#define DLM_IO_END 1
196#define DLM_IO_EOF 2
197#define DLM_IO_RESCHED 3
198#define DLM_IO_FLUSH 4
199
200static void process_recv_sockets(struct work_struct *work);
201static void process_send_sockets(struct work_struct *work);
202static void process_dlm_messages(struct work_struct *work);
203
204static DECLARE_WORK(process_work, process_dlm_messages);
205static DEFINE_SPINLOCK(processqueue_lock);
206static bool process_dlm_messages_pending;
207static atomic_t processqueue_count;
208static LIST_HEAD(processqueue);
209
210bool dlm_lowcomms_is_running(void)
211{
212 return !!listen_con.sock;
213}
214
215static void lowcomms_queue_swork(struct connection *con)
216{
217 assert_spin_locked(&con->writequeue_lock);
218
219 if (!test_bit(CF_IO_STOP, &con->flags) &&
220 !test_bit(CF_APP_LIMITED, &con->flags) &&
221 !test_and_set_bit(CF_SEND_PENDING, &con->flags))
222 queue_work(io_workqueue, &con->swork);
223}
224
225static void lowcomms_queue_rwork(struct connection *con)
226{
227#ifdef CONFIG_LOCKDEP
228 WARN_ON_ONCE(!lockdep_sock_is_held(con->sock->sk));
229#endif
230
231 if (!test_bit(CF_IO_STOP, &con->flags) &&
232 !test_and_set_bit(CF_RECV_PENDING, &con->flags))
233 queue_work(io_workqueue, &con->rwork);
234}
235
236static void writequeue_entry_ctor(void *data)
237{
238 struct writequeue_entry *entry = data;
239
240 INIT_LIST_HEAD(&entry->msgs);
241}
242
243struct kmem_cache *dlm_lowcomms_writequeue_cache_create(void)
244{
245 return kmem_cache_create("dlm_writequeue", sizeof(struct writequeue_entry),
246 0, 0, writequeue_entry_ctor);
247}
248
249struct kmem_cache *dlm_lowcomms_msg_cache_create(void)
250{
251 return kmem_cache_create("dlm_msg", sizeof(struct dlm_msg), 0, 0, NULL);
252}
253
254/* need to held writequeue_lock */
255static struct writequeue_entry *con_next_wq(struct connection *con)
256{
257 struct writequeue_entry *e;
258
259 e = list_first_entry_or_null(&con->writequeue, struct writequeue_entry,
260 list);
261 /* if len is zero nothing is to send, if there are users filling
262 * buffers we wait until the users are done so we can send more.
263 */
264 if (!e || e->users || e->len == 0)
265 return NULL;
266
267 return e;
268}
269
270static struct connection *__find_con(int nodeid, int r)
271{
272 struct connection *con;
273
274 hlist_for_each_entry_rcu(con, &connection_hash[r], list) {
275 if (con->nodeid == nodeid)
276 return con;
277 }
278
279 return NULL;
280}
281
282static void dlm_con_init(struct connection *con, int nodeid)
283{
284 con->nodeid = nodeid;
285 init_rwsem(&con->sock_lock);
286 INIT_LIST_HEAD(&con->writequeue);
287 spin_lock_init(&con->writequeue_lock);
288 INIT_WORK(&con->swork, process_send_sockets);
289 INIT_WORK(&con->rwork, process_recv_sockets);
290 spin_lock_init(&con->addrs_lock);
291 init_waitqueue_head(&con->shutdown_wait);
292}
293
294/*
295 * If 'allocation' is zero then we don't attempt to create a new
296 * connection structure for this node.
297 */
298static struct connection *nodeid2con(int nodeid, gfp_t alloc)
299{
300 struct connection *con, *tmp;
301 int r;
302
303 r = nodeid_hash(nodeid);
304 con = __find_con(nodeid, r);
305 if (con || !alloc)
306 return con;
307
308 con = kzalloc(sizeof(*con), alloc);
309 if (!con)
310 return NULL;
311
312 dlm_con_init(con, nodeid);
313
314 spin_lock(&connections_lock);
315 /* Because multiple workqueues/threads calls this function it can
316 * race on multiple cpu's. Instead of locking hot path __find_con()
317 * we just check in rare cases of recently added nodes again
318 * under protection of connections_lock. If this is the case we
319 * abort our connection creation and return the existing connection.
320 */
321 tmp = __find_con(nodeid, r);
322 if (tmp) {
323 spin_unlock(&connections_lock);
324 kfree(con);
325 return tmp;
326 }
327
328 hlist_add_head_rcu(&con->list, &connection_hash[r]);
329 spin_unlock(&connections_lock);
330
331 return con;
332}
333
334static int addr_compare(const struct sockaddr_storage *x,
335 const struct sockaddr_storage *y)
336{
337 switch (x->ss_family) {
338 case AF_INET: {
339 struct sockaddr_in *sinx = (struct sockaddr_in *)x;
340 struct sockaddr_in *siny = (struct sockaddr_in *)y;
341 if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr)
342 return 0;
343 if (sinx->sin_port != siny->sin_port)
344 return 0;
345 break;
346 }
347 case AF_INET6: {
348 struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x;
349 struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y;
350 if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr))
351 return 0;
352 if (sinx->sin6_port != siny->sin6_port)
353 return 0;
354 break;
355 }
356 default:
357 return 0;
358 }
359 return 1;
360}
361
362static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
363 struct sockaddr *sa_out, bool try_new_addr,
364 unsigned int *mark)
365{
366 struct sockaddr_storage sas;
367 struct connection *con;
368 int idx;
369
370 if (!dlm_local_count)
371 return -1;
372
373 idx = srcu_read_lock(&connections_srcu);
374 con = nodeid2con(nodeid, 0);
375 if (!con) {
376 srcu_read_unlock(&connections_srcu, idx);
377 return -ENOENT;
378 }
379
380 spin_lock(&con->addrs_lock);
381 if (!con->addr_count) {
382 spin_unlock(&con->addrs_lock);
383 srcu_read_unlock(&connections_srcu, idx);
384 return -ENOENT;
385 }
386
387 memcpy(&sas, &con->addr[con->curr_addr_index],
388 sizeof(struct sockaddr_storage));
389
390 if (try_new_addr) {
391 con->curr_addr_index++;
392 if (con->curr_addr_index == con->addr_count)
393 con->curr_addr_index = 0;
394 }
395
396 *mark = con->mark;
397 spin_unlock(&con->addrs_lock);
398
399 if (sas_out)
400 memcpy(sas_out, &sas, sizeof(struct sockaddr_storage));
401
402 if (!sa_out) {
403 srcu_read_unlock(&connections_srcu, idx);
404 return 0;
405 }
406
407 if (dlm_local_addr[0].ss_family == AF_INET) {
408 struct sockaddr_in *in4 = (struct sockaddr_in *) &sas;
409 struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;
410 ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
411 } else {
412 struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &sas;
413 struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;
414 ret6->sin6_addr = in6->sin6_addr;
415 }
416
417 srcu_read_unlock(&connections_srcu, idx);
418 return 0;
419}
420
421static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid,
422 unsigned int *mark)
423{
424 struct connection *con;
425 int i, idx, addr_i;
426
427 idx = srcu_read_lock(&connections_srcu);
428 for (i = 0; i < CONN_HASH_SIZE; i++) {
429 hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
430 WARN_ON_ONCE(!con->addr_count);
431
432 spin_lock(&con->addrs_lock);
433 for (addr_i = 0; addr_i < con->addr_count; addr_i++) {
434 if (addr_compare(&con->addr[addr_i], addr)) {
435 *nodeid = con->nodeid;
436 *mark = con->mark;
437 spin_unlock(&con->addrs_lock);
438 srcu_read_unlock(&connections_srcu, idx);
439 return 0;
440 }
441 }
442 spin_unlock(&con->addrs_lock);
443 }
444 }
445 srcu_read_unlock(&connections_srcu, idx);
446
447 return -ENOENT;
448}
449
450static bool dlm_lowcomms_con_has_addr(const struct connection *con,
451 const struct sockaddr_storage *addr)
452{
453 int i;
454
455 for (i = 0; i < con->addr_count; i++) {
456 if (addr_compare(&con->addr[i], addr))
457 return true;
458 }
459
460 return false;
461}
462
463int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len)
464{
465 struct connection *con;
466 bool ret, idx;
467
468 idx = srcu_read_lock(&connections_srcu);
469 con = nodeid2con(nodeid, GFP_NOFS);
470 if (!con) {
471 srcu_read_unlock(&connections_srcu, idx);
472 return -ENOMEM;
473 }
474
475 spin_lock(&con->addrs_lock);
476 if (!con->addr_count) {
477 memcpy(&con->addr[0], addr, sizeof(*addr));
478 con->addr_count = 1;
479 con->mark = dlm_config.ci_mark;
480 spin_unlock(&con->addrs_lock);
481 srcu_read_unlock(&connections_srcu, idx);
482 return 0;
483 }
484
485 ret = dlm_lowcomms_con_has_addr(con, addr);
486 if (ret) {
487 spin_unlock(&con->addrs_lock);
488 srcu_read_unlock(&connections_srcu, idx);
489 return -EEXIST;
490 }
491
492 if (con->addr_count >= DLM_MAX_ADDR_COUNT) {
493 spin_unlock(&con->addrs_lock);
494 srcu_read_unlock(&connections_srcu, idx);
495 return -ENOSPC;
496 }
497
498 memcpy(&con->addr[con->addr_count++], addr, sizeof(*addr));
499 srcu_read_unlock(&connections_srcu, idx);
500 spin_unlock(&con->addrs_lock);
501 return 0;
502}
503
504/* Data available on socket or listen socket received a connect */
505static void lowcomms_data_ready(struct sock *sk)
506{
507 struct connection *con = sock2con(sk);
508
509 trace_sk_data_ready(sk);
510
511 set_bit(CF_RECV_INTR, &con->flags);
512 lowcomms_queue_rwork(con);
513}
514
515static void lowcomms_write_space(struct sock *sk)
516{
517 struct connection *con = sock2con(sk);
518
519 clear_bit(SOCK_NOSPACE, &con->sock->flags);
520
521 spin_lock_bh(&con->writequeue_lock);
522 if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
523 con->sock->sk->sk_write_pending--;
524 clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags);
525 }
526
527 lowcomms_queue_swork(con);
528 spin_unlock_bh(&con->writequeue_lock);
529}
530
531static void lowcomms_state_change(struct sock *sk)
532{
533 /* SCTP layer is not calling sk_data_ready when the connection
534 * is done, so we catch the signal through here.
535 */
536 if (sk->sk_shutdown == RCV_SHUTDOWN)
537 lowcomms_data_ready(sk);
538}
539
540static void lowcomms_listen_data_ready(struct sock *sk)
541{
542 trace_sk_data_ready(sk);
543
544 queue_work(io_workqueue, &listen_con.rwork);
545}
546
547int dlm_lowcomms_connect_node(int nodeid)
548{
549 struct connection *con;
550 int idx;
551
552 idx = srcu_read_lock(&connections_srcu);
553 con = nodeid2con(nodeid, 0);
554 if (WARN_ON_ONCE(!con)) {
555 srcu_read_unlock(&connections_srcu, idx);
556 return -ENOENT;
557 }
558
559 down_read(&con->sock_lock);
560 if (!con->sock) {
561 spin_lock_bh(&con->writequeue_lock);
562 lowcomms_queue_swork(con);
563 spin_unlock_bh(&con->writequeue_lock);
564 }
565 up_read(&con->sock_lock);
566 srcu_read_unlock(&connections_srcu, idx);
567
568 cond_resched();
569 return 0;
570}
571
572int dlm_lowcomms_nodes_set_mark(int nodeid, unsigned int mark)
573{
574 struct connection *con;
575 int idx;
576
577 idx = srcu_read_lock(&connections_srcu);
578 con = nodeid2con(nodeid, 0);
579 if (!con) {
580 srcu_read_unlock(&connections_srcu, idx);
581 return -ENOENT;
582 }
583
584 spin_lock(&con->addrs_lock);
585 con->mark = mark;
586 spin_unlock(&con->addrs_lock);
587 srcu_read_unlock(&connections_srcu, idx);
588 return 0;
589}
590
591static void lowcomms_error_report(struct sock *sk)
592{
593 struct connection *con = sock2con(sk);
594 struct inet_sock *inet;
595
596 inet = inet_sk(sk);
597 switch (sk->sk_family) {
598 case AF_INET:
599 printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
600 "sending to node %d at %pI4, dport %d, "
601 "sk_err=%d/%d\n", dlm_our_nodeid(),
602 con->nodeid, &inet->inet_daddr,
603 ntohs(inet->inet_dport), sk->sk_err,
604 READ_ONCE(sk->sk_err_soft));
605 break;
606#if IS_ENABLED(CONFIG_IPV6)
607 case AF_INET6:
608 printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
609 "sending to node %d at %pI6c, "
610 "dport %d, sk_err=%d/%d\n", dlm_our_nodeid(),
611 con->nodeid, &sk->sk_v6_daddr,
612 ntohs(inet->inet_dport), sk->sk_err,
613 READ_ONCE(sk->sk_err_soft));
614 break;
615#endif
616 default:
617 printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
618 "invalid socket family %d set, "
619 "sk_err=%d/%d\n", dlm_our_nodeid(),
620 sk->sk_family, sk->sk_err,
621 READ_ONCE(sk->sk_err_soft));
622 break;
623 }
624
625 dlm_midcomms_unack_msg_resend(con->nodeid);
626
627 listen_sock.sk_error_report(sk);
628}
629
630static void restore_callbacks(struct sock *sk)
631{
632#ifdef CONFIG_LOCKDEP
633 WARN_ON_ONCE(!lockdep_sock_is_held(sk));
634#endif
635
636 sk->sk_user_data = NULL;
637 sk->sk_data_ready = listen_sock.sk_data_ready;
638 sk->sk_state_change = listen_sock.sk_state_change;
639 sk->sk_write_space = listen_sock.sk_write_space;
640 sk->sk_error_report = listen_sock.sk_error_report;
641}
642
643/* Make a socket active */
644static void add_sock(struct socket *sock, struct connection *con)
645{
646 struct sock *sk = sock->sk;
647
648 lock_sock(sk);
649 con->sock = sock;
650
651 sk->sk_user_data = con;
652 sk->sk_data_ready = lowcomms_data_ready;
653 sk->sk_write_space = lowcomms_write_space;
654 if (dlm_config.ci_protocol == DLM_PROTO_SCTP)
655 sk->sk_state_change = lowcomms_state_change;
656 sk->sk_allocation = GFP_NOFS;
657 sk->sk_use_task_frag = false;
658 sk->sk_error_report = lowcomms_error_report;
659 release_sock(sk);
660}
661
662/* Add the port number to an IPv6 or 4 sockaddr and return the address
663 length */
664static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
665 int *addr_len)
666{
667 saddr->ss_family = dlm_local_addr[0].ss_family;
668 if (saddr->ss_family == AF_INET) {
669 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
670 in4_addr->sin_port = cpu_to_be16(port);
671 *addr_len = sizeof(struct sockaddr_in);
672 memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
673 } else {
674 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
675 in6_addr->sin6_port = cpu_to_be16(port);
676 *addr_len = sizeof(struct sockaddr_in6);
677 }
678 memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
679}
680
681static void dlm_page_release(struct kref *kref)
682{
683 struct writequeue_entry *e = container_of(kref, struct writequeue_entry,
684 ref);
685
686 __free_page(e->page);
687 dlm_free_writequeue(e);
688}
689
690static void dlm_msg_release(struct kref *kref)
691{
692 struct dlm_msg *msg = container_of(kref, struct dlm_msg, ref);
693
694 kref_put(&msg->entry->ref, dlm_page_release);
695 dlm_free_msg(msg);
696}
697
698static void free_entry(struct writequeue_entry *e)
699{
700 struct dlm_msg *msg, *tmp;
701
702 list_for_each_entry_safe(msg, tmp, &e->msgs, list) {
703 if (msg->orig_msg) {
704 msg->orig_msg->retransmit = false;
705 kref_put(&msg->orig_msg->ref, dlm_msg_release);
706 }
707
708 list_del(&msg->list);
709 kref_put(&msg->ref, dlm_msg_release);
710 }
711
712 list_del(&e->list);
713 kref_put(&e->ref, dlm_page_release);
714}
715
716static void dlm_close_sock(struct socket **sock)
717{
718 lock_sock((*sock)->sk);
719 restore_callbacks((*sock)->sk);
720 release_sock((*sock)->sk);
721
722 sock_release(*sock);
723 *sock = NULL;
724}
725
726static void allow_connection_io(struct connection *con)
727{
728 if (con->othercon)
729 clear_bit(CF_IO_STOP, &con->othercon->flags);
730 clear_bit(CF_IO_STOP, &con->flags);
731}
732
733static void stop_connection_io(struct connection *con)
734{
735 if (con->othercon)
736 stop_connection_io(con->othercon);
737
738 spin_lock_bh(&con->writequeue_lock);
739 set_bit(CF_IO_STOP, &con->flags);
740 spin_unlock_bh(&con->writequeue_lock);
741
742 down_write(&con->sock_lock);
743 if (con->sock) {
744 lock_sock(con->sock->sk);
745 restore_callbacks(con->sock->sk);
746 release_sock(con->sock->sk);
747 }
748 up_write(&con->sock_lock);
749
750 cancel_work_sync(&con->swork);
751 cancel_work_sync(&con->rwork);
752}
753
754/* Close a remote connection and tidy up */
755static void close_connection(struct connection *con, bool and_other)
756{
757 struct writequeue_entry *e;
758
759 if (con->othercon && and_other)
760 close_connection(con->othercon, false);
761
762 down_write(&con->sock_lock);
763 if (!con->sock) {
764 up_write(&con->sock_lock);
765 return;
766 }
767
768 dlm_close_sock(&con->sock);
769
770 /* if we send a writequeue entry only a half way, we drop the
771 * whole entry because reconnection and that we not start of the
772 * middle of a msg which will confuse the other end.
773 *
774 * we can always drop messages because retransmits, but what we
775 * cannot allow is to transmit half messages which may be processed
776 * at the other side.
777 *
778 * our policy is to start on a clean state when disconnects, we don't
779 * know what's send/received on transport layer in this case.
780 */
781 spin_lock_bh(&con->writequeue_lock);
782 if (!list_empty(&con->writequeue)) {
783 e = list_first_entry(&con->writequeue, struct writequeue_entry,
784 list);
785 if (e->dirty)
786 free_entry(e);
787 }
788 spin_unlock_bh(&con->writequeue_lock);
789
790 con->rx_leftover = 0;
791 con->retries = 0;
792 clear_bit(CF_APP_LIMITED, &con->flags);
793 clear_bit(CF_RECV_PENDING, &con->flags);
794 clear_bit(CF_SEND_PENDING, &con->flags);
795 up_write(&con->sock_lock);
796}
797
798static void shutdown_connection(struct connection *con, bool and_other)
799{
800 int ret;
801
802 if (con->othercon && and_other)
803 shutdown_connection(con->othercon, false);
804
805 flush_workqueue(io_workqueue);
806 down_read(&con->sock_lock);
807 /* nothing to shutdown */
808 if (!con->sock) {
809 up_read(&con->sock_lock);
810 return;
811 }
812
813 ret = kernel_sock_shutdown(con->sock, SHUT_WR);
814 up_read(&con->sock_lock);
815 if (ret) {
816 log_print("Connection %p failed to shutdown: %d will force close",
817 con, ret);
818 goto force_close;
819 } else {
820 ret = wait_event_timeout(con->shutdown_wait, !con->sock,
821 DLM_SHUTDOWN_WAIT_TIMEOUT);
822 if (ret == 0) {
823 log_print("Connection %p shutdown timed out, will force close",
824 con);
825 goto force_close;
826 }
827 }
828
829 return;
830
831force_close:
832 close_connection(con, false);
833}
834
835static struct processqueue_entry *new_processqueue_entry(int nodeid,
836 int buflen)
837{
838 struct processqueue_entry *pentry;
839
840 pentry = kmalloc(sizeof(*pentry), GFP_NOFS);
841 if (!pentry)
842 return NULL;
843
844 pentry->buf = kmalloc(buflen, GFP_NOFS);
845 if (!pentry->buf) {
846 kfree(pentry);
847 return NULL;
848 }
849
850 pentry->nodeid = nodeid;
851 return pentry;
852}
853
854static void free_processqueue_entry(struct processqueue_entry *pentry)
855{
856 kfree(pentry->buf);
857 kfree(pentry);
858}
859
860struct dlm_processed_nodes {
861 int nodeid;
862
863 struct list_head list;
864};
865
866static void process_dlm_messages(struct work_struct *work)
867{
868 struct processqueue_entry *pentry;
869
870 spin_lock(&processqueue_lock);
871 pentry = list_first_entry_or_null(&processqueue,
872 struct processqueue_entry, list);
873 if (WARN_ON_ONCE(!pentry)) {
874 process_dlm_messages_pending = false;
875 spin_unlock(&processqueue_lock);
876 return;
877 }
878
879 list_del(&pentry->list);
880 atomic_dec(&processqueue_count);
881 spin_unlock(&processqueue_lock);
882
883 for (;;) {
884 dlm_process_incoming_buffer(pentry->nodeid, pentry->buf,
885 pentry->buflen);
886 free_processqueue_entry(pentry);
887
888 spin_lock(&processqueue_lock);
889 pentry = list_first_entry_or_null(&processqueue,
890 struct processqueue_entry, list);
891 if (!pentry) {
892 process_dlm_messages_pending = false;
893 spin_unlock(&processqueue_lock);
894 break;
895 }
896
897 list_del(&pentry->list);
898 atomic_dec(&processqueue_count);
899 spin_unlock(&processqueue_lock);
900 }
901}
902
903/* Data received from remote end */
904static int receive_from_sock(struct connection *con, int buflen)
905{
906 struct processqueue_entry *pentry;
907 int ret, buflen_real;
908 struct msghdr msg;
909 struct kvec iov;
910
911 pentry = new_processqueue_entry(con->nodeid, buflen);
912 if (!pentry)
913 return DLM_IO_RESCHED;
914
915 memcpy(pentry->buf, con->rx_leftover_buf, con->rx_leftover);
916
917 /* calculate new buffer parameter regarding last receive and
918 * possible leftover bytes
919 */
920 iov.iov_base = pentry->buf + con->rx_leftover;
921 iov.iov_len = buflen - con->rx_leftover;
922
923 memset(&msg, 0, sizeof(msg));
924 msg.msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
925 clear_bit(CF_RECV_INTR, &con->flags);
926again:
927 ret = kernel_recvmsg(con->sock, &msg, &iov, 1, iov.iov_len,
928 msg.msg_flags);
929 trace_dlm_recv(con->nodeid, ret);
930 if (ret == -EAGAIN) {
931 lock_sock(con->sock->sk);
932 if (test_and_clear_bit(CF_RECV_INTR, &con->flags)) {
933 release_sock(con->sock->sk);
934 goto again;
935 }
936
937 clear_bit(CF_RECV_PENDING, &con->flags);
938 release_sock(con->sock->sk);
939 free_processqueue_entry(pentry);
940 return DLM_IO_END;
941 } else if (ret == 0) {
942 /* close will clear CF_RECV_PENDING */
943 free_processqueue_entry(pentry);
944 return DLM_IO_EOF;
945 } else if (ret < 0) {
946 free_processqueue_entry(pentry);
947 return ret;
948 }
949
950 /* new buflen according readed bytes and leftover from last receive */
951 buflen_real = ret + con->rx_leftover;
952 ret = dlm_validate_incoming_buffer(con->nodeid, pentry->buf,
953 buflen_real);
954 if (ret < 0) {
955 free_processqueue_entry(pentry);
956 return ret;
957 }
958
959 pentry->buflen = ret;
960
961 /* calculate leftover bytes from process and put it into begin of
962 * the receive buffer, so next receive we have the full message
963 * at the start address of the receive buffer.
964 */
965 con->rx_leftover = buflen_real - ret;
966 memmove(con->rx_leftover_buf, pentry->buf + ret,
967 con->rx_leftover);
968
969 spin_lock(&processqueue_lock);
970 ret = atomic_inc_return(&processqueue_count);
971 list_add_tail(&pentry->list, &processqueue);
972 if (!process_dlm_messages_pending) {
973 process_dlm_messages_pending = true;
974 queue_work(process_workqueue, &process_work);
975 }
976 spin_unlock(&processqueue_lock);
977
978 if (ret > DLM_MAX_PROCESS_BUFFERS)
979 return DLM_IO_FLUSH;
980
981 return DLM_IO_SUCCESS;
982}
983
984/* Listening socket is busy, accept a connection */
985static int accept_from_sock(void)
986{
987 struct sockaddr_storage peeraddr;
988 int len, idx, result, nodeid;
989 struct connection *newcon;
990 struct socket *newsock;
991 unsigned int mark;
992
993 result = kernel_accept(listen_con.sock, &newsock, O_NONBLOCK);
994 if (result == -EAGAIN)
995 return DLM_IO_END;
996 else if (result < 0)
997 goto accept_err;
998
999 /* Get the connected socket's peer */
1000 memset(&peeraddr, 0, sizeof(peeraddr));
1001 len = newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 2);
1002 if (len < 0) {
1003 result = -ECONNABORTED;
1004 goto accept_err;
1005 }
1006
1007 /* Get the new node's NODEID */
1008 make_sockaddr(&peeraddr, 0, &len);
1009 if (addr_to_nodeid(&peeraddr, &nodeid, &mark)) {
1010 switch (peeraddr.ss_family) {
1011 case AF_INET: {
1012 struct sockaddr_in *sin = (struct sockaddr_in *)&peeraddr;
1013
1014 log_print("connect from non cluster IPv4 node %pI4",
1015 &sin->sin_addr);
1016 break;
1017 }
1018#if IS_ENABLED(CONFIG_IPV6)
1019 case AF_INET6: {
1020 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&peeraddr;
1021
1022 log_print("connect from non cluster IPv6 node %pI6c",
1023 &sin6->sin6_addr);
1024 break;
1025 }
1026#endif
1027 default:
1028 log_print("invalid family from non cluster node");
1029 break;
1030 }
1031
1032 sock_release(newsock);
1033 return -1;
1034 }
1035
1036 log_print("got connection from %d", nodeid);
1037
1038 /* Check to see if we already have a connection to this node. This
1039 * could happen if the two nodes initiate a connection at roughly
1040 * the same time and the connections cross on the wire.
1041 * In this case we store the incoming one in "othercon"
1042 */
1043 idx = srcu_read_lock(&connections_srcu);
1044 newcon = nodeid2con(nodeid, 0);
1045 if (WARN_ON_ONCE(!newcon)) {
1046 srcu_read_unlock(&connections_srcu, idx);
1047 result = -ENOENT;
1048 goto accept_err;
1049 }
1050
1051 sock_set_mark(newsock->sk, mark);
1052
1053 down_write(&newcon->sock_lock);
1054 if (newcon->sock) {
1055 struct connection *othercon = newcon->othercon;
1056
1057 if (!othercon) {
1058 othercon = kzalloc(sizeof(*othercon), GFP_NOFS);
1059 if (!othercon) {
1060 log_print("failed to allocate incoming socket");
1061 up_write(&newcon->sock_lock);
1062 srcu_read_unlock(&connections_srcu, idx);
1063 result = -ENOMEM;
1064 goto accept_err;
1065 }
1066
1067 dlm_con_init(othercon, nodeid);
1068 lockdep_set_subclass(&othercon->sock_lock, 1);
1069 newcon->othercon = othercon;
1070 set_bit(CF_IS_OTHERCON, &othercon->flags);
1071 } else {
1072 /* close other sock con if we have something new */
1073 close_connection(othercon, false);
1074 }
1075
1076 down_write(&othercon->sock_lock);
1077 add_sock(newsock, othercon);
1078
1079 /* check if we receved something while adding */
1080 lock_sock(othercon->sock->sk);
1081 lowcomms_queue_rwork(othercon);
1082 release_sock(othercon->sock->sk);
1083 up_write(&othercon->sock_lock);
1084 }
1085 else {
1086 /* accept copies the sk after we've saved the callbacks, so we
1087 don't want to save them a second time or comm errors will
1088 result in calling sk_error_report recursively. */
1089 add_sock(newsock, newcon);
1090
1091 /* check if we receved something while adding */
1092 lock_sock(newcon->sock->sk);
1093 lowcomms_queue_rwork(newcon);
1094 release_sock(newcon->sock->sk);
1095 }
1096 up_write(&newcon->sock_lock);
1097 srcu_read_unlock(&connections_srcu, idx);
1098
1099 return DLM_IO_SUCCESS;
1100
1101accept_err:
1102 if (newsock)
1103 sock_release(newsock);
1104
1105 return result;
1106}
1107
1108/*
1109 * writequeue_entry_complete - try to delete and free write queue entry
1110 * @e: write queue entry to try to delete
1111 * @completed: bytes completed
1112 *
1113 * writequeue_lock must be held.
1114 */
1115static void writequeue_entry_complete(struct writequeue_entry *e, int completed)
1116{
1117 e->offset += completed;
1118 e->len -= completed;
1119 /* signal that page was half way transmitted */
1120 e->dirty = true;
1121
1122 if (e->len == 0 && e->users == 0)
1123 free_entry(e);
1124}
1125
1126/*
1127 * sctp_bind_addrs - bind a SCTP socket to all our addresses
1128 */
1129static int sctp_bind_addrs(struct socket *sock, uint16_t port)
1130{
1131 struct sockaddr_storage localaddr;
1132 struct sockaddr *addr = (struct sockaddr *)&localaddr;
1133 int i, addr_len, result = 0;
1134
1135 for (i = 0; i < dlm_local_count; i++) {
1136 memcpy(&localaddr, &dlm_local_addr[i], sizeof(localaddr));
1137 make_sockaddr(&localaddr, port, &addr_len);
1138
1139 if (!i)
1140 result = kernel_bind(sock, addr, addr_len);
1141 else
1142 result = sock_bind_add(sock->sk, addr, addr_len);
1143
1144 if (result < 0) {
1145 log_print("Can't bind to %d addr number %d, %d.\n",
1146 port, i + 1, result);
1147 break;
1148 }
1149 }
1150 return result;
1151}
1152
1153/* Get local addresses */
1154static void init_local(void)
1155{
1156 struct sockaddr_storage sas;
1157 int i;
1158
1159 dlm_local_count = 0;
1160 for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
1161 if (dlm_our_addr(&sas, i))
1162 break;
1163
1164 memcpy(&dlm_local_addr[dlm_local_count++], &sas, sizeof(sas));
1165 }
1166}
1167
1168static struct writequeue_entry *new_writequeue_entry(struct connection *con)
1169{
1170 struct writequeue_entry *entry;
1171
1172 entry = dlm_allocate_writequeue();
1173 if (!entry)
1174 return NULL;
1175
1176 entry->page = alloc_page(GFP_ATOMIC | __GFP_ZERO);
1177 if (!entry->page) {
1178 dlm_free_writequeue(entry);
1179 return NULL;
1180 }
1181
1182 entry->offset = 0;
1183 entry->len = 0;
1184 entry->end = 0;
1185 entry->dirty = false;
1186 entry->con = con;
1187 entry->users = 1;
1188 kref_init(&entry->ref);
1189 return entry;
1190}
1191
1192static struct writequeue_entry *new_wq_entry(struct connection *con, int len,
1193 char **ppc, void (*cb)(void *data),
1194 void *data)
1195{
1196 struct writequeue_entry *e;
1197
1198 spin_lock_bh(&con->writequeue_lock);
1199 if (!list_empty(&con->writequeue)) {
1200 e = list_last_entry(&con->writequeue, struct writequeue_entry, list);
1201 if (DLM_WQ_REMAIN_BYTES(e) >= len) {
1202 kref_get(&e->ref);
1203
1204 *ppc = page_address(e->page) + e->end;
1205 if (cb)
1206 cb(data);
1207
1208 e->end += len;
1209 e->users++;
1210 goto out;
1211 }
1212 }
1213
1214 e = new_writequeue_entry(con);
1215 if (!e)
1216 goto out;
1217
1218 kref_get(&e->ref);
1219 *ppc = page_address(e->page);
1220 e->end += len;
1221 if (cb)
1222 cb(data);
1223
1224 list_add_tail(&e->list, &con->writequeue);
1225
1226out:
1227 spin_unlock_bh(&con->writequeue_lock);
1228 return e;
1229};
1230
1231static struct dlm_msg *dlm_lowcomms_new_msg_con(struct connection *con, int len,
1232 gfp_t allocation, char **ppc,
1233 void (*cb)(void *data),
1234 void *data)
1235{
1236 struct writequeue_entry *e;
1237 struct dlm_msg *msg;
1238
1239 msg = dlm_allocate_msg(allocation);
1240 if (!msg)
1241 return NULL;
1242
1243 kref_init(&msg->ref);
1244
1245 e = new_wq_entry(con, len, ppc, cb, data);
1246 if (!e) {
1247 dlm_free_msg(msg);
1248 return NULL;
1249 }
1250
1251 msg->retransmit = false;
1252 msg->orig_msg = NULL;
1253 msg->ppc = *ppc;
1254 msg->len = len;
1255 msg->entry = e;
1256
1257 return msg;
1258}
1259
1260/* avoid false positive for nodes_srcu, unlock happens in
1261 * dlm_lowcomms_commit_msg which is a must call if success
1262 */
1263#ifndef __CHECKER__
1264struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation,
1265 char **ppc, void (*cb)(void *data),
1266 void *data)
1267{
1268 struct connection *con;
1269 struct dlm_msg *msg;
1270 int idx;
1271
1272 if (len > DLM_MAX_SOCKET_BUFSIZE ||
1273 len < sizeof(struct dlm_header)) {
1274 BUILD_BUG_ON(PAGE_SIZE < DLM_MAX_SOCKET_BUFSIZE);
1275 log_print("failed to allocate a buffer of size %d", len);
1276 WARN_ON_ONCE(1);
1277 return NULL;
1278 }
1279
1280 idx = srcu_read_lock(&connections_srcu);
1281 con = nodeid2con(nodeid, 0);
1282 if (WARN_ON_ONCE(!con)) {
1283 srcu_read_unlock(&connections_srcu, idx);
1284 return NULL;
1285 }
1286
1287 msg = dlm_lowcomms_new_msg_con(con, len, allocation, ppc, cb, data);
1288 if (!msg) {
1289 srcu_read_unlock(&connections_srcu, idx);
1290 return NULL;
1291 }
1292
1293 /* for dlm_lowcomms_commit_msg() */
1294 kref_get(&msg->ref);
1295 /* we assume if successful commit must called */
1296 msg->idx = idx;
1297 return msg;
1298}
1299#endif
1300
1301static void _dlm_lowcomms_commit_msg(struct dlm_msg *msg)
1302{
1303 struct writequeue_entry *e = msg->entry;
1304 struct connection *con = e->con;
1305 int users;
1306
1307 spin_lock_bh(&con->writequeue_lock);
1308 kref_get(&msg->ref);
1309 list_add(&msg->list, &e->msgs);
1310
1311 users = --e->users;
1312 if (users)
1313 goto out;
1314
1315 e->len = DLM_WQ_LENGTH_BYTES(e);
1316
1317 lowcomms_queue_swork(con);
1318
1319out:
1320 spin_unlock_bh(&con->writequeue_lock);
1321 return;
1322}
1323
1324/* avoid false positive for nodes_srcu, lock was happen in
1325 * dlm_lowcomms_new_msg
1326 */
1327#ifndef __CHECKER__
1328void dlm_lowcomms_commit_msg(struct dlm_msg *msg)
1329{
1330 _dlm_lowcomms_commit_msg(msg);
1331 srcu_read_unlock(&connections_srcu, msg->idx);
1332 /* because dlm_lowcomms_new_msg() */
1333 kref_put(&msg->ref, dlm_msg_release);
1334}
1335#endif
1336
1337void dlm_lowcomms_put_msg(struct dlm_msg *msg)
1338{
1339 kref_put(&msg->ref, dlm_msg_release);
1340}
1341
1342/* does not held connections_srcu, usage lowcomms_error_report only */
1343int dlm_lowcomms_resend_msg(struct dlm_msg *msg)
1344{
1345 struct dlm_msg *msg_resend;
1346 char *ppc;
1347
1348 if (msg->retransmit)
1349 return 1;
1350
1351 msg_resend = dlm_lowcomms_new_msg_con(msg->entry->con, msg->len,
1352 GFP_ATOMIC, &ppc, NULL, NULL);
1353 if (!msg_resend)
1354 return -ENOMEM;
1355
1356 msg->retransmit = true;
1357 kref_get(&msg->ref);
1358 msg_resend->orig_msg = msg;
1359
1360 memcpy(ppc, msg->ppc, msg->len);
1361 _dlm_lowcomms_commit_msg(msg_resend);
1362 dlm_lowcomms_put_msg(msg_resend);
1363
1364 return 0;
1365}
1366
1367/* Send a message */
1368static int send_to_sock(struct connection *con)
1369{
1370 struct writequeue_entry *e;
1371 struct bio_vec bvec;
1372 struct msghdr msg = {
1373 .msg_flags = MSG_SPLICE_PAGES | MSG_DONTWAIT | MSG_NOSIGNAL,
1374 };
1375 int len, offset, ret;
1376
1377 spin_lock_bh(&con->writequeue_lock);
1378 e = con_next_wq(con);
1379 if (!e) {
1380 clear_bit(CF_SEND_PENDING, &con->flags);
1381 spin_unlock_bh(&con->writequeue_lock);
1382 return DLM_IO_END;
1383 }
1384
1385 len = e->len;
1386 offset = e->offset;
1387 WARN_ON_ONCE(len == 0 && e->users == 0);
1388 spin_unlock_bh(&con->writequeue_lock);
1389
1390 bvec_set_page(&bvec, e->page, len, offset);
1391 iov_iter_bvec(&msg.msg_iter, ITER_SOURCE, &bvec, 1, len);
1392 ret = sock_sendmsg(con->sock, &msg);
1393 trace_dlm_send(con->nodeid, ret);
1394 if (ret == -EAGAIN || ret == 0) {
1395 lock_sock(con->sock->sk);
1396 spin_lock_bh(&con->writequeue_lock);
1397 if (test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) &&
1398 !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1399 /* Notify TCP that we're limited by the
1400 * application window size.
1401 */
1402 set_bit(SOCK_NOSPACE, &con->sock->sk->sk_socket->flags);
1403 con->sock->sk->sk_write_pending++;
1404
1405 clear_bit(CF_SEND_PENDING, &con->flags);
1406 spin_unlock_bh(&con->writequeue_lock);
1407 release_sock(con->sock->sk);
1408
1409 /* wait for write_space() event */
1410 return DLM_IO_END;
1411 }
1412 spin_unlock_bh(&con->writequeue_lock);
1413 release_sock(con->sock->sk);
1414
1415 return DLM_IO_RESCHED;
1416 } else if (ret < 0) {
1417 return ret;
1418 }
1419
1420 spin_lock_bh(&con->writequeue_lock);
1421 writequeue_entry_complete(e, ret);
1422 spin_unlock_bh(&con->writequeue_lock);
1423
1424 return DLM_IO_SUCCESS;
1425}
1426
1427static void clean_one_writequeue(struct connection *con)
1428{
1429 struct writequeue_entry *e, *safe;
1430
1431 spin_lock_bh(&con->writequeue_lock);
1432 list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1433 free_entry(e);
1434 }
1435 spin_unlock_bh(&con->writequeue_lock);
1436}
1437
1438static void connection_release(struct rcu_head *rcu)
1439{
1440 struct connection *con = container_of(rcu, struct connection, rcu);
1441
1442 WARN_ON_ONCE(!list_empty(&con->writequeue));
1443 WARN_ON_ONCE(con->sock);
1444 kfree(con);
1445}
1446
1447/* Called from recovery when it knows that a node has
1448 left the cluster */
1449int dlm_lowcomms_close(int nodeid)
1450{
1451 struct connection *con;
1452 int idx;
1453
1454 log_print("closing connection to node %d", nodeid);
1455
1456 idx = srcu_read_lock(&connections_srcu);
1457 con = nodeid2con(nodeid, 0);
1458 if (WARN_ON_ONCE(!con)) {
1459 srcu_read_unlock(&connections_srcu, idx);
1460 return -ENOENT;
1461 }
1462
1463 stop_connection_io(con);
1464 log_print("io handling for node: %d stopped", nodeid);
1465 close_connection(con, true);
1466
1467 spin_lock(&connections_lock);
1468 hlist_del_rcu(&con->list);
1469 spin_unlock(&connections_lock);
1470
1471 clean_one_writequeue(con);
1472 call_srcu(&connections_srcu, &con->rcu, connection_release);
1473 if (con->othercon) {
1474 clean_one_writequeue(con->othercon);
1475 call_srcu(&connections_srcu, &con->othercon->rcu, connection_release);
1476 }
1477 srcu_read_unlock(&connections_srcu, idx);
1478
1479 /* for debugging we print when we are done to compare with other
1480 * messages in between. This function need to be correctly synchronized
1481 * with io handling
1482 */
1483 log_print("closing connection to node %d done", nodeid);
1484
1485 return 0;
1486}
1487
1488/* Receive worker function */
1489static void process_recv_sockets(struct work_struct *work)
1490{
1491 struct connection *con = container_of(work, struct connection, rwork);
1492 int ret, buflen;
1493
1494 down_read(&con->sock_lock);
1495 if (!con->sock) {
1496 up_read(&con->sock_lock);
1497 return;
1498 }
1499
1500 buflen = READ_ONCE(dlm_config.ci_buffer_size);
1501 do {
1502 ret = receive_from_sock(con, buflen);
1503 } while (ret == DLM_IO_SUCCESS);
1504 up_read(&con->sock_lock);
1505
1506 switch (ret) {
1507 case DLM_IO_END:
1508 /* CF_RECV_PENDING cleared */
1509 break;
1510 case DLM_IO_EOF:
1511 close_connection(con, false);
1512 wake_up(&con->shutdown_wait);
1513 /* CF_RECV_PENDING cleared */
1514 break;
1515 case DLM_IO_FLUSH:
1516 flush_workqueue(process_workqueue);
1517 fallthrough;
1518 case DLM_IO_RESCHED:
1519 cond_resched();
1520 queue_work(io_workqueue, &con->rwork);
1521 /* CF_RECV_PENDING not cleared */
1522 break;
1523 default:
1524 if (ret < 0) {
1525 if (test_bit(CF_IS_OTHERCON, &con->flags)) {
1526 close_connection(con, false);
1527 } else {
1528 spin_lock_bh(&con->writequeue_lock);
1529 lowcomms_queue_swork(con);
1530 spin_unlock_bh(&con->writequeue_lock);
1531 }
1532
1533 /* CF_RECV_PENDING cleared for othercon
1534 * we trigger send queue if not already done
1535 * and process_send_sockets will handle it
1536 */
1537 break;
1538 }
1539
1540 WARN_ON_ONCE(1);
1541 break;
1542 }
1543}
1544
1545static void process_listen_recv_socket(struct work_struct *work)
1546{
1547 int ret;
1548
1549 if (WARN_ON_ONCE(!listen_con.sock))
1550 return;
1551
1552 do {
1553 ret = accept_from_sock();
1554 } while (ret == DLM_IO_SUCCESS);
1555
1556 if (ret < 0)
1557 log_print("critical error accepting connection: %d", ret);
1558}
1559
1560static int dlm_connect(struct connection *con)
1561{
1562 struct sockaddr_storage addr;
1563 int result, addr_len;
1564 struct socket *sock;
1565 unsigned int mark;
1566
1567 memset(&addr, 0, sizeof(addr));
1568 result = nodeid_to_addr(con->nodeid, &addr, NULL,
1569 dlm_proto_ops->try_new_addr, &mark);
1570 if (result < 0) {
1571 log_print("no address for nodeid %d", con->nodeid);
1572 return result;
1573 }
1574
1575 /* Create a socket to communicate with */
1576 result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family,
1577 SOCK_STREAM, dlm_proto_ops->proto, &sock);
1578 if (result < 0)
1579 return result;
1580
1581 sock_set_mark(sock->sk, mark);
1582 dlm_proto_ops->sockopts(sock);
1583
1584 result = dlm_proto_ops->bind(sock);
1585 if (result < 0) {
1586 sock_release(sock);
1587 return result;
1588 }
1589
1590 add_sock(sock, con);
1591
1592 log_print_ratelimited("connecting to %d", con->nodeid);
1593 make_sockaddr(&addr, dlm_config.ci_tcp_port, &addr_len);
1594 result = dlm_proto_ops->connect(con, sock, (struct sockaddr *)&addr,
1595 addr_len);
1596 switch (result) {
1597 case -EINPROGRESS:
1598 /* not an error */
1599 fallthrough;
1600 case 0:
1601 break;
1602 default:
1603 if (result < 0)
1604 dlm_close_sock(&con->sock);
1605
1606 break;
1607 }
1608
1609 return result;
1610}
1611
1612/* Send worker function */
1613static void process_send_sockets(struct work_struct *work)
1614{
1615 struct connection *con = container_of(work, struct connection, swork);
1616 int ret;
1617
1618 WARN_ON_ONCE(test_bit(CF_IS_OTHERCON, &con->flags));
1619
1620 down_read(&con->sock_lock);
1621 if (!con->sock) {
1622 up_read(&con->sock_lock);
1623 down_write(&con->sock_lock);
1624 if (!con->sock) {
1625 ret = dlm_connect(con);
1626 switch (ret) {
1627 case 0:
1628 break;
1629 case -EINPROGRESS:
1630 /* avoid spamming resched on connection
1631 * we might can switch to a state_change
1632 * event based mechanism if established
1633 */
1634 msleep(100);
1635 break;
1636 default:
1637 /* CF_SEND_PENDING not cleared */
1638 up_write(&con->sock_lock);
1639 log_print("connect to node %d try %d error %d",
1640 con->nodeid, con->retries++, ret);
1641 msleep(1000);
1642 /* For now we try forever to reconnect. In
1643 * future we should send a event to cluster
1644 * manager to fence itself after certain amount
1645 * of retries.
1646 */
1647 queue_work(io_workqueue, &con->swork);
1648 return;
1649 }
1650 }
1651 downgrade_write(&con->sock_lock);
1652 }
1653
1654 do {
1655 ret = send_to_sock(con);
1656 } while (ret == DLM_IO_SUCCESS);
1657 up_read(&con->sock_lock);
1658
1659 switch (ret) {
1660 case DLM_IO_END:
1661 /* CF_SEND_PENDING cleared */
1662 break;
1663 case DLM_IO_RESCHED:
1664 /* CF_SEND_PENDING not cleared */
1665 cond_resched();
1666 queue_work(io_workqueue, &con->swork);
1667 break;
1668 default:
1669 if (ret < 0) {
1670 close_connection(con, false);
1671
1672 /* CF_SEND_PENDING cleared */
1673 spin_lock_bh(&con->writequeue_lock);
1674 lowcomms_queue_swork(con);
1675 spin_unlock_bh(&con->writequeue_lock);
1676 break;
1677 }
1678
1679 WARN_ON_ONCE(1);
1680 break;
1681 }
1682}
1683
1684static void work_stop(void)
1685{
1686 if (io_workqueue) {
1687 destroy_workqueue(io_workqueue);
1688 io_workqueue = NULL;
1689 }
1690
1691 if (process_workqueue) {
1692 destroy_workqueue(process_workqueue);
1693 process_workqueue = NULL;
1694 }
1695}
1696
1697static int work_start(void)
1698{
1699 io_workqueue = alloc_workqueue("dlm_io", WQ_HIGHPRI | WQ_MEM_RECLAIM |
1700 WQ_UNBOUND, 0);
1701 if (!io_workqueue) {
1702 log_print("can't start dlm_io");
1703 return -ENOMEM;
1704 }
1705
1706 /* ordered dlm message process queue,
1707 * should be converted to a tasklet
1708 */
1709 process_workqueue = alloc_ordered_workqueue("dlm_process",
1710 WQ_HIGHPRI | WQ_MEM_RECLAIM);
1711 if (!process_workqueue) {
1712 log_print("can't start dlm_process");
1713 destroy_workqueue(io_workqueue);
1714 io_workqueue = NULL;
1715 return -ENOMEM;
1716 }
1717
1718 return 0;
1719}
1720
1721void dlm_lowcomms_shutdown(void)
1722{
1723 struct connection *con;
1724 int i, idx;
1725
1726 /* stop lowcomms_listen_data_ready calls */
1727 lock_sock(listen_con.sock->sk);
1728 listen_con.sock->sk->sk_data_ready = listen_sock.sk_data_ready;
1729 release_sock(listen_con.sock->sk);
1730
1731 cancel_work_sync(&listen_con.rwork);
1732 dlm_close_sock(&listen_con.sock);
1733
1734 idx = srcu_read_lock(&connections_srcu);
1735 for (i = 0; i < CONN_HASH_SIZE; i++) {
1736 hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
1737 shutdown_connection(con, true);
1738 stop_connection_io(con);
1739 flush_workqueue(process_workqueue);
1740 close_connection(con, true);
1741
1742 clean_one_writequeue(con);
1743 if (con->othercon)
1744 clean_one_writequeue(con->othercon);
1745 allow_connection_io(con);
1746 }
1747 }
1748 srcu_read_unlock(&connections_srcu, idx);
1749}
1750
1751void dlm_lowcomms_stop(void)
1752{
1753 work_stop();
1754 dlm_proto_ops = NULL;
1755}
1756
1757static int dlm_listen_for_all(void)
1758{
1759 struct socket *sock;
1760 int result;
1761
1762 log_print("Using %s for communications",
1763 dlm_proto_ops->name);
1764
1765 result = dlm_proto_ops->listen_validate();
1766 if (result < 0)
1767 return result;
1768
1769 result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family,
1770 SOCK_STREAM, dlm_proto_ops->proto, &sock);
1771 if (result < 0) {
1772 log_print("Can't create comms socket: %d", result);
1773 return result;
1774 }
1775
1776 sock_set_mark(sock->sk, dlm_config.ci_mark);
1777 dlm_proto_ops->listen_sockopts(sock);
1778
1779 result = dlm_proto_ops->listen_bind(sock);
1780 if (result < 0)
1781 goto out;
1782
1783 lock_sock(sock->sk);
1784 listen_sock.sk_data_ready = sock->sk->sk_data_ready;
1785 listen_sock.sk_write_space = sock->sk->sk_write_space;
1786 listen_sock.sk_error_report = sock->sk->sk_error_report;
1787 listen_sock.sk_state_change = sock->sk->sk_state_change;
1788
1789 listen_con.sock = sock;
1790
1791 sock->sk->sk_allocation = GFP_NOFS;
1792 sock->sk->sk_use_task_frag = false;
1793 sock->sk->sk_data_ready = lowcomms_listen_data_ready;
1794 release_sock(sock->sk);
1795
1796 result = sock->ops->listen(sock, 128);
1797 if (result < 0) {
1798 dlm_close_sock(&listen_con.sock);
1799 return result;
1800 }
1801
1802 return 0;
1803
1804out:
1805 sock_release(sock);
1806 return result;
1807}
1808
1809static int dlm_tcp_bind(struct socket *sock)
1810{
1811 struct sockaddr_storage src_addr;
1812 int result, addr_len;
1813
1814 /* Bind to our cluster-known address connecting to avoid
1815 * routing problems.
1816 */
1817 memcpy(&src_addr, &dlm_local_addr[0], sizeof(src_addr));
1818 make_sockaddr(&src_addr, 0, &addr_len);
1819
1820 result = kernel_bind(sock, (struct sockaddr *)&src_addr,
1821 addr_len);
1822 if (result < 0) {
1823 /* This *may* not indicate a critical error */
1824 log_print("could not bind for connect: %d", result);
1825 }
1826
1827 return 0;
1828}
1829
1830static int dlm_tcp_connect(struct connection *con, struct socket *sock,
1831 struct sockaddr *addr, int addr_len)
1832{
1833 return kernel_connect(sock, addr, addr_len, O_NONBLOCK);
1834}
1835
1836static int dlm_tcp_listen_validate(void)
1837{
1838 /* We don't support multi-homed hosts */
1839 if (dlm_local_count > 1) {
1840 log_print("TCP protocol can't handle multi-homed hosts, try SCTP");
1841 return -EINVAL;
1842 }
1843
1844 return 0;
1845}
1846
1847static void dlm_tcp_sockopts(struct socket *sock)
1848{
1849 /* Turn off Nagle's algorithm */
1850 tcp_sock_set_nodelay(sock->sk);
1851}
1852
1853static void dlm_tcp_listen_sockopts(struct socket *sock)
1854{
1855 dlm_tcp_sockopts(sock);
1856 sock_set_reuseaddr(sock->sk);
1857}
1858
1859static int dlm_tcp_listen_bind(struct socket *sock)
1860{
1861 int addr_len;
1862
1863 /* Bind to our port */
1864 make_sockaddr(&dlm_local_addr[0], dlm_config.ci_tcp_port, &addr_len);
1865 return kernel_bind(sock, (struct sockaddr *)&dlm_local_addr[0],
1866 addr_len);
1867}
1868
1869static const struct dlm_proto_ops dlm_tcp_ops = {
1870 .name = "TCP",
1871 .proto = IPPROTO_TCP,
1872 .connect = dlm_tcp_connect,
1873 .sockopts = dlm_tcp_sockopts,
1874 .bind = dlm_tcp_bind,
1875 .listen_validate = dlm_tcp_listen_validate,
1876 .listen_sockopts = dlm_tcp_listen_sockopts,
1877 .listen_bind = dlm_tcp_listen_bind,
1878};
1879
1880static int dlm_sctp_bind(struct socket *sock)
1881{
1882 return sctp_bind_addrs(sock, 0);
1883}
1884
1885static int dlm_sctp_connect(struct connection *con, struct socket *sock,
1886 struct sockaddr *addr, int addr_len)
1887{
1888 int ret;
1889
1890 /*
1891 * Make kernel_connect() function return in specified time,
1892 * since O_NONBLOCK argument in connect() function does not work here,
1893 * then, we should restore the default value of this attribute.
1894 */
1895 sock_set_sndtimeo(sock->sk, 5);
1896 ret = kernel_connect(sock, addr, addr_len, 0);
1897 sock_set_sndtimeo(sock->sk, 0);
1898 return ret;
1899}
1900
1901static int dlm_sctp_listen_validate(void)
1902{
1903 if (!IS_ENABLED(CONFIG_IP_SCTP)) {
1904 log_print("SCTP is not enabled by this kernel");
1905 return -EOPNOTSUPP;
1906 }
1907
1908 request_module("sctp");
1909 return 0;
1910}
1911
1912static int dlm_sctp_bind_listen(struct socket *sock)
1913{
1914 return sctp_bind_addrs(sock, dlm_config.ci_tcp_port);
1915}
1916
1917static void dlm_sctp_sockopts(struct socket *sock)
1918{
1919 /* Turn off Nagle's algorithm */
1920 sctp_sock_set_nodelay(sock->sk);
1921 sock_set_rcvbuf(sock->sk, NEEDED_RMEM);
1922}
1923
1924static const struct dlm_proto_ops dlm_sctp_ops = {
1925 .name = "SCTP",
1926 .proto = IPPROTO_SCTP,
1927 .try_new_addr = true,
1928 .connect = dlm_sctp_connect,
1929 .sockopts = dlm_sctp_sockopts,
1930 .bind = dlm_sctp_bind,
1931 .listen_validate = dlm_sctp_listen_validate,
1932 .listen_sockopts = dlm_sctp_sockopts,
1933 .listen_bind = dlm_sctp_bind_listen,
1934};
1935
1936int dlm_lowcomms_start(void)
1937{
1938 int error;
1939
1940 init_local();
1941 if (!dlm_local_count) {
1942 error = -ENOTCONN;
1943 log_print("no local IP address has been set");
1944 goto fail;
1945 }
1946
1947 error = work_start();
1948 if (error)
1949 goto fail;
1950
1951 /* Start listening */
1952 switch (dlm_config.ci_protocol) {
1953 case DLM_PROTO_TCP:
1954 dlm_proto_ops = &dlm_tcp_ops;
1955 break;
1956 case DLM_PROTO_SCTP:
1957 dlm_proto_ops = &dlm_sctp_ops;
1958 break;
1959 default:
1960 log_print("Invalid protocol identifier %d set",
1961 dlm_config.ci_protocol);
1962 error = -EINVAL;
1963 goto fail_proto_ops;
1964 }
1965
1966 error = dlm_listen_for_all();
1967 if (error)
1968 goto fail_listen;
1969
1970 return 0;
1971
1972fail_listen:
1973 dlm_proto_ops = NULL;
1974fail_proto_ops:
1975 work_stop();
1976fail:
1977 return error;
1978}
1979
1980void dlm_lowcomms_init(void)
1981{
1982 int i;
1983
1984 for (i = 0; i < CONN_HASH_SIZE; i++)
1985 INIT_HLIST_HEAD(&connection_hash[i]);
1986
1987 INIT_WORK(&listen_con.rwork, process_listen_recv_socket);
1988}
1989
1990void dlm_lowcomms_exit(void)
1991{
1992 struct connection *con;
1993 int i, idx;
1994
1995 idx = srcu_read_lock(&connections_srcu);
1996 for (i = 0; i < CONN_HASH_SIZE; i++) {
1997 hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
1998 spin_lock(&connections_lock);
1999 hlist_del_rcu(&con->list);
2000 spin_unlock(&connections_lock);
2001
2002 if (con->othercon)
2003 call_srcu(&connections_srcu, &con->othercon->rcu,
2004 connection_release);
2005 call_srcu(&connections_srcu, &con->rcu, connection_release);
2006 }
2007 }
2008 srcu_read_unlock(&connections_srcu, idx);
2009}