lnet/ulnds/socklnd/usocklnd_cb.c

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