Whamcloud - gitweb
b9e2bc0446afe787ecc06ffe77adcfad088a5529
[fs/lustre-release.git] / lnet / klnds / ralnd / ralnd_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) 2004, 2010, Oracle and/or its affiliates. All rights reserved.
28  * Use is subject to license terms.
29  *
30  * Copyright (c) 2012, Intel Corporation.
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/klnds/ralnd/ralnd_cb.c
37  *
38  * Author: Eric Barton <eric@bartonsoftware.com>
39  */
40
41 #include "ralnd.h"
42
43 void
44 kranal_device_callback(RAP_INT32 devid, RAP_PVOID arg)
45 {
46         kra_device_t *dev;
47         int           i;
48         unsigned long flags;
49
50         CDEBUG(D_NET, "callback for device %d\n", devid);
51
52         for (i = 0; i < kranal_data.kra_ndevs; i++) {
53
54                 dev = &kranal_data.kra_devices[i];
55                 if (dev->rad_id != devid)
56                         continue;
57
58                 spin_lock_irqsave(&dev->rad_lock, flags);
59
60                 if (!dev->rad_ready) {
61                         dev->rad_ready = 1;
62                         wake_up(&dev->rad_waitq);
63                 }
64
65                 spin_unlock_irqrestore(&dev->rad_lock, flags);
66                 return;
67         }
68
69         CWARN("callback for unknown device %d\n", devid);
70 }
71
72 void
73 kranal_schedule_conn(kra_conn_t *conn)
74 {
75         kra_device_t    *dev = conn->rac_device;
76         unsigned long    flags;
77
78         spin_lock_irqsave(&dev->rad_lock, flags);
79
80         if (!conn->rac_scheduled) {
81                 kranal_conn_addref(conn);       /* +1 ref for scheduler */
82                 conn->rac_scheduled = 1;
83                 cfs_list_add_tail(&conn->rac_schedlist, &dev->rad_ready_conns);
84                 wake_up(&dev->rad_waitq);
85         }
86
87         spin_unlock_irqrestore(&dev->rad_lock, flags);
88 }
89
90 kra_tx_t *
91 kranal_get_idle_tx (void)
92 {
93         unsigned long  flags;
94         kra_tx_t      *tx;
95
96         spin_lock_irqsave(&kranal_data.kra_tx_lock, flags);
97
98         if (cfs_list_empty(&kranal_data.kra_idle_txs)) {
99                 spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
100                 return NULL;
101         }
102
103         tx = cfs_list_entry(kranal_data.kra_idle_txs.next, kra_tx_t, tx_list);
104         cfs_list_del(&tx->tx_list);
105
106         /* Allocate a new completion cookie.  It might not be needed, but we've
107          * got a lock right now... */
108         tx->tx_cookie = kranal_data.kra_next_tx_cookie++;
109
110         spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
111
112         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
113         LASSERT (tx->tx_msg.ram_type == RANAL_MSG_NONE);
114         LASSERT (tx->tx_conn == NULL);
115         LASSERT (tx->tx_lntmsg[0] == NULL);
116         LASSERT (tx->tx_lntmsg[1] == NULL);
117
118         return tx;
119 }
120
121 void
122 kranal_init_msg(kra_msg_t *msg, int type)
123 {
124         msg->ram_magic = RANAL_MSG_MAGIC;
125         msg->ram_version = RANAL_MSG_VERSION;
126         msg->ram_type = type;
127         msg->ram_srcnid = kranal_data.kra_ni->ni_nid;
128         /* ram_connstamp gets set when FMA is sent */
129 }
130
131 kra_tx_t *
132 kranal_new_tx_msg (int type)
133 {
134         kra_tx_t *tx = kranal_get_idle_tx();
135
136         if (tx != NULL)
137                 kranal_init_msg(&tx->tx_msg, type);
138
139         return tx;
140 }
141
142 int
143 kranal_setup_immediate_buffer (kra_tx_t *tx, 
144                                unsigned int niov, struct iovec *iov,
145                                int offset, int nob)
146
147 {
148         /* For now this is almost identical to kranal_setup_virt_buffer, but we
149          * could "flatten" the payload into a single contiguous buffer ready
150          * for sending direct over an FMA if we ever needed to. */
151
152         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
153         LASSERT (nob >= 0);
154
155         if (nob == 0) {
156                 tx->tx_buffer = NULL;
157         } else {
158                 LASSERT (niov > 0);
159
160                 while (offset >= iov->iov_len) {
161                         offset -= iov->iov_len;
162                         niov--;
163                         iov++;
164                         LASSERT (niov > 0);
165                 }
166
167                 if (nob > iov->iov_len - offset) {
168                         CERROR("Can't handle multiple vaddr fragments\n");
169                         return -EMSGSIZE;
170                 }
171
172                 tx->tx_buffer = (void *)(((unsigned long)iov->iov_base) + offset);
173         }
174
175         tx->tx_buftype = RANAL_BUF_IMMEDIATE;
176         tx->tx_nob = nob;
177         return 0;
178 }
179
180 int
181 kranal_setup_virt_buffer (kra_tx_t *tx, 
182                           unsigned int niov, struct iovec *iov,
183                           int offset, int nob)
184
185 {
186         LASSERT (nob > 0);
187         LASSERT (niov > 0);
188         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
189
190         while (offset >= iov->iov_len) {
191                 offset -= iov->iov_len;
192                 niov--;
193                 iov++;
194                 LASSERT (niov > 0);
195         }
196
197         if (nob > iov->iov_len - offset) {
198                 CERROR("Can't handle multiple vaddr fragments\n");
199                 return -EMSGSIZE;
200         }
201
202         tx->tx_buftype = RANAL_BUF_VIRT_UNMAPPED;
203         tx->tx_nob = nob;
204         tx->tx_buffer = (void *)(((unsigned long)iov->iov_base) + offset);
205         return 0;
206 }
207
208 int
209 kranal_setup_phys_buffer (kra_tx_t *tx, int nkiov, lnet_kiov_t *kiov,
210                           int offset, int nob)
211 {
212         RAP_PHYS_REGION *phys = tx->tx_phys;
213         int              resid;
214
215         CDEBUG(D_NET, "niov %d offset %d nob %d\n", nkiov, offset, nob);
216
217         LASSERT (nob > 0);
218         LASSERT (nkiov > 0);
219         LASSERT (tx->tx_buftype == RANAL_BUF_NONE);
220
221         while (offset >= kiov->kiov_len) {
222                 offset -= kiov->kiov_len;
223                 nkiov--;
224                 kiov++;
225                 LASSERT (nkiov > 0);
226         }
227
228         tx->tx_buftype = RANAL_BUF_PHYS_UNMAPPED;
229         tx->tx_nob = nob;
230         tx->tx_buffer = (void *)((unsigned long)(kiov->kiov_offset + offset));
231
232         phys->Address = lnet_page2phys(kiov->kiov_page);
233         phys++;
234
235         resid = nob - (kiov->kiov_len - offset);
236         while (resid > 0) {
237                 kiov++;
238                 nkiov--;
239                 LASSERT (nkiov > 0);
240
241                 if (kiov->kiov_offset != 0 ||
242                     ((resid > PAGE_SIZE) &&
243                      kiov->kiov_len < PAGE_SIZE)) {
244                         /* Can't have gaps */
245                         CERROR("Can't make payload contiguous in I/O VM:"
246                                "page %d, offset %d, len %d \n",
247                                (int)(phys - tx->tx_phys),
248                                kiov->kiov_offset, kiov->kiov_len);
249                         return -EINVAL;
250                 }
251
252                 if ((phys - tx->tx_phys) == LNET_MAX_IOV) {
253                         CERROR ("payload too big (%d)\n", (int)(phys - tx->tx_phys));
254                         return -EMSGSIZE;
255                 }
256
257                 phys->Address = lnet_page2phys(kiov->kiov_page);
258                 phys++;
259
260                 resid -= PAGE_SIZE;
261         }
262
263         tx->tx_phys_npages = phys - tx->tx_phys;
264         return 0;
265 }
266
267 static inline int
268 kranal_setup_rdma_buffer (kra_tx_t *tx, unsigned int niov,
269                           struct iovec *iov, lnet_kiov_t *kiov,
270                           int offset, int nob)
271 {
272         LASSERT ((iov == NULL) != (kiov == NULL));
273
274         if (kiov != NULL)
275                 return kranal_setup_phys_buffer(tx, niov, kiov, offset, nob);
276
277         return kranal_setup_virt_buffer(tx, niov, iov, offset, nob);
278 }
279
280 int
281 kranal_map_buffer (kra_tx_t *tx)
282 {
283         kra_conn_t     *conn = tx->tx_conn;
284         kra_device_t   *dev = conn->rac_device;
285         RAP_RETURN      rrc;
286
287         LASSERT (current == dev->rad_scheduler);
288
289         switch (tx->tx_buftype) {
290         default:
291                 LBUG();
292
293         case RANAL_BUF_NONE:
294         case RANAL_BUF_IMMEDIATE:
295         case RANAL_BUF_PHYS_MAPPED:
296         case RANAL_BUF_VIRT_MAPPED:
297                 return 0;
298
299         case RANAL_BUF_PHYS_UNMAPPED:
300                 rrc = RapkRegisterPhys(dev->rad_handle,
301                                        tx->tx_phys, tx->tx_phys_npages,
302                                        &tx->tx_map_key);
303                 if (rrc != RAP_SUCCESS) {
304                         CERROR ("Can't map %d pages: dev %d "
305                                 "phys %u pp %u, virt %u nob %lu\n",
306                                 tx->tx_phys_npages, dev->rad_id, 
307                                 dev->rad_nphysmap, dev->rad_nppphysmap,
308                                 dev->rad_nvirtmap, dev->rad_nobvirtmap);
309                         return -ENOMEM; /* assume insufficient resources */
310                 }
311
312                 dev->rad_nphysmap++;
313                 dev->rad_nppphysmap += tx->tx_phys_npages;
314
315                 tx->tx_buftype = RANAL_BUF_PHYS_MAPPED;
316                 return 0;
317
318         case RANAL_BUF_VIRT_UNMAPPED:
319                 rrc = RapkRegisterMemory(dev->rad_handle,
320                                          tx->tx_buffer, tx->tx_nob,
321                                          &tx->tx_map_key);
322                 if (rrc != RAP_SUCCESS) {
323                         CERROR ("Can't map %d bytes: dev %d "
324                                 "phys %u pp %u, virt %u nob %lu\n",
325                                 tx->tx_nob, dev->rad_id, 
326                                 dev->rad_nphysmap, dev->rad_nppphysmap,
327                                 dev->rad_nvirtmap, dev->rad_nobvirtmap);
328                         return -ENOMEM; /* assume insufficient resources */
329                 }
330
331                 dev->rad_nvirtmap++;
332                 dev->rad_nobvirtmap += tx->tx_nob;
333
334                 tx->tx_buftype = RANAL_BUF_VIRT_MAPPED;
335                 return 0;
336         }
337 }
338
339 void
340 kranal_unmap_buffer (kra_tx_t *tx)
341 {
342         kra_device_t   *dev;
343         RAP_RETURN      rrc;
344
345         switch (tx->tx_buftype) {
346         default:
347                 LBUG();
348
349         case RANAL_BUF_NONE:
350         case RANAL_BUF_IMMEDIATE:
351         case RANAL_BUF_PHYS_UNMAPPED:
352         case RANAL_BUF_VIRT_UNMAPPED:
353                 break;
354
355         case RANAL_BUF_PHYS_MAPPED:
356                 LASSERT (tx->tx_conn != NULL);
357                 dev = tx->tx_conn->rac_device;
358                 LASSERT (current == dev->rad_scheduler);
359                 rrc = RapkDeregisterMemory(dev->rad_handle, NULL,
360                                            &tx->tx_map_key);
361                 LASSERT (rrc == RAP_SUCCESS);
362
363                 dev->rad_nphysmap--;
364                 dev->rad_nppphysmap -= tx->tx_phys_npages;
365
366                 tx->tx_buftype = RANAL_BUF_PHYS_UNMAPPED;
367                 break;
368
369         case RANAL_BUF_VIRT_MAPPED:
370                 LASSERT (tx->tx_conn != NULL);
371                 dev = tx->tx_conn->rac_device;
372                 LASSERT (current == dev->rad_scheduler);
373                 rrc = RapkDeregisterMemory(dev->rad_handle, tx->tx_buffer,
374                                            &tx->tx_map_key);
375                 LASSERT (rrc == RAP_SUCCESS);
376
377                 dev->rad_nvirtmap--;
378                 dev->rad_nobvirtmap -= tx->tx_nob;
379
380                 tx->tx_buftype = RANAL_BUF_VIRT_UNMAPPED;
381                 break;
382         }
383 }
384
385 void
386 kranal_tx_done (kra_tx_t *tx, int completion)
387 {
388         lnet_msg_t      *lnetmsg[2];
389         unsigned long    flags;
390         int              i;
391
392         LASSERT (!in_interrupt());
393
394         kranal_unmap_buffer(tx);
395
396         lnetmsg[0] = tx->tx_lntmsg[0]; tx->tx_lntmsg[0] = NULL;
397         lnetmsg[1] = tx->tx_lntmsg[1]; tx->tx_lntmsg[1] = NULL;
398
399         tx->tx_buftype = RANAL_BUF_NONE;
400         tx->tx_msg.ram_type = RANAL_MSG_NONE;
401         tx->tx_conn = NULL;
402
403         spin_lock_irqsave(&kranal_data.kra_tx_lock, flags);
404
405         cfs_list_add_tail(&tx->tx_list, &kranal_data.kra_idle_txs);
406
407         spin_unlock_irqrestore(&kranal_data.kra_tx_lock, flags);
408
409         /* finalize AFTER freeing lnet msgs */
410         for (i = 0; i < 2; i++) {
411                 if (lnetmsg[i] == NULL)
412                         continue;
413
414                 lnet_finalize(kranal_data.kra_ni, lnetmsg[i], completion);
415         }
416 }
417
418 kra_conn_t *
419 kranal_find_conn_locked (kra_peer_t *peer)
420 {
421         cfs_list_t *tmp;
422
423         /* just return the first connection */
424         cfs_list_for_each (tmp, &peer->rap_conns) {
425                 return cfs_list_entry(tmp, kra_conn_t, rac_list);
426         }
427
428         return NULL;
429 }
430
431 void
432 kranal_post_fma (kra_conn_t *conn, kra_tx_t *tx)
433 {
434         unsigned long    flags;
435
436         tx->tx_conn = conn;
437
438         spin_lock_irqsave(&conn->rac_lock, flags);
439         cfs_list_add_tail(&tx->tx_list, &conn->rac_fmaq);
440         tx->tx_qtime = jiffies;
441         spin_unlock_irqrestore(&conn->rac_lock, flags);
442
443         kranal_schedule_conn(conn);
444 }
445
446 void
447 kranal_launch_tx (kra_tx_t *tx, lnet_nid_t nid)
448 {
449         unsigned long    flags;
450         kra_peer_t      *peer;
451         kra_conn_t      *conn;
452         int              rc;
453         int              retry;
454         rwlock_t    *g_lock = &kranal_data.kra_global_lock;
455
456         /* If I get here, I've committed to send, so I complete the tx with
457          * failure on any problems */
458
459         LASSERT (tx->tx_conn == NULL);      /* only set when assigned a conn */
460
461         for (retry = 0; ; retry = 1) {
462
463                 read_lock(g_lock);
464
465                 peer = kranal_find_peer_locked(nid);
466                 if (peer != NULL) {
467                         conn = kranal_find_conn_locked(peer);
468                         if (conn != NULL) {
469                                 kranal_post_fma(conn, tx);
470                                 read_unlock(g_lock);
471                                 return;
472                         }
473                 }
474                 
475                 /* Making connections; I'll need a write lock... */
476                 read_unlock(g_lock);
477                 write_lock_irqsave(g_lock, flags);
478
479                 peer = kranal_find_peer_locked(nid);
480                 if (peer != NULL)
481                         break;
482                 
483                 write_unlock_irqrestore(g_lock, flags);
484                 
485                 if (retry) {
486                         CERROR("Can't find peer %s\n", libcfs_nid2str(nid));
487                         kranal_tx_done(tx, -EHOSTUNREACH);
488                         return;
489                 }
490
491                 rc = kranal_add_persistent_peer(nid, LNET_NIDADDR(nid),
492                                                 lnet_acceptor_port());
493                 if (rc != 0) {
494                         CERROR("Can't add peer %s: %d\n",
495                                libcfs_nid2str(nid), rc);
496                         kranal_tx_done(tx, rc);
497                         return;
498                 }
499         }
500         
501         conn = kranal_find_conn_locked(peer);
502         if (conn != NULL) {
503                 /* Connection exists; queue message on it */
504                 kranal_post_fma(conn, tx);
505                 write_unlock_irqrestore(g_lock, flags);
506                 return;
507         }
508                         
509         LASSERT (peer->rap_persistence > 0);
510
511         if (!peer->rap_connecting) {
512                 LASSERT (cfs_list_empty(&peer->rap_tx_queue));
513
514                 if (!(peer->rap_reconnect_interval == 0 || /* first attempt */
515                       cfs_time_aftereq(jiffies, peer->rap_reconnect_time))) {
516                         write_unlock_irqrestore(g_lock, flags);
517                         kranal_tx_done(tx, -EHOSTUNREACH);
518                         return;
519                 }
520
521                 peer->rap_connecting = 1;
522                 kranal_peer_addref(peer); /* extra ref for connd */
523
524                 spin_lock(&kranal_data.kra_connd_lock);
525
526                 cfs_list_add_tail(&peer->rap_connd_list,
527                               &kranal_data.kra_connd_peers);
528                 wake_up(&kranal_data.kra_connd_waitq);
529
530                 spin_unlock(&kranal_data.kra_connd_lock);
531         }
532
533         /* A connection is being established; queue the message... */
534         cfs_list_add_tail(&tx->tx_list, &peer->rap_tx_queue);
535
536         write_unlock_irqrestore(g_lock, flags);
537 }
538
539 void
540 kranal_rdma(kra_tx_t *tx, int type,
541             kra_rdma_desc_t *sink, int nob, __u64 cookie)
542 {
543         kra_conn_t   *conn = tx->tx_conn;
544         RAP_RETURN    rrc;
545         unsigned long flags;
546
547         LASSERT (kranal_tx_mapped(tx));
548         LASSERT (nob <= sink->rard_nob);
549         LASSERT (nob <= tx->tx_nob);
550
551         /* No actual race with scheduler sending CLOSE (I'm she!) */
552         LASSERT (current == conn->rac_device->rad_scheduler);
553
554         memset(&tx->tx_rdma_desc, 0, sizeof(tx->tx_rdma_desc));
555         tx->tx_rdma_desc.SrcPtr.AddressBits = (__u64)((unsigned long)tx->tx_buffer);
556         tx->tx_rdma_desc.SrcKey = tx->tx_map_key;
557         tx->tx_rdma_desc.DstPtr = sink->rard_addr;
558         tx->tx_rdma_desc.DstKey = sink->rard_key;
559         tx->tx_rdma_desc.Length = nob;
560         tx->tx_rdma_desc.AppPtr = tx;
561
562         /* prep final completion message */
563         kranal_init_msg(&tx->tx_msg, type);
564         tx->tx_msg.ram_u.completion.racm_cookie = cookie;
565
566         if (nob == 0) { /* Immediate completion */
567                 kranal_post_fma(conn, tx);
568                 return;
569         }
570
571         LASSERT (!conn->rac_close_sent); /* Don't lie (CLOSE == RDMA idle) */
572
573         rrc = RapkPostRdma(conn->rac_rihandle, &tx->tx_rdma_desc);
574         LASSERT (rrc == RAP_SUCCESS);
575
576         spin_lock_irqsave(&conn->rac_lock, flags);
577         cfs_list_add_tail(&tx->tx_list, &conn->rac_rdmaq);
578         tx->tx_qtime = jiffies;
579         spin_unlock_irqrestore(&conn->rac_lock, flags);
580 }
581
582 int
583 kranal_consume_rxmsg (kra_conn_t *conn, void *buffer, int nob)
584 {
585         __u32      nob_received = nob;
586         RAP_RETURN rrc;
587
588         LASSERT (conn->rac_rxmsg != NULL);
589         CDEBUG(D_NET, "Consuming %p\n", conn);
590
591         rrc = RapkFmaCopyOut(conn->rac_rihandle, buffer,
592                              &nob_received, sizeof(kra_msg_t));
593         LASSERT (rrc == RAP_SUCCESS);
594
595         conn->rac_rxmsg = NULL;
596
597         if (nob_received < nob) {
598                 CWARN("Incomplete immediate msg from %s: expected %d, got %d\n",
599                       libcfs_nid2str(conn->rac_peer->rap_nid), 
600                       nob, nob_received);
601                 return -EPROTO;
602         }
603
604         return 0;
605 }
606
607 int
608 kranal_send (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg)
609 {
610         lnet_hdr_t       *hdr = &lntmsg->msg_hdr;
611         int               type = lntmsg->msg_type;
612         lnet_process_id_t target = lntmsg->msg_target;
613         int               target_is_router = lntmsg->msg_target_is_router;
614         int               routing = lntmsg->msg_routing;
615         unsigned int      niov = lntmsg->msg_niov;
616         struct iovec     *iov = lntmsg->msg_iov;
617         lnet_kiov_t      *kiov = lntmsg->msg_kiov;
618         unsigned int      offset = lntmsg->msg_offset;
619         unsigned int      nob = lntmsg->msg_len;
620         kra_tx_t         *tx;
621         int               rc;
622
623         /* NB 'private' is different depending on what we're sending.... */
624
625         CDEBUG(D_NET, "sending %d bytes in %d frags to %s\n",
626                nob, niov, libcfs_id2str(target));
627
628         LASSERT (nob == 0 || niov > 0);
629         LASSERT (niov <= LNET_MAX_IOV);
630
631         LASSERT (!in_interrupt());
632         /* payload is either all vaddrs or all pages */
633         LASSERT (!(kiov != NULL && iov != NULL));
634
635         if (routing) {
636                 CERROR ("Can't route\n");
637                 return -EIO;
638         }
639
640         switch(type) {
641         default:
642                 LBUG();
643
644         case LNET_MSG_ACK:
645                 LASSERT (nob == 0);
646                 break;
647
648         case LNET_MSG_GET:
649                 LASSERT (niov == 0);
650                 LASSERT (nob == 0);
651                 /* We have to consider the eventual sink buffer rather than any
652                  * payload passed here (there isn't any, and strictly, looking
653                  * inside lntmsg is a layering violation).  We send a simple
654                  * IMMEDIATE GET if the sink buffer is mapped already and small
655                  * enough for FMA */
656
657                 if (routing || target_is_router)
658                         break;                  /* send IMMEDIATE */
659
660                 if ((lntmsg->msg_md->md_options & LNET_MD_KIOV) == 0 &&
661                     lntmsg->msg_md->md_length <= RANAL_FMA_MAX_DATA &&
662                     lntmsg->msg_md->md_length <= *kranal_tunables.kra_max_immediate)
663                         break;                  /* send IMMEDIATE */
664
665                 tx = kranal_new_tx_msg(RANAL_MSG_GET_REQ);
666                 if (tx == NULL)
667                         return -ENOMEM;
668
669                 if ((lntmsg->msg_md->md_options & LNET_MD_KIOV) == 0)
670                         rc = kranal_setup_virt_buffer(tx, lntmsg->msg_md->md_niov,
671                                                       lntmsg->msg_md->md_iov.iov,
672                                                       0, lntmsg->msg_md->md_length);
673                 else
674                         rc = kranal_setup_phys_buffer(tx, lntmsg->msg_md->md_niov,
675                                                       lntmsg->msg_md->md_iov.kiov,
676                                                       0, lntmsg->msg_md->md_length);
677                 if (rc != 0) {
678                         kranal_tx_done(tx, rc);
679                         return -EIO;
680                 }
681
682                 tx->tx_lntmsg[1] = lnet_create_reply_msg(ni, lntmsg);
683                 if (tx->tx_lntmsg[1] == NULL) {
684                         CERROR("Can't create reply for GET to %s\n", 
685                                libcfs_nid2str(target.nid));
686                         kranal_tx_done(tx, rc);
687                         return -EIO;
688                 }
689
690                 tx->tx_lntmsg[0] = lntmsg;
691                 tx->tx_msg.ram_u.get.ragm_hdr = *hdr;
692                 /* rest of tx_msg is setup just before it is sent */
693                 kranal_launch_tx(tx, target.nid);
694                 return 0;
695
696         case LNET_MSG_REPLY:
697         case LNET_MSG_PUT:
698                 if (kiov == NULL &&             /* not paged */
699                     nob <= RANAL_FMA_MAX_DATA && /* small enough */
700                     nob <= *kranal_tunables.kra_max_immediate)
701                         break;                  /* send IMMEDIATE */
702
703                 tx = kranal_new_tx_msg(RANAL_MSG_PUT_REQ);
704                 if (tx == NULL)
705                         return -ENOMEM;
706
707                 rc = kranal_setup_rdma_buffer(tx, niov, iov, kiov, offset, nob);
708                 if (rc != 0) {
709                         kranal_tx_done(tx, rc);
710                         return -EIO;
711                 }
712
713                 tx->tx_lntmsg[0] = lntmsg;
714                 tx->tx_msg.ram_u.putreq.raprm_hdr = *hdr;
715                 /* rest of tx_msg is setup just before it is sent */
716                 kranal_launch_tx(tx, target.nid);
717                 return 0;
718         }
719
720         /* send IMMEDIATE */
721
722         LASSERT (kiov == NULL);
723         LASSERT (nob <= RANAL_FMA_MAX_DATA);
724
725         tx = kranal_new_tx_msg(RANAL_MSG_IMMEDIATE);
726         if (tx == NULL)
727                 return -ENOMEM;
728
729         rc = kranal_setup_immediate_buffer(tx, niov, iov, offset, nob);
730         if (rc != 0) {
731                 kranal_tx_done(tx, rc);
732                 return -EIO;
733         }
734
735         tx->tx_msg.ram_u.immediate.raim_hdr = *hdr;
736         tx->tx_lntmsg[0] = lntmsg;
737         kranal_launch_tx(tx, target.nid);
738         return 0;
739 }
740
741 void
742 kranal_reply(lnet_ni_t *ni, kra_conn_t *conn, lnet_msg_t *lntmsg)
743 {
744         kra_msg_t     *rxmsg = conn->rac_rxmsg;
745         unsigned int   niov = lntmsg->msg_niov;
746         struct iovec  *iov = lntmsg->msg_iov;
747         lnet_kiov_t   *kiov = lntmsg->msg_kiov;
748         unsigned int   offset = lntmsg->msg_offset;
749         unsigned int   nob = lntmsg->msg_len;
750         kra_tx_t      *tx;
751         int            rc;
752
753         tx = kranal_get_idle_tx();
754         if (tx == NULL)
755                 goto failed_0;
756
757         rc = kranal_setup_rdma_buffer(tx, niov, iov, kiov, offset, nob);
758         if (rc != 0)
759                 goto failed_1;
760
761         tx->tx_conn = conn;
762
763         rc = kranal_map_buffer(tx);
764         if (rc != 0)
765                 goto failed_1;
766
767         tx->tx_lntmsg[0] = lntmsg;
768
769         kranal_rdma(tx, RANAL_MSG_GET_DONE,
770                     &rxmsg->ram_u.get.ragm_desc, nob,
771                     rxmsg->ram_u.get.ragm_cookie);
772         return;
773
774  failed_1:
775         kranal_tx_done(tx, -EIO);
776  failed_0:
777         lnet_finalize(ni, lntmsg, -EIO);
778 }
779
780 int
781 kranal_eager_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg,
782                    void **new_private)
783 {
784         kra_conn_t *conn = (kra_conn_t *)private;
785
786         LCONSOLE_ERROR_MSG(0x12b, "Dropping message from %s: no buffers free.\n",
787                            libcfs_nid2str(conn->rac_peer->rap_nid));
788
789         return -EDEADLK;
790 }
791
792 int
793 kranal_recv (lnet_ni_t *ni, void *private, lnet_msg_t *lntmsg,
794              int delayed, unsigned int niov, 
795              struct iovec *iov, lnet_kiov_t *kiov,
796              unsigned int offset, unsigned int mlen, unsigned int rlen)
797 {
798         kra_conn_t  *conn = private;
799         kra_msg_t   *rxmsg = conn->rac_rxmsg;
800         kra_tx_t    *tx;
801         void        *buffer;
802         int          rc;
803
804         LASSERT (mlen <= rlen);
805         LASSERT (!in_interrupt());
806         /* Either all pages or all vaddrs */
807         LASSERT (!(kiov != NULL && iov != NULL));
808
809         CDEBUG(D_NET, "conn %p, rxmsg %p, lntmsg %p\n", conn, rxmsg, lntmsg);
810
811         switch(rxmsg->ram_type) {
812         default:
813                 LBUG();
814
815         case RANAL_MSG_IMMEDIATE:
816                 if (mlen == 0) {
817                         buffer = NULL;
818                 } else if (kiov != NULL) {
819                         CERROR("Can't recv immediate into paged buffer\n");
820                         return -EIO;
821                 } else {
822                         LASSERT (niov > 0);
823                         while (offset >= iov->iov_len) {
824                                 offset -= iov->iov_len;
825                                 iov++;
826                                 niov--;
827                                 LASSERT (niov > 0);
828                         }
829                         if (mlen > iov->iov_len - offset) {
830                                 CERROR("Can't handle immediate frags\n");
831                                 return -EIO;
832                         }
833                         buffer = ((char *)iov->iov_base) + offset;
834                 }
835                 rc = kranal_consume_rxmsg(conn, buffer, mlen);
836                 lnet_finalize(ni, lntmsg, (rc == 0) ? 0 : -EIO);
837                 return 0;
838
839         case RANAL_MSG_PUT_REQ:
840                 tx = kranal_new_tx_msg(RANAL_MSG_PUT_ACK);
841                 if (tx == NULL) {
842                         kranal_consume_rxmsg(conn, NULL, 0);
843                         return -ENOMEM;
844                 }
845                 
846                 rc = kranal_setup_rdma_buffer(tx, niov, iov, kiov, offset, mlen);
847                 if (rc != 0) {
848                         kranal_tx_done(tx, rc);
849                         kranal_consume_rxmsg(conn, NULL, 0);
850                         return -EIO;
851                 }
852
853                 tx->tx_conn = conn;
854                 rc = kranal_map_buffer(tx);
855                 if (rc != 0) {
856                         kranal_tx_done(tx, rc);
857                         kranal_consume_rxmsg(conn, NULL, 0);
858                         return -EIO;
859                 }
860
861                 tx->tx_msg.ram_u.putack.rapam_src_cookie =
862                         conn->rac_rxmsg->ram_u.putreq.raprm_cookie;
863                 tx->tx_msg.ram_u.putack.rapam_dst_cookie = tx->tx_cookie;
864                 tx->tx_msg.ram_u.putack.rapam_desc.rard_key = tx->tx_map_key;
865                 tx->tx_msg.ram_u.putack.rapam_desc.rard_addr.AddressBits =
866                         (__u64)((unsigned long)tx->tx_buffer);
867                 tx->tx_msg.ram_u.putack.rapam_desc.rard_nob = mlen;
868
869                 tx->tx_lntmsg[0] = lntmsg; /* finalize this on RDMA_DONE */
870
871                 kranal_post_fma(conn, tx);
872                 kranal_consume_rxmsg(conn, NULL, 0);
873                 return 0;
874
875         case RANAL_MSG_GET_REQ:
876                 if (lntmsg != NULL) {
877                         /* Matched! */
878                         kranal_reply(ni, conn, lntmsg);
879                 } else {
880                         /* No match */
881                         tx = kranal_new_tx_msg(RANAL_MSG_GET_NAK);
882                         if (tx != NULL) {
883                                 tx->tx_msg.ram_u.completion.racm_cookie =
884                                         rxmsg->ram_u.get.ragm_cookie;
885                                 kranal_post_fma(conn, tx);
886                         }
887                 }
888                 kranal_consume_rxmsg(conn, NULL, 0);
889                 return 0;
890         }
891 }
892
893 int
894 kranal_thread_start(int(*fn)(void *arg), void *arg, char *name)
895 {
896         struct task_struct *task = cfs_thread_run(fn, arg, name);
897
898         if (!IS_ERR(task))
899                 atomic_inc(&kranal_data.kra_nthreads);
900         return PTR_ERR(task);
901 }
902
903 void
904 kranal_thread_fini (void)
905 {
906         atomic_dec(&kranal_data.kra_nthreads);
907 }
908
909 int
910 kranal_check_conn_timeouts (kra_conn_t *conn)
911 {
912         kra_tx_t          *tx;
913         cfs_list_t        *ttmp;
914         unsigned long      flags;
915         long               timeout;
916         unsigned long      now = jiffies;
917
918         LASSERT (conn->rac_state == RANAL_CONN_ESTABLISHED ||
919                  conn->rac_state == RANAL_CONN_CLOSING);
920
921         if (!conn->rac_close_sent &&
922             cfs_time_aftereq(now, conn->rac_last_tx +
923                              msecs_to_jiffies(conn->rac_keepalive *
924                                               MSEC_PER_SEC))) {
925                 /* not sent in a while; schedule conn so scheduler sends a keepalive */
926                 CDEBUG(D_NET, "Scheduling keepalive %p->%s\n",
927                        conn, libcfs_nid2str(conn->rac_peer->rap_nid));
928                 kranal_schedule_conn(conn);
929         }
930
931         timeout = msecs_to_jiffies(conn->rac_timeout * MSEC_PER_SEC);
932
933         if (!conn->rac_close_recvd &&
934             cfs_time_aftereq(now, conn->rac_last_rx + timeout)) {
935                 CERROR("%s received from %s within %lu seconds\n",
936                        (conn->rac_state == RANAL_CONN_ESTABLISHED) ?
937                        "Nothing" : "CLOSE not",
938                        libcfs_nid2str(conn->rac_peer->rap_nid),
939                        jiffies_to_msecs(now - conn->rac_last_rx)/MSEC_PER_SEC);
940                 return -ETIMEDOUT;
941         }
942
943         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
944                 return 0;
945
946         /* Check the conn's queues are moving.  These are "belt+braces" checks,
947          * in case of hardware/software errors that make this conn seem
948          * responsive even though it isn't progressing its message queues. */
949
950         spin_lock_irqsave(&conn->rac_lock, flags);
951
952         cfs_list_for_each (ttmp, &conn->rac_fmaq) {
953                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
954
955                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
956                         spin_unlock_irqrestore(&conn->rac_lock, flags);
957                         CERROR("tx on fmaq for %s blocked %lu seconds\n",
958                                libcfs_nid2str(conn->rac_peer->rap_nid),
959                                jiffies_to_msecs(now-tx->tx_qtime)/MSEC_PER_SEC);
960                         return -ETIMEDOUT;
961                 }
962         }
963
964         cfs_list_for_each (ttmp, &conn->rac_rdmaq) {
965                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
966
967                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
968                         spin_unlock_irqrestore(&conn->rac_lock, flags);
969                         CERROR("tx on rdmaq for %s blocked %lu seconds\n",
970                                libcfs_nid2str(conn->rac_peer->rap_nid),
971                                jiffies_to_msecs(now-tx->tx_qtime)/MSEC_PER_SEC);
972                         return -ETIMEDOUT;
973                 }
974         }
975
976         cfs_list_for_each (ttmp, &conn->rac_replyq) {
977                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
978
979                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
980                         spin_unlock_irqrestore(&conn->rac_lock, flags);
981                         CERROR("tx on replyq for %s blocked %lu seconds\n",
982                                libcfs_nid2str(conn->rac_peer->rap_nid),
983                                jiffies_to_msecs(now-tx->tx_qtime)/MSEC_PER_SEC);
984                         return -ETIMEDOUT;
985                 }
986         }
987
988         spin_unlock_irqrestore(&conn->rac_lock, flags);
989         return 0;
990 }
991
992 void
993 kranal_reaper_check (int idx, unsigned long *min_timeoutp)
994 {
995         cfs_list_t        *conns = &kranal_data.kra_conns[idx];
996         cfs_list_t        *ctmp;
997         kra_conn_t        *conn;
998         unsigned long      flags;
999         int                rc;
1000
1001  again:
1002         /* NB. We expect to check all the conns and not find any problems, so
1003          * we just use a shared lock while we take a look... */
1004         read_lock(&kranal_data.kra_global_lock);
1005
1006         cfs_list_for_each (ctmp, conns) {
1007                 conn = cfs_list_entry(ctmp, kra_conn_t, rac_hashlist);
1008
1009                 if (conn->rac_timeout < *min_timeoutp )
1010                         *min_timeoutp = conn->rac_timeout;
1011                 if (conn->rac_keepalive < *min_timeoutp )
1012                         *min_timeoutp = conn->rac_keepalive;
1013
1014                 rc = kranal_check_conn_timeouts(conn);
1015                 if (rc == 0)
1016                         continue;
1017
1018                 kranal_conn_addref(conn);
1019                 read_unlock(&kranal_data.kra_global_lock);
1020
1021                 CERROR("Conn to %s, cqid %d timed out\n",
1022                        libcfs_nid2str(conn->rac_peer->rap_nid), 
1023                        conn->rac_cqid);
1024
1025                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1026
1027                 switch (conn->rac_state) {
1028                 default:
1029                         LBUG();
1030
1031                 case RANAL_CONN_ESTABLISHED:
1032                         kranal_close_conn_locked(conn, -ETIMEDOUT);
1033                         break;
1034
1035                 case RANAL_CONN_CLOSING:
1036                         kranal_terminate_conn_locked(conn);
1037                         break;
1038                 }
1039
1040                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1041                                             flags);
1042
1043                 kranal_conn_decref(conn);
1044
1045                 /* start again now I've dropped the lock */
1046                 goto again;
1047         }
1048
1049         read_unlock(&kranal_data.kra_global_lock);
1050 }
1051
1052 int
1053 kranal_connd (void *arg)
1054 {
1055         long               id = (long)arg;
1056         wait_queue_t     wait;
1057         unsigned long      flags;
1058         kra_peer_t        *peer;
1059         kra_acceptsock_t  *ras;
1060         int                did_something;
1061
1062         cfs_block_allsigs();
1063
1064         init_waitqueue_entry_current(&wait);
1065
1066         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1067
1068         while (!kranal_data.kra_shutdown) {
1069                 did_something = 0;
1070
1071                 if (!cfs_list_empty(&kranal_data.kra_connd_acceptq)) {
1072                         ras = cfs_list_entry(kranal_data.kra_connd_acceptq.next,
1073                                              kra_acceptsock_t, ras_list);
1074                         cfs_list_del(&ras->ras_list);
1075
1076                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1077                                                    flags);
1078
1079                         CDEBUG(D_NET,"About to handshake someone\n");
1080
1081                         kranal_conn_handshake(ras->ras_sock, NULL);
1082                         kranal_free_acceptsock(ras);
1083
1084                         CDEBUG(D_NET,"Finished handshaking someone\n");
1085
1086                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1087                                               flags);
1088                         did_something = 1;
1089                 }
1090
1091                 if (!cfs_list_empty(&kranal_data.kra_connd_peers)) {
1092                         peer = cfs_list_entry(kranal_data.kra_connd_peers.next,
1093                                               kra_peer_t, rap_connd_list);
1094
1095                         cfs_list_del_init(&peer->rap_connd_list);
1096                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1097                                                    flags);
1098
1099                         kranal_connect(peer);
1100                         kranal_peer_decref(peer);
1101
1102                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1103                                               flags);
1104                         did_something = 1;
1105                 }
1106
1107                 if (did_something)
1108                         continue;
1109
1110                 set_current_state(TASK_INTERRUPTIBLE);
1111                 add_wait_queue_exclusive(&kranal_data.kra_connd_waitq, &wait);
1112
1113                 spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1114
1115                 waitq_wait(&wait, TASK_INTERRUPTIBLE);
1116
1117                 set_current_state(TASK_RUNNING);
1118                 remove_wait_queue(&kranal_data.kra_connd_waitq, &wait);
1119
1120                 spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1121         }
1122
1123         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1124
1125         kranal_thread_fini();
1126         return 0;
1127 }
1128
1129 void
1130 kranal_update_reaper_timeout(long timeout)
1131 {
1132         unsigned long   flags;
1133
1134         LASSERT (timeout > 0);
1135
1136         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1137
1138         if (timeout < kranal_data.kra_new_min_timeout)
1139                 kranal_data.kra_new_min_timeout = timeout;
1140
1141         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1142 }
1143
1144 int
1145 kranal_reaper (void *arg)
1146 {
1147         wait_queue_t     wait;
1148         unsigned long      flags;
1149         long               timeout;
1150         int                i;
1151         int                conn_entries = kranal_data.kra_conn_hash_size;
1152         int                conn_index = 0;
1153         int                base_index = conn_entries - 1;
1154         unsigned long      next_check_time = jiffies;
1155         long               next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1156         long               current_min_timeout = 1;
1157
1158         cfs_block_allsigs();
1159
1160         init_waitqueue_entry_current(&wait);
1161
1162         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1163
1164         while (!kranal_data.kra_shutdown) {
1165                 /* I wake up every 'p' seconds to check for timeouts on some
1166                  * more peers.  I try to check every connection 'n' times
1167                  * within the global minimum of all keepalive and timeout
1168                  * intervals, to ensure I attend to every connection within
1169                  * (n+1)/n times its timeout intervals. */
1170                 const int     p = 1;
1171                 const int     n = 3;
1172                 unsigned long min_timeout;
1173                 int           chunk;
1174
1175                 /* careful with the jiffy wrap... */
1176                 timeout = (long)(next_check_time - jiffies);
1177                 if (timeout > 0) {
1178                         set_current_state(TASK_INTERRUPTIBLE);
1179                         add_wait_queue(&kranal_data.kra_reaper_waitq, &wait);
1180
1181                         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock,
1182                                                    flags);
1183
1184                         waitq_timedwait(&wait, TASK_INTERRUPTIBLE,
1185                                             timeout);
1186
1187                         spin_lock_irqsave(&kranal_data.kra_reaper_lock,
1188                                               flags);
1189
1190                         set_current_state(TASK_RUNNING);
1191                         remove_wait_queue(&kranal_data.kra_reaper_waitq, &wait);
1192                         continue;
1193                 }
1194
1195                 if (kranal_data.kra_new_min_timeout !=
1196                     MAX_SCHEDULE_TIMEOUT) {
1197                         /* new min timeout set: restart min timeout scan */
1198                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1199                         base_index = conn_index - 1;
1200                         if (base_index < 0)
1201                                 base_index = conn_entries - 1;
1202
1203                         if (kranal_data.kra_new_min_timeout <
1204                             current_min_timeout) {
1205                                 current_min_timeout =
1206                                         kranal_data.kra_new_min_timeout;
1207                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1208                                        current_min_timeout);
1209                         }
1210
1211                         kranal_data.kra_new_min_timeout =
1212                                 MAX_SCHEDULE_TIMEOUT;
1213                 }
1214                 min_timeout = current_min_timeout;
1215
1216                 spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1217
1218                 LASSERT (min_timeout > 0);
1219
1220                 /* Compute how many table entries to check now so I get round
1221                  * the whole table fast enough given that I do this at fixed
1222                  * intervals of 'p' seconds) */
1223                 chunk = conn_entries;
1224                 if (min_timeout > n * p)
1225                         chunk = (chunk * n * p) / min_timeout;
1226                 if (chunk == 0)
1227                         chunk = 1;
1228
1229                 for (i = 0; i < chunk; i++) {
1230                         kranal_reaper_check(conn_index,
1231                                             &next_min_timeout);
1232                         conn_index = (conn_index + 1) % conn_entries;
1233                 }
1234
1235                 next_check_time += msecs_to_jiffies(p * MSEC_PER_SEC);
1236
1237                 spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1238
1239                 if (((conn_index - chunk <= base_index &&
1240                       base_index < conn_index) ||
1241                      (conn_index - conn_entries - chunk <= base_index &&
1242                       base_index < conn_index - conn_entries))) {
1243
1244                         /* Scanned all conns: set current_min_timeout... */
1245                         if (current_min_timeout != next_min_timeout) {
1246                                 current_min_timeout = next_min_timeout;
1247                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1248                                        current_min_timeout);
1249                         }
1250
1251                         /* ...and restart min timeout scan */
1252                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1253                         base_index = conn_index - 1;
1254                         if (base_index < 0)
1255                                 base_index = conn_entries - 1;
1256                 }
1257         }
1258
1259         kranal_thread_fini();
1260         return 0;
1261 }
1262
1263 void
1264 kranal_check_rdma_cq (kra_device_t *dev)
1265 {
1266         kra_conn_t          *conn;
1267         kra_tx_t            *tx;
1268         RAP_RETURN           rrc;
1269         unsigned long        flags;
1270         RAP_RDMA_DESCRIPTOR *desc;
1271         __u32                cqid;
1272         __u32                event_type;
1273
1274         for (;;) {
1275                 rrc = RapkCQDone(dev->rad_rdma_cqh, &cqid, &event_type);
1276                 if (rrc == RAP_NOT_DONE) {
1277                         CDEBUG(D_NET, "RDMA CQ %d empty\n", dev->rad_id);
1278                         return;
1279                 }
1280
1281                 LASSERT (rrc == RAP_SUCCESS);
1282                 LASSERT ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0);
1283
1284                 read_lock(&kranal_data.kra_global_lock);
1285
1286                 conn = kranal_cqid2conn_locked(cqid);
1287                 if (conn == NULL) {
1288                         /* Conn was destroyed? */
1289                         CDEBUG(D_NET, "RDMA CQID lookup %d failed\n", cqid);
1290                         read_unlock(&kranal_data.kra_global_lock);
1291                         continue;
1292                 }
1293
1294                 rrc = RapkRdmaDone(conn->rac_rihandle, &desc);
1295                 LASSERT (rrc == RAP_SUCCESS);
1296
1297                 CDEBUG(D_NET, "Completed %p\n",
1298                        cfs_list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list));
1299
1300                 spin_lock_irqsave(&conn->rac_lock, flags);
1301
1302                 LASSERT (!cfs_list_empty(&conn->rac_rdmaq));
1303                 tx = cfs_list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list);
1304                 cfs_list_del(&tx->tx_list);
1305
1306                 LASSERT(desc->AppPtr == (void *)tx);
1307                 LASSERT(tx->tx_msg.ram_type == RANAL_MSG_PUT_DONE ||
1308                         tx->tx_msg.ram_type == RANAL_MSG_GET_DONE);
1309
1310                 cfs_list_add_tail(&tx->tx_list, &conn->rac_fmaq);
1311                 tx->tx_qtime = jiffies;
1312
1313                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1314
1315                 /* Get conn's fmaq processed, now I've just put something
1316                  * there */
1317                 kranal_schedule_conn(conn);
1318
1319                 read_unlock(&kranal_data.kra_global_lock);
1320         }
1321 }
1322
1323 void
1324 kranal_check_fma_cq (kra_device_t *dev)
1325 {
1326         kra_conn_t         *conn;
1327         RAP_RETURN          rrc;
1328         __u32               cqid;
1329         __u32               event_type;
1330         cfs_list_t         *conns;
1331         cfs_list_t         *tmp;
1332         int                 i;
1333
1334         for (;;) {
1335                 rrc = RapkCQDone(dev->rad_fma_cqh, &cqid, &event_type);
1336                 if (rrc == RAP_NOT_DONE) {
1337                         CDEBUG(D_NET, "FMA CQ %d empty\n", dev->rad_id);
1338                         return;
1339                 }
1340
1341                 LASSERT (rrc == RAP_SUCCESS);
1342
1343                 if ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0) {
1344
1345                         read_lock(&kranal_data.kra_global_lock);
1346
1347                         conn = kranal_cqid2conn_locked(cqid);
1348                         if (conn == NULL) {
1349                                 CDEBUG(D_NET, "FMA CQID lookup %d failed\n",
1350                                        cqid);
1351                         } else {
1352                                 CDEBUG(D_NET, "FMA completed: %p CQID %d\n",
1353                                        conn, cqid);
1354                                 kranal_schedule_conn(conn);
1355                         }
1356
1357                         read_unlock(&kranal_data.kra_global_lock);
1358                         continue;
1359                 }
1360
1361                 /* FMA CQ has overflowed: check ALL conns */
1362                 CWARN("FMA CQ overflow: scheduling ALL conns on device %d\n", 
1363                       dev->rad_id);
1364
1365                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++) {
1366
1367                         read_lock(&kranal_data.kra_global_lock);
1368
1369                         conns = &kranal_data.kra_conns[i];
1370
1371                         cfs_list_for_each (tmp, conns) {
1372                                 conn = cfs_list_entry(tmp, kra_conn_t,
1373                                                       rac_hashlist);
1374
1375                                 if (conn->rac_device == dev)
1376                                         kranal_schedule_conn(conn);
1377                         }
1378
1379                         /* don't block write lockers for too long... */
1380                         read_unlock(&kranal_data.kra_global_lock);
1381                 }
1382         }
1383 }
1384
1385 int
1386 kranal_sendmsg(kra_conn_t *conn, kra_msg_t *msg,
1387                void *immediate, int immediatenob)
1388 {
1389         int        sync = (msg->ram_type & RANAL_MSG_FENCE) != 0;
1390         RAP_RETURN rrc;
1391
1392         CDEBUG(D_NET,"%p sending msg %p %02x%s [%p for %d]\n",
1393                conn, msg, msg->ram_type, sync ? "(sync)" : "",
1394                immediate, immediatenob);
1395
1396         LASSERT (sizeof(*msg) <= RANAL_FMA_MAX_PREFIX);
1397         LASSERT ((msg->ram_type == RANAL_MSG_IMMEDIATE) ?
1398                  immediatenob <= RANAL_FMA_MAX_DATA :
1399                  immediatenob == 0);
1400
1401         msg->ram_connstamp = conn->rac_my_connstamp;
1402         msg->ram_seq = conn->rac_tx_seq;
1403
1404         if (sync)
1405                 rrc = RapkFmaSyncSend(conn->rac_rihandle,
1406                                       immediate, immediatenob,
1407                                       msg, sizeof(*msg));
1408         else
1409                 rrc = RapkFmaSend(conn->rac_rihandle,
1410                                   immediate, immediatenob,
1411                                   msg, sizeof(*msg));
1412
1413         switch (rrc) {
1414         default:
1415                 LBUG();
1416
1417         case RAP_SUCCESS:
1418                 conn->rac_last_tx = jiffies;
1419                 conn->rac_tx_seq++;
1420                 return 0;
1421
1422         case RAP_NOT_DONE:
1423                 if (cfs_time_aftereq(jiffies,
1424                                      conn->rac_last_tx +
1425                                      msecs_to_jiffies(conn->rac_keepalive *
1426                                                       MSEC_PER_SEC)))
1427                         CWARN("EAGAIN sending %02x (idle %lu secs)\n",
1428                               msg->ram_type,
1429                               jiffies_to_msecs(jiffies - conn->rac_last_tx) /
1430                               MSEC_PER_SEC);
1431                 return -EAGAIN;
1432         }
1433 }
1434
1435 void
1436 kranal_process_fmaq (kra_conn_t *conn)
1437 {
1438         unsigned long flags;
1439         int           more_to_do;
1440         kra_tx_t     *tx;
1441         int           rc;
1442         int           expect_reply;
1443
1444         /* NB 1. kranal_sendmsg() may fail if I'm out of credits right now.
1445          *       However I will be rescheduled by an FMA completion event
1446          *       when I eventually get some.
1447          * NB 2. Sampling rac_state here races with setting it elsewhere.
1448          *       But it doesn't matter if I try to send a "real" message just
1449          *       as I start closing because I'll get scheduled to send the
1450          *       close anyway. */
1451
1452         /* Not racing with incoming message processing! */
1453         LASSERT (current == conn->rac_device->rad_scheduler);
1454
1455         if (conn->rac_state != RANAL_CONN_ESTABLISHED) {
1456                 if (!cfs_list_empty(&conn->rac_rdmaq)) {
1457                         /* RDMAs in progress */
1458                         LASSERT (!conn->rac_close_sent);
1459
1460                         if (cfs_time_aftereq(jiffies,
1461                                              conn->rac_last_tx +
1462                                              msecs_to_jiffies(conn->rac_keepalive *
1463                                                               MSEC_PER_SEC))) {
1464                                 CDEBUG(D_NET, "sending NOOP (rdma in progress)\n");
1465                                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1466                                 kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1467                         }
1468                         return;
1469                 }
1470
1471                 if (conn->rac_close_sent)
1472                         return;
1473
1474                 CWARN("sending CLOSE to %s\n", 
1475                       libcfs_nid2str(conn->rac_peer->rap_nid));
1476                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_CLOSE);
1477                 rc = kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1478                 if (rc != 0)
1479                         return;
1480
1481                 conn->rac_close_sent = 1;
1482                 if (!conn->rac_close_recvd)
1483                         return;
1484
1485                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1486
1487                 if (conn->rac_state == RANAL_CONN_CLOSING)
1488                         kranal_terminate_conn_locked(conn);
1489
1490                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1491                                             flags);
1492                 return;
1493         }
1494
1495         spin_lock_irqsave(&conn->rac_lock, flags);
1496
1497         if (cfs_list_empty(&conn->rac_fmaq)) {
1498
1499                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1500
1501                 if (cfs_time_aftereq(jiffies,
1502                                      conn->rac_last_tx +
1503                                      msecs_to_jiffies(conn->rac_keepalive *
1504                                                       MSEC_PER_SEC))) {
1505                         CDEBUG(D_NET, "sending NOOP -> %s (%p idle %lu(%ld))\n",
1506                                libcfs_nid2str(conn->rac_peer->rap_nid), conn,
1507                                jiffies_to_msecs(jiffies - conn->rac_last_tx) /
1508                                MSEC_PER_SEC,
1509                                conn->rac_keepalive);
1510                         kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1511                         kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1512                 }
1513                 return;
1514         }
1515
1516         tx = cfs_list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1517         cfs_list_del(&tx->tx_list);
1518         more_to_do = !cfs_list_empty(&conn->rac_fmaq);
1519
1520         spin_unlock_irqrestore(&conn->rac_lock, flags);
1521
1522         expect_reply = 0;
1523         CDEBUG(D_NET, "sending regular msg: %p, type %02x, cookie "LPX64"\n",
1524                tx, tx->tx_msg.ram_type, tx->tx_cookie);
1525         switch (tx->tx_msg.ram_type) {
1526         default:
1527                 LBUG();
1528
1529         case RANAL_MSG_IMMEDIATE:
1530                 rc = kranal_sendmsg(conn, &tx->tx_msg,
1531                                     tx->tx_buffer, tx->tx_nob);
1532                 break;
1533
1534         case RANAL_MSG_PUT_NAK:
1535         case RANAL_MSG_PUT_DONE:
1536         case RANAL_MSG_GET_NAK:
1537         case RANAL_MSG_GET_DONE:
1538                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1539                 break;
1540
1541         case RANAL_MSG_PUT_REQ:
1542                 rc = kranal_map_buffer(tx);
1543                 LASSERT (rc != -EAGAIN);
1544                 if (rc != 0)
1545                         break;
1546
1547                 tx->tx_msg.ram_u.putreq.raprm_cookie = tx->tx_cookie;
1548                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1549                 expect_reply = 1;
1550                 break;
1551
1552         case RANAL_MSG_PUT_ACK:
1553                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1554                 expect_reply = 1;
1555                 break;
1556
1557         case RANAL_MSG_GET_REQ:
1558                 rc = kranal_map_buffer(tx);
1559                 LASSERT (rc != -EAGAIN);
1560                 if (rc != 0)
1561                         break;
1562
1563                 tx->tx_msg.ram_u.get.ragm_cookie = tx->tx_cookie;
1564                 tx->tx_msg.ram_u.get.ragm_desc.rard_key = tx->tx_map_key;
1565                 tx->tx_msg.ram_u.get.ragm_desc.rard_addr.AddressBits =
1566                         (__u64)((unsigned long)tx->tx_buffer);
1567                 tx->tx_msg.ram_u.get.ragm_desc.rard_nob = tx->tx_nob;
1568                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1569                 expect_reply = 1;
1570                 break;
1571         }
1572
1573         if (rc == -EAGAIN) {
1574                 /* I need credits to send this.  Replace tx at the head of the
1575                  * fmaq and I'll get rescheduled when credits appear */
1576                 CDEBUG(D_NET, "EAGAIN on %p\n", conn);
1577                 spin_lock_irqsave(&conn->rac_lock, flags);
1578                 cfs_list_add(&tx->tx_list, &conn->rac_fmaq);
1579                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1580                 return;
1581         }
1582
1583         if (!expect_reply || rc != 0) {
1584                 kranal_tx_done(tx, rc);
1585         } else {
1586                 /* LASSERT(current) above ensures this doesn't race with reply
1587                  * processing */
1588                 spin_lock_irqsave(&conn->rac_lock, flags);
1589                 cfs_list_add_tail(&tx->tx_list, &conn->rac_replyq);
1590                 tx->tx_qtime = jiffies;
1591                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1592         }
1593
1594         if (more_to_do) {
1595                 CDEBUG(D_NET, "Rescheduling %p (more to do)\n", conn);
1596                 kranal_schedule_conn(conn);
1597         }
1598 }
1599
1600 static inline void
1601 kranal_swab_rdma_desc (kra_rdma_desc_t *d)
1602 {
1603         __swab64s(&d->rard_key.Key);
1604         __swab16s(&d->rard_key.Cookie);
1605         __swab16s(&d->rard_key.MdHandle);
1606         __swab32s(&d->rard_key.Flags);
1607         __swab64s(&d->rard_addr.AddressBits);
1608         __swab32s(&d->rard_nob);
1609 }
1610
1611 kra_tx_t *
1612 kranal_match_reply(kra_conn_t *conn, int type, __u64 cookie)
1613 {
1614         cfs_list_t       *ttmp;
1615         kra_tx_t         *tx;
1616         unsigned long     flags;
1617
1618         spin_lock_irqsave(&conn->rac_lock, flags);
1619
1620         cfs_list_for_each(ttmp, &conn->rac_replyq) {
1621                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
1622
1623                 CDEBUG(D_NET,"Checking %p %02x/"LPX64"\n",
1624                        tx, tx->tx_msg.ram_type, tx->tx_cookie);
1625
1626                 if (tx->tx_cookie != cookie)
1627                         continue;
1628
1629                 if (tx->tx_msg.ram_type != type) {
1630                         spin_unlock_irqrestore(&conn->rac_lock, flags);
1631                         CWARN("Unexpected type %x (%x expected) "
1632                               "matched reply from %s\n",
1633                               tx->tx_msg.ram_type, type,
1634                               libcfs_nid2str(conn->rac_peer->rap_nid));
1635                         return NULL;
1636                 }
1637
1638                 cfs_list_del(&tx->tx_list);
1639                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1640                 return tx;
1641         }
1642
1643         spin_unlock_irqrestore(&conn->rac_lock, flags);
1644         CWARN("Unmatched reply %02x/"LPX64" from %s\n",
1645               type, cookie, libcfs_nid2str(conn->rac_peer->rap_nid));
1646         return NULL;
1647 }
1648
1649 void
1650 kranal_check_fma_rx (kra_conn_t *conn)
1651 {
1652         unsigned long flags;
1653         __u32         seq;
1654         kra_tx_t     *tx;
1655         kra_msg_t    *msg;
1656         void         *prefix;
1657         RAP_RETURN    rrc = RapkFmaGetPrefix(conn->rac_rihandle, &prefix);
1658         kra_peer_t   *peer = conn->rac_peer;
1659         int           rc = 0;
1660         int           repost = 1;
1661
1662         if (rrc == RAP_NOT_DONE)
1663                 return;
1664
1665         CDEBUG(D_NET, "RX on %p\n", conn);
1666
1667         LASSERT (rrc == RAP_SUCCESS);
1668         conn->rac_last_rx = jiffies;
1669         seq = conn->rac_rx_seq++;
1670         msg = (kra_msg_t *)prefix;
1671
1672         /* stash message for portals callbacks they'll NULL
1673          * rac_rxmsg if they consume it */
1674         LASSERT (conn->rac_rxmsg == NULL);
1675         conn->rac_rxmsg = msg;
1676
1677         if (msg->ram_magic != RANAL_MSG_MAGIC) {
1678                 if (__swab32(msg->ram_magic) != RANAL_MSG_MAGIC) {
1679                         CERROR("Unexpected magic %08x from %s\n",
1680                                msg->ram_magic, libcfs_nid2str(peer->rap_nid));
1681                         rc = -EPROTO;
1682                         goto out;
1683                 }
1684
1685                 __swab32s(&msg->ram_magic);
1686                 __swab16s(&msg->ram_version);
1687                 __swab16s(&msg->ram_type);
1688                 __swab64s(&msg->ram_srcnid);
1689                 __swab64s(&msg->ram_connstamp);
1690                 __swab32s(&msg->ram_seq);
1691
1692                 /* NB message type checked below; NOT here... */
1693                 switch (msg->ram_type) {
1694                 case RANAL_MSG_PUT_ACK:
1695                         kranal_swab_rdma_desc(&msg->ram_u.putack.rapam_desc);
1696                         break;
1697
1698                 case RANAL_MSG_GET_REQ:
1699                         kranal_swab_rdma_desc(&msg->ram_u.get.ragm_desc);
1700                         break;
1701
1702                 default:
1703                         break;
1704                 }
1705         }
1706
1707         if (msg->ram_version != RANAL_MSG_VERSION) {
1708                 CERROR("Unexpected protocol version %d from %s\n",
1709                        msg->ram_version, libcfs_nid2str(peer->rap_nid));
1710                 rc = -EPROTO;
1711                 goto out;
1712         }
1713
1714         if (msg->ram_srcnid != peer->rap_nid) {
1715                 CERROR("Unexpected peer %s from %s\n",
1716                        libcfs_nid2str(msg->ram_srcnid), 
1717                        libcfs_nid2str(peer->rap_nid));
1718                 rc = -EPROTO;
1719                 goto out;
1720         }
1721
1722         if (msg->ram_connstamp != conn->rac_peer_connstamp) {
1723                 CERROR("Unexpected connstamp "LPX64"("LPX64
1724                        " expected) from %s\n",
1725                        msg->ram_connstamp, conn->rac_peer_connstamp,
1726                        libcfs_nid2str(peer->rap_nid));
1727                 rc = -EPROTO;
1728                 goto out;
1729         }
1730
1731         if (msg->ram_seq != seq) {
1732                 CERROR("Unexpected sequence number %d(%d expected) from %s\n",
1733                        msg->ram_seq, seq, libcfs_nid2str(peer->rap_nid));
1734                 rc = -EPROTO;
1735                 goto out;
1736         }
1737
1738         if ((msg->ram_type & RANAL_MSG_FENCE) != 0) {
1739                 /* This message signals RDMA completion... */
1740                 rrc = RapkFmaSyncWait(conn->rac_rihandle);
1741                 if (rrc != RAP_SUCCESS) {
1742                         CERROR("RapkFmaSyncWait failed: %d\n", rrc);
1743                         rc = -ENETDOWN;
1744                         goto out;
1745                 }
1746         }
1747
1748         if (conn->rac_close_recvd) {
1749                 CERROR("Unexpected message %d after CLOSE from %s\n",
1750                        msg->ram_type, libcfs_nid2str(conn->rac_peer->rap_nid));
1751                 rc = -EPROTO;
1752                 goto out;
1753         }
1754
1755         if (msg->ram_type == RANAL_MSG_CLOSE) {
1756                 CWARN("RX CLOSE from %s\n", libcfs_nid2str(conn->rac_peer->rap_nid));
1757                 conn->rac_close_recvd = 1;
1758                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1759
1760                 if (conn->rac_state == RANAL_CONN_ESTABLISHED)
1761                         kranal_close_conn_locked(conn, 0);
1762                 else if (conn->rac_state == RANAL_CONN_CLOSING &&
1763                          conn->rac_close_sent)
1764                         kranal_terminate_conn_locked(conn);
1765
1766                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1767                                             flags);
1768                 goto out;
1769         }
1770
1771         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
1772                 goto out;
1773
1774         switch (msg->ram_type) {
1775         case RANAL_MSG_NOOP:
1776                 /* Nothing to do; just a keepalive */
1777                 CDEBUG(D_NET, "RX NOOP on %p\n", conn);
1778                 break;
1779
1780         case RANAL_MSG_IMMEDIATE:
1781                 CDEBUG(D_NET, "RX IMMEDIATE on %p\n", conn);
1782                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.immediate.raim_hdr, 
1783                                 msg->ram_srcnid, conn, 0);
1784                 repost = rc < 0;
1785                 break;
1786
1787         case RANAL_MSG_PUT_REQ:
1788                 CDEBUG(D_NET, "RX PUT_REQ on %p\n", conn);
1789                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.putreq.raprm_hdr, 
1790                                 msg->ram_srcnid, conn, 1);
1791                 repost = rc < 0;
1792                 break;
1793
1794         case RANAL_MSG_PUT_NAK:
1795                 CDEBUG(D_NET, "RX PUT_NAK on %p\n", conn);
1796                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1797                                         msg->ram_u.completion.racm_cookie);
1798                 if (tx == NULL)
1799                         break;
1800
1801                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1802                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1803                 kranal_tx_done(tx, -ENOENT);    /* no match */
1804                 break;
1805
1806         case RANAL_MSG_PUT_ACK:
1807                 CDEBUG(D_NET, "RX PUT_ACK on %p\n", conn);
1808                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1809                                         msg->ram_u.putack.rapam_src_cookie);
1810                 if (tx == NULL)
1811                         break;
1812
1813                 kranal_rdma(tx, RANAL_MSG_PUT_DONE,
1814                             &msg->ram_u.putack.rapam_desc,
1815                             msg->ram_u.putack.rapam_desc.rard_nob,
1816                             msg->ram_u.putack.rapam_dst_cookie);
1817                 break;
1818
1819         case RANAL_MSG_PUT_DONE:
1820                 CDEBUG(D_NET, "RX PUT_DONE on %p\n", conn);
1821                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_ACK,
1822                                         msg->ram_u.completion.racm_cookie);
1823                 if (tx == NULL)
1824                         break;
1825
1826                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1827                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1828                 kranal_tx_done(tx, 0);
1829                 break;
1830
1831         case RANAL_MSG_GET_REQ:
1832                 CDEBUG(D_NET, "RX GET_REQ on %p\n", conn);
1833                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.get.ragm_hdr, 
1834                                 msg->ram_srcnid, conn, 1);
1835                 repost = rc < 0;
1836                 break;
1837
1838         case RANAL_MSG_GET_NAK:
1839                 CDEBUG(D_NET, "RX GET_NAK on %p\n", conn);
1840                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1841                                         msg->ram_u.completion.racm_cookie);
1842                 if (tx == NULL)
1843                         break;
1844
1845                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1846                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1847                 kranal_tx_done(tx, -ENOENT);    /* no match */
1848                 break;
1849
1850         case RANAL_MSG_GET_DONE:
1851                 CDEBUG(D_NET, "RX GET_DONE on %p\n", conn);
1852                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1853                                         msg->ram_u.completion.racm_cookie);
1854                 if (tx == NULL)
1855                         break;
1856
1857                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1858                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1859 #if 0
1860                 /* completion message should send rdma length if we ever allow
1861                  * GET truncation */
1862                 lnet_set_reply_msg_len(kranal_data.kra_ni, tx->tx_lntmsg[1], ???);
1863 #endif
1864                 kranal_tx_done(tx, 0);
1865                 break;
1866         }
1867
1868  out:
1869         if (rc < 0)                             /* protocol/comms error */
1870                 kranal_close_conn (conn, rc);
1871
1872         if (repost && conn->rac_rxmsg != NULL)
1873                 kranal_consume_rxmsg(conn, NULL, 0);
1874
1875         /* check again later */
1876         kranal_schedule_conn(conn);
1877 }
1878
1879 void
1880 kranal_complete_closed_conn (kra_conn_t *conn)
1881 {
1882         kra_tx_t   *tx;
1883         int         nfma;
1884         int         nreplies;
1885
1886         LASSERT (conn->rac_state == RANAL_CONN_CLOSED);
1887         LASSERT (cfs_list_empty(&conn->rac_list));
1888         LASSERT (cfs_list_empty(&conn->rac_hashlist));
1889
1890         for (nfma = 0; !cfs_list_empty(&conn->rac_fmaq); nfma++) {
1891                 tx = cfs_list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1892
1893                 cfs_list_del(&tx->tx_list);
1894                 kranal_tx_done(tx, -ECONNABORTED);
1895         }
1896
1897         LASSERT (cfs_list_empty(&conn->rac_rdmaq));
1898
1899         for (nreplies = 0; !cfs_list_empty(&conn->rac_replyq); nreplies++) {
1900                 tx = cfs_list_entry(conn->rac_replyq.next, kra_tx_t, tx_list);
1901
1902                 cfs_list_del(&tx->tx_list);
1903                 kranal_tx_done(tx, -ECONNABORTED);
1904         }
1905
1906         CWARN("Closed conn %p -> %s: nmsg %d nreplies %d\n",
1907                conn, libcfs_nid2str(conn->rac_peer->rap_nid), nfma, nreplies);
1908 }
1909
1910 int kranal_process_new_conn (kra_conn_t *conn)
1911 {
1912         RAP_RETURN   rrc;
1913
1914         rrc = RapkCompleteSync(conn->rac_rihandle, 1);
1915         if (rrc == RAP_SUCCESS)
1916                 return 0;
1917
1918         LASSERT (rrc == RAP_NOT_DONE);
1919         if (!cfs_time_aftereq(jiffies, conn->rac_last_tx +
1920                               msecs_to_jiffies(conn->rac_timeout*MSEC_PER_SEC)))
1921                 return -EAGAIN;
1922
1923         /* Too late */
1924         rrc = RapkCompleteSync(conn->rac_rihandle, 0);
1925         LASSERT (rrc == RAP_SUCCESS);
1926         return -ETIMEDOUT;
1927 }
1928
1929 int
1930 kranal_scheduler (void *arg)
1931 {
1932         kra_device_t     *dev = (kra_device_t *)arg;
1933         wait_queue_t    wait;
1934         kra_conn_t       *conn;
1935         unsigned long     flags;
1936         unsigned long     deadline;
1937         unsigned long     soonest;
1938         int               nsoonest;
1939         long              timeout;
1940         cfs_list_t       *tmp;
1941         cfs_list_t       *nxt;
1942         int               rc;
1943         int               dropped_lock;
1944         int               busy_loops = 0;
1945
1946         cfs_block_allsigs();
1947
1948         dev->rad_scheduler = current;
1949         init_waitqueue_entry_current(&wait);
1950
1951         spin_lock_irqsave(&dev->rad_lock, flags);
1952
1953         while (!kranal_data.kra_shutdown) {
1954                 /* Safe: kra_shutdown only set when quiescent */
1955
1956                 if (busy_loops++ >= RANAL_RESCHED) {
1957                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1958
1959                         cond_resched();
1960                         busy_loops = 0;
1961
1962                         spin_lock_irqsave(&dev->rad_lock, flags);
1963                 }
1964
1965                 dropped_lock = 0;
1966
1967                 if (dev->rad_ready) {
1968                         /* Device callback fired since I last checked it */
1969                         dev->rad_ready = 0;
1970                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1971                         dropped_lock = 1;
1972
1973                         kranal_check_rdma_cq(dev);
1974                         kranal_check_fma_cq(dev);
1975
1976                         spin_lock_irqsave(&dev->rad_lock, flags);
1977                 }
1978
1979                 cfs_list_for_each_safe(tmp, nxt, &dev->rad_ready_conns) {
1980                         conn = cfs_list_entry(tmp, kra_conn_t, rac_schedlist);
1981
1982                         cfs_list_del_init(&conn->rac_schedlist);
1983                         LASSERT (conn->rac_scheduled);
1984                         conn->rac_scheduled = 0;
1985                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1986                         dropped_lock = 1;
1987
1988                         kranal_check_fma_rx(conn);
1989                         kranal_process_fmaq(conn);
1990
1991                         if (conn->rac_state == RANAL_CONN_CLOSED)
1992                                 kranal_complete_closed_conn(conn);
1993
1994                         kranal_conn_decref(conn);
1995                         spin_lock_irqsave(&dev->rad_lock, flags);
1996                 }
1997
1998                 nsoonest = 0;
1999                 soonest = jiffies;
2000
2001                 cfs_list_for_each_safe(tmp, nxt, &dev->rad_new_conns) {
2002                         conn = cfs_list_entry(tmp, kra_conn_t, rac_schedlist);
2003
2004                         deadline = conn->rac_last_tx + conn->rac_keepalive;
2005                         if (cfs_time_aftereq(jiffies, deadline)) {
2006                                 /* Time to process this new conn */
2007                                 spin_unlock_irqrestore(&dev->rad_lock,
2008                                                            flags);
2009                                 dropped_lock = 1;
2010
2011                                 rc = kranal_process_new_conn(conn);
2012                                 if (rc != -EAGAIN) {
2013                                         /* All done with this conn */
2014                                         spin_lock_irqsave(&dev->rad_lock,
2015                                                               flags);
2016                                         cfs_list_del_init(&conn->rac_schedlist);
2017                                         spin_unlock_irqrestore(&dev-> \
2018                                                                    rad_lock,
2019                                                                    flags);
2020
2021                                         kranal_conn_decref(conn);
2022                                         spin_lock_irqsave(&dev->rad_lock,
2023                                                               flags);
2024                                         continue;
2025                                 }
2026
2027                                 /* retry with exponential backoff until HZ */
2028                                 if (conn->rac_keepalive == 0)
2029                                         conn->rac_keepalive = 1;
2030                                 else if (conn->rac_keepalive <=
2031                                          msecs_to_jiffies(MSEC_PER_SEC))
2032                                         conn->rac_keepalive *= 2;
2033                                 else
2034                                         conn->rac_keepalive +=
2035                                                 msecs_to_jiffies(MSEC_PER_SEC);
2036
2037                                 deadline = conn->rac_last_tx + conn->rac_keepalive;
2038                                 spin_lock_irqsave(&dev->rad_lock, flags);
2039                         }
2040
2041                         /* Does this conn need attention soonest? */
2042                         if (nsoonest++ == 0 ||
2043                             !cfs_time_aftereq(deadline, soonest))
2044                                 soonest = deadline;
2045                 }
2046
2047                 if (dropped_lock)               /* may sleep iff I didn't drop the lock */
2048                         continue;
2049
2050                 set_current_state(TASK_INTERRUPTIBLE);
2051                 add_wait_queue_exclusive(&dev->rad_waitq, &wait);
2052                 spin_unlock_irqrestore(&dev->rad_lock, flags);
2053
2054                 if (nsoonest == 0) {
2055                         busy_loops = 0;
2056                         waitq_wait(&wait, TASK_INTERRUPTIBLE);
2057                 } else {
2058                         timeout = (long)(soonest - jiffies);
2059                         if (timeout > 0) {
2060                                 busy_loops = 0;
2061                                 waitq_timedwait(&wait,
2062                                                     TASK_INTERRUPTIBLE,
2063                                                     timeout);
2064                         }
2065                 }
2066
2067                 remove_wait_queue(&dev->rad_waitq, &wait);
2068                 set_current_state(TASK_RUNNING);
2069                 spin_lock_irqsave(&dev->rad_lock, flags);
2070         }
2071
2072         spin_unlock_irqrestore(&dev->rad_lock, flags);
2073
2074         dev->rad_scheduler = NULL;
2075         kranal_thread_fini();
2076         return 0;
2077 }