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
69 struct 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
117 struct 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 */
126 struct 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
139 struct 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
151 struct processqueue_entry {
152 unsigned char *buf;
153 int nodeid;
154 int buflen;
155
156 struct list_head list;
157 };
158
159 struct 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
173 static 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
180 static struct listen_connection listen_con;
181 static struct sockaddr_storage dlm_local_addr[DLM_MAX_ADDR_COUNT];
182 static int dlm_local_count;
183
184 /* Work queues */
185 static struct workqueue_struct *io_workqueue;
186 static struct workqueue_struct *process_workqueue;
187
188 static struct hlist_head connection_hash[CONN_HASH_SIZE];
189 static DEFINE_SPINLOCK(connections_lock);
190 DEFINE_STATIC_SRCU(connections_srcu);
191
192 static 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
200 static void process_recv_sockets(struct work_struct *work);
201 static void process_send_sockets(struct work_struct *work);
202 static void process_dlm_messages(struct work_struct *work);
203
204 static DECLARE_WORK(process_work, process_dlm_messages);
205 static DEFINE_SPINLOCK(processqueue_lock);
206 static bool process_dlm_messages_pending;
207 static DECLARE_WAIT_QUEUE_HEAD(processqueue_wq);
208 static atomic_t processqueue_count;
209 static LIST_HEAD(processqueue);
210
211 bool dlm_lowcomms_is_running(void)
212 {
213 return !!listen_con.sock;
214 }
215
216 static void lowcomms_queue_swork(struct connection *con)
217 {
218 assert_spin_locked(&con->writequeue_lock);
219
220 if (!test_bit(CF_IO_STOP, &con->flags) &&
221 !test_bit(CF_APP_LIMITED, &con->flags) &&
222 !test_and_set_bit(CF_SEND_PENDING, &con->flags))
223 queue_work(io_workqueue, &con->swork);
224 }
225
226 static void lowcomms_queue_rwork(struct connection *con)
227 {
228 #ifdef CONFIG_LOCKDEP
229 WARN_ON_ONCE(!lockdep_sock_is_held(con->sock->sk));
230 #endif
231
232 if (!test_bit(CF_IO_STOP, &con->flags) &&
233 !test_and_set_bit(CF_RECV_PENDING, &con->flags))
234 queue_work(io_workqueue, &con->rwork);
235 }
236
237 static void writequeue_entry_ctor(void *data)
238 {
239 struct writequeue_entry *entry = data;
240
241 INIT_LIST_HEAD(&entry->msgs);
242 }
243
244 struct kmem_cache *dlm_lowcomms_writequeue_cache_create(void)
245 {
246 return kmem_cache_create("dlm_writequeue", sizeof(struct writequeue_entry),
247 0, 0, writequeue_entry_ctor);
248 }
249
250 struct kmem_cache *dlm_lowcomms_msg_cache_create(void)
251 {
252 return KMEM_CACHE(dlm_msg, 0);
253 }
254
255 /* need to held writequeue_lock */
256 static struct writequeue_entry *con_next_wq(struct connection *con)
257 {
258 struct writequeue_entry *e;
259
260 e = list_first_entry_or_null(&con->writequeue, struct writequeue_entry,
261 list);
262 /* if len is zero nothing is to send, if there are users filling
263 * buffers we wait until the users are done so we can send more.
264 */
265 if (!e || e->users || e->len == 0)
266 return NULL;
267
268 return e;
269 }
270
271 static struct connection *__find_con(int nodeid, int r)
272 {
273 struct connection *con;
274
275 hlist_for_each_entry_rcu(con, &connection_hash[r], list) {
276 if (con->nodeid == nodeid)
277 return con;
278 }
279
280 return NULL;
281 }
282
283 static void dlm_con_init(struct connection *con, int nodeid)
284 {
285 con->nodeid = nodeid;
286 init_rwsem(&con->sock_lock);
287 INIT_LIST_HEAD(&con->writequeue);
288 spin_lock_init(&con->writequeue_lock);
289 INIT_WORK(&con->swork, process_send_sockets);
290 INIT_WORK(&con->rwork, process_recv_sockets);
291 spin_lock_init(&con->addrs_lock);
292 init_waitqueue_head(&con->shutdown_wait);
293 }
294
295 /*
296 * If 'allocation' is zero then we don't attempt to create a new
297 * connection structure for this node.
298 */
299 static struct connection *nodeid2con(int nodeid, gfp_t alloc)
300 {
301 struct connection *con, *tmp;
302 int r;
303
304 r = nodeid_hash(nodeid);
305 con = __find_con(nodeid, r);
306 if (con || !alloc)
307 return con;
308
309 con = kzalloc(sizeof(*con), alloc);
310 if (!con)
311 return NULL;
312
313 dlm_con_init(con, nodeid);
314
315 spin_lock(&connections_lock);
316 /* Because multiple workqueues/threads calls this function it can
317 * race on multiple cpu's. Instead of locking hot path __find_con()
318 * we just check in rare cases of recently added nodes again
319 * under protection of connections_lock. If this is the case we
320 * abort our connection creation and return the existing connection.
321 */
322 tmp = __find_con(nodeid, r);
323 if (tmp) {
324 spin_unlock(&connections_lock);
325 kfree(con);
326 return tmp;
327 }
328
329 hlist_add_head_rcu(&con->list, &connection_hash[r]);
330 spin_unlock(&connections_lock);
331
332 return con;
333 }
334
335 static int addr_compare(const struct sockaddr_storage *x,
336 const struct sockaddr_storage *y)
337 {
338 switch (x->ss_family) {
339 case AF_INET: {
340 struct sockaddr_in *sinx = (struct sockaddr_in *)x;
341 struct sockaddr_in *siny = (struct sockaddr_in *)y;
342 if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr)
343 return 0;
344 if (sinx->sin_port != siny->sin_port)
345 return 0;
346 break;
347 }
348 case AF_INET6: {
349 struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x;
350 struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y;
351 if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr))
352 return 0;
353 if (sinx->sin6_port != siny->sin6_port)
354 return 0;
355 break;
356 }
357 default:
358 return 0;
359 }
360 return 1;
361 }
362
363 static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out,
364 struct sockaddr *sa_out, bool try_new_addr,
365 unsigned int *mark)
366 {
367 struct sockaddr_storage sas;
368 struct connection *con;
369 int idx;
370
371 if (!dlm_local_count)
372 return -1;
373
374 idx = srcu_read_lock(&connections_srcu);
375 con = nodeid2con(nodeid, 0);
376 if (!con) {
377 srcu_read_unlock(&connections_srcu, idx);
378 return -ENOENT;
379 }
380
381 spin_lock(&con->addrs_lock);
382 if (!con->addr_count) {
383 spin_unlock(&con->addrs_lock);
384 srcu_read_unlock(&connections_srcu, idx);
385 return -ENOENT;
386 }
387
388 memcpy(&sas, &con->addr[con->curr_addr_index],
389 sizeof(struct sockaddr_storage));
390
391 if (try_new_addr) {
392 con->curr_addr_index++;
393 if (con->curr_addr_index == con->addr_count)
394 con->curr_addr_index = 0;
395 }
396
397 *mark = con->mark;
398 spin_unlock(&con->addrs_lock);
399
400 if (sas_out)
401 memcpy(sas_out, &sas, sizeof(struct sockaddr_storage));
402
403 if (!sa_out) {
404 srcu_read_unlock(&connections_srcu, idx);
405 return 0;
406 }
407
408 if (dlm_local_addr[0].ss_family == AF_INET) {
409 struct sockaddr_in *in4 = (struct sockaddr_in *) &sas;
410 struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out;
411 ret4->sin_addr.s_addr = in4->sin_addr.s_addr;
412 } else {
413 struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &sas;
414 struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out;
415 ret6->sin6_addr = in6->sin6_addr;
416 }
417
418 srcu_read_unlock(&connections_srcu, idx);
419 return 0;
420 }
421
422 static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid,
423 unsigned int *mark)
424 {
425 struct connection *con;
426 int i, idx, addr_i;
427
428 idx = srcu_read_lock(&connections_srcu);
429 for (i = 0; i < CONN_HASH_SIZE; i++) {
430 hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
431 WARN_ON_ONCE(!con->addr_count);
432
433 spin_lock(&con->addrs_lock);
434 for (addr_i = 0; addr_i < con->addr_count; addr_i++) {
435 if (addr_compare(&con->addr[addr_i], addr)) {
436 *nodeid = con->nodeid;
437 *mark = con->mark;
438 spin_unlock(&con->addrs_lock);
439 srcu_read_unlock(&connections_srcu, idx);
440 return 0;
441 }
442 }
443 spin_unlock(&con->addrs_lock);
444 }
445 }
446 srcu_read_unlock(&connections_srcu, idx);
447
448 return -ENOENT;
449 }
450
451 static bool dlm_lowcomms_con_has_addr(const struct connection *con,
452 const struct sockaddr_storage *addr)
453 {
454 int i;
455
456 for (i = 0; i < con->addr_count; i++) {
457 if (addr_compare(&con->addr[i], addr))
458 return true;
459 }
460
461 return false;
462 }
463
464 int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr)
465 {
466 struct connection *con;
467 bool ret, idx;
468
469 idx = srcu_read_lock(&connections_srcu);
470 con = nodeid2con(nodeid, GFP_NOFS);
471 if (!con) {
472 srcu_read_unlock(&connections_srcu, idx);
473 return -ENOMEM;
474 }
475
476 spin_lock(&con->addrs_lock);
477 if (!con->addr_count) {
478 memcpy(&con->addr[0], addr, sizeof(*addr));
479 con->addr_count = 1;
480 con->mark = dlm_config.ci_mark;
481 spin_unlock(&con->addrs_lock);
482 srcu_read_unlock(&connections_srcu, idx);
483 return 0;
484 }
485
486 ret = dlm_lowcomms_con_has_addr(con, addr);
487 if (ret) {
488 spin_unlock(&con->addrs_lock);
489 srcu_read_unlock(&connections_srcu, idx);
490 return -EEXIST;
491 }
492
493 if (con->addr_count >= DLM_MAX_ADDR_COUNT) {
494 spin_unlock(&con->addrs_lock);
495 srcu_read_unlock(&connections_srcu, idx);
496 return -ENOSPC;
497 }
498
499 memcpy(&con->addr[con->addr_count++], addr, sizeof(*addr));
500 srcu_read_unlock(&connections_srcu, idx);
501 spin_unlock(&con->addrs_lock);
502 return 0;
503 }
504
505 /* Data available on socket or listen socket received a connect */
506 static void lowcomms_data_ready(struct sock *sk)
507 {
508 struct connection *con = sock2con(sk);
509
510 trace_sk_data_ready(sk);
511
512 set_bit(CF_RECV_INTR, &con->flags);
513 lowcomms_queue_rwork(con);
514 }
515
516 static void lowcomms_write_space(struct sock *sk)
517 {
518 struct connection *con = sock2con(sk);
519
520 clear_bit(SOCK_NOSPACE, &con->sock->flags);
521
522 spin_lock_bh(&con->writequeue_lock);
523 if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
524 con->sock->sk->sk_write_pending--;
525 clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags);
526 }
527
528 lowcomms_queue_swork(con);
529 spin_unlock_bh(&con->writequeue_lock);
530 }
531
532 static void lowcomms_state_change(struct sock *sk)
533 {
534 /* SCTP layer is not calling sk_data_ready when the connection
535 * is done, so we catch the signal through here.
536 */
537 if (sk->sk_shutdown == RCV_SHUTDOWN)
538 lowcomms_data_ready(sk);
539 }
540
541 static void lowcomms_listen_data_ready(struct sock *sk)
542 {
543 trace_sk_data_ready(sk);
544
545 queue_work(io_workqueue, &listen_con.rwork);
546 }
547
548 int dlm_lowcomms_connect_node(int nodeid)
549 {
550 struct connection *con;
551 int idx;
552
553 idx = srcu_read_lock(&connections_srcu);
554 con = nodeid2con(nodeid, 0);
555 if (WARN_ON_ONCE(!con)) {
556 srcu_read_unlock(&connections_srcu, idx);
557 return -ENOENT;
558 }
559
560 down_read(&con->sock_lock);
561 if (!con->sock) {
562 spin_lock_bh(&con->writequeue_lock);
563 lowcomms_queue_swork(con);
564 spin_unlock_bh(&con->writequeue_lock);
565 }
566 up_read(&con->sock_lock);
567 srcu_read_unlock(&connections_srcu, idx);
568
569 cond_resched();
570 return 0;
571 }
572
573 int dlm_lowcomms_nodes_set_mark(int nodeid, unsigned int mark)
574 {
575 struct connection *con;
576 int idx;
577
578 idx = srcu_read_lock(&connections_srcu);
579 con = nodeid2con(nodeid, 0);
580 if (!con) {
581 srcu_read_unlock(&connections_srcu, idx);
582 return -ENOENT;
583 }
584
585 spin_lock(&con->addrs_lock);
586 con->mark = mark;
587 spin_unlock(&con->addrs_lock);
588 srcu_read_unlock(&connections_srcu, idx);
589 return 0;
590 }
591
592 static void lowcomms_error_report(struct sock *sk)
593 {
594 struct connection *con = sock2con(sk);
595 struct inet_sock *inet;
596
597 inet = inet_sk(sk);
598 switch (sk->sk_family) {
599 case AF_INET:
600 printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
601 "sending to node %d at %pI4, dport %d, "
602 "sk_err=%d/%d\n", dlm_our_nodeid(),
603 con->nodeid, &inet->inet_daddr,
604 ntohs(inet->inet_dport), sk->sk_err,
605 READ_ONCE(sk->sk_err_soft));
606 break;
607 #if IS_ENABLED(CONFIG_IPV6)
608 case AF_INET6:
609 printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
610 "sending to node %d at %pI6c, "
611 "dport %d, sk_err=%d/%d\n", dlm_our_nodeid(),
612 con->nodeid, &sk->sk_v6_daddr,
613 ntohs(inet->inet_dport), sk->sk_err,
614 READ_ONCE(sk->sk_err_soft));
615 break;
616 #endif
617 default:
618 printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
619 "invalid socket family %d set, "
620 "sk_err=%d/%d\n", dlm_our_nodeid(),
621 sk->sk_family, sk->sk_err,
622 READ_ONCE(sk->sk_err_soft));
623 break;
624 }
625
626 dlm_midcomms_unack_msg_resend(con->nodeid);
627
628 listen_sock.sk_error_report(sk);
629 }
630
631 static void restore_callbacks(struct sock *sk)
632 {
633 #ifdef CONFIG_LOCKDEP
634 WARN_ON_ONCE(!lockdep_sock_is_held(sk));
635 #endif
636
637 sk->sk_user_data = NULL;
638 sk->sk_data_ready = listen_sock.sk_data_ready;
639 sk->sk_state_change = listen_sock.sk_state_change;
640 sk->sk_write_space = listen_sock.sk_write_space;
641 sk->sk_error_report = listen_sock.sk_error_report;
642 }
643
644 /* Make a socket active */
645 static void add_sock(struct socket *sock, struct connection *con)
646 {
647 struct sock *sk = sock->sk;
648
649 lock_sock(sk);
650 con->sock = sock;
651
652 sk->sk_user_data = con;
653 sk->sk_data_ready = lowcomms_data_ready;
654 sk->sk_write_space = lowcomms_write_space;
655 if (dlm_config.ci_protocol == DLM_PROTO_SCTP)
656 sk->sk_state_change = lowcomms_state_change;
657 sk->sk_allocation = GFP_NOFS;
658 sk->sk_use_task_frag = false;
659 sk->sk_error_report = lowcomms_error_report;
660 release_sock(sk);
661 }
662
663 /* Add the port number to an IPv6 or 4 sockaddr and return the address
664 length */
665 static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
666 int *addr_len)
667 {
668 saddr->ss_family = dlm_local_addr[0].ss_family;
669 if (saddr->ss_family == AF_INET) {
670 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
671 in4_addr->sin_port = cpu_to_be16(port);
672 *addr_len = sizeof(struct sockaddr_in);
673 memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
674 } else {
675 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
676 in6_addr->sin6_port = cpu_to_be16(port);
677 *addr_len = sizeof(struct sockaddr_in6);
678 }
679 memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
680 }
681
682 static void dlm_page_release(struct kref *kref)
683 {
684 struct writequeue_entry *e = container_of(kref, struct writequeue_entry,
685 ref);
686
687 __free_page(e->page);
688 dlm_free_writequeue(e);
689 }
690
691 static void dlm_msg_release(struct kref *kref)
692 {
693 struct dlm_msg *msg = container_of(kref, struct dlm_msg, ref);
694
695 kref_put(&msg->entry->ref, dlm_page_release);
696 dlm_free_msg(msg);
697 }
698
699 static void free_entry(struct writequeue_entry *e)
700 {
701 struct dlm_msg *msg, *tmp;
702
703 list_for_each_entry_safe(msg, tmp, &e->msgs, list) {
704 if (msg->orig_msg) {
705 msg->orig_msg->retransmit = false;
706 kref_put(&msg->orig_msg->ref, dlm_msg_release);
707 }
708
709 list_del(&msg->list);
710 kref_put(&msg->ref, dlm_msg_release);
711 }
712
713 list_del(&e->list);
714 kref_put(&e->ref, dlm_page_release);
715 }
716
717 static void dlm_close_sock(struct socket **sock)
718 {
719 lock_sock((*sock)->sk);
720 restore_callbacks((*sock)->sk);
721 release_sock((*sock)->sk);
722
723 sock_release(*sock);
724 *sock = NULL;
725 }
726
727 static void allow_connection_io(struct connection *con)
728 {
729 if (con->othercon)
730 clear_bit(CF_IO_STOP, &con->othercon->flags);
731 clear_bit(CF_IO_STOP, &con->flags);
732 }
733
734 static void stop_connection_io(struct connection *con)
735 {
736 if (con->othercon)
737 stop_connection_io(con->othercon);
738
739 spin_lock_bh(&con->writequeue_lock);
740 set_bit(CF_IO_STOP, &con->flags);
741 spin_unlock_bh(&con->writequeue_lock);
742
743 down_write(&con->sock_lock);
744 if (con->sock) {
745 lock_sock(con->sock->sk);
746 restore_callbacks(con->sock->sk);
747 release_sock(con->sock->sk);
748 }
749 up_write(&con->sock_lock);
750
751 cancel_work_sync(&con->swork);
752 cancel_work_sync(&con->rwork);
753 }
754
755 /* Close a remote connection and tidy up */
756 static void close_connection(struct connection *con, bool and_other)
757 {
758 struct writequeue_entry *e;
759
760 if (con->othercon && and_other)
761 close_connection(con->othercon, false);
762
763 down_write(&con->sock_lock);
764 if (!con->sock) {
765 up_write(&con->sock_lock);
766 return;
767 }
768
769 dlm_close_sock(&con->sock);
770
771 /* if we send a writequeue entry only a half way, we drop the
772 * whole entry because reconnection and that we not start of the
773 * middle of a msg which will confuse the other end.
774 *
775 * we can always drop messages because retransmits, but what we
776 * cannot allow is to transmit half messages which may be processed
777 * at the other side.
778 *
779 * our policy is to start on a clean state when disconnects, we don't
780 * know what's send/received on transport layer in this case.
781 */
782 spin_lock_bh(&con->writequeue_lock);
783 if (!list_empty(&con->writequeue)) {
784 e = list_first_entry(&con->writequeue, struct writequeue_entry,
785 list);
786 if (e->dirty)
787 free_entry(e);
788 }
789 spin_unlock_bh(&con->writequeue_lock);
790
791 con->rx_leftover = 0;
792 con->retries = 0;
793 clear_bit(CF_APP_LIMITED, &con->flags);
794 clear_bit(CF_RECV_PENDING, &con->flags);
795 clear_bit(CF_SEND_PENDING, &con->flags);
796 up_write(&con->sock_lock);
797 }
798
799 static void shutdown_connection(struct connection *con, bool and_other)
800 {
801 int ret;
802
803 if (con->othercon && and_other)
804 shutdown_connection(con->othercon, false);
805
806 flush_workqueue(io_workqueue);
807 down_read(&con->sock_lock);
808 /* nothing to shutdown */
809 if (!con->sock) {
810 up_read(&con->sock_lock);
811 return;
812 }
813
814 ret = kernel_sock_shutdown(con->sock, SHUT_WR);
815 up_read(&con->sock_lock);
816 if (ret) {
817 log_print("Connection %p failed to shutdown: %d will force close",
818 con, ret);
819 goto force_close;
820 } else {
821 ret = wait_event_timeout(con->shutdown_wait, !con->sock,
822 DLM_SHUTDOWN_WAIT_TIMEOUT);
823 if (ret == 0) {
824 log_print("Connection %p shutdown timed out, will force close",
825 con);
826 goto force_close;
827 }
828 }
829
830 return;
831
832 force_close:
833 close_connection(con, false);
834 }
835
836 static struct processqueue_entry *new_processqueue_entry(int nodeid,
837 int buflen)
838 {
839 struct processqueue_entry *pentry;
840
841 pentry = kmalloc(sizeof(*pentry), GFP_NOFS);
842 if (!pentry)
843 return NULL;
844
845 pentry->buf = kmalloc(buflen, GFP_NOFS);
846 if (!pentry->buf) {
847 kfree(pentry);
848 return NULL;
849 }
850
851 pentry->nodeid = nodeid;
852 return pentry;
853 }
854
855 static void free_processqueue_entry(struct processqueue_entry *pentry)
856 {
857 kfree(pentry->buf);
858 kfree(pentry);
859 }
860
861 static void process_dlm_messages(struct work_struct *work)
862 {
863 struct processqueue_entry *pentry;
864
865 spin_lock_bh(&processqueue_lock);
866 pentry = list_first_entry_or_null(&processqueue,
867 struct processqueue_entry, list);
868 if (WARN_ON_ONCE(!pentry)) {
869 process_dlm_messages_pending = false;
870 spin_unlock_bh(&processqueue_lock);
871 return;
872 }
873
874 list_del(&pentry->list);
875 if (atomic_dec_and_test(&processqueue_count))
876 wake_up(&processqueue_wq);
877 spin_unlock_bh(&processqueue_lock);
878
879 for (;;) {
880 dlm_process_incoming_buffer(pentry->nodeid, pentry->buf,
881 pentry->buflen);
882 free_processqueue_entry(pentry);
883
884 spin_lock_bh(&processqueue_lock);
885 pentry = list_first_entry_or_null(&processqueue,
886 struct processqueue_entry, list);
887 if (!pentry) {
888 process_dlm_messages_pending = false;
889 spin_unlock_bh(&processqueue_lock);
890 break;
891 }
892
893 list_del(&pentry->list);
894 if (atomic_dec_and_test(&processqueue_count))
895 wake_up(&processqueue_wq);
896 spin_unlock_bh(&processqueue_lock);
897 }
898 }
899
900 /* Data received from remote end */
901 static int receive_from_sock(struct connection *con, int buflen)
902 {
903 struct processqueue_entry *pentry;
904 int ret, buflen_real;
905 struct msghdr msg;
906 struct kvec iov;
907
908 pentry = new_processqueue_entry(con->nodeid, buflen);
909 if (!pentry)
910 return DLM_IO_RESCHED;
911
912 memcpy(pentry->buf, con->rx_leftover_buf, con->rx_leftover);
913
914 /* calculate new buffer parameter regarding last receive and
915 * possible leftover bytes
916 */
917 iov.iov_base = pentry->buf + con->rx_leftover;
918 iov.iov_len = buflen - con->rx_leftover;
919
920 memset(&msg, 0, sizeof(msg));
921 msg.msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
922 clear_bit(CF_RECV_INTR, &con->flags);
923 again:
924 ret = kernel_recvmsg(con->sock, &msg, &iov, 1, iov.iov_len,
925 msg.msg_flags);
926 trace_dlm_recv(con->nodeid, ret);
927 if (ret == -EAGAIN) {
928 lock_sock(con->sock->sk);
929 if (test_and_clear_bit(CF_RECV_INTR, &con->flags)) {
930 release_sock(con->sock->sk);
931 goto again;
932 }
933
934 clear_bit(CF_RECV_PENDING, &con->flags);
935 release_sock(con->sock->sk);
936 free_processqueue_entry(pentry);
937 return DLM_IO_END;
938 } else if (ret == 0) {
939 /* close will clear CF_RECV_PENDING */
940 free_processqueue_entry(pentry);
941 return DLM_IO_EOF;
942 } else if (ret < 0) {
943 free_processqueue_entry(pentry);
944 return ret;
945 }
946
947 /* new buflen according readed bytes and leftover from last receive */
948 buflen_real = ret + con->rx_leftover;
949 ret = dlm_validate_incoming_buffer(con->nodeid, pentry->buf,
950 buflen_real);
951 if (ret < 0) {
952 free_processqueue_entry(pentry);
953 return ret;
954 }
955
956 pentry->buflen = ret;
957
958 /* calculate leftover bytes from process and put it into begin of
959 * the receive buffer, so next receive we have the full message
960 * at the start address of the receive buffer.
961 */
962 con->rx_leftover = buflen_real - ret;
963 memmove(con->rx_leftover_buf, pentry->buf + ret,
964 con->rx_leftover);
965
966 spin_lock_bh(&processqueue_lock);
967 ret = atomic_inc_return(&processqueue_count);
968 list_add_tail(&pentry->list, &processqueue);
969 if (!process_dlm_messages_pending) {
970 process_dlm_messages_pending = true;
971 queue_work(process_workqueue, &process_work);
972 }
973 spin_unlock_bh(&processqueue_lock);
974
975 if (ret > DLM_MAX_PROCESS_BUFFERS)
976 return DLM_IO_FLUSH;
977
978 return DLM_IO_SUCCESS;
979 }
980
981 /* Listening socket is busy, accept a connection */
982 static int accept_from_sock(void)
983 {
984 struct sockaddr_storage peeraddr;
985 int len, idx, result, nodeid;
986 struct connection *newcon;
987 struct socket *newsock;
988 unsigned int mark;
989
990 result = kernel_accept(listen_con.sock, &newsock, O_NONBLOCK);
991 if (result == -EAGAIN)
992 return DLM_IO_END;
993 else if (result < 0)
994 goto accept_err;
995
996 /* Get the connected socket's peer */
997 memset(&peeraddr, 0, sizeof(peeraddr));
998 len = newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 2);
999 if (len < 0) {
1000 result = -ECONNABORTED;
1001 goto accept_err;
1002 }
1003
1004 /* Get the new node's NODEID */
1005 make_sockaddr(&peeraddr, 0, &len);
1006 if (addr_to_nodeid(&peeraddr, &nodeid, &mark)) {
1007 switch (peeraddr.ss_family) {
1008 case AF_INET: {
1009 struct sockaddr_in *sin = (struct sockaddr_in *)&peeraddr;
1010
1011 log_print("connect from non cluster IPv4 node %pI4",
1012 &sin->sin_addr);
1013 break;
1014 }
1015 #if IS_ENABLED(CONFIG_IPV6)
1016 case AF_INET6: {
1017 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&peeraddr;
1018
1019 log_print("connect from non cluster IPv6 node %pI6c",
1020 &sin6->sin6_addr);
1021 break;
1022 }
1023 #endif
1024 default:
1025 log_print("invalid family from non cluster node");
1026 break;
1027 }
1028
1029 sock_release(newsock);
1030 return -1;
1031 }
1032
1033 log_print("got connection from %d", nodeid);
1034
1035 /* Check to see if we already have a connection to this node. This
1036 * could happen if the two nodes initiate a connection at roughly
1037 * the same time and the connections cross on the wire.
1038 * In this case we store the incoming one in "othercon"
1039 */
1040 idx = srcu_read_lock(&connections_srcu);
1041 newcon = nodeid2con(nodeid, 0);
1042 if (WARN_ON_ONCE(!newcon)) {
1043 srcu_read_unlock(&connections_srcu, idx);
1044 result = -ENOENT;
1045 goto accept_err;
1046 }
1047
1048 sock_set_mark(newsock->sk, mark);
1049
1050 down_write(&newcon->sock_lock);
1051 if (newcon->sock) {
1052 struct connection *othercon = newcon->othercon;
1053
1054 if (!othercon) {
1055 othercon = kzalloc(sizeof(*othercon), GFP_NOFS);
1056 if (!othercon) {
1057 log_print("failed to allocate incoming socket");
1058 up_write(&newcon->sock_lock);
1059 srcu_read_unlock(&connections_srcu, idx);
1060 result = -ENOMEM;
1061 goto accept_err;
1062 }
1063
1064 dlm_con_init(othercon, nodeid);
1065 lockdep_set_subclass(&othercon->sock_lock, 1);
1066 newcon->othercon = othercon;
1067 set_bit(CF_IS_OTHERCON, &othercon->flags);
1068 } else {
1069 /* close other sock con if we have something new */
1070 close_connection(othercon, false);
1071 }
1072
1073 down_write(&othercon->sock_lock);
1074 add_sock(newsock, othercon);
1075
1076 /* check if we receved something while adding */
1077 lock_sock(othercon->sock->sk);
1078 lowcomms_queue_rwork(othercon);
1079 release_sock(othercon->sock->sk);
1080 up_write(&othercon->sock_lock);
1081 }
1082 else {
1083 /* accept copies the sk after we've saved the callbacks, so we
1084 don't want to save them a second time or comm errors will
1085 result in calling sk_error_report recursively. */
1086 add_sock(newsock, newcon);
1087
1088 /* check if we receved something while adding */
1089 lock_sock(newcon->sock->sk);
1090 lowcomms_queue_rwork(newcon);
1091 release_sock(newcon->sock->sk);
1092 }
1093 up_write(&newcon->sock_lock);
1094 srcu_read_unlock(&connections_srcu, idx);
1095
1096 return DLM_IO_SUCCESS;
1097
1098 accept_err:
1099 if (newsock)
1100 sock_release(newsock);
1101
1102 return result;
1103 }
1104
1105 /*
1106 * writequeue_entry_complete - try to delete and free write queue entry
1107 * @e: write queue entry to try to delete
1108 * @completed: bytes completed
1109 *
1110 * writequeue_lock must be held.
1111 */
1112 static void writequeue_entry_complete(struct writequeue_entry *e, int completed)
1113 {
1114 e->offset += completed;
1115 e->len -= completed;
1116 /* signal that page was half way transmitted */
1117 e->dirty = true;
1118
1119 if (e->len == 0 && e->users == 0)
1120 free_entry(e);
1121 }
1122
1123 /*
1124 * sctp_bind_addrs - bind a SCTP socket to all our addresses
1125 */
1126 static int sctp_bind_addrs(struct socket *sock, uint16_t port)
1127 {
1128 struct sockaddr_storage localaddr;
1129 struct sockaddr *addr = (struct sockaddr *)&localaddr;
1130 int i, addr_len, result = 0;
1131
1132 for (i = 0; i < dlm_local_count; i++) {
1133 memcpy(&localaddr, &dlm_local_addr[i], sizeof(localaddr));
1134 make_sockaddr(&localaddr, port, &addr_len);
1135
1136 if (!i)
1137 result = kernel_bind(sock, addr, addr_len);
1138 else
1139 result = sock_bind_add(sock->sk, addr, addr_len);
1140
1141 if (result < 0) {
1142 log_print("Can't bind to %d addr number %d, %d.\n",
1143 port, i + 1, result);
1144 break;
1145 }
1146 }
1147 return result;
1148 }
1149
1150 /* Get local addresses */
1151 static void init_local(void)
1152 {
1153 struct sockaddr_storage sas;
1154 int i;
1155
1156 dlm_local_count = 0;
1157 for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
1158 if (dlm_our_addr(&sas, i))
1159 break;
1160
1161 memcpy(&dlm_local_addr[dlm_local_count++], &sas, sizeof(sas));
1162 }
1163 }
1164
1165 static struct writequeue_entry *new_writequeue_entry(struct connection *con)
1166 {
1167 struct writequeue_entry *entry;
1168
1169 entry = dlm_allocate_writequeue();
1170 if (!entry)
1171 return NULL;
1172
1173 entry->page = alloc_page(GFP_ATOMIC | __GFP_ZERO);
1174 if (!entry->page) {
1175 dlm_free_writequeue(entry);
1176 return NULL;
1177 }
1178
1179 entry->offset = 0;
1180 entry->len = 0;
1181 entry->end = 0;
1182 entry->dirty = false;
1183 entry->con = con;
1184 entry->users = 1;
1185 kref_init(&entry->ref);
1186 return entry;
1187 }
1188
1189 static struct writequeue_entry *new_wq_entry(struct connection *con, int len,
1190 char **ppc, void (*cb)(void *data),
1191 void *data)
1192 {
1193 struct writequeue_entry *e;
1194
1195 spin_lock_bh(&con->writequeue_lock);
1196 if (!list_empty(&con->writequeue)) {
1197 e = list_last_entry(&con->writequeue, struct writequeue_entry, list);
1198 if (DLM_WQ_REMAIN_BYTES(e) >= len) {
1199 kref_get(&e->ref);
1200
1201 *ppc = page_address(e->page) + e->end;
1202 if (cb)
1203 cb(data);
1204
1205 e->end += len;
1206 e->users++;
1207 goto out;
1208 }
1209 }
1210
1211 e = new_writequeue_entry(con);
1212 if (!e)
1213 goto out;
1214
1215 kref_get(&e->ref);
1216 *ppc = page_address(e->page);
1217 e->end += len;
1218 if (cb)
1219 cb(data);
1220
1221 list_add_tail(&e->list, &con->writequeue);
1222
1223 out:
1224 spin_unlock_bh(&con->writequeue_lock);
1225 return e;
1226 };
1227
1228 static struct dlm_msg *dlm_lowcomms_new_msg_con(struct connection *con, int len,
1229 char **ppc, void (*cb)(void *data),
1230 void *data)
1231 {
1232 struct writequeue_entry *e;
1233 struct dlm_msg *msg;
1234
1235 msg = dlm_allocate_msg();
1236 if (!msg)
1237 return NULL;
1238
1239 kref_init(&msg->ref);
1240
1241 e = new_wq_entry(con, len, ppc, cb, data);
1242 if (!e) {
1243 dlm_free_msg(msg);
1244 return NULL;
1245 }
1246
1247 msg->retransmit = false;
1248 msg->orig_msg = NULL;
1249 msg->ppc = *ppc;
1250 msg->len = len;
1251 msg->entry = e;
1252
1253 return msg;
1254 }
1255
1256 /* avoid false positive for nodes_srcu, unlock happens in
1257 * dlm_lowcomms_commit_msg which is a must call if success
1258 */
1259 #ifndef __CHECKER__
1260 struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, char **ppc,
1261 void (*cb)(void *data), void *data)
1262 {
1263 struct connection *con;
1264 struct dlm_msg *msg;
1265 int idx;
1266
1267 if (len > DLM_MAX_SOCKET_BUFSIZE ||
1268 len < sizeof(struct dlm_header)) {
1269 BUILD_BUG_ON(PAGE_SIZE < DLM_MAX_SOCKET_BUFSIZE);
1270 log_print("failed to allocate a buffer of size %d", len);
1271 WARN_ON_ONCE(1);
1272 return NULL;
1273 }
1274
1275 idx = srcu_read_lock(&connections_srcu);
1276 con = nodeid2con(nodeid, 0);
1277 if (WARN_ON_ONCE(!con)) {
1278 srcu_read_unlock(&connections_srcu, idx);
1279 return NULL;
1280 }
1281
1282 msg = dlm_lowcomms_new_msg_con(con, len, ppc, cb, data);
1283 if (!msg) {
1284 srcu_read_unlock(&connections_srcu, idx);
1285 return NULL;
1286 }
1287
1288 /* for dlm_lowcomms_commit_msg() */
1289 kref_get(&msg->ref);
1290 /* we assume if successful commit must called */
1291 msg->idx = idx;
1292 return msg;
1293 }
1294 #endif
1295
1296 static void _dlm_lowcomms_commit_msg(struct dlm_msg *msg)
1297 {
1298 struct writequeue_entry *e = msg->entry;
1299 struct connection *con = e->con;
1300 int users;
1301
1302 spin_lock_bh(&con->writequeue_lock);
1303 kref_get(&msg->ref);
1304 list_add(&msg->list, &e->msgs);
1305
1306 users = --e->users;
1307 if (users)
1308 goto out;
1309
1310 e->len = DLM_WQ_LENGTH_BYTES(e);
1311
1312 lowcomms_queue_swork(con);
1313
1314 out:
1315 spin_unlock_bh(&con->writequeue_lock);
1316 return;
1317 }
1318
1319 /* avoid false positive for nodes_srcu, lock was happen in
1320 * dlm_lowcomms_new_msg
1321 */
1322 #ifndef __CHECKER__
1323 void dlm_lowcomms_commit_msg(struct dlm_msg *msg)
1324 {
1325 _dlm_lowcomms_commit_msg(msg);
1326 srcu_read_unlock(&connections_srcu, msg->idx);
1327 /* because dlm_lowcomms_new_msg() */
1328 kref_put(&msg->ref, dlm_msg_release);
1329 }
1330 #endif
1331
1332 void dlm_lowcomms_put_msg(struct dlm_msg *msg)
1333 {
1334 kref_put(&msg->ref, dlm_msg_release);
1335 }
1336
1337 /* does not held connections_srcu, usage lowcomms_error_report only */
1338 int dlm_lowcomms_resend_msg(struct dlm_msg *msg)
1339 {
1340 struct dlm_msg *msg_resend;
1341 char *ppc;
1342
1343 if (msg->retransmit)
1344 return 1;
1345
1346 msg_resend = dlm_lowcomms_new_msg_con(msg->entry->con, msg->len, &ppc,
1347 NULL, NULL);
1348 if (!msg_resend)
1349 return -ENOMEM;
1350
1351 msg->retransmit = true;
1352 kref_get(&msg->ref);
1353 msg_resend->orig_msg = msg;
1354
1355 memcpy(ppc, msg->ppc, msg->len);
1356 _dlm_lowcomms_commit_msg(msg_resend);
1357 dlm_lowcomms_put_msg(msg_resend);
1358
1359 return 0;
1360 }
1361
1362 /* Send a message */
1363 static int send_to_sock(struct connection *con)
1364 {
1365 struct writequeue_entry *e;
1366 struct bio_vec bvec;
1367 struct msghdr msg = {
1368 .msg_flags = MSG_SPLICE_PAGES | MSG_DONTWAIT | MSG_NOSIGNAL,
1369 };
1370 int len, offset, ret;
1371
1372 spin_lock_bh(&con->writequeue_lock);
1373 e = con_next_wq(con);
1374 if (!e) {
1375 clear_bit(CF_SEND_PENDING, &con->flags);
1376 spin_unlock_bh(&con->writequeue_lock);
1377 return DLM_IO_END;
1378 }
1379
1380 len = e->len;
1381 offset = e->offset;
1382 WARN_ON_ONCE(len == 0 && e->users == 0);
1383 spin_unlock_bh(&con->writequeue_lock);
1384
1385 bvec_set_page(&bvec, e->page, len, offset);
1386 iov_iter_bvec(&msg.msg_iter, ITER_SOURCE, &bvec, 1, len);
1387 ret = sock_sendmsg(con->sock, &msg);
1388 trace_dlm_send(con->nodeid, ret);
1389 if (ret == -EAGAIN || ret == 0) {
1390 lock_sock(con->sock->sk);
1391 spin_lock_bh(&con->writequeue_lock);
1392 if (test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) &&
1393 !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1394 /* Notify TCP that we're limited by the
1395 * application window size.
1396 */
1397 set_bit(SOCK_NOSPACE, &con->sock->sk->sk_socket->flags);
1398 con->sock->sk->sk_write_pending++;
1399
1400 clear_bit(CF_SEND_PENDING, &con->flags);
1401 spin_unlock_bh(&con->writequeue_lock);
1402 release_sock(con->sock->sk);
1403
1404 /* wait for write_space() event */
1405 return DLM_IO_END;
1406 }
1407 spin_unlock_bh(&con->writequeue_lock);
1408 release_sock(con->sock->sk);
1409
1410 return DLM_IO_RESCHED;
1411 } else if (ret < 0) {
1412 return ret;
1413 }
1414
1415 spin_lock_bh(&con->writequeue_lock);
1416 writequeue_entry_complete(e, ret);
1417 spin_unlock_bh(&con->writequeue_lock);
1418
1419 return DLM_IO_SUCCESS;
1420 }
1421
1422 static void clean_one_writequeue(struct connection *con)
1423 {
1424 struct writequeue_entry *e, *safe;
1425
1426 spin_lock_bh(&con->writequeue_lock);
1427 list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1428 free_entry(e);
1429 }
1430 spin_unlock_bh(&con->writequeue_lock);
1431 }
1432
1433 static void connection_release(struct rcu_head *rcu)
1434 {
1435 struct connection *con = container_of(rcu, struct connection, rcu);
1436
1437 WARN_ON_ONCE(!list_empty(&con->writequeue));
1438 WARN_ON_ONCE(con->sock);
1439 kfree(con);
1440 }
1441
1442 /* Called from recovery when it knows that a node has
1443 left the cluster */
1444 int dlm_lowcomms_close(int nodeid)
1445 {
1446 struct connection *con;
1447 int idx;
1448
1449 log_print("closing connection to node %d", nodeid);
1450
1451 idx = srcu_read_lock(&connections_srcu);
1452 con = nodeid2con(nodeid, 0);
1453 if (WARN_ON_ONCE(!con)) {
1454 srcu_read_unlock(&connections_srcu, idx);
1455 return -ENOENT;
1456 }
1457
1458 stop_connection_io(con);
1459 log_print("io handling for node: %d stopped", nodeid);
1460 close_connection(con, true);
1461
1462 spin_lock(&connections_lock);
1463 hlist_del_rcu(&con->list);
1464 spin_unlock(&connections_lock);
1465
1466 clean_one_writequeue(con);
1467 call_srcu(&connections_srcu, &con->rcu, connection_release);
1468 if (con->othercon) {
1469 clean_one_writequeue(con->othercon);
1470 call_srcu(&connections_srcu, &con->othercon->rcu, connection_release);
1471 }
1472 srcu_read_unlock(&connections_srcu, idx);
1473
1474 /* for debugging we print when we are done to compare with other
1475 * messages in between. This function need to be correctly synchronized
1476 * with io handling
1477 */
1478 log_print("closing connection to node %d done", nodeid);
1479
1480 return 0;
1481 }
1482
1483 /* Receive worker function */
1484 static void process_recv_sockets(struct work_struct *work)
1485 {
1486 struct connection *con = container_of(work, struct connection, rwork);
1487 int ret, buflen;
1488
1489 down_read(&con->sock_lock);
1490 if (!con->sock) {
1491 up_read(&con->sock_lock);
1492 return;
1493 }
1494
1495 buflen = READ_ONCE(dlm_config.ci_buffer_size);
1496 do {
1497 ret = receive_from_sock(con, buflen);
1498 } while (ret == DLM_IO_SUCCESS);
1499 up_read(&con->sock_lock);
1500
1501 switch (ret) {
1502 case DLM_IO_END:
1503 /* CF_RECV_PENDING cleared */
1504 break;
1505 case DLM_IO_EOF:
1506 close_connection(con, false);
1507 wake_up(&con->shutdown_wait);
1508 /* CF_RECV_PENDING cleared */
1509 break;
1510 case DLM_IO_FLUSH:
1511 /* we can't flush the process_workqueue here because a
1512 * WQ_MEM_RECLAIM workequeue can occurr a deadlock for a non
1513 * WQ_MEM_RECLAIM workqueue such as process_workqueue. Instead
1514 * we have a waitqueue to wait until all messages are
1515 * processed.
1516 *
1517 * This handling is only necessary to backoff the sender and
1518 * not queue all messages from the socket layer into DLM
1519 * processqueue. When DLM is capable to parse multiple messages
1520 * on an e.g. per socket basis this handling can might be
1521 * removed. Especially in a message burst we are too slow to
1522 * process messages and the queue will fill up memory.
1523 */
1524 wait_event(processqueue_wq, !atomic_read(&processqueue_count));
1525 fallthrough;
1526 case DLM_IO_RESCHED:
1527 cond_resched();
1528 queue_work(io_workqueue, &con->rwork);
1529 /* CF_RECV_PENDING not cleared */
1530 break;
1531 default:
1532 if (ret < 0) {
1533 if (test_bit(CF_IS_OTHERCON, &con->flags)) {
1534 close_connection(con, false);
1535 } else {
1536 spin_lock_bh(&con->writequeue_lock);
1537 lowcomms_queue_swork(con);
1538 spin_unlock_bh(&con->writequeue_lock);
1539 }
1540
1541 /* CF_RECV_PENDING cleared for othercon
1542 * we trigger send queue if not already done
1543 * and process_send_sockets will handle it
1544 */
1545 break;
1546 }
1547
1548 WARN_ON_ONCE(1);
1549 break;
1550 }
1551 }
1552
1553 static void process_listen_recv_socket(struct work_struct *work)
1554 {
1555 int ret;
1556
1557 if (WARN_ON_ONCE(!listen_con.sock))
1558 return;
1559
1560 do {
1561 ret = accept_from_sock();
1562 } while (ret == DLM_IO_SUCCESS);
1563
1564 if (ret < 0)
1565 log_print("critical error accepting connection: %d", ret);
1566 }
1567
1568 static int dlm_connect(struct connection *con)
1569 {
1570 struct sockaddr_storage addr;
1571 int result, addr_len;
1572 struct socket *sock;
1573 unsigned int mark;
1574
1575 memset(&addr, 0, sizeof(addr));
1576 result = nodeid_to_addr(con->nodeid, &addr, NULL,
1577 dlm_proto_ops->try_new_addr, &mark);
1578 if (result < 0) {
1579 log_print("no address for nodeid %d", con->nodeid);
1580 return result;
1581 }
1582
1583 /* Create a socket to communicate with */
1584 result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family,
1585 SOCK_STREAM, dlm_proto_ops->proto, &sock);
1586 if (result < 0)
1587 return result;
1588
1589 sock_set_mark(sock->sk, mark);
1590 dlm_proto_ops->sockopts(sock);
1591
1592 result = dlm_proto_ops->bind(sock);
1593 if (result < 0) {
1594 sock_release(sock);
1595 return result;
1596 }
1597
1598 add_sock(sock, con);
1599
1600 log_print_ratelimited("connecting to %d", con->nodeid);
1601 make_sockaddr(&addr, dlm_config.ci_tcp_port, &addr_len);
1602 result = dlm_proto_ops->connect(con, sock, (struct sockaddr *)&addr,
1603 addr_len);
1604 switch (result) {
1605 case -EINPROGRESS:
1606 /* not an error */
1607 fallthrough;
1608 case 0:
1609 break;
1610 default:
1611 if (result < 0)
1612 dlm_close_sock(&con->sock);
1613
1614 break;
1615 }
1616
1617 return result;
1618 }
1619
1620 /* Send worker function */
1621 static void process_send_sockets(struct work_struct *work)
1622 {
1623 struct connection *con = container_of(work, struct connection, swork);
1624 int ret;
1625
1626 WARN_ON_ONCE(test_bit(CF_IS_OTHERCON, &con->flags));
1627
1628 down_read(&con->sock_lock);
1629 if (!con->sock) {
1630 up_read(&con->sock_lock);
1631 down_write(&con->sock_lock);
1632 if (!con->sock) {
1633 ret = dlm_connect(con);
1634 switch (ret) {
1635 case 0:
1636 break;
1637 case -EINPROGRESS:
1638 /* avoid spamming resched on connection
1639 * we might can switch to a state_change
1640 * event based mechanism if established
1641 */
1642 msleep(100);
1643 break;
1644 default:
1645 /* CF_SEND_PENDING not cleared */
1646 up_write(&con->sock_lock);
1647 log_print("connect to node %d try %d error %d",
1648 con->nodeid, con->retries++, ret);
1649 msleep(1000);
1650 /* For now we try forever to reconnect. In
1651 * future we should send a event to cluster
1652 * manager to fence itself after certain amount
1653 * of retries.
1654 */
1655 queue_work(io_workqueue, &con->swork);
1656 return;
1657 }
1658 }
1659 downgrade_write(&con->sock_lock);
1660 }
1661
1662 do {
1663 ret = send_to_sock(con);
1664 } while (ret == DLM_IO_SUCCESS);
1665 up_read(&con->sock_lock);
1666
1667 switch (ret) {
1668 case DLM_IO_END:
1669 /* CF_SEND_PENDING cleared */
1670 break;
1671 case DLM_IO_RESCHED:
1672 /* CF_SEND_PENDING not cleared */
1673 cond_resched();
1674 queue_work(io_workqueue, &con->swork);
1675 break;
1676 default:
1677 if (ret < 0) {
1678 close_connection(con, false);
1679
1680 /* CF_SEND_PENDING cleared */
1681 spin_lock_bh(&con->writequeue_lock);
1682 lowcomms_queue_swork(con);
1683 spin_unlock_bh(&con->writequeue_lock);
1684 break;
1685 }
1686
1687 WARN_ON_ONCE(1);
1688 break;
1689 }
1690 }
1691
1692 static void work_stop(void)
1693 {
1694 if (io_workqueue) {
1695 destroy_workqueue(io_workqueue);
1696 io_workqueue = NULL;
1697 }
1698
1699 if (process_workqueue) {
1700 destroy_workqueue(process_workqueue);
1701 process_workqueue = NULL;
1702 }
1703 }
1704
1705 static int work_start(void)
1706 {
1707 io_workqueue = alloc_workqueue("dlm_io", WQ_HIGHPRI | WQ_MEM_RECLAIM |
1708 WQ_UNBOUND, 0);
1709 if (!io_workqueue) {
1710 log_print("can't start dlm_io");
1711 return -ENOMEM;
1712 }
1713
1714 process_workqueue = alloc_workqueue("dlm_process", WQ_HIGHPRI | WQ_BH, 0);
1715 if (!process_workqueue) {
1716 log_print("can't start dlm_process");
1717 destroy_workqueue(io_workqueue);
1718 io_workqueue = NULL;
1719 return -ENOMEM;
1720 }
1721
1722 return 0;
1723 }
1724
1725 void dlm_lowcomms_shutdown(void)
1726 {
1727 struct connection *con;
1728 int i, idx;
1729
1730 /* stop lowcomms_listen_data_ready calls */
1731 lock_sock(listen_con.sock->sk);
1732 listen_con.sock->sk->sk_data_ready = listen_sock.sk_data_ready;
1733 release_sock(listen_con.sock->sk);
1734
1735 cancel_work_sync(&listen_con.rwork);
1736 dlm_close_sock(&listen_con.sock);
1737
1738 idx = srcu_read_lock(&connections_srcu);
1739 for (i = 0; i < CONN_HASH_SIZE; i++) {
1740 hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
1741 shutdown_connection(con, true);
1742 stop_connection_io(con);
1743 flush_workqueue(process_workqueue);
1744 close_connection(con, true);
1745
1746 clean_one_writequeue(con);
1747 if (con->othercon)
1748 clean_one_writequeue(con->othercon);
1749 allow_connection_io(con);
1750 }
1751 }
1752 srcu_read_unlock(&connections_srcu, idx);
1753 }
1754
1755 void dlm_lowcomms_stop(void)
1756 {
1757 work_stop();
1758 dlm_proto_ops = NULL;
1759 }
1760
1761 static int dlm_listen_for_all(void)
1762 {
1763 struct socket *sock;
1764 int result;
1765
1766 log_print("Using %s for communications",
1767 dlm_proto_ops->name);
1768
1769 result = dlm_proto_ops->listen_validate();
1770 if (result < 0)
1771 return result;
1772
1773 result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family,
1774 SOCK_STREAM, dlm_proto_ops->proto, &sock);
1775 if (result < 0) {
1776 log_print("Can't create comms socket: %d", result);
1777 return result;
1778 }
1779
1780 sock_set_mark(sock->sk, dlm_config.ci_mark);
1781 dlm_proto_ops->listen_sockopts(sock);
1782
1783 result = dlm_proto_ops->listen_bind(sock);
1784 if (result < 0)
1785 goto out;
1786
1787 lock_sock(sock->sk);
1788 listen_sock.sk_data_ready = sock->sk->sk_data_ready;
1789 listen_sock.sk_write_space = sock->sk->sk_write_space;
1790 listen_sock.sk_error_report = sock->sk->sk_error_report;
1791 listen_sock.sk_state_change = sock->sk->sk_state_change;
1792
1793 listen_con.sock = sock;
1794
1795 sock->sk->sk_allocation = GFP_NOFS;
1796 sock->sk->sk_use_task_frag = false;
1797 sock->sk->sk_data_ready = lowcomms_listen_data_ready;
1798 release_sock(sock->sk);
1799
1800 result = sock->ops->listen(sock, 128);
1801 if (result < 0) {
1802 dlm_close_sock(&listen_con.sock);
1803 return result;
1804 }
1805
1806 return 0;
1807
1808 out:
1809 sock_release(sock);
1810 return result;
1811 }
1812
1813 static int dlm_tcp_bind(struct socket *sock)
1814 {
1815 struct sockaddr_storage src_addr;
1816 int result, addr_len;
1817
1818 /* Bind to our cluster-known address connecting to avoid
1819 * routing problems.
1820 */
1821 memcpy(&src_addr, &dlm_local_addr[0], sizeof(src_addr));
1822 make_sockaddr(&src_addr, 0, &addr_len);
1823
1824 result = kernel_bind(sock, (struct sockaddr *)&src_addr,
1825 addr_len);
1826 if (result < 0) {
1827 /* This *may* not indicate a critical error */
1828 log_print("could not bind for connect: %d", result);
1829 }
1830
1831 return 0;
1832 }
1833
1834 static int dlm_tcp_connect(struct connection *con, struct socket *sock,
1835 struct sockaddr *addr, int addr_len)
1836 {
1837 return kernel_connect(sock, addr, addr_len, O_NONBLOCK);
1838 }
1839
1840 static int dlm_tcp_listen_validate(void)
1841 {
1842 /* We don't support multi-homed hosts */
1843 if (dlm_local_count > 1) {
1844 log_print("TCP protocol can't handle multi-homed hosts, try SCTP");
1845 return -EINVAL;
1846 }
1847
1848 return 0;
1849 }
1850
1851 static void dlm_tcp_sockopts(struct socket *sock)
1852 {
1853 /* Turn off Nagle's algorithm */
1854 tcp_sock_set_nodelay(sock->sk);
1855 }
1856
1857 static void dlm_tcp_listen_sockopts(struct socket *sock)
1858 {
1859 dlm_tcp_sockopts(sock);
1860 sock_set_reuseaddr(sock->sk);
1861 }
1862
1863 static int dlm_tcp_listen_bind(struct socket *sock)
1864 {
1865 int addr_len;
1866
1867 /* Bind to our port */
1868 make_sockaddr(&dlm_local_addr[0], dlm_config.ci_tcp_port, &addr_len);
1869 return kernel_bind(sock, (struct sockaddr *)&dlm_local_addr[0],
1870 addr_len);
1871 }
1872
1873 static const struct dlm_proto_ops dlm_tcp_ops = {
1874 .name = "TCP",
1875 .proto = IPPROTO_TCP,
1876 .connect = dlm_tcp_connect,
1877 .sockopts = dlm_tcp_sockopts,
1878 .bind = dlm_tcp_bind,
1879 .listen_validate = dlm_tcp_listen_validate,
1880 .listen_sockopts = dlm_tcp_listen_sockopts,
1881 .listen_bind = dlm_tcp_listen_bind,
1882 };
1883
1884 static int dlm_sctp_bind(struct socket *sock)
1885 {
1886 return sctp_bind_addrs(sock, 0);
1887 }
1888
1889 static int dlm_sctp_connect(struct connection *con, struct socket *sock,
1890 struct sockaddr *addr, int addr_len)
1891 {
1892 int ret;
1893
1894 /*
1895 * Make kernel_connect() function return in specified time,
1896 * since O_NONBLOCK argument in connect() function does not work here,
1897 * then, we should restore the default value of this attribute.
1898 */
1899 sock_set_sndtimeo(sock->sk, 5);
1900 ret = kernel_connect(sock, addr, addr_len, 0);
1901 sock_set_sndtimeo(sock->sk, 0);
1902 return ret;
1903 }
1904
1905 static int dlm_sctp_listen_validate(void)
1906 {
1907 if (!IS_ENABLED(CONFIG_IP_SCTP)) {
1908 log_print("SCTP is not enabled by this kernel");
1909 return -EOPNOTSUPP;
1910 }
1911
1912 request_module("sctp");
1913 return 0;
1914 }
1915
1916 static int dlm_sctp_bind_listen(struct socket *sock)
1917 {
1918 return sctp_bind_addrs(sock, dlm_config.ci_tcp_port);
1919 }
1920
1921 static void dlm_sctp_sockopts(struct socket *sock)
1922 {
1923 /* Turn off Nagle's algorithm */
1924 sctp_sock_set_nodelay(sock->sk);
1925 sock_set_rcvbuf(sock->sk, NEEDED_RMEM);
1926 }
1927
1928 static const struct dlm_proto_ops dlm_sctp_ops = {
1929 .name = "SCTP",
1930 .proto = IPPROTO_SCTP,
1931 .try_new_addr = true,
1932 .connect = dlm_sctp_connect,
1933 .sockopts = dlm_sctp_sockopts,
1934 .bind = dlm_sctp_bind,
1935 .listen_validate = dlm_sctp_listen_validate,
1936 .listen_sockopts = dlm_sctp_sockopts,
1937 .listen_bind = dlm_sctp_bind_listen,
1938 };
1939
1940 int dlm_lowcomms_start(void)
1941 {
1942 int error;
1943
1944 init_local();
1945 if (!dlm_local_count) {
1946 error = -ENOTCONN;
1947 log_print("no local IP address has been set");
1948 goto fail;
1949 }
1950
1951 error = work_start();
1952 if (error)
1953 goto fail;
1954
1955 /* Start listening */
1956 switch (dlm_config.ci_protocol) {
1957 case DLM_PROTO_TCP:
1958 dlm_proto_ops = &dlm_tcp_ops;
1959 break;
1960 case DLM_PROTO_SCTP:
1961 dlm_proto_ops = &dlm_sctp_ops;
1962 break;
1963 default:
1964 log_print("Invalid protocol identifier %d set",
1965 dlm_config.ci_protocol);
1966 error = -EINVAL;
1967 goto fail_proto_ops;
1968 }
1969
1970 error = dlm_listen_for_all();
1971 if (error)
1972 goto fail_listen;
1973
1974 return 0;
1975
1976 fail_listen:
1977 dlm_proto_ops = NULL;
1978 fail_proto_ops:
1979 work_stop();
1980 fail:
1981 return error;
1982 }
1983
1984 void dlm_lowcomms_init(void)
1985 {
1986 int i;
1987
1988 for (i = 0; i < CONN_HASH_SIZE; i++)
1989 INIT_HLIST_HEAD(&connection_hash[i]);
1990
1991 INIT_WORK(&listen_con.rwork, process_listen_recv_socket);
1992 }
1993
1994 void dlm_lowcomms_exit(void)
1995 {
1996 struct connection *con;
1997 int i, idx;
1998
1999 idx = srcu_read_lock(&connections_srcu);
2000 for (i = 0; i < CONN_HASH_SIZE; i++) {
2001 hlist_for_each_entry_rcu(con, &connection_hash[i], list) {
2002 spin_lock(&connections_lock);
2003 hlist_del_rcu(&con->list);
2004 spin_unlock(&connections_lock);
2005
2006 if (con->othercon)
2007 call_srcu(&connections_srcu, &con->othercon->rcu,
2008 connection_release);
2009 call_srcu(&connections_srcu, &con->rcu, connection_release);
2010 }
2011 }
2012 srcu_read_unlock(&connections_srcu, idx);
2013 }
2014
Linux® is a registered trademark of Linus Torvalds in the United States and other countries.
TOMOYO® is a registered trademark of NTT DATA CORPORATION.