lnet/ulnds/socklnd/usocklnd_cb.c

   1 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
   2  * vim:expandtab:shiftwidth=8:tabstop=8:
   3  *
   4  * GPL HEADER START
   5  *
   6  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
   7  *
   8  * This program is free software; you can redistribute it and/or modify
   9  * it under the terms of the GNU General Public License version 2 only,
  10  * as published by the Free Software Foundation.
  11  *
  12  * This program is distributed in the hope that it will be useful, but
  13  * WITHOUT ANY WARRANTY; without even the implied warranty of
  14  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  15  * General Public License version 2 for more details (a copy is included
  16  * in the LICENSE file that accompanied this code).
  17  *
  18  * You should have received a copy of the GNU General Public License
  19  * version 2 along with this program; If not, see
  20  * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
  21  *
  22  * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
  23  * CA 95054 USA or visit www.sun.com if you need additional information or
  24  * have any questions.
  25  *
  26  * GPL HEADER END
  27  */
  28 /*
  29  * Copyright (c) 2007, 2010, Oracle and/or its affiliates. All rights reserved.
  30  * Use is subject to license terms.
  31  */
  32 /*
  33  * This file is part of Lustre, http://www.lustre.org/
  34  * Lustre is a trademark of Sun Microsystems, Inc.
  35  *
  36  * lnet/ulnds/socklnd/usocklnd_cb.c
  37  *
  38  * Author: Maxim Patlasov <maxim@clusterfs.com>
  39  */
  40
  41 #include "usocklnd.h"
  42
  43 static int
  44 usocklnd_send_tx_immediately(usock_conn_t *conn, usock_tx_t *tx)
  45 {
  46         int           rc;
  47         int           rc2;
  48         int           partial_send = 0;
  49         usock_peer_t *peer         = conn->uc_peer;
  50
  51         LASSERT (peer != NULL);
  52
  53         /* usocklnd_enqueue_tx() turned it on for us */
  54         LASSERT(conn->uc_sending);
  55
  56         //counter_imm_start++;
  57         rc = usocklnd_send_tx(conn, tx);
  58         if (rc == 0) { /* partial send or connection closed */
  59                 pthread_mutex_lock(&conn->uc_lock);
  60                 cfs_list_add(&tx->tx_list, &conn->uc_tx_list);
  61                 conn->uc_sending = 0;
  62                 pthread_mutex_unlock(&conn->uc_lock);
  63                 partial_send = 1;
  64         } else {
  65                 usocklnd_destroy_tx(peer->up_ni, tx);
  66                 /* NB: lnetmsg was finalized, so we *must* return 0 */
  67
  68                 if (rc < 0) { /* real error */
  69                         usocklnd_conn_kill(conn);
  70                         return 0;
  71                 }
  72
  73                 /* rc == 1: tx was sent completely */
  74                 rc = 0; /* let's say to caller 'Ok' */
  75                 //counter_imm_complete++;
  76         }
  77
  78         pthread_mutex_lock(&conn->uc_lock);
  79         conn->uc_sending = 0;
  80
  81         /* schedule write handler */
  82         if (partial_send ||
  83             (conn->uc_state == UC_READY &&
  84              (!cfs_list_empty(&conn->uc_tx_list) ||
  85               !cfs_list_empty(&conn->uc_zcack_list)))) {
  86                 conn->uc_tx_deadline =
  87                         cfs_time_shift(usock_tuns.ut_timeout);
  88                 conn->uc_tx_flag = 1;
  89                 rc2 = usocklnd_add_pollrequest(conn, POLL_TX_SET_REQUEST, POLLOUT);
  90                 if (rc2 != 0)
  91                         usocklnd_conn_kill_locked(conn);
  92                 else
  93                         usocklnd_wakeup_pollthread(conn->uc_pt_idx);
  94         }
  95
  96         pthread_mutex_unlock(&conn->uc_lock);
  97
  98         return rc;
  99 }
 100
 101 int
 102 usocklnd_send(lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
 103 {
 104         usock_tx_t       *tx;
 105         lnet_process_id_t target = lntmsg->msg_target;
 106         usock_peer_t     *peer;
 107         int               type;
 108         int               rc;
 109         usock_conn_t     *conn;
 110         int               send_immediately;
 111
 112         tx = usocklnd_create_tx(lntmsg);
 113         if (tx == NULL)
 114                 return -ENOMEM;
 115
 116         rc = usocklnd_find_or_create_peer(ni, target, &peer);
 117         if (rc) {
 118                 LIBCFS_FREE (tx, tx->tx_size);
 119                 return rc;
 120         }
 121         /* peer cannot disappear now because its refcount was incremented */
 122
 123         type = usocklnd_get_conn_type(lntmsg);
 124         rc = usocklnd_find_or_create_conn(peer, type, &conn, tx, NULL,
 125                                           &send_immediately);
 126         if (rc != 0) {
 127                 usocklnd_peer_decref(peer);
 128                 usocklnd_check_peer_stale(ni, target);
 129                 LIBCFS_FREE (tx, tx->tx_size);
 130                 return rc;
 131         }
 132         /* conn cannot disappear now because its refcount was incremented */
 133
 134         if (send_immediately)
 135                 rc = usocklnd_send_tx_immediately(conn, tx);
 136
 137         usocklnd_conn_decref(conn);
 138         usocklnd_peer_decref(peer);
 139         return rc;
 140 }
 141
 142 int
 143 usocklnd_recv(lnet_ni_t *ni, void *private, lnet_msg_t *msg, int delayed,
 144               unsigned int niov, struct iovec *iov, lnet_kiov_t *kiov,
 145               unsigned int offset, unsigned int mlen, unsigned int rlen)
 146 {
 147         int           rc   = 0;
 148         usock_conn_t *conn = (usock_conn_t *)private;
 149
 150         /* I don't think that we'll win much concurrency moving lock()
 151          * call below lnet_extract_iov() */
 152         pthread_mutex_lock(&conn->uc_lock);
 153
 154         conn->uc_rx_lnetmsg = msg;
 155         conn->uc_rx_nob_wanted = mlen;
 156         conn->uc_rx_nob_left = rlen;
 157         conn->uc_rx_iov = conn->uc_rx_iova;
 158         conn->uc_rx_niov =
 159                 lnet_extract_iov(LNET_MAX_IOV, conn->uc_rx_iov,
 160                                  niov, iov, offset, mlen);
 161
 162         /* the gap between lnet_parse() and usocklnd_recv() happened? */
 163         if (conn->uc_rx_state == UC_RX_PARSE_WAIT) {
 164                 conn->uc_rx_flag = 1; /* waiting for incoming lnet payload */
 165                 conn->uc_rx_deadline =
 166                         cfs_time_shift(usock_tuns.ut_timeout);
 167                 rc = usocklnd_add_pollrequest(conn, POLL_RX_SET_REQUEST, POLLIN);
 168                 if (rc != 0) {
 169                         usocklnd_conn_kill_locked(conn);
 170                         goto recv_out;
 171                 }
 172                 usocklnd_wakeup_pollthread(conn->uc_pt_idx);
 173         }
 174
 175         conn->uc_rx_state = UC_RX_LNET_PAYLOAD;
 176   recv_out:
 177         pthread_mutex_unlock(&conn->uc_lock);
 178         usocklnd_conn_decref(conn);
 179         return rc;
 180 }
 181
 182 int
 183 usocklnd_accept(lnet_ni_t *ni, cfs_socket_t *sock)
 184 {
 185         int           rc;
 186         usock_conn_t *conn;
 187
 188         rc = usocklnd_create_passive_conn(ni, sock, &conn);
 189         if (rc)
 190                 return rc;
 191         LASSERT(conn != NULL);
 192
 193         /* disable shutdown event temporarily */
 194         lnet_ni_addref(ni);
 195
 196         rc = usocklnd_add_pollrequest(conn, POLL_ADD_REQUEST, POLLIN);
 197         if (rc == 0)
 198                 usocklnd_wakeup_pollthread(conn->uc_pt_idx);
 199
 200         /* NB: conn reference counter was incremented while adding
 201          * poll request if rc == 0 */
 202
 203         usocklnd_conn_decref(conn); /* should destroy conn if rc != 0 */
 204         return rc;
 205 }