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_WRITE_PENDING 2 110 #define CF_INIT_PENDING 4 111 #define CF_IS_OTHERCON 5 112 #define CF_CLOSE 6 113 #define CF_APP_LIMITED 7 114 #define CF_CLOSING 8 115 struct list_head writequeue; /* List of outgoing writequeue_entries */ 116 spinlock_t writequeue_lock; 117 int (*rx_action) (struct connection *); /* What to do when active */ 118 void (*connect_action) (struct connection *); /* What to do to connect */ 119 struct page *rx_page; 120 struct cbuf cb; 121 int retries; 122 #define MAX_CONNECT_RETRIES 3 123 struct hlist_node list; 124 struct connection *othercon; 125 struct work_struct rwork; /* Receive workqueue */ 126 struct work_struct swork; /* Send workqueue */ 127 }; 128 #define sock2con(x) ((struct connection *)(x)->sk_user_data) 129 130 /* An entry waiting to be sent */ 131 struct writequeue_entry { 132 struct list_head list; 133 struct page *page; 134 int offset; 135 int len; 136 int end; 137 int users; 138 struct connection *con; 139 }; 140 141 struct dlm_node_addr { 142 struct list_head list; 143 int nodeid; 144 int addr_count; 145 int curr_addr_index; 146 struct sockaddr_storage *addr[DLM_MAX_ADDR_COUNT]; 147 }; 148 149 static struct listen_sock_callbacks { 150 void (*sk_error_report)(struct sock *); 151 void (*sk_data_ready)(struct sock *); 152 void (*sk_state_change)(struct sock *); 153 void (*sk_write_space)(struct sock *); 154 } listen_sock; 155 156 static LIST_HEAD(dlm_node_addrs); 157 static DEFINE_SPINLOCK(dlm_node_addrs_spin); 158 159 static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT]; 160 static int dlm_local_count; 161 static int dlm_allow_conn; 162 163 /* Work queues */ 164 static struct workqueue_struct *recv_workqueue; 165 static struct workqueue_struct *send_workqueue; 166 167 static struct hlist_head connection_hash[CONN_HASH_SIZE]; 168 static DEFINE_MUTEX(connections_lock); 169 static struct kmem_cache *con_cache; 170 171 static void process_recv_sockets(struct work_struct *work); 172 static void process_send_sockets(struct work_struct *work); 173 174 175 /* This is deliberately very simple because most clusters have simple 176 sequential nodeids, so we should be able to go straight to a connection 177 struct in the array */ 178 static inline int nodeid_hash(int nodeid) 179 { 180 return nodeid & (CONN_HASH_SIZE-1); 181 } 182 183 static struct connection *__find_con(int nodeid) 184 { 185 int r; 186 struct connection *con; 187 188 r = nodeid_hash(nodeid); 189 190 hlist_for_each_entry(con, &connection_hash[r], list) { 191 if (con->nodeid == nodeid) 192 return con; 193 } 194 return NULL; 195 } 196 197 /* 198 * If 'allocation' is zero then we don't attempt to create a new 199 * connection structure for this node. 200 */ 201 static struct connection *__nodeid2con(int nodeid, gfp_t alloc) 202 { 203 struct connection *con = NULL; 204 int r; 205 206 con = __find_con(nodeid); 207 if (con || !alloc) 208 return con; 209 210 con = kmem_cache_zalloc(con_cache, alloc); 211 if (!con) 212 return NULL; 213 214 r = nodeid_hash(nodeid); 215 hlist_add_head(&con->list, &connection_hash[r]); 216 217 con->nodeid = nodeid; 218 mutex_init(&con->sock_mutex); 219 INIT_LIST_HEAD(&con->writequeue); 220 spin_lock_init(&con->writequeue_lock); 221 INIT_WORK(&con->swork, process_send_sockets); 222 INIT_WORK(&con->rwork, process_recv_sockets); 223 224 /* Setup action pointers for child sockets */ 225 if (con->nodeid) { 226 struct connection *zerocon = __find_con(0); 227 228 con->connect_action = zerocon->connect_action; 229 if (!con->rx_action) 230 con->rx_action = zerocon->rx_action; 231 } 232 233 return con; 234 } 235 236 /* Loop round all connections */ 237 static void foreach_conn(void (*conn_func)(struct connection *c)) 238 { 239 int i; 240 struct hlist_node *n; 241 struct connection *con; 242 243 for (i = 0; i < CONN_HASH_SIZE; i++) { 244 hlist_for_each_entry_safe(con, n, &connection_hash[i], list) 245 conn_func(con); 246 } 247 } 248 249 static struct connection *nodeid2con(int nodeid, gfp_t allocation) 250 { 251 struct connection *con; 252 253 mutex_lock(&connections_lock); 254 con = __nodeid2con(nodeid, allocation); 255 mutex_unlock(&connections_lock); 256 257 return con; 258 } 259 260 static struct dlm_node_addr *find_node_addr(int nodeid) 261 { 262 struct dlm_node_addr *na; 263 264 list_for_each_entry(na, &dlm_node_addrs, list) { 265 if (na->nodeid == nodeid) 266 return na; 267 } 268 return NULL; 269 } 270 271 static int addr_compare(struct sockaddr_storage *x, struct sockaddr_storage *y) 272 { 273 switch (x->ss_family) { 274 case AF_INET: { 275 struct sockaddr_in *sinx = (struct sockaddr_in *)x; 276 struct sockaddr_in *siny = (struct sockaddr_in *)y; 277 if (sinx->sin_addr.s_addr != siny->sin_addr.s_addr) 278 return 0; 279 if (sinx->sin_port != siny->sin_port) 280 return 0; 281 break; 282 } 283 case AF_INET6: { 284 struct sockaddr_in6 *sinx = (struct sockaddr_in6 *)x; 285 struct sockaddr_in6 *siny = (struct sockaddr_in6 *)y; 286 if (!ipv6_addr_equal(&sinx->sin6_addr, &siny->sin6_addr)) 287 return 0; 288 if (sinx->sin6_port != siny->sin6_port) 289 return 0; 290 break; 291 } 292 default: 293 return 0; 294 } 295 return 1; 296 } 297 298 static int nodeid_to_addr(int nodeid, struct sockaddr_storage *sas_out, 299 struct sockaddr *sa_out, bool try_new_addr) 300 { 301 struct sockaddr_storage sas; 302 struct dlm_node_addr *na; 303 304 if (!dlm_local_count) 305 return -1; 306 307 spin_lock(&dlm_node_addrs_spin); 308 na = find_node_addr(nodeid); 309 if (na && na->addr_count) { 310 memcpy(&sas, na->addr[na->curr_addr_index], 311 sizeof(struct sockaddr_storage)); 312 313 if (try_new_addr) { 314 na->curr_addr_index++; 315 if (na->curr_addr_index == na->addr_count) 316 na->curr_addr_index = 0; 317 } 318 } 319 spin_unlock(&dlm_node_addrs_spin); 320 321 if (!na) 322 return -EEXIST; 323 324 if (!na->addr_count) 325 return -ENOENT; 326 327 if (sas_out) 328 memcpy(sas_out, &sas, sizeof(struct sockaddr_storage)); 329 330 if (!sa_out) 331 return 0; 332 333 if (dlm_local_addr[0]->ss_family == AF_INET) { 334 struct sockaddr_in *in4 = (struct sockaddr_in *) &sas; 335 struct sockaddr_in *ret4 = (struct sockaddr_in *) sa_out; 336 ret4->sin_addr.s_addr = in4->sin_addr.s_addr; 337 } else { 338 struct sockaddr_in6 *in6 = (struct sockaddr_in6 *) &sas; 339 struct sockaddr_in6 *ret6 = (struct sockaddr_in6 *) sa_out; 340 ret6->sin6_addr = in6->sin6_addr; 341 } 342 343 return 0; 344 } 345 346 static int addr_to_nodeid(struct sockaddr_storage *addr, int *nodeid) 347 { 348 struct dlm_node_addr *na; 349 int rv = -EEXIST; 350 int addr_i; 351 352 spin_lock(&dlm_node_addrs_spin); 353 list_for_each_entry(na, &dlm_node_addrs, list) { 354 if (!na->addr_count) 355 continue; 356 357 for (addr_i = 0; addr_i < na->addr_count; addr_i++) { 358 if (addr_compare(na->addr[addr_i], addr)) { 359 *nodeid = na->nodeid; 360 rv = 0; 361 goto unlock; 362 } 363 } 364 } 365 unlock: 366 spin_unlock(&dlm_node_addrs_spin); 367 return rv; 368 } 369 370 int dlm_lowcomms_addr(int nodeid, struct sockaddr_storage *addr, int len) 371 { 372 struct sockaddr_storage *new_addr; 373 struct dlm_node_addr *new_node, *na; 374 375 new_node = kzalloc(sizeof(struct dlm_node_addr), GFP_NOFS); 376 if (!new_node) 377 return -ENOMEM; 378 379 new_addr = kzalloc(sizeof(struct sockaddr_storage), GFP_NOFS); 380 if (!new_addr) { 381 kfree(new_node); 382 return -ENOMEM; 383 } 384 385 memcpy(new_addr, addr, len); 386 387 spin_lock(&dlm_node_addrs_spin); 388 na = find_node_addr(nodeid); 389 if (!na) { 390 new_node->nodeid = nodeid; 391 new_node->addr[0] = new_addr; 392 new_node->addr_count = 1; 393 list_add(&new_node->list, &dlm_node_addrs); 394 spin_unlock(&dlm_node_addrs_spin); 395 return 0; 396 } 397 398 if (na->addr_count >= DLM_MAX_ADDR_COUNT) { 399 spin_unlock(&dlm_node_addrs_spin); 400 kfree(new_addr); 401 kfree(new_node); 402 return -ENOSPC; 403 } 404 405 na->addr[na->addr_count++] = new_addr; 406 spin_unlock(&dlm_node_addrs_spin); 407 kfree(new_node); 408 return 0; 409 } 410 411 /* Data available on socket or listen socket received a connect */ 412 static void lowcomms_data_ready(struct sock *sk) 413 { 414 struct connection *con = sock2con(sk); 415 if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags)) 416 queue_work(recv_workqueue, &con->rwork); 417 } 418 419 static void lowcomms_write_space(struct sock *sk) 420 { 421 struct connection *con = sock2con(sk); 422 423 if (!con) 424 return; 425 426 clear_bit(SOCK_NOSPACE, &con->sock->flags); 427 428 if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) { 429 con->sock->sk->sk_write_pending--; 430 clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags); 431 } 432 433 queue_work(send_workqueue, &con->swork); 434 } 435 436 static inline void lowcomms_connect_sock(struct connection *con) 437 { 438 if (test_bit(CF_CLOSE, &con->flags)) 439 return; 440 queue_work(send_workqueue, &con->swork); 441 cond_resched(); 442 } 443 444 static void lowcomms_state_change(struct sock *sk) 445 { 446 /* SCTP layer is not calling sk_data_ready when the connection 447 * is done, so we catch the signal through here. Also, it 448 * doesn't switch socket state when entering shutdown, so we 449 * skip the write in that case. 450 */ 451 if (sk->sk_shutdown) { 452 if (sk->sk_shutdown == RCV_SHUTDOWN) 453 lowcomms_data_ready(sk); 454 } else if (sk->sk_state == TCP_ESTABLISHED) { 455 lowcomms_write_space(sk); 456 } 457 } 458 459 int dlm_lowcomms_connect_node(int nodeid) 460 { 461 struct connection *con; 462 463 if (nodeid == dlm_our_nodeid()) 464 return 0; 465 466 con = nodeid2con(nodeid, GFP_NOFS); 467 if (!con) 468 return -ENOMEM; 469 lowcomms_connect_sock(con); 470 return 0; 471 } 472 473 static void lowcomms_error_report(struct sock *sk) 474 { 475 struct connection *con; 476 struct sockaddr_storage saddr; 477 int buflen; 478 void (*orig_report)(struct sock *) = NULL; 479 480 read_lock_bh(&sk->sk_callback_lock); 481 con = sock2con(sk); 482 if (con == NULL) 483 goto out; 484 485 orig_report = listen_sock.sk_error_report; 486 if (con->sock == NULL || 487 kernel_getpeername(con->sock, (struct sockaddr *)&saddr, &buflen)) { 488 printk_ratelimited(KERN_ERR "dlm: node %d: socket error " 489 "sending to node %d, port %d, " 490 "sk_err=%d/%d\n", dlm_our_nodeid(), 491 con->nodeid, dlm_config.ci_tcp_port, 492 sk->sk_err, sk->sk_err_soft); 493 } else if (saddr.ss_family == AF_INET) { 494 struct sockaddr_in *sin4 = (struct sockaddr_in *)&saddr; 495 496 printk_ratelimited(KERN_ERR "dlm: node %d: socket error " 497 "sending to node %d at %pI4, port %d, " 498 "sk_err=%d/%d\n", dlm_our_nodeid(), 499 con->nodeid, &sin4->sin_addr.s_addr, 500 dlm_config.ci_tcp_port, sk->sk_err, 501 sk->sk_err_soft); 502 } else { 503 struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&saddr; 504 505 printk_ratelimited(KERN_ERR "dlm: node %d: socket error " 506 "sending to node %d at %u.%u.%u.%u, " 507 "port %d, sk_err=%d/%d\n", dlm_our_nodeid(), 508 con->nodeid, sin6->sin6_addr.s6_addr32[0], 509 sin6->sin6_addr.s6_addr32[1], 510 sin6->sin6_addr.s6_addr32[2], 511 sin6->sin6_addr.s6_addr32[3], 512 dlm_config.ci_tcp_port, sk->sk_err, 513 sk->sk_err_soft); 514 } 515 out: 516 read_unlock_bh(&sk->sk_callback_lock); 517 if (orig_report) 518 orig_report(sk); 519 } 520 521 /* Note: sk_callback_lock must be locked before calling this function. */ 522 static void save_listen_callbacks(struct socket *sock) 523 { 524 struct sock *sk = sock->sk; 525 526 listen_sock.sk_data_ready = sk->sk_data_ready; 527 listen_sock.sk_state_change = sk->sk_state_change; 528 listen_sock.sk_write_space = sk->sk_write_space; 529 listen_sock.sk_error_report = sk->sk_error_report; 530 } 531 532 static void restore_callbacks(struct socket *sock) 533 { 534 struct sock *sk = sock->sk; 535 536 write_lock_bh(&sk->sk_callback_lock); 537 sk->sk_user_data = NULL; 538 sk->sk_data_ready = listen_sock.sk_data_ready; 539 sk->sk_state_change = listen_sock.sk_state_change; 540 sk->sk_write_space = listen_sock.sk_write_space; 541 sk->sk_error_report = listen_sock.sk_error_report; 542 write_unlock_bh(&sk->sk_callback_lock); 543 } 544 545 /* Make a socket active */ 546 static void add_sock(struct socket *sock, struct connection *con) 547 { 548 struct sock *sk = sock->sk; 549 550 write_lock_bh(&sk->sk_callback_lock); 551 con->sock = sock; 552 553 sk->sk_user_data = con; 554 /* Install a data_ready callback */ 555 sk->sk_data_ready = lowcomms_data_ready; 556 sk->sk_write_space = lowcomms_write_space; 557 sk->sk_state_change = lowcomms_state_change; 558 sk->sk_allocation = GFP_NOFS; 559 sk->sk_error_report = lowcomms_error_report; 560 write_unlock_bh(&sk->sk_callback_lock); 561 } 562 563 /* Add the port number to an IPv6 or 4 sockaddr and return the address 564 length */ 565 static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port, 566 int *addr_len) 567 { 568 saddr->ss_family = dlm_local_addr[0]->ss_family; 569 if (saddr->ss_family == AF_INET) { 570 struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr; 571 in4_addr->sin_port = cpu_to_be16(port); 572 *addr_len = sizeof(struct sockaddr_in); 573 memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero)); 574 } else { 575 struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr; 576 in6_addr->sin6_port = cpu_to_be16(port); 577 *addr_len = sizeof(struct sockaddr_in6); 578 } 579 memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len); 580 } 581 582 /* Close a remote connection and tidy up */ 583 static void close_connection(struct connection *con, bool and_other, 584 bool tx, bool rx) 585 { 586 bool closing = test_and_set_bit(CF_CLOSING, &con->flags); 587 588 if (tx && !closing && cancel_work_sync(&con->swork)) 589 log_print("canceled swork for node %d", con->nodeid); 590 if (rx && !closing && cancel_work_sync(&con->rwork)) 591 log_print("canceled rwork for node %d", con->nodeid); 592 593 mutex_lock(&con->sock_mutex); 594 if (con->sock) { 595 restore_callbacks(con->sock); 596 sock_release(con->sock); 597 con->sock = NULL; 598 } 599 if (con->othercon && and_other) { 600 /* Will only re-enter once. */ 601 close_connection(con->othercon, false, true, true); 602 } 603 if (con->rx_page) { 604 __free_page(con->rx_page); 605 con->rx_page = NULL; 606 } 607 608 con->retries = 0; 609 mutex_unlock(&con->sock_mutex); 610 clear_bit(CF_CLOSING, &con->flags); 611 } 612 613 /* Data received from remote end */ 614 static int receive_from_sock(struct connection *con) 615 { 616 int ret = 0; 617 struct msghdr msg = {}; 618 struct kvec iov[2]; 619 unsigned len; 620 int r; 621 int call_again_soon = 0; 622 int nvec; 623 624 mutex_lock(&con->sock_mutex); 625 626 if (con->sock == NULL) { 627 ret = -EAGAIN; 628 goto out_close; 629 } 630 if (con->nodeid == 0) { 631 ret = -EINVAL; 632 goto out_close; 633 } 634 635 if (con->rx_page == NULL) { 636 /* 637 * This doesn't need to be atomic, but I think it should 638 * improve performance if it is. 639 */ 640 con->rx_page = alloc_page(GFP_ATOMIC); 641 if (con->rx_page == NULL) 642 goto out_resched; 643 cbuf_init(&con->cb, PAGE_SIZE); 644 } 645 646 /* 647 * iov[0] is the bit of the circular buffer between the current end 648 * point (cb.base + cb.len) and the end of the buffer. 649 */ 650 iov[0].iov_len = con->cb.base - cbuf_data(&con->cb); 651 iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb); 652 iov[1].iov_len = 0; 653 nvec = 1; 654 655 /* 656 * iov[1] is the bit of the circular buffer between the start of the 657 * buffer and the start of the currently used section (cb.base) 658 */ 659 if (cbuf_data(&con->cb) >= con->cb.base) { 660 iov[0].iov_len = PAGE_SIZE - cbuf_data(&con->cb); 661 iov[1].iov_len = con->cb.base; 662 iov[1].iov_base = page_address(con->rx_page); 663 nvec = 2; 664 } 665 len = iov[0].iov_len + iov[1].iov_len; 666 667 r = ret = kernel_recvmsg(con->sock, &msg, iov, nvec, len, 668 MSG_DONTWAIT | MSG_NOSIGNAL); 669 if (ret <= 0) 670 goto out_close; 671 else if (ret == len) 672 call_again_soon = 1; 673 674 cbuf_add(&con->cb, ret); 675 ret = dlm_process_incoming_buffer(con->nodeid, 676 page_address(con->rx_page), 677 con->cb.base, con->cb.len, 678 PAGE_SIZE); 679 if (ret == -EBADMSG) { 680 log_print("lowcomms: addr=%p, base=%u, len=%u, read=%d", 681 page_address(con->rx_page), con->cb.base, 682 con->cb.len, r); 683 } 684 if (ret < 0) 685 goto out_close; 686 cbuf_eat(&con->cb, ret); 687 688 if (cbuf_empty(&con->cb) && !call_again_soon) { 689 __free_page(con->rx_page); 690 con->rx_page = NULL; 691 } 692 693 if (call_again_soon) 694 goto out_resched; 695 mutex_unlock(&con->sock_mutex); 696 return 0; 697 698 out_resched: 699 if (!test_and_set_bit(CF_READ_PENDING, &con->flags)) 700 queue_work(recv_workqueue, &con->rwork); 701 mutex_unlock(&con->sock_mutex); 702 return -EAGAIN; 703 704 out_close: 705 mutex_unlock(&con->sock_mutex); 706 if (ret != -EAGAIN) { 707 close_connection(con, true, true, false); 708 /* Reconnect when there is something to send */ 709 } 710 /* Don't return success if we really got EOF */ 711 if (ret == 0) 712 ret = -EAGAIN; 713 714 return ret; 715 } 716 717 /* Listening socket is busy, accept a connection */ 718 static int tcp_accept_from_sock(struct connection *con) 719 { 720 int result; 721 struct sockaddr_storage peeraddr; 722 struct socket *newsock; 723 int len; 724 int nodeid; 725 struct connection *newcon; 726 struct connection *addcon; 727 728 mutex_lock(&connections_lock); 729 if (!dlm_allow_conn) { 730 mutex_unlock(&connections_lock); 731 return -1; 732 } 733 mutex_unlock(&connections_lock); 734 735 mutex_lock_nested(&con->sock_mutex, 0); 736 737 if (!con->sock) { 738 mutex_unlock(&con->sock_mutex); 739 return -ENOTCONN; 740 } 741 742 result = kernel_accept(con->sock, &newsock, O_NONBLOCK); 743 if (result < 0) 744 goto accept_err; 745 746 /* Get the connected socket's peer */ 747 memset(&peeraddr, 0, sizeof(peeraddr)); 748 if (newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 749 &len, 2)) { 750 result = -ECONNABORTED; 751 goto accept_err; 752 } 753 754 /* Get the new node's NODEID */ 755 make_sockaddr(&peeraddr, 0, &len); 756 if (addr_to_nodeid(&peeraddr, &nodeid)) { 757 unsigned char *b=(unsigned char *)&peeraddr; 758 log_print("connect from non cluster node"); 759 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 760 b, sizeof(struct sockaddr_storage)); 761 sock_release(newsock); 762 mutex_unlock(&con->sock_mutex); 763 return -1; 764 } 765 766 log_print("got connection from %d", nodeid); 767 768 /* Check to see if we already have a connection to this node. This 769 * could happen if the two nodes initiate a connection at roughly 770 * the same time and the connections cross on the wire. 771 * In this case we store the incoming one in "othercon" 772 */ 773 newcon = nodeid2con(nodeid, GFP_NOFS); 774 if (!newcon) { 775 result = -ENOMEM; 776 goto accept_err; 777 } 778 mutex_lock_nested(&newcon->sock_mutex, 1); 779 if (newcon->sock) { 780 struct connection *othercon = newcon->othercon; 781 782 if (!othercon) { 783 othercon = kmem_cache_zalloc(con_cache, GFP_NOFS); 784 if (!othercon) { 785 log_print("failed to allocate incoming socket"); 786 mutex_unlock(&newcon->sock_mutex); 787 result = -ENOMEM; 788 goto accept_err; 789 } 790 othercon->nodeid = nodeid; 791 othercon->rx_action = receive_from_sock; 792 mutex_init(&othercon->sock_mutex); 793 INIT_WORK(&othercon->swork, process_send_sockets); 794 INIT_WORK(&othercon->rwork, process_recv_sockets); 795 set_bit(CF_IS_OTHERCON, &othercon->flags); 796 } 797 mutex_lock_nested(&othercon->sock_mutex, 2); 798 if (!othercon->sock) { 799 newcon->othercon = othercon; 800 othercon->sock = newsock; 801 newsock->sk->sk_user_data = othercon; 802 add_sock(newsock, othercon); 803 addcon = othercon; 804 mutex_unlock(&othercon->sock_mutex); 805 } 806 else { 807 printk("Extra connection from node %d attempted\n", nodeid); 808 result = -EAGAIN; 809 mutex_unlock(&othercon->sock_mutex); 810 mutex_unlock(&newcon->sock_mutex); 811 goto accept_err; 812 } 813 } 814 else { 815 newsock->sk->sk_user_data = newcon; 816 newcon->rx_action = receive_from_sock; 817 /* accept copies the sk after we've saved the callbacks, so we 818 don't want to save them a second time or comm errors will 819 result in calling sk_error_report recursively. */ 820 add_sock(newsock, newcon); 821 addcon = newcon; 822 } 823 824 mutex_unlock(&newcon->sock_mutex); 825 826 /* 827 * Add it to the active queue in case we got data 828 * between processing the accept adding the socket 829 * to the read_sockets list 830 */ 831 if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags)) 832 queue_work(recv_workqueue, &addcon->rwork); 833 mutex_unlock(&con->sock_mutex); 834 835 return 0; 836 837 accept_err: 838 mutex_unlock(&con->sock_mutex); 839 if (newsock) 840 sock_release(newsock); 841 842 if (result != -EAGAIN) 843 log_print("error accepting connection from node: %d", result); 844 return result; 845 } 846 847 static int sctp_accept_from_sock(struct connection *con) 848 { 849 /* Check that the new node is in the lockspace */ 850 struct sctp_prim prim; 851 int nodeid; 852 int prim_len, ret; 853 int addr_len; 854 struct connection *newcon; 855 struct connection *addcon; 856 struct socket *newsock; 857 858 mutex_lock(&connections_lock); 859 if (!dlm_allow_conn) { 860 mutex_unlock(&connections_lock); 861 return -1; 862 } 863 mutex_unlock(&connections_lock); 864 865 mutex_lock_nested(&con->sock_mutex, 0); 866 867 ret = kernel_accept(con->sock, &newsock, O_NONBLOCK); 868 if (ret < 0) 869 goto accept_err; 870 871 memset(&prim, 0, sizeof(struct sctp_prim)); 872 prim_len = sizeof(struct sctp_prim); 873 874 ret = kernel_getsockopt(newsock, IPPROTO_SCTP, SCTP_PRIMARY_ADDR, 875 (char *)&prim, &prim_len); 876 if (ret < 0) { 877 log_print("getsockopt/sctp_primary_addr failed: %d", ret); 878 goto accept_err; 879 } 880 881 make_sockaddr(&prim.ssp_addr, 0, &addr_len); 882 ret = addr_to_nodeid(&prim.ssp_addr, &nodeid); 883 if (ret) { 884 unsigned char *b = (unsigned char *)&prim.ssp_addr; 885 886 log_print("reject connect from unknown addr"); 887 print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 888 b, sizeof(struct sockaddr_storage)); 889 goto accept_err; 890 } 891 892 newcon = nodeid2con(nodeid, GFP_NOFS); 893 if (!newcon) { 894 ret = -ENOMEM; 895 goto accept_err; 896 } 897 898 mutex_lock_nested(&newcon->sock_mutex, 1); 899 900 if (newcon->sock) { 901 struct connection *othercon = newcon->othercon; 902 903 if (!othercon) { 904 othercon = kmem_cache_zalloc(con_cache, GFP_NOFS); 905 if (!othercon) { 906 log_print("failed to allocate incoming socket"); 907 mutex_unlock(&newcon->sock_mutex); 908 ret = -ENOMEM; 909 goto accept_err; 910 } 911 othercon->nodeid = nodeid; 912 othercon->rx_action = receive_from_sock; 913 mutex_init(&othercon->sock_mutex); 914 INIT_WORK(&othercon->swork, process_send_sockets); 915 INIT_WORK(&othercon->rwork, process_recv_sockets); 916 set_bit(CF_IS_OTHERCON, &othercon->flags); 917 } 918 mutex_lock_nested(&othercon->sock_mutex, 2); 919 if (!othercon->sock) { 920 newcon->othercon = othercon; 921 othercon->sock = newsock; 922 newsock->sk->sk_user_data = othercon; 923 add_sock(newsock, othercon); 924 addcon = othercon; 925 mutex_unlock(&othercon->sock_mutex); 926 } else { 927 printk("Extra connection from node %d attempted\n", nodeid); 928 ret = -EAGAIN; 929 mutex_unlock(&othercon->sock_mutex); 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, true, 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 clear_bit(CF_WRITE_PENDING, &con->flags); 1597 if (con->sock == NULL) /* not mutex protected so check it inside too */ 1598 con->connect_action(con); 1599 if (!list_empty(&con->writequeue)) 1600 send_to_sock(con); 1601 } 1602 1603 1604 /* Discard all entries on the write queues */ 1605 static void clean_writequeues(void) 1606 { 1607 foreach_conn(clean_one_writequeue); 1608 } 1609 1610 static void work_stop(void) 1611 { 1612 destroy_workqueue(recv_workqueue); 1613 destroy_workqueue(send_workqueue); 1614 } 1615 1616 static int work_start(void) 1617 { 1618 recv_workqueue = alloc_workqueue("dlm_recv", 1619 WQ_UNBOUND | WQ_MEM_RECLAIM, 1); 1620 if (!recv_workqueue) { 1621 log_print("can't start dlm_recv"); 1622 return -ENOMEM; 1623 } 1624 1625 send_workqueue = alloc_workqueue("dlm_send", 1626 WQ_UNBOUND | WQ_MEM_RECLAIM, 1); 1627 if (!send_workqueue) { 1628 log_print("can't start dlm_send"); 1629 destroy_workqueue(recv_workqueue); 1630 return -ENOMEM; 1631 } 1632 1633 return 0; 1634 } 1635 1636 static void _stop_conn(struct connection *con, bool and_other) 1637 { 1638 mutex_lock(&con->sock_mutex); 1639 set_bit(CF_CLOSE, &con->flags); 1640 set_bit(CF_READ_PENDING, &con->flags); 1641 set_bit(CF_WRITE_PENDING, &con->flags); 1642 if (con->sock && con->sock->sk) 1643 con->sock->sk->sk_user_data = NULL; 1644 if (con->othercon && and_other) 1645 _stop_conn(con->othercon, false); 1646 mutex_unlock(&con->sock_mutex); 1647 } 1648 1649 static void stop_conn(struct connection *con) 1650 { 1651 _stop_conn(con, true); 1652 } 1653 1654 static void free_conn(struct connection *con) 1655 { 1656 close_connection(con, true, true, true); 1657 if (con->othercon) 1658 kmem_cache_free(con_cache, con->othercon); 1659 hlist_del(&con->list); 1660 kmem_cache_free(con_cache, con); 1661 } 1662 1663 static void work_flush(void) 1664 { 1665 int ok; 1666 int i; 1667 struct hlist_node *n; 1668 struct connection *con; 1669 1670 flush_workqueue(recv_workqueue); 1671 flush_workqueue(send_workqueue); 1672 do { 1673 ok = 1; 1674 foreach_conn(stop_conn); 1675 flush_workqueue(recv_workqueue); 1676 flush_workqueue(send_workqueue); 1677 for (i = 0; i < CONN_HASH_SIZE && ok; i++) { 1678 hlist_for_each_entry_safe(con, n, 1679 &connection_hash[i], list) { 1680 ok &= test_bit(CF_READ_PENDING, &con->flags); 1681 ok &= test_bit(CF_WRITE_PENDING, &con->flags); 1682 if (con->othercon) { 1683 ok &= test_bit(CF_READ_PENDING, 1684 &con->othercon->flags); 1685 ok &= test_bit(CF_WRITE_PENDING, 1686 &con->othercon->flags); 1687 } 1688 } 1689 } 1690 } while (!ok); 1691 } 1692 1693 void dlm_lowcomms_stop(void) 1694 { 1695 /* Set all the flags to prevent any 1696 socket activity. 1697 */ 1698 mutex_lock(&connections_lock); 1699 dlm_allow_conn = 0; 1700 mutex_unlock(&connections_lock); 1701 work_flush(); 1702 clean_writequeues(); 1703 foreach_conn(free_conn); 1704 work_stop(); 1705 1706 kmem_cache_destroy(con_cache); 1707 } 1708 1709 int dlm_lowcomms_start(void) 1710 { 1711 int error = -EINVAL; 1712 struct connection *con; 1713 int i; 1714 1715 for (i = 0; i < CONN_HASH_SIZE; i++) 1716 INIT_HLIST_HEAD(&connection_hash[i]); 1717 1718 init_local(); 1719 if (!dlm_local_count) { 1720 error = -ENOTCONN; 1721 log_print("no local IP address has been set"); 1722 goto fail; 1723 } 1724 1725 error = -ENOMEM; 1726 con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection), 1727 __alignof__(struct connection), 0, 1728 NULL); 1729 if (!con_cache) 1730 goto fail; 1731 1732 error = work_start(); 1733 if (error) 1734 goto fail_destroy; 1735 1736 dlm_allow_conn = 1; 1737 1738 /* Start listening */ 1739 if (dlm_config.ci_protocol == 0) 1740 error = tcp_listen_for_all(); 1741 else 1742 error = sctp_listen_for_all(); 1743 if (error) 1744 goto fail_unlisten; 1745 1746 return 0; 1747 1748 fail_unlisten: 1749 dlm_allow_conn = 0; 1750 con = nodeid2con(0,0); 1751 if (con) { 1752 close_connection(con, false, true, true); 1753 kmem_cache_free(con_cache, con); 1754 } 1755 fail_destroy: 1756 kmem_cache_destroy(con_cache); 1757 fail: 1758 return error; 1759 } 1760 1761 void dlm_lowcomms_exit(void) 1762 { 1763 struct dlm_node_addr *na, *safe; 1764 1765 spin_lock(&dlm_node_addrs_spin); 1766 list_for_each_entry_safe(na, safe, &dlm_node_addrs, list) { 1767 list_del(&na->list); 1768 while (na->addr_count--) 1769 kfree(na->addr[na->addr_count]); 1770 kfree(na); 1771 } 1772 spin_unlock(&dlm_node_addrs_spin); 1773 } 1774