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