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