Whamcloud - gitweb
b=5498
[fs/lustre-release.git] / lnet / ulnds / ptllnd / ptllnd.c
1
2 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
3  * vim:expandtab:shiftwidth=8:tabstop=8:
4  *
5  * Copyright (C) 2005 Cluster File Systems, Inc. All rights reserved.
6  *   Author: Eric Barton <eeb@bartonsoftware.com>
7  *
8  *   This file is part of the Lustre file system, http://www.lustre.org
9  *   Lustre is a trademark of Cluster File Systems, Inc.
10  *
11  *   This file is confidential source code owned by Cluster File Systems.
12  *   No viewing, modification, compilation, redistribution, or any other
13  *   form of use is permitted except through a signed license agreement.
14  *
15  *   If you have not signed such an agreement, then you have no rights to
16  *   this file.  Please destroy it immediately and contact CFS.
17  *
18  */
19
20 #include "ptllnd.h"
21
22 lnd_t               the_ptllnd = {
23         .lnd_type       = PTLLND,
24         .lnd_startup    = ptllnd_startup,
25         .lnd_shutdown   = ptllnd_shutdown,
26         .lnd_ctl        = ptllnd_ctl,
27         .lnd_send       = ptllnd_send,
28         .lnd_recv       = ptllnd_recv,
29         .lnd_eager_recv = ptllnd_eager_recv,
30         .lnd_notify     = ptllnd_notify,
31         .lnd_wait       = ptllnd_wait,
32         .lnd_setasync   = ptllnd_setasync,
33 };
34
35 static int ptllnd_ni_count = 0;
36
37 static struct list_head ptllnd_idle_history;
38 static struct list_head ptllnd_history_list;
39
40 void
41 ptllnd_history_fini(void)
42 {
43         ptllnd_he_t *he;
44
45         while (!list_empty(&ptllnd_idle_history)) {
46                 he = list_entry(ptllnd_idle_history.next,
47                                 ptllnd_he_t, he_list);
48                 
49                 list_del(&he->he_list);
50                 LIBCFS_FREE(he, sizeof(*he));
51         }
52         
53         while (!list_empty(&ptllnd_history_list)) {
54                 he = list_entry(ptllnd_history_list.next,
55                                 ptllnd_he_t, he_list);
56                 
57                 list_del(&he->he_list);
58                 LIBCFS_FREE(he, sizeof(*he));
59         }
60 }
61
62 int
63 ptllnd_history_init(void)
64 {
65         int          i;
66         ptllnd_he_t *he;
67         int          n;
68         int          rc;
69         
70         CFS_INIT_LIST_HEAD(&ptllnd_idle_history);
71         CFS_INIT_LIST_HEAD(&ptllnd_history_list);
72         
73         rc = ptllnd_parse_int_tunable(&n, "PTLLND_HISTORY", 0);
74         if (rc != 0)
75                 return rc;
76         
77         for (i = 0; i < n; i++) {
78                 LIBCFS_ALLOC(he, sizeof(*he));
79                 if (he == NULL) {
80                         ptllnd_history_fini();
81                         return -ENOMEM;
82                 }
83                 
84                 list_add(&he->he_list, &ptllnd_idle_history);
85         }
86
87         PTLLND_HISTORY("Init");
88
89         return 0;
90 }
91
92 void
93 ptllnd_history(const char *fn, const char *file, const int line,
94                const char *fmt, ...)
95 {
96         static int     seq;
97         
98         va_list        ap;
99         ptllnd_he_t   *he;
100         
101         if (!list_empty(&ptllnd_idle_history)) {
102                 he = list_entry(ptllnd_idle_history.next,
103                                 ptllnd_he_t, he_list);
104         } else if (!list_empty(&ptllnd_history_list)) {
105                 he = list_entry(ptllnd_history_list.next,
106                                 ptllnd_he_t, he_list);
107         } else {
108                 return;
109         }
110
111         list_del(&he->he_list);
112         list_add_tail(&he->he_list, &ptllnd_history_list);
113
114         he->he_seq = seq++;
115         he->he_fn = fn;
116         he->he_file = file;
117         he->he_line = line;
118         gettimeofday(&he->he_time, NULL);
119         
120         va_start(ap, fmt);
121         vsnprintf(he->he_msg, sizeof(he->he_msg), fmt, ap);
122         va_end(ap);
123 }
124
125 void
126 ptllnd_dump_history(void)
127 {
128         ptllnd_he_t    *he;
129
130         PTLLND_HISTORY("dumping...");
131         
132         while (!list_empty(&ptllnd_history_list)) {
133                 he = list_entry(ptllnd_history_list.next,
134                                 ptllnd_he_t, he_list);
135
136                 list_del(&he->he_list);
137                 
138                 CDEBUG(D_WARNING, "%d %d.%06d (%s:%d:%s()) %s\n", he->he_seq,
139                        (int)he->he_time.tv_sec, (int)he->he_time.tv_usec,
140                        he->he_file, he->he_line, he->he_fn, he->he_msg);
141
142                 list_add_tail(&he->he_list, &ptllnd_idle_history);
143         }
144
145         PTLLND_HISTORY("complete");
146 }
147
148 void 
149 ptllnd_assert_wire_constants (void)
150 {
151         /* Wire protocol assertions generated by 'wirecheck'
152          * running on Linux fedora 2.6.11-co-0.6.4 #1 Mon Jun 19 05:36:13 UTC 2006 i686 i686 i386 GNU
153          * with gcc version 4.1.1 20060525 (Red Hat 4.1.1-1) */
154
155
156         /* Constants... */
157         CLASSERT (PTL_RESERVED_MATCHBITS == 0x100);
158         CLASSERT (LNET_MSG_MATCHBITS == 0);
159         CLASSERT (PTLLND_MSG_MAGIC == 0x50746C4E);
160         CLASSERT (PTLLND_MSG_VERSION == 0x04);
161         CLASSERT (PTLLND_RDMA_OK == 0x00);
162         CLASSERT (PTLLND_RDMA_FAIL == 0x01);
163         CLASSERT (PTLLND_MSG_TYPE_INVALID == 0x00);
164         CLASSERT (PTLLND_MSG_TYPE_PUT == 0x01);
165         CLASSERT (PTLLND_MSG_TYPE_GET == 0x02);
166         CLASSERT (PTLLND_MSG_TYPE_IMMEDIATE == 0x03);
167         CLASSERT (PTLLND_MSG_TYPE_NOOP == 0x04);
168         CLASSERT (PTLLND_MSG_TYPE_HELLO == 0x05);
169         CLASSERT (PTLLND_MSG_TYPE_NAK == 0x06);
170
171         /* Checks for struct kptl_msg_t */
172         CLASSERT ((int)sizeof(kptl_msg_t) == 136);
173         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_magic) == 0);
174         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_magic) == 4);
175         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_version) == 4);
176         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_version) == 2);
177         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_type) == 6);
178         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_type) == 1);
179         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_credits) == 7);
180         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_credits) == 1);
181         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_nob) == 8);
182         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_nob) == 4);
183         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_cksum) == 12);
184         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_cksum) == 4);
185         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_srcnid) == 16);
186         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_srcnid) == 8);
187         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_srcstamp) == 24);
188         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_srcstamp) == 8);
189         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_dstnid) == 32);
190         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_dstnid) == 8);
191         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_dststamp) == 40);
192         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_dststamp) == 8);
193         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_srcpid) == 48);
194         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_srcpid) == 4);
195         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_dstpid) == 52);
196         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_dstpid) == 4);
197         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_u.immediate) == 56);
198         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_u.immediate) == 72);
199         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_u.rdma) == 56);
200         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_u.rdma) == 80);
201         CLASSERT ((int)offsetof(kptl_msg_t, ptlm_u.hello) == 56);
202         CLASSERT ((int)sizeof(((kptl_msg_t *)0)->ptlm_u.hello) == 12);
203
204         /* Checks for struct kptl_immediate_msg_t */
205         CLASSERT ((int)sizeof(kptl_immediate_msg_t) == 72);
206         CLASSERT ((int)offsetof(kptl_immediate_msg_t, kptlim_hdr) == 0);
207         CLASSERT ((int)sizeof(((kptl_immediate_msg_t *)0)->kptlim_hdr) == 72);
208         CLASSERT ((int)offsetof(kptl_immediate_msg_t, kptlim_payload[13]) == 85);
209         CLASSERT ((int)sizeof(((kptl_immediate_msg_t *)0)->kptlim_payload[13]) == 1);
210
211         /* Checks for struct kptl_rdma_msg_t */
212         CLASSERT ((int)sizeof(kptl_rdma_msg_t) == 80);
213         CLASSERT ((int)offsetof(kptl_rdma_msg_t, kptlrm_hdr) == 0);
214         CLASSERT ((int)sizeof(((kptl_rdma_msg_t *)0)->kptlrm_hdr) == 72);
215         CLASSERT ((int)offsetof(kptl_rdma_msg_t, kptlrm_matchbits) == 72);
216         CLASSERT ((int)sizeof(((kptl_rdma_msg_t *)0)->kptlrm_matchbits) == 8);
217
218         /* Checks for struct kptl_hello_msg_t */
219         CLASSERT ((int)sizeof(kptl_hello_msg_t) == 12);
220         CLASSERT ((int)offsetof(kptl_hello_msg_t, kptlhm_matchbits) == 0);
221         CLASSERT ((int)sizeof(((kptl_hello_msg_t *)0)->kptlhm_matchbits) == 8);
222         CLASSERT ((int)offsetof(kptl_hello_msg_t, kptlhm_max_msg_size) == 8);
223         CLASSERT ((int)sizeof(((kptl_hello_msg_t *)0)->kptlhm_max_msg_size) == 4);
224 }
225
226 int
227 ptllnd_parse_int_tunable(int *value, char *name, int dflt)
228 {
229         char    *env = getenv(name);
230         char    *end;
231
232         if (env == NULL) {
233                 *value = dflt;
234                 return 0;
235         }
236
237         *value = strtoull(env, &end, 0);
238         if (*end == 0)
239                 return 0;
240
241         CERROR("Can't parse tunable %s=%s\n", name, env);
242         return -EINVAL;
243 }
244
245 int
246 ptllnd_get_tunables(lnet_ni_t *ni)
247 {
248         ptllnd_ni_t *plni = ni->ni_data;
249         int          max_msg_size;
250         int          msgs_per_buffer;
251         int          rc;
252         int          temp;
253
254         /*  Other tunable defaults depend on this */
255         rc = ptllnd_parse_int_tunable(&plni->plni_debug, "PTLLND_DEBUG", 0);
256         if (rc != 0)
257                 return rc;
258
259         rc = ptllnd_parse_int_tunable(&plni->plni_portal,
260                                       "PTLLND_PORTAL", PTLLND_PORTAL);
261         if (rc != 0)
262                 return rc;
263
264         rc = ptllnd_parse_int_tunable(&temp,
265                                       "PTLLND_PID", PTLLND_PID);
266         if (rc != 0)
267                 return rc;
268         plni->plni_ptllnd_pid = (ptl_pid_t)temp;
269
270         rc = ptllnd_parse_int_tunable(&plni->plni_peer_credits,
271                                       "PTLLND_PEERCREDITS", PTLLND_PEERCREDITS);
272         if (rc != 0)
273                 return rc;
274
275         rc = ptllnd_parse_int_tunable(&max_msg_size,
276                                       "PTLLND_MAX_MSG_SIZE",
277                                       PTLLND_MAX_ULND_MSG_SIZE);
278         if (rc != 0)
279                 return rc;
280
281         rc = ptllnd_parse_int_tunable(&msgs_per_buffer,
282                                       "PTLLND_MSGS_PER_BUFFER", 64);
283         if (rc != 0)
284                 return rc;
285
286         rc = ptllnd_parse_int_tunable(&plni->plni_msgs_spare,
287                                       "PTLLND_MSGS_SPARE", 256);
288         if (rc != 0)
289                 return rc;
290
291         rc = ptllnd_parse_int_tunable(&plni->plni_peer_hash_size,
292                                       "PTLLND_PEER_HASH_SIZE", 101);
293         if (rc != 0)
294                 return rc;
295
296
297         rc = ptllnd_parse_int_tunable(&plni->plni_eq_size,
298                                       "PTLLND_EQ_SIZE", 1024);
299         if (rc != 0)
300                 return rc;
301
302         rc = ptllnd_parse_int_tunable(&plni->plni_checksum,
303                                       "PTLLND_CHECKSUM", 0);
304         if (rc != 0)
305                 return rc;
306
307         rc = ptllnd_parse_int_tunable(&plni->plni_max_tx_history,
308                                       "PTLLND_TX_HISTORY",
309                                       plni->plni_debug ? 1024 : 0);
310         if (rc != 0)
311                 return rc;
312
313         rc = ptllnd_parse_int_tunable(&plni->plni_abort_on_protocol_mismatch,
314                                       "PTLLND_ABORT_ON_PROTOCOL_MISMATCH", 1);
315         if (rc != 0)
316                 return rc;
317
318         rc = ptllnd_parse_int_tunable(&plni->plni_abort_on_nak,
319                                       "PTLLND_ABORT_ON_NAK", 0);
320         if (rc != 0)
321                 return rc;
322
323         rc = ptllnd_parse_int_tunable(&plni->plni_dump_on_nak,
324                                       "PTLLND_DUMP_ON_NAK", plni->plni_debug);
325         if (rc != 0)
326                 return rc;
327
328         rc = ptllnd_parse_int_tunable(&plni->plni_watchdog_interval,
329                                       "PTLLND_WATCHDOG_INTERVAL", 1);
330         if (rc != 0)
331                 return rc;
332         if (plni->plni_watchdog_interval <= 0)
333                 plni->plni_watchdog_interval = 1;
334
335         rc = ptllnd_parse_int_tunable(&plni->plni_timeout,
336                                       "PTLLND_TIMEOUT", 50);
337         if (rc != 0)
338                 return rc;
339
340         rc = ptllnd_parse_int_tunable(&plni->plni_long_wait,
341                                       "PTLLND_LONG_WAIT",
342                                       plni->plni_debug ? 5 : plni->plni_timeout);
343         if (rc != 0)
344                 return rc;
345         plni->plni_long_wait *= 1000;           /* convert to mS */
346
347         plni->plni_max_msg_size = max_msg_size & ~7;
348         if (plni->plni_max_msg_size < PTLLND_MIN_BUFFER_SIZE)
349                 plni->plni_max_msg_size = PTLLND_MIN_BUFFER_SIZE;
350         CLASSERT ((PTLLND_MIN_BUFFER_SIZE & 7) == 0);
351         CLASSERT (sizeof(kptl_msg_t) <= PTLLND_MIN_BUFFER_SIZE);
352
353         plni->plni_buffer_size = plni->plni_max_msg_size * msgs_per_buffer;
354
355         CDEBUG(D_NET, "portal          = %d\n",plni->plni_portal);
356         CDEBUG(D_NET, "ptllnd_pid      = %d\n",plni->plni_ptllnd_pid);
357         CDEBUG(D_NET, "max_msg_size    = %d\n",max_msg_size);
358         CDEBUG(D_NET, "msgs_per_buffer = %d\n",msgs_per_buffer);
359         CDEBUG(D_NET, "msgs_spare      = %d\n",plni->plni_msgs_spare);
360         CDEBUG(D_NET, "peer_hash_size  = %d\n",plni->plni_peer_hash_size);
361         CDEBUG(D_NET, "eq_size         = %d\n",plni->plni_eq_size);
362         CDEBUG(D_NET, "max_msg_size    = %d\n",plni->plni_max_msg_size);
363         CDEBUG(D_NET, "buffer_size     = %d\n",plni->plni_buffer_size);
364
365         return 0;
366 }
367
368 ptllnd_buffer_t *
369 ptllnd_create_buffer (lnet_ni_t *ni)
370 {
371         ptllnd_ni_t     *plni = ni->ni_data;
372         ptllnd_buffer_t *buf;
373
374         LIBCFS_ALLOC(buf, sizeof(*buf));
375         if (buf == NULL) {
376                 CERROR("Can't allocate buffer descriptor\n");
377                 return NULL;
378         }
379
380         buf->plb_ni = ni;
381         buf->plb_posted = 0;
382         CFS_INIT_LIST_HEAD(&buf->plb_list);
383
384         LIBCFS_ALLOC(buf->plb_buffer, plni->plni_buffer_size);
385         if (buf->plb_buffer == NULL) {
386                 CERROR("Can't allocate buffer size %d\n",
387                        plni->plni_buffer_size);
388                 LIBCFS_FREE(buf, sizeof(*buf));
389                 return NULL;
390         }
391
392         list_add(&buf->plb_list, &plni->plni_buffers);
393         plni->plni_nbuffers++;
394
395         return buf;
396 }
397
398 void
399 ptllnd_destroy_buffer (ptllnd_buffer_t *buf)
400 {
401         ptllnd_ni_t     *plni = buf->plb_ni->ni_data;
402
403         LASSERT (!buf->plb_posted);
404
405         plni->plni_nbuffers--;
406         list_del(&buf->plb_list);
407         LIBCFS_FREE(buf->plb_buffer, plni->plni_buffer_size);
408         LIBCFS_FREE(buf, sizeof(*buf));
409 }
410
411 int
412 ptllnd_size_buffers (lnet_ni_t *ni, int delta)
413 {
414         ptllnd_ni_t     *plni = ni->ni_data;
415         ptllnd_buffer_t *buf;
416         int              nmsgs;
417         int              nbufs;
418         int              rc;
419
420         CDEBUG(D_NET, "nposted_buffers = %d (before)\n",plni->plni_nposted_buffers);
421         CDEBUG(D_NET, "nbuffers = %d (before)\n",plni->plni_nbuffers);
422
423         plni->plni_nmsgs += delta;
424         LASSERT(plni->plni_nmsgs >= 0);
425         
426         nmsgs = plni->plni_nmsgs + plni->plni_msgs_spare;
427
428         nbufs = (nmsgs * plni->plni_max_msg_size + plni->plni_buffer_size - 1) /
429                 plni->plni_buffer_size;
430
431         while (nbufs > plni->plni_nbuffers) {
432                 buf = ptllnd_create_buffer(ni);
433
434                 if (buf == NULL)
435                         return -ENOMEM;
436
437                 rc = ptllnd_post_buffer(buf);
438                 if (rc != 0) {
439                         /* TODO - this path seems to orpahn the buffer
440                          * in a state where its not posted and will never be
441                          * However it does not leak the buffer as it's
442                          * already been put onto the global buffer list
443                          * and will be cleaned up
444                          */
445                         return rc;
446                 }
447         }
448
449         CDEBUG(D_NET, "nposted_buffers = %d (after)\n",plni->plni_nposted_buffers);
450         CDEBUG(D_NET, "nbuffers = %d (after)\n",plni->plni_nbuffers);
451         return 0;
452 }
453
454 void
455 ptllnd_destroy_buffers (lnet_ni_t *ni)
456 {
457         ptllnd_ni_t       *plni = ni->ni_data;
458         ptllnd_buffer_t   *buf;
459         struct list_head  *tmp;
460         struct list_head  *nxt;
461
462         CDEBUG(D_NET, "nposted_buffers = %d (before)\n",plni->plni_nposted_buffers);
463         CDEBUG(D_NET, "nbuffers = %d (before)\n",plni->plni_nbuffers);
464
465         list_for_each_safe(tmp, nxt, &plni->plni_buffers) {
466                 buf = list_entry(tmp, ptllnd_buffer_t, plb_list);
467
468                 //CDEBUG(D_NET, "buf=%p posted=%d\n",buf,buf->plb_posted);
469
470                 LASSERT (plni->plni_nbuffers > 0);
471                 if (buf->plb_posted) {
472                         time_t   start = cfs_time_current_sec();
473                         int      w = plni->plni_long_wait;
474
475                         LASSERT (plni->plni_nposted_buffers > 0);
476
477 #ifdef LUSTRE_PORTALS_UNLINK_SEMANTICS
478                         (void) PtlMDUnlink(buf->plb_md);
479
480                         while (buf->plb_posted) {
481                                 if (w > 0 && cfs_time_current_sec() > start + w/1000) {
482                                         CWARN("Waited %ds to unlink buffer\n",
483                                               (int)(cfs_time_current_sec() - start));
484                                         w *= 2;
485                                 }
486                                 ptllnd_wait(ni, w);
487                         }
488 #else
489                         while (buf->plb_posted) {
490                                 rc = PtlMDUnlink(buf->plb_md);
491                                 if (rc == PTL_OK) {
492                                         buf->plb_posted = 0;
493                                         plni->plni_nposted_buffers--;
494                                         break;
495                                 }
496                                 LASSERT (rc == PTL_MD_IN_USE);
497                                 if (w > 0 && cfs_time_current_sec() > start + w/1000) {
498                                         CWARN("Waited %ds to unlink buffer\n",
499                                               cfs_time_current_sec() - start);
500                                         w *= 2;
501                                 }
502                                 ptllnd_wait(ni, w);
503                         }
504 #endif
505                 }
506                 ptllnd_destroy_buffer(buf);
507         }
508
509         CDEBUG(D_NET, "nposted_buffers = %d (after)\n",plni->plni_nposted_buffers);
510         CDEBUG(D_NET, "nbuffers = %d (after)\n",plni->plni_nbuffers);
511
512         LASSERT (plni->plni_nposted_buffers == 0);
513         LASSERT (plni->plni_nbuffers == 0);
514 }
515
516 int
517 ptllnd_create_peer_hash (lnet_ni_t *ni)
518 {
519         ptllnd_ni_t *plni = ni->ni_data;
520         int          i;
521
522         plni->plni_npeers = 0;
523
524         LIBCFS_ALLOC(plni->plni_peer_hash,
525                      plni->plni_peer_hash_size * sizeof(*plni->plni_peer_hash));
526         if (plni->plni_peer_hash == NULL) {
527                 CERROR("Can't allocate ptllnd peer hash (size %d)\n",
528                        plni->plni_peer_hash_size);
529                 return -ENOMEM;
530         }
531
532         for (i = 0; i < plni->plni_peer_hash_size; i++)
533                 CFS_INIT_LIST_HEAD(&plni->plni_peer_hash[i]);
534
535         return 0;
536 }
537
538 void
539 ptllnd_destroy_peer_hash (lnet_ni_t *ni)
540 {
541         ptllnd_ni_t    *plni = ni->ni_data;
542         int             i;
543
544         LASSERT( plni->plni_npeers == 0);
545
546         for (i = 0; i < plni->plni_peer_hash_size; i++)
547                 LASSERT (list_empty(&plni->plni_peer_hash[i]));
548
549         LIBCFS_FREE(plni->plni_peer_hash,
550                     plni->plni_peer_hash_size * sizeof(*plni->plni_peer_hash));
551 }
552
553 void
554 ptllnd_close_peers (lnet_ni_t *ni)
555 {
556         ptllnd_ni_t    *plni = ni->ni_data;
557         ptllnd_peer_t  *plp;
558         int             i;
559
560         for (i = 0; i < plni->plni_peer_hash_size; i++)
561                 while (!list_empty(&plni->plni_peer_hash[i])) {
562                         plp = list_entry(plni->plni_peer_hash[i].next,
563                                          ptllnd_peer_t, plp_list);
564
565                         ptllnd_close_peer(plp, 0);
566                 }
567 }
568
569 int
570 ptllnd_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
571 {
572         switch (cmd) {
573         case IOC_LIBCFS_DEBUG_PEER:
574                 ptllnd_dump_debug(ni, *((lnet_process_id_t *)arg));
575                 return 0;
576                 
577         default:
578                 return -EINVAL;
579         }
580 }
581
582 __u64
583 ptllnd_get_timestamp(void)
584 {
585         struct timeval  tv;
586         int             rc = gettimeofday(&tv, NULL);
587
588         LASSERT (rc == 0);
589         return ((__u64)tv.tv_sec) * 1000000 + tv.tv_usec;
590 }
591
592 void
593 ptllnd_shutdown (lnet_ni_t *ni)
594 {
595         ptllnd_ni_t *plni = ni->ni_data;
596         int          rc;
597         time_t       start = cfs_time_current_sec();
598         int          w = plni->plni_long_wait;
599
600         LASSERT (ptllnd_ni_count == 1);
601         plni->plni_max_tx_history = 0;
602
603         ptllnd_cull_tx_history(plni);
604
605         ptllnd_close_peers(ni);
606         ptllnd_destroy_buffers(ni);
607
608         while (plni->plni_npeers > 0) {
609                 if (w > 0 && cfs_time_current_sec() > start + w/1000) {
610                         CWARN("Waited %ds for peers to shutdown\n",
611                               (int)(cfs_time_current_sec() - start));
612                         w *= 2;
613                 }
614                 ptllnd_wait(ni, w);
615         }
616
617         LASSERT (plni->plni_ntxs == 0);
618         LASSERT (plni->plni_nrxs == 0);
619
620         rc = PtlEQFree(plni->plni_eqh);
621         LASSERT (rc == PTL_OK);
622
623         rc = PtlNIFini(plni->plni_nih);
624         LASSERT (rc == PTL_OK);
625
626         ptllnd_destroy_peer_hash(ni);
627         LIBCFS_FREE(plni, sizeof(*plni));
628         ptllnd_ni_count--;
629 }
630
631 int
632 ptllnd_startup (lnet_ni_t *ni)
633 {
634         ptllnd_ni_t *plni;
635         int          rc;
636
637         /* could get limits from portals I guess... */
638         ni->ni_maxtxcredits =
639         ni->ni_peertxcredits = 1000;
640
641         if (ptllnd_ni_count != 0) {
642                 CERROR("Can't have > 1 instance of ptllnd\n");
643                 return -EPERM;
644         }
645
646         ptllnd_ni_count++;
647
648         rc = ptllnd_history_init();
649         if (rc != 0) {
650                 CERROR("Can't init history\n");
651                 goto failed0;
652         }
653         
654         LIBCFS_ALLOC(plni, sizeof(*plni));
655         if (plni == NULL) {
656                 CERROR("Can't allocate ptllnd state\n");
657                 rc = -ENOMEM;
658                 goto failed0;
659         }
660
661         ni->ni_data = plni;
662
663         plni->plni_stamp = ptllnd_get_timestamp();
664         plni->plni_nrxs = 0;
665         plni->plni_ntxs = 0;
666         plni->plni_ntx_history = 0;
667         plni->plni_watchdog_peeridx = 0;
668         plni->plni_watchdog_nextt = cfs_time_current_sec();
669         CFS_INIT_LIST_HEAD(&plni->plni_zombie_txs);
670         CFS_INIT_LIST_HEAD(&plni->plni_tx_history);
671
672         /*
673          *  Initilize buffer related data structures
674          */
675         CFS_INIT_LIST_HEAD(&plni->plni_buffers);
676         plni->plni_nbuffers = 0;
677         plni->plni_nposted_buffers = 0;
678
679         rc = ptllnd_get_tunables(ni);
680         if (rc != 0)
681                 goto failed1;
682
683         rc = ptllnd_create_peer_hash(ni);
684         if (rc != 0)
685                 goto failed1;
686
687         /* NB I most probably won't get the PID I requested here.  It doesn't
688          * matter because I don't need a fixed PID (only connection acceptors
689          * need a "well known" PID). */
690
691         rc = PtlNIInit(PTL_IFACE_DEFAULT, plni->plni_ptllnd_pid,
692                        NULL, NULL, &plni->plni_nih);
693         if (rc != PTL_OK && rc != PTL_IFACE_DUP) {
694                 CERROR("PtlNIInit failed: %s(%d)\n",
695                        ptllnd_errtype2str(rc), rc);
696                 rc = -ENODEV;
697                 goto failed2;
698         }
699
700         rc = PtlEQAlloc(plni->plni_nih, plni->plni_eq_size,
701                         PTL_EQ_HANDLER_NONE, &plni->plni_eqh);
702         if (rc != PTL_OK) {
703                 CERROR("PtlEQAlloc failed: %s(%d)\n",
704                        ptllnd_errtype2str(rc), rc);
705                 rc = -ENODEV;
706                 goto failed3;
707         }
708
709         /*
710          * Fetch the Portals NID
711          */
712         rc = PtlGetId(plni->plni_nih, &plni->plni_portals_id);
713         if (rc != PTL_OK) {
714                 CERROR ("PtlGetID failed : %s(%d)\n",
715                         ptllnd_errtype2str(rc), rc);
716                 rc = -EINVAL;
717                 goto failed4;
718         }
719
720         /*
721          * Create the new NID.  Based on the LND network type
722          * and the lower ni's address data.
723          */
724         ni->ni_nid = ptllnd_ptl2lnetnid(ni, plni->plni_portals_id.nid);
725
726         CDEBUG(D_NET, "ptl id  =%s\n", ptllnd_ptlid2str(plni->plni_portals_id));
727         CDEBUG(D_NET, "lnet id =%s (passed back)\n",
728                libcfs_id2str((lnet_process_id_t) {
729                        .nid = ni->ni_nid, .pid = the_lnet.ln_pid}));
730
731         rc = ptllnd_size_buffers(ni, 0);
732         if (rc != 0)
733                 goto failed4;
734
735         return 0;
736
737  failed4:
738         ptllnd_destroy_buffers(ni);
739         PtlEQFree(plni->plni_eqh);
740  failed3:
741         PtlNIFini(plni->plni_nih);
742  failed2:
743         ptllnd_destroy_peer_hash(ni);
744  failed1:
745         LIBCFS_FREE(plni, sizeof(*plni));
746  failed0:
747         ptllnd_history_fini();
748         ptllnd_ni_count--;
749         CDEBUG(D_NET, "<<< rc=%d\n",rc);
750         return rc;
751 }
752
753 const char *ptllnd_evtype2str(int type)
754 {
755 #define DO_TYPE(x) case x: return #x;
756         switch(type)
757         {
758                 DO_TYPE(PTL_EVENT_GET_START);
759                 DO_TYPE(PTL_EVENT_GET_END);
760                 DO_TYPE(PTL_EVENT_PUT_START);
761                 DO_TYPE(PTL_EVENT_PUT_END);
762                 DO_TYPE(PTL_EVENT_REPLY_START);
763                 DO_TYPE(PTL_EVENT_REPLY_END);
764                 DO_TYPE(PTL_EVENT_ACK);
765                 DO_TYPE(PTL_EVENT_SEND_START);
766                 DO_TYPE(PTL_EVENT_SEND_END);
767                 DO_TYPE(PTL_EVENT_UNLINK);
768         default:
769                 return "<unknown event type>";
770         }
771 #undef DO_TYPE
772 }
773
774 const char *ptllnd_msgtype2str(int type)
775 {
776 #define DO_TYPE(x) case x: return #x;
777         switch(type)
778         {
779                 DO_TYPE(PTLLND_MSG_TYPE_INVALID);
780                 DO_TYPE(PTLLND_MSG_TYPE_PUT);
781                 DO_TYPE(PTLLND_MSG_TYPE_GET);
782                 DO_TYPE(PTLLND_MSG_TYPE_IMMEDIATE);
783                 DO_TYPE(PTLLND_MSG_TYPE_HELLO);
784                 DO_TYPE(PTLLND_MSG_TYPE_NOOP);
785                 DO_TYPE(PTLLND_MSG_TYPE_NAK);
786         default:
787                 return "<unknown msg type>";
788         }
789 #undef DO_TYPE
790 }
791
792 const char *ptllnd_errtype2str(int type)
793 {
794 #define DO_TYPE(x) case x: return #x;
795         switch(type)
796         {
797                 DO_TYPE(PTL_OK);
798                 DO_TYPE(PTL_SEGV);
799                 DO_TYPE(PTL_NO_SPACE);
800                 DO_TYPE(PTL_ME_IN_USE);
801                 DO_TYPE(PTL_NAL_FAILED);
802                 DO_TYPE(PTL_NO_INIT);
803                 DO_TYPE(PTL_IFACE_DUP);
804                 DO_TYPE(PTL_IFACE_INVALID);
805                 DO_TYPE(PTL_HANDLE_INVALID);
806                 DO_TYPE(PTL_MD_INVALID);
807                 DO_TYPE(PTL_ME_INVALID);
808                 DO_TYPE(PTL_PROCESS_INVALID);
809                 DO_TYPE(PTL_PT_INDEX_INVALID);
810                 DO_TYPE(PTL_SR_INDEX_INVALID);
811                 DO_TYPE(PTL_EQ_INVALID);
812                 DO_TYPE(PTL_EQ_DROPPED);
813                 DO_TYPE(PTL_EQ_EMPTY);
814                 DO_TYPE(PTL_MD_NO_UPDATE);
815                 DO_TYPE(PTL_FAIL);
816                 DO_TYPE(PTL_AC_INDEX_INVALID);
817                 DO_TYPE(PTL_MD_ILLEGAL);
818                 DO_TYPE(PTL_ME_LIST_TOO_LONG);
819                 DO_TYPE(PTL_MD_IN_USE);
820                 DO_TYPE(PTL_NI_INVALID);
821                 DO_TYPE(PTL_PID_INVALID);
822                 DO_TYPE(PTL_PT_FULL);
823                 DO_TYPE(PTL_VAL_FAILED);
824                 DO_TYPE(PTL_NOT_IMPLEMENTED);
825                 DO_TYPE(PTL_NO_ACK);
826                 DO_TYPE(PTL_EQ_IN_USE);
827                 DO_TYPE(PTL_PID_IN_USE);
828                 DO_TYPE(PTL_INV_EQ_SIZE);
829                 DO_TYPE(PTL_AGAIN);
830         default:
831                 return "<unknown error type>";
832         }
833 #undef DO_TYPE
834 }