linux/fs/dlm/lowcomms.c
<<
>>
Prefs
   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
  69struct cbuf {
  70        unsigned int base;
  71        unsigned int len;
  72        unsigned int mask;
  73};
  74
  75static void cbuf_add(struct cbuf *cb, int n)
  76{
  77        cb->len += n;
  78}
  79
  80static int cbuf_data(struct cbuf *cb)
  81{
  82        return ((cb->base + cb->len) & cb->mask);
  83}
  84
  85static void cbuf_init(struct cbuf *cb, int size)
  86{
  87        cb->base = cb->len = 0;
  88        cb->mask = size-1;
  89}
  90
  91static void cbuf_eat(struct cbuf *cb, int n)
  92{
  93        cb->len  -= n;
  94        cb->base += n;
  95        cb->base &= cb->mask;
  96}
  97
  98static bool cbuf_empty(struct cbuf *cb)
  99{
 100        return cb->len == 0;
 101}
 102
 103struct 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 */
 131struct 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
 141struct 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
 149static 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
 156static LIST_HEAD(dlm_node_addrs);
 157static DEFINE_SPINLOCK(dlm_node_addrs_spin);
 158
 159static struct sockaddr_storage *dlm_local_addr[DLM_MAX_ADDR_COUNT];
 160static int dlm_local_count;
 161static int dlm_allow_conn;
 162
 163/* Work queues */
 164static struct workqueue_struct *recv_workqueue;
 165static struct workqueue_struct *send_workqueue;
 166
 167static struct hlist_head connection_hash[CONN_HASH_SIZE];
 168static DEFINE_MUTEX(connections_lock);
 169static struct kmem_cache *con_cache;
 170
 171static void process_recv_sockets(struct work_struct *work);
 172static 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 */
 178static inline int nodeid_hash(int nodeid)
 179{
 180        return nodeid & (CONN_HASH_SIZE-1);
 181}
 182
 183static 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 */
 201static 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 */
 237static 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
 249static 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
 260static 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
 271static 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
 298static 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
 346static 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        }
 365unlock:
 366        spin_unlock(&dlm_node_addrs_spin);
 367        return rv;
 368}
 369
 370int 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 */
 412static void lowcomms_data_ready(struct sock *sk)
 413{
 414        struct connection *con;
 415
 416        read_lock_bh(&sk->sk_callback_lock);
 417        con = sock2con(sk);
 418        if (con && !test_and_set_bit(CF_READ_PENDING, &con->flags))
 419                queue_work(recv_workqueue, &con->rwork);
 420        read_unlock_bh(&sk->sk_callback_lock);
 421}
 422
 423static void lowcomms_write_space(struct sock *sk)
 424{
 425        struct connection *con;
 426
 427        read_lock_bh(&sk->sk_callback_lock);
 428        con = sock2con(sk);
 429        if (!con)
 430                goto out;
 431
 432        clear_bit(SOCK_NOSPACE, &con->sock->flags);
 433
 434        if (test_and_clear_bit(CF_APP_LIMITED, &con->flags)) {
 435                con->sock->sk->sk_write_pending--;
 436                clear_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags);
 437        }
 438
 439        queue_work(send_workqueue, &con->swork);
 440out:
 441        read_unlock_bh(&sk->sk_callback_lock);
 442}
 443
 444static inline void lowcomms_connect_sock(struct connection *con)
 445{
 446        if (test_bit(CF_CLOSE, &con->flags))
 447                return;
 448        queue_work(send_workqueue, &con->swork);
 449        cond_resched();
 450}
 451
 452static void lowcomms_state_change(struct sock *sk)
 453{
 454        /* SCTP layer is not calling sk_data_ready when the connection
 455         * is done, so we catch the signal through here. Also, it
 456         * doesn't switch socket state when entering shutdown, so we
 457         * skip the write in that case.
 458         */
 459        if (sk->sk_shutdown) {
 460                if (sk->sk_shutdown == RCV_SHUTDOWN)
 461                        lowcomms_data_ready(sk);
 462        } else if (sk->sk_state == TCP_ESTABLISHED) {
 463                lowcomms_write_space(sk);
 464        }
 465}
 466
 467int dlm_lowcomms_connect_node(int nodeid)
 468{
 469        struct connection *con;
 470
 471        if (nodeid == dlm_our_nodeid())
 472                return 0;
 473
 474        con = nodeid2con(nodeid, GFP_NOFS);
 475        if (!con)
 476                return -ENOMEM;
 477        lowcomms_connect_sock(con);
 478        return 0;
 479}
 480
 481static void lowcomms_error_report(struct sock *sk)
 482{
 483        struct connection *con;
 484        struct sockaddr_storage saddr;
 485        void (*orig_report)(struct sock *) = NULL;
 486
 487        read_lock_bh(&sk->sk_callback_lock);
 488        con = sock2con(sk);
 489        if (con == NULL)
 490                goto out;
 491
 492        orig_report = listen_sock.sk_error_report;
 493        if (con->sock == NULL ||
 494            kernel_getpeername(con->sock, (struct sockaddr *)&saddr) < 0) {
 495                printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
 496                                   "sending to node %d, port %d, "
 497                                   "sk_err=%d/%d\n", dlm_our_nodeid(),
 498                                   con->nodeid, dlm_config.ci_tcp_port,
 499                                   sk->sk_err, sk->sk_err_soft);
 500        } else if (saddr.ss_family == AF_INET) {
 501                struct sockaddr_in *sin4 = (struct sockaddr_in *)&saddr;
 502
 503                printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
 504                                   "sending to node %d at %pI4, port %d, "
 505                                   "sk_err=%d/%d\n", dlm_our_nodeid(),
 506                                   con->nodeid, &sin4->sin_addr.s_addr,
 507                                   dlm_config.ci_tcp_port, sk->sk_err,
 508                                   sk->sk_err_soft);
 509        } else {
 510                struct sockaddr_in6 *sin6 = (struct sockaddr_in6 *)&saddr;
 511
 512                printk_ratelimited(KERN_ERR "dlm: node %d: socket error "
 513                                   "sending to node %d at %u.%u.%u.%u, "
 514                                   "port %d, sk_err=%d/%d\n", dlm_our_nodeid(),
 515                                   con->nodeid, sin6->sin6_addr.s6_addr32[0],
 516                                   sin6->sin6_addr.s6_addr32[1],
 517                                   sin6->sin6_addr.s6_addr32[2],
 518                                   sin6->sin6_addr.s6_addr32[3],
 519                                   dlm_config.ci_tcp_port, sk->sk_err,
 520                                   sk->sk_err_soft);
 521        }
 522out:
 523        read_unlock_bh(&sk->sk_callback_lock);
 524        if (orig_report)
 525                orig_report(sk);
 526}
 527
 528/* Note: sk_callback_lock must be locked before calling this function. */
 529static void save_listen_callbacks(struct socket *sock)
 530{
 531        struct sock *sk = sock->sk;
 532
 533        listen_sock.sk_data_ready = sk->sk_data_ready;
 534        listen_sock.sk_state_change = sk->sk_state_change;
 535        listen_sock.sk_write_space = sk->sk_write_space;
 536        listen_sock.sk_error_report = sk->sk_error_report;
 537}
 538
 539static void restore_callbacks(struct socket *sock)
 540{
 541        struct sock *sk = sock->sk;
 542
 543        write_lock_bh(&sk->sk_callback_lock);
 544        sk->sk_user_data = NULL;
 545        sk->sk_data_ready = listen_sock.sk_data_ready;
 546        sk->sk_state_change = listen_sock.sk_state_change;
 547        sk->sk_write_space = listen_sock.sk_write_space;
 548        sk->sk_error_report = listen_sock.sk_error_report;
 549        write_unlock_bh(&sk->sk_callback_lock);
 550}
 551
 552/* Make a socket active */
 553static void add_sock(struct socket *sock, struct connection *con)
 554{
 555        struct sock *sk = sock->sk;
 556
 557        write_lock_bh(&sk->sk_callback_lock);
 558        con->sock = sock;
 559
 560        sk->sk_user_data = con;
 561        /* Install a data_ready callback */
 562        sk->sk_data_ready = lowcomms_data_ready;
 563        sk->sk_write_space = lowcomms_write_space;
 564        sk->sk_state_change = lowcomms_state_change;
 565        sk->sk_allocation = GFP_NOFS;
 566        sk->sk_error_report = lowcomms_error_report;
 567        write_unlock_bh(&sk->sk_callback_lock);
 568}
 569
 570/* Add the port number to an IPv6 or 4 sockaddr and return the address
 571   length */
 572static void make_sockaddr(struct sockaddr_storage *saddr, uint16_t port,
 573                          int *addr_len)
 574{
 575        saddr->ss_family =  dlm_local_addr[0]->ss_family;
 576        if (saddr->ss_family == AF_INET) {
 577                struct sockaddr_in *in4_addr = (struct sockaddr_in *)saddr;
 578                in4_addr->sin_port = cpu_to_be16(port);
 579                *addr_len = sizeof(struct sockaddr_in);
 580                memset(&in4_addr->sin_zero, 0, sizeof(in4_addr->sin_zero));
 581        } else {
 582                struct sockaddr_in6 *in6_addr = (struct sockaddr_in6 *)saddr;
 583                in6_addr->sin6_port = cpu_to_be16(port);
 584                *addr_len = sizeof(struct sockaddr_in6);
 585        }
 586        memset((char *)saddr + *addr_len, 0, sizeof(struct sockaddr_storage) - *addr_len);
 587}
 588
 589/* Close a remote connection and tidy up */
 590static void close_connection(struct connection *con, bool and_other,
 591                             bool tx, bool rx)
 592{
 593        bool closing = test_and_set_bit(CF_CLOSING, &con->flags);
 594
 595        if (tx && !closing && cancel_work_sync(&con->swork)) {
 596                log_print("canceled swork for node %d", con->nodeid);
 597                clear_bit(CF_WRITE_PENDING, &con->flags);
 598        }
 599        if (rx && !closing && cancel_work_sync(&con->rwork)) {
 600                log_print("canceled rwork for node %d", con->nodeid);
 601                clear_bit(CF_READ_PENDING, &con->flags);
 602        }
 603
 604        mutex_lock(&con->sock_mutex);
 605        if (con->sock) {
 606                restore_callbacks(con->sock);
 607                sock_release(con->sock);
 608                con->sock = NULL;
 609        }
 610        if (con->othercon && and_other) {
 611                /* Will only re-enter once. */
 612                close_connection(con->othercon, false, true, true);
 613        }
 614        if (con->rx_page) {
 615                __free_page(con->rx_page);
 616                con->rx_page = NULL;
 617        }
 618
 619        con->retries = 0;
 620        mutex_unlock(&con->sock_mutex);
 621        clear_bit(CF_CLOSING, &con->flags);
 622}
 623
 624/* Data received from remote end */
 625static int receive_from_sock(struct connection *con)
 626{
 627        int ret = 0;
 628        struct msghdr msg = {};
 629        struct kvec iov[2];
 630        unsigned len;
 631        int r;
 632        int call_again_soon = 0;
 633        int nvec;
 634
 635        mutex_lock(&con->sock_mutex);
 636
 637        if (con->sock == NULL) {
 638                ret = -EAGAIN;
 639                goto out_close;
 640        }
 641        if (con->nodeid == 0) {
 642                ret = -EINVAL;
 643                goto out_close;
 644        }
 645
 646        if (con->rx_page == NULL) {
 647                /*
 648                 * This doesn't need to be atomic, but I think it should
 649                 * improve performance if it is.
 650                 */
 651                con->rx_page = alloc_page(GFP_ATOMIC);
 652                if (con->rx_page == NULL)
 653                        goto out_resched;
 654                cbuf_init(&con->cb, PAGE_SIZE);
 655        }
 656
 657        /*
 658         * iov[0] is the bit of the circular buffer between the current end
 659         * point (cb.base + cb.len) and the end of the buffer.
 660         */
 661        iov[0].iov_len = con->cb.base - cbuf_data(&con->cb);
 662        iov[0].iov_base = page_address(con->rx_page) + cbuf_data(&con->cb);
 663        iov[1].iov_len = 0;
 664        nvec = 1;
 665
 666        /*
 667         * iov[1] is the bit of the circular buffer between the start of the
 668         * buffer and the start of the currently used section (cb.base)
 669         */
 670        if (cbuf_data(&con->cb) >= con->cb.base) {
 671                iov[0].iov_len = PAGE_SIZE - cbuf_data(&con->cb);
 672                iov[1].iov_len = con->cb.base;
 673                iov[1].iov_base = page_address(con->rx_page);
 674                nvec = 2;
 675        }
 676        len = iov[0].iov_len + iov[1].iov_len;
 677        iov_iter_kvec(&msg.msg_iter, READ | ITER_KVEC, iov, nvec, len);
 678
 679        r = ret = sock_recvmsg(con->sock, &msg, MSG_DONTWAIT | MSG_NOSIGNAL);
 680        if (ret <= 0)
 681                goto out_close;
 682        else if (ret == len)
 683                call_again_soon = 1;
 684
 685        cbuf_add(&con->cb, ret);
 686        ret = dlm_process_incoming_buffer(con->nodeid,
 687                                          page_address(con->rx_page),
 688                                          con->cb.base, con->cb.len,
 689                                          PAGE_SIZE);
 690        if (ret == -EBADMSG) {
 691                log_print("lowcomms: addr=%p, base=%u, len=%u, read=%d",
 692                          page_address(con->rx_page), con->cb.base,
 693                          con->cb.len, r);
 694        }
 695        if (ret < 0)
 696                goto out_close;
 697        cbuf_eat(&con->cb, ret);
 698
 699        if (cbuf_empty(&con->cb) && !call_again_soon) {
 700                __free_page(con->rx_page);
 701                con->rx_page = NULL;
 702        }
 703
 704        if (call_again_soon)
 705                goto out_resched;
 706        mutex_unlock(&con->sock_mutex);
 707        return 0;
 708
 709out_resched:
 710        if (!test_and_set_bit(CF_READ_PENDING, &con->flags))
 711                queue_work(recv_workqueue, &con->rwork);
 712        mutex_unlock(&con->sock_mutex);
 713        return -EAGAIN;
 714
 715out_close:
 716        mutex_unlock(&con->sock_mutex);
 717        if (ret != -EAGAIN) {
 718                close_connection(con, true, true, false);
 719                /* Reconnect when there is something to send */
 720        }
 721        /* Don't return success if we really got EOF */
 722        if (ret == 0)
 723                ret = -EAGAIN;
 724
 725        return ret;
 726}
 727
 728/* Listening socket is busy, accept a connection */
 729static int tcp_accept_from_sock(struct connection *con)
 730{
 731        int result;
 732        struct sockaddr_storage peeraddr;
 733        struct socket *newsock;
 734        int len;
 735        int nodeid;
 736        struct connection *newcon;
 737        struct connection *addcon;
 738
 739        mutex_lock(&connections_lock);
 740        if (!dlm_allow_conn) {
 741                mutex_unlock(&connections_lock);
 742                return -1;
 743        }
 744        mutex_unlock(&connections_lock);
 745
 746        mutex_lock_nested(&con->sock_mutex, 0);
 747
 748        if (!con->sock) {
 749                mutex_unlock(&con->sock_mutex);
 750                return -ENOTCONN;
 751        }
 752
 753        result = kernel_accept(con->sock, &newsock, O_NONBLOCK);
 754        if (result < 0)
 755                goto accept_err;
 756
 757        /* Get the connected socket's peer */
 758        memset(&peeraddr, 0, sizeof(peeraddr));
 759        len = newsock->ops->getname(newsock, (struct sockaddr *)&peeraddr, 2);
 760        if (len < 0) {
 761                result = -ECONNABORTED;
 762                goto accept_err;
 763        }
 764
 765        /* Get the new node's NODEID */
 766        make_sockaddr(&peeraddr, 0, &len);
 767        if (addr_to_nodeid(&peeraddr, &nodeid)) {
 768                unsigned char *b=(unsigned char *)&peeraddr;
 769                log_print("connect from non cluster node");
 770                print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE, 
 771                                     b, sizeof(struct sockaddr_storage));
 772                sock_release(newsock);
 773                mutex_unlock(&con->sock_mutex);
 774                return -1;
 775        }
 776
 777        log_print("got connection from %d", nodeid);
 778
 779        /*  Check to see if we already have a connection to this node. This
 780         *  could happen if the two nodes initiate a connection at roughly
 781         *  the same time and the connections cross on the wire.
 782         *  In this case we store the incoming one in "othercon"
 783         */
 784        newcon = nodeid2con(nodeid, GFP_NOFS);
 785        if (!newcon) {
 786                result = -ENOMEM;
 787                goto accept_err;
 788        }
 789        mutex_lock_nested(&newcon->sock_mutex, 1);
 790        if (newcon->sock) {
 791                struct connection *othercon = newcon->othercon;
 792
 793                if (!othercon) {
 794                        othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
 795                        if (!othercon) {
 796                                log_print("failed to allocate incoming socket");
 797                                mutex_unlock(&newcon->sock_mutex);
 798                                result = -ENOMEM;
 799                                goto accept_err;
 800                        }
 801                        othercon->nodeid = nodeid;
 802                        othercon->rx_action = receive_from_sock;
 803                        mutex_init(&othercon->sock_mutex);
 804                        INIT_LIST_HEAD(&othercon->writequeue);
 805                        spin_lock_init(&othercon->writequeue_lock);
 806                        INIT_WORK(&othercon->swork, process_send_sockets);
 807                        INIT_WORK(&othercon->rwork, process_recv_sockets);
 808                        set_bit(CF_IS_OTHERCON, &othercon->flags);
 809                }
 810                mutex_lock_nested(&othercon->sock_mutex, 2);
 811                if (!othercon->sock) {
 812                        newcon->othercon = othercon;
 813                        add_sock(newsock, othercon);
 814                        addcon = othercon;
 815                        mutex_unlock(&othercon->sock_mutex);
 816                }
 817                else {
 818                        printk("Extra connection from node %d attempted\n", nodeid);
 819                        result = -EAGAIN;
 820                        mutex_unlock(&othercon->sock_mutex);
 821                        mutex_unlock(&newcon->sock_mutex);
 822                        goto accept_err;
 823                }
 824        }
 825        else {
 826                newcon->rx_action = receive_from_sock;
 827                /* accept copies the sk after we've saved the callbacks, so we
 828                   don't want to save them a second time or comm errors will
 829                   result in calling sk_error_report recursively. */
 830                add_sock(newsock, newcon);
 831                addcon = newcon;
 832        }
 833
 834        mutex_unlock(&newcon->sock_mutex);
 835
 836        /*
 837         * Add it to the active queue in case we got data
 838         * between processing the accept adding the socket
 839         * to the read_sockets list
 840         */
 841        if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
 842                queue_work(recv_workqueue, &addcon->rwork);
 843        mutex_unlock(&con->sock_mutex);
 844
 845        return 0;
 846
 847accept_err:
 848        mutex_unlock(&con->sock_mutex);
 849        if (newsock)
 850                sock_release(newsock);
 851
 852        if (result != -EAGAIN)
 853                log_print("error accepting connection from node: %d", result);
 854        return result;
 855}
 856
 857static int sctp_accept_from_sock(struct connection *con)
 858{
 859        /* Check that the new node is in the lockspace */
 860        struct sctp_prim prim;
 861        int nodeid;
 862        int prim_len, ret;
 863        int addr_len;
 864        struct connection *newcon;
 865        struct connection *addcon;
 866        struct socket *newsock;
 867
 868        mutex_lock(&connections_lock);
 869        if (!dlm_allow_conn) {
 870                mutex_unlock(&connections_lock);
 871                return -1;
 872        }
 873        mutex_unlock(&connections_lock);
 874
 875        mutex_lock_nested(&con->sock_mutex, 0);
 876
 877        ret = kernel_accept(con->sock, &newsock, O_NONBLOCK);
 878        if (ret < 0)
 879                goto accept_err;
 880
 881        memset(&prim, 0, sizeof(struct sctp_prim));
 882        prim_len = sizeof(struct sctp_prim);
 883
 884        ret = kernel_getsockopt(newsock, IPPROTO_SCTP, SCTP_PRIMARY_ADDR,
 885                                (char *)&prim, &prim_len);
 886        if (ret < 0) {
 887                log_print("getsockopt/sctp_primary_addr failed: %d", ret);
 888                goto accept_err;
 889        }
 890
 891        make_sockaddr(&prim.ssp_addr, 0, &addr_len);
 892        ret = addr_to_nodeid(&prim.ssp_addr, &nodeid);
 893        if (ret) {
 894                unsigned char *b = (unsigned char *)&prim.ssp_addr;
 895
 896                log_print("reject connect from unknown addr");
 897                print_hex_dump_bytes("ss: ", DUMP_PREFIX_NONE,
 898                                     b, sizeof(struct sockaddr_storage));
 899                goto accept_err;
 900        }
 901
 902        newcon = nodeid2con(nodeid, GFP_NOFS);
 903        if (!newcon) {
 904                ret = -ENOMEM;
 905                goto accept_err;
 906        }
 907
 908        mutex_lock_nested(&newcon->sock_mutex, 1);
 909
 910        if (newcon->sock) {
 911                struct connection *othercon = newcon->othercon;
 912
 913                if (!othercon) {
 914                        othercon = kmem_cache_zalloc(con_cache, GFP_NOFS);
 915                        if (!othercon) {
 916                                log_print("failed to allocate incoming socket");
 917                                mutex_unlock(&newcon->sock_mutex);
 918                                ret = -ENOMEM;
 919                                goto accept_err;
 920                        }
 921                        othercon->nodeid = nodeid;
 922                        othercon->rx_action = receive_from_sock;
 923                        mutex_init(&othercon->sock_mutex);
 924                        INIT_LIST_HEAD(&othercon->writequeue);
 925                        spin_lock_init(&othercon->writequeue_lock);
 926                        INIT_WORK(&othercon->swork, process_send_sockets);
 927                        INIT_WORK(&othercon->rwork, process_recv_sockets);
 928                        set_bit(CF_IS_OTHERCON, &othercon->flags);
 929                }
 930                mutex_lock_nested(&othercon->sock_mutex, 2);
 931                if (!othercon->sock) {
 932                        newcon->othercon = othercon;
 933                        add_sock(newsock, othercon);
 934                        addcon = othercon;
 935                        mutex_unlock(&othercon->sock_mutex);
 936                } else {
 937                        printk("Extra connection from node %d attempted\n", nodeid);
 938                        ret = -EAGAIN;
 939                        mutex_unlock(&othercon->sock_mutex);
 940                        mutex_unlock(&newcon->sock_mutex);
 941                        goto accept_err;
 942                }
 943        } else {
 944                newcon->rx_action = receive_from_sock;
 945                add_sock(newsock, newcon);
 946                addcon = newcon;
 947        }
 948
 949        log_print("connected to %d", nodeid);
 950
 951        mutex_unlock(&newcon->sock_mutex);
 952
 953        /*
 954         * Add it to the active queue in case we got data
 955         * between processing the accept adding the socket
 956         * to the read_sockets list
 957         */
 958        if (!test_and_set_bit(CF_READ_PENDING, &addcon->flags))
 959                queue_work(recv_workqueue, &addcon->rwork);
 960        mutex_unlock(&con->sock_mutex);
 961
 962        return 0;
 963
 964accept_err:
 965        mutex_unlock(&con->sock_mutex);
 966        if (newsock)
 967                sock_release(newsock);
 968        if (ret != -EAGAIN)
 969                log_print("error accepting connection from node: %d", ret);
 970
 971        return ret;
 972}
 973
 974static void free_entry(struct writequeue_entry *e)
 975{
 976        __free_page(e->page);
 977        kfree(e);
 978}
 979
 980/*
 981 * writequeue_entry_complete - try to delete and free write queue entry
 982 * @e: write queue entry to try to delete
 983 * @completed: bytes completed
 984 *
 985 * writequeue_lock must be held.
 986 */
 987static void writequeue_entry_complete(struct writequeue_entry *e, int completed)
 988{
 989        e->offset += completed;
 990        e->len -= completed;
 991
 992        if (e->len == 0 && e->users == 0) {
 993                list_del(&e->list);
 994                free_entry(e);
 995        }
 996}
 997
 998/*
 999 * sctp_bind_addrs - bind a SCTP socket to all our addresses
1000 */
1001static int sctp_bind_addrs(struct connection *con, uint16_t port)
1002{
1003        struct sockaddr_storage localaddr;
1004        int i, addr_len, result = 0;
1005
1006        for (i = 0; i < dlm_local_count; i++) {
1007                memcpy(&localaddr, dlm_local_addr[i], sizeof(localaddr));
1008                make_sockaddr(&localaddr, port, &addr_len);
1009
1010                if (!i)
1011                        result = kernel_bind(con->sock,
1012                                             (struct sockaddr *)&localaddr,
1013                                             addr_len);
1014                else
1015                        result = kernel_setsockopt(con->sock, SOL_SCTP,
1016                                                   SCTP_SOCKOPT_BINDX_ADD,
1017                                                   (char *)&localaddr, addr_len);
1018
1019                if (result < 0) {
1020                        log_print("Can't bind to %d addr number %d, %d.\n",
1021                                  port, i + 1, result);
1022                        break;
1023                }
1024        }
1025        return result;
1026}
1027
1028/* Initiate an SCTP association.
1029   This is a special case of send_to_sock() in that we don't yet have a
1030   peeled-off socket for this association, so we use the listening socket
1031   and add the primary IP address of the remote node.
1032 */
1033static void sctp_connect_to_sock(struct connection *con)
1034{
1035        struct sockaddr_storage daddr;
1036        int one = 1;
1037        int result;
1038        int addr_len;
1039        struct socket *sock;
1040        struct timeval tv = { .tv_sec = 5, .tv_usec = 0 };
1041
1042        if (con->nodeid == 0) {
1043                log_print("attempt to connect sock 0 foiled");
1044                return;
1045        }
1046
1047        mutex_lock(&con->sock_mutex);
1048
1049        /* Some odd races can cause double-connects, ignore them */
1050        if (con->retries++ > MAX_CONNECT_RETRIES)
1051                goto out;
1052
1053        if (con->sock) {
1054                log_print("node %d already connected.", con->nodeid);
1055                goto out;
1056        }
1057
1058        memset(&daddr, 0, sizeof(daddr));
1059        result = nodeid_to_addr(con->nodeid, &daddr, NULL, true);
1060        if (result < 0) {
1061                log_print("no address for nodeid %d", con->nodeid);
1062                goto out;
1063        }
1064
1065        /* Create a socket to communicate with */
1066        result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1067                                  SOCK_STREAM, IPPROTO_SCTP, &sock);
1068        if (result < 0)
1069                goto socket_err;
1070
1071        con->rx_action = receive_from_sock;
1072        con->connect_action = sctp_connect_to_sock;
1073        add_sock(sock, con);
1074
1075        /* Bind to all addresses. */
1076        if (sctp_bind_addrs(con, 0))
1077                goto bind_err;
1078
1079        make_sockaddr(&daddr, dlm_config.ci_tcp_port, &addr_len);
1080
1081        log_print("connecting to %d", con->nodeid);
1082
1083        /* Turn off Nagle's algorithm */
1084        kernel_setsockopt(sock, SOL_SCTP, SCTP_NODELAY, (char *)&one,
1085                          sizeof(one));
1086
1087        /*
1088         * Make sock->ops->connect() function return in specified time,
1089         * since O_NONBLOCK argument in connect() function does not work here,
1090         * then, we should restore the default value of this attribute.
1091         */
1092        kernel_setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv,
1093                          sizeof(tv));
1094        result = sock->ops->connect(sock, (struct sockaddr *)&daddr, addr_len,
1095                                   0);
1096        memset(&tv, 0, sizeof(tv));
1097        kernel_setsockopt(sock, SOL_SOCKET, SO_SNDTIMEO, (char *)&tv,
1098                          sizeof(tv));
1099
1100        if (result == -EINPROGRESS)
1101                result = 0;
1102        if (result == 0)
1103                goto out;
1104
1105bind_err:
1106        con->sock = NULL;
1107        sock_release(sock);
1108
1109socket_err:
1110        /*
1111         * Some errors are fatal and this list might need adjusting. For other
1112         * errors we try again until the max number of retries is reached.
1113         */
1114        if (result != -EHOSTUNREACH &&
1115            result != -ENETUNREACH &&
1116            result != -ENETDOWN &&
1117            result != -EINVAL &&
1118            result != -EPROTONOSUPPORT) {
1119                log_print("connect %d try %d error %d", con->nodeid,
1120                          con->retries, result);
1121                mutex_unlock(&con->sock_mutex);
1122                msleep(1000);
1123                lowcomms_connect_sock(con);
1124                return;
1125        }
1126
1127out:
1128        mutex_unlock(&con->sock_mutex);
1129}
1130
1131/* Connect a new socket to its peer */
1132static void tcp_connect_to_sock(struct connection *con)
1133{
1134        struct sockaddr_storage saddr, src_addr;
1135        int addr_len;
1136        struct socket *sock = NULL;
1137        int one = 1;
1138        int result;
1139
1140        if (con->nodeid == 0) {
1141                log_print("attempt to connect sock 0 foiled");
1142                return;
1143        }
1144
1145        mutex_lock(&con->sock_mutex);
1146        if (con->retries++ > MAX_CONNECT_RETRIES)
1147                goto out;
1148
1149        /* Some odd races can cause double-connects, ignore them */
1150        if (con->sock)
1151                goto out;
1152
1153        /* Create a socket to communicate with */
1154        result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1155                                  SOCK_STREAM, IPPROTO_TCP, &sock);
1156        if (result < 0)
1157                goto out_err;
1158
1159        memset(&saddr, 0, sizeof(saddr));
1160        result = nodeid_to_addr(con->nodeid, &saddr, NULL, false);
1161        if (result < 0) {
1162                log_print("no address for nodeid %d", con->nodeid);
1163                goto out_err;
1164        }
1165
1166        con->rx_action = receive_from_sock;
1167        con->connect_action = tcp_connect_to_sock;
1168        add_sock(sock, con);
1169
1170        /* Bind to our cluster-known address connecting to avoid
1171           routing problems */
1172        memcpy(&src_addr, dlm_local_addr[0], sizeof(src_addr));
1173        make_sockaddr(&src_addr, 0, &addr_len);
1174        result = sock->ops->bind(sock, (struct sockaddr *) &src_addr,
1175                                 addr_len);
1176        if (result < 0) {
1177                log_print("could not bind for connect: %d", result);
1178                /* This *may* not indicate a critical error */
1179        }
1180
1181        make_sockaddr(&saddr, dlm_config.ci_tcp_port, &addr_len);
1182
1183        log_print("connecting to %d", con->nodeid);
1184
1185        /* Turn off Nagle's algorithm */
1186        kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1187                          sizeof(one));
1188
1189        result = sock->ops->connect(sock, (struct sockaddr *)&saddr, addr_len,
1190                                   O_NONBLOCK);
1191        if (result == -EINPROGRESS)
1192                result = 0;
1193        if (result == 0)
1194                goto out;
1195
1196out_err:
1197        if (con->sock) {
1198                sock_release(con->sock);
1199                con->sock = NULL;
1200        } else if (sock) {
1201                sock_release(sock);
1202        }
1203        /*
1204         * Some errors are fatal and this list might need adjusting. For other
1205         * errors we try again until the max number of retries is reached.
1206         */
1207        if (result != -EHOSTUNREACH &&
1208            result != -ENETUNREACH &&
1209            result != -ENETDOWN && 
1210            result != -EINVAL &&
1211            result != -EPROTONOSUPPORT) {
1212                log_print("connect %d try %d error %d", con->nodeid,
1213                          con->retries, result);
1214                mutex_unlock(&con->sock_mutex);
1215                msleep(1000);
1216                lowcomms_connect_sock(con);
1217                return;
1218        }
1219out:
1220        mutex_unlock(&con->sock_mutex);
1221        return;
1222}
1223
1224static struct socket *tcp_create_listen_sock(struct connection *con,
1225                                             struct sockaddr_storage *saddr)
1226{
1227        struct socket *sock = NULL;
1228        int result = 0;
1229        int one = 1;
1230        int addr_len;
1231
1232        if (dlm_local_addr[0]->ss_family == AF_INET)
1233                addr_len = sizeof(struct sockaddr_in);
1234        else
1235                addr_len = sizeof(struct sockaddr_in6);
1236
1237        /* Create a socket to communicate with */
1238        result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1239                                  SOCK_STREAM, IPPROTO_TCP, &sock);
1240        if (result < 0) {
1241                log_print("Can't create listening comms socket");
1242                goto create_out;
1243        }
1244
1245        /* Turn off Nagle's algorithm */
1246        kernel_setsockopt(sock, SOL_TCP, TCP_NODELAY, (char *)&one,
1247                          sizeof(one));
1248
1249        result = kernel_setsockopt(sock, SOL_SOCKET, SO_REUSEADDR,
1250                                   (char *)&one, sizeof(one));
1251
1252        if (result < 0) {
1253                log_print("Failed to set SO_REUSEADDR on socket: %d", result);
1254        }
1255        write_lock_bh(&sock->sk->sk_callback_lock);
1256        sock->sk->sk_user_data = con;
1257        save_listen_callbacks(sock);
1258        con->rx_action = tcp_accept_from_sock;
1259        con->connect_action = tcp_connect_to_sock;
1260        write_unlock_bh(&sock->sk->sk_callback_lock);
1261
1262        /* Bind to our port */
1263        make_sockaddr(saddr, dlm_config.ci_tcp_port, &addr_len);
1264        result = sock->ops->bind(sock, (struct sockaddr *) saddr, addr_len);
1265        if (result < 0) {
1266                log_print("Can't bind to port %d", dlm_config.ci_tcp_port);
1267                sock_release(sock);
1268                sock = NULL;
1269                con->sock = NULL;
1270                goto create_out;
1271        }
1272        result = kernel_setsockopt(sock, SOL_SOCKET, SO_KEEPALIVE,
1273                                 (char *)&one, sizeof(one));
1274        if (result < 0) {
1275                log_print("Set keepalive failed: %d", result);
1276        }
1277
1278        result = sock->ops->listen(sock, 5);
1279        if (result < 0) {
1280                log_print("Can't listen on port %d", dlm_config.ci_tcp_port);
1281                sock_release(sock);
1282                sock = NULL;
1283                goto create_out;
1284        }
1285
1286create_out:
1287        return sock;
1288}
1289
1290/* Get local addresses */
1291static void init_local(void)
1292{
1293        struct sockaddr_storage sas, *addr;
1294        int i;
1295
1296        dlm_local_count = 0;
1297        for (i = 0; i < DLM_MAX_ADDR_COUNT; i++) {
1298                if (dlm_our_addr(&sas, i))
1299                        break;
1300
1301                addr = kmemdup(&sas, sizeof(*addr), GFP_NOFS);
1302                if (!addr)
1303                        break;
1304                dlm_local_addr[dlm_local_count++] = addr;
1305        }
1306}
1307
1308/* Initialise SCTP socket and bind to all interfaces */
1309static int sctp_listen_for_all(void)
1310{
1311        struct socket *sock = NULL;
1312        int result = -EINVAL;
1313        struct connection *con = nodeid2con(0, GFP_NOFS);
1314        int bufsize = NEEDED_RMEM;
1315        int one = 1;
1316
1317        if (!con)
1318                return -ENOMEM;
1319
1320        log_print("Using SCTP for communications");
1321
1322        result = sock_create_kern(&init_net, dlm_local_addr[0]->ss_family,
1323                                  SOCK_STREAM, IPPROTO_SCTP, &sock);
1324        if (result < 0) {
1325                log_print("Can't create comms socket, check SCTP is loaded");
1326                goto out;
1327        }
1328
1329        result = kernel_setsockopt(sock, SOL_SOCKET, SO_RCVBUFFORCE,
1330                                 (char *)&bufsize, sizeof(bufsize));
1331        if (result)
1332                log_print("Error increasing buffer space on socket %d", result);
1333
1334        result = kernel_setsockopt(sock, SOL_SCTP, SCTP_NODELAY, (char *)&one,
1335                                   sizeof(one));
1336        if (result < 0)
1337                log_print("Could not set SCTP NODELAY error %d\n", result);
1338
1339        write_lock_bh(&sock->sk->sk_callback_lock);
1340        /* Init con struct */
1341        sock->sk->sk_user_data = con;
1342        save_listen_callbacks(sock);
1343        con->sock = sock;
1344        con->sock->sk->sk_data_ready = lowcomms_data_ready;
1345        con->rx_action = sctp_accept_from_sock;
1346        con->connect_action = sctp_connect_to_sock;
1347
1348        write_unlock_bh(&sock->sk->sk_callback_lock);
1349
1350        /* Bind to all addresses. */
1351        if (sctp_bind_addrs(con, dlm_config.ci_tcp_port))
1352                goto create_delsock;
1353
1354        result = sock->ops->listen(sock, 5);
1355        if (result < 0) {
1356                log_print("Can't set socket listening");
1357                goto create_delsock;
1358        }
1359
1360        return 0;
1361
1362create_delsock:
1363        sock_release(sock);
1364        con->sock = NULL;
1365out:
1366        return result;
1367}
1368
1369static int tcp_listen_for_all(void)
1370{
1371        struct socket *sock = NULL;
1372        struct connection *con = nodeid2con(0, GFP_NOFS);
1373        int result = -EINVAL;
1374
1375        if (!con)
1376                return -ENOMEM;
1377
1378        /* We don't support multi-homed hosts */
1379        if (dlm_local_addr[1] != NULL) {
1380                log_print("TCP protocol can't handle multi-homed hosts, "
1381                          "try SCTP");
1382                return -EINVAL;
1383        }
1384
1385        log_print("Using TCP for communications");
1386
1387        sock = tcp_create_listen_sock(con, dlm_local_addr[0]);
1388        if (sock) {
1389                add_sock(sock, con);
1390                result = 0;
1391        }
1392        else {
1393                result = -EADDRINUSE;
1394        }
1395
1396        return result;
1397}
1398
1399
1400
1401static struct writequeue_entry *new_writequeue_entry(struct connection *con,
1402                                                     gfp_t allocation)
1403{
1404        struct writequeue_entry *entry;
1405
1406        entry = kmalloc(sizeof(struct writequeue_entry), allocation);
1407        if (!entry)
1408                return NULL;
1409
1410        entry->page = alloc_page(allocation);
1411        if (!entry->page) {
1412                kfree(entry);
1413                return NULL;
1414        }
1415
1416        entry->offset = 0;
1417        entry->len = 0;
1418        entry->end = 0;
1419        entry->users = 0;
1420        entry->con = con;
1421
1422        return entry;
1423}
1424
1425void *dlm_lowcomms_get_buffer(int nodeid, int len, gfp_t allocation, char **ppc)
1426{
1427        struct connection *con;
1428        struct writequeue_entry *e;
1429        int offset = 0;
1430
1431        con = nodeid2con(nodeid, allocation);
1432        if (!con)
1433                return NULL;
1434
1435        spin_lock(&con->writequeue_lock);
1436        e = list_entry(con->writequeue.prev, struct writequeue_entry, list);
1437        if ((&e->list == &con->writequeue) ||
1438            (PAGE_SIZE - e->end < len)) {
1439                e = NULL;
1440        } else {
1441                offset = e->end;
1442                e->end += len;
1443                e->users++;
1444        }
1445        spin_unlock(&con->writequeue_lock);
1446
1447        if (e) {
1448        got_one:
1449                *ppc = page_address(e->page) + offset;
1450                return e;
1451        }
1452
1453        e = new_writequeue_entry(con, allocation);
1454        if (e) {
1455                spin_lock(&con->writequeue_lock);
1456                offset = e->end;
1457                e->end += len;
1458                e->users++;
1459                list_add_tail(&e->list, &con->writequeue);
1460                spin_unlock(&con->writequeue_lock);
1461                goto got_one;
1462        }
1463        return NULL;
1464}
1465
1466void dlm_lowcomms_commit_buffer(void *mh)
1467{
1468        struct writequeue_entry *e = (struct writequeue_entry *)mh;
1469        struct connection *con = e->con;
1470        int users;
1471
1472        spin_lock(&con->writequeue_lock);
1473        users = --e->users;
1474        if (users)
1475                goto out;
1476        e->len = e->end - e->offset;
1477        spin_unlock(&con->writequeue_lock);
1478
1479        queue_work(send_workqueue, &con->swork);
1480        return;
1481
1482out:
1483        spin_unlock(&con->writequeue_lock);
1484        return;
1485}
1486
1487/* Send a message */
1488static void send_to_sock(struct connection *con)
1489{
1490        int ret = 0;
1491        const int msg_flags = MSG_DONTWAIT | MSG_NOSIGNAL;
1492        struct writequeue_entry *e;
1493        int len, offset;
1494        int count = 0;
1495
1496        mutex_lock(&con->sock_mutex);
1497        if (con->sock == NULL)
1498                goto out_connect;
1499
1500        spin_lock(&con->writequeue_lock);
1501        for (;;) {
1502                e = list_entry(con->writequeue.next, struct writequeue_entry,
1503                               list);
1504                if ((struct list_head *) e == &con->writequeue)
1505                        break;
1506
1507                len = e->len;
1508                offset = e->offset;
1509                BUG_ON(len == 0 && e->users == 0);
1510                spin_unlock(&con->writequeue_lock);
1511
1512                ret = 0;
1513                if (len) {
1514                        ret = kernel_sendpage(con->sock, e->page, offset, len,
1515                                              msg_flags);
1516                        if (ret == -EAGAIN || ret == 0) {
1517                                if (ret == -EAGAIN &&
1518                                    test_bit(SOCKWQ_ASYNC_NOSPACE, &con->sock->flags) &&
1519                                    !test_and_set_bit(CF_APP_LIMITED, &con->flags)) {
1520                                        /* Notify TCP that we're limited by the
1521                                         * application window size.
1522                                         */
1523                                        set_bit(SOCK_NOSPACE, &con->sock->flags);
1524                                        con->sock->sk->sk_write_pending++;
1525                                }
1526                                cond_resched();
1527                                goto out;
1528                        } else if (ret < 0)
1529                                goto send_error;
1530                }
1531
1532                /* Don't starve people filling buffers */
1533                if (++count >= MAX_SEND_MSG_COUNT) {
1534                        cond_resched();
1535                        count = 0;
1536                }
1537
1538                spin_lock(&con->writequeue_lock);
1539                writequeue_entry_complete(e, ret);
1540        }
1541        spin_unlock(&con->writequeue_lock);
1542out:
1543        mutex_unlock(&con->sock_mutex);
1544        return;
1545
1546send_error:
1547        mutex_unlock(&con->sock_mutex);
1548        close_connection(con, true, false, true);
1549        /* Requeue the send work. When the work daemon runs again, it will try
1550           a new connection, then call this function again. */
1551        queue_work(send_workqueue, &con->swork);
1552        return;
1553
1554out_connect:
1555        mutex_unlock(&con->sock_mutex);
1556        queue_work(send_workqueue, &con->swork);
1557        cond_resched();
1558}
1559
1560static void clean_one_writequeue(struct connection *con)
1561{
1562        struct writequeue_entry *e, *safe;
1563
1564        spin_lock(&con->writequeue_lock);
1565        list_for_each_entry_safe(e, safe, &con->writequeue, list) {
1566                list_del(&e->list);
1567                free_entry(e);
1568        }
1569        spin_unlock(&con->writequeue_lock);
1570}
1571
1572/* Called from recovery when it knows that a node has
1573   left the cluster */
1574int dlm_lowcomms_close(int nodeid)
1575{
1576        struct connection *con;
1577        struct dlm_node_addr *na;
1578
1579        log_print("closing connection to node %d", nodeid);
1580        con = nodeid2con(nodeid, 0);
1581        if (con) {
1582                set_bit(CF_CLOSE, &con->flags);
1583                close_connection(con, true, true, true);
1584                clean_one_writequeue(con);
1585        }
1586
1587        spin_lock(&dlm_node_addrs_spin);
1588        na = find_node_addr(nodeid);
1589        if (na) {
1590                list_del(&na->list);
1591                while (na->addr_count--)
1592                        kfree(na->addr[na->addr_count]);
1593                kfree(na);
1594        }
1595        spin_unlock(&dlm_node_addrs_spin);
1596
1597        return 0;
1598}
1599
1600/* Receive workqueue function */
1601static void process_recv_sockets(struct work_struct *work)
1602{
1603        struct connection *con = container_of(work, struct connection, rwork);
1604        int err;
1605
1606        clear_bit(CF_READ_PENDING, &con->flags);
1607        do {
1608                err = con->rx_action(con);
1609        } while (!err);
1610}
1611
1612/* Send workqueue function */
1613static void process_send_sockets(struct work_struct *work)
1614{
1615        struct connection *con = container_of(work, struct connection, swork);
1616
1617        clear_bit(CF_WRITE_PENDING, &con->flags);
1618        if (con->sock == NULL) /* not mutex protected so check it inside too */
1619                con->connect_action(con);
1620        if (!list_empty(&con->writequeue))
1621                send_to_sock(con);
1622}
1623
1624
1625/* Discard all entries on the write queues */
1626static void clean_writequeues(void)
1627{
1628        foreach_conn(clean_one_writequeue);
1629}
1630
1631static void work_stop(void)
1632{
1633        destroy_workqueue(recv_workqueue);
1634        destroy_workqueue(send_workqueue);
1635}
1636
1637static int work_start(void)
1638{
1639        recv_workqueue = alloc_workqueue("dlm_recv",
1640                                         WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1641        if (!recv_workqueue) {
1642                log_print("can't start dlm_recv");
1643                return -ENOMEM;
1644        }
1645
1646        send_workqueue = alloc_workqueue("dlm_send",
1647                                         WQ_UNBOUND | WQ_MEM_RECLAIM, 1);
1648        if (!send_workqueue) {
1649                log_print("can't start dlm_send");
1650                destroy_workqueue(recv_workqueue);
1651                return -ENOMEM;
1652        }
1653
1654        return 0;
1655}
1656
1657static void _stop_conn(struct connection *con, bool and_other)
1658{
1659        mutex_lock(&con->sock_mutex);
1660        set_bit(CF_CLOSE, &con->flags);
1661        set_bit(CF_READ_PENDING, &con->flags);
1662        set_bit(CF_WRITE_PENDING, &con->flags);
1663        if (con->sock && con->sock->sk) {
1664                write_lock_bh(&con->sock->sk->sk_callback_lock);
1665                con->sock->sk->sk_user_data = NULL;
1666                write_unlock_bh(&con->sock->sk->sk_callback_lock);
1667        }
1668        if (con->othercon && and_other)
1669                _stop_conn(con->othercon, false);
1670        mutex_unlock(&con->sock_mutex);
1671}
1672
1673static void stop_conn(struct connection *con)
1674{
1675        _stop_conn(con, true);
1676}
1677
1678static void free_conn(struct connection *con)
1679{
1680        close_connection(con, true, true, true);
1681        if (con->othercon)
1682                kmem_cache_free(con_cache, con->othercon);
1683        hlist_del(&con->list);
1684        kmem_cache_free(con_cache, con);
1685}
1686
1687static void work_flush(void)
1688{
1689        int ok;
1690        int i;
1691        struct hlist_node *n;
1692        struct connection *con;
1693
1694        flush_workqueue(recv_workqueue);
1695        flush_workqueue(send_workqueue);
1696        do {
1697                ok = 1;
1698                foreach_conn(stop_conn);
1699                flush_workqueue(recv_workqueue);
1700                flush_workqueue(send_workqueue);
1701                for (i = 0; i < CONN_HASH_SIZE && ok; i++) {
1702                        hlist_for_each_entry_safe(con, n,
1703                                                  &connection_hash[i], list) {
1704                                ok &= test_bit(CF_READ_PENDING, &con->flags);
1705                                ok &= test_bit(CF_WRITE_PENDING, &con->flags);
1706                                if (con->othercon) {
1707                                        ok &= test_bit(CF_READ_PENDING,
1708                                                       &con->othercon->flags);
1709                                        ok &= test_bit(CF_WRITE_PENDING,
1710                                                       &con->othercon->flags);
1711                                }
1712                        }
1713                }
1714        } while (!ok);
1715}
1716
1717void dlm_lowcomms_stop(void)
1718{
1719        /* Set all the flags to prevent any
1720           socket activity.
1721        */
1722        mutex_lock(&connections_lock);
1723        dlm_allow_conn = 0;
1724        mutex_unlock(&connections_lock);
1725        work_flush();
1726        clean_writequeues();
1727        foreach_conn(free_conn);
1728        work_stop();
1729
1730        kmem_cache_destroy(con_cache);
1731}
1732
1733int dlm_lowcomms_start(void)
1734{
1735        int error = -EINVAL;
1736        struct connection *con;
1737        int i;
1738
1739        for (i = 0; i < CONN_HASH_SIZE; i++)
1740                INIT_HLIST_HEAD(&connection_hash[i]);
1741
1742        init_local();
1743        if (!dlm_local_count) {
1744                error = -ENOTCONN;
1745                log_print("no local IP address has been set");
1746                goto fail;
1747        }
1748
1749        error = -ENOMEM;
1750        con_cache = kmem_cache_create("dlm_conn", sizeof(struct connection),
1751                                      __alignof__(struct connection), 0,
1752                                      NULL);
1753        if (!con_cache)
1754                goto fail;
1755
1756        error = work_start();
1757        if (error)
1758                goto fail_destroy;
1759
1760        dlm_allow_conn = 1;
1761
1762        /* Start listening */
1763        if (dlm_config.ci_protocol == 0)
1764                error = tcp_listen_for_all();
1765        else
1766                error = sctp_listen_for_all();
1767        if (error)
1768                goto fail_unlisten;
1769
1770        return 0;
1771
1772fail_unlisten:
1773        dlm_allow_conn = 0;
1774        con = nodeid2con(0,0);
1775        if (con) {
1776                close_connection(con, false, true, true);
1777                kmem_cache_free(con_cache, con);
1778        }
1779fail_destroy:
1780        kmem_cache_destroy(con_cache);
1781fail:
1782        return error;
1783}
1784
1785void dlm_lowcomms_exit(void)
1786{
1787        struct dlm_node_addr *na, *safe;
1788
1789        spin_lock(&dlm_node_addrs_spin);
1790        list_for_each_entry_safe(na, safe, &dlm_node_addrs, list) {
1791                list_del(&na->list);
1792                while (na->addr_count--)
1793                        kfree(na->addr[na->addr_count]);
1794                kfree(na);
1795        }
1796        spin_unlock(&dlm_node_addrs_spin);
1797}
1798