Whamcloud - gitweb
i=liang,b=18450:
[fs/lustre-release.git] / lnet / klnds / gmlnd / gmlnd_api.c
index bdf5cfe..f6b4001 100644 (file)
 /* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
  * vim:expandtab:shiftwidth=8:tabstop=8:
  *
- *  Copyright (c) 2003 Los Alamos National Laboratory (LANL)
+ * GPL HEADER START
  *
- *   This file is part of Lustre, http://www.lustre.org/
+ * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
  *
- *   Lustre is free software; you can redistribute it and/or
- *   modify it under the terms of version 2 of the GNU General Public
- *   License as published by the Free Software Foundation.
+ * This program is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License version 2 only,
+ * as published by the Free Software Foundation.
  *
- *   Lustre is distributed in the hope that it will be useful,
- *   but WITHOUT ANY WARRANTY; without even the implied warranty of
- *   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
- *   GNU General Public License for more details.
+ * This program is distributed in the hope that it will be useful, but
+ * WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+ * General Public License version 2 for more details (a copy is included
+ * in the LICENSE file that accompanied this code).
  *
- *   You should have received a copy of the GNU General Public License
- *   along with Lustre; if not, write to the Free Software
- *   Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA.
+ * You should have received a copy of the GNU General Public License
+ * version 2 along with this program; If not, see
+ * http://www.sun.com/software/products/lustre/docs/GPLv2.pdf
+ *
+ * Please contact Sun Microsystems, Inc., 4150 Network Circle, Santa Clara,
+ * CA 95054 USA or visit www.sun.com if you need additional information or
+ * have any questions.
+ *
+ * GPL HEADER END
+ */
+/*
+ * Copyright  2008 Sun Microsystems, Inc. All rights reserved
+ * Use is subject to license terms.
+ *
+ * Copyright (c) 2003 Los Alamos National Laboratory (LANL)
+ */
+/*
+ * This file is part of Lustre, http://www.lustre.org/
+ * Lustre is a trademark of Sun Microsystems, Inc.
  */
 
 /*
  *     Implements the API NAL functions
  */
 
-#include "gmnal.h"
+#include "gmlnd.h"
+
+lnd_t the_gmlnd =
+{
+        .lnd_type            = GMLND,
+        .lnd_startup         = gmnal_startup,
+        .lnd_shutdown        = gmnal_shutdown,
+        .lnd_ctl             = gmnal_ctl,
+        .lnd_send            = gmnal_send,
+        .lnd_recv            = gmnal_recv,
+};
 
+gmnal_ni_t *the_gmni = NULL;
 
+int
+gmnal_ctl(lnet_ni_t *ni, unsigned int cmd, void *arg)
+{
+        struct libcfs_ioctl_data *data = arg;
 
-gmnal_data_t   *global_nal_data = NULL;
-#define         GLOBAL_NID_STR_LEN      16
-char            global_nid_str[GLOBAL_NID_STR_LEN] = {0};
+        switch (cmd) {
+        case IOC_LIBCFS_REGISTER_MYNID:
+                if (data->ioc_nid == ni->ni_nid)
+                        return 0;
 
-/*
- *      Write the global nid /proc/sys/gmnal/globalnid
- */
-#define GMNAL_SYSCTL    201
-#define GMNAL_SYSCTL_GLOBALNID  1
-
-static ctl_table gmnal_sysctl_table[] = {
-        {GMNAL_SYSCTL_GLOBALNID, "globalnid",
-         global_nid_str, GLOBAL_NID_STR_LEN,
-         0444, NULL, &proc_dostring},
-        { 0 }
-};
+                LASSERT (LNET_NIDNET(data->ioc_nid) == LNET_NIDNET(ni->ni_nid));
 
+                CERROR("obsolete IOC_LIBCFS_REGISTER_MYNID for %s(%s)\n",
+                       libcfs_nid2str(data->ioc_nid),
+                       libcfs_nid2str(ni->ni_nid));
+                return 0;
 
-static ctl_table gmnalnal_top_sysctl_table[] = {
-        {GMNAL_SYSCTL, "gmnal", NULL, 0, 0555, gmnal_sysctl_table},
-        { 0 }
-};
+        default:
+                return (-EINVAL);
+        }
+}
+
+int
+gmnal_set_local_nid (gmnal_ni_t *gmni)
+{
+        lnet_ni_t       *ni = gmni->gmni_ni;
+        __u32            local_gmid;
+        __u32            global_gmid;
+        gm_status_t      gm_status;
+
+        /* Called before anything initialised: no need to lock */
+        gm_status = gm_get_node_id(gmni->gmni_port, &local_gmid);
+        if (gm_status != GM_SUCCESS)
+                return 0;
+
+        CDEBUG(D_NET, "Local node id is [%u]\n", local_gmid);
+
+        gm_status = gm_node_id_to_global_id(gmni->gmni_port, 
+                                            local_gmid, 
+                                            &global_gmid);
+        if (gm_status != GM_SUCCESS)
+                return 0;
+
+        CDEBUG(D_NET, "Global node id is [%u]\n", global_gmid);
+
+        ni->ni_nid = LNET_MKNID(LNET_NIDNET(ni->ni_nid), global_gmid);
+        return 1;
+}
 
-/*
- *     gmnal_api_shutdown
- *      nal_refct == 0 => called on last matching PtlNIFini()
- *     Close down this interface and free any resources associated with it
- *     nal_t   nal     our nal to shutdown
- */
 void
-gmnal_api_shutdown(nal_t *nal, int interface)
+gmnal_shutdown(lnet_ni_t *ni)
 {
-       gmnal_data_t    *nal_data;
-       lib_nal_t       *libnal;
-
-        if (nal->nal_refct != 0)
-                return;
-        
-       CDEBUG(D_TRACE, "gmnal_api_shutdown: nal_data [%p]\n", nal_data);
-
-        LASSERT(nal == global_nal_data->nal);
-        libnal = (lib_nal_t *)nal->nal_data;
-        nal_data = (gmnal_data_t *)libnal->libnal_data;
-        LASSERT(nal_data == global_nal_data);
-
-        /* Stop portals calling our ioctl handler */
-        libcfs_nal_cmd_unregister(GMNAL);
-
-        /* XXX for shutdown "under fire" we probably need to set a shutdown
-         * flag so when lib calls us we fail immediately and dont queue any
-         * more work but our threads can still call into lib OK.  THEN
-         * shutdown our threads, THEN lib_fini() */
-        lib_fini(libnal);
-
-       gmnal_stop_rxthread(nal_data);
-       gmnal_stop_ctthread(nal_data);
-       gmnal_free_txd(nal_data);
-       gmnal_free_srxd(nal_data);
-       GMNAL_GM_LOCK(nal_data);
-       gm_close(nal_data->gm_port);
-       gm_finalize();
-       GMNAL_GM_UNLOCK(nal_data);
-        if (nal_data->sysctl)
-                unregister_sysctl_table (nal_data->sysctl);
-        /* Don't free 'nal'; it's a static struct */
-       PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-       PORTAL_FREE(libnal, sizeof(lib_nal_t));
-
-        global_nal_data = NULL;
+        gmnal_ni_t *gmni = ni->ni_data;
+
+        CDEBUG(D_TRACE, "gmnal_api_shutdown: gmni [%p]\n", gmni);
+
+        LASSERT (gmni == the_gmni);
+
+        /* stop processing messages */
+        gmnal_stop_threads(gmni);
+
+        /* stop all network callbacks */
+        gm_close(gmni->gmni_port);
+        gmni->gmni_port = NULL;
+
+        gm_finalize();
+
+        gmnal_free_ltxbufs(gmni);
+        gmnal_free_txs(gmni);
+        gmnal_free_rxs(gmni);
+
+        LIBCFS_FREE(gmni, sizeof(*gmni));
+
+        the_gmni = NULL;
         PORTAL_MODULE_UNUSE;
 }
 
-
 int
-gmnal_api_startup(nal_t *nal, ptl_pid_t requested_pid,
-                  ptl_ni_limits_t *requested_limits,
-                  ptl_ni_limits_t *actual_limits)
+gmnal_startup(lnet_ni_t *ni)
 {
+        gmnal_ni_t  *gmni = NULL;
+        gmnal_rx_t  *rx = NULL;
+        gm_status_t  gm_status;
+        int          rc;
+
+        LASSERT (ni->ni_lnd == &the_gmlnd);
 
-       lib_nal_t       *libnal = NULL;
-       gmnal_data_t    *nal_data = NULL;
-       gmnal_srxd_t    *srxd = NULL;
-       gm_status_t     gm_status;
-       unsigned int    local_nid = 0, global_nid = 0;
-        ptl_process_id_t process_id;
-
-        if (nal->nal_refct != 0) {
-                if (actual_limits != NULL) {
-                        libnal = (lib_nal_t *)nal->nal_data;
-                        *actual_limits = nal->libnal_ni.ni_actual_limits;
-                return (PTL_OK);
+        ni->ni_maxtxcredits = *gmnal_tunables.gm_credits;
+        ni->ni_peertxcredits = *gmnal_tunables.gm_peer_credits;
+
+        if (the_gmni != NULL) {
+                CERROR("Only 1 instance supported\n");
+                return -EINVAL;
         }
 
-        /* Called on first PtlNIInit() */
-
-       CDEBUG(D_TRACE, "startup\n");
-
-        LASSERT(global_nal_data == NULL);
-
-       PORTAL_ALLOC(nal_data, sizeof(gmnal_data_t));
-       if (!nal_data) {
-               CDEBUG(D_ERROR, "can't get memory\n");
-               return(PTL_NO_SPACE);
-       }       
-       memset(nal_data, 0, sizeof(gmnal_data_t));
-       /*
-        *      set the small message buffer size 
-        */
-
-       CDEBUG(D_INFO, "Allocd and reset nal_data[%p]\n", nal_data);
-       CDEBUG(D_INFO, "small_msg_size is [%d]\n", nal_data->small_msg_size);
-
-       PORTAL_ALLOC(libnal, sizeof(lib_nal_t));
-       if (!libnal) {
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));
-               return(PTL_NO_SPACE);
-       }
-       memset(libnal, 0, sizeof(lib_nal_t));
-       CDEBUG(D_INFO, "Allocd and reset libnal[%p]\n", libnal);
-
-       GMNAL_INIT_NAL_CB(libnal);
-       /*
-        *      String them all together
-        */
-       libnal->libnal_data = (void*)nal_data;
-       nal_data->nal = nal;
-       nal_data->libnal = libnal;
-
-       GMNAL_GM_LOCK_INIT(nal_data);
-
-
-       /*
-        *      initialise the interface, 
-        */
-       CDEBUG(D_INFO, "Calling gm_init\n");
-       if (gm_init() != GM_SUCCESS) {
-               CDEBUG(D_ERROR, "call to gm_init failed\n");
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
-       }
-
-
-       CDEBUG(D_NET, "Calling gm_open with interface [%d], port [%d], "
-                      "name [%s], version [%d]\n", interface, GMNAL_GM_PORT, 
-              "gmnal", GM_API_VERSION);
-
-       GMNAL_GM_LOCK(nal_data);
-       gm_status = gm_open(&nal_data->gm_port, 0, GMNAL_GM_PORT, "gmnal", 
-                           GM_API_VERSION);
-       GMNAL_GM_UNLOCK(nal_data);
-
-       CDEBUG(D_INFO, "gm_open returned [%d]\n", gm_status);
-       if (gm_status == GM_SUCCESS) {
-               CDEBUG(D_INFO, "gm_open succeeded port[%p]\n", 
-                      nal_data->gm_port);
-       } else {
-               switch(gm_status) {
-               case(GM_INVALID_PARAMETER):
-                       CDEBUG(D_ERROR, "gm_open Failure. Invalid Parameter\n");
-                       break;
-               case(GM_BUSY):
-                       CDEBUG(D_ERROR, "gm_open Failure. GM Busy\n");
-                       break;
-               case(GM_NO_SUCH_DEVICE):
-                       CDEBUG(D_ERROR, "gm_open Failure. No such device\n");
-                       break;
-               case(GM_INCOMPATIBLE_LIB_AND_DRIVER):
-                       CDEBUG(D_ERROR, "gm_open Failure. Incompatile lib "
-                              "and driver\n");
-                       break;
-               case(GM_OUT_OF_MEMORY):
-                       CDEBUG(D_ERROR, "gm_open Failure. Out of Memory\n");
-                       break;
-               default:
-                       CDEBUG(D_ERROR, "gm_open Failure. Unknow error "
-                              "code [%d]\n", gm_status);
-                       break;
-               }       
-               GMNAL_GM_LOCK(nal_data);
-               gm_finalize();
-               GMNAL_GM_UNLOCK(nal_data);
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
-       }
-
-       
-       nal_data->small_msg_size = gmnal_small_msg_size;
-       nal_data->small_msg_gmsize = 
-                       gm_min_size_for_length(gmnal_small_msg_size);
-
-       if (gmnal_alloc_srxd(nal_data) != GMNAL_STATUS_OK) {
-               CDEBUG(D_ERROR, "Failed to allocate small rx descriptors\n");
-               gmnal_free_txd(nal_data);
-               GMNAL_GM_LOCK(nal_data);
-               gm_close(nal_data->gm_port);
-               gm_finalize();
-               GMNAL_GM_UNLOCK(nal_data);
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
-       }
-
-
-       /*
-        *      Hang out a bunch of small receive buffers
-        *      In fact hang them all out
-        */
-       while((srxd = gmnal_get_srxd(nal_data, 0))) {
-               CDEBUG(D_NET, "giving [%p] to gm_provide_recvive_buffer\n", 
-                      srxd->buffer);
-               GMNAL_GM_LOCK(nal_data);
-               gm_provide_receive_buffer_with_tag(nal_data->gm_port, 
-                                                  srxd->buffer, srxd->gmsize, 
-                                                  GM_LOW_PRIORITY, 0);
-               GMNAL_GM_UNLOCK(nal_data);
-       }
-       
-       /*
-        *      Allocate pools of small tx buffers and descriptors
-        */
-       if (gmnal_alloc_txd(nal_data) != GMNAL_STATUS_OK) {
-               CDEBUG(D_ERROR, "Failed to allocate small tx descriptors\n");
-               GMNAL_GM_LOCK(nal_data);
-               gm_close(nal_data->gm_port);
-               gm_finalize();
-               GMNAL_GM_UNLOCK(nal_data);
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
-       }
-
-       gmnal_start_kernel_threads(nal_data);
-
-       while (nal_data->rxthread_flag != GMNAL_RXTHREADS_STARTED) {
-               gmnal_yield(1);
-               CDEBUG(D_INFO, "Waiting for receive thread signs of life\n");
-       }
-
-       CDEBUG(D_INFO, "receive thread seems to have started\n");
-
-
-       /*
-        *      Initialise the portals library
-        */
-       CDEBUG(D_NET, "Getting node id\n");
-       GMNAL_GM_LOCK(nal_data);
-       gm_status = gm_get_node_id(nal_data->gm_port, &local_nid);
-       GMNAL_GM_UNLOCK(nal_data);
-       if (gm_status != GM_SUCCESS) {
-               gmnal_stop_rxthread(nal_data);
-               gmnal_stop_ctthread(nal_data);
-               CDEBUG(D_ERROR, "can't determine node id\n");
-               gmnal_free_txd(nal_data);
-               gmnal_free_srxd(nal_data);
-               GMNAL_GM_LOCK(nal_data);
-               gm_close(nal_data->gm_port);
-               gm_finalize();
-               GMNAL_GM_UNLOCK(nal_data);
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
-       }
-       nal_data->gm_local_nid = local_nid;
-       CDEBUG(D_INFO, "Local node id is [%u]\n", local_nid);
-       GMNAL_GM_LOCK(nal_data);
-       gm_status = gm_node_id_to_global_id(nal_data->gm_port, local_nid, 
-                                           &global_nid);
-       GMNAL_GM_UNLOCK(nal_data);
-       if (gm_status != GM_SUCCESS) {
-               CDEBUG(D_ERROR, "failed to obtain global id\n");
-               gmnal_stop_rxthread(nal_data);
-               gmnal_stop_ctthread(nal_data);
-               gmnal_free_txd(nal_data);
-               gmnal_free_srxd(nal_data);
-               GMNAL_GM_LOCK(nal_data);
-               gm_close(nal_data->gm_port);
-               gm_finalize();
-               GMNAL_GM_UNLOCK(nal_data);
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
-       }
-       CDEBUG(D_INFO, "Global node id is [%u]\n", global_nid);
-       nal_data->gm_global_nid = global_nid;
-        snprintf(global_nid_str, GLOBAL_NID_STR_LEN, "%u", global_nid);
+        LIBCFS_ALLOC(gmni, sizeof(*gmni));
+        if (gmni == NULL) {
+                CERROR("can't allocate gmni\n");
+                return -ENOMEM;
+        }
 
-/*
-       pid = gm_getpid();
-*/
-        process_id.pid = requested_pid;
-        process_id.nid = global_nid;
-        
-       CDEBUG(D_INFO, "portals_pid is [%u]\n", process_id.pid);
-       CDEBUG(D_INFO, "portals_nid is ["LPU64"]\n", process_id.nid);
-       
-       CDEBUG(D_PORTALS, "calling lib_init\n");
-       if (lib_init(libnal, nal, process_id, 
-                     requested_limits, actual_limits) != PTL_OK) {
-               CDEBUG(D_ERROR, "lib_init failed\n");
-               gmnal_stop_rxthread(nal_data);
-               gmnal_stop_ctthread(nal_data);
-               gmnal_free_txd(nal_data);
-               gmnal_free_srxd(nal_data);
-               GMNAL_GM_LOCK(nal_data);
-               gm_close(nal_data->gm_port);
-               gm_finalize();
-               GMNAL_GM_UNLOCK(nal_data);
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
-               
-       }
-
-       if (libcfs_nal_cmd_register(GMNAL, &gmnal_cmd, nal->nal_data) != 0) {
-               CDEBUG(D_INFO, "libcfs_nal_cmd_register failed\n");
-
-                /* XXX these cleanup cases should be restructured to
-                 * minimise duplication... */
-                lib_fini(libnal);
-                
-               gmnal_stop_rxthread(nal_data);
-               gmnal_stop_ctthread(nal_data);
-               gmnal_free_txd(nal_data);
-               gmnal_free_srxd(nal_data);
-               GMNAL_GM_LOCK(nal_data);
-               gm_close(nal_data->gm_port);
-               gm_finalize();
-               GMNAL_GM_UNLOCK(nal_data);
-               PORTAL_FREE(nal_data, sizeof(gmnal_data_t));    
-               PORTAL_FREE(libnal, sizeof(lib_nal_t));
-               return(PTL_FAIL);
+        ni->ni_data = gmni;
+
+        memset(gmni, 0, sizeof(*gmni));
+        gmni->gmni_ni = ni;
+        spin_lock_init(&gmni->gmni_tx_lock);
+        spin_lock_init(&gmni->gmni_gm_lock);
+        INIT_LIST_HEAD(&gmni->gmni_idle_txs);
+        INIT_LIST_HEAD(&gmni->gmni_idle_ltxbs);
+        INIT_LIST_HEAD(&gmni->gmni_buf_txq);
+        INIT_LIST_HEAD(&gmni->gmni_cred_txq);
+        sema_init(&gmni->gmni_rx_mutex, 1);
+        PORTAL_MODULE_USE;
+
+        /*
+         * initialise the interface,
+         */
+        CDEBUG(D_NET, "Calling gm_init\n");
+        if (gm_init() != GM_SUCCESS) {
+                CERROR("call to gm_init failed\n");
+                goto failed_0;
         }
 
-        /* might be better to initialise this at module load rather than in
-         * NAL startup */
-        nal_data->sysctl = NULL;
-        nal_data->sysctl = register_sysctl_table (gmnalnal_top_sysctl_table, 0);
+        CDEBUG(D_NET, "Calling gm_open with port [%d], version [%d]\n",
+               *gmnal_tunables.gm_port, GM_API_VERSION);
 
-       
-       CDEBUG(D_INFO, "gmnal_init finished\n");
-       global_nal_data = nal->nal_data;
+        gm_status = gm_open(&gmni->gmni_port, 0, *gmnal_tunables.gm_port, 
+                            "gmnal", GM_API_VERSION);
 
-        /* no unload now until shutdown */
-        PORTAL_MODULE_USE;
-        
-       return(PTL_OK);
-}
+        if (gm_status != GM_SUCCESS) {
+                CERROR("Can't open GM port %d: %d (%s)\n",
+                       *gmnal_tunables.gm_port, gm_status, 
+                       gmnal_gmstatus2str(gm_status));
+                goto failed_1;
+        }
+
+        CDEBUG(D_NET,"gm_open succeeded port[%p]\n",gmni->gmni_port);
+
+        if (!gmnal_set_local_nid(gmni))
+                goto failed_2;
+
+        CDEBUG(D_NET, "portals_nid is %s\n", libcfs_nid2str(ni->ni_nid));
+
+        gmni->gmni_large_msgsize = 
+                offsetof(gmnal_msg_t, gmm_u.immediate.gmim_payload[LNET_MAX_PAYLOAD]);
+        gmni->gmni_large_gmsize = 
+                gm_min_size_for_length(gmni->gmni_large_msgsize);
+        gmni->gmni_large_pages =
+                (gmni->gmni_large_msgsize + PAGE_SIZE - 1)/PAGE_SIZE;
+
+        gmni->gmni_small_msgsize = MIN(GM_MTU, PAGE_SIZE);
+        gmni->gmni_small_gmsize = 
+                gm_min_size_for_length(gmni->gmni_small_msgsize);
+
+        gmni->gmni_netaddr_base = GMNAL_NETADDR_BASE;
+        gmni->gmni_netaddr_size = 0;
+
+        CDEBUG(D_NET, "Msg size %08x/%08x [%d/%d]\n", 
+               gmni->gmni_large_msgsize, gmni->gmni_small_msgsize,
+               gmni->gmni_large_gmsize, gmni->gmni_small_gmsize);
+
+        if (gmnal_alloc_rxs(gmni) != 0) {
+                CERROR("Failed to allocate rx descriptors\n");
+                goto failed_2;
+        }
+
+        if (gmnal_alloc_txs(gmni) != 0) {
+                CERROR("Failed to allocate tx descriptors\n");
+                goto failed_2;
+        }
 
-nal_t the_gm_nal;
+        if (gmnal_alloc_ltxbufs(gmni) != 0) {
+                CERROR("Failed to allocate large tx buffers\n");
+                goto failed_2;
+        }
+
+        rc = gmnal_start_threads(gmni);
+        if (rc != 0) {
+                CERROR("Can't start threads: %d\n", rc);
+                goto failed_2;
+        }
+
+        /* Start listening */
+        for (rx = gmni->gmni_rxs; rx != NULL; rx = rx->rx_next)
+                gmnal_post_rx(gmni, rx);
+
+        the_gmni = gmni;
+
+        CDEBUG(D_NET, "gmnal_init finished\n");
+        return 0;
+
+ failed_2:
+        gm_close(gmni->gmni_port);
+        gmni->gmni_port = NULL;
+
+ failed_1:
+        gm_finalize();
+
+ failed_0:
+        /* safe to free descriptors after network has been shut down */
+        gmnal_free_ltxbufs(gmni);
+        gmnal_free_txs(gmni);
+        gmnal_free_rxs(gmni);
+
+        LIBCFS_FREE(gmni, sizeof(*gmni));
+        PORTAL_MODULE_UNUSE;
+
+        return -EIO;
+}
 
 /* 
  *        Called when module loaded
  */
 int gmnal_init(void)
 {
-        int    rc;
-
-       memset(&the_gm_nal, 0, sizeof(nal_t));
-       CDEBUG(D_INFO, "reset nal[%p]\n", &the_gm_nal);
-       GMNAL_INIT_NAL(&the_gm_nal);
-
-        rc = ptl_register_nal(GMNAL, &the_gm_nal);
-        if (rc != PTL_OK)
-                CERROR("Can't register GMNAL: %d\n", rc);
-
-        return (rc);
+        lnet_register_lnd(&the_gmlnd);
+        return 0;
 }
 
-                
-
 /*
  *     Called when module removed
  */
 void gmnal_fini()
 {
-       CDEBUG(D_TRACE, "gmnal_fini\n");
-
-        LASSERT(global_nal_data == NULL);
-
-        ptl_unregister_nal(GMNAL);
+        lnet_unregister_lnd(&the_gmlnd);
 }