Whamcloud - gitweb
b=14132
[fs/lustre-release.git] / lnet / ulnds / socklnd / handlers.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  2008 Sun Microsystems, Inc. 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/handlers.c
37  *
38  * Author: Maxim Patlasov <maxim@clusterfs.com>
39  */
40
41 #include "usocklnd.h"
42 #include <unistd.h>
43 #include <sys/syscall.h>
44
45 int
46 usocklnd_notifier_handler(int fd)
47 {
48         int notification;
49         return syscall(SYS_read, fd, &notification, sizeof(notification));
50 }
51
52 void
53 usocklnd_exception_handler(usock_conn_t *conn)
54 {
55         pthread_mutex_lock(&conn->uc_lock);
56
57         if (conn->uc_state == UC_CONNECTING ||
58             conn->uc_state == UC_SENDING_HELLO)
59                 usocklnd_conn_kill_locked(conn);
60
61         pthread_mutex_unlock(&conn->uc_lock);
62 }
63
64 int
65 usocklnd_read_handler(usock_conn_t *conn)
66 {
67         int rc;
68         int continue_reading;
69         int state;
70
71   read_again:
72         rc = 0;
73         pthread_mutex_lock(&conn->uc_lock);
74         state = conn->uc_state;
75
76         /* process special case: LNET calls lnd_recv() asyncronously */
77         if (state == UC_READY && conn->uc_rx_state == UC_RX_PARSE) {
78                 /* still don't have usocklnd_recv() called */
79                 rc = usocklnd_add_pollrequest(conn, POLL_RX_SET_REQUEST, 0);
80                 if (rc == 0)
81                         conn->uc_rx_state = UC_RX_PARSE_WAIT;
82                 else
83                         usocklnd_conn_kill_locked(conn);
84
85                 pthread_mutex_unlock(&conn->uc_lock);
86                 return rc;
87         }
88
89         pthread_mutex_unlock(&conn->uc_lock);
90         /* From here and below the conn cannot be changed
91          * asyncronously, except:
92          * 1) usocklnd_send() can work with uc_tx_list and uc_zcack_list,
93          * 2) usocklnd_shutdown() can change uc_state to UC_DEAD */
94
95         switch (state) {
96
97         case UC_RECEIVING_HELLO:
98         case UC_READY:
99                 if (conn->uc_rx_nob_wanted != 0) {
100                         /* read from conn fd as much wanted data as possible */
101                         rc = usocklnd_read_data(conn);
102                         if (rc == 0) /* partial read */
103                                 break;
104                         if (rc < 0) {/* error happened or EOF */
105                                 usocklnd_conn_kill(conn);
106                                 break;
107                         }
108                 }
109
110                 /* process incoming data */
111                 if (state == UC_READY )
112                         rc = usocklnd_read_msg(conn, &continue_reading);
113                 else /* state == UC_RECEIVING_HELLO */
114                         rc = usocklnd_read_hello(conn, &continue_reading);
115
116                 if (rc < 0) {
117                         usocklnd_conn_kill(conn);
118                         break;
119                 }
120
121                 if (continue_reading)
122                         goto read_again;
123
124                 break;
125
126         case UC_DEAD:
127                 break;
128
129         default:
130                 LBUG();
131         }
132
133         return rc;
134 }
135
136 /* Switch on rx_state.
137  * Return 0 on success, 1 if whole packet is read, else return <0
138  * Always set cont_flag: 1 if we're ready to continue reading, else 0
139  * NB: If whole packet is read, cont_flag will be set to zero to take
140  * care of fairess
141  */
142 int
143 usocklnd_read_msg(usock_conn_t *conn, int *cont_flag)
144 {
145         int   rc = 0;
146         __u64 cookie;
147
148        *cont_flag = 0;
149
150         /* smth. new emerged in RX part - let's process it */
151         switch (conn->uc_rx_state) {
152         case UC_RX_KSM_HEADER:
153                 if (conn->uc_flip) {
154                         __swab32s(&conn->uc_rx_msg.ksm_type);
155                         __swab32s(&conn->uc_rx_msg.ksm_csum);
156                         __swab64s(&conn->uc_rx_msg.ksm_zc_cookies[0]);
157                         __swab64s(&conn->uc_rx_msg.ksm_zc_cookies[1]);
158                 }
159
160                 /* we never send packets for wich zc-acking is required */
161                 if (conn->uc_rx_msg.ksm_type != KSOCK_MSG_LNET ||
162                     conn->uc_rx_msg.ksm_zc_cookies[1] != 0) {
163                         conn->uc_errored = 1;
164                         return -EPROTO;
165                 }
166
167                 /* zc_req will be processed later, when
168                    lnet payload will be received */
169
170                 usocklnd_rx_lnethdr_state_transition(conn);
171                 *cont_flag = 1;
172                 break;
173
174         case UC_RX_LNET_HEADER:
175                 if (the_lnet.ln_pid & LNET_PID_USERFLAG) {
176                         /* replace dest_nid,pid (ksocknal sets its own) */
177                         conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_nid =
178                                 cpu_to_le64(conn->uc_peer->up_ni->ni_nid);
179                         conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr.dest_pid =
180                                 cpu_to_le32(the_lnet.ln_pid);
181
182                 } else if (conn->uc_peer->up_peerid.pid & LNET_PID_USERFLAG) {
183                         /* Userspace peer */
184                         lnet_process_id_t *id = &conn->uc_peer->up_peerid;
185                         lnet_hdr_t        *lhdr = &conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr;
186
187                         /* Substitute process ID assigned at connection time */
188                         lhdr->src_pid = cpu_to_le32(id->pid);
189                         lhdr->src_nid = cpu_to_le64(id->nid);
190                 }
191
192                 conn->uc_rx_state = UC_RX_PARSE;
193                 usocklnd_conn_addref(conn); /* ++ref while parsing */
194
195                 rc = lnet_parse(conn->uc_peer->up_ni,
196                                 &conn->uc_rx_msg.ksm_u.lnetmsg.ksnm_hdr,
197                                 conn->uc_peerid.nid, conn, 0);
198
199                 if (rc < 0) {
200                         /* I just received garbage: give up on this conn */
201                         conn->uc_errored = 1;
202                         usocklnd_conn_decref(conn);
203                         return -EPROTO;
204                 }
205
206                 /* Race with usocklnd_recv() is possible */
207                 pthread_mutex_lock(&conn->uc_lock);
208                 LASSERT (conn->uc_rx_state == UC_RX_PARSE ||
209                          conn->uc_rx_state == UC_RX_LNET_PAYLOAD);
210
211                 /* check whether usocklnd_recv() got called */
212                 if (conn->uc_rx_state == UC_RX_LNET_PAYLOAD)
213                         *cont_flag = 1;
214                 pthread_mutex_unlock(&conn->uc_lock);
215                 break;
216
217         case UC_RX_PARSE:
218                 LBUG(); /* it's error to be here, because this special
219                          * case is handled by caller */
220                 break;
221
222         case UC_RX_PARSE_WAIT:
223                 LBUG(); /* it's error to be here, because the conn
224                          * shouldn't wait for POLLIN event in this
225                          * state */
226                 break;
227
228         case UC_RX_LNET_PAYLOAD:
229                 /* payload all received */
230
231                 lnet_finalize(conn->uc_peer->up_ni, conn->uc_rx_lnetmsg, 0);
232
233                 cookie = conn->uc_rx_msg.ksm_zc_cookies[0];
234                 if (cookie != 0)
235                         rc = usocklnd_handle_zc_req(conn->uc_peer, cookie);
236
237                 if (rc != 0) {
238                         /* change state not to finalize twice */
239                         conn->uc_rx_state = UC_RX_KSM_HEADER;
240                         return -EPROTO;
241                 }
242
243                 /* Fall through */
244
245         case UC_RX_SKIPPING:
246                 if (conn->uc_rx_nob_left != 0) {
247                         usocklnd_rx_skipping_state_transition(conn);
248                         *cont_flag = 1;
249                 } else {
250                         usocklnd_rx_ksmhdr_state_transition(conn);
251                         rc = 1; /* whole packet is read */
252                 }
253
254                 break;
255
256         default:
257                 LBUG(); /* unknown state */
258         }
259
260         return rc;
261 }
262
263 /* Handle incoming ZC request from sender.
264  * NB: it's called only from read_handler, so we're sure that
265  * the conn cannot become zombie in the middle of processing */
266 int
267 usocklnd_handle_zc_req(usock_peer_t *peer, __u64 cookie)
268 {
269         usock_conn_t   *conn;
270         usock_zc_ack_t *zc_ack;
271         int             type;
272         int             rc;
273         int             dummy;
274
275         LIBCFS_ALLOC (zc_ack, sizeof(*zc_ack));
276         if (zc_ack == NULL)
277                 return -ENOMEM;
278         zc_ack->zc_cookie = cookie;
279
280         /* Let's assume that CONTROL is the best type for zcack,
281          * but userspace clients don't use typed connections */
282         if (the_lnet.ln_pid & LNET_PID_USERFLAG)
283                 type = SOCKLND_CONN_ANY;
284         else
285                 type = SOCKLND_CONN_CONTROL;
286
287         rc = usocklnd_find_or_create_conn(peer, type, &conn, NULL, zc_ack,
288                                           &dummy);
289         if (rc != 0) {
290                 LIBCFS_FREE (zc_ack, sizeof(*zc_ack));
291                 return rc;
292         }
293         usocklnd_conn_decref(conn);
294
295         return 0;
296 }
297
298 /* Switch on rx_state.
299  * Return 0 on success, else return <0
300  * Always set cont_flag: 1 if we're ready to continue reading, else 0
301  */
302 int
303 usocklnd_read_hello(usock_conn_t *conn, int *cont_flag)
304 {
305         int                rc = 0;
306         ksock_hello_msg_t *hello = conn->uc_rx_hello;
307
308         *cont_flag = 0;
309
310         /* smth. new emerged in hello - let's process it */
311         switch (conn->uc_rx_state) {
312         case UC_RX_HELLO_MAGIC:
313                 if (hello->kshm_magic == LNET_PROTO_MAGIC)
314                         conn->uc_flip = 0;
315                 else if (hello->kshm_magic == __swab32(LNET_PROTO_MAGIC))
316                         conn->uc_flip = 1;
317                 else
318                         return -EPROTO;
319
320                 usocklnd_rx_helloversion_state_transition(conn);
321                 *cont_flag = 1;
322                 break;
323
324         case UC_RX_HELLO_VERSION:
325                 if ((!conn->uc_flip &&
326                      (hello->kshm_version != KSOCK_PROTO_V2)) ||
327                     (conn->uc_flip &&
328                      (hello->kshm_version != __swab32(KSOCK_PROTO_V2))))
329                         return -EPROTO;
330
331                 usocklnd_rx_hellobody_state_transition(conn);
332                 *cont_flag = 1;
333                 break;
334
335         case UC_RX_HELLO_BODY:
336                 if (conn->uc_flip) {
337                         ksock_hello_msg_t *hello = conn->uc_rx_hello;
338                         __swab32s(&hello->kshm_src_pid);
339                         __swab64s(&hello->kshm_src_nid);
340                         __swab32s(&hello->kshm_dst_pid);
341                         __swab64s(&hello->kshm_dst_nid);
342                         __swab64s(&hello->kshm_src_incarnation);
343                         __swab64s(&hello->kshm_dst_incarnation);
344                         __swab32s(&hello->kshm_ctype);
345                         __swab32s(&hello->kshm_nips);
346                 }
347
348                 if (conn->uc_rx_hello->kshm_nips > LNET_MAX_INTERFACES) {
349                         CERROR("Bad nips %d from ip %u.%u.%u.%u port %d\n",
350                                conn->uc_rx_hello->kshm_nips,
351                                HIPQUAD(conn->uc_peer_ip), conn->uc_peer_port);
352                         return -EPROTO;
353                 }
354
355                 if (conn->uc_rx_hello->kshm_nips) {
356                         usocklnd_rx_helloIPs_state_transition(conn);
357                         *cont_flag = 1;
358                         break;
359                 }
360                 /* fall through */
361
362         case UC_RX_HELLO_IPS:
363                 if (conn->uc_activeflag == 1) /* active conn */
364                         rc = usocklnd_activeconn_hellorecv(conn);
365                 else                          /* passive conn */
366                         rc = usocklnd_passiveconn_hellorecv(conn);
367
368                 break;
369
370         default:
371                 LBUG(); /* unknown state */
372         }
373
374         return rc;
375 }
376
377 /* All actions that we need after receiving hello on active conn:
378  * 1) Schedule removing if we're zombie
379  * 2) Restart active conn if we lost the race
380  * 3) Else: update RX part to receive KSM header
381  */
382 int
383 usocklnd_activeconn_hellorecv(usock_conn_t *conn)
384 {
385         int                rc    = 0;
386         ksock_hello_msg_t *hello = conn->uc_rx_hello;
387         usock_peer_t      *peer  = conn->uc_peer;
388
389         /* Active conn with peer==NULL is zombie.
390          * Don't try to link it to peer because the conn
391          * has already had a chance to proceed at the beginning */
392         if (peer == NULL) {
393                 LASSERT(list_empty(&conn->uc_tx_list) &&
394                         list_empty(&conn->uc_zcack_list));
395
396                 usocklnd_conn_kill(conn);
397                 return 0;
398         }
399
400         peer->up_last_alive = cfs_time_current();
401
402         /* peer says that we lost the race */
403         if (hello->kshm_ctype == SOCKLND_CONN_NONE) {
404                 /* Start new active conn, relink txs and zc_acks from
405                  * the conn to new conn, schedule removing the conn.
406                  * Actually, we're expecting that a passive conn will
407                  * make us zombie soon and take care of our txs and
408                  * zc_acks */
409
410                 struct list_head tx_list, zcack_list;
411                 usock_conn_t *conn2;
412                 int idx = usocklnd_type2idx(conn->uc_type);
413
414                 CFS_INIT_LIST_HEAD (&tx_list);
415                 CFS_INIT_LIST_HEAD (&zcack_list);
416
417                 /* Block usocklnd_send() to check peer->up_conns[idx]
418                  * and to enqueue more txs */
419                 pthread_mutex_lock(&peer->up_lock);
420                 pthread_mutex_lock(&conn->uc_lock);
421
422                 /* usocklnd_shutdown() could kill us */
423                 if (conn->uc_state == UC_DEAD) {
424                         pthread_mutex_unlock(&conn->uc_lock);
425                         pthread_mutex_unlock(&peer->up_lock);
426                         return 0;
427                 }
428
429                 LASSERT (peer == conn->uc_peer);
430                 LASSERT (peer->up_conns[idx] == conn);
431
432                 rc = usocklnd_create_active_conn(peer, conn->uc_type, &conn2);
433                 if (rc) {
434                         conn->uc_errored = 1;
435                         pthread_mutex_unlock(&conn->uc_lock);
436                         pthread_mutex_unlock(&peer->up_lock);
437                         return rc;
438                 }
439
440                 usocklnd_link_conn_to_peer(conn2, peer, idx);
441                 conn2->uc_peer = peer;
442
443                 /* unlink txs and zcack from the conn */
444                 list_add(&tx_list, &conn->uc_tx_list);
445                 list_del_init(&conn->uc_tx_list);
446                 list_add(&zcack_list, &conn->uc_zcack_list);
447                 list_del_init(&conn->uc_zcack_list);
448
449                 /* link they to the conn2 */
450                 list_add(&conn2->uc_tx_list, &tx_list);
451                 list_del_init(&tx_list);
452                 list_add(&conn2->uc_zcack_list, &zcack_list);
453                 list_del_init(&zcack_list);
454
455                 /* make conn zombie */
456                 conn->uc_peer = NULL;
457                 usocklnd_peer_decref(peer);
458
459                 /* schedule conn2 for processing */
460                 rc = usocklnd_add_pollrequest(conn2, POLL_ADD_REQUEST, POLLOUT);
461                 if (rc) {
462                         peer->up_conns[idx] = NULL;
463                         usocklnd_conn_decref(conn2); /* should destroy conn */
464                 } else {
465                         usocklnd_conn_kill_locked(conn);
466                 }
467
468                 pthread_mutex_unlock(&conn->uc_lock);
469                 pthread_mutex_unlock(&peer->up_lock);
470                 usocklnd_conn_decref(conn);
471
472         } else { /* hello->kshm_ctype != SOCKLND_CONN_NONE */
473                 if (conn->uc_type != usocklnd_invert_type(hello->kshm_ctype))
474                         return -EPROTO;
475
476                 pthread_mutex_lock(&peer->up_lock);
477                 usocklnd_cleanup_stale_conns(peer, hello->kshm_src_incarnation,
478                                              conn);
479                 pthread_mutex_unlock(&peer->up_lock);
480
481                 /* safely transit to UC_READY state */
482                 /* rc == 0 */
483                 pthread_mutex_lock(&conn->uc_lock);
484                 if (conn->uc_state != UC_DEAD) {
485                         usocklnd_rx_ksmhdr_state_transition(conn);
486
487                         /* POLLIN is already set because we just
488                          * received hello, but maybe we've smth. to
489                          * send? */
490                         LASSERT (conn->uc_sending == 0);
491                         if ( !list_empty(&conn->uc_tx_list) ||
492                              !list_empty(&conn->uc_zcack_list) ) {
493
494                                 conn->uc_tx_deadline =
495                                         cfs_time_shift(usock_tuns.ut_timeout);
496                                 conn->uc_tx_flag = 1;
497                                 rc = usocklnd_add_pollrequest(conn,
498                                                               POLL_SET_REQUEST,
499                                                               POLLIN | POLLOUT);
500                         }
501
502                         if (rc == 0)
503                                 conn->uc_state = UC_READY;
504                 }
505                 pthread_mutex_unlock(&conn->uc_lock);
506         }
507
508         return rc;
509 }
510
511 /* All actions that we need after receiving hello on passive conn:
512  * 1) Stash peer's nid, pid, incarnation and conn type
513  * 2) Cope with easy case: conn[idx] is empty - just save conn there
514  * 3) Resolve race:
515  *    a) if our nid is higher - reply with CONN_NONE and make us zombie
516  *    b) if peer's nid is higher - postpone race resolution till
517  *       READY state
518  * 4) Anyhow, send reply hello
519 */
520 int
521 usocklnd_passiveconn_hellorecv(usock_conn_t *conn)
522 {
523         ksock_hello_msg_t *hello = conn->uc_rx_hello;
524         int                type;
525         int                idx;
526         int                rc;
527         usock_peer_t      *peer;
528         lnet_ni_t         *ni        = conn->uc_ni;
529         __u32              peer_ip   = conn->uc_peer_ip;
530         __u16              peer_port = conn->uc_peer_port;
531
532         /* don't know parent peer yet and not zombie */
533         LASSERT (conn->uc_peer == NULL &&
534                  ni != NULL);
535
536         /* don't know peer's nid and incarnation yet */
537         if (peer_port > LNET_ACCEPTOR_MAX_RESERVED_PORT) {
538                 /* do not trust liblustre clients */
539                 conn->uc_peerid.pid = peer_port | LNET_PID_USERFLAG;
540                 conn->uc_peerid.nid = LNET_MKNID(LNET_NIDNET(ni->ni_nid),
541                                                  peer_ip);
542                 if (hello->kshm_ctype != SOCKLND_CONN_ANY) {
543                         lnet_ni_decref(ni);
544                         conn->uc_ni = NULL;
545                         CERROR("Refusing to accept connection of type=%d from "
546                                "userspace process %u.%u.%u.%u:%d\n", hello->kshm_ctype,
547                                HIPQUAD(peer_ip), peer_port);
548                         return -EINVAL;
549                 }
550         } else {
551                 conn->uc_peerid.pid = hello->kshm_src_pid;
552                 conn->uc_peerid.nid = hello->kshm_src_nid;
553         }
554         conn->uc_type = type = usocklnd_invert_type(hello->kshm_ctype);
555
556         rc = usocklnd_find_or_create_peer(ni, conn->uc_peerid, &peer);
557         if (rc) {
558                 lnet_ni_decref(ni);
559                 conn->uc_ni = NULL;
560                 return rc;
561         }
562
563         peer->up_last_alive = cfs_time_current();
564
565         idx = usocklnd_type2idx(conn->uc_type);
566
567         /* safely check whether we're first */
568         pthread_mutex_lock(&peer->up_lock);
569
570         usocklnd_cleanup_stale_conns(peer, hello->kshm_src_incarnation, NULL);
571
572         if (peer->up_conns[idx] == NULL) {
573                 peer->up_last_alive = cfs_time_current();
574                 conn->uc_peer = peer;
575                 conn->uc_ni = NULL;
576                 usocklnd_link_conn_to_peer(conn, peer, idx);
577                 usocklnd_conn_addref(conn);
578         } else {
579                 usocklnd_peer_decref(peer);
580
581                 /* Resolve race in favour of higher NID */
582                 if (conn->uc_peerid.nid < conn->uc_ni->ni_nid) {
583                         /* make us zombie */
584                         conn->uc_ni = NULL;
585                         type = SOCKLND_CONN_NONE;
586                 }
587
588                 /* if conn->uc_peerid.nid > conn->uc_ni->ni_nid,
589                  * postpone race resolution till READY state
590                  * (hopefully that conn[idx] will die because of
591                  * incoming hello of CONN_NONE type) */
592         }
593         pthread_mutex_unlock(&peer->up_lock);
594
595         /* allocate and initialize fake tx with hello */
596         conn->uc_tx_hello = usocklnd_create_hello_tx(ni, type,
597                                                      conn->uc_peerid.nid);
598         if (conn->uc_ni == NULL)
599                 lnet_ni_decref(ni);
600
601         if (conn->uc_tx_hello == NULL)
602                 return -ENOMEM;
603
604         /* rc == 0 */
605         pthread_mutex_lock(&conn->uc_lock);
606         if (conn->uc_state == UC_DEAD)
607                 goto passive_hellorecv_done;
608
609         conn->uc_state = UC_SENDING_HELLO;
610         conn->uc_tx_deadline = cfs_time_shift(usock_tuns.ut_timeout);
611         conn->uc_tx_flag = 1;
612         rc = usocklnd_add_pollrequest(conn, POLL_SET_REQUEST, POLLOUT);
613
614   passive_hellorecv_done:
615         pthread_mutex_unlock(&conn->uc_lock);
616         return rc;
617 }
618
619 int
620 usocklnd_write_handler(usock_conn_t *conn)
621 {
622         usock_tx_t   *tx;
623         int           ret;
624         int           rc = 0;
625         int           state;
626         usock_peer_t *peer;
627         lnet_ni_t    *ni;
628
629         pthread_mutex_lock(&conn->uc_lock); /* like membar */
630         state = conn->uc_state;
631         pthread_mutex_unlock(&conn->uc_lock);
632
633         switch (state) {
634         case UC_CONNECTING:
635                 /* hello_tx has already been initialized
636                  * in usocklnd_create_active_conn() */
637                 usocklnd_conn_new_state(conn, UC_SENDING_HELLO);
638                 /* fall through */
639
640         case UC_SENDING_HELLO:
641                 rc = usocklnd_send_tx(conn, conn->uc_tx_hello);
642                 if (rc <= 0) /* error or partial send or connection closed */
643                         break;
644
645                 /* tx with hello was sent successfully */
646                 usocklnd_destroy_tx(NULL, conn->uc_tx_hello);
647                 conn->uc_tx_hello = NULL;
648
649                 if (conn->uc_activeflag == 1) /* active conn */
650                         rc = usocklnd_activeconn_hellosent(conn);
651                 else                          /* passive conn */
652                         rc = usocklnd_passiveconn_hellosent(conn);
653
654                 break;
655
656         case UC_READY:
657                 pthread_mutex_lock(&conn->uc_lock);
658
659                 peer = conn->uc_peer;
660                 LASSERT (peer != NULL);
661                 ni = peer->up_ni;
662
663                 if (list_empty(&conn->uc_tx_list) &&
664                     list_empty(&conn->uc_zcack_list)) {
665                         LASSERT(usock_tuns.ut_fair_limit > 1);
666                         pthread_mutex_unlock(&conn->uc_lock);
667                         return 0;
668                 }
669
670                 tx = usocklnd_try_piggyback(&conn->uc_tx_list,
671                                             &conn->uc_zcack_list);
672                 if (tx != NULL)
673                         conn->uc_sending = 1;
674                 else
675                         rc = -ENOMEM;
676
677                 pthread_mutex_unlock(&conn->uc_lock);
678
679                 if (rc)
680                         break;
681
682                 rc = usocklnd_send_tx(conn, tx);
683                 if (rc == 0) { /* partial send or connection closed */
684                         pthread_mutex_lock(&conn->uc_lock);
685                         list_add(&tx->tx_list, &conn->uc_tx_list);
686                         conn->uc_sending = 0;
687                         pthread_mutex_unlock(&conn->uc_lock);
688                         break;
689                 }
690                 if (rc < 0) { /* real error */
691                         usocklnd_destroy_tx(ni, tx);
692                         break;
693                 }
694
695                 /* rc == 1: tx was sent completely */
696                 usocklnd_destroy_tx(ni, tx);
697
698                 pthread_mutex_lock(&conn->uc_lock);
699                 conn->uc_sending = 0;
700                 if (conn->uc_state != UC_DEAD &&
701                     list_empty(&conn->uc_tx_list) &&
702                     list_empty(&conn->uc_zcack_list)) {
703                         conn->uc_tx_flag = 0;
704                         ret = usocklnd_add_pollrequest(conn,
705                                                       POLL_TX_SET_REQUEST, 0);
706                         if (ret)
707                                 rc = ret;
708                 }
709                 pthread_mutex_unlock(&conn->uc_lock);
710
711                 break;
712
713         case UC_DEAD:
714                 break;
715
716         default:
717                 LBUG();
718         }
719
720         if (rc < 0)
721                 usocklnd_conn_kill(conn);
722
723         return rc;
724 }
725
726 /* Return the first tx from tx_list with piggybacked zc_ack
727  * from zcack_list when possible. If tx_list is empty, return
728  * brand new noop tx for zc_ack from zcack_list. Return NULL
729  * if an error happened */
730 usock_tx_t *
731 usocklnd_try_piggyback(struct list_head *tx_list_p,
732                        struct list_head *zcack_list_p)
733 {
734         usock_tx_t     *tx;
735         usock_zc_ack_t *zc_ack;
736
737         /* assign tx and zc_ack */
738         if (list_empty(tx_list_p))
739                 tx = NULL;
740         else {
741                 tx = list_entry(tx_list_p->next, usock_tx_t, tx_list);
742                 list_del(&tx->tx_list);
743
744                 /* already piggybacked or partially send */
745                 if (tx->tx_msg.ksm_zc_cookies[1] != 0 ||
746                     tx->tx_resid != tx->tx_nob)
747                         return tx;
748         }
749
750         if (list_empty(zcack_list_p)) {
751                 /* nothing to piggyback */
752                 return tx;
753         } else {
754                 zc_ack = list_entry(zcack_list_p->next,
755                                     usock_zc_ack_t, zc_list);
756                 list_del(&zc_ack->zc_list);
757         }
758
759         if (tx != NULL)
760                 /* piggyback the zc-ack cookie */
761                 tx->tx_msg.ksm_zc_cookies[1] = zc_ack->zc_cookie;
762         else
763                 /* cannot piggyback, need noop */
764                 tx = usocklnd_create_noop_tx(zc_ack->zc_cookie);
765
766         LIBCFS_FREE (zc_ack, sizeof(*zc_ack));
767         return tx;
768 }
769
770 /* All actions that we need after sending hello on active conn:
771  * 1) update RX iov to receive hello
772  * 2) state transition to UC_RECEIVING_HELLO
773  * 3) notify poll_thread that we're waiting for incoming hello */
774 int
775 usocklnd_activeconn_hellosent(usock_conn_t *conn)
776 {
777         int rc = 0;
778
779         pthread_mutex_lock(&conn->uc_lock);
780
781         if (conn->uc_state != UC_DEAD) {
782                 usocklnd_rx_hellomagic_state_transition(conn);
783                 conn->uc_state = UC_RECEIVING_HELLO;
784                 conn->uc_tx_flag = 0;
785                 rc = usocklnd_add_pollrequest(conn, POLL_SET_REQUEST, POLLIN);
786         }
787
788         pthread_mutex_unlock(&conn->uc_lock);
789
790         return rc;
791 }
792
793 /* All actions that we need after sending hello on passive conn:
794  * 1) Cope with 1st easy case: conn is already linked to a peer
795  * 2) Cope with 2nd easy case: remove zombie conn
796   * 3) Resolve race:
797  *    a) find the peer
798  *    b) link the conn to the peer if conn[idx] is empty
799  *    c) if the conn[idx] isn't empty and is in READY state,
800  *       remove the conn as duplicated
801  *    d) if the conn[idx] isn't empty and isn't in READY state,
802  *       override conn[idx] with the conn
803  */
804 int
805 usocklnd_passiveconn_hellosent(usock_conn_t *conn)
806 {
807         usock_conn_t    *conn2;
808         usock_peer_t    *peer;
809         struct list_head tx_list;
810         struct list_head zcack_list;
811         int              idx;
812         int              rc = 0;
813
814         /* almost nothing to do if conn is already linked to peer hash table */
815         if (conn->uc_peer != NULL)
816                 goto passive_hellosent_done;
817
818         /* conn->uc_peer == NULL, so the conn isn't accessible via
819          * peer hash list, so nobody can touch the conn but us */
820
821         if (conn->uc_ni == NULL) /* remove zombie conn */
822                 goto passive_hellosent_connkill;
823
824         /* all code below is race resolution, because normally
825          * passive conn is linked to peer just after receiving hello */
826         CFS_INIT_LIST_HEAD (&tx_list);
827         CFS_INIT_LIST_HEAD (&zcack_list);
828
829         /* conn is passive and isn't linked to any peer,
830            so its tx and zc_ack lists have to be empty */
831         LASSERT (list_empty(&conn->uc_tx_list) &&
832                  list_empty(&conn->uc_zcack_list) &&
833                  conn->uc_sending == 0);
834
835         rc = usocklnd_find_or_create_peer(conn->uc_ni, conn->uc_peerid, &peer);
836         if (rc)
837                 return rc;
838
839         idx = usocklnd_type2idx(conn->uc_type);
840
841         /* try to link conn to peer */
842         pthread_mutex_lock(&peer->up_lock);
843         if (peer->up_conns[idx] == NULL) {
844                 usocklnd_link_conn_to_peer(conn, peer, idx);
845                 usocklnd_conn_addref(conn);
846                 conn->uc_peer = peer;
847                 usocklnd_peer_addref(peer);
848         } else {
849                 conn2 = peer->up_conns[idx];
850                 pthread_mutex_lock(&conn2->uc_lock);
851
852                 if (conn2->uc_state == UC_READY) {
853                         /* conn2 is in READY state, so conn is "duplicated" */
854                         pthread_mutex_unlock(&conn2->uc_lock);
855                         pthread_mutex_unlock(&peer->up_lock);
856                         usocklnd_peer_decref(peer);
857                         goto passive_hellosent_connkill;
858                 }
859
860                 /* uc_state != UC_READY => switch conn and conn2 */
861                 /* Relink txs and zc_acks from conn2 to conn.
862                  * We're sure that nobody but us can access to conn,
863                  * nevertheless we use mutex (if we're wrong yet,
864                  * deadlock is easy to see that corrupted list */
865                 list_add(&tx_list, &conn2->uc_tx_list);
866                 list_del_init(&conn2->uc_tx_list);
867                 list_add(&zcack_list, &conn2->uc_zcack_list);
868                 list_del_init(&conn2->uc_zcack_list);
869
870                 pthread_mutex_lock(&conn->uc_lock);
871                 list_add_tail(&conn->uc_tx_list, &tx_list);
872                 list_del_init(&tx_list);
873                 list_add_tail(&conn->uc_zcack_list, &zcack_list);
874                 list_del_init(&zcack_list);
875                 conn->uc_peer = peer;
876                 pthread_mutex_unlock(&conn->uc_lock);
877
878                 conn2->uc_peer = NULL; /* make conn2 zombie */
879                 pthread_mutex_unlock(&conn2->uc_lock);
880                 usocklnd_conn_decref(conn2);
881
882                 usocklnd_link_conn_to_peer(conn, peer, idx);
883                 usocklnd_conn_addref(conn);
884                 conn->uc_peer = peer;
885         }
886
887         lnet_ni_decref(conn->uc_ni);
888         conn->uc_ni = NULL;
889         pthread_mutex_unlock(&peer->up_lock);
890         usocklnd_peer_decref(peer);
891
892   passive_hellosent_done:
893         /* safely transit to UC_READY state */
894         /* rc == 0 */
895         pthread_mutex_lock(&conn->uc_lock);
896         if (conn->uc_state != UC_DEAD) {
897                 usocklnd_rx_ksmhdr_state_transition(conn);
898
899                 /* we're ready to recive incoming packets and maybe
900                    already have smth. to transmit */
901                 LASSERT (conn->uc_sending == 0);
902                 if ( list_empty(&conn->uc_tx_list) &&
903                      list_empty(&conn->uc_zcack_list) ) {
904                         conn->uc_tx_flag = 0;
905                         rc = usocklnd_add_pollrequest(conn, POLL_SET_REQUEST,
906                                                  POLLIN);
907                 } else {
908                         conn->uc_tx_deadline =
909                                 cfs_time_shift(usock_tuns.ut_timeout);
910                         conn->uc_tx_flag = 1;
911                         rc = usocklnd_add_pollrequest(conn, POLL_SET_REQUEST,
912                                                       POLLIN | POLLOUT);
913                 }
914
915                 if (rc == 0)
916                         conn->uc_state = UC_READY;
917         }
918         pthread_mutex_unlock(&conn->uc_lock);
919         return rc;
920
921   passive_hellosent_connkill:
922         usocklnd_conn_kill(conn);
923         return 0;
924 }
925
926 /* Send as much tx data as possible.
927  * Returns 0 or 1 on succsess, <0 if fatal error.
928  * 0 means partial send or non-fatal error, 1 - complete.
929  * Rely on libcfs_sock_writev() for differentiating fatal and
930  * non-fatal errors. An error should be considered as non-fatal if:
931  * 1) it still makes sense to continue reading &&
932  * 2) anyway, poll() will set up POLLHUP|POLLERR flags */
933 int
934 usocklnd_send_tx(usock_conn_t *conn, usock_tx_t *tx)
935 {
936         struct iovec *iov;
937         int           nob;
938         cfs_time_t    t;
939
940         LASSERT (tx->tx_resid != 0);
941
942         do {
943                 usock_peer_t *peer = conn->uc_peer;
944
945                 LASSERT (tx->tx_niov > 0);
946
947                 nob = libcfs_sock_writev(conn->uc_sock,
948                                          tx->tx_iov, tx->tx_niov);
949                 if (nob < 0)
950                         conn->uc_errored = 1;
951                 if (nob <= 0) /* write queue is flow-controlled or error */
952                         return nob;
953
954                 LASSERT (nob <= tx->tx_resid);
955                 tx->tx_resid -= nob;
956                 t = cfs_time_current();
957                 conn->uc_tx_deadline = cfs_time_add(t, cfs_time_seconds(usock_tuns.ut_timeout));
958
959                 if(peer != NULL)
960                         peer->up_last_alive = t;
961
962                 /* "consume" iov */
963                 iov = tx->tx_iov;
964                 do {
965                         LASSERT (tx->tx_niov > 0);
966
967                         if (nob < iov->iov_len) {
968                                 iov->iov_base = (void *)(((unsigned long)(iov->iov_base)) + nob);
969                                 iov->iov_len -= nob;
970                                 break;
971                         }
972
973                         nob -= iov->iov_len;
974                         tx->tx_iov = ++iov;
975                         tx->tx_niov--;
976                 } while (nob != 0);
977
978         } while (tx->tx_resid != 0);
979
980         return 1; /* send complete */
981 }
982
983 /* Read from wire as much data as possible.
984  * Returns 0 or 1 on succsess, <0 if error or EOF.
985  * 0 means partial read, 1 - complete */
986 int
987 usocklnd_read_data(usock_conn_t *conn)
988 {
989         struct iovec *iov;
990         int           nob;
991         cfs_time_t    t;
992
993         LASSERT (conn->uc_rx_nob_wanted != 0);
994
995         do {
996                 usock_peer_t *peer = conn->uc_peer;
997
998                 LASSERT (conn->uc_rx_niov > 0);
999
1000                 nob = libcfs_sock_readv(conn->uc_sock,
1001                                         conn->uc_rx_iov, conn->uc_rx_niov);
1002                 if (nob <= 0) {/* read nothing or error */
1003                         conn->uc_errored = 1;
1004                         return nob;
1005                 }
1006
1007                 LASSERT (nob <= conn->uc_rx_nob_wanted);
1008                 conn->uc_rx_nob_wanted -= nob;
1009                 conn->uc_rx_nob_left -= nob;
1010                 t = cfs_time_current();
1011                 conn->uc_rx_deadline = cfs_time_add(t, cfs_time_seconds(usock_tuns.ut_timeout));
1012
1013                 if(peer != NULL)
1014                         peer->up_last_alive = t;
1015
1016                 /* "consume" iov */
1017                 iov = conn->uc_rx_iov;
1018                 do {
1019                         LASSERT (conn->uc_rx_niov > 0);
1020
1021                         if (nob < iov->iov_len) {
1022                                 iov->iov_base = (void *)(((unsigned long)(iov->iov_base)) + nob);
1023                                 iov->iov_len -= nob;
1024                                 break;
1025                         }
1026
1027                         nob -= iov->iov_len;
1028                         conn->uc_rx_iov = ++iov;
1029                         conn->uc_rx_niov--;
1030                 } while (nob != 0);
1031
1032         } while (conn->uc_rx_nob_wanted != 0);
1033
1034         return 1; /* read complete */
1035 }