Whamcloud - gitweb
Land from b_hd_pid to HEAD
[fs/lustre-release.git] / lustre / portals / knals / qswnal / qswnal.c
1 /*
2  * Copyright (C) 2002 Cluster File Systems, Inc.
3  *   Author: Eric Barton <eric@bartonsoftware.com>
4  *
5  * Copyright (C) 2002, Lawrence Livermore National Labs (LLNL)
6  * W. Marcus Miller - Based on ksocknal
7  *
8  * This file is part of Portals, http://www.sf.net/projects/lustre/
9  *
10  * Portals is free software; you can redistribute it and/or
11  * modify it under the terms of version 2 of the GNU General Public
12  * License as published by the Free Software Foundation.
13  *
14  * Portals is distributed in the hope that it will be useful,
15  * but WITHOUT ANY WARRANTY; without even the implied warranty of
16  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
17  * GNU General Public License for more details.
18  *
19  * You should have received a copy of the GNU General Public License
20  * along with Portals; if not, write to the Free Software
21  * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
22  *
23  */
24
25 #include "qswnal.h"
26
27 nal_t                   kqswnal_api;
28 kqswnal_data_t          kqswnal_data;
29 ptl_handle_ni_t         kqswnal_ni;
30 kqswnal_tunables_t      kqswnal_tunables;
31
32 kpr_nal_interface_t kqswnal_router_interface = {
33         kprni_nalid:    QSWNAL,
34         kprni_arg:      NULL,
35         kprni_fwd:      kqswnal_fwd_packet,
36         kprni_notify:   NULL,                   /* we're connectionless */
37 };
38
39 #if CONFIG_SYSCTL
40 #define QSWNAL_SYSCTL  201
41
42 #define QSWNAL_SYSCTL_OPTIMIZED_GETS     1
43 #define QSWNAL_SYSCTL_COPY_SMALL_FWD     2
44
45 static ctl_table kqswnal_ctl_table[] = {
46         {QSWNAL_SYSCTL_OPTIMIZED_GETS, "optimized_puts",
47          &kqswnal_tunables.kqn_optimized_puts, sizeof (int),
48          0644, NULL, &proc_dointvec},
49         {QSWNAL_SYSCTL_OPTIMIZED_GETS, "optimized_gets",
50          &kqswnal_tunables.kqn_optimized_gets, sizeof (int),
51          0644, NULL, &proc_dointvec},
52         {0}
53 };
54
55 static ctl_table kqswnal_top_ctl_table[] = {
56         {QSWNAL_SYSCTL, "qswnal", NULL, 0, 0555, kqswnal_ctl_table},
57         {0}
58 };
59 #endif
60
61 int
62 kqswnal_get_tx_desc (struct portals_cfg *pcfg)
63 {
64         unsigned long      flags;
65         struct list_head  *tmp;
66         kqswnal_tx_t      *ktx;
67         int                index = pcfg->pcfg_count;
68         int                rc = -ENOENT;
69
70         spin_lock_irqsave (&kqswnal_data.kqn_idletxd_lock, flags);
71
72         list_for_each (tmp, &kqswnal_data.kqn_activetxds) {
73                 if (index-- != 0)
74                         continue;
75
76                 ktx = list_entry (tmp, kqswnal_tx_t, ktx_list);
77
78                 pcfg->pcfg_pbuf1 = (char *)ktx;
79                 pcfg->pcfg_count = NTOH__u32(ktx->ktx_wire_hdr->type);
80                 pcfg->pcfg_size  = NTOH__u32(ktx->ktx_wire_hdr->payload_length);
81                 pcfg->pcfg_nid   = NTOH__u64(ktx->ktx_wire_hdr->dest_nid);
82                 pcfg->pcfg_nid2  = ktx->ktx_nid;
83                 pcfg->pcfg_misc  = ktx->ktx_launcher;
84                 pcfg->pcfg_flags = (list_empty (&ktx->ktx_delayed_list) ? 0 : 1) |
85                                   (!ktx->ktx_isnblk                    ? 0 : 2) |
86                                   (ktx->ktx_state << 2);
87                 rc = 0;
88                 break;
89         }
90         
91         spin_unlock_irqrestore (&kqswnal_data.kqn_idletxd_lock, flags);
92         return (rc);
93 }
94
95 int
96 kqswnal_cmd (struct portals_cfg *pcfg, void *private)
97 {
98         LASSERT (pcfg != NULL);
99         
100         switch (pcfg->pcfg_command) {
101         case NAL_CMD_GET_TXDESC:
102                 return (kqswnal_get_tx_desc (pcfg));
103
104         case NAL_CMD_REGISTER_MYNID:
105                 CDEBUG (D_IOCTL, "setting NID offset to "LPX64" (was "LPX64")\n",
106                         pcfg->pcfg_nid - kqswnal_data.kqn_elanid,
107                         kqswnal_data.kqn_nid_offset);
108                 kqswnal_data.kqn_nid_offset =
109                         pcfg->pcfg_nid - kqswnal_data.kqn_elanid;
110                 kqswnal_lib.libnal_ni.ni_pid.nid = pcfg->pcfg_nid;
111                 return (0);
112                 
113         default:
114                 return (-EINVAL);
115         }
116 }
117
118 static void
119 kqswnal_shutdown(nal_t *nal)
120 {
121         unsigned long flags;
122         int           do_lib_fini = 0;
123
124         /* NB The first ref was this module! */
125         if (nal->nal_refct != 0) {
126                 PORTAL_MODULE_UNUSE;
127                 return;
128         }
129
130         CDEBUG (D_NET, "shutdown\n");
131         LASSERT (nal == &kqswnal_api);
132
133         switch (kqswnal_data.kqn_init)
134         {
135         default:
136                 LASSERT (0);
137
138         case KQN_INIT_ALL:
139                 libcfs_nal_cmd_unregister(QSWNAL);
140                 /* fall through */
141
142         case KQN_INIT_LIB:
143                 do_lib_fini = 1;
144                 /* fall through */
145
146         case KQN_INIT_DATA:
147                 break;
148
149         case KQN_INIT_NOTHING:
150                 return;
151         }
152
153         /**********************************************************************/
154         /* Tell router we're shutting down.  Any router calls my threads
155          * make will now fail immediately and the router will stop calling
156          * into me. */
157         kpr_shutdown (&kqswnal_data.kqn_router);
158         
159         /**********************************************************************/
160         /* Signal the start of shutdown... */
161         spin_lock_irqsave(&kqswnal_data.kqn_idletxd_lock, flags);
162         kqswnal_data.kqn_shuttingdown = 1;
163         spin_unlock_irqrestore(&kqswnal_data.kqn_idletxd_lock, flags);
164
165         wake_up_all(&kqswnal_data.kqn_idletxd_waitq);
166
167         /**********************************************************************/
168         /* wait for sends that have allocated a tx desc to launch or give up */
169         while (atomic_read (&kqswnal_data.kqn_pending_txs) != 0) {
170                 CDEBUG(D_NET, "waiting for %d pending sends\n",
171                        atomic_read (&kqswnal_data.kqn_pending_txs));
172                 set_current_state (TASK_UNINTERRUPTIBLE);
173                 schedule_timeout (HZ);
174         }
175
176         /**********************************************************************/
177         /* close elan comms */
178 #if MULTIRAIL_EKC
179         /* Shut down receivers first; rx callbacks might try sending... */
180         if (kqswnal_data.kqn_eprx_small != NULL)
181                 ep_free_rcvr (kqswnal_data.kqn_eprx_small);
182
183         if (kqswnal_data.kqn_eprx_large != NULL)
184                 ep_free_rcvr (kqswnal_data.kqn_eprx_large);
185
186         /* NB ep_free_rcvr() returns only after we've freed off all receive
187          * buffers (see shutdown handling in kqswnal_requeue_rx()).  This
188          * means we must have completed any messages we passed to
189          * lib_parse() or kpr_fwd_start(). */
190
191         if (kqswnal_data.kqn_eptx != NULL)
192                 ep_free_xmtr (kqswnal_data.kqn_eptx);
193
194         /* NB ep_free_xmtr() returns only after all outstanding transmits
195          * have called their callback... */
196         LASSERT(list_empty(&kqswnal_data.kqn_activetxds));
197 #else
198         /* "Old" EKC just pretends to shutdown cleanly but actually
199          * provides no guarantees */
200         if (kqswnal_data.kqn_eprx_small != NULL)
201                 ep_remove_large_rcvr (kqswnal_data.kqn_eprx_small);
202
203         if (kqswnal_data.kqn_eprx_large != NULL)
204                 ep_remove_large_rcvr (kqswnal_data.kqn_eprx_large);
205
206         /* wait for transmits to complete */
207         while (!list_empty(&kqswnal_data.kqn_activetxds)) {
208                 CWARN("waiting for active transmits to complete\n");
209                 set_current_state(TASK_UNINTERRUPTIBLE);
210                 schedule_timeout(HZ);
211         }
212
213         if (kqswnal_data.kqn_eptx != NULL)
214                 ep_free_large_xmtr (kqswnal_data.kqn_eptx);
215 #endif
216         /**********************************************************************/
217         /* flag threads to terminate, wake them and wait for them to die */
218         kqswnal_data.kqn_shuttingdown = 2;
219         wake_up_all (&kqswnal_data.kqn_sched_waitq);
220
221         while (atomic_read (&kqswnal_data.kqn_nthreads) != 0) {
222                 CDEBUG(D_NET, "waiting for %d threads to terminate\n",
223                        atomic_read (&kqswnal_data.kqn_nthreads));
224                 set_current_state (TASK_UNINTERRUPTIBLE);
225                 schedule_timeout (HZ);
226         }
227
228         /**********************************************************************/
229         /* No more threads.  No more portals, router or comms callbacks!
230          * I control the horizontals and the verticals...
231          */
232
233 #if MULTIRAIL_EKC
234         LASSERT (list_empty (&kqswnal_data.kqn_readyrxds));
235         LASSERT (list_empty (&kqswnal_data.kqn_delayedtxds));
236         LASSERT (list_empty (&kqswnal_data.kqn_delayedfwds));
237 #endif
238
239         /**********************************************************************/
240         /* Complete any blocked forwarding packets, with error
241          */
242
243         while (!list_empty (&kqswnal_data.kqn_idletxd_fwdq))
244         {
245                 kpr_fwd_desc_t *fwd = list_entry (kqswnal_data.kqn_idletxd_fwdq.next,
246                                                   kpr_fwd_desc_t, kprfd_list);
247                 list_del (&fwd->kprfd_list);
248                 kpr_fwd_done (&kqswnal_data.kqn_router, fwd, -ESHUTDOWN);
249         }
250
251         /**********************************************************************/
252         /* finalise router and portals lib */
253
254         kpr_deregister (&kqswnal_data.kqn_router);
255
256         if (do_lib_fini)
257                 lib_fini (&kqswnal_lib);
258
259         /**********************************************************************/
260         /* Unmap message buffers and free all descriptors and buffers
261          */
262
263 #if MULTIRAIL_EKC
264         /* FTTB, we need to unmap any remaining mapped memory.  When
265          * ep_dvma_release() get fixed (and releases any mappings in the
266          * region), we can delete all the code from here -------->  */
267
268         if (kqswnal_data.kqn_txds != NULL) {
269                 int  i;
270
271                 for (i = 0; i < KQSW_NTXMSGS + KQSW_NNBLK_TXMSGS; i++) {
272                         kqswnal_tx_t *ktx = &kqswnal_data.kqn_txds[i];
273
274                         /* If ktx has a buffer, it got mapped; unmap now.
275                          * NB only the pre-mapped stuff is still mapped
276                          * since all tx descs must be idle */
277
278                         if (ktx->ktx_buffer != NULL)
279                                 ep_dvma_unload(kqswnal_data.kqn_ep,
280                                                kqswnal_data.kqn_ep_tx_nmh,
281                                                &ktx->ktx_ebuffer);
282                 }
283         }
284
285         if (kqswnal_data.kqn_rxds != NULL) {
286                 int   i;
287
288                 for (i = 0; i < KQSW_NRXMSGS_SMALL + KQSW_NRXMSGS_LARGE; i++) {
289                         kqswnal_rx_t *krx = &kqswnal_data.kqn_rxds[i];
290
291                         /* If krx_kiov[0].kiov_page got allocated, it got mapped.  
292                          * NB subsequent pages get merged */
293
294                         if (krx->krx_kiov[0].kiov_page != NULL)
295                                 ep_dvma_unload(kqswnal_data.kqn_ep,
296                                                kqswnal_data.kqn_ep_rx_nmh,
297                                                &krx->krx_elanbuffer);
298                 }
299         }
300         /* <----------- to here */
301
302         if (kqswnal_data.kqn_ep_rx_nmh != NULL)
303                 ep_dvma_release(kqswnal_data.kqn_ep, kqswnal_data.kqn_ep_rx_nmh);
304
305         if (kqswnal_data.kqn_ep_tx_nmh != NULL)
306                 ep_dvma_release(kqswnal_data.kqn_ep, kqswnal_data.kqn_ep_tx_nmh);
307 #else
308         if (kqswnal_data.kqn_eprxdmahandle != NULL)
309         {
310                 elan3_dvma_unload(kqswnal_data.kqn_ep->DmaState,
311                                   kqswnal_data.kqn_eprxdmahandle, 0,
312                                   KQSW_NRXMSGPAGES_SMALL * KQSW_NRXMSGS_SMALL +
313                                   KQSW_NRXMSGPAGES_LARGE * KQSW_NRXMSGS_LARGE);
314
315                 elan3_dma_release(kqswnal_data.kqn_ep->DmaState,
316                                   kqswnal_data.kqn_eprxdmahandle);
317         }
318
319         if (kqswnal_data.kqn_eptxdmahandle != NULL)
320         {
321                 elan3_dvma_unload(kqswnal_data.kqn_ep->DmaState,
322                                   kqswnal_data.kqn_eptxdmahandle, 0,
323                                   KQSW_NTXMSGPAGES * (KQSW_NTXMSGS +
324                                                       KQSW_NNBLK_TXMSGS));
325
326                 elan3_dma_release(kqswnal_data.kqn_ep->DmaState,
327                                   kqswnal_data.kqn_eptxdmahandle);
328         }
329 #endif
330
331         if (kqswnal_data.kqn_txds != NULL)
332         {
333                 int   i;
334
335                 for (i = 0; i < KQSW_NTXMSGS + KQSW_NNBLK_TXMSGS; i++)
336                 {
337                         kqswnal_tx_t *ktx = &kqswnal_data.kqn_txds[i];
338
339                         if (ktx->ktx_buffer != NULL)
340                                 PORTAL_FREE(ktx->ktx_buffer,
341                                             KQSW_TX_BUFFER_SIZE);
342                 }
343
344                 PORTAL_FREE(kqswnal_data.kqn_txds,
345                             sizeof (kqswnal_tx_t) * (KQSW_NTXMSGS +
346                                                      KQSW_NNBLK_TXMSGS));
347         }
348
349         if (kqswnal_data.kqn_rxds != NULL)
350         {
351                 int   i;
352                 int   j;
353
354                 for (i = 0; i < KQSW_NRXMSGS_SMALL + KQSW_NRXMSGS_LARGE; i++)
355                 {
356                         kqswnal_rx_t *krx = &kqswnal_data.kqn_rxds[i];
357
358                         for (j = 0; j < krx->krx_npages; j++)
359                                 if (krx->krx_kiov[j].kiov_page != NULL)
360                                         __free_page (krx->krx_kiov[j].kiov_page);
361                 }
362
363                 PORTAL_FREE(kqswnal_data.kqn_rxds,
364                             sizeof(kqswnal_rx_t) * (KQSW_NRXMSGS_SMALL +
365                                                     KQSW_NRXMSGS_LARGE));
366         }
367
368         /* resets flags, pointers to NULL etc */
369         memset(&kqswnal_data, 0, sizeof (kqswnal_data));
370
371         CDEBUG (D_MALLOC, "done kmem %d\n", atomic_read(&portal_kmemory));
372
373         printk (KERN_INFO "Lustre: Routing QSW NAL unloaded (final mem %d)\n",
374                 atomic_read(&portal_kmemory));
375 }
376
377 static int __init
378 kqswnal_startup (nal_t *nal, ptl_pid_t requested_pid,
379                  ptl_ni_limits_t *requested_limits, 
380                  ptl_ni_limits_t *actual_limits)
381 {
382 #if MULTIRAIL_EKC
383         EP_RAILMASK       all_rails = EP_RAILMASK_ALL;
384 #else
385         ELAN3_DMA_REQUEST dmareq;
386 #endif
387         int               rc;
388         int               i;
389         int               elan_page_idx;
390         ptl_process_id_t  my_process_id;
391         int               pkmem = atomic_read(&portal_kmemory);
392
393         LASSERT (nal == &kqswnal_api);
394
395         if (nal->nal_refct != 0) {
396                 if (actual_limits != NULL)
397                         *actual_limits = kqswnal_lib.libnal_ni.ni_actual_limits;
398                 /* This module got the first ref */
399                 PORTAL_MODULE_USE;
400                 return (PTL_OK);
401         }
402
403         LASSERT (kqswnal_data.kqn_init == KQN_INIT_NOTHING);
404
405         CDEBUG (D_MALLOC, "start kmem %d\n", atomic_read(&portal_kmemory));
406
407         /* ensure all pointers NULL etc */
408         memset (&kqswnal_data, 0, sizeof (kqswnal_data));
409
410         INIT_LIST_HEAD (&kqswnal_data.kqn_idletxds);
411         INIT_LIST_HEAD (&kqswnal_data.kqn_nblk_idletxds);
412         INIT_LIST_HEAD (&kqswnal_data.kqn_activetxds);
413         spin_lock_init (&kqswnal_data.kqn_idletxd_lock);
414         init_waitqueue_head (&kqswnal_data.kqn_idletxd_waitq);
415         INIT_LIST_HEAD (&kqswnal_data.kqn_idletxd_fwdq);
416
417         INIT_LIST_HEAD (&kqswnal_data.kqn_delayedfwds);
418         INIT_LIST_HEAD (&kqswnal_data.kqn_delayedtxds);
419         INIT_LIST_HEAD (&kqswnal_data.kqn_readyrxds);
420
421         spin_lock_init (&kqswnal_data.kqn_sched_lock);
422         init_waitqueue_head (&kqswnal_data.kqn_sched_waitq);
423
424         /* Leave kqn_rpc_success zeroed */
425 #if MULTIRAIL_EKC
426         kqswnal_data.kqn_rpc_failed.Data[0] = -ECONNREFUSED;
427 #else
428         kqswnal_data.kqn_rpc_failed.Status = -ECONNREFUSED;
429 #endif
430
431         /* pointers/lists/locks initialised */
432         kqswnal_data.kqn_init = KQN_INIT_DATA;
433         
434 #if MULTIRAIL_EKC
435         kqswnal_data.kqn_ep = ep_system();
436         if (kqswnal_data.kqn_ep == NULL) {
437                 CERROR("Can't initialise EKC\n");
438                 kqswnal_shutdown(nal);
439                 return (PTL_IFACE_INVALID);
440         }
441
442         if (ep_waitfor_nodeid(kqswnal_data.kqn_ep) == ELAN_INVALID_NODE) {
443                 CERROR("Can't get elan ID\n");
444                 kqswnal_shutdown(nal);
445                 return (PTL_IFACE_INVALID);
446         }
447 #else
448         /**********************************************************************/
449         /* Find the first Elan device */
450
451         kqswnal_data.kqn_ep = ep_device (0);
452         if (kqswnal_data.kqn_ep == NULL)
453         {
454                 CERROR ("Can't get elan device 0\n");
455                 kqswnal_shutdown(nal);
456                 return (PTL_IFACE_INVALID);
457         }
458 #endif
459
460         kqswnal_data.kqn_nid_offset = 0;
461         kqswnal_data.kqn_nnodes     = ep_numnodes (kqswnal_data.kqn_ep);
462         kqswnal_data.kqn_elanid     = ep_nodeid (kqswnal_data.kqn_ep);
463         
464         /**********************************************************************/
465         /* Get the transmitter */
466
467         kqswnal_data.kqn_eptx = ep_alloc_xmtr (kqswnal_data.kqn_ep);
468         if (kqswnal_data.kqn_eptx == NULL)
469         {
470                 CERROR ("Can't allocate transmitter\n");
471                 kqswnal_shutdown (nal);
472                 return (PTL_NO_SPACE);
473         }
474
475         /**********************************************************************/
476         /* Get the receivers */
477
478         kqswnal_data.kqn_eprx_small = ep_alloc_rcvr (kqswnal_data.kqn_ep,
479                                                      EP_MSG_SVC_PORTALS_SMALL,
480                                                      KQSW_EP_ENVELOPES_SMALL);
481         if (kqswnal_data.kqn_eprx_small == NULL)
482         {
483                 CERROR ("Can't install small msg receiver\n");
484                 kqswnal_shutdown (nal);
485                 return (PTL_NO_SPACE);
486         }
487
488         kqswnal_data.kqn_eprx_large = ep_alloc_rcvr (kqswnal_data.kqn_ep,
489                                                      EP_MSG_SVC_PORTALS_LARGE,
490                                                      KQSW_EP_ENVELOPES_LARGE);
491         if (kqswnal_data.kqn_eprx_large == NULL)
492         {
493                 CERROR ("Can't install large msg receiver\n");
494                 kqswnal_shutdown (nal);
495                 return (PTL_NO_SPACE);
496         }
497
498         /**********************************************************************/
499         /* Reserve Elan address space for transmit descriptors NB we may
500          * either send the contents of associated buffers immediately, or
501          * map them for the peer to suck/blow... */
502 #if MULTIRAIL_EKC
503         kqswnal_data.kqn_ep_tx_nmh = 
504                 ep_dvma_reserve(kqswnal_data.kqn_ep,
505                                 KQSW_NTXMSGPAGES*(KQSW_NTXMSGS+KQSW_NNBLK_TXMSGS),
506                                 EP_PERM_WRITE);
507         if (kqswnal_data.kqn_ep_tx_nmh == NULL) {
508                 CERROR("Can't reserve tx dma space\n");
509                 kqswnal_shutdown(nal);
510                 return (PTL_NO_SPACE);
511         }
512 #else
513         dmareq.Waitfn   = DDI_DMA_SLEEP;
514         dmareq.ElanAddr = (E3_Addr) 0;
515         dmareq.Attr     = PTE_LOAD_LITTLE_ENDIAN;
516         dmareq.Perm     = ELAN_PERM_REMOTEWRITE;
517
518         rc = elan3_dma_reserve(kqswnal_data.kqn_ep->DmaState,
519                               KQSW_NTXMSGPAGES*(KQSW_NTXMSGS+KQSW_NNBLK_TXMSGS),
520                               &dmareq, &kqswnal_data.kqn_eptxdmahandle);
521         if (rc != DDI_SUCCESS)
522         {
523                 CERROR ("Can't reserve rx dma space\n");
524                 kqswnal_shutdown (nal);
525                 return (PTL_NO_SPACE);
526         }
527 #endif
528         /**********************************************************************/
529         /* Reserve Elan address space for receive buffers */
530 #if MULTIRAIL_EKC
531         kqswnal_data.kqn_ep_rx_nmh =
532                 ep_dvma_reserve(kqswnal_data.kqn_ep,
533                                 KQSW_NRXMSGPAGES_SMALL * KQSW_NRXMSGS_SMALL +
534                                 KQSW_NRXMSGPAGES_LARGE * KQSW_NRXMSGS_LARGE,
535                                 EP_PERM_WRITE);
536         if (kqswnal_data.kqn_ep_tx_nmh == NULL) {
537                 CERROR("Can't reserve rx dma space\n");
538                 kqswnal_shutdown(nal);
539                 return (PTL_NO_SPACE);
540         }
541 #else
542         dmareq.Waitfn   = DDI_DMA_SLEEP;
543         dmareq.ElanAddr = (E3_Addr) 0;
544         dmareq.Attr     = PTE_LOAD_LITTLE_ENDIAN;
545         dmareq.Perm     = ELAN_PERM_REMOTEWRITE;
546
547         rc = elan3_dma_reserve (kqswnal_data.kqn_ep->DmaState,
548                                 KQSW_NRXMSGPAGES_SMALL * KQSW_NRXMSGS_SMALL +
549                                 KQSW_NRXMSGPAGES_LARGE * KQSW_NRXMSGS_LARGE,
550                                 &dmareq, &kqswnal_data.kqn_eprxdmahandle);
551         if (rc != DDI_SUCCESS)
552         {
553                 CERROR ("Can't reserve rx dma space\n");
554                 kqswnal_shutdown (nal);
555                 return (PTL_NO_SPACE);
556         }
557 #endif
558         /**********************************************************************/
559         /* Allocate/Initialise transmit descriptors */
560
561         PORTAL_ALLOC(kqswnal_data.kqn_txds,
562                      sizeof(kqswnal_tx_t) * (KQSW_NTXMSGS + KQSW_NNBLK_TXMSGS));
563         if (kqswnal_data.kqn_txds == NULL)
564         {
565                 kqswnal_shutdown (nal);
566                 return (PTL_NO_SPACE);
567         }
568
569         /* clear flags, null pointers etc */
570         memset(kqswnal_data.kqn_txds, 0,
571                sizeof(kqswnal_tx_t) * (KQSW_NTXMSGS + KQSW_NNBLK_TXMSGS));
572         for (i = 0; i < (KQSW_NTXMSGS + KQSW_NNBLK_TXMSGS); i++)
573         {
574                 int           premapped_pages;
575                 kqswnal_tx_t *ktx = &kqswnal_data.kqn_txds[i];
576                 int           basepage = i * KQSW_NTXMSGPAGES;
577
578                 PORTAL_ALLOC (ktx->ktx_buffer, KQSW_TX_BUFFER_SIZE);
579                 if (ktx->ktx_buffer == NULL)
580                 {
581                         kqswnal_shutdown (nal);
582                         return (PTL_NO_SPACE);
583                 }
584
585                 /* Map pre-allocated buffer NOW, to save latency on transmit */
586                 premapped_pages = kqswnal_pages_spanned(ktx->ktx_buffer,
587                                                         KQSW_TX_BUFFER_SIZE);
588 #if MULTIRAIL_EKC
589                 ep_dvma_load(kqswnal_data.kqn_ep, NULL, 
590                              ktx->ktx_buffer, KQSW_TX_BUFFER_SIZE, 
591                              kqswnal_data.kqn_ep_tx_nmh, basepage,
592                              &all_rails, &ktx->ktx_ebuffer);
593 #else
594                 elan3_dvma_kaddr_load (kqswnal_data.kqn_ep->DmaState,
595                                        kqswnal_data.kqn_eptxdmahandle,
596                                        ktx->ktx_buffer, KQSW_TX_BUFFER_SIZE,
597                                        basepage, &ktx->ktx_ebuffer);
598 #endif
599                 ktx->ktx_basepage = basepage + premapped_pages; /* message mapping starts here */
600                 ktx->ktx_npages = KQSW_NTXMSGPAGES - premapped_pages; /* for this many pages */
601
602                 INIT_LIST_HEAD (&ktx->ktx_delayed_list);
603
604                 ktx->ktx_state = KTX_IDLE;
605                 ktx->ktx_isnblk = (i >= KQSW_NTXMSGS);
606                 list_add_tail (&ktx->ktx_list, 
607                                ktx->ktx_isnblk ? &kqswnal_data.kqn_nblk_idletxds :
608                                                  &kqswnal_data.kqn_idletxds);
609         }
610
611         /**********************************************************************/
612         /* Allocate/Initialise receive descriptors */
613
614         PORTAL_ALLOC (kqswnal_data.kqn_rxds,
615                       sizeof (kqswnal_rx_t) * (KQSW_NRXMSGS_SMALL + KQSW_NRXMSGS_LARGE));
616         if (kqswnal_data.kqn_rxds == NULL)
617         {
618                 kqswnal_shutdown (nal);
619                 return (PTL_NO_SPACE);
620         }
621
622         memset(kqswnal_data.kqn_rxds, 0, /* clear flags, null pointers etc */
623                sizeof(kqswnal_rx_t) * (KQSW_NRXMSGS_SMALL+KQSW_NRXMSGS_LARGE));
624
625         elan_page_idx = 0;
626         for (i = 0; i < KQSW_NRXMSGS_SMALL + KQSW_NRXMSGS_LARGE; i++)
627         {
628 #if MULTIRAIL_EKC
629                 EP_NMD        elanbuffer;
630 #else
631                 E3_Addr       elanbuffer;
632 #endif
633                 int           j;
634                 kqswnal_rx_t *krx = &kqswnal_data.kqn_rxds[i];
635
636                 if (i < KQSW_NRXMSGS_SMALL)
637                 {
638                         krx->krx_npages = KQSW_NRXMSGPAGES_SMALL;
639                         krx->krx_eprx   = kqswnal_data.kqn_eprx_small;
640                 }
641                 else
642                 {
643                         krx->krx_npages = KQSW_NRXMSGPAGES_LARGE;
644                         krx->krx_eprx   = kqswnal_data.kqn_eprx_large;
645                 }
646
647                 LASSERT (krx->krx_npages > 0);
648                 for (j = 0; j < krx->krx_npages; j++)
649                 {
650                         struct page *page = alloc_page(GFP_KERNEL);
651                         
652                         if (page == NULL) {
653                                 kqswnal_shutdown (nal);
654                                 return (PTL_NO_SPACE);
655                         }
656
657                         krx->krx_kiov[j].kiov_page = page;
658                         LASSERT(page_address(page) != NULL);
659
660 #if MULTIRAIL_EKC
661                         ep_dvma_load(kqswnal_data.kqn_ep, NULL,
662                                      page_address(page),
663                                      PAGE_SIZE, kqswnal_data.kqn_ep_rx_nmh,
664                                      elan_page_idx, &all_rails, &elanbuffer);
665                         
666                         if (j == 0) {
667                                 krx->krx_elanbuffer = elanbuffer;
668                         } else {
669                                 rc = ep_nmd_merge(&krx->krx_elanbuffer,
670                                                   &krx->krx_elanbuffer, 
671                                                   &elanbuffer);
672                                 /* NB contiguous mapping */
673                                 LASSERT(rc);
674                         }
675 #else
676                         elan3_dvma_kaddr_load(kqswnal_data.kqn_ep->DmaState,
677                                               kqswnal_data.kqn_eprxdmahandle,
678                                               page_address(page),
679                                               PAGE_SIZE, elan_page_idx,
680                                               &elanbuffer);
681                         if (j == 0)
682                                 krx->krx_elanbuffer = elanbuffer;
683
684                         /* NB contiguous mapping */
685                         LASSERT (elanbuffer == krx->krx_elanbuffer + j * PAGE_SIZE);
686 #endif
687                         elan_page_idx++;
688
689                 }
690         }
691         LASSERT (elan_page_idx ==
692                  (KQSW_NRXMSGS_SMALL * KQSW_NRXMSGPAGES_SMALL) +
693                  (KQSW_NRXMSGS_LARGE * KQSW_NRXMSGPAGES_LARGE));
694
695         /**********************************************************************/
696         /* Network interface ready to initialise */
697
698         my_process_id.nid = kqswnal_elanid2nid(kqswnal_data.kqn_elanid);
699         my_process_id.pid = requested_pid;
700
701         rc = lib_init(&kqswnal_lib, nal, my_process_id,
702                       requested_limits, actual_limits);
703         if (rc != PTL_OK)
704         {
705                 CERROR ("lib_init failed %d\n", rc);
706                 kqswnal_shutdown (nal);
707                 return (rc);
708         }
709
710         kqswnal_data.kqn_init = KQN_INIT_LIB;
711
712         /**********************************************************************/
713         /* Queue receives, now that it's OK to run their completion callbacks */
714
715         for (i = 0; i < KQSW_NRXMSGS_SMALL + KQSW_NRXMSGS_LARGE; i++)
716         {
717                 kqswnal_rx_t *krx = &kqswnal_data.kqn_rxds[i];
718
719                 /* NB this enqueue can allocate/sleep (attr == 0) */
720                 krx->krx_state = KRX_POSTED;
721 #if MULTIRAIL_EKC
722                 rc = ep_queue_receive(krx->krx_eprx, kqswnal_rxhandler, krx,
723                                       &krx->krx_elanbuffer, 0);
724 #else
725                 rc = ep_queue_receive(krx->krx_eprx, kqswnal_rxhandler, krx,
726                                       krx->krx_elanbuffer,
727                                       krx->krx_npages * PAGE_SIZE, 0);
728 #endif
729                 if (rc != EP_SUCCESS)
730                 {
731                         CERROR ("failed ep_queue_receive %d\n", rc);
732                         kqswnal_shutdown (nal);
733                         return (PTL_FAIL);
734                 }
735         }
736
737         /**********************************************************************/
738         /* Spawn scheduling threads */
739         for (i = 0; i < num_online_cpus(); i++) {
740                 rc = kqswnal_thread_start (kqswnal_scheduler, NULL);
741                 if (rc != 0)
742                 {
743                         CERROR ("failed to spawn scheduling thread: %d\n", rc);
744                         kqswnal_shutdown (nal);
745                         return (PTL_FAIL);
746                 }
747         }
748
749         /**********************************************************************/
750         /* Connect to the router */
751         rc = kpr_register (&kqswnal_data.kqn_router, &kqswnal_router_interface);
752         CDEBUG(D_NET, "Can't initialise routing interface (rc = %d): not routing\n",rc);
753
754         rc = libcfs_nal_cmd_register (QSWNAL, &kqswnal_cmd, NULL);
755         if (rc != 0) {
756                 CERROR ("Can't initialise command interface (rc = %d)\n", rc);
757                 kqswnal_shutdown (nal);
758                 return (PTL_FAIL);
759         }
760
761         kqswnal_data.kqn_init = KQN_INIT_ALL;
762
763         printk(KERN_INFO "Lustre: Routing QSW NAL loaded on node %d of %d "
764                "(Routing %s, initial mem %d)\n", 
765                kqswnal_data.kqn_elanid, kqswnal_data.kqn_nnodes,
766                kpr_routing (&kqswnal_data.kqn_router) ? "enabled" : "disabled",
767                pkmem);
768
769         return (PTL_OK);
770 }
771
772 void __exit
773 kqswnal_finalise (void)
774 {
775 #if CONFIG_SYSCTL
776         if (kqswnal_tunables.kqn_sysctl != NULL)
777                 unregister_sysctl_table (kqswnal_tunables.kqn_sysctl);
778 #endif
779         PtlNIFini(kqswnal_ni);
780
781         ptl_unregister_nal(QSWNAL);
782 }
783
784 static int __init
785 kqswnal_initialise (void)
786 {
787         int   rc;
788
789         kqswnal_api.nal_ni_init = kqswnal_startup;
790         kqswnal_api.nal_ni_fini = kqswnal_shutdown;
791
792         /* Initialise dynamic tunables to defaults once only */
793         kqswnal_tunables.kqn_optimized_puts = KQSW_OPTIMIZED_PUTS;
794         kqswnal_tunables.kqn_optimized_gets = KQSW_OPTIMIZED_GETS;
795         
796         rc = ptl_register_nal(QSWNAL, &kqswnal_api);
797         if (rc != PTL_OK) {
798                 CERROR("Can't register QSWNAL: %d\n", rc);
799                 return (-ENOMEM);               /* or something... */
800         }
801
802         /* Pure gateways, and the workaround for 'EKC blocks forever until
803          * the service is active' want the NAL started up at module load
804          * time... */
805         rc = PtlNIInit(QSWNAL, LUSTRE_SRV_PTL_PID, NULL, NULL, &kqswnal_ni);
806         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
807                 ptl_unregister_nal(QSWNAL);
808                 return (-ENODEV);
809         }
810
811 #if CONFIG_SYSCTL
812         /* Press on regardless even if registering sysctl doesn't work */
813         kqswnal_tunables.kqn_sysctl = 
814                 register_sysctl_table (kqswnal_top_ctl_table, 0);
815 #endif
816         return (0);
817 }
818
819 MODULE_AUTHOR("Cluster File Systems, Inc. <info@clusterfs.com>");
820 MODULE_DESCRIPTION("Kernel Quadrics/Elan NAL v1.01");
821 MODULE_LICENSE("GPL");
822
823 module_init (kqswnal_initialise);
824 module_exit (kqswnal_finalise);