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