Whamcloud - gitweb
533e8bb71c8102bcbcf6cde6dce2083447106a5a
[fs/lustre-release.git] / lnet / klnds / socklnd / socklnd_proto.c
1 /*
2  * Copyright (c) 2009, 2010, Oracle and/or its affiliates. All rights reserved.
3  *
4  * Copyright (c) 2012, 2017, Intel Corporation.
5  *
6  *   Author: Zach Brown <zab@zabbo.net>
7  *   Author: Peter J. Braam <braam@clusterfs.com>
8  *   Author: Phil Schwan <phil@clusterfs.com>
9  *   Author: Eric Barton <eric@bartonsoftware.com>
10  *
11  *   This file is part of Lustre, https://wiki.whamcloud.com/
12  *
13  *   Portals is free software; you can redistribute it and/or
14  *   modify it under the terms of version 2 of the GNU General Public
15  *   License as published by the Free Software Foundation.
16  *
17  *   Portals is distributed in the hope that it will be useful,
18  *   but WITHOUT ANY WARRANTY; without even the implied warranty of
19  *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
20  *   GNU General Public License for more details.
21  *
22  *   You should have received a copy of the GNU General Public License
23  *   along with Portals; if not, write to the Free Software
24  *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
25  */
26
27 #include "socklnd.h"
28
29 /*
30  * Protocol entries :
31  *   pro_send_hello       : send hello message
32  *   pro_recv_hello       : receive hello message
33  *   pro_pack             : pack message header
34  *   pro_unpack           : unpack message header
35  *   pro_queue_tx_zcack() : Called holding BH lock: kss_lock
36  *                          return 1 if ACK is piggybacked, otherwise return 0
37  *   pro_queue_tx_msg()   : Called holding BH lock: kss_lock
38  *                          return the ACK that piggybacked by my message, or NULL
39  *   pro_handle_zcreq()   : handler of incoming ZC-REQ
40  *   pro_handle_zcack()   : handler of incoming ZC-ACK
41  *   pro_match_tx()       : Called holding glock
42  */
43
44 static struct ksock_tx *
45 ksocknal_queue_tx_msg_v1(struct ksock_conn *conn, struct ksock_tx *tx_msg)
46 {
47         /* V1.x, just enqueue it */
48         list_add_tail(&tx_msg->tx_list, &conn->ksnc_tx_queue);
49         return NULL;
50 }
51
52 void
53 ksocknal_next_tx_carrier(struct ksock_conn *conn)
54 {
55         struct ksock_tx *tx = conn->ksnc_tx_carrier;
56
57         /* Called holding BH lock: conn->ksnc_scheduler->kss_lock */
58         LASSERT(!list_empty(&conn->ksnc_tx_queue));
59         LASSERT(tx != NULL);
60
61         /* Next TX that can carry ZC-ACK or LNet message */
62         if (tx->tx_list.next == &conn->ksnc_tx_queue) {
63                 /* no more packets queued */
64                 conn->ksnc_tx_carrier = NULL;
65         } else {
66                 conn->ksnc_tx_carrier = list_next_entry(tx, tx_list);
67                 LASSERT(conn->ksnc_tx_carrier->tx_msg.ksm_type ==
68                         tx->tx_msg.ksm_type);
69         }
70 }
71
72 static int
73 ksocknal_queue_tx_zcack_v2(struct ksock_conn *conn,
74                            struct ksock_tx *tx_ack, __u64 cookie)
75 {
76         struct ksock_tx *tx = conn->ksnc_tx_carrier;
77
78         LASSERT (tx_ack == NULL ||
79                  tx_ack->tx_msg.ksm_type == KSOCK_MSG_NOOP);
80
81         /*
82          * Enqueue or piggyback tx_ack / cookie
83          * . no tx can piggyback cookie of tx_ack (or cookie), just
84          *   enqueue the tx_ack (if tx_ack != NUL) and return NULL.
85          * . There is tx can piggyback cookie of tx_ack (or cookie),
86          *   piggyback the cookie and return the tx.
87          */
88         if (tx == NULL) {
89                 if (tx_ack != NULL) {
90                         list_add_tail(&tx_ack->tx_list,
91                                           &conn->ksnc_tx_queue);
92                         conn->ksnc_tx_carrier = tx_ack;
93                 }
94                 return 0;
95         }
96
97         if (tx->tx_msg.ksm_type == KSOCK_MSG_NOOP) {
98                 /* tx is noop zc-ack, can't piggyback zc-ack cookie */
99                 if (tx_ack != NULL)
100                         list_add_tail(&tx_ack->tx_list,
101                                           &conn->ksnc_tx_queue);
102                 return 0;
103         }
104
105         LASSERT(tx->tx_msg.ksm_type == KSOCK_MSG_LNET);
106         LASSERT(tx->tx_msg.ksm_zc_cookies[1] == 0);
107
108         if (tx_ack != NULL)
109                 cookie = tx_ack->tx_msg.ksm_zc_cookies[1];
110
111         /* piggyback the zc-ack cookie */
112         tx->tx_msg.ksm_zc_cookies[1] = cookie;
113         /* move on to the next TX which can carry cookie */
114         ksocknal_next_tx_carrier(conn);
115
116         return 1;
117 }
118
119 static struct ksock_tx *
120 ksocknal_queue_tx_msg_v2(struct ksock_conn *conn, struct ksock_tx *tx_msg)
121 {
122         struct ksock_tx  *tx  = conn->ksnc_tx_carrier;
123
124         /*
125          * Enqueue tx_msg:
126          * . If there is no NOOP on the connection, just enqueue
127          *   tx_msg and return NULL
128          * . If there is NOOP on the connection, piggyback the cookie
129          *   and replace the NOOP tx, and return the NOOP tx.
130          */
131         if (tx == NULL) { /* nothing on queue */
132                 list_add_tail(&tx_msg->tx_list, &conn->ksnc_tx_queue);
133                 conn->ksnc_tx_carrier = tx_msg;
134                 return NULL;
135         }
136
137         if (tx->tx_msg.ksm_type == KSOCK_MSG_LNET) { /* nothing to carry */
138                 list_add_tail(&tx_msg->tx_list, &conn->ksnc_tx_queue);
139                 return NULL;
140         }
141
142         LASSERT (tx->tx_msg.ksm_type == KSOCK_MSG_NOOP);
143
144         /* There is a noop zc-ack can be piggybacked */
145         tx_msg->tx_msg.ksm_zc_cookies[1] = tx->tx_msg.ksm_zc_cookies[1];
146         ksocknal_next_tx_carrier(conn);
147
148         /* use new_tx to replace the noop zc-ack packet */
149         list_splice(&tx->tx_list, &tx_msg->tx_list);
150
151         return tx;
152 }
153
154 static int
155 ksocknal_queue_tx_zcack_v3(struct ksock_conn *conn,
156                            struct ksock_tx *tx_ack, __u64 cookie)
157 {
158         struct ksock_tx *tx;
159
160         if (conn->ksnc_type != SOCKLND_CONN_ACK)
161                 return ksocknal_queue_tx_zcack_v2(conn, tx_ack, cookie);
162
163         /* non-blocking ZC-ACK (to router) */
164         LASSERT (tx_ack == NULL ||
165                  tx_ack->tx_msg.ksm_type == KSOCK_MSG_NOOP);
166
167         if ((tx = conn->ksnc_tx_carrier) == NULL) {
168                 if (tx_ack != NULL) {
169                         list_add_tail(&tx_ack->tx_list,
170                                           &conn->ksnc_tx_queue);
171                         conn->ksnc_tx_carrier = tx_ack;
172                 }
173                 return 0;
174         }
175
176         /* conn->ksnc_tx_carrier != NULL */
177
178         if (tx_ack != NULL)
179                 cookie = tx_ack->tx_msg.ksm_zc_cookies[1];
180
181         if (cookie == SOCKNAL_KEEPALIVE_PING) /* ignore keepalive PING */
182                 return 1;
183
184         if (tx->tx_msg.ksm_zc_cookies[1] == SOCKNAL_KEEPALIVE_PING) {
185                 /* replace the keepalive PING with a real ACK */
186                 LASSERT (tx->tx_msg.ksm_zc_cookies[0] == 0);
187                 tx->tx_msg.ksm_zc_cookies[1] = cookie;
188                 return 1;
189         }
190
191         if (cookie == tx->tx_msg.ksm_zc_cookies[0] ||
192             cookie == tx->tx_msg.ksm_zc_cookies[1]) {
193                 CWARN("%s: duplicated ZC cookie: %llu\n",
194                       libcfs_id2str(conn->ksnc_peer->ksnp_id), cookie);
195                 return 1; /* XXX return error in the future */
196         }
197
198         if (tx->tx_msg.ksm_zc_cookies[0] == 0) {
199                 /* NOOP tx has only one ZC-ACK cookie, can carry at least one more */
200                 if (tx->tx_msg.ksm_zc_cookies[1] > cookie) {
201                         tx->tx_msg.ksm_zc_cookies[0] = tx->tx_msg.ksm_zc_cookies[1];
202                         tx->tx_msg.ksm_zc_cookies[1] = cookie;
203                 } else {
204                         tx->tx_msg.ksm_zc_cookies[0] = cookie;
205                 }
206
207                 if (tx->tx_msg.ksm_zc_cookies[0] - tx->tx_msg.ksm_zc_cookies[1] > 2) {
208                         /* not likely to carry more ACKs, skip it to simplify logic */
209                         ksocknal_next_tx_carrier(conn);
210                 }
211
212                 return 1;
213         }
214
215         /* takes two or more cookies already */
216
217         if (tx->tx_msg.ksm_zc_cookies[0] > tx->tx_msg.ksm_zc_cookies[1]) {
218                 __u64   tmp = 0;
219
220                 /* two separated cookies: (a+2, a) or (a+1, a) */
221                 LASSERT (tx->tx_msg.ksm_zc_cookies[0] -
222                          tx->tx_msg.ksm_zc_cookies[1] <= 2);
223
224                 if (tx->tx_msg.ksm_zc_cookies[0] -
225                     tx->tx_msg.ksm_zc_cookies[1] == 2) {
226                         if (cookie == tx->tx_msg.ksm_zc_cookies[1] + 1)
227                                 tmp = cookie;
228                 } else if (cookie == tx->tx_msg.ksm_zc_cookies[1] - 1) {
229                         tmp = tx->tx_msg.ksm_zc_cookies[1];
230                 } else if (cookie == tx->tx_msg.ksm_zc_cookies[0] + 1) {
231                         tmp = tx->tx_msg.ksm_zc_cookies[0];
232                 }
233
234                 if (tmp != 0) {
235                         /* range of cookies */
236                         tx->tx_msg.ksm_zc_cookies[0] = tmp - 1;
237                         tx->tx_msg.ksm_zc_cookies[1] = tmp + 1;
238                         return 1;
239                 }
240
241         } else {
242                 /* ksm_zc_cookies[0] < ksm_zc_cookies[1], it is range of cookies */
243                 if (cookie >= tx->tx_msg.ksm_zc_cookies[0] &&
244                     cookie <= tx->tx_msg.ksm_zc_cookies[1]) {
245                         CWARN("%s: duplicated ZC cookie: %llu\n",
246                               libcfs_id2str(conn->ksnc_peer->ksnp_id), cookie);
247                         return 1; /* XXX: return error in the future */
248                 }
249
250                 if (cookie == tx->tx_msg.ksm_zc_cookies[1] + 1) {
251                         tx->tx_msg.ksm_zc_cookies[1] = cookie;
252                         return 1;
253                 }
254
255                 if (cookie == tx->tx_msg.ksm_zc_cookies[0] - 1) {
256                         tx->tx_msg.ksm_zc_cookies[0] = cookie;
257                         return 1;
258                 }
259         }
260
261         /* failed to piggyback ZC-ACK */
262         if (tx_ack != NULL) {
263                 list_add_tail(&tx_ack->tx_list, &conn->ksnc_tx_queue);
264                 /* the next tx can piggyback at least 1 ACK */
265                 ksocknal_next_tx_carrier(conn);
266         }
267
268         return 0;
269 }
270
271 static int
272 ksocknal_match_tx(struct ksock_conn *conn, struct ksock_tx *tx, int nonblk)
273 {
274         int nob;
275
276 #if SOCKNAL_VERSION_DEBUG
277         if (!*ksocknal_tunables.ksnd_typed_conns)
278                 return SOCKNAL_MATCH_YES;
279 #endif
280
281         if (tx == NULL || tx->tx_lnetmsg == NULL) {
282                 /* noop packet */
283                 nob = offsetof(struct ksock_msg, ksm_u);
284         } else {
285                 nob = tx->tx_lnetmsg->msg_len +
286                       ((conn->ksnc_proto == &ksocknal_protocol_v1x) ?
287                        sizeof(struct lnet_hdr) : sizeof(struct ksock_msg));
288         }
289
290         /* default checking for typed connection */
291         switch (conn->ksnc_type) {
292         default:
293                 CERROR("ksnc_type bad: %u\n", conn->ksnc_type);
294                 LBUG();
295         case SOCKLND_CONN_ANY:
296                 return SOCKNAL_MATCH_YES;
297
298         case SOCKLND_CONN_BULK_IN:
299                 return SOCKNAL_MATCH_MAY;
300
301         case SOCKLND_CONN_BULK_OUT:
302                 if (nob < *ksocknal_tunables.ksnd_min_bulk)
303                         return SOCKNAL_MATCH_MAY;
304                 else
305                         return SOCKNAL_MATCH_YES;
306
307         case SOCKLND_CONN_CONTROL:
308                 if (nob >= *ksocknal_tunables.ksnd_min_bulk)
309                         return SOCKNAL_MATCH_MAY;
310                 else
311                         return SOCKNAL_MATCH_YES;
312         }
313 }
314
315 static int
316 ksocknal_match_tx_v3(struct ksock_conn *conn, struct ksock_tx *tx, int nonblk)
317 {
318         int nob;
319
320         if (tx == NULL || tx->tx_lnetmsg == NULL)
321                 nob = offsetof(struct ksock_msg, ksm_u);
322         else
323                 nob = tx->tx_lnetmsg->msg_len + sizeof(struct ksock_msg);
324
325         switch (conn->ksnc_type) {
326         default:
327                 CERROR("ksnc_type bad: %u\n", conn->ksnc_type);
328                 LBUG();
329         case SOCKLND_CONN_ANY:
330                 return SOCKNAL_MATCH_NO;
331
332         case SOCKLND_CONN_ACK:
333                 if (nonblk)
334                         return SOCKNAL_MATCH_YES;
335                 else if (tx == NULL || tx->tx_lnetmsg == NULL)
336                         return SOCKNAL_MATCH_MAY;
337                 else
338                         return SOCKNAL_MATCH_NO;
339
340         case SOCKLND_CONN_BULK_OUT:
341                 if (nonblk)
342                         return SOCKNAL_MATCH_NO;
343                 else if (nob < *ksocknal_tunables.ksnd_min_bulk)
344                         return SOCKNAL_MATCH_MAY;
345                 else
346                         return SOCKNAL_MATCH_YES;
347
348         case SOCKLND_CONN_CONTROL:
349                 if (nonblk)
350                         return SOCKNAL_MATCH_NO;
351                 else if (nob >= *ksocknal_tunables.ksnd_min_bulk)
352                         return SOCKNAL_MATCH_MAY;
353                 else
354                         return SOCKNAL_MATCH_YES;
355         }
356 }
357
358 /* (Sink) handle incoming ZC request from sender */
359 static int
360 ksocknal_handle_zcreq(struct ksock_conn *c, __u64 cookie, int remote)
361 {
362         struct ksock_peer_ni *peer_ni = c->ksnc_peer;
363         struct ksock_conn *conn;
364         struct ksock_tx *tx;
365         int rc;
366
367         read_lock(&ksocknal_data.ksnd_global_lock);
368
369         conn = ksocknal_find_conn_locked(peer_ni, NULL, !!remote);
370         if (conn != NULL) {
371                 struct ksock_sched *sched = conn->ksnc_scheduler;
372
373                 LASSERT(conn->ksnc_proto->pro_queue_tx_zcack != NULL);
374
375                 spin_lock_bh(&sched->kss_lock);
376
377                 rc = conn->ksnc_proto->pro_queue_tx_zcack(conn, NULL, cookie);
378
379                 spin_unlock_bh(&sched->kss_lock);
380
381                 if (rc) { /* piggybacked */
382                         read_unlock(&ksocknal_data.ksnd_global_lock);
383                         return 0;
384                 }
385         }
386
387         read_unlock(&ksocknal_data.ksnd_global_lock);
388
389         /* ACK connection is not ready, or can't piggyback the ACK */
390         tx = ksocknal_alloc_tx_noop(cookie, !!remote);
391         if (tx == NULL)
392                 return -ENOMEM;
393
394         if ((rc = ksocknal_launch_packet(peer_ni->ksnp_ni, tx, peer_ni->ksnp_id)) == 0)
395                 return 0;
396
397         ksocknal_free_tx(tx);
398         return rc;
399 }
400
401 /* (Sender) handle ZC_ACK from sink */
402 static int
403 ksocknal_handle_zcack(struct ksock_conn *conn, __u64 cookie1, __u64 cookie2)
404 {
405         struct ksock_peer_ni *peer_ni = conn->ksnc_peer;
406         struct ksock_tx *tx;
407         struct ksock_tx *tmp;
408         LIST_HEAD(zlist);
409         int count;
410
411         if (cookie1 == 0)
412                 cookie1 = cookie2;
413
414         count = (cookie1 > cookie2) ? 2 : (cookie2 - cookie1 + 1);
415
416         if (cookie2 == SOCKNAL_KEEPALIVE_PING &&
417             conn->ksnc_proto == &ksocknal_protocol_v3x) {
418                 /* keepalive PING for V3.x, just ignore it */
419                 return count == 1 ? 0 : -EPROTO;
420         }
421
422         spin_lock(&peer_ni->ksnp_lock);
423
424         list_for_each_entry_safe(tx, tmp, &peer_ni->ksnp_zc_req_list,
425                                  tx_zc_list) {
426                 __u64 c = tx->tx_msg.ksm_zc_cookies[0];
427
428                 if (c == cookie1 || c == cookie2 || (cookie1 < c && c < cookie2)) {
429                         tx->tx_msg.ksm_zc_cookies[0] = 0;
430                         list_move(&tx->tx_zc_list, &zlist);
431
432                         if (--count == 0)
433                                 break;
434                 }
435         }
436
437         spin_unlock(&peer_ni->ksnp_lock);
438
439         while ((tx = list_first_entry_or_null(&zlist, struct ksock_tx,
440                                               tx_zc_list)) != NULL) {
441                 list_del(&tx->tx_zc_list);
442                 ksocknal_tx_decref(tx);
443         }
444
445         return count == 0 ? 0 : -EPROTO;
446 }
447
448 static int
449 ksocknal_send_hello_v1(struct ksock_conn *conn, struct ksock_hello_msg *hello)
450 {
451         struct socket *sock = conn->ksnc_sock;
452         struct lnet_hdr *hdr;
453         struct lnet_magicversion *hmv;
454         int rc;
455         int i;
456
457         BUILD_BUG_ON(sizeof(struct lnet_magicversion) !=
458                      offsetof(struct lnet_hdr, src_nid));
459
460         LIBCFS_ALLOC(hdr, sizeof(*hdr));
461         if (hdr == NULL) {
462                 CERROR("Can't allocate struct lnet_hdr\n");
463                 return -ENOMEM;
464         }
465
466         hmv = (struct lnet_magicversion *)&hdr->dest_nid;
467
468         /* Re-organize V2.x message header to V1.x (struct lnet_hdr)
469          * header and send out */
470         hmv->magic         = cpu_to_le32 (LNET_PROTO_TCP_MAGIC);
471         hmv->version_major = cpu_to_le16 (KSOCK_PROTO_V1_MAJOR);
472         hmv->version_minor = cpu_to_le16 (KSOCK_PROTO_V1_MINOR);
473
474         if (the_lnet.ln_testprotocompat) {
475                 /* single-shot proto check */
476                 if (test_and_clear_bit(0, &the_lnet.ln_testprotocompat))
477                         hmv->version_major++;   /* just different! */
478
479                 if (test_and_clear_bit(1, &the_lnet.ln_testprotocompat))
480                         hmv->magic = LNET_PROTO_MAGIC;
481         }
482
483         hdr->src_nid        = cpu_to_le64 (hello->kshm_src_nid);
484         hdr->src_pid        = cpu_to_le32 (hello->kshm_src_pid);
485         hdr->type           = cpu_to_le32 (LNET_MSG_HELLO);
486         hdr->payload_length = cpu_to_le32 (hello->kshm_nips * sizeof(__u32));
487         hdr->msg.hello.type = cpu_to_le32 (hello->kshm_ctype);
488         hdr->msg.hello.incarnation = cpu_to_le64 (hello->kshm_src_incarnation);
489
490         rc = lnet_sock_write(sock, hdr, sizeof(*hdr), lnet_acceptor_timeout());
491         if (rc != 0) {
492                 CNETERR("Error %d sending HELLO hdr to %pISp\n",
493                         rc, &conn->ksnc_peeraddr);
494                 goto out;
495         }
496
497         if (hello->kshm_nips == 0)
498                 goto out;
499
500         for (i = 0; i < (int) hello->kshm_nips; i++) {
501                 hello->kshm_ips[i] = __cpu_to_le32 (hello->kshm_ips[i]);
502         }
503
504         rc = lnet_sock_write(sock, hello->kshm_ips,
505                              hello->kshm_nips * sizeof(__u32),
506                              lnet_acceptor_timeout());
507         if (rc != 0) {
508                 CNETERR("Error %d sending HELLO payload (%d) to %pISp\n",
509                         rc, hello->kshm_nips,
510                         &conn->ksnc_peeraddr);
511         }
512 out:
513         LIBCFS_FREE(hdr, sizeof(*hdr));
514
515         return rc;
516 }
517
518 static int
519 ksocknal_send_hello_v2(struct ksock_conn *conn, struct ksock_hello_msg *hello)
520 {
521         struct socket *sock = conn->ksnc_sock;
522         int rc;
523
524         hello->kshm_magic   = LNET_PROTO_MAGIC;
525         hello->kshm_version = conn->ksnc_proto->pro_version;
526
527         if (the_lnet.ln_testprotocompat) {
528                 /* single-shot proto check */
529                 if (test_and_clear_bit(0, &the_lnet.ln_testprotocompat))
530                         hello->kshm_version++;   /* just different! */
531         }
532
533         rc = lnet_sock_write(sock, hello, offsetof(struct ksock_hello_msg, kshm_ips),
534                                lnet_acceptor_timeout());
535
536         if (rc != 0) {
537                 CNETERR("Error %d sending HELLO hdr to %pISp\n",
538                         rc, &conn->ksnc_peeraddr);
539                 return rc;
540         }
541
542         if (hello->kshm_nips == 0)
543                 return 0;
544
545         rc = lnet_sock_write(sock, hello->kshm_ips,
546                              hello->kshm_nips * sizeof(__u32),
547                              lnet_acceptor_timeout());
548         if (rc != 0) {
549                 CNETERR("Error %d sending HELLO payload (%d) to %pISp\n", rc,
550                         hello->kshm_nips,
551                         &conn->ksnc_peeraddr);
552         }
553
554         return rc;
555 }
556
557 static int
558 ksocknal_recv_hello_v1(struct ksock_conn *conn, struct ksock_hello_msg *hello,
559                        int timeout)
560 {
561         struct socket *sock = conn->ksnc_sock;
562         struct lnet_hdr *hdr;
563         int rc;
564         int i;
565
566         LIBCFS_ALLOC(hdr, sizeof(*hdr));
567         if (hdr == NULL) {
568                 CERROR("Can't allocate struct lnet_hdr\n");
569                 return -ENOMEM;
570         }
571
572         rc = lnet_sock_read(sock, &hdr->src_nid,
573                               sizeof(*hdr) - offsetof(struct lnet_hdr, src_nid),
574                               timeout);
575         if (rc != 0) {
576                 CERROR("Error %d reading rest of HELLO hdr from %pIS\n",
577                        rc, &conn->ksnc_peeraddr);
578                 LASSERT(rc < 0 && rc != -EALREADY);
579                 goto out;
580         }
581
582         /* ...and check we got what we expected */
583         if (hdr->type != cpu_to_le32 (LNET_MSG_HELLO)) {
584                 CERROR("Expecting a HELLO hdr, but got type %d from %pIS\n",
585                        le32_to_cpu(hdr->type),
586                        &conn->ksnc_peeraddr);
587                 rc = -EPROTO;
588                 goto out;
589         }
590
591         hello->kshm_src_nid         = le64_to_cpu (hdr->src_nid);
592         hello->kshm_src_pid         = le32_to_cpu (hdr->src_pid);
593         hello->kshm_src_incarnation = le64_to_cpu (hdr->msg.hello.incarnation);
594         hello->kshm_ctype           = le32_to_cpu (hdr->msg.hello.type);
595         hello->kshm_nips            = le32_to_cpu (hdr->payload_length) /
596                                          sizeof (__u32);
597
598         if (hello->kshm_nips > LNET_INTERFACES_NUM) {
599                 CERROR("Bad nips %d from ip %pIS\n",
600                        hello->kshm_nips, &conn->ksnc_peeraddr);
601                 rc = -EPROTO;
602                 goto out;
603         }
604
605         if (hello->kshm_nips == 0)
606                 goto out;
607
608         rc = lnet_sock_read(sock, hello->kshm_ips,
609                               hello->kshm_nips * sizeof(__u32), timeout);
610         if (rc != 0) {
611                 CERROR("Error %d reading IPs from ip %pIS\n",
612                        rc, &conn->ksnc_peeraddr);
613                 LASSERT(rc < 0 && rc != -EALREADY);
614                 goto out;
615         }
616
617         for (i = 0; i < (int) hello->kshm_nips; i++) {
618                 hello->kshm_ips[i] = __le32_to_cpu(hello->kshm_ips[i]);
619
620                 if (hello->kshm_ips[i] == 0) {
621                         CERROR("Zero IP[%d] from ip %pIS\n",
622                                i, &conn->ksnc_peeraddr);
623                         rc = -EPROTO;
624                         break;
625                 }
626         }
627 out:
628         LIBCFS_FREE(hdr, sizeof(*hdr));
629
630         return rc;
631 }
632
633 static int
634 ksocknal_recv_hello_v2(struct ksock_conn *conn, struct ksock_hello_msg *hello,
635                        int timeout)
636 {
637         struct socket     *sock = conn->ksnc_sock;
638         int                rc;
639         int                i;
640
641         if (hello->kshm_magic == LNET_PROTO_MAGIC)
642                 conn->ksnc_flip = 0;
643         else
644                 conn->ksnc_flip = 1;
645
646         rc = lnet_sock_read(sock, &hello->kshm_src_nid,
647                               offsetof(struct ksock_hello_msg, kshm_ips) -
648                                        offsetof(struct ksock_hello_msg, kshm_src_nid),
649                               timeout);
650         if (rc != 0) {
651                 CERROR("Error %d reading HELLO from %pIS\n",
652                        rc, &conn->ksnc_peeraddr);
653                 LASSERT(rc < 0 && rc != -EALREADY);
654                 return rc;
655         }
656
657         if (conn->ksnc_flip) {
658                 __swab32s(&hello->kshm_src_pid);
659                 __swab64s(&hello->kshm_src_nid);
660                 __swab32s(&hello->kshm_dst_pid);
661                 __swab64s(&hello->kshm_dst_nid);
662                 __swab64s(&hello->kshm_src_incarnation);
663                 __swab64s(&hello->kshm_dst_incarnation);
664                 __swab32s(&hello->kshm_ctype);
665                 __swab32s(&hello->kshm_nips);
666         }
667
668         if (hello->kshm_nips > LNET_INTERFACES_NUM) {
669                 CERROR("Bad nips %d from ip %pIS\n",
670                        hello->kshm_nips, &conn->ksnc_peeraddr);
671                 return -EPROTO;
672         }
673
674         if (hello->kshm_nips == 0)
675                 return 0;
676
677         rc = lnet_sock_read(sock, hello->kshm_ips,
678                             hello->kshm_nips * sizeof(__u32), timeout);
679         if (rc != 0) {
680                 CERROR("Error %d reading IPs from ip %pIS\n",
681                        rc, &conn->ksnc_peeraddr);
682                 LASSERT(rc < 0 && rc != -EALREADY);
683                 return rc;
684         }
685
686         for (i = 0; i < (int) hello->kshm_nips; i++) {
687                 if (conn->ksnc_flip)
688                         __swab32s(&hello->kshm_ips[i]);
689
690                 if (hello->kshm_ips[i] == 0) {
691                         CERROR("Zero IP[%d] from ip %pIS\n",
692                                i, &conn->ksnc_peeraddr);
693                         return -EPROTO;
694                 }
695         }
696
697         return 0;
698 }
699
700 static void
701 ksocknal_pack_msg_v1(struct ksock_tx *tx)
702 {
703         /* V1.x has no KSOCK_MSG_NOOP */
704         LASSERT(tx->tx_msg.ksm_type != KSOCK_MSG_NOOP);
705         LASSERT(tx->tx_lnetmsg != NULL);
706
707         tx->tx_hdr.iov_base = (void *)&tx->tx_lnetmsg->msg_hdr;
708         tx->tx_hdr.iov_len  = sizeof(struct lnet_hdr);
709
710         tx->tx_nob = tx->tx_lnetmsg->msg_len + sizeof(struct lnet_hdr);
711         tx->tx_resid = tx->tx_nob;
712 }
713
714 static void
715 ksocknal_pack_msg_v2(struct ksock_tx *tx)
716 {
717         tx->tx_hdr.iov_base = (void *)&tx->tx_msg;
718
719         if (tx->tx_lnetmsg != NULL) {
720                 LASSERT(tx->tx_msg.ksm_type != KSOCK_MSG_NOOP);
721
722                 tx->tx_msg.ksm_u.lnetmsg.ksnm_hdr = tx->tx_lnetmsg->msg_hdr;
723                 tx->tx_hdr.iov_len = sizeof(struct ksock_msg);
724                 tx->tx_resid = tx->tx_nob = sizeof(struct ksock_msg) + tx->tx_lnetmsg->msg_len;
725         } else {
726                 LASSERT(tx->tx_msg.ksm_type == KSOCK_MSG_NOOP);
727
728                 tx->tx_hdr.iov_len = offsetof(struct ksock_msg,
729                                               ksm_u.lnetmsg.ksnm_hdr);
730                 tx->tx_resid = tx->tx_nob = offsetof(struct ksock_msg,  ksm_u.lnetmsg.ksnm_hdr);
731         }
732         /* Don't checksum before start sending, because packet can be piggybacked with ACK */
733 }
734
735 static void
736 ksocknal_unpack_msg_v1(struct ksock_msg *msg)
737 {
738         msg->ksm_csum           = 0;
739         msg->ksm_type           = KSOCK_MSG_LNET;
740         msg->ksm_zc_cookies[0]  = msg->ksm_zc_cookies[1]  = 0;
741 }
742
743 static void
744 ksocknal_unpack_msg_v2(struct ksock_msg *msg)
745 {
746         return;  /* Do nothing */
747 }
748
749 const struct ksock_proto ksocknal_protocol_v1x =
750 {
751         .pro_version            = KSOCK_PROTO_V1,
752         .pro_send_hello         = ksocknal_send_hello_v1,
753         .pro_recv_hello         = ksocknal_recv_hello_v1,
754         .pro_pack               = ksocknal_pack_msg_v1,
755         .pro_unpack             = ksocknal_unpack_msg_v1,
756         .pro_queue_tx_msg       = ksocknal_queue_tx_msg_v1,
757         .pro_handle_zcreq       = NULL,
758         .pro_handle_zcack       = NULL,
759         .pro_queue_tx_zcack     = NULL,
760         .pro_match_tx           = ksocknal_match_tx
761 };
762
763 const struct ksock_proto ksocknal_protocol_v2x =
764 {
765         .pro_version            = KSOCK_PROTO_V2,
766         .pro_send_hello         = ksocknal_send_hello_v2,
767         .pro_recv_hello         = ksocknal_recv_hello_v2,
768         .pro_pack               = ksocknal_pack_msg_v2,
769         .pro_unpack             = ksocknal_unpack_msg_v2,
770         .pro_queue_tx_msg       = ksocknal_queue_tx_msg_v2,
771         .pro_queue_tx_zcack     = ksocknal_queue_tx_zcack_v2,
772         .pro_handle_zcreq       = ksocknal_handle_zcreq,
773         .pro_handle_zcack       = ksocknal_handle_zcack,
774         .pro_match_tx           = ksocknal_match_tx
775 };
776
777 const struct ksock_proto ksocknal_protocol_v3x =
778 {
779         .pro_version            = KSOCK_PROTO_V3,
780         .pro_send_hello         = ksocknal_send_hello_v2,
781         .pro_recv_hello         = ksocknal_recv_hello_v2,
782         .pro_pack               = ksocknal_pack_msg_v2,
783         .pro_unpack             = ksocknal_unpack_msg_v2,
784         .pro_queue_tx_msg       = ksocknal_queue_tx_msg_v2,
785         .pro_queue_tx_zcack     = ksocknal_queue_tx_zcack_v3,
786         .pro_handle_zcreq       = ksocknal_handle_zcreq,
787         .pro_handle_zcack       = ksocknal_handle_zcack,
788         .pro_match_tx           = ksocknal_match_tx_v3
789 };
790