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