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