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