Whamcloud - gitweb
LU-1346 gnilnd: remove libcfs abstractions
[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 (!cfs_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 (!cfs_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 (!cfs_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                 cfs_atomic_inc(&kranal_data.kra_nthreads);
900         return PTR_ERR(task);
901 }
902
903 void
904 kranal_thread_fini (void)
905 {
906         cfs_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 + conn->rac_keepalive *
923                              HZ)) {
924                 /* not sent in a while; schedule conn so scheduler sends a keepalive */
925                 CDEBUG(D_NET, "Scheduling keepalive %p->%s\n",
926                        conn, libcfs_nid2str(conn->rac_peer->rap_nid));
927                 kranal_schedule_conn(conn);
928         }
929
930         timeout = conn->rac_timeout * HZ;
931
932         if (!conn->rac_close_recvd &&
933             cfs_time_aftereq(now, conn->rac_last_rx + timeout)) {
934                 CERROR("%s received from %s within %lu seconds\n",
935                        (conn->rac_state == RANAL_CONN_ESTABLISHED) ?
936                        "Nothing" : "CLOSE not",
937                        libcfs_nid2str(conn->rac_peer->rap_nid),
938                        (now - conn->rac_last_rx)/HZ);
939                 return -ETIMEDOUT;
940         }
941
942         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
943                 return 0;
944
945         /* Check the conn's queues are moving.  These are "belt+braces" checks,
946          * in case of hardware/software errors that make this conn seem
947          * responsive even though it isn't progressing its message queues. */
948
949         spin_lock_irqsave(&conn->rac_lock, flags);
950
951         cfs_list_for_each (ttmp, &conn->rac_fmaq) {
952                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
953
954                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
955                         spin_unlock_irqrestore(&conn->rac_lock, flags);
956                         CERROR("tx on fmaq for %s blocked %lu seconds\n",
957                                libcfs_nid2str(conn->rac_peer->rap_nid),
958                                (now - tx->tx_qtime)/HZ);
959                         return -ETIMEDOUT;
960                 }
961         }
962
963         cfs_list_for_each (ttmp, &conn->rac_rdmaq) {
964                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
965
966                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
967                         spin_unlock_irqrestore(&conn->rac_lock, flags);
968                         CERROR("tx on rdmaq for %s blocked %lu seconds\n",
969                                libcfs_nid2str(conn->rac_peer->rap_nid),
970                                (now - tx->tx_qtime)/HZ);
971                         return -ETIMEDOUT;
972                 }
973         }
974
975         cfs_list_for_each (ttmp, &conn->rac_replyq) {
976                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
977
978                 if (cfs_time_aftereq(now, tx->tx_qtime + timeout)) {
979                         spin_unlock_irqrestore(&conn->rac_lock, flags);
980                         CERROR("tx on replyq for %s blocked %lu seconds\n",
981                                libcfs_nid2str(conn->rac_peer->rap_nid),
982                                (now - tx->tx_qtime)/HZ);
983                         return -ETIMEDOUT;
984                 }
985         }
986
987         spin_unlock_irqrestore(&conn->rac_lock, flags);
988         return 0;
989 }
990
991 void
992 kranal_reaper_check (int idx, unsigned long *min_timeoutp)
993 {
994         cfs_list_t        *conns = &kranal_data.kra_conns[idx];
995         cfs_list_t        *ctmp;
996         kra_conn_t        *conn;
997         unsigned long      flags;
998         int                rc;
999
1000  again:
1001         /* NB. We expect to check all the conns and not find any problems, so
1002          * we just use a shared lock while we take a look... */
1003         read_lock(&kranal_data.kra_global_lock);
1004
1005         cfs_list_for_each (ctmp, conns) {
1006                 conn = cfs_list_entry(ctmp, kra_conn_t, rac_hashlist);
1007
1008                 if (conn->rac_timeout < *min_timeoutp )
1009                         *min_timeoutp = conn->rac_timeout;
1010                 if (conn->rac_keepalive < *min_timeoutp )
1011                         *min_timeoutp = conn->rac_keepalive;
1012
1013                 rc = kranal_check_conn_timeouts(conn);
1014                 if (rc == 0)
1015                         continue;
1016
1017                 kranal_conn_addref(conn);
1018                 read_unlock(&kranal_data.kra_global_lock);
1019
1020                 CERROR("Conn to %s, cqid %d timed out\n",
1021                        libcfs_nid2str(conn->rac_peer->rap_nid), 
1022                        conn->rac_cqid);
1023
1024                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1025
1026                 switch (conn->rac_state) {
1027                 default:
1028                         LBUG();
1029
1030                 case RANAL_CONN_ESTABLISHED:
1031                         kranal_close_conn_locked(conn, -ETIMEDOUT);
1032                         break;
1033
1034                 case RANAL_CONN_CLOSING:
1035                         kranal_terminate_conn_locked(conn);
1036                         break;
1037                 }
1038
1039                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1040                                             flags);
1041
1042                 kranal_conn_decref(conn);
1043
1044                 /* start again now I've dropped the lock */
1045                 goto again;
1046         }
1047
1048         read_unlock(&kranal_data.kra_global_lock);
1049 }
1050
1051 int
1052 kranal_connd (void *arg)
1053 {
1054         long               id = (long)arg;
1055         wait_queue_t     wait;
1056         unsigned long      flags;
1057         kra_peer_t        *peer;
1058         kra_acceptsock_t  *ras;
1059         int                did_something;
1060
1061         cfs_block_allsigs();
1062
1063         init_waitqueue_entry_current(&wait);
1064
1065         spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1066
1067         while (!kranal_data.kra_shutdown) {
1068                 did_something = 0;
1069
1070                 if (!cfs_list_empty(&kranal_data.kra_connd_acceptq)) {
1071                         ras = cfs_list_entry(kranal_data.kra_connd_acceptq.next,
1072                                              kra_acceptsock_t, ras_list);
1073                         cfs_list_del(&ras->ras_list);
1074
1075                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1076                                                    flags);
1077
1078                         CDEBUG(D_NET,"About to handshake someone\n");
1079
1080                         kranal_conn_handshake(ras->ras_sock, NULL);
1081                         kranal_free_acceptsock(ras);
1082
1083                         CDEBUG(D_NET,"Finished handshaking someone\n");
1084
1085                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1086                                               flags);
1087                         did_something = 1;
1088                 }
1089
1090                 if (!cfs_list_empty(&kranal_data.kra_connd_peers)) {
1091                         peer = cfs_list_entry(kranal_data.kra_connd_peers.next,
1092                                               kra_peer_t, rap_connd_list);
1093
1094                         cfs_list_del_init(&peer->rap_connd_list);
1095                         spin_unlock_irqrestore(&kranal_data.kra_connd_lock,
1096                                                    flags);
1097
1098                         kranal_connect(peer);
1099                         kranal_peer_decref(peer);
1100
1101                         spin_lock_irqsave(&kranal_data.kra_connd_lock,
1102                                               flags);
1103                         did_something = 1;
1104                 }
1105
1106                 if (did_something)
1107                         continue;
1108
1109                 set_current_state(TASK_INTERRUPTIBLE);
1110                 add_wait_queue_exclusive(&kranal_data.kra_connd_waitq, &wait);
1111
1112                 spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1113
1114                 waitq_wait(&wait, TASK_INTERRUPTIBLE);
1115
1116                 set_current_state(TASK_RUNNING);
1117                 remove_wait_queue(&kranal_data.kra_connd_waitq, &wait);
1118
1119                 spin_lock_irqsave(&kranal_data.kra_connd_lock, flags);
1120         }
1121
1122         spin_unlock_irqrestore(&kranal_data.kra_connd_lock, flags);
1123
1124         kranal_thread_fini();
1125         return 0;
1126 }
1127
1128 void
1129 kranal_update_reaper_timeout(long timeout)
1130 {
1131         unsigned long   flags;
1132
1133         LASSERT (timeout > 0);
1134
1135         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1136
1137         if (timeout < kranal_data.kra_new_min_timeout)
1138                 kranal_data.kra_new_min_timeout = timeout;
1139
1140         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1141 }
1142
1143 int
1144 kranal_reaper (void *arg)
1145 {
1146         wait_queue_t     wait;
1147         unsigned long      flags;
1148         long               timeout;
1149         int                i;
1150         int                conn_entries = kranal_data.kra_conn_hash_size;
1151         int                conn_index = 0;
1152         int                base_index = conn_entries - 1;
1153         unsigned long      next_check_time = jiffies;
1154         long               next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1155         long               current_min_timeout = 1;
1156
1157         cfs_block_allsigs();
1158
1159         init_waitqueue_entry_current(&wait);
1160
1161         spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1162
1163         while (!kranal_data.kra_shutdown) {
1164                 /* I wake up every 'p' seconds to check for timeouts on some
1165                  * more peers.  I try to check every connection 'n' times
1166                  * within the global minimum of all keepalive and timeout
1167                  * intervals, to ensure I attend to every connection within
1168                  * (n+1)/n times its timeout intervals. */
1169                 const int     p = 1;
1170                 const int     n = 3;
1171                 unsigned long min_timeout;
1172                 int           chunk;
1173
1174                 /* careful with the jiffy wrap... */
1175                 timeout = (long)(next_check_time - jiffies);
1176                 if (timeout > 0) {
1177                         set_current_state(TASK_INTERRUPTIBLE);
1178                         add_wait_queue(&kranal_data.kra_reaper_waitq, &wait);
1179
1180                         spin_unlock_irqrestore(&kranal_data.kra_reaper_lock,
1181                                                    flags);
1182
1183                         waitq_timedwait(&wait, TASK_INTERRUPTIBLE,
1184                                             timeout);
1185
1186                         spin_lock_irqsave(&kranal_data.kra_reaper_lock,
1187                                               flags);
1188
1189                         set_current_state(TASK_RUNNING);
1190                         remove_wait_queue(&kranal_data.kra_reaper_waitq, &wait);
1191                         continue;
1192                 }
1193
1194                 if (kranal_data.kra_new_min_timeout !=
1195                     MAX_SCHEDULE_TIMEOUT) {
1196                         /* new min timeout set: restart min timeout scan */
1197                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1198                         base_index = conn_index - 1;
1199                         if (base_index < 0)
1200                                 base_index = conn_entries - 1;
1201
1202                         if (kranal_data.kra_new_min_timeout <
1203                             current_min_timeout) {
1204                                 current_min_timeout =
1205                                         kranal_data.kra_new_min_timeout;
1206                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1207                                        current_min_timeout);
1208                         }
1209
1210                         kranal_data.kra_new_min_timeout =
1211                                 MAX_SCHEDULE_TIMEOUT;
1212                 }
1213                 min_timeout = current_min_timeout;
1214
1215                 spin_unlock_irqrestore(&kranal_data.kra_reaper_lock, flags);
1216
1217                 LASSERT (min_timeout > 0);
1218
1219                 /* Compute how many table entries to check now so I get round
1220                  * the whole table fast enough given that I do this at fixed
1221                  * intervals of 'p' seconds) */
1222                 chunk = conn_entries;
1223                 if (min_timeout > n * p)
1224                         chunk = (chunk * n * p) / min_timeout;
1225                 if (chunk == 0)
1226                         chunk = 1;
1227
1228                 for (i = 0; i < chunk; i++) {
1229                         kranal_reaper_check(conn_index,
1230                                             &next_min_timeout);
1231                         conn_index = (conn_index + 1) % conn_entries;
1232                 }
1233
1234                 next_check_time += p * HZ;
1235
1236                 spin_lock_irqsave(&kranal_data.kra_reaper_lock, flags);
1237
1238                 if (((conn_index - chunk <= base_index &&
1239                       base_index < conn_index) ||
1240                      (conn_index - conn_entries - chunk <= base_index &&
1241                       base_index < conn_index - conn_entries))) {
1242
1243                         /* Scanned all conns: set current_min_timeout... */
1244                         if (current_min_timeout != next_min_timeout) {
1245                                 current_min_timeout = next_min_timeout;
1246                                 CDEBUG(D_NET, "Set new min timeout %ld\n",
1247                                        current_min_timeout);
1248                         }
1249
1250                         /* ...and restart min timeout scan */
1251                         next_min_timeout = MAX_SCHEDULE_TIMEOUT;
1252                         base_index = conn_index - 1;
1253                         if (base_index < 0)
1254                                 base_index = conn_entries - 1;
1255                 }
1256         }
1257
1258         kranal_thread_fini();
1259         return 0;
1260 }
1261
1262 void
1263 kranal_check_rdma_cq (kra_device_t *dev)
1264 {
1265         kra_conn_t          *conn;
1266         kra_tx_t            *tx;
1267         RAP_RETURN           rrc;
1268         unsigned long        flags;
1269         RAP_RDMA_DESCRIPTOR *desc;
1270         __u32                cqid;
1271         __u32                event_type;
1272
1273         for (;;) {
1274                 rrc = RapkCQDone(dev->rad_rdma_cqh, &cqid, &event_type);
1275                 if (rrc == RAP_NOT_DONE) {
1276                         CDEBUG(D_NET, "RDMA CQ %d empty\n", dev->rad_id);
1277                         return;
1278                 }
1279
1280                 LASSERT (rrc == RAP_SUCCESS);
1281                 LASSERT ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0);
1282
1283                 read_lock(&kranal_data.kra_global_lock);
1284
1285                 conn = kranal_cqid2conn_locked(cqid);
1286                 if (conn == NULL) {
1287                         /* Conn was destroyed? */
1288                         CDEBUG(D_NET, "RDMA CQID lookup %d failed\n", cqid);
1289                         read_unlock(&kranal_data.kra_global_lock);
1290                         continue;
1291                 }
1292
1293                 rrc = RapkRdmaDone(conn->rac_rihandle, &desc);
1294                 LASSERT (rrc == RAP_SUCCESS);
1295
1296                 CDEBUG(D_NET, "Completed %p\n",
1297                        cfs_list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list));
1298
1299                 spin_lock_irqsave(&conn->rac_lock, flags);
1300
1301                 LASSERT (!cfs_list_empty(&conn->rac_rdmaq));
1302                 tx = cfs_list_entry(conn->rac_rdmaq.next, kra_tx_t, tx_list);
1303                 cfs_list_del(&tx->tx_list);
1304
1305                 LASSERT(desc->AppPtr == (void *)tx);
1306                 LASSERT(tx->tx_msg.ram_type == RANAL_MSG_PUT_DONE ||
1307                         tx->tx_msg.ram_type == RANAL_MSG_GET_DONE);
1308
1309                 cfs_list_add_tail(&tx->tx_list, &conn->rac_fmaq);
1310                 tx->tx_qtime = jiffies;
1311
1312                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1313
1314                 /* Get conn's fmaq processed, now I've just put something
1315                  * there */
1316                 kranal_schedule_conn(conn);
1317
1318                 read_unlock(&kranal_data.kra_global_lock);
1319         }
1320 }
1321
1322 void
1323 kranal_check_fma_cq (kra_device_t *dev)
1324 {
1325         kra_conn_t         *conn;
1326         RAP_RETURN          rrc;
1327         __u32               cqid;
1328         __u32               event_type;
1329         cfs_list_t         *conns;
1330         cfs_list_t         *tmp;
1331         int                 i;
1332
1333         for (;;) {
1334                 rrc = RapkCQDone(dev->rad_fma_cqh, &cqid, &event_type);
1335                 if (rrc == RAP_NOT_DONE) {
1336                         CDEBUG(D_NET, "FMA CQ %d empty\n", dev->rad_id);
1337                         return;
1338                 }
1339
1340                 LASSERT (rrc == RAP_SUCCESS);
1341
1342                 if ((event_type & RAPK_CQ_EVENT_OVERRUN) == 0) {
1343
1344                         read_lock(&kranal_data.kra_global_lock);
1345
1346                         conn = kranal_cqid2conn_locked(cqid);
1347                         if (conn == NULL) {
1348                                 CDEBUG(D_NET, "FMA CQID lookup %d failed\n",
1349                                        cqid);
1350                         } else {
1351                                 CDEBUG(D_NET, "FMA completed: %p CQID %d\n",
1352                                        conn, cqid);
1353                                 kranal_schedule_conn(conn);
1354                         }
1355
1356                         read_unlock(&kranal_data.kra_global_lock);
1357                         continue;
1358                 }
1359
1360                 /* FMA CQ has overflowed: check ALL conns */
1361                 CWARN("FMA CQ overflow: scheduling ALL conns on device %d\n", 
1362                       dev->rad_id);
1363
1364                 for (i = 0; i < kranal_data.kra_conn_hash_size; i++) {
1365
1366                         read_lock(&kranal_data.kra_global_lock);
1367
1368                         conns = &kranal_data.kra_conns[i];
1369
1370                         cfs_list_for_each (tmp, conns) {
1371                                 conn = cfs_list_entry(tmp, kra_conn_t,
1372                                                       rac_hashlist);
1373
1374                                 if (conn->rac_device == dev)
1375                                         kranal_schedule_conn(conn);
1376                         }
1377
1378                         /* don't block write lockers for too long... */
1379                         read_unlock(&kranal_data.kra_global_lock);
1380                 }
1381         }
1382 }
1383
1384 int
1385 kranal_sendmsg(kra_conn_t *conn, kra_msg_t *msg,
1386                void *immediate, int immediatenob)
1387 {
1388         int        sync = (msg->ram_type & RANAL_MSG_FENCE) != 0;
1389         RAP_RETURN rrc;
1390
1391         CDEBUG(D_NET,"%p sending msg %p %02x%s [%p for %d]\n",
1392                conn, msg, msg->ram_type, sync ? "(sync)" : "",
1393                immediate, immediatenob);
1394
1395         LASSERT (sizeof(*msg) <= RANAL_FMA_MAX_PREFIX);
1396         LASSERT ((msg->ram_type == RANAL_MSG_IMMEDIATE) ?
1397                  immediatenob <= RANAL_FMA_MAX_DATA :
1398                  immediatenob == 0);
1399
1400         msg->ram_connstamp = conn->rac_my_connstamp;
1401         msg->ram_seq = conn->rac_tx_seq;
1402
1403         if (sync)
1404                 rrc = RapkFmaSyncSend(conn->rac_rihandle,
1405                                       immediate, immediatenob,
1406                                       msg, sizeof(*msg));
1407         else
1408                 rrc = RapkFmaSend(conn->rac_rihandle,
1409                                   immediate, immediatenob,
1410                                   msg, sizeof(*msg));
1411
1412         switch (rrc) {
1413         default:
1414                 LBUG();
1415
1416         case RAP_SUCCESS:
1417                 conn->rac_last_tx = jiffies;
1418                 conn->rac_tx_seq++;
1419                 return 0;
1420
1421         case RAP_NOT_DONE:
1422                 if (cfs_time_aftereq(jiffies,
1423                                      conn->rac_last_tx + conn->rac_keepalive *
1424                                      HZ))
1425                         CWARN("EAGAIN sending %02x (idle %lu secs)\n",
1426                               msg->ram_type,
1427                               (jiffies - conn->rac_last_tx)/HZ);
1428                 return -EAGAIN;
1429         }
1430 }
1431
1432 void
1433 kranal_process_fmaq (kra_conn_t *conn)
1434 {
1435         unsigned long flags;
1436         int           more_to_do;
1437         kra_tx_t     *tx;
1438         int           rc;
1439         int           expect_reply;
1440
1441         /* NB 1. kranal_sendmsg() may fail if I'm out of credits right now.
1442          *       However I will be rescheduled by an FMA completion event
1443          *       when I eventually get some.
1444          * NB 2. Sampling rac_state here races with setting it elsewhere.
1445          *       But it doesn't matter if I try to send a "real" message just
1446          *       as I start closing because I'll get scheduled to send the
1447          *       close anyway. */
1448
1449         /* Not racing with incoming message processing! */
1450         LASSERT (current == conn->rac_device->rad_scheduler);
1451
1452         if (conn->rac_state != RANAL_CONN_ESTABLISHED) {
1453                 if (!cfs_list_empty(&conn->rac_rdmaq)) {
1454                         /* RDMAs in progress */
1455                         LASSERT (!conn->rac_close_sent);
1456
1457                         if (cfs_time_aftereq(jiffies,
1458                                              conn->rac_last_tx +
1459                                              conn->rac_keepalive * HZ)) {
1460                                 CDEBUG(D_NET, "sending NOOP (rdma in progress)\n");
1461                                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1462                                 kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1463                         }
1464                         return;
1465                 }
1466
1467                 if (conn->rac_close_sent)
1468                         return;
1469
1470                 CWARN("sending CLOSE to %s\n", 
1471                       libcfs_nid2str(conn->rac_peer->rap_nid));
1472                 kranal_init_msg(&conn->rac_msg, RANAL_MSG_CLOSE);
1473                 rc = kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1474                 if (rc != 0)
1475                         return;
1476
1477                 conn->rac_close_sent = 1;
1478                 if (!conn->rac_close_recvd)
1479                         return;
1480
1481                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1482
1483                 if (conn->rac_state == RANAL_CONN_CLOSING)
1484                         kranal_terminate_conn_locked(conn);
1485
1486                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1487                                             flags);
1488                 return;
1489         }
1490
1491         spin_lock_irqsave(&conn->rac_lock, flags);
1492
1493         if (cfs_list_empty(&conn->rac_fmaq)) {
1494
1495                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1496
1497                 if (cfs_time_aftereq(jiffies,
1498                                      conn->rac_last_tx + conn->rac_keepalive *
1499                                      HZ)) {
1500                         CDEBUG(D_NET, "sending NOOP -> %s (%p idle %lu(%ld))\n",
1501                                libcfs_nid2str(conn->rac_peer->rap_nid), conn,
1502                                (jiffies - conn->rac_last_tx)/HZ,
1503                                conn->rac_keepalive);
1504                         kranal_init_msg(&conn->rac_msg, RANAL_MSG_NOOP);
1505                         kranal_sendmsg(conn, &conn->rac_msg, NULL, 0);
1506                 }
1507                 return;
1508         }
1509
1510         tx = cfs_list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1511         cfs_list_del(&tx->tx_list);
1512         more_to_do = !cfs_list_empty(&conn->rac_fmaq);
1513
1514         spin_unlock_irqrestore(&conn->rac_lock, flags);
1515
1516         expect_reply = 0;
1517         CDEBUG(D_NET, "sending regular msg: %p, type %02x, cookie "LPX64"\n",
1518                tx, tx->tx_msg.ram_type, tx->tx_cookie);
1519         switch (tx->tx_msg.ram_type) {
1520         default:
1521                 LBUG();
1522
1523         case RANAL_MSG_IMMEDIATE:
1524                 rc = kranal_sendmsg(conn, &tx->tx_msg,
1525                                     tx->tx_buffer, tx->tx_nob);
1526                 break;
1527
1528         case RANAL_MSG_PUT_NAK:
1529         case RANAL_MSG_PUT_DONE:
1530         case RANAL_MSG_GET_NAK:
1531         case RANAL_MSG_GET_DONE:
1532                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1533                 break;
1534
1535         case RANAL_MSG_PUT_REQ:
1536                 rc = kranal_map_buffer(tx);
1537                 LASSERT (rc != -EAGAIN);
1538                 if (rc != 0)
1539                         break;
1540
1541                 tx->tx_msg.ram_u.putreq.raprm_cookie = tx->tx_cookie;
1542                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1543                 expect_reply = 1;
1544                 break;
1545
1546         case RANAL_MSG_PUT_ACK:
1547                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1548                 expect_reply = 1;
1549                 break;
1550
1551         case RANAL_MSG_GET_REQ:
1552                 rc = kranal_map_buffer(tx);
1553                 LASSERT (rc != -EAGAIN);
1554                 if (rc != 0)
1555                         break;
1556
1557                 tx->tx_msg.ram_u.get.ragm_cookie = tx->tx_cookie;
1558                 tx->tx_msg.ram_u.get.ragm_desc.rard_key = tx->tx_map_key;
1559                 tx->tx_msg.ram_u.get.ragm_desc.rard_addr.AddressBits =
1560                         (__u64)((unsigned long)tx->tx_buffer);
1561                 tx->tx_msg.ram_u.get.ragm_desc.rard_nob = tx->tx_nob;
1562                 rc = kranal_sendmsg(conn, &tx->tx_msg, NULL, 0);
1563                 expect_reply = 1;
1564                 break;
1565         }
1566
1567         if (rc == -EAGAIN) {
1568                 /* I need credits to send this.  Replace tx at the head of the
1569                  * fmaq and I'll get rescheduled when credits appear */
1570                 CDEBUG(D_NET, "EAGAIN on %p\n", conn);
1571                 spin_lock_irqsave(&conn->rac_lock, flags);
1572                 cfs_list_add(&tx->tx_list, &conn->rac_fmaq);
1573                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1574                 return;
1575         }
1576
1577         if (!expect_reply || rc != 0) {
1578                 kranal_tx_done(tx, rc);
1579         } else {
1580                 /* LASSERT(current) above ensures this doesn't race with reply
1581                  * processing */
1582                 spin_lock_irqsave(&conn->rac_lock, flags);
1583                 cfs_list_add_tail(&tx->tx_list, &conn->rac_replyq);
1584                 tx->tx_qtime = jiffies;
1585                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1586         }
1587
1588         if (more_to_do) {
1589                 CDEBUG(D_NET, "Rescheduling %p (more to do)\n", conn);
1590                 kranal_schedule_conn(conn);
1591         }
1592 }
1593
1594 static inline void
1595 kranal_swab_rdma_desc (kra_rdma_desc_t *d)
1596 {
1597         __swab64s(&d->rard_key.Key);
1598         __swab16s(&d->rard_key.Cookie);
1599         __swab16s(&d->rard_key.MdHandle);
1600         __swab32s(&d->rard_key.Flags);
1601         __swab64s(&d->rard_addr.AddressBits);
1602         __swab32s(&d->rard_nob);
1603 }
1604
1605 kra_tx_t *
1606 kranal_match_reply(kra_conn_t *conn, int type, __u64 cookie)
1607 {
1608         cfs_list_t       *ttmp;
1609         kra_tx_t         *tx;
1610         unsigned long     flags;
1611
1612         spin_lock_irqsave(&conn->rac_lock, flags);
1613
1614         cfs_list_for_each(ttmp, &conn->rac_replyq) {
1615                 tx = cfs_list_entry(ttmp, kra_tx_t, tx_list);
1616
1617                 CDEBUG(D_NET,"Checking %p %02x/"LPX64"\n",
1618                        tx, tx->tx_msg.ram_type, tx->tx_cookie);
1619
1620                 if (tx->tx_cookie != cookie)
1621                         continue;
1622
1623                 if (tx->tx_msg.ram_type != type) {
1624                         spin_unlock_irqrestore(&conn->rac_lock, flags);
1625                         CWARN("Unexpected type %x (%x expected) "
1626                               "matched reply from %s\n",
1627                               tx->tx_msg.ram_type, type,
1628                               libcfs_nid2str(conn->rac_peer->rap_nid));
1629                         return NULL;
1630                 }
1631
1632                 cfs_list_del(&tx->tx_list);
1633                 spin_unlock_irqrestore(&conn->rac_lock, flags);
1634                 return tx;
1635         }
1636
1637         spin_unlock_irqrestore(&conn->rac_lock, flags);
1638         CWARN("Unmatched reply %02x/"LPX64" from %s\n",
1639               type, cookie, libcfs_nid2str(conn->rac_peer->rap_nid));
1640         return NULL;
1641 }
1642
1643 void
1644 kranal_check_fma_rx (kra_conn_t *conn)
1645 {
1646         unsigned long flags;
1647         __u32         seq;
1648         kra_tx_t     *tx;
1649         kra_msg_t    *msg;
1650         void         *prefix;
1651         RAP_RETURN    rrc = RapkFmaGetPrefix(conn->rac_rihandle, &prefix);
1652         kra_peer_t   *peer = conn->rac_peer;
1653         int           rc = 0;
1654         int           repost = 1;
1655
1656         if (rrc == RAP_NOT_DONE)
1657                 return;
1658
1659         CDEBUG(D_NET, "RX on %p\n", conn);
1660
1661         LASSERT (rrc == RAP_SUCCESS);
1662         conn->rac_last_rx = jiffies;
1663         seq = conn->rac_rx_seq++;
1664         msg = (kra_msg_t *)prefix;
1665
1666         /* stash message for portals callbacks they'll NULL
1667          * rac_rxmsg if they consume it */
1668         LASSERT (conn->rac_rxmsg == NULL);
1669         conn->rac_rxmsg = msg;
1670
1671         if (msg->ram_magic != RANAL_MSG_MAGIC) {
1672                 if (__swab32(msg->ram_magic) != RANAL_MSG_MAGIC) {
1673                         CERROR("Unexpected magic %08x from %s\n",
1674                                msg->ram_magic, libcfs_nid2str(peer->rap_nid));
1675                         rc = -EPROTO;
1676                         goto out;
1677                 }
1678
1679                 __swab32s(&msg->ram_magic);
1680                 __swab16s(&msg->ram_version);
1681                 __swab16s(&msg->ram_type);
1682                 __swab64s(&msg->ram_srcnid);
1683                 __swab64s(&msg->ram_connstamp);
1684                 __swab32s(&msg->ram_seq);
1685
1686                 /* NB message type checked below; NOT here... */
1687                 switch (msg->ram_type) {
1688                 case RANAL_MSG_PUT_ACK:
1689                         kranal_swab_rdma_desc(&msg->ram_u.putack.rapam_desc);
1690                         break;
1691
1692                 case RANAL_MSG_GET_REQ:
1693                         kranal_swab_rdma_desc(&msg->ram_u.get.ragm_desc);
1694                         break;
1695
1696                 default:
1697                         break;
1698                 }
1699         }
1700
1701         if (msg->ram_version != RANAL_MSG_VERSION) {
1702                 CERROR("Unexpected protocol version %d from %s\n",
1703                        msg->ram_version, libcfs_nid2str(peer->rap_nid));
1704                 rc = -EPROTO;
1705                 goto out;
1706         }
1707
1708         if (msg->ram_srcnid != peer->rap_nid) {
1709                 CERROR("Unexpected peer %s from %s\n",
1710                        libcfs_nid2str(msg->ram_srcnid), 
1711                        libcfs_nid2str(peer->rap_nid));
1712                 rc = -EPROTO;
1713                 goto out;
1714         }
1715
1716         if (msg->ram_connstamp != conn->rac_peer_connstamp) {
1717                 CERROR("Unexpected connstamp "LPX64"("LPX64
1718                        " expected) from %s\n",
1719                        msg->ram_connstamp, conn->rac_peer_connstamp,
1720                        libcfs_nid2str(peer->rap_nid));
1721                 rc = -EPROTO;
1722                 goto out;
1723         }
1724
1725         if (msg->ram_seq != seq) {
1726                 CERROR("Unexpected sequence number %d(%d expected) from %s\n",
1727                        msg->ram_seq, seq, libcfs_nid2str(peer->rap_nid));
1728                 rc = -EPROTO;
1729                 goto out;
1730         }
1731
1732         if ((msg->ram_type & RANAL_MSG_FENCE) != 0) {
1733                 /* This message signals RDMA completion... */
1734                 rrc = RapkFmaSyncWait(conn->rac_rihandle);
1735                 if (rrc != RAP_SUCCESS) {
1736                         CERROR("RapkFmaSyncWait failed: %d\n", rrc);
1737                         rc = -ENETDOWN;
1738                         goto out;
1739                 }
1740         }
1741
1742         if (conn->rac_close_recvd) {
1743                 CERROR("Unexpected message %d after CLOSE from %s\n",
1744                        msg->ram_type, libcfs_nid2str(conn->rac_peer->rap_nid));
1745                 rc = -EPROTO;
1746                 goto out;
1747         }
1748
1749         if (msg->ram_type == RANAL_MSG_CLOSE) {
1750                 CWARN("RX CLOSE from %s\n", libcfs_nid2str(conn->rac_peer->rap_nid));
1751                 conn->rac_close_recvd = 1;
1752                 write_lock_irqsave(&kranal_data.kra_global_lock, flags);
1753
1754                 if (conn->rac_state == RANAL_CONN_ESTABLISHED)
1755                         kranal_close_conn_locked(conn, 0);
1756                 else if (conn->rac_state == RANAL_CONN_CLOSING &&
1757                          conn->rac_close_sent)
1758                         kranal_terminate_conn_locked(conn);
1759
1760                 write_unlock_irqrestore(&kranal_data.kra_global_lock,
1761                                             flags);
1762                 goto out;
1763         }
1764
1765         if (conn->rac_state != RANAL_CONN_ESTABLISHED)
1766                 goto out;
1767
1768         switch (msg->ram_type) {
1769         case RANAL_MSG_NOOP:
1770                 /* Nothing to do; just a keepalive */
1771                 CDEBUG(D_NET, "RX NOOP on %p\n", conn);
1772                 break;
1773
1774         case RANAL_MSG_IMMEDIATE:
1775                 CDEBUG(D_NET, "RX IMMEDIATE on %p\n", conn);
1776                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.immediate.raim_hdr, 
1777                                 msg->ram_srcnid, conn, 0);
1778                 repost = rc < 0;
1779                 break;
1780
1781         case RANAL_MSG_PUT_REQ:
1782                 CDEBUG(D_NET, "RX PUT_REQ on %p\n", conn);
1783                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.putreq.raprm_hdr, 
1784                                 msg->ram_srcnid, conn, 1);
1785                 repost = rc < 0;
1786                 break;
1787
1788         case RANAL_MSG_PUT_NAK:
1789                 CDEBUG(D_NET, "RX PUT_NAK on %p\n", conn);
1790                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1791                                         msg->ram_u.completion.racm_cookie);
1792                 if (tx == NULL)
1793                         break;
1794
1795                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1796                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1797                 kranal_tx_done(tx, -ENOENT);    /* no match */
1798                 break;
1799
1800         case RANAL_MSG_PUT_ACK:
1801                 CDEBUG(D_NET, "RX PUT_ACK on %p\n", conn);
1802                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_REQ,
1803                                         msg->ram_u.putack.rapam_src_cookie);
1804                 if (tx == NULL)
1805                         break;
1806
1807                 kranal_rdma(tx, RANAL_MSG_PUT_DONE,
1808                             &msg->ram_u.putack.rapam_desc,
1809                             msg->ram_u.putack.rapam_desc.rard_nob,
1810                             msg->ram_u.putack.rapam_dst_cookie);
1811                 break;
1812
1813         case RANAL_MSG_PUT_DONE:
1814                 CDEBUG(D_NET, "RX PUT_DONE on %p\n", conn);
1815                 tx = kranal_match_reply(conn, RANAL_MSG_PUT_ACK,
1816                                         msg->ram_u.completion.racm_cookie);
1817                 if (tx == NULL)
1818                         break;
1819
1820                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1821                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1822                 kranal_tx_done(tx, 0);
1823                 break;
1824
1825         case RANAL_MSG_GET_REQ:
1826                 CDEBUG(D_NET, "RX GET_REQ on %p\n", conn);
1827                 rc = lnet_parse(kranal_data.kra_ni, &msg->ram_u.get.ragm_hdr, 
1828                                 msg->ram_srcnid, conn, 1);
1829                 repost = rc < 0;
1830                 break;
1831
1832         case RANAL_MSG_GET_NAK:
1833                 CDEBUG(D_NET, "RX GET_NAK on %p\n", conn);
1834                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1835                                         msg->ram_u.completion.racm_cookie);
1836                 if (tx == NULL)
1837                         break;
1838
1839                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1840                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1841                 kranal_tx_done(tx, -ENOENT);    /* no match */
1842                 break;
1843
1844         case RANAL_MSG_GET_DONE:
1845                 CDEBUG(D_NET, "RX GET_DONE on %p\n", conn);
1846                 tx = kranal_match_reply(conn, RANAL_MSG_GET_REQ,
1847                                         msg->ram_u.completion.racm_cookie);
1848                 if (tx == NULL)
1849                         break;
1850
1851                 LASSERT (tx->tx_buftype == RANAL_BUF_PHYS_MAPPED ||
1852                          tx->tx_buftype == RANAL_BUF_VIRT_MAPPED);
1853 #if 0
1854                 /* completion message should send rdma length if we ever allow
1855                  * GET truncation */
1856                 lnet_set_reply_msg_len(kranal_data.kra_ni, tx->tx_lntmsg[1], ???);
1857 #endif
1858                 kranal_tx_done(tx, 0);
1859                 break;
1860         }
1861
1862  out:
1863         if (rc < 0)                             /* protocol/comms error */
1864                 kranal_close_conn (conn, rc);
1865
1866         if (repost && conn->rac_rxmsg != NULL)
1867                 kranal_consume_rxmsg(conn, NULL, 0);
1868
1869         /* check again later */
1870         kranal_schedule_conn(conn);
1871 }
1872
1873 void
1874 kranal_complete_closed_conn (kra_conn_t *conn)
1875 {
1876         kra_tx_t   *tx;
1877         int         nfma;
1878         int         nreplies;
1879
1880         LASSERT (conn->rac_state == RANAL_CONN_CLOSED);
1881         LASSERT (cfs_list_empty(&conn->rac_list));
1882         LASSERT (cfs_list_empty(&conn->rac_hashlist));
1883
1884         for (nfma = 0; !cfs_list_empty(&conn->rac_fmaq); nfma++) {
1885                 tx = cfs_list_entry(conn->rac_fmaq.next, kra_tx_t, tx_list);
1886
1887                 cfs_list_del(&tx->tx_list);
1888                 kranal_tx_done(tx, -ECONNABORTED);
1889         }
1890
1891         LASSERT (cfs_list_empty(&conn->rac_rdmaq));
1892
1893         for (nreplies = 0; !cfs_list_empty(&conn->rac_replyq); nreplies++) {
1894                 tx = cfs_list_entry(conn->rac_replyq.next, kra_tx_t, tx_list);
1895
1896                 cfs_list_del(&tx->tx_list);
1897                 kranal_tx_done(tx, -ECONNABORTED);
1898         }
1899
1900         CWARN("Closed conn %p -> %s: nmsg %d nreplies %d\n",
1901                conn, libcfs_nid2str(conn->rac_peer->rap_nid), nfma, nreplies);
1902 }
1903
1904 int kranal_process_new_conn (kra_conn_t *conn)
1905 {
1906         RAP_RETURN   rrc;
1907
1908         rrc = RapkCompleteSync(conn->rac_rihandle, 1);
1909         if (rrc == RAP_SUCCESS)
1910                 return 0;
1911
1912         LASSERT (rrc == RAP_NOT_DONE);
1913         if (!cfs_time_aftereq(jiffies, conn->rac_last_tx +
1914                               conn->rac_timeout * HZ))
1915                 return -EAGAIN;
1916
1917         /* Too late */
1918         rrc = RapkCompleteSync(conn->rac_rihandle, 0);
1919         LASSERT (rrc == RAP_SUCCESS);
1920         return -ETIMEDOUT;
1921 }
1922
1923 int
1924 kranal_scheduler (void *arg)
1925 {
1926         kra_device_t     *dev = (kra_device_t *)arg;
1927         wait_queue_t    wait;
1928         kra_conn_t       *conn;
1929         unsigned long     flags;
1930         unsigned long     deadline;
1931         unsigned long     soonest;
1932         int               nsoonest;
1933         long              timeout;
1934         cfs_list_t       *tmp;
1935         cfs_list_t       *nxt;
1936         int               rc;
1937         int               dropped_lock;
1938         int               busy_loops = 0;
1939
1940         cfs_block_allsigs();
1941
1942         dev->rad_scheduler = current;
1943         init_waitqueue_entry_current(&wait);
1944
1945         spin_lock_irqsave(&dev->rad_lock, flags);
1946
1947         while (!kranal_data.kra_shutdown) {
1948                 /* Safe: kra_shutdown only set when quiescent */
1949
1950                 if (busy_loops++ >= RANAL_RESCHED) {
1951                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1952
1953                         cond_resched();
1954                         busy_loops = 0;
1955
1956                         spin_lock_irqsave(&dev->rad_lock, flags);
1957                 }
1958
1959                 dropped_lock = 0;
1960
1961                 if (dev->rad_ready) {
1962                         /* Device callback fired since I last checked it */
1963                         dev->rad_ready = 0;
1964                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1965                         dropped_lock = 1;
1966
1967                         kranal_check_rdma_cq(dev);
1968                         kranal_check_fma_cq(dev);
1969
1970                         spin_lock_irqsave(&dev->rad_lock, flags);
1971                 }
1972
1973                 cfs_list_for_each_safe(tmp, nxt, &dev->rad_ready_conns) {
1974                         conn = cfs_list_entry(tmp, kra_conn_t, rac_schedlist);
1975
1976                         cfs_list_del_init(&conn->rac_schedlist);
1977                         LASSERT (conn->rac_scheduled);
1978                         conn->rac_scheduled = 0;
1979                         spin_unlock_irqrestore(&dev->rad_lock, flags);
1980                         dropped_lock = 1;
1981
1982                         kranal_check_fma_rx(conn);
1983                         kranal_process_fmaq(conn);
1984
1985                         if (conn->rac_state == RANAL_CONN_CLOSED)
1986                                 kranal_complete_closed_conn(conn);
1987
1988                         kranal_conn_decref(conn);
1989                         spin_lock_irqsave(&dev->rad_lock, flags);
1990                 }
1991
1992                 nsoonest = 0;
1993                 soonest = jiffies;
1994
1995                 cfs_list_for_each_safe(tmp, nxt, &dev->rad_new_conns) {
1996                         conn = cfs_list_entry(tmp, kra_conn_t, rac_schedlist);
1997
1998                         deadline = conn->rac_last_tx + conn->rac_keepalive;
1999                         if (cfs_time_aftereq(jiffies, deadline)) {
2000                                 /* Time to process this new conn */
2001                                 spin_unlock_irqrestore(&dev->rad_lock,
2002                                                            flags);
2003                                 dropped_lock = 1;
2004
2005                                 rc = kranal_process_new_conn(conn);
2006                                 if (rc != -EAGAIN) {
2007                                         /* All done with this conn */
2008                                         spin_lock_irqsave(&dev->rad_lock,
2009                                                               flags);
2010                                         cfs_list_del_init(&conn->rac_schedlist);
2011                                         spin_unlock_irqrestore(&dev-> \
2012                                                                    rad_lock,
2013                                                                    flags);
2014
2015                                         kranal_conn_decref(conn);
2016                                         spin_lock_irqsave(&dev->rad_lock,
2017                                                               flags);
2018                                         continue;
2019                                 }
2020
2021                                 /* retry with exponential backoff until HZ */
2022                                 if (conn->rac_keepalive == 0)
2023                                         conn->rac_keepalive = 1;
2024                                 else if (conn->rac_keepalive <= HZ)
2025                                         conn->rac_keepalive *= 2;
2026                                 else
2027                                         conn->rac_keepalive += HZ;
2028
2029                                 deadline = conn->rac_last_tx + conn->rac_keepalive;
2030                                 spin_lock_irqsave(&dev->rad_lock, flags);
2031                         }
2032
2033                         /* Does this conn need attention soonest? */
2034                         if (nsoonest++ == 0 ||
2035                             !cfs_time_aftereq(deadline, soonest))
2036                                 soonest = deadline;
2037                 }
2038
2039                 if (dropped_lock)               /* may sleep iff I didn't drop the lock */
2040                         continue;
2041
2042                 set_current_state(TASK_INTERRUPTIBLE);
2043                 add_wait_queue_exclusive(&dev->rad_waitq, &wait);
2044                 spin_unlock_irqrestore(&dev->rad_lock, flags);
2045
2046                 if (nsoonest == 0) {
2047                         busy_loops = 0;
2048                         waitq_wait(&wait, TASK_INTERRUPTIBLE);
2049                 } else {
2050                         timeout = (long)(soonest - jiffies);
2051                         if (timeout > 0) {
2052                                 busy_loops = 0;
2053                                 waitq_timedwait(&wait,
2054                                                     TASK_INTERRUPTIBLE,
2055                                                     timeout);
2056                         }
2057                 }
2058
2059                 remove_wait_queue(&dev->rad_waitq, &wait);
2060                 set_current_state(TASK_RUNNING);
2061                 spin_lock_irqsave(&dev->rad_lock, flags);
2062         }
2063
2064         spin_unlock_irqrestore(&dev->rad_lock, flags);
2065
2066         dev->rad_scheduler = NULL;
2067         kranal_thread_fini();
2068         return 0;
2069 }