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 spin_lock_bh(&con->writequeue_lock); 739 set_bit(CF_IO_STOP, &con->flags); 740 spin_unlock_bh(&con->writequeue_lock); 741 742 down_write(&con->sock_lock); 743 if (con->sock) { 744 lock_sock(con->sock->sk); 745 restore_callbacks(con->sock->sk); 746 release_sock(con->sock->sk); 747 } 748 up_write(&con->sock_lock); 749 750 cancel_work_sync(&con->swork); 751 cancel_work_sync(&con->rwork); 752 } 753 754 /* Close a remote connection and tidy up */ 755 static void close_connection(struct connection *con, bool and_other) 756 { 757 struct writequeue_entry *e; 758 759 if (con->othercon && and_other) 760 close_connection(con->othercon, false); 761 762 down_write(&con->sock_lock); 763 if (!con->sock) { 764 up_write(&con->sock_lock); 765 return; 766 } 767 768 dlm_close_sock(&con->sock); 769 770 /* if we send a writequeue entry only a half way, we drop the 771 * whole entry because reconnection and that we not start of the 772 * middle of a msg which will confuse the other end. 773 * 774 * we can always drop messages because retransmits, but what we 775 * cannot allow is to transmit half messages which may be processed 776 * at the other side. 777 * 778 * our policy is to start on a clean state when disconnects, we don't 779 * know what's send/received on transport layer in this case. 780 */ 781 spin_lock_bh(&con->writequeue_lock); 782 if (!list_empty(&con->writequeue)) { 783 e = list_first_entry(&con->writequeue, struct writequeue_entry, 784 list); 785 if (e->dirty) 786 free_entry(e); 787 } 788 spin_unlock_bh(&con->writequeue_lock); 789 790 con->rx_leftover = 0; 791 con->retries = 0; 792 clear_bit(CF_APP_LIMITED, &con->flags); 793 clear_bit(CF_RECV_PENDING, &con->flags); 794 clear_bit(CF_SEND_PENDING, &con->flags); 795 up_write(&con->sock_lock); 796 } 797 798 static void shutdown_connection(struct connection *con, bool and_other) 799 { 800 int ret; 801 802 if (con->othercon && and_other) 803 shutdown_connection(con->othercon, false); 804 805 flush_workqueue(io_workqueue); 806 down_read(&con->sock_lock); 807 /* nothing to shutdown */ 808 if (!con->sock) { 809 up_read(&con->sock_lock); 810 return; 811 } 812 813 ret = kernel_sock_shutdown(con->sock, SHUT_WR); 814 up_read(&con->sock_lock); 815 if (ret) { 816 log_print("Connection %p failed to shutdown: %d will force close", 817 con, ret); 818 goto force_close; 819 } else { 820 ret = wait_event_timeout(con->shutdown_wait, !con->sock, 821 DLM_SHUTDOWN_WAIT_TIMEOUT); 822 if (ret == 0) { 823 log_print("Connection %p shutdown timed out, will force close", 824 con); 825 goto force_close; 826 } 827 } 828 829 return; 830 831 force_close: 832 close_connection(con, false); 833 } 834 835 static struct processqueue_entry *new_processqueue_entry(int nodeid, 836 int buflen) 837 { 838 struct processqueue_entry *pentry; 839 840 pentry = kmalloc(sizeof(*pentry), GFP_NOFS); 841 if (!pentry) 842 return NULL; 843 844 pentry->buf = kmalloc(buflen, GFP_NOFS); 845 if (!pentry->buf) { 846 kfree(pentry); 847 return NULL; 848 } 849 850 pentry->nodeid = nodeid; 851 return pentry; 852 } 853 854 static void free_processqueue_entry(struct processqueue_entry *pentry) 855 { 856 kfree(pentry->buf); 857 kfree(pentry); 858 } 859 860 struct dlm_processed_nodes { 861 int nodeid; 862 863 struct list_head list; 864 }; 865 866 static void add_processed_node(int nodeid, struct list_head *processed_nodes) 867 { 868 struct dlm_processed_nodes *n; 869 870 list_for_each_entry(n, processed_nodes, list) { 871 /* we already remembered this node */ 872 if (n->nodeid == nodeid) 873 return; 874 } 875 876 /* if it's fails in worst case we simple don't send an ack back. 877 * We try it next time. 878 */ 879 n = kmalloc(sizeof(*n), GFP_NOFS); 880 if (!n) 881 return; 882 883 n->nodeid = nodeid; 884 list_add(&n->list, processed_nodes); 885 } 886 887 static void process_dlm_messages(struct work_struct *work) 888 { 889 struct dlm_processed_nodes *n, *n_tmp; 890 struct processqueue_entry *pentry; 891 LIST_HEAD(processed_nodes); 892 893 spin_lock(&processqueue_lock); 894 pentry = list_first_entry_or_null(&processqueue, 895 struct processqueue_entry, list); 896 if (WARN_ON_ONCE(!pentry)) { 897 process_dlm_messages_pending = false; 898 spin_unlock(&processqueue_lock); 899 return; 900 } 901 902 list_del(&pentry->list); 903 spin_unlock(&processqueue_lock); 904 905 for (;;) { 906 dlm_process_incoming_buffer(pentry->nodeid, pentry->buf, 907 pentry->buflen); 908 add_processed_node(pentry->nodeid, &processed_nodes); 909 free_processqueue_entry(pentry); 910 911 spin_lock(&processqueue_lock); 912 pentry = list_first_entry_or_null(&processqueue, 913 struct processqueue_entry, list); 914 if (!pentry) { 915 process_dlm_messages_pending = false; 916 spin_unlock(&processqueue_lock); 917 break; 918 } 919 920 list_del(&pentry->list); 921 spin_unlock(&processqueue_lock); 922 } 923 924 /* send ack back after we processed couple of messages */ 925 list_for_each_entry_safe(n, n_tmp, &processed_nodes, list) { 926 list_del(&n->list); 927 dlm_midcomms_receive_done(n->nodeid); 928 kfree(n); 929 } 930 } 931 932 /* Data received from remote end */ 933 static int receive_from_sock(struct connection *con, int buflen) 934 { 935 struct processqueue_entry *pentry; 936 int ret, buflen_real; 937 struct msghdr msg; 938 struct kvec iov; 939 940 pentry = new_processqueue_entry(con->nodeid, buflen); 941 if (!pentry) 942 return DLM_IO_RESCHED; 943 944 memcpy(pentry->buf, con->rx_leftover_buf, con->rx_leftover); 945 946 /* calculate new buffer parameter regarding last receive and 947 * possible leftover bytes 948 */ 949 iov.iov_base = pentry->buf + con->rx_leftover; 950 iov.iov_len = buflen - con->rx_leftover; 951 952 memset(&msg, 0, sizeof(msg)); 953 msg.msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL; 954 clear_bit(CF_RECV_INTR, &con->flags); 955 again: 956 ret = kernel_recvmsg(con->sock, &msg, &iov, 1, iov.iov_len, 957 msg.msg_flags); 958 trace_dlm_recv(con->nodeid, ret); 959 if (ret == -EAGAIN) { 960 lock_sock(con->sock->sk); 961 if (test_and_clear_bit(CF_RECV_INTR, &con->flags)) { 962 release_sock(con->sock->sk); 963 goto again; 964 } 965 966 clear_bit(CF_RECV_PENDING, &con->flags); 967 release_sock(con->sock->sk); 968 free_processqueue_entry(pentry); 969 return DLM_IO_END; 970 } else if (ret == 0) { 971 /* close will clear CF_RECV_PENDING */ 972 free_processqueue_entry(pentry); 973 return DLM_IO_EOF; 974 } else if (ret < 0) { 975 free_processqueue_entry(pentry); 976 return ret; 977 } 978 979 /* new buflen according readed bytes and leftover from last receive */ 980 buflen_real = ret + con->rx_leftover; 981 ret = dlm_validate_incoming_buffer(con->nodeid, pentry->buf, 982 buflen_real); 983 if (ret < 0) { 984 free_processqueue_entry(pentry); 985 return ret; 986 } 987 988 pentry->buflen = ret; 989 990 /* calculate leftover bytes from process and put it into begin of 991 * the receive buffer, so next receive we have the full message 992 * at the start address of the receive buffer. 993 */ 994 con->rx_leftover = buflen_real - ret; 995 memmove(con->rx_leftover_buf, pentry->buf + ret, 996 con->rx_leftover); 997 998 spin_lock(&processqueue_lock); 999 list_add_tail(&pentry->list, &processqueue); 1000 if (!process_dlm_messages_pending) { 1001 process_dlm_messages_pending = true; 1002 queue_work(process_workqueue, &process_work); 1003 } 1004 spin_unlock(&processqueue_lock); 1005 1006 return DLM_IO_SUCCESS; 1007 } 1008 1009 /* Listening socket is busy, accept a connection */ 1010 static int accept_from_sock(void) 1011 { 1012 struct sockaddr_storage peeraddr; 1013 int len, idx, result, nodeid; 1014 struct connection *newcon; 1015 struct socket *newsock; 1016 unsigned int mark; 1017 1018 result = kernel_accept(listen_con.sock, &newsock, O_NONBLOCK); 1019 if (result == -EAGAIN) 1020 return DLM_IO_END; 1021 else if (result < 0) 1022 goto accept_err; 1023 1024 /* Get the connected socket's peer */ 1025 memset(&peeraddr, 0, sizeof(peeraddr)); 1026 len = newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 2); 1027 if (len < 0) { 1028 result = -ECONNABORTED; 1029 goto accept_err; 1030 } 1031 1032 /* Get the new node's NODEID */ 1033 make_sockaddr(&peeraddr, 0, &len); 1034 if (addr_to_nodeid(&peeraddr, &nodeid, &mark)) { 1035 switch (peeraddr.ss_family) { 1036 case AF_INET: { 1037 struct sockaddr_in *sin = (struct sockaddr_in *)&peeraddr; 1038 1039 log_print("connect from non cluster IPv4 node %pI4", 1040 &sin->sin_addr); 1041 break; 1042 } 1043 #if IS_ENABLED(CONFIG_IPV6) 1044 case AF_INET6: { 1045 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&peeraddr; 1046 1047 log_print("connect from non cluster IPv6 node %pI6c", 1048 &sin6->sin6_addr); 1049 break; 1050 } 1051 #endif 1052 default: 1053 log_print("invalid family from non cluster node"); 1054 break; 1055 } 1056 1057 sock_release(newsock); 1058 return -1; 1059 } 1060 1061 log_print("got connection from %d", nodeid); 1062 1063 /* Check to see if we already have a connection to this node. This 1064 * could happen if the two nodes initiate a connection at roughly 1065 * the same time and the connections cross on the wire. 1066 * In this case we store the incoming one in "othercon" 1067 */ 1068 idx = srcu_read_lock(&connections_srcu); 1069 newcon = nodeid2con(nodeid, 0); 1070 if (WARN_ON_ONCE(!newcon)) { 1071 srcu_read_unlock(&connections_srcu, idx); 1072 result = -ENOENT; 1073 goto accept_err; 1074 } 1075 1076 sock_set_mark(newsock->sk, mark); 1077 1078 down_write(&newcon->sock_lock); 1079 if (newcon->sock) { 1080 struct connection *othercon = newcon->othercon; 1081 1082 if (!othercon) { 1083 othercon = kzalloc(sizeof(*othercon), GFP_NOFS); 1084 if (!othercon) { 1085 log_print("failed to allocate incoming socket"); 1086 up_write(&newcon->sock_lock); 1087 srcu_read_unlock(&connections_srcu, idx); 1088 result = -ENOMEM; 1089 goto accept_err; 1090 } 1091 1092 dlm_con_init(othercon, nodeid); 1093 lockdep_set_subclass(&othercon->sock_lock, 1); 1094 newcon->othercon = othercon; 1095 set_bit(CF_IS_OTHERCON, &othercon->flags); 1096 } else { 1097 /* close other sock con if we have something new */ 1098 close_connection(othercon, false); 1099 } 1100 1101 down_write(&othercon->sock_lock); 1102 add_sock(newsock, othercon); 1103 1104 /* check if we receved something while adding */ 1105 lock_sock(othercon->sock->sk); 1106 lowcomms_queue_rwork(othercon); 1107 release_sock(othercon->sock->sk); 1108 up_write(&othercon->sock_lock); 1109 } 1110 else { 1111 /* accept copies the sk after we've saved the callbacks, so we 1112 don't want to save them a second time or comm errors will 1113 result in calling sk_error_report recursively. */ 1114 add_sock(newsock, newcon); 1115 1116 /* check if we receved something while adding */ 1117 lock_sock(newcon->sock->sk); 1118 lowcomms_queue_rwork(newcon); 1119 release_sock(newcon->sock->sk); 1120 } 1121 up_write(&newcon->sock_lock); 1122 srcu_read_unlock(&connections_srcu, idx); 1123 1124 return DLM_IO_SUCCESS; 1125 1126 accept_err: 1127 if (newsock) 1128 sock_release(newsock); 1129 1130 return result; 1131 } 1132 1133 /* 1134 * writequeue_entry_complete - try to delete and free write queue entry 1135 * @e: write queue entry to try to delete 1136 * @completed: bytes completed 1137 * 1138 * writequeue_lock must be held. 1139 */ 1140 static void writequeue_entry_complete(struct writequeue_entry *e, int completed) 1141 { 1142 e->offset += completed; 1143 e->len -= completed; 1144 /* signal that page was half way transmitted */ 1145 e->dirty = true; 1146 1147 if (e->len == 0 && e->users == 0) 1148 free_entry(e); 1149 } 1150 1151 /* 1152 * sctp_bind_addrs - bind a SCTP socket to all our addresses 1153 */ 1154 static int sctp_bind_addrs(struct socket *sock, uint16_t port) 1155 { 1156 struct sockaddr_storage localaddr; 1157 struct sockaddr *addr = (struct sockaddr *)&localaddr; 1158 int i, addr_len, result = 0; 1159 1160 for (i = 0; i < dlm_local_count; i++) { 1161 memcpy(&localaddr, &dlm_local_addr[i], sizeof(localaddr)); 1162 make_sockaddr(&localaddr, port, &addr_len); 1163 1164 if (!i) 1165 result = kernel_bind(sock, addr, addr_len); 1166 else 1167 result = sock_bind_add(sock->sk, addr, addr_len); 1168 1169 if (result < 0) { 1170 log_print("Can't bind to %d addr number %d, %d.\n", 1171 port, i + 1, result); 1172 break; 1173 } 1174 } 1175 return result; 1176 } 1177 1178 /* Get local addresses */ 1179 static void init_local(void) 1180 { 1181 struct sockaddr_storage sas; 1182 int i; 1183 1184 dlm_local_count = 0; 1185 for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) { 1186 if (dlm_our_addr(&sas, i)) 1187 break; 1188 1189 memcpy(&dlm_local_addr[dlm_local_count++], &sas, sizeof(sas)); 1190 } 1191 } 1192 1193 static struct writequeue_entry *new_writequeue_entry(struct connection *con) 1194 { 1195 struct writequeue_entry *entry; 1196 1197 entry = dlm_allocate_writequeue(); 1198 if (!entry) 1199 return NULL; 1200 1201 entry->page = alloc_page(GFP_ATOMIC | __GFP_ZERO); 1202 if (!entry->page) { 1203 dlm_free_writequeue(entry); 1204 return NULL; 1205 } 1206 1207 entry->offset = 0; 1208 entry->len = 0; 1209 entry->end = 0; 1210 entry->dirty = false; 1211 entry->con = con; 1212 entry->users = 1; 1213 kref_init(&entry->ref); 1214 return entry; 1215 } 1216 1217 static struct writequeue_entry *new_wq_entry(struct connection *con, int len, 1218 char **ppc, void (*cb)(void *data), 1219 void *data) 1220 { 1221 struct writequeue_entry *e; 1222 1223 spin_lock_bh(&con->writequeue_lock); 1224 if (!list_empty(&con->writequeue)) { 1225 e = list_last_entry(&con->writequeue, struct writequeue_entry, list); 1226 if (DLM_WQ_REMAIN_BYTES(e) >= len) { 1227 kref_get(&e->ref); 1228 1229 *ppc = page_address(e->page) + e->end; 1230 if (cb) 1231 cb(data); 1232 1233 e->end += len; 1234 e->users++; 1235 goto out; 1236 } 1237 } 1238 1239 e = new_writequeue_entry(con); 1240 if (!e) 1241 goto out; 1242 1243 kref_get(&e->ref); 1244 *ppc = page_address(e->page); 1245 e->end += len; 1246 if (cb) 1247 cb(data); 1248 1249 list_add_tail(&e->list, &con->writequeue); 1250 1251 out: 1252 spin_unlock_bh(&con->writequeue_lock); 1253 return e; 1254 }; 1255 1256 static struct dlm_msg *dlm_lowcomms_new_msg_con(struct connection *con, int len, 1257 gfp_t allocation, char **ppc, 1258 void (*cb)(void *data), 1259 void *data) 1260 { 1261 struct writequeue_entry *e; 1262 struct dlm_msg *msg; 1263 1264 msg = dlm_allocate_msg(allocation); 1265 if (!msg) 1266 return NULL; 1267 1268 kref_init(&msg->ref); 1269 1270 e = new_wq_entry(con, len, ppc, cb, data); 1271 if (!e) { 1272 dlm_free_msg(msg); 1273 return NULL; 1274 } 1275 1276 msg->retransmit = false; 1277 msg->orig_msg = NULL; 1278 msg->ppc = *ppc; 1279 msg->len = len; 1280 msg->entry = e; 1281 1282 return msg; 1283 } 1284 1285 /* avoid false positive for nodes_srcu, unlock happens in 1286 * dlm_lowcomms_commit_msg which is a must call if success 1287 */ 1288 #ifndef __CHECKER__ 1289 struct dlm_msg *dlm_lowcomms_new_msg(int nodeid, int len, gfp_t allocation, 1290 char **ppc, void (*cb)(void *data), 1291 void *data) 1292 { 1293 struct connection *con; 1294 struct dlm_msg *msg; 1295 int idx; 1296 1297 if (len > DLM_MAX_SOCKET_BUFSIZE || 1298 len < sizeof(struct dlm_header)) { 1299 BUILD_BUG_ON(PAGE_SIZE < DLM_MAX_SOCKET_BUFSIZE); 1300 log_print("failed to allocate a buffer of size %d", len); 1301 WARN_ON_ONCE(1); 1302 return NULL; 1303 } 1304 1305 idx = srcu_read_lock(&connections_srcu); 1306 con = nodeid2con(nodeid, 0); 1307 if (WARN_ON_ONCE(!con)) { 1308 srcu_read_unlock(&connections_srcu, idx); 1309 return NULL; 1310 } 1311 1312 msg = dlm_lowcomms_new_msg_con(con, len, allocation, ppc, cb, data); 1313 if (!msg) { 1314 srcu_read_unlock(&connections_srcu, idx); 1315 return NULL; 1316 } 1317 1318 /* for dlm_lowcomms_commit_msg() */ 1319 kref_get(&msg->ref); 1320 /* we assume if successful commit must called */ 1321 msg->idx = idx; 1322 return msg; 1323 } 1324 #endif 1325 1326 static void _dlm_lowcomms_commit_msg(struct dlm_msg *msg) 1327 { 1328 struct writequeue_entry *e = msg->entry; 1329 struct connection *con = e->con; 1330 int users; 1331 1332 spin_lock_bh(&con->writequeue_lock); 1333 kref_get(&msg->ref); 1334 list_add(&msg->list, &e->msgs); 1335 1336 users = --e->users; 1337 if (users) 1338 goto out; 1339 1340 e->len = DLM_WQ_LENGTH_BYTES(e); 1341 1342 lowcomms_queue_swork(con); 1343 1344 out: 1345 spin_unlock_bh(&con->writequeue_lock); 1346 return; 1347 } 1348 1349 /* avoid false positive for nodes_srcu, lock was happen in 1350 * dlm_lowcomms_new_msg 1351 */ 1352 #ifndef __CHECKER__ 1353 void dlm_lowcomms_commit_msg(struct dlm_msg *msg) 1354 { 1355 _dlm_lowcomms_commit_msg(msg); 1356 srcu_read_unlock(&connections_srcu, msg->idx); 1357 /* because dlm_lowcomms_new_msg() */ 1358 kref_put(&msg->ref, dlm_msg_release); 1359 } 1360 #endif 1361 1362 void dlm_lowcomms_put_msg(struct dlm_msg *msg) 1363 { 1364 kref_put(&msg->ref, dlm_msg_release); 1365 } 1366 1367 /* does not held connections_srcu, usage lowcomms_error_report only */ 1368 int dlm_lowcomms_resend_msg(struct dlm_msg *msg) 1369 { 1370 struct dlm_msg *msg_resend; 1371 char *ppc; 1372 1373 if (msg->retransmit) 1374 return 1; 1375 1376 msg_resend = dlm_lowcomms_new_msg_con(msg->entry->con, msg->len, 1377 GFP_ATOMIC, &ppc, NULL, NULL); 1378 if (!msg_resend) 1379 return -ENOMEM; 1380 1381 msg->retransmit = true; 1382 kref_get(&msg->ref); 1383 msg_resend->orig_msg = msg; 1384 1385 memcpy(ppc, msg->ppc, msg->len); 1386 _dlm_lowcomms_commit_msg(msg_resend); 1387 dlm_lowcomms_put_msg(msg_resend); 1388 1389 return 0; 1390 } 1391 1392 /* Send a message */ 1393 static int send_to_sock(struct connection *con) 1394 { 1395 const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL; 1396 struct writequeue_entry *e; 1397 int len, offset, ret; 1398 1399 spin_lock_bh(&con->writequeue_lock); 1400 e = con_next_wq(con); 1401 if (!e) { 1402 clear_bit(CF_SEND_PENDING, &con->flags); 1403 spin_unlock_bh(&con->writequeue_lock); 1404 return DLM_IO_END; 1405 } 1406 1407 len = e->len; 1408 offset = e->offset; 1409 WARN_ON_ONCE(len == 0 && e->users == 0); 1410 spin_unlock_bh(&con->writequeue_lock); 1411 1412 ret = kernel_sendpage(con->sock, e->page, offset, len, 1413 msg_flags); 1414 trace_dlm_send(con->nodeid, ret); 1415 if (ret == -EAGAIN || ret == 0) { 1416 lock_sock(con->sock->sk); 1417 spin_lock_bh(&con->writequeue_lock); 1418 if (test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) && 1419 !test_and_set_bit(CF_APP_LIMITED, &con->flags)) { 1420 /* Notify TCP that we're limited by the 1421 * application window size. 1422 */ 1423 set_bit(SOCK_NOSPACE, &con->sock->sk->sk_socket->flags); 1424 con->sock->sk->sk_write_pending++; 1425 1426 clear_bit(CF_SEND_PENDING, &con->flags); 1427 spin_unlock_bh(&con->writequeue_lock); 1428 release_sock(con->sock->sk); 1429 1430 /* wait for write_space() event */ 1431 return DLM_IO_END; 1432 } 1433 spin_unlock_bh(&con->writequeue_lock); 1434 release_sock(con->sock->sk); 1435 1436 return DLM_IO_RESCHED; 1437 } else if (ret < 0) { 1438 return ret; 1439 } 1440 1441 spin_lock_bh(&con->writequeue_lock); 1442 writequeue_entry_complete(e, ret); 1443 spin_unlock_bh(&con->writequeue_lock); 1444 1445 return DLM_IO_SUCCESS; 1446 } 1447 1448 static void clean_one_writequeue(struct connection *con) 1449 { 1450 struct writequeue_entry *e, *safe; 1451 1452 spin_lock_bh(&con->writequeue_lock); 1453 list_for_each_entry_safe(e, safe, &con->writequeue, list) { 1454 free_entry(e); 1455 } 1456 spin_unlock_bh(&con->writequeue_lock); 1457 } 1458 1459 static void connection_release(struct rcu_head *rcu) 1460 { 1461 struct connection *con = container_of(rcu, struct connection, rcu); 1462 1463 WARN_ON_ONCE(!list_empty(&con->writequeue)); 1464 WARN_ON_ONCE(con->sock); 1465 kfree(con); 1466 } 1467 1468 /* Called from recovery when it knows that a node has 1469 left the cluster */ 1470 int dlm_lowcomms_close(int nodeid) 1471 { 1472 struct connection *con; 1473 int idx; 1474 1475 log_print("closing connection to node %d", nodeid); 1476 1477 idx = srcu_read_lock(&connections_srcu); 1478 con = nodeid2con(nodeid, 0); 1479 if (WARN_ON_ONCE(!con)) { 1480 srcu_read_unlock(&connections_srcu, idx); 1481 return -ENOENT; 1482 } 1483 1484 stop_connection_io(con); 1485 log_print("io handling for node: %d stopped", nodeid); 1486 close_connection(con, true); 1487 1488 spin_lock(&connections_lock); 1489 hlist_del_rcu(&con->list); 1490 spin_unlock(&connections_lock); 1491 1492 clean_one_writequeue(con); 1493 call_srcu(&connections_srcu, &con->rcu, connection_release); 1494 if (con->othercon) { 1495 clean_one_writequeue(con->othercon); 1496 call_srcu(&connections_srcu, &con->othercon->rcu, connection_release); 1497 } 1498 srcu_read_unlock(&connections_srcu, idx); 1499 1500 /* for debugging we print when we are done to compare with other 1501 * messages in between. This function need to be correctly synchronized 1502 * with io handling 1503 */ 1504 log_print("closing connection to node %d done", nodeid); 1505 1506 return 0; 1507 } 1508 1509 /* Receive worker function */ 1510 static void process_recv_sockets(struct work_struct *work) 1511 { 1512 struct connection *con = container_of(work, struct connection, rwork); 1513 int ret, buflen; 1514 1515 down_read(&con->sock_lock); 1516 if (!con->sock) { 1517 up_read(&con->sock_lock); 1518 return; 1519 } 1520 1521 buflen = READ_ONCE(dlm_config.ci_buffer_size); 1522 do { 1523 ret = receive_from_sock(con, buflen); 1524 } while (ret == DLM_IO_SUCCESS); 1525 up_read(&con->sock_lock); 1526 1527 switch (ret) { 1528 case DLM_IO_END: 1529 /* CF_RECV_PENDING cleared */ 1530 break; 1531 case DLM_IO_EOF: 1532 close_connection(con, false); 1533 wake_up(&con->shutdown_wait); 1534 /* CF_RECV_PENDING cleared */ 1535 break; 1536 case DLM_IO_RESCHED: 1537 cond_resched(); 1538 queue_work(io_workqueue, &con->rwork); 1539 /* CF_RECV_PENDING not cleared */ 1540 break; 1541 default: 1542 if (ret < 0) { 1543 if (test_bit(CF_IS_OTHERCON, &con->flags)) { 1544 close_connection(con, false); 1545 } else { 1546 spin_lock_bh(&con->writequeue_lock); 1547 lowcomms_queue_swork(con); 1548 spin_unlock_bh(&con->writequeue_lock); 1549 } 1550 1551 /* CF_RECV_PENDING cleared for othercon 1552 * we trigger send queue if not already done 1553 * and process_send_sockets will handle it 1554 */ 1555 break; 1556 } 1557 1558 WARN_ON_ONCE(1); 1559 break; 1560 } 1561 } 1562 1563 static void process_listen_recv_socket(struct work_struct *work) 1564 { 1565 int ret; 1566 1567 if (WARN_ON_ONCE(!listen_con.sock)) 1568 return; 1569 1570 do { 1571 ret = accept_from_sock(); 1572 } while (ret == DLM_IO_SUCCESS); 1573 1574 if (ret < 0) 1575 log_print("critical error accepting connection: %d", ret); 1576 } 1577 1578 static int dlm_connect(struct connection *con) 1579 { 1580 struct sockaddr_storage addr; 1581 int result, addr_len; 1582 struct socket *sock; 1583 unsigned int mark; 1584 1585 memset(&addr, 0, sizeof(addr)); 1586 result = nodeid_to_addr(con->nodeid, &addr, NULL, 1587 dlm_proto_ops->try_new_addr, &mark); 1588 if (result < 0) { 1589 log_print("no address for nodeid %d", con->nodeid); 1590 return result; 1591 } 1592 1593 /* Create a socket to communicate with */ 1594 result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family, 1595 SOCK_STREAM, dlm_proto_ops->proto, &sock); 1596 if (result < 0) 1597 return result; 1598 1599 sock_set_mark(sock->sk, mark); 1600 dlm_proto_ops->sockopts(sock); 1601 1602 result = dlm_proto_ops->bind(sock); 1603 if (result < 0) { 1604 sock_release(sock); 1605 return result; 1606 } 1607 1608 add_sock(sock, con); 1609 1610 log_print_ratelimited("connecting to %d", con->nodeid); 1611 make_sockaddr(&addr, dlm_config.ci_tcp_port, &addr_len); 1612 result = dlm_proto_ops->connect(con, sock, (struct sockaddr *)&addr, 1613 addr_len); 1614 switch (result) { 1615 case -EINPROGRESS: 1616 /* not an error */ 1617 fallthrough; 1618 case 0: 1619 break; 1620 default: 1621 if (result < 0) 1622 dlm_close_sock(&con->sock); 1623 1624 break; 1625 } 1626 1627 return result; 1628 } 1629 1630 /* Send worker function */ 1631 static void process_send_sockets(struct work_struct *work) 1632 { 1633 struct connection *con = container_of(work, struct connection, swork); 1634 int ret; 1635 1636 WARN_ON_ONCE(test_bit(CF_IS_OTHERCON, &con->flags)); 1637 1638 down_read(&con->sock_lock); 1639 if (!con->sock) { 1640 up_read(&con->sock_lock); 1641 down_write(&con->sock_lock); 1642 if (!con->sock) { 1643 ret = dlm_connect(con); 1644 switch (ret) { 1645 case 0: 1646 break; 1647 case -EINPROGRESS: 1648 /* avoid spamming resched on connection 1649 * we might can switch to a state_change 1650 * event based mechanism if established 1651 */ 1652 msleep(100); 1653 break; 1654 default: 1655 /* CF_SEND_PENDING not cleared */ 1656 up_write(&con->sock_lock); 1657 log_print("connect to node %d try %d error %d", 1658 con->nodeid, con->retries++, ret); 1659 msleep(1000); 1660 /* For now we try forever to reconnect. In 1661 * future we should send a event to cluster 1662 * manager to fence itself after certain amount 1663 * of retries. 1664 */ 1665 queue_work(io_workqueue, &con->swork); 1666 return; 1667 } 1668 } 1669 downgrade_write(&con->sock_lock); 1670 } 1671 1672 do { 1673 ret = send_to_sock(con); 1674 } while (ret == DLM_IO_SUCCESS); 1675 up_read(&con->sock_lock); 1676 1677 switch (ret) { 1678 case DLM_IO_END: 1679 /* CF_SEND_PENDING cleared */ 1680 break; 1681 case DLM_IO_RESCHED: 1682 /* CF_SEND_PENDING not cleared */ 1683 cond_resched(); 1684 queue_work(io_workqueue, &con->swork); 1685 break; 1686 default: 1687 if (ret < 0) { 1688 close_connection(con, false); 1689 1690 /* CF_SEND_PENDING cleared */ 1691 spin_lock_bh(&con->writequeue_lock); 1692 lowcomms_queue_swork(con); 1693 spin_unlock_bh(&con->writequeue_lock); 1694 break; 1695 } 1696 1697 WARN_ON_ONCE(1); 1698 break; 1699 } 1700 } 1701 1702 static void work_stop(void) 1703 { 1704 if (io_workqueue) { 1705 destroy_workqueue(io_workqueue); 1706 io_workqueue = NULL; 1707 } 1708 1709 if (process_workqueue) { 1710 destroy_workqueue(process_workqueue); 1711 process_workqueue = NULL; 1712 } 1713 } 1714 1715 static int work_start(void) 1716 { 1717 io_workqueue = alloc_workqueue("dlm_io", WQ_HIGHPRI | WQ_MEM_RECLAIM | 1718 WQ_UNBOUND, 0); 1719 if (!io_workqueue) { 1720 log_print("can't start dlm_io"); 1721 return -ENOMEM; 1722 } 1723 1724 /* ordered dlm message process queue, 1725 * should be converted to a tasklet 1726 */ 1727 process_workqueue = alloc_ordered_workqueue("dlm_process", 1728 WQ_HIGHPRI | WQ_MEM_RECLAIM); 1729 if (!process_workqueue) { 1730 log_print("can't start dlm_process"); 1731 destroy_workqueue(io_workqueue); 1732 io_workqueue = NULL; 1733 return -ENOMEM; 1734 } 1735 1736 return 0; 1737 } 1738 1739 void dlm_lowcomms_shutdown(void) 1740 { 1741 struct connection *con; 1742 int i, idx; 1743 1744 /* stop lowcomms_listen_data_ready calls */ 1745 lock_sock(listen_con.sock->sk); 1746 listen_con.sock->sk->sk_data_ready = listen_sock.sk_data_ready; 1747 release_sock(listen_con.sock->sk); 1748 1749 cancel_work_sync(&listen_con.rwork); 1750 dlm_close_sock(&listen_con.sock); 1751 1752 idx = srcu_read_lock(&connections_srcu); 1753 for (i = 0; i < CONN_HASH_SIZE; i++) { 1754 hlist_for_each_entry_rcu(con, &connection_hash[i], list) { 1755 shutdown_connection(con, true); 1756 stop_connection_io(con); 1757 flush_workqueue(process_workqueue); 1758 close_connection(con, true); 1759 1760 clean_one_writequeue(con); 1761 if (con->othercon) 1762 clean_one_writequeue(con->othercon); 1763 allow_connection_io(con); 1764 } 1765 } 1766 srcu_read_unlock(&connections_srcu, idx); 1767 } 1768 1769 void dlm_lowcomms_stop(void) 1770 { 1771 work_stop(); 1772 dlm_proto_ops = NULL; 1773 } 1774 1775 static int dlm_listen_for_all(void) 1776 { 1777 struct socket *sock; 1778 int result; 1779 1780 log_print("Using %s for communications", 1781 dlm_proto_ops->name); 1782 1783 result = dlm_proto_ops->listen_validate(); 1784 if (result < 0) 1785 return result; 1786 1787 result = sock_create_kern(&init_net, dlm_local_addr[0].ss_family, 1788 SOCK_STREAM, dlm_proto_ops->proto, &sock); 1789 if (result < 0) { 1790 log_print("Can't create comms socket: %d", result); 1791 return result; 1792 } 1793 1794 sock_set_mark(sock->sk, dlm_config.ci_mark); 1795 dlm_proto_ops->listen_sockopts(sock); 1796 1797 result = dlm_proto_ops->listen_bind(sock); 1798 if (result < 0) 1799 goto out; 1800 1801 lock_sock(sock->sk); 1802 listen_sock.sk_data_ready = sock->sk->sk_data_ready; 1803 listen_sock.sk_write_space = sock->sk->sk_write_space; 1804 listen_sock.sk_error_report = sock->sk->sk_error_report; 1805 listen_sock.sk_state_change = sock->sk->sk_state_change; 1806 1807 listen_con.sock = sock; 1808 1809 sock->sk->sk_allocation = GFP_NOFS; 1810 sock->sk->sk_use_task_frag = false; 1811 sock->sk->sk_data_ready = lowcomms_listen_data_ready; 1812 release_sock(sock->sk); 1813 1814 result = sock->ops->listen(sock, 128); 1815 if (result < 0) { 1816 dlm_close_sock(&listen_con.sock); 1817 return result; 1818 } 1819 1820 return 0; 1821 1822 out: 1823 sock_release(sock); 1824 return result; 1825 } 1826 1827 static int dlm_tcp_bind(struct socket *sock) 1828 { 1829 struct sockaddr_storage src_addr; 1830 int result, addr_len; 1831 1832 /* Bind to our cluster-known address connecting to avoid 1833 * routing problems. 1834 */ 1835 memcpy(&src_addr, &dlm_local_addr[0], sizeof(src_addr)); 1836 make_sockaddr(&src_addr, 0, &addr_len); 1837 1838 result = sock->ops->bind(sock, (struct sockaddr *)&src_addr, 1839 addr_len); 1840 if (result < 0) { 1841 /* This *may* not indicate a critical error */ 1842 log_print("could not bind for connect: %d", result); 1843 } 1844 1845 return 0; 1846 } 1847 1848 static int dlm_tcp_connect(struct connection *con, struct socket *sock, 1849 struct sockaddr *addr, int addr_len) 1850 { 1851 return sock->ops->connect(sock, addr, addr_len, O_NONBLOCK); 1852 } 1853 1854 static int dlm_tcp_listen_validate(void) 1855 { 1856 /* We don't support multi-homed hosts */ 1857 if (dlm_local_count > 1) { 1858 log_print("TCP protocol can't handle multi-homed hosts, try SCTP"); 1859 return -EINVAL; 1860 } 1861 1862 return 0; 1863 } 1864 1865 static void dlm_tcp_sockopts(struct socket *sock) 1866 { 1867 /* Turn off Nagle's algorithm */ 1868 tcp_sock_set_nodelay(sock->sk); 1869 } 1870 1871 static void dlm_tcp_listen_sockopts(struct socket *sock) 1872 { 1873 dlm_tcp_sockopts(sock); 1874 sock_set_reuseaddr(sock->sk); 1875 } 1876 1877 static int dlm_tcp_listen_bind(struct socket *sock) 1878 { 1879 int addr_len; 1880 1881 /* Bind to our port */ 1882 make_sockaddr(&dlm_local_addr[0], dlm_config.ci_tcp_port, &addr_len); 1883 return sock->ops->bind(sock, (struct sockaddr *)&dlm_local_addr[0], 1884 addr_len); 1885 } 1886 1887 static const struct dlm_proto_ops dlm_tcp_ops = { 1888 .name = "TCP", 1889 .proto = IPPROTO_TCP, 1890 .connect = dlm_tcp_connect, 1891 .sockopts = dlm_tcp_sockopts, 1892 .bind = dlm_tcp_bind, 1893 .listen_validate = dlm_tcp_listen_validate, 1894 .listen_sockopts = dlm_tcp_listen_sockopts, 1895 .listen_bind = dlm_tcp_listen_bind, 1896 }; 1897 1898 static int dlm_sctp_bind(struct socket *sock) 1899 { 1900 return sctp_bind_addrs(sock, 0); 1901 } 1902 1903 static int dlm_sctp_connect(struct connection *con, struct socket *sock, 1904 struct sockaddr *addr, int addr_len) 1905 { 1906 int ret; 1907 1908 /* 1909 * Make sock->ops->connect() function return in specified time, 1910 * since O_NONBLOCK argument in connect() function does not work here, 1911 * then, we should restore the default value of this attribute. 1912 */ 1913 sock_set_sndtimeo(sock->sk, 5); 1914 ret = sock->ops->connect(sock, addr, addr_len, 0); 1915 sock_set_sndtimeo(sock->sk, 0); 1916 return ret; 1917 } 1918 1919 static int dlm_sctp_listen_validate(void) 1920 { 1921 if (!IS_ENABLED(CONFIG_IP_SCTP)) { 1922 log_print("SCTP is not enabled by this kernel"); 1923 return -EOPNOTSUPP; 1924 } 1925 1926 request_module("sctp"); 1927 return 0; 1928 } 1929 1930 static int dlm_sctp_bind_listen(struct socket *sock) 1931 { 1932 return sctp_bind_addrs(sock, dlm_config.ci_tcp_port); 1933 } 1934 1935 static void dlm_sctp_sockopts(struct socket *sock) 1936 { 1937 /* Turn off Nagle's algorithm */ 1938 sctp_sock_set_nodelay(sock->sk); 1939 sock_set_rcvbuf(sock->sk, NEEDED_RMEM); 1940 } 1941 1942 static const struct dlm_proto_ops dlm_sctp_ops = { 1943 .name = "SCTP", 1944 .proto = IPPROTO_SCTP, 1945 .try_new_addr = true, 1946 .connect = dlm_sctp_connect, 1947 .sockopts = dlm_sctp_sockopts, 1948 .bind = dlm_sctp_bind, 1949 .listen_validate = dlm_sctp_listen_validate, 1950 .listen_sockopts = dlm_sctp_sockopts, 1951 .listen_bind = dlm_sctp_bind_listen, 1952 }; 1953 1954 int dlm_lowcomms_start(void) 1955 { 1956 int error; 1957 1958 init_local(); 1959 if (!dlm_local_count) { 1960 error = -ENOTCONN; 1961 log_print("no local IP address has been set"); 1962 goto fail; 1963 } 1964 1965 error = work_start(); 1966 if (error) 1967 goto fail; 1968 1969 /* Start listening */ 1970 switch (dlm_config.ci_protocol) { 1971 case DLM_PROTO_TCP: 1972 dlm_proto_ops = &dlm_tcp_ops; 1973 break; 1974 case DLM_PROTO_SCTP: 1975 dlm_proto_ops = &dlm_sctp_ops; 1976 break; 1977 default: 1978 log_print("Invalid protocol identifier %d set", 1979 dlm_config.ci_protocol); 1980 error = -EINVAL; 1981 goto fail_proto_ops; 1982 } 1983 1984 error = dlm_listen_for_all(); 1985 if (error) 1986 goto fail_listen; 1987 1988 return 0; 1989 1990 fail_listen: 1991 dlm_proto_ops = NULL; 1992 fail_proto_ops: 1993 work_stop(); 1994 fail: 1995 return error; 1996 } 1997 1998 void dlm_lowcomms_init(void) 1999 { 2000 int i; 2001 2002 for (i = 0; i < CONN_HASH_SIZE; i++) 2003 INIT_HLIST_HEAD(&connection_hash[i]); 2004 2005 INIT_WORK(&listen_con.rwork, process_listen_recv_socket); 2006 } 2007 2008 void dlm_lowcomms_exit(void) 2009 { 2010 struct connection *con; 2011 int i, idx; 2012 2013 idx = srcu_read_lock(&connections_srcu); 2014 for (i = 0; i < CONN_HASH_SIZE; i++) { 2015 hlist_for_each_entry_rcu(con, &connection_hash[i], list) { 2016 spin_lock(&connections_lock); 2017 hlist_del_rcu(&con->list); 2018 spin_unlock(&connections_lock); 2019 2020 if (con->othercon) 2021 call_srcu(&connections_srcu, &con->othercon->rcu, 2022 connection_release); 2023 call_srcu(&connections_srcu, &con->rcu, connection_release); 2024 } 2025 } 2026 srcu_read_unlock(&connections_srcu, idx); 2027 } 2028