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