1 /****************************************************************************** 2 ******************************************************************************* 3 ** 4 ** Copyright (C) Sistina Software, Inc. 1997-2003 All rights reserved. 5 ** Copyright (C) 2004-2009 Red Hat, Inc. All rights reserved. 6 ** 7 ** This copyrighted material is made available to anyone wishing to use, 8 ** modify, copy, or redistribute it subject to the terms and conditions 9 ** of the GNU General Public License v.2. 10 ** 11 ******************************************************************************* 12 ******************************************************************************/ 13 14 /* 15 * lowcomms.c 16 * 17 * This is the "low-level" comms layer. 18 * 19 * It is responsible for sending/receiving messages 20 * from other nodes in the cluster. 21 * 22 * Cluster nodes are referred to by their nodeids. nodeids are 23 * simply 32 bit numbers to the locking module - if they need to 24 * be expanded for the cluster infrastructure then that is its 25 * responsibility. It is this layer's 26 * responsibility to resolve these into IP address or 27 * whatever it needs for inter-node communication. 28 * 29 * The comms level is two kernel threads that deal mainly with 30 * the receiving of messages from other nodes and passing them 31 * up to the mid-level comms layer (which understands the 32 * message format) for execution by the locking core, and 33 * a send thread which does all the setting up of connections 34 * to remote nodes and the sending of data. Threads are not allowed 35 * to send their own data because it may cause them to wait in times 36 * of high load. Also, this way, the sending thread can collect together 37 * messages bound for one node and send them in one block. 38 * 39 * lowcomms will choose to use either TCP or SCTP as its transport layer 40 * depending on the configuration variable 'protocol'. This should be set 41 * to 0 (default) for TCP or 1 for SCTP. It should be configured using a 42 * cluster-wide mechanism as it must be the same on all nodes of the cluster 43 * for the DLM to function. 44 * 45 */ 46 47 #include <asm/ioctls.h> 48 #include <net/sock.h> 49 #include <net/tcp.h> 50 #include <linux/pagemap.h> 51 #include <linux/file.h> 52 #include <linux/mutex.h> 53 #include <linux/sctp.h> 54 #include <linux/slab.h> 55 #include <net/sctp/sctp.h> 56 #include <net/ipv6.h> 57 58 #include "dlm_internal.h" 59 #include "lowcomms.h" 60 #include "midcomms.h" 61 #include "config.h" 62 63 #define NEEDED_RMEM (4*1024*1024) 64 #define CONN_HASH_SIZE 32 65 66 /* Number of messages to send before rescheduling */ 67 #define MAX_SEND_MSG_COUNT 25 68 69 struct cbuf { 70 unsigned int base; 71 unsigned int len; 72 unsigned int mask; 73 }; 74 75 static void cbuf_add(struct cbuf *cb, int n) 76 { 77 cb->len += n; 78 } 79 80 static int cbuf_data(struct cbuf *cb) 81 { 82 return ((cb->base + cb->len) & cb->mask); 83 } 84 85 static void cbuf_init(struct cbuf *cb, int size) 86 { 87 cb->base = cb->len = 0; 88 cb->mask = size-1; 89 } 90 91 static void cbuf_eat(struct cbuf *cb, int n) 92 { 93 cb->len -= n; 94 cb->base += n; 95 cb->base &= cb->mask; 96 } 97 98 static bool cbuf_empty(struct cbuf *cb) 99 { 100 return cb->len == 0; 101 } 102 103 struct connection { 104 struct socket *sock; /* NULL if not connected */ 105 uint32_t nodeid; /* So we know who we are in the list */ 106 struct mutex sock_mutex; 107 unsigned long flags; 108 #define CF_READ_PENDING 1 109 #define CF_INIT_PENDING 4 110 #define CF_IS_OTHERCON 5 111 #define CF_CLOSE 6 112 #define CF_APP_LIMITED 7 113 #define CF_CLOSING 8 114 struct list_head writequeue; /* List of outgoing writequeue_entries */ 115 spinlock_t writequeue_lock; 116 int (*rx_action) (struct connection *); /* What to do when active */ 117 void (*connect_action) (struct connection *); /* What to do to connect */ 118 struct page *rx_page; 119 struct cbuf cb; 120 int retries; 121 #define MAX_CONNECT_RETRIES 3 122 struct hlist_node list; 123 struct connection *othercon; 124 struct work_struct rwork; /* Receive workqueue */ 125 struct work_struct swork; /* Send workqueue */ 126 }; 127 #define sock2con(x) ((struct connection *)(x)->sk_user_data) 128 129 /* An entry waiting to be sent */ 130 struct writequeue_entry { 131 struct list_head list; 132 struct page *page; 133 int offset; 134 int len; 135 int end; 136 int users; 137 struct connection *con; 138 }; 139 140 struct dlm_node_addr { 141 struct list_head list; 142 int nodeid; 143 int addr_count; 144 int curr_addr_index; 145 struct sockaddr_storage *addr[DLM_MAX_ADDR_COUNT]; 146 }; 147 148 static struct listen_sock_callbacks { 149 void (*sk_error_report)(struct sock *); 150 void (*sk_data_ready)(struct sock *); 151 void (*sk_state_change)(struct sock *); 152 void (*sk_write_space)(struct sock *); 153 } listen_sock; 154 155 static LIST_HEAD(dlm_node_addrs); 156 static DEFINE_SPINLOCK(dlm_node_addrs_spin); 157 158 static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT]; 159 static int dlm_local_count; 160 static int dlm_allow_conn; 161 162 /* Work queues */ 163 static struct workqueue_struct *recv_workqueue; 164 static struct workqueue_struct *send_workqueue; 165 166 static struct hlist_head connection_hash[CONN_HASH_SIZE]; 167 static DEFINE_MUTEX(connections_lock); 168 static struct kmem_cache *con_cache; 169 170 static void process_recv_sockets(struct work_struct *work); 171 static void process_send_sockets(struct work_struct *work); 172 173 174 /* This is deliberately very simple because most clusters have simple 175 sequential nodeids, so we should be able to go straight to a connection 176 struct in the array */ 177 static inline int nodeid_hash(int nodeid) 178 { 179 return nodeid & (CONN_HASH_SIZE-1); 180 } 181 182 static struct connection *__find_con(int nodeid) 183 { 184 int r; 185 struct connection *con; 186 187 r = nodeid_hash(nodeid); 188 189 hlist_for_each_entry(con, &connection_hash[r], list) { 190 if (con->nodeid == nodeid) 191 return con; 192 } 193 return NULL; 194 } 195 196 /* 197 * If 'allocation' is zero then we don't attempt to create a new 198 * connection structure for this node. 199 */ 200 static struct connection *__nodeid2con(int nodeid, gfp_t alloc) 201 { 202 struct connection *con = NULL; 203 int r; 204 205 con = __find_con(nodeid); 206 if (con || !alloc) 207 return con; 208 209 con = kmem_cache_zalloc(con_cache, alloc); 210 if (!con) 211 return NULL; 212 213 r = nodeid_hash(nodeid); 214 hlist_add_head(&con->list, &connection_hash[r]); 215 216 con->nodeid = nodeid; 217 mutex_init(&con->sock_mutex); 218 INIT_LIST_HEAD(&con->writequeue); 219 spin_lock_init(&con->writequeue_lock); 220 INIT_WORK(&con->swork, process_send_sockets); 221 INIT_WORK(&con->rwork, process_recv_sockets); 222 223 /* Setup action pointers for child sockets */ 224 if (con->nodeid) { 225 struct connection *zerocon = __find_con(0); 226 227 con->connect_action = zerocon->connect_action; 228 if (!con->rx_action) 229 con->rx_action = zerocon->rx_action; 230 } 231 232 return con; 233 } 234 235 /* Loop round all connections */ 236 static void foreach_conn(void (*conn_func)(struct connection *c)) 237 { 238 int i; 239 struct hlist_node *n; 240 struct connection *con; 241 242 for (i = 0; i < CONN_HASH_SIZE; i++) { 243 hlist_for_each_entry_safe(con, n, &connection_hash[i], list) 244 conn_func(con); 245 } 246 } 247 248 static struct connection *nodeid2con(int nodeid, gfp_t allocation) 249 { 250 struct connection *con; 251 252 mutex_lock(&connections_lock); 253 con = __nodeid2con(nodeid, allocation); 254 mutex_unlock(&connections_lock); 255 256 return con; 257 } 258 259 static struct dlm_node_addr *find_node_addr(int nodeid) 260 { 261 struct dlm_node_addr *na; 262 263 list_for_each_entry(na, &dlm_node_addrs, list) { 264 if (na->nodeid == nodeid) 265 return na; 266 } 267 return NULL; 268 } 269 270 static int addr_compare(struct sockaddr_storage *x, struct sockaddr_storage *y) 271 { 272 switch (x->ss_family) { 273 case AF_INET: { 274 struct sockaddr_in *sinx = (struct sockaddr_in *)x; 275 struct sockaddr_in *siny = (struct sockaddr_in *)y; 276 if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr) 277 return 0; 278 if (sinx->sin_port != siny->sin_port) 279 return 0; 280 break; 281 } 282 case AF_INET6: { 283 struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x; 284 struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y; 285 if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr)) 286 return 0; 287 if (sinx->sin6_port != siny->sin6_port) 288 return 0; 289 break; 290 } 291 default: 292 return 0; 293 } 294 return 1; 295 } 296 297 static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out, 298 struct sockaddr *sa_out, bool try_new_addr) 299 { 300 struct sockaddr_storage sas; 301 struct dlm_node_addr *na; 302 303 if (!dlm_local_count) 304 return -1; 305 306 spin_lock(&dlm_node_addrs_spin); 307 na = find_node_addr(nodeid); 308 if (na && na->addr_count) { 309 memcpy(&sas, na->addr[na->curr_addr_index], 310 sizeof(struct sockaddr_storage)); 311 312 if (try_new_addr) { 313 na->curr_addr_index++; 314 if (na->curr_addr_index == na->addr_count) 315 na->curr_addr_index = 0; 316 } 317 } 318 spin_unlock(&dlm_node_addrs_spin); 319 320 if (!na) 321 return -EEXIST; 322 323 if (!na->addr_count) 324 return -ENOENT; 325 326 if (sas_out) 327 memcpy(sas_out, &sas, sizeof(struct sockaddr_storage)); 328 329 if (!sa_out) 330 return 0; 331 332 if (dlm_local_addr[0]->ss_family == AF_INET) { 333 struct sockaddr_in *in4 = (struct sockaddr_in *) &sas; 334 struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out; 335 ret4->sin_addr.s_addr = in4->sin_addr.s_addr; 336 } else { 337 struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &sas; 338 struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out; 339 ret6->sin6_addr = in6->sin6_addr; 340 } 341 342 return 0; 343 } 344 345 static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid) 346 { 347 struct dlm_node_addr *na; 348 int rv = -EEXIST; 349 int addr_i; 350 351 spin_lock(&dlm_node_addrs_spin); 352 list_for_each_entry(na, &dlm_node_addrs, list) { 353 if (!na->addr_count) 354 continue; 355 356 for (addr_i = 0; addr_i < na->addr_count; addr_i++) { 357 if (addr_compare(na->addr[addr_i], addr)) { 358 *nodeid = na->nodeid; 359 rv = 0; 360 goto unlock; 361 } 362 } 363 } 364 unlock: 365 spin_unlock(&dlm_node_addrs_spin); 366 return rv; 367 } 368 369 int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len) 370 { 371 struct sockaddr_storage *new_addr; 372 struct dlm_node_addr *new_node, *na; 373 374 new_node = kzalloc(sizeof(struct dlm_node_addr), GFP_NOFS); 375 if (!new_node) 376 return -ENOMEM; 377 378 new_addr = kzalloc(sizeof(struct sockaddr_storage), GFP_NOFS); 379 if (!new_addr) { 380 kfree(new_node); 381 return -ENOMEM; 382 } 383 384 memcpy(new_addr, addr, len); 385 386 spin_lock(&dlm_node_addrs_spin); 387 na = find_node_addr(nodeid); 388 if (!na) { 389 new_node->nodeid = nodeid; 390 new_node->addr[0] = new_addr; 391 new_node->addr_count = 1; 392 list_add(&new_node->list, &dlm_node_addrs); 393 spin_unlock(&dlm_node_addrs_spin); 394 return 0; 395 } 396 397 if (na->addr_count >= DLM_MAX_ADDR_COUNT) { 398 spin_unlock(&dlm_node_addrs_spin); 399 kfree(new_addr); 400 kfree(new_node); 401 return -ENOSPC; 402 } 403 404 na->addr[na->addr_count++] = new_addr; 405 spin_unlock(&dlm_node_addrs_spin); 406 kfree(new_node); 407 return 0; 408 } 409 410 /* Data available on socket or listen socket received a connect */ 411 static void lowcomms_data_ready(struct sock *sk) 412 { 413 struct connection *con = sock2con(sk); 414 if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags)) 415 queue_work(recv_workqueue, &con->rwork); 416 } 417 418 static void lowcomms_write_space(struct sock *sk) 419 { 420 struct connection *con = sock2con(sk); 421 422 if (!con) 423 return; 424 425 clear_bit(SOCK_NOSPACE, &con->sock->flags); 426 427 if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) { 428 con->sock->sk->sk_write_pending--; 429 clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags); 430 } 431 432 queue_work(send_workqueue, &con->swork); 433 } 434 435 static inline void lowcomms_connect_sock(struct connection *con) 436 { 437 if (test_bit(CF_CLOSE, &con->flags)) 438 return; 439 queue_work(send_workqueue, &con->swork); 440 cond_resched(); 441 } 442 443 static void lowcomms_state_change(struct sock *sk) 444 { 445 /* SCTP layer is not calling sk_data_ready when the connection 446 * is done, so we catch the signal through here. Also, it 447 * doesn't switch socket state when entering shutdown, so we 448 * skip the write in that case. 449 */ 450 if (sk->sk_shutdown) { 451 if (sk->sk_shutdown == RCV_SHUTDOWN) 452 lowcomms_data_ready(sk); 453 } else if (sk->sk_state == TCP_ESTABLISHED) { 454 lowcomms_write_space(sk); 455 } 456 } 457 458 int dlm_lowcomms_connect_node(int nodeid) 459 { 460 struct connection *con; 461 462 if (nodeid == dlm_our_nodeid()) 463 return 0; 464 465 con = nodeid2con(nodeid, GFP_NOFS); 466 if (!con) 467 return -ENOMEM; 468 lowcomms_connect_sock(con); 469 return 0; 470 } 471 472 static void lowcomms_error_report(struct sock *sk) 473 { 474 struct connection *con; 475 struct sockaddr_storage saddr; 476 int buflen; 477 void (*orig_report)(struct sock *) = NULL; 478 479 read_lock_bh(&sk->sk_callback_lock); 480 con = sock2con(sk); 481 if (con == NULL) 482 goto out; 483 484 orig_report = listen_sock.sk_error_report; 485 if (con->sock == NULL || 486 kernel_getpeername(con->sock, (struct sockaddr *)&saddr, &buflen)) { 487 printk_ratelimited(KERN_ERR "dlm: node %d: socket error " 488 "sending to node %d, port %d, " 489 "sk_err=%d/%d\n", dlm_our_nodeid(), 490 con->nodeid, dlm_config.ci_tcp_port, 491 sk->sk_err, sk->sk_err_soft); 492 } else if (saddr.ss_family == AF_INET) { 493 struct sockaddr_in *sin4 = (struct sockaddr_in *)&saddr; 494 495 printk_ratelimited(KERN_ERR "dlm: node %d: socket error " 496 "sending to node %d at %pI4, port %d, " 497 "sk_err=%d/%d\n", dlm_our_nodeid(), 498 con->nodeid, &sin4->sin_addr.s_addr, 499 dlm_config.ci_tcp_port, sk->sk_err, 500 sk->sk_err_soft); 501 } else { 502 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&saddr; 503 504 printk_ratelimited(KERN_ERR "dlm: node %d: socket error " 505 "sending to node %d at %u.%u.%u.%u, " 506 "port %d, sk_err=%d/%d\n", dlm_our_nodeid(), 507 con->nodeid, sin6->sin6_addr.s6_addr32[0], 508 sin6->sin6_addr.s6_addr32[1], 509 sin6->sin6_addr.s6_addr32[2], 510 sin6->sin6_addr.s6_addr32[3], 511 dlm_config.ci_tcp_port, sk->sk_err, 512 sk->sk_err_soft); 513 } 514 out: 515 read_unlock_bh(&sk->sk_callback_lock); 516 if (orig_report) 517 orig_report(sk); 518 } 519 520 /* Note: sk_callback_lock must be locked before calling this function. */ 521 static void save_listen_callbacks(struct socket *sock) 522 { 523 struct sock *sk = sock->sk; 524 525 listen_sock.sk_data_ready = sk->sk_data_ready; 526 listen_sock.sk_state_change = sk->sk_state_change; 527 listen_sock.sk_write_space = sk->sk_write_space; 528 listen_sock.sk_error_report = sk->sk_error_report; 529 } 530 531 static void restore_callbacks(struct socket *sock) 532 { 533 struct sock *sk = sock->sk; 534 535 write_lock_bh(&sk->sk_callback_lock); 536 sk->sk_user_data = NULL; 537 sk->sk_data_ready = listen_sock.sk_data_ready; 538 sk->sk_state_change = listen_sock.sk_state_change; 539 sk->sk_write_space = listen_sock.sk_write_space; 540 sk->sk_error_report = listen_sock.sk_error_report; 541 write_unlock_bh(&sk->sk_callback_lock); 542 } 543 544 /* Make a socket active */ 545 static void add_sock(struct socket *sock, struct connection *con) 546 { 547 struct sock *sk = sock->sk; 548 549 write_lock_bh(&sk->sk_callback_lock); 550 con->sock = sock; 551 552 sk->sk_user_data = con; 553 /* Install a data_ready callback */ 554 sk->sk_data_ready = lowcomms_data_ready; 555 sk->sk_write_space = lowcomms_write_space; 556 sk->sk_state_change = lowcomms_state_change; 557 sk->sk_allocation = GFP_NOFS; 558 sk->sk_error_report = lowcomms_error_report; 559 write_unlock_bh(&sk->sk_callback_lock); 560 } 561 562 /* Add the port number to an IPv6 or 4 sockaddr and return the address 563 length */ 564 static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port, 565 int *addr_len) 566 { 567 saddr->ss_family = dlm_local_addr[0]->ss_family; 568 if (saddr->ss_family == AF_INET) { 569 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr; 570 in4_addr->sin_port = cpu_to_be16(port); 571 *addr_len = sizeof(struct sockaddr_in); 572 memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero)); 573 } else { 574 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr; 575 in6_addr->sin6_port = cpu_to_be16(port); 576 *addr_len = sizeof(struct sockaddr_in6); 577 } 578 memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len); 579 } 580 581 /* Close a remote connection and tidy up */ 582 static void close_connection(struct connection *con, bool and_other, 583 bool tx, bool rx) 584 { 585 bool closing = test_and_set_bit(CF_CLOSING, &con->flags); 586 587 if (tx && !closing && cancel_work_sync(&con->swork)) 588 log_print("canceled swork for node %d", con->nodeid); 589 if (rx && !closing && cancel_work_sync(&con->rwork)) 590 log_print("canceled rwork for node %d", con->nodeid); 591 592 mutex_lock(&con->sock_mutex); 593 if (con->sock) { 594 restore_callbacks(con->sock); 595 sock_release(con->sock); 596 con->sock = NULL; 597 } 598 if (con->othercon && and_other) { 599 /* Will only re-enter once. */ 600 close_connection(con->othercon, false, true, true); 601 } 602 if (con->rx_page) { 603 __free_page(con->rx_page); 604 con->rx_page = NULL; 605 } 606 607 con->retries = 0; 608 mutex_unlock(&con->sock_mutex); 609 clear_bit(CF_CLOSING, &con->flags); 610 } 611 612 /* Data received from remote end */ 613 static int receive_from_sock(struct connection *con) 614 { 615 int ret = 0; 616 struct msghdr msg = {}; 617 struct kvec iov[2]; 618 unsigned len; 619 int r; 620 int call_again_soon = 0; 621 int nvec; 622 623 mutex_lock(&con->sock_mutex); 624 625 if (con->sock == NULL) { 626 ret = -EAGAIN; 627 goto out_close; 628 } 629 if (con->nodeid == 0) { 630 ret = -EINVAL; 631 goto out_close; 632 } 633 634 if (con->rx_page == NULL) { 635 /* 636 * This doesn't need to be atomic, but I think it should 637 * improve performance if it is. 638 */ 639 con->rx_page = alloc_page(GFP_ATOMIC); 640 if (con->rx_page == NULL) 641 goto out_resched; 642 cbuf_init(&con->cb, PAGE_SIZE); 643 } 644 645 /* 646 * iov[0] is the bit of the circular buffer between the current end 647 * point (cb.base + cb.len) and the end of the buffer. 648 */ 649 iov[0].iov_len = con->cb.base - cbuf_data(&con->cb); 650 iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb); 651 iov[1].iov_len = 0; 652 nvec = 1; 653 654 /* 655 * iov[1] is the bit of the circular buffer between the start of the 656 * buffer and the start of the currently used section (cb.base) 657 */ 658 if (cbuf_data(&con->cb) >= con->cb.base) { 659 iov[0].iov_len = PAGE_SIZE - cbuf_data(&con->cb); 660 iov[1].iov_len = con->cb.base; 661 iov[1].iov_base = page_address(con->rx_page); 662 nvec = 2; 663 } 664 len = iov[0].iov_len + iov[1].iov_len; 665 666 r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len, 667 MSG_DONTWAIT | MSG_NOSIGNAL); 668 if (ret <= 0) 669 goto out_close; 670 else if (ret == len) 671 call_again_soon = 1; 672 673 cbuf_add(&con->cb, ret); 674 ret = dlm_process_incoming_buffer(con->nodeid, 675 page_address(con->rx_page), 676 con->cb.base, con->cb.len, 677 PAGE_SIZE); 678 if (ret == -EBADMSG) { 679 log_print("lowcomms: addr=%p, base=%u, len=%u, read=%d", 680 page_address(con->rx_page), con->cb.base, 681 con->cb.len, r); 682 } 683 if (ret < 0) 684 goto out_close; 685 cbuf_eat(&con->cb, ret); 686 687 if (cbuf_empty(&con->cb) && !call_again_soon) { 688 __free_page(con->rx_page); 689 con->rx_page = NULL; 690 } 691 692 if (call_again_soon) 693 goto out_resched; 694 mutex_unlock(&con->sock_mutex); 695 return 0; 696 697 out_resched: 698 if (!test_and_set_bit(CF_READ_PENDING, &con->flags)) 699 queue_work(recv_workqueue, &con->rwork); 700 mutex_unlock(&con->sock_mutex); 701 return -EAGAIN; 702 703 out_close: 704 mutex_unlock(&con->sock_mutex); 705 if (ret != -EAGAIN) { 706 close_connection(con, false, true, false); 707 /* Reconnect when there is something to send */ 708 } 709 /* Don't return success if we really got EOF */ 710 if (ret == 0) 711 ret = -EAGAIN; 712 713 return ret; 714 } 715 716 /* Listening socket is busy, accept a connection */ 717 static int tcp_accept_from_sock(struct connection *con) 718 { 719 int result; 720 struct sockaddr_storage peeraddr; 721 struct socket *newsock; 722 int len; 723 int nodeid; 724 struct connection *newcon; 725 struct connection *addcon; 726 727 mutex_lock(&connections_lock); 728 if (!dlm_allow_conn) { 729 mutex_unlock(&connections_lock); 730 return -1; 731 } 732 mutex_unlock(&connections_lock); 733 734 memset(&peeraddr, 0, sizeof(peeraddr)); 735 result = sock_create_lite(dlm_local_addr[0]->ss_family, 736 SOCK_STREAM, IPPROTO_TCP, &newsock); 737 if (result < 0) 738 return -ENOMEM; 739 740 mutex_lock_nested(&con->sock_mutex, 0); 741 742 result = -ENOTCONN; 743 if (con->sock == NULL) 744 goto accept_err; 745 746 newsock->type = con->sock->type; 747 newsock->ops = con->sock->ops; 748 749 result = con->sock->ops->accept(con->sock, newsock, O_NONBLOCK, true); 750 if (result < 0) 751 goto accept_err; 752 753 /* Get the connected socket's peer */ 754 memset(&peeraddr, 0, sizeof(peeraddr)); 755 if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 756 &len, 2)) { 757 result = -ECONNABORTED; 758 goto accept_err; 759 } 760 761 /* Get the new node's NODEID */ 762 make_sockaddr(&peeraddr, 0, &len); 763 if (addr_to_nodeid(&peeraddr, &nodeid)) { 764 unsigned char *b=(unsigned char *)&peeraddr; 765 log_print("connect from non cluster node"); 766 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 767 b, sizeof(struct sockaddr_storage)); 768 sock_release(newsock); 769 mutex_unlock(&con->sock_mutex); 770 return -1; 771 } 772 773 log_print("got connection from %d", nodeid); 774 775 /* Check to see if we already have a connection to this node. This 776 * could happen if the two nodes initiate a connection at roughly 777 * the same time and the connections cross on the wire. 778 * In this case we store the incoming one in "othercon" 779 */ 780 newcon = nodeid2con(nodeid, GFP_NOFS); 781 if (!newcon) { 782 result = -ENOMEM; 783 goto accept_err; 784 } 785 mutex_lock_nested(&newcon->sock_mutex, 1); 786 if (newcon->sock) { 787 struct connection *othercon = newcon->othercon; 788 789 if (!othercon) { 790 othercon = kmem_cache_zalloc(con_cache, GFP_NOFS); 791 if (!othercon) { 792 log_print("failed to allocate incoming socket"); 793 mutex_unlock(&newcon->sock_mutex); 794 result = -ENOMEM; 795 goto accept_err; 796 } 797 othercon->nodeid = nodeid; 798 othercon->rx_action = receive_from_sock; 799 mutex_init(&othercon->sock_mutex); 800 INIT_WORK(&othercon->swork, process_send_sockets); 801 INIT_WORK(&othercon->rwork, process_recv_sockets); 802 set_bit(CF_IS_OTHERCON, &othercon->flags); 803 } 804 if (!othercon->sock) { 805 newcon->othercon = othercon; 806 othercon->sock = newsock; 807 newsock->sk->sk_user_data = othercon; 808 add_sock(newsock, othercon); 809 addcon = othercon; 810 } 811 else { 812 printk("Extra connection from node %d attempted\n", nodeid); 813 result = -EAGAIN; 814 mutex_unlock(&newcon->sock_mutex); 815 goto accept_err; 816 } 817 } 818 else { 819 newsock->sk->sk_user_data = newcon; 820 newcon->rx_action = receive_from_sock; 821 /* accept copies the sk after we've saved the callbacks, so we 822 don't want to save them a second time or comm errors will 823 result in calling sk_error_report recursively. */ 824 add_sock(newsock, newcon); 825 addcon = newcon; 826 } 827 828 mutex_unlock(&newcon->sock_mutex); 829 830 /* 831 * Add it to the active queue in case we got data 832 * between processing the accept adding the socket 833 * to the read_sockets list 834 */ 835 if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags)) 836 queue_work(recv_workqueue, &addcon->rwork); 837 mutex_unlock(&con->sock_mutex); 838 839 return 0; 840 841 accept_err: 842 mutex_unlock(&con->sock_mutex); 843 sock_release(newsock); 844 845 if (result != -EAGAIN) 846 log_print("error accepting connection from node: %d", result); 847 return result; 848 } 849 850 static int sctp_accept_from_sock(struct connection *con) 851 { 852 /* Check that the new node is in the lockspace */ 853 struct sctp_prim prim; 854 int nodeid; 855 int prim_len, ret; 856 int addr_len; 857 struct connection *newcon; 858 struct connection *addcon; 859 struct socket *newsock; 860 861 mutex_lock(&connections_lock); 862 if (!dlm_allow_conn) { 863 mutex_unlock(&connections_lock); 864 return -1; 865 } 866 mutex_unlock(&connections_lock); 867 868 mutex_lock_nested(&con->sock_mutex, 0); 869 870 ret = kernel_accept(con->sock, &newsock, O_NONBLOCK); 871 if (ret < 0) 872 goto accept_err; 873 874 memset(&prim, 0, sizeof(struct sctp_prim)); 875 prim_len = sizeof(struct sctp_prim); 876 877 ret = kernel_getsockopt(newsock, IPPROTO_SCTP, SCTP_PRIMARY_ADDR, 878 (char *)&prim, &prim_len); 879 if (ret < 0) { 880 log_print("getsockopt/sctp_primary_addr failed: %d", ret); 881 goto accept_err; 882 } 883 884 make_sockaddr(&prim.ssp_addr, 0, &addr_len); 885 ret = addr_to_nodeid(&prim.ssp_addr, &nodeid); 886 if (ret) { 887 unsigned char *b = (unsigned char *)&prim.ssp_addr; 888 889 log_print("reject connect from unknown addr"); 890 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 891 b, sizeof(struct sockaddr_storage)); 892 goto accept_err; 893 } 894 895 newcon = nodeid2con(nodeid, GFP_NOFS); 896 if (!newcon) { 897 ret = -ENOMEM; 898 goto accept_err; 899 } 900 901 mutex_lock_nested(&newcon->sock_mutex, 1); 902 903 if (newcon->sock) { 904 struct connection *othercon = newcon->othercon; 905 906 if (!othercon) { 907 othercon = kmem_cache_zalloc(con_cache, GFP_NOFS); 908 if (!othercon) { 909 log_print("failed to allocate incoming socket"); 910 mutex_unlock(&newcon->sock_mutex); 911 ret = -ENOMEM; 912 goto accept_err; 913 } 914 othercon->nodeid = nodeid; 915 othercon->rx_action = receive_from_sock; 916 mutex_init(&othercon->sock_mutex); 917 INIT_WORK(&othercon->swork, process_send_sockets); 918 INIT_WORK(&othercon->rwork, process_recv_sockets); 919 set_bit(CF_IS_OTHERCON, &othercon->flags); 920 } 921 if (!othercon->sock) { 922 newcon->othercon = othercon; 923 othercon->sock = newsock; 924 newsock->sk->sk_user_data = othercon; 925 add_sock(newsock, othercon); 926 addcon = othercon; 927 } else { 928 printk("Extra connection from node %d attempted\n", nodeid); 929 ret = -EAGAIN; 930 mutex_unlock(&newcon->sock_mutex); 931 goto accept_err; 932 } 933 } else { 934 newsock->sk->sk_user_data = newcon; 935 newcon->rx_action = receive_from_sock; 936 add_sock(newsock, newcon); 937 addcon = newcon; 938 } 939 940 log_print("connected to %d", nodeid); 941 942 mutex_unlock(&newcon->sock_mutex); 943 944 /* 945 * Add it to the active queue in case we got data 946 * between processing the accept adding the socket 947 * to the read_sockets list 948 */ 949 if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags)) 950 queue_work(recv_workqueue, &addcon->rwork); 951 mutex_unlock(&con->sock_mutex); 952 953 return 0; 954 955 accept_err: 956 mutex_unlock(&con->sock_mutex); 957 if (newsock) 958 sock_release(newsock); 959 if (ret != -EAGAIN) 960 log_print("error accepting connection from node: %d", ret); 961 962 return ret; 963 } 964 965 static void free_entry(struct writequeue_entry *e) 966 { 967 __free_page(e->page); 968 kfree(e); 969 } 970 971 /* 972 * writequeue_entry_complete - try to delete and free write queue entry 973 * @e: write queue entry to try to delete 974 * @completed: bytes completed 975 * 976 * writequeue_lock must be held. 977 */ 978 static void writequeue_entry_complete(struct writequeue_entry *e, int completed) 979 { 980 e->offset += completed; 981 e->len -= completed; 982 983 if (e->len == 0 && e->users == 0) { 984 list_del(&e->list); 985 free_entry(e); 986 } 987 } 988 989 /* 990 * sctp_bind_addrs - bind a SCTP socket to all our addresses 991 */ 992 static int sctp_bind_addrs(struct connection *con, uint16_t port) 993 { 994 struct sockaddr_storage localaddr; 995 int i, addr_len, result = 0; 996 997 for (i = 0; i < dlm_local_count; i++) { 998 memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr)); 999 make_sockaddr(&localaddr, port, &addr_len); 1000 1001 if (!i) 1002 result = kernel_bind(con->sock, 1003 (struct sockaddr *)&localaddr, 1004 addr_len); 1005 else 1006 result = kernel_setsockopt(con->sock, SOL_SCTP, 1007 SCTP_SOCKOPT_BINDX_ADD, 1008 (char *)&localaddr, addr_len); 1009 1010 if (result < 0) { 1011 log_print("Can't bind to %d addr number %d, %d.\n", 1012 port, i + 1, result); 1013 break; 1014 } 1015 } 1016 return result; 1017 } 1018 1019 /* Initiate an SCTP association. 1020 This is a special case of send_to_sock() in that we don't yet have a 1021 peeled-off socket for this association, so we use the listening socket 1022 and add the primary IP address of the remote node. 1023 */ 1024 static void sctp_connect_to_sock(struct connection *con) 1025 { 1026 struct sockaddr_storage daddr; 1027 int one = 1; 1028 int result; 1029 int addr_len; 1030 struct socket *sock; 1031 1032 if (con->nodeid == 0) { 1033 log_print("attempt to connect sock 0 foiled"); 1034 return; 1035 } 1036 1037 mutex_lock(&con->sock_mutex); 1038 1039 /* Some odd races can cause double-connects, ignore them */ 1040 if (con->retries++ > MAX_CONNECT_RETRIES) 1041 goto out; 1042 1043 if (con->sock) { 1044 log_print("node %d already connected.", con->nodeid); 1045 goto out; 1046 } 1047 1048 memset(&daddr, 0, sizeof(daddr)); 1049 result = nodeid_to_addr(con->nodeid, &daddr, NULL, true); 1050 if (result < 0) { 1051 log_print("no address for nodeid %d", con->nodeid); 1052 goto out; 1053 } 1054 1055 /* Create a socket to communicate with */ 1056 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family, 1057 SOCK_STREAM, IPPROTO_SCTP, &sock); 1058 if (result < 0) 1059 goto socket_err; 1060 1061 sock->sk->sk_user_data = con; 1062 con->rx_action = receive_from_sock; 1063 con->connect_action = sctp_connect_to_sock; 1064 add_sock(sock, con); 1065 1066 /* Bind to all addresses. */ 1067 if (sctp_bind_addrs(con, 0)) 1068 goto bind_err; 1069 1070 make_sockaddr(&daddr, dlm_config.ci_tcp_port, &addr_len); 1071 1072 log_print("connecting to %d", con->nodeid); 1073 1074 /* Turn off Nagle's algorithm */ 1075 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one, 1076 sizeof(one)); 1077 1078 result = sock->ops->connect(sock, (struct sockaddr *)&daddr, addr_len, 1079 O_NONBLOCK); 1080 if (result == -EINPROGRESS) 1081 result = 0; 1082 if (result == 0) 1083 goto out; 1084 1085 bind_err: 1086 con->sock = NULL; 1087 sock_release(sock); 1088 1089 socket_err: 1090 /* 1091 * Some errors are fatal and this list might need adjusting. For other 1092 * errors we try again until the max number of retries is reached. 1093 */ 1094 if (result != -EHOSTUNREACH && 1095 result != -ENETUNREACH && 1096 result != -ENETDOWN && 1097 result != -EINVAL && 1098 result != -EPROTONOSUPPORT) { 1099 log_print("connect %d try %d error %d", con->nodeid, 1100 con->retries, result); 1101 mutex_unlock(&con->sock_mutex); 1102 msleep(1000); 1103 lowcomms_connect_sock(con); 1104 return; 1105 } 1106 1107 out: 1108 mutex_unlock(&con->sock_mutex); 1109 } 1110 1111 /* Connect a new socket to its peer */ 1112 static void tcp_connect_to_sock(struct connection *con) 1113 { 1114 struct sockaddr_storage saddr, src_addr; 1115 int addr_len; 1116 struct socket *sock = NULL; 1117 int one = 1; 1118 int result; 1119 1120 if (con->nodeid == 0) { 1121 log_print("attempt to connect sock 0 foiled"); 1122 return; 1123 } 1124 1125 mutex_lock(&con->sock_mutex); 1126 if (con->retries++ > MAX_CONNECT_RETRIES) 1127 goto out; 1128 1129 /* Some odd races can cause double-connects, ignore them */ 1130 if (con->sock) 1131 goto out; 1132 1133 /* Create a socket to communicate with */ 1134 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family, 1135 SOCK_STREAM, IPPROTO_TCP, &sock); 1136 if (result < 0) 1137 goto out_err; 1138 1139 memset(&saddr, 0, sizeof(saddr)); 1140 result = nodeid_to_addr(con->nodeid, &saddr, NULL, false); 1141 if (result < 0) { 1142 log_print("no address for nodeid %d", con->nodeid); 1143 goto out_err; 1144 } 1145 1146 sock->sk->sk_user_data = con; 1147 con->rx_action = receive_from_sock; 1148 con->connect_action = tcp_connect_to_sock; 1149 add_sock(sock, con); 1150 1151 /* Bind to our cluster-known address connecting to avoid 1152 routing problems */ 1153 memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr)); 1154 make_sockaddr(&src_addr, 0, &addr_len); 1155 result = sock->ops->bind(sock, (struct sockaddr *) &src_addr, 1156 addr_len); 1157 if (result < 0) { 1158 log_print("could not bind for connect: %d", result); 1159 /* This *may* not indicate a critical error */ 1160 } 1161 1162 make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len); 1163 1164 log_print("connecting to %d", con->nodeid); 1165 1166 /* Turn off Nagle's algorithm */ 1167 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one, 1168 sizeof(one)); 1169 1170 result = sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len, 1171 O_NONBLOCK); 1172 if (result == -EINPROGRESS) 1173 result = 0; 1174 if (result == 0) 1175 goto out; 1176 1177 out_err: 1178 if (con->sock) { 1179 sock_release(con->sock); 1180 con->sock = NULL; 1181 } else if (sock) { 1182 sock_release(sock); 1183 } 1184 /* 1185 * Some errors are fatal and this list might need adjusting. For other 1186 * errors we try again until the max number of retries is reached. 1187 */ 1188 if (result != -EHOSTUNREACH && 1189 result != -ENETUNREACH && 1190 result != -ENETDOWN && 1191 result != -EINVAL && 1192 result != -EPROTONOSUPPORT) { 1193 log_print("connect %d try %d error %d", con->nodeid, 1194 con->retries, result); 1195 mutex_unlock(&con->sock_mutex); 1196 msleep(1000); 1197 lowcomms_connect_sock(con); 1198 return; 1199 } 1200 out: 1201 mutex_unlock(&con->sock_mutex); 1202 return; 1203 } 1204 1205 static struct socket *tcp_create_listen_sock(struct connection *con, 1206 struct sockaddr_storage *saddr) 1207 { 1208 struct socket *sock = NULL; 1209 int result = 0; 1210 int one = 1; 1211 int addr_len; 1212 1213 if (dlm_local_addr[0]->ss_family == AF_INET) 1214 addr_len = sizeof(struct sockaddr_in); 1215 else 1216 addr_len = sizeof(struct sockaddr_in6); 1217 1218 /* Create a socket to communicate with */ 1219 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family, 1220 SOCK_STREAM, IPPROTO_TCP, &sock); 1221 if (result < 0) { 1222 log_print("Can't create listening comms socket"); 1223 goto create_out; 1224 } 1225 1226 /* Turn off Nagle's algorithm */ 1227 kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one, 1228 sizeof(one)); 1229 1230 result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR, 1231 (char *)&one, sizeof(one)); 1232 1233 if (result < 0) { 1234 log_print("Failed to set SO_REUSEADDR on socket: %d", result); 1235 } 1236 sock->sk->sk_user_data = con; 1237 save_listen_callbacks(sock); 1238 con->rx_action = tcp_accept_from_sock; 1239 con->connect_action = tcp_connect_to_sock; 1240 1241 /* Bind to our port */ 1242 make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len); 1243 result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len); 1244 if (result < 0) { 1245 log_print("Can't bind to port %d", dlm_config.ci_tcp_port); 1246 sock_release(sock); 1247 sock = NULL; 1248 con->sock = NULL; 1249 goto create_out; 1250 } 1251 result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE, 1252 (char *)&one, sizeof(one)); 1253 if (result < 0) { 1254 log_print("Set keepalive failed: %d", result); 1255 } 1256 1257 result = sock->ops->listen(sock, 5); 1258 if (result < 0) { 1259 log_print("Can't listen on port %d", dlm_config.ci_tcp_port); 1260 sock_release(sock); 1261 sock = NULL; 1262 goto create_out; 1263 } 1264 1265 create_out: 1266 return sock; 1267 } 1268 1269 /* Get local addresses */ 1270 static void init_local(void) 1271 { 1272 struct sockaddr_storage sas, *addr; 1273 int i; 1274 1275 dlm_local_count = 0; 1276 for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) { 1277 if (dlm_our_addr(&sas, i)) 1278 break; 1279 1280 addr = kmemdup(&sas, sizeof(*addr), GFP_NOFS); 1281 if (!addr) 1282 break; 1283 dlm_local_addr[dlm_local_count++] = addr; 1284 } 1285 } 1286 1287 /* Initialise SCTP socket and bind to all interfaces */ 1288 static int sctp_listen_for_all(void) 1289 { 1290 struct socket *sock = NULL; 1291 int result = -EINVAL; 1292 struct connection *con = nodeid2con(0, GFP_NOFS); 1293 int bufsize = NEEDED_RMEM; 1294 int one = 1; 1295 1296 if (!con) 1297 return -ENOMEM; 1298 1299 log_print("Using SCTP for communications"); 1300 1301 result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family, 1302 SOCK_STREAM, IPPROTO_SCTP, &sock); 1303 if (result < 0) { 1304 log_print("Can't create comms socket, check SCTP is loaded"); 1305 goto out; 1306 } 1307 1308 result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE, 1309 (char *)&bufsize, sizeof(bufsize)); 1310 if (result) 1311 log_print("Error increasing buffer space on socket %d", result); 1312 1313 result = kernel_setsockopt(sock, SOL_SCTP, SCTP_NODELAY, (char *)&one, 1314 sizeof(one)); 1315 if (result < 0) 1316 log_print("Could not set SCTP NODELAY error %d\n", result); 1317 1318 write_lock_bh(&sock->sk->sk_callback_lock); 1319 /* Init con struct */ 1320 sock->sk->sk_user_data = con; 1321 save_listen_callbacks(sock); 1322 con->sock = sock; 1323 con->sock->sk->sk_data_ready = lowcomms_data_ready; 1324 con->rx_action = sctp_accept_from_sock; 1325 con->connect_action = sctp_connect_to_sock; 1326 1327 write_unlock_bh(&sock->sk->sk_callback_lock); 1328 1329 /* Bind to all addresses. */ 1330 if (sctp_bind_addrs(con, dlm_config.ci_tcp_port)) 1331 goto create_delsock; 1332 1333 result = sock->ops->listen(sock, 5); 1334 if (result < 0) { 1335 log_print("Can't set socket listening"); 1336 goto create_delsock; 1337 } 1338 1339 return 0; 1340 1341 create_delsock: 1342 sock_release(sock); 1343 con->sock = NULL; 1344 out: 1345 return result; 1346 } 1347 1348 static int tcp_listen_for_all(void) 1349 { 1350 struct socket *sock = NULL; 1351 struct connection *con = nodeid2con(0, GFP_NOFS); 1352 int result = -EINVAL; 1353 1354 if (!con) 1355 return -ENOMEM; 1356 1357 /* We don't support multi-homed hosts */ 1358 if (dlm_local_addr[1] != NULL) { 1359 log_print("TCP protocol can't handle multi-homed hosts, " 1360 "try SCTP"); 1361 return -EINVAL; 1362 } 1363 1364 log_print("Using TCP for communications"); 1365 1366 sock = tcp_create_listen_sock(con, dlm_local_addr[0]); 1367 if (sock) { 1368 add_sock(sock, con); 1369 result = 0; 1370 } 1371 else { 1372 result = -EADDRINUSE; 1373 } 1374 1375 return result; 1376 } 1377 1378 1379 1380 static struct writequeue_entry *new_writequeue_entry(struct connection *con, 1381 gfp_t allocation) 1382 { 1383 struct writequeue_entry *entry; 1384 1385 entry = kmalloc(sizeof(struct writequeue_entry), allocation); 1386 if (!entry) 1387 return NULL; 1388 1389 entry->page = alloc_page(allocation); 1390 if (!entry->page) { 1391 kfree(entry); 1392 return NULL; 1393 } 1394 1395 entry->offset = 0; 1396 entry->len = 0; 1397 entry->end = 0; 1398 entry->users = 0; 1399 entry->con = con; 1400 1401 return entry; 1402 } 1403 1404 void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc) 1405 { 1406 struct connection *con; 1407 struct writequeue_entry *e; 1408 int offset = 0; 1409 1410 con = nodeid2con(nodeid, allocation); 1411 if (!con) 1412 return NULL; 1413 1414 spin_lock(&con->writequeue_lock); 1415 e = list_entry(con->writequeue.prev, struct writequeue_entry, list); 1416 if ((&e->list == &con->writequeue) || 1417 (PAGE_SIZE - e->end < len)) { 1418 e = NULL; 1419 } else { 1420 offset = e->end; 1421 e->end += len; 1422 e->users++; 1423 } 1424 spin_unlock(&con->writequeue_lock); 1425 1426 if (e) { 1427 got_one: 1428 *ppc = page_address(e->page) + offset; 1429 return e; 1430 } 1431 1432 e = new_writequeue_entry(con, allocation); 1433 if (e) { 1434 spin_lock(&con->writequeue_lock); 1435 offset = e->end; 1436 e->end += len; 1437 e->users++; 1438 list_add_tail(&e->list, &con->writequeue); 1439 spin_unlock(&con->writequeue_lock); 1440 goto got_one; 1441 } 1442 return NULL; 1443 } 1444 1445 void dlm_lowcomms_commit_buffer(void *mh) 1446 { 1447 struct writequeue_entry *e = (struct writequeue_entry *)mh; 1448 struct connection *con = e->con; 1449 int users; 1450 1451 spin_lock(&con->writequeue_lock); 1452 users = --e->users; 1453 if (users) 1454 goto out; 1455 e->len = e->end - e->offset; 1456 spin_unlock(&con->writequeue_lock); 1457 1458 queue_work(send_workqueue, &con->swork); 1459 return; 1460 1461 out: 1462 spin_unlock(&con->writequeue_lock); 1463 return; 1464 } 1465 1466 /* Send a message */ 1467 static void send_to_sock(struct connection *con) 1468 { 1469 int ret = 0; 1470 const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL; 1471 struct writequeue_entry *e; 1472 int len, offset; 1473 int count = 0; 1474 1475 mutex_lock(&con->sock_mutex); 1476 if (con->sock == NULL) 1477 goto out_connect; 1478 1479 spin_lock(&con->writequeue_lock); 1480 for (;;) { 1481 e = list_entry(con->writequeue.next, struct writequeue_entry, 1482 list); 1483 if ((struct list_head *) e == &con->writequeue) 1484 break; 1485 1486 len = e->len; 1487 offset = e->offset; 1488 BUG_ON(len == 0 && e->users == 0); 1489 spin_unlock(&con->writequeue_lock); 1490 1491 ret = 0; 1492 if (len) { 1493 ret = kernel_sendpage(con->sock, e->page, offset, len, 1494 msg_flags); 1495 if (ret == -EAGAIN || ret == 0) { 1496 if (ret == -EAGAIN && 1497 test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) && 1498 !test_and_set_bit(CF_APP_LIMITED, &con->flags)) { 1499 /* Notify TCP that we're limited by the 1500 * application window size. 1501 */ 1502 set_bit(SOCK_NOSPACE, &con->sock->flags); 1503 con->sock->sk->sk_write_pending++; 1504 } 1505 cond_resched(); 1506 goto out; 1507 } else if (ret < 0) 1508 goto send_error; 1509 } 1510 1511 /* Don't starve people filling buffers */ 1512 if (++count >= MAX_SEND_MSG_COUNT) { 1513 cond_resched(); 1514 count = 0; 1515 } 1516 1517 spin_lock(&con->writequeue_lock); 1518 writequeue_entry_complete(e, ret); 1519 } 1520 spin_unlock(&con->writequeue_lock); 1521 out: 1522 mutex_unlock(&con->sock_mutex); 1523 return; 1524 1525 send_error: 1526 mutex_unlock(&con->sock_mutex); 1527 close_connection(con, false, false, true); 1528 /* Requeue the send work. When the work daemon runs again, it will try 1529 a new connection, then call this function again. */ 1530 queue_work(send_workqueue, &con->swork); 1531 return; 1532 1533 out_connect: 1534 mutex_unlock(&con->sock_mutex); 1535 queue_work(send_workqueue, &con->swork); 1536 cond_resched(); 1537 } 1538 1539 static void clean_one_writequeue(struct connection *con) 1540 { 1541 struct writequeue_entry *e, *safe; 1542 1543 spin_lock(&con->writequeue_lock); 1544 list_for_each_entry_safe(e, safe, &con->writequeue, list) { 1545 list_del(&e->list); 1546 free_entry(e); 1547 } 1548 spin_unlock(&con->writequeue_lock); 1549 } 1550 1551 /* Called from recovery when it knows that a node has 1552 left the cluster */ 1553 int dlm_lowcomms_close(int nodeid) 1554 { 1555 struct connection *con; 1556 struct dlm_node_addr *na; 1557 1558 log_print("closing connection to node %d", nodeid); 1559 con = nodeid2con(nodeid, 0); 1560 if (con) { 1561 set_bit(CF_CLOSE, &con->flags); 1562 close_connection(con, true, true, true); 1563 clean_one_writequeue(con); 1564 } 1565 1566 spin_lock(&dlm_node_addrs_spin); 1567 na = find_node_addr(nodeid); 1568 if (na) { 1569 list_del(&na->list); 1570 while (na->addr_count--) 1571 kfree(na->addr[na->addr_count]); 1572 kfree(na); 1573 } 1574 spin_unlock(&dlm_node_addrs_spin); 1575 1576 return 0; 1577 } 1578 1579 /* Receive workqueue function */ 1580 static void process_recv_sockets(struct work_struct *work) 1581 { 1582 struct connection *con = container_of(work, struct connection, rwork); 1583 int err; 1584 1585 clear_bit(CF_READ_PENDING, &con->flags); 1586 do { 1587 err = con->rx_action(con); 1588 } while (!err); 1589 } 1590 1591 /* Send workqueue function */ 1592 static void process_send_sockets(struct work_struct *work) 1593 { 1594 struct connection *con = container_of(work, struct connection, swork); 1595 1596 if (con->sock == NULL) /* not mutex protected so check it inside too */ 1597 con->connect_action(con); 1598 if (!list_empty(&con->writequeue)) 1599 send_to_sock(con); 1600 } 1601 1602 1603 /* Discard all entries on the write queues */ 1604 static void clean_writequeues(void) 1605 { 1606 foreach_conn(clean_one_writequeue); 1607 } 1608 1609 static void work_stop(void) 1610 { 1611 destroy_workqueue(recv_workqueue); 1612 destroy_workqueue(send_workqueue); 1613 } 1614 1615 static int work_start(void) 1616 { 1617 recv_workqueue = alloc_workqueue("dlm_recv", 1618 WQ_UNBOUND | WQ_MEM_RECLAIM, 1); 1619 if (!recv_workqueue) { 1620 log_print("can't start dlm_recv"); 1621 return -ENOMEM; 1622 } 1623 1624 send_workqueue = alloc_workqueue("dlm_send", 1625 WQ_UNBOUND | WQ_MEM_RECLAIM, 1); 1626 if (!send_workqueue) { 1627 log_print("can't start dlm_send"); 1628 destroy_workqueue(recv_workqueue); 1629 return -ENOMEM; 1630 } 1631 1632 return 0; 1633 } 1634 1635 static void _stop_conn(struct connection *con, bool and_other) 1636 { 1637 mutex_lock(&con->sock_mutex); 1638 set_bit(CF_READ_PENDING, &con->flags); 1639 if (con->sock && con->sock->sk) 1640 con->sock->sk->sk_user_data = NULL; 1641 if (con->othercon && and_other) 1642 _stop_conn(con->othercon, false); 1643 mutex_unlock(&con->sock_mutex); 1644 } 1645 1646 static void stop_conn(struct connection *con) 1647 { 1648 _stop_conn(con, true); 1649 } 1650 1651 static void free_conn(struct connection *con) 1652 { 1653 close_connection(con, true, true, true); 1654 if (con->othercon) 1655 kmem_cache_free(con_cache, con->othercon); 1656 hlist_del(&con->list); 1657 kmem_cache_free(con_cache, con); 1658 } 1659 1660 static void work_flush(void) 1661 { 1662 int ok; 1663 int i; 1664 struct hlist_node *n; 1665 struct connection *con; 1666 1667 flush_workqueue(recv_workqueue); 1668 flush_workqueue(send_workqueue); 1669 do { 1670 ok = 1; 1671 foreach_conn(stop_conn); 1672 flush_workqueue(recv_workqueue); 1673 flush_workqueue(send_workqueue); 1674 for (i = 0; i < CONN_HASH_SIZE && ok; i++) { 1675 hlist_for_each_entry_safe(con, n, 1676 &connection_hash[i], list) { 1677 ok &= test_bit(CF_READ_PENDING, &con->flags); 1678 if (con->othercon) 1679 ok &= test_bit(CF_READ_PENDING, 1680 &con->othercon->flags); 1681 } 1682 } 1683 } while (!ok); 1684 } 1685 1686 void dlm_lowcomms_stop(void) 1687 { 1688 /* Set all the flags to prevent any 1689 socket activity. 1690 */ 1691 mutex_lock(&connections_lock); 1692 dlm_allow_conn = 0; 1693 mutex_unlock(&connections_lock); 1694 work_flush(); 1695 clean_writequeues(); 1696 foreach_conn(free_conn); 1697 work_stop(); 1698 1699 kmem_cache_destroy(con_cache); 1700 } 1701 1702 int dlm_lowcomms_start(void) 1703 { 1704 int error = -EINVAL; 1705 struct connection *con; 1706 int i; 1707 1708 for (i = 0; i < CONN_HASH_SIZE; i++) 1709 INIT_HLIST_HEAD(&connection_hash[i]); 1710 1711 init_local(); 1712 if (!dlm_local_count) { 1713 error = -ENOTCONN; 1714 log_print("no local IP address has been set"); 1715 goto fail; 1716 } 1717 1718 error = -ENOMEM; 1719 con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection), 1720 __alignof__(struct connection), 0, 1721 NULL); 1722 if (!con_cache) 1723 goto fail; 1724 1725 error = work_start(); 1726 if (error) 1727 goto fail_destroy; 1728 1729 dlm_allow_conn = 1; 1730 1731 /* Start listening */ 1732 if (dlm_config.ci_protocol == 0) 1733 error = tcp_listen_for_all(); 1734 else 1735 error = sctp_listen_for_all(); 1736 if (error) 1737 goto fail_unlisten; 1738 1739 return 0; 1740 1741 fail_unlisten: 1742 dlm_allow_conn = 0; 1743 con = nodeid2con(0,0); 1744 if (con) { 1745 close_connection(con, false, true, true); 1746 kmem_cache_free(con_cache, con); 1747 } 1748 fail_destroy: 1749 kmem_cache_destroy(con_cache); 1750 fail: 1751 return error; 1752 } 1753 1754 void dlm_lowcomms_exit(void) 1755 { 1756 struct dlm_node_addr *na, *safe; 1757 1758 spin_lock(&dlm_node_addrs_spin); 1759 list_for_each_entry_safe(na, safe, &dlm_node_addrs, list) { 1760 list_del(&na->list); 1761 while (na->addr_count--) 1762 kfree(na->addr[na->addr_count]); 1763 kfree(na); 1764 } 1765 spin_unlock(&dlm_node_addrs_spin); 1766 } 1767