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