<commit_before>/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * 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.
 *
 * 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
 * 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.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 */

#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE 500
#endif

#define DEBUG_SUBSYSTEM S_LNET
#include <lnet/lib-lnet.h>

#include "psc_util/lock.h"

#ifdef __KERNEL__
#define D_LNI D_CONSOLE
#else
#define D_LNI D_CONFIG
#endif

lnet_t      the_lnet;                           /* THE state of the network */

#ifdef __KERNEL__

static char *ip2nets = "";
CFS_MODULE_PARM(ip2nets, "s", charp, 0444,
                "LNET network <- IP table");

static char *networks = "";
CFS_MODULE_PARM(networks, "s", charp, 0444,
                "local networks");

static char *routes = "";
CFS_MODULE_PARM(routes, "s", charp, 0444,
                "routes to non-local networks");

static char *portals_compatibility = "none";
CFS_MODULE_PARM(portals_compatibility, "s", charp, 0444,
                "wire protocol compatibility: 'strong'|'weak'|'none'");

char *
lnet_get_routes(void)
{
        return routes;
}

char *
lnet_get_networks(void)
{
        char   *nets;
        int     rc;

        if (*networks != 0 && *ip2nets != 0) {
                LCONSOLE_ERROR_MSG(0x101, "Please specify EITHER 'networks' or "
                                   "'ip2nets' but not both at once\n");
                return NULL;
        }

        if (*ip2nets != 0) {
                rc = lnet_parse_ip2nets(&nets, ip2nets);
                return (rc == 0) ? nets : NULL;
        }

        if (*networks != 0)
                return networks;

        return "tcp";
}

int
lnet_get_portals_compatibility(void)
{
        if (!strcmp(portals_compatibility, "none")) {
                return 0;
        }

        if (!strcmp(portals_compatibility, "weak")) {
                return 1;
                LCONSOLE_WARN("Starting in weak portals-compatible mode\n");
        }

        if (!strcmp(portals_compatibility, "strong")) {
                return 2;
                LCONSOLE_WARN("Starting in strong portals-compatible mode\n");
        }

        LCONSOLE_ERROR_MSG(0x102, "portals_compatibility=\"%s\" not supported\n",
                           portals_compatibility);
        return -EINVAL;
}

void
lnet_init_locks(void)
{
        spin_lock_init (&the_lnet.ln_lock);
        cfs_waitq_init (&the_lnet.ln_waitq);
        init_mutex(&the_lnet.ln_lnd_mutex);
        init_mutex(&the_lnet.ln_api_mutex);
}

void
lnet_fini_locks(void)
{
}

#else

int
lnet_get_usesdp(void)
{
	static psc_spinlock_t lock = LOCK_INITIALIZER;
	static int usesdp;
	static int init;

	if (init)
		return (usesdp);
	spinlock(&lock);
	if (init == 0) {
		char *val;

		val = getenv("LNET_USESDP");
		if (val && *val != '0')
			usesdp = 1;
		init = 1;
	}
	freelock(&lock);
	return (usesdp);
}

char *
lnet_get_routes(void)
{
        char *str = getenv("LNET_ROUTES");

        return (str == NULL) ? "" : str;
}

char *
lnet_get_networks (void)
{
        static char       default_networks[256];
        char             *networks = getenv ("LNET_NETWORKS");
        char             *ip2nets  = getenv ("LNET_IP2NETS");
        char             *str;
        char             *sep;
        int               len;
        int               nob;
        int               rc;
        struct list_head *tmp;

#ifdef NOT_YET
        if (networks != NULL && ip2nets != NULL) {
                LCONSOLE_ERROR_MSG(0x103, "Please set EITHER 'LNET_NETWORKS' or"
                                   " 'LNET_IP2NETS' but not both at once\n");
                return NULL;
        }

        if (ip2nets != NULL) {
                rc = lnet_parse_ip2nets(&networks, ip2nets);
                return (rc == 0) ? networks : NULL;
        }
#else
        ip2nets = NULL;
        rc = 0;
#endif
        if (networks != NULL)
                return networks;

        /* In userland, the default 'networks=' is the list of known net types */

        len = sizeof(default_networks);
        str = default_networks;
        *str = 0;
        sep = "";

        list_for_each (tmp, &the_lnet.ln_lnds) {
                lnd_t *lnd = list_entry(tmp, lnd_t, lnd_list);

                nob = snprintf(str, len, "%s%s", sep,
                               libcfs_lnd2str(lnd->lnd_type));
                len -= nob;
                if (len < 0) {
                        /* overflowed the string; leave it where it was */
                        *str = 0;
                        break;
                }

                str += nob;
                sep = ",";
        }

        return default_networks;
}

int
lnet_get_portals_compatibility(void)
{
        return 0;
}

# ifndef HAVE_LIBPTHREAD

void lnet_init_locks(void)
{
        the_lnet.ln_lock = 0;
        the_lnet.ln_lnd_mutex = 0;
        the_lnet.ln_api_mutex = 0;
}

void lnet_fini_locks(void)
{
        LASSERT (the_lnet.ln_api_mutex == 0);
        LASSERT (the_lnet.ln_lnd_mutex == 0);
        LASSERT (the_lnet.ln_lock == 0);
}

# else

void lnet_init_locks(void)
{
	pthread_mutexattr_t attr;

        pthread_cond_init(&the_lnet.ln_cond, NULL);
	pthread_mutexattr_init(&attr);
	pthread_mutexattr_settype(&attr,
	    PTHREAD_MUTEX_ERRORCHECK_NP);
        pthread_mutex_init(&the_lnet.ln_lock, &attr);
        pthread_mutex_init(&the_lnet.ln_lnd_mutex, &attr);
        pthread_mutex_init(&the_lnet.ln_api_mutex, &attr);
}

void lnet_fini_locks(void)
{
        pthread_mutex_destroy(&the_lnet.ln_api_mutex);
        pthread_mutex_destroy(&the_lnet.ln_lnd_mutex);
        pthread_mutex_destroy(&the_lnet.ln_lock);
        pthread_cond_destroy(&the_lnet.ln_cond);
}

# endif
#endif

void lnet_assert_wire_constants (void)
{
        /* Wire protocol assertions generated by 'wirecheck'
         * running on Linux robert.bartonsoftware.com 2.6.8-1.521
         * #1 Mon Aug 16 09:01:18 EDT 2004 i686 athlon i386 GNU/Linux
         * with gcc version 3.3.3 20040412 (Red Hat Linux 3.3.3-7) */

        /* Constants... */
        CLASSERT (LNET_PROTO_TCP_MAGIC == 0xeebc0ded);
        CLASSERT (LNET_PROTO_TCP_VERSION_MAJOR == 1);
        CLASSERT (LNET_PROTO_TCP_VERSION_MINOR == 0);
        CLASSERT (LNET_MSG_ACK == 0);
        CLASSERT (LNET_MSG_PUT == 1);
        CLASSERT (LNET_MSG_GET == 2);
        CLASSERT (LNET_MSG_REPLY == 3);
        CLASSERT (LNET_MSG_HELLO == 4);

        /* Checks for struct ptl_handle_wire_t */
        CLASSERT ((int)sizeof(lnet_handle_wire_t) == 16);
        CLASSERT ((int)offsetof(lnet_handle_wire_t, wh_interface_cookie) == 0);
        CLASSERT ((int)sizeof(((lnet_handle_wire_t *)0)->wh_interface_cookie) == 8);
        CLASSERT ((int)offsetof(lnet_handle_wire_t, wh_object_cookie) == 8);
        CLASSERT ((int)sizeof(((lnet_handle_wire_t *)0)->wh_object_cookie) == 8);

        /* Checks for struct lnet_magicversion_t */
        CLASSERT ((int)sizeof(lnet_magicversion_t) == 8);
        CLASSERT ((int)offsetof(lnet_magicversion_t, magic) == 0);
        CLASSERT ((int)sizeof(((lnet_magicversion_t *)0)->magic) == 4);
        CLASSERT ((int)offsetof(lnet_magicversion_t, version_major) == 4);
        CLASSERT ((int)sizeof(((lnet_magicversion_t *)0)->version_major) == 2);
        CLASSERT ((int)offsetof(lnet_magicversion_t, version_minor) == 6);
        CLASSERT ((int)sizeof(((lnet_magicversion_t *)0)->version_minor) == 2);

        /* Checks for struct lnet_hdr_t */
        CLASSERT ((int)sizeof(lnet_hdr_t) == 72);
        CLASSERT ((int)offsetof(lnet_hdr_t, dest_nid) == 0);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->dest_nid) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, src_nid) == 8);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->src_nid) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, dest_pid) == 16);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->dest_pid) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, src_pid) == 20);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->src_pid) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, type) == 24);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->type) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, payload_length) == 28);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->payload_length) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg) == 40);

        /* Ack */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.ack.dst_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.ack.dst_wmd) == 16);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.ack.match_bits) == 48);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.ack.match_bits) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.ack.mlength) == 56);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.ack.mlength) == 4);

        /* Put */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.ack_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.ack_wmd) == 16);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.match_bits) == 48);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.match_bits) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.hdr_data) == 56);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.hdr_data) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.ptl_index) == 64);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.ptl_index) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.offset) == 68);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.offset) == 4);

        /* Get */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.return_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.return_wmd) == 16);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.match_bits) == 48);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.match_bits) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.ptl_index) == 56);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.ptl_index) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.src_offset) == 60);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.src_offset) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.sink_length) == 64);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.sink_length) == 4);

        /* Reply */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.reply.dst_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.reply.dst_wmd) == 16);

        /* Hello */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.hello.incarnation) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.hello.incarnation) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.hello.type) == 40);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.hello.type) == 4);
}

lnd_t *
lnet_find_lnd_by_type (int type)
{
        lnd_t              *lnd;
        struct list_head   *tmp;

        /* holding lnd mutex */
        list_for_each (tmp, &the_lnet.ln_lnds) {
                lnd = list_entry(tmp, lnd_t, lnd_list);

                if (lnd->lnd_type == (unsigned int)type)
                        return lnd;
        }

        return NULL;
}

void
lnet_register_lnd (lnd_t *lnd)
{
        LNET_MUTEX_DOWN(&the_lnet.ln_lnd_mutex);

        LASSERT (the_lnet.ln_init);
        LASSERT (libcfs_isknown_lnd(lnd->lnd_type));
        LASSERT (lnet_find_lnd_by_type(lnd->lnd_type) == NULL);

        list_add_tail (&lnd->lnd_list, &the_lnet.ln_lnds);
        lnd->lnd_refcount = 0;

        CDEBUG(D_NET, "%s LND registered\n", libcfs_lnd2str(lnd->lnd_type));

        LNET_MUTEX_UP(&the_lnet.ln_lnd_mutex);
}

void
lnet_unregister_lnd (lnd_t *lnd)
{
        LNET_MUTEX_DOWN(&the_lnet.ln_lnd_mutex);

        LASSERT (the_lnet.ln_init);
        LASSERT (lnet_find_lnd_by_type(lnd->lnd_type) == lnd);
        LASSERT (lnd->lnd_refcount == 0);

        list_del (&lnd->lnd_list);
        CDEBUG(D_NET, "%s LND unregistered\n", libcfs_lnd2str(lnd->lnd_type));

        LNET_MUTEX_UP(&the_lnet.ln_lnd_mutex);
}

#ifndef LNET_USE_LIB_FREELIST

int
lnet_descriptor_setup (void)
{
        return 0;
}

void
lnet_descriptor_cleanup (void)
{
}

#else

int
lnet_freelist_init (lnet_freelist_t *fl, int n, int size)
{
        char *space;

        LASSERT (n > 0);

        size += offsetof (lnet_freeobj_t, fo_contents);

        LIBCFS_ALLOC(space, n * size);
        if (space == NULL)
                return (-ENOMEM);

        CFS_INIT_LIST_HEAD (&fl->fl_list);
        fl->fl_objs = space;
        fl->fl_nobjs = n;
        fl->fl_objsize = size;

        do
        {
                memset (space, 0, size);
                list_add ((struct list_head *)space, &fl->fl_list);
                space += size;
        } while (--n != 0);

        return (0);
}

void
lnet_freelist_fini (lnet_freelist_t *fl)
{
        struct list_head *el;
        int               count;

        if (fl->fl_nobjs == 0)
                return;

        count = 0;
        for (el = fl->fl_list.next; el != &fl->fl_list; el = el->next)
                count++;

        LASSERT (count == fl->fl_nobjs);

        LIBCFS_FREE(fl->fl_objs, fl->fl_nobjs * fl->fl_objsize);
        memset (fl, 0, sizeof (*fl));
}

int
lnet_descriptor_setup (void)
{
        /* NB on failure caller must still call lnet_descriptor_cleanup */
        /*               ******                                         */
        int        rc;

        memset (&the_lnet.ln_free_mes,  0, sizeof (the_lnet.ln_free_mes));
        memset (&the_lnet.ln_free_msgs, 0, sizeof (the_lnet.ln_free_msgs));
        memset (&the_lnet.ln_free_mds,  0, sizeof (the_lnet.ln_free_mds));
        memset (&the_lnet.ln_free_eqs,  0, sizeof (the_lnet.ln_free_eqs));

        rc = lnet_freelist_init(&the_lnet.ln_free_mes,
                                MAX_MES, sizeof (lnet_me_t));
        if (rc != 0)
                return (rc);

        rc = lnet_freelist_init(&the_lnet.ln_free_msgs,
                                MAX_MSGS, sizeof (lnet_msg_t));
        if (rc != 0)
                return (rc);

        rc = lnet_freelist_init(&the_lnet.ln_free_mds,
                                MAX_MDS, sizeof (lnet_libmd_t));
        if (rc != 0)
                return (rc);

        rc = lnet_freelist_init(&the_lnet.ln_free_eqs,
                                MAX_EQS, sizeof (lnet_eq_t));
        return (rc);
}

void
lnet_descriptor_cleanup (void)
{
        lnet_freelist_fini (&the_lnet.ln_free_mes);
        lnet_freelist_fini (&the_lnet.ln_free_msgs);
        lnet_freelist_fini (&the_lnet.ln_free_mds);
        lnet_freelist_fini (&the_lnet.ln_free_eqs);
}

#endif

__u64
lnet_create_interface_cookie (void)
{
        /* NB the interface cookie in wire handles guards against delayed
         * replies and ACKs appearing valid after reboot. Initialisation time,
         * even if it's only implemented to millisecond resolution is probably
         * easily good enough. */
        struct timeval tv;
        __u64          cookie;
#ifndef __KERNEL__
        int            rc = gettimeofday (&tv, NULL);
        LASSERT (rc == 0);
#else
        do_gettimeofday(&tv);
#endif
        cookie = tv.tv_sec;
        cookie *= 1000000;
        cookie += tv.tv_usec;
        return cookie;
}

int
lnet_setup_handle_hash (void)
{
        int       i;

        /* Arbitrary choice of hash table size */
#ifdef __KERNEL__
        the_lnet.ln_lh_hash_size = CFS_PAGE_SIZE / sizeof (struct list_head);
#else
        the_lnet.ln_lh_hash_size = (MAX_MES + MAX_MDS + MAX_EQS)/4;
#endif
        LIBCFS_ALLOC(the_lnet.ln_lh_hash_table,
                     the_lnet.ln_lh_hash_size * sizeof (struct list_head));
        if (the_lnet.ln_lh_hash_table == NULL)
                return (-ENOMEM);

        for (i = 0; i < the_lnet.ln_lh_hash_size; i++)
                CFS_INIT_LIST_HEAD (&the_lnet.ln_lh_hash_table[i]);

        the_lnet.ln_next_object_cookie = LNET_COOKIE_TYPES;

        return (0);
}

void
lnet_cleanup_handle_hash (void)
{
        if (the_lnet.ln_lh_hash_table == NULL)
                return;

        LIBCFS_FREE(the_lnet.ln_lh_hash_table,
                    the_lnet.ln_lh_hash_size * sizeof (struct list_head));
}

lnet_libhandle_t *
lnet_lookup_cookie (__u64 cookie, int type)
{
        /* ALWAYS called with LNET_LOCK held */
        struct list_head    *list;
        struct list_head    *el;
        unsigned int         hash;

        if ((cookie & (LNET_COOKIE_TYPES - 1)) != (__u64)type)
                return (NULL);

        hash = ((unsigned int)cookie) % the_lnet.ln_lh_hash_size;
        list = &the_lnet.ln_lh_hash_table[hash];

        list_for_each (el, list) {
                lnet_libhandle_t *lh = list_entry (el, lnet_libhandle_t,
                                                  lh_hash_chain);

                if (lh->lh_cookie == cookie)
                        return (lh);
        }

        return (NULL);
}

void
lnet_initialise_handle (lnet_libhandle_t *lh, int type)
{
        /* ALWAYS called with LNET_LOCK held */
        unsigned int    hash;

        LASSERT (type >= 0 && type < LNET_COOKIE_TYPES);
        lh->lh_cookie = the_lnet.ln_next_object_cookie | type;
        the_lnet.ln_next_object_cookie += LNET_COOKIE_TYPES;

        hash = ((unsigned int)lh->lh_cookie) % the_lnet.ln_lh_hash_size;
        list_add (&lh->lh_hash_chain, &the_lnet.ln_lh_hash_table[hash]);
}

void
lnet_invalidate_handle (lnet_libhandle_t *lh)
{
        /* ALWAYS called with LNET_LOCK held */
        list_del (&lh->lh_hash_chain);
}

int
lnet_init_finalizers(void)
{
#ifdef __KERNEL__
        int    i;

        the_lnet.ln_nfinalizers = num_online_cpus();

        LIBCFS_ALLOC(the_lnet.ln_finalizers,
                     the_lnet.ln_nfinalizers *
                     sizeof(*the_lnet.ln_finalizers));
        if (the_lnet.ln_finalizers == NULL) {
                CERROR("Can't allocate ln_finalizers\n");
                return -ENOMEM;
        }

        for (i = 0; i < the_lnet.ln_nfinalizers; i++)
                the_lnet.ln_finalizers[i] = NULL;
#else
        the_lnet.ln_finalizing = 0;
#endif

        CFS_INIT_LIST_HEAD(&the_lnet.ln_finalizeq);
        return 0;
}

void
lnet_fini_finalizers(void)
{
#ifdef __KERNEL__
        int    i;

        for (i = 0; i < the_lnet.ln_nfinalizers; i++)
                LASSERT (the_lnet.ln_finalizers[i] == NULL);

        LIBCFS_FREE(the_lnet.ln_finalizers,
                    the_lnet.ln_nfinalizers *
                    sizeof(*the_lnet.ln_finalizers));
#else
        LASSERT (!the_lnet.ln_finalizing);
#endif
        LASSERT (list_empty(&the_lnet.ln_finalizeq));
}

#ifndef __KERNEL__
/* Temporary workaround to allow uOSS and test programs force server
 * mode in userspace. See comments near ln_server_mode_flag in
 * lnet/lib-types.h */

void
lnet_server_mode() {
        the_lnet.ln_server_mode_flag = 1;
}
#endif

int
lnet_prepare(lnet_pid_t requested_pid)
{
        /* Prepare to bring up the network */
        int               rc = 0;
        int               i;

        LASSERT (the_lnet.ln_refcount == 0);

        the_lnet.ln_routing = 0;

#ifdef __KERNEL__
        LASSERT ((requested_pid & LNET_PID_USERFLAG) == 0);
        the_lnet.ln_pid = requested_pid;
#else
        if (the_lnet.ln_server_mode_flag) {/* server case (uOSS) */
                LASSERT ((requested_pid & LNET_PID_USERFLAG) == 0);

                if (cfs_curproc_uid())/* Only root can run user-space server */
                        return -EPERM;
                the_lnet.ln_pid = requested_pid;

        } else {/* client case (liblustre) */

                /* My PID must be unique on this node and flag I'm userspace */
                //the_lnet.ln_pid = getpid() | LNET_PID_USERFLAG;
                the_lnet.ln_pid = getpid();
        }

        CERROR("my pid=%u\n", the_lnet.ln_pid);
#endif

        rc = lnet_descriptor_setup();
        if (rc != 0)
                goto failed0;

        memset(&the_lnet.ln_counters, 0,
               sizeof(the_lnet.ln_counters));

        CFS_INIT_LIST_HEAD (&the_lnet.ln_active_msgs);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_active_mds);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_active_eqs);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_test_peers);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_nis);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_zombie_nis);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_remote_nets);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_routers);

        the_lnet.ln_interface_cookie = lnet_create_interface_cookie();

        lnet_init_rtrpools();

        rc = lnet_setup_handle_hash ();
        if (rc != 0)<commit_msg>undoing r6999, because it prevents intercept client from connecting<commit_after>/* -*- mode: c; c-basic-offset: 8; indent-tabs-mode: nil; -*-
 * vim:expandtab:shiftwidth=8:tabstop=8:
 *
 * GPL HEADER START
 *
 * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
 *
 * 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.
 *
 * 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
 * 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.
 */
/*
 * This file is part of Lustre, http://www.lustre.org/
 * Lustre is a trademark of Sun Microsystems, Inc.
 */

#ifndef _XOPEN_SOURCE
#define _XOPEN_SOURCE 500
#endif

#define DEBUG_SUBSYSTEM S_LNET
#include <lnet/lib-lnet.h>

#include "psc_util/lock.h"

#ifdef __KERNEL__
#define D_LNI D_CONSOLE
#else
#define D_LNI D_CONFIG
#endif

lnet_t      the_lnet;                           /* THE state of the network */

#ifdef __KERNEL__

static char *ip2nets = "";
CFS_MODULE_PARM(ip2nets, "s", charp, 0444,
                "LNET network <- IP table");

static char *networks = "";
CFS_MODULE_PARM(networks, "s", charp, 0444,
                "local networks");

static char *routes = "";
CFS_MODULE_PARM(routes, "s", charp, 0444,
                "routes to non-local networks");

static char *portals_compatibility = "none";
CFS_MODULE_PARM(portals_compatibility, "s", charp, 0444,
                "wire protocol compatibility: 'strong'|'weak'|'none'");

char *
lnet_get_routes(void)
{
        return routes;
}

char *
lnet_get_networks(void)
{
        char   *nets;
        int     rc;

        if (*networks != 0 && *ip2nets != 0) {
                LCONSOLE_ERROR_MSG(0x101, "Please specify EITHER 'networks' or "
                                   "'ip2nets' but not both at once\n");
                return NULL;
        }

        if (*ip2nets != 0) {
                rc = lnet_parse_ip2nets(&nets, ip2nets);
                return (rc == 0) ? nets : NULL;
        }

        if (*networks != 0)
                return networks;

        return "tcp";
}

int
lnet_get_portals_compatibility(void)
{
        if (!strcmp(portals_compatibility, "none")) {
                return 0;
        }

        if (!strcmp(portals_compatibility, "weak")) {
                return 1;
                LCONSOLE_WARN("Starting in weak portals-compatible mode\n");
        }

        if (!strcmp(portals_compatibility, "strong")) {
                return 2;
                LCONSOLE_WARN("Starting in strong portals-compatible mode\n");
        }

        LCONSOLE_ERROR_MSG(0x102, "portals_compatibility=\"%s\" not supported\n",
                           portals_compatibility);
        return -EINVAL;
}

void
lnet_init_locks(void)
{
        spin_lock_init (&the_lnet.ln_lock);
        cfs_waitq_init (&the_lnet.ln_waitq);
        init_mutex(&the_lnet.ln_lnd_mutex);
        init_mutex(&the_lnet.ln_api_mutex);
}

void
lnet_fini_locks(void)
{
}

#else

int
lnet_get_usesdp(void)
{
	static psc_spinlock_t lock = LOCK_INITIALIZER;
	static int usesdp;
	static int init;

	if (init)
		return (usesdp);
	spinlock(&lock);
	if (init == 0) {
		char *val;

		val = getenv("LNET_USESDP");
		if (val && *val != '0')
			usesdp = 1;
		init = 1;
	}
	freelock(&lock);
	return (usesdp);
}

char *
lnet_get_routes(void)
{
        char *str = getenv("LNET_ROUTES");

        return (str == NULL) ? "" : str;
}

char *
lnet_get_networks (void)
{
        static char       default_networks[256];
        char             *networks = getenv ("LNET_NETWORKS");
        char             *ip2nets  = getenv ("LNET_IP2NETS");
        char             *str;
        char             *sep;
        int               len;
        int               nob;
        int               rc;
        struct list_head *tmp;

#ifdef NOT_YET
        if (networks != NULL && ip2nets != NULL) {
                LCONSOLE_ERROR_MSG(0x103, "Please set EITHER 'LNET_NETWORKS' or"
                                   " 'LNET_IP2NETS' but not both at once\n");
                return NULL;
        }

        if (ip2nets != NULL) {
                rc = lnet_parse_ip2nets(&networks, ip2nets);
                return (rc == 0) ? networks : NULL;
        }
#else
        ip2nets = NULL;
        rc = 0;
#endif
        if (networks != NULL)
                return networks;

        /* In userland, the default 'networks=' is the list of known net types */

        len = sizeof(default_networks);
        str = default_networks;
        *str = 0;
        sep = "";

        list_for_each (tmp, &the_lnet.ln_lnds) {
                lnd_t *lnd = list_entry(tmp, lnd_t, lnd_list);

                nob = snprintf(str, len, "%s%s", sep,
                               libcfs_lnd2str(lnd->lnd_type));
                len -= nob;
                if (len < 0) {
                        /* overflowed the string; leave it where it was */
                        *str = 0;
                        break;
                }

                str += nob;
                sep = ",";
        }

        return default_networks;
}

int
lnet_get_portals_compatibility(void)
{
        return 0;
}

# ifndef HAVE_LIBPTHREAD

void lnet_init_locks(void)
{
        the_lnet.ln_lock = 0;
        the_lnet.ln_lnd_mutex = 0;
        the_lnet.ln_api_mutex = 0;
}

void lnet_fini_locks(void)
{
        LASSERT (the_lnet.ln_api_mutex == 0);
        LASSERT (the_lnet.ln_lnd_mutex == 0);
        LASSERT (the_lnet.ln_lock == 0);
}

# else

void lnet_init_locks(void)
{
	pthread_mutexattr_t attr;

        pthread_cond_init(&the_lnet.ln_cond, NULL);
	pthread_mutexattr_init(&attr);
	pthread_mutexattr_settype(&attr,
	    PTHREAD_MUTEX_ERRORCHECK_NP);
        pthread_mutex_init(&the_lnet.ln_lock, &attr);
        pthread_mutex_init(&the_lnet.ln_lnd_mutex, &attr);
        pthread_mutex_init(&the_lnet.ln_api_mutex, &attr);
}

void lnet_fini_locks(void)
{
        pthread_mutex_destroy(&the_lnet.ln_api_mutex);
        pthread_mutex_destroy(&the_lnet.ln_lnd_mutex);
        pthread_mutex_destroy(&the_lnet.ln_lock);
        pthread_cond_destroy(&the_lnet.ln_cond);
}

# endif
#endif

void lnet_assert_wire_constants (void)
{
        /* Wire protocol assertions generated by 'wirecheck'
         * running on Linux robert.bartonsoftware.com 2.6.8-1.521
         * #1 Mon Aug 16 09:01:18 EDT 2004 i686 athlon i386 GNU/Linux
         * with gcc version 3.3.3 20040412 (Red Hat Linux 3.3.3-7) */

        /* Constants... */
        CLASSERT (LNET_PROTO_TCP_MAGIC == 0xeebc0ded);
        CLASSERT (LNET_PROTO_TCP_VERSION_MAJOR == 1);
        CLASSERT (LNET_PROTO_TCP_VERSION_MINOR == 0);
        CLASSERT (LNET_MSG_ACK == 0);
        CLASSERT (LNET_MSG_PUT == 1);
        CLASSERT (LNET_MSG_GET == 2);
        CLASSERT (LNET_MSG_REPLY == 3);
        CLASSERT (LNET_MSG_HELLO == 4);

        /* Checks for struct ptl_handle_wire_t */
        CLASSERT ((int)sizeof(lnet_handle_wire_t) == 16);
        CLASSERT ((int)offsetof(lnet_handle_wire_t, wh_interface_cookie) == 0);
        CLASSERT ((int)sizeof(((lnet_handle_wire_t *)0)->wh_interface_cookie) == 8);
        CLASSERT ((int)offsetof(lnet_handle_wire_t, wh_object_cookie) == 8);
        CLASSERT ((int)sizeof(((lnet_handle_wire_t *)0)->wh_object_cookie) == 8);

        /* Checks for struct lnet_magicversion_t */
        CLASSERT ((int)sizeof(lnet_magicversion_t) == 8);
        CLASSERT ((int)offsetof(lnet_magicversion_t, magic) == 0);
        CLASSERT ((int)sizeof(((lnet_magicversion_t *)0)->magic) == 4);
        CLASSERT ((int)offsetof(lnet_magicversion_t, version_major) == 4);
        CLASSERT ((int)sizeof(((lnet_magicversion_t *)0)->version_major) == 2);
        CLASSERT ((int)offsetof(lnet_magicversion_t, version_minor) == 6);
        CLASSERT ((int)sizeof(((lnet_magicversion_t *)0)->version_minor) == 2);

        /* Checks for struct lnet_hdr_t */
        CLASSERT ((int)sizeof(lnet_hdr_t) == 72);
        CLASSERT ((int)offsetof(lnet_hdr_t, dest_nid) == 0);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->dest_nid) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, src_nid) == 8);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->src_nid) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, dest_pid) == 16);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->dest_pid) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, src_pid) == 20);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->src_pid) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, type) == 24);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->type) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, payload_length) == 28);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->payload_length) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg) == 40);

        /* Ack */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.ack.dst_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.ack.dst_wmd) == 16);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.ack.match_bits) == 48);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.ack.match_bits) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.ack.mlength) == 56);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.ack.mlength) == 4);

        /* Put */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.ack_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.ack_wmd) == 16);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.match_bits) == 48);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.match_bits) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.hdr_data) == 56);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.hdr_data) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.ptl_index) == 64);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.ptl_index) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.put.offset) == 68);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.put.offset) == 4);

        /* Get */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.return_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.return_wmd) == 16);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.match_bits) == 48);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.match_bits) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.ptl_index) == 56);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.ptl_index) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.src_offset) == 60);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.src_offset) == 4);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.get.sink_length) == 64);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.get.sink_length) == 4);

        /* Reply */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.reply.dst_wmd) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.reply.dst_wmd) == 16);

        /* Hello */
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.hello.incarnation) == 32);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.hello.incarnation) == 8);
        CLASSERT ((int)offsetof(lnet_hdr_t, msg.hello.type) == 40);
        CLASSERT ((int)sizeof(((lnet_hdr_t *)0)->msg.hello.type) == 4);
}

lnd_t *
lnet_find_lnd_by_type (int type)
{
        lnd_t              *lnd;
        struct list_head   *tmp;

        /* holding lnd mutex */
        list_for_each (tmp, &the_lnet.ln_lnds) {
                lnd = list_entry(tmp, lnd_t, lnd_list);

                if (lnd->lnd_type == (unsigned int)type)
                        return lnd;
        }

        return NULL;
}

void
lnet_register_lnd (lnd_t *lnd)
{
        LNET_MUTEX_DOWN(&the_lnet.ln_lnd_mutex);

        LASSERT (the_lnet.ln_init);
        LASSERT (libcfs_isknown_lnd(lnd->lnd_type));
        LASSERT (lnet_find_lnd_by_type(lnd->lnd_type) == NULL);

        list_add_tail (&lnd->lnd_list, &the_lnet.ln_lnds);
        lnd->lnd_refcount = 0;

        CDEBUG(D_NET, "%s LND registered\n", libcfs_lnd2str(lnd->lnd_type));

        LNET_MUTEX_UP(&the_lnet.ln_lnd_mutex);
}

void
lnet_unregister_lnd (lnd_t *lnd)
{
        LNET_MUTEX_DOWN(&the_lnet.ln_lnd_mutex);

        LASSERT (the_lnet.ln_init);
        LASSERT (lnet_find_lnd_by_type(lnd->lnd_type) == lnd);
        LASSERT (lnd->lnd_refcount == 0);

        list_del (&lnd->lnd_list);
        CDEBUG(D_NET, "%s LND unregistered\n", libcfs_lnd2str(lnd->lnd_type));

        LNET_MUTEX_UP(&the_lnet.ln_lnd_mutex);
}

#ifndef LNET_USE_LIB_FREELIST

int
lnet_descriptor_setup (void)
{
        return 0;
}

void
lnet_descriptor_cleanup (void)
{
}

#else

int
lnet_freelist_init (lnet_freelist_t *fl, int n, int size)
{
        char *space;

        LASSERT (n > 0);

        size += offsetof (lnet_freeobj_t, fo_contents);

        LIBCFS_ALLOC(space, n * size);
        if (space == NULL)
                return (-ENOMEM);

        CFS_INIT_LIST_HEAD (&fl->fl_list);
        fl->fl_objs = space;
        fl->fl_nobjs = n;
        fl->fl_objsize = size;

        do
        {
                memset (space, 0, size);
                list_add ((struct list_head *)space, &fl->fl_list);
                space += size;
        } while (--n != 0);

        return (0);
}

void
lnet_freelist_fini (lnet_freelist_t *fl)
{
        struct list_head *el;
        int               count;

        if (fl->fl_nobjs == 0)
                return;

        count = 0;
        for (el = fl->fl_list.next; el != &fl->fl_list; el = el->next)
                count++;

        LASSERT (count == fl->fl_nobjs);

        LIBCFS_FREE(fl->fl_objs, fl->fl_nobjs * fl->fl_objsize);
        memset (fl, 0, sizeof (*fl));
}

int
lnet_descriptor_setup (void)
{
        /* NB on failure caller must still call lnet_descriptor_cleanup */
        /*               ******                                         */
        int        rc;

        memset (&the_lnet.ln_free_mes,  0, sizeof (the_lnet.ln_free_mes));
        memset (&the_lnet.ln_free_msgs, 0, sizeof (the_lnet.ln_free_msgs));
        memset (&the_lnet.ln_free_mds,  0, sizeof (the_lnet.ln_free_mds));
        memset (&the_lnet.ln_free_eqs,  0, sizeof (the_lnet.ln_free_eqs));

        rc = lnet_freelist_init(&the_lnet.ln_free_mes,
                                MAX_MES, sizeof (lnet_me_t));
        if (rc != 0)
                return (rc);

        rc = lnet_freelist_init(&the_lnet.ln_free_msgs,
                                MAX_MSGS, sizeof (lnet_msg_t));
        if (rc != 0)
                return (rc);

        rc = lnet_freelist_init(&the_lnet.ln_free_mds,
                                MAX_MDS, sizeof (lnet_libmd_t));
        if (rc != 0)
                return (rc);

        rc = lnet_freelist_init(&the_lnet.ln_free_eqs,
                                MAX_EQS, sizeof (lnet_eq_t));
        return (rc);
}

void
lnet_descriptor_cleanup (void)
{
        lnet_freelist_fini (&the_lnet.ln_free_mes);
        lnet_freelist_fini (&the_lnet.ln_free_msgs);
        lnet_freelist_fini (&the_lnet.ln_free_mds);
        lnet_freelist_fini (&the_lnet.ln_free_eqs);
}

#endif

__u64
lnet_create_interface_cookie (void)
{
        /* NB the interface cookie in wire handles guards against delayed
         * replies and ACKs appearing valid after reboot. Initialisation time,
         * even if it's only implemented to millisecond resolution is probably
         * easily good enough. */
        struct timeval tv;
        __u64          cookie;
#ifndef __KERNEL__
        int            rc = gettimeofday (&tv, NULL);
        LASSERT (rc == 0);
#else
        do_gettimeofday(&tv);
#endif
        cookie = tv.tv_sec;
        cookie *= 1000000;
        cookie += tv.tv_usec;
        return cookie;
}

int
lnet_setup_handle_hash (void)
{
        int       i;

        /* Arbitrary choice of hash table size */
#ifdef __KERNEL__
        the_lnet.ln_lh_hash_size = CFS_PAGE_SIZE / sizeof (struct list_head);
#else
        the_lnet.ln_lh_hash_size = (MAX_MES + MAX_MDS + MAX_EQS)/4;
#endif
        LIBCFS_ALLOC(the_lnet.ln_lh_hash_table,
                     the_lnet.ln_lh_hash_size * sizeof (struct list_head));
        if (the_lnet.ln_lh_hash_table == NULL)
                return (-ENOMEM);

        for (i = 0; i < the_lnet.ln_lh_hash_size; i++)
                CFS_INIT_LIST_HEAD (&the_lnet.ln_lh_hash_table[i]);

        the_lnet.ln_next_object_cookie = LNET_COOKIE_TYPES;

        return (0);
}

void
lnet_cleanup_handle_hash (void)
{
        if (the_lnet.ln_lh_hash_table == NULL)
                return;

        LIBCFS_FREE(the_lnet.ln_lh_hash_table,
                    the_lnet.ln_lh_hash_size * sizeof (struct list_head));
}

lnet_libhandle_t *
lnet_lookup_cookie (__u64 cookie, int type)
{
        /* ALWAYS called with LNET_LOCK held */
        struct list_head    *list;
        struct list_head    *el;
        unsigned int         hash;

        if ((cookie & (LNET_COOKIE_TYPES - 1)) != (__u64)type)
                return (NULL);

        hash = ((unsigned int)cookie) % the_lnet.ln_lh_hash_size;
        list = &the_lnet.ln_lh_hash_table[hash];

        list_for_each (el, list) {
                lnet_libhandle_t *lh = list_entry (el, lnet_libhandle_t,
                                                  lh_hash_chain);

                if (lh->lh_cookie == cookie)
                        return (lh);
        }

        return (NULL);
}

void
lnet_initialise_handle (lnet_libhandle_t *lh, int type)
{
        /* ALWAYS called with LNET_LOCK held */
        unsigned int    hash;

        LASSERT (type >= 0 && type < LNET_COOKIE_TYPES);
        lh->lh_cookie = the_lnet.ln_next_object_cookie | type;
        the_lnet.ln_next_object_cookie += LNET_COOKIE_TYPES;

        hash = ((unsigned int)lh->lh_cookie) % the_lnet.ln_lh_hash_size;
        list_add (&lh->lh_hash_chain, &the_lnet.ln_lh_hash_table[hash]);
}

void
lnet_invalidate_handle (lnet_libhandle_t *lh)
{
        /* ALWAYS called with LNET_LOCK held */
        list_del (&lh->lh_hash_chain);
}

int
lnet_init_finalizers(void)
{
#ifdef __KERNEL__
        int    i;

        the_lnet.ln_nfinalizers = num_online_cpus();

        LIBCFS_ALLOC(the_lnet.ln_finalizers,
                     the_lnet.ln_nfinalizers *
                     sizeof(*the_lnet.ln_finalizers));
        if (the_lnet.ln_finalizers == NULL) {
                CERROR("Can't allocate ln_finalizers\n");
                return -ENOMEM;
        }

        for (i = 0; i < the_lnet.ln_nfinalizers; i++)
                the_lnet.ln_finalizers[i] = NULL;
#else
        the_lnet.ln_finalizing = 0;
#endif

        CFS_INIT_LIST_HEAD(&the_lnet.ln_finalizeq);
        return 0;
}

void
lnet_fini_finalizers(void)
{
#ifdef __KERNEL__
        int    i;

        for (i = 0; i < the_lnet.ln_nfinalizers; i++)
                LASSERT (the_lnet.ln_finalizers[i] == NULL);

        LIBCFS_FREE(the_lnet.ln_finalizers,
                    the_lnet.ln_nfinalizers *
                    sizeof(*the_lnet.ln_finalizers));
#else
        LASSERT (!the_lnet.ln_finalizing);
#endif
        LASSERT (list_empty(&the_lnet.ln_finalizeq));
}

#ifndef __KERNEL__
/* Temporary workaround to allow uOSS and test programs force server
 * mode in userspace. See comments near ln_server_mode_flag in
 * lnet/lib-types.h */

void
lnet_server_mode() {
        the_lnet.ln_server_mode_flag = 1;
}
#endif

int
lnet_prepare(lnet_pid_t requested_pid)
{
        /* Prepare to bring up the network */
        int               rc = 0;
        int               i;

        LASSERT (the_lnet.ln_refcount == 0);

        the_lnet.ln_routing = 0;

#ifdef __KERNEL__
        LASSERT ((requested_pid & LNET_PID_USERFLAG) == 0);
        the_lnet.ln_pid = requested_pid;
#else
        if (the_lnet.ln_server_mode_flag) {/* server case (uOSS) */
                LASSERT ((requested_pid & LNET_PID_USERFLAG) == 0);

                if (cfs_curproc_uid())/* Only root can run user-space server */
                        return -EPERM;
                the_lnet.ln_pid = requested_pid;

        } else {/* client case (liblustre) */

                /* My PID must be unique on this node and flag I'm userspace */
                the_lnet.ln_pid = getpid() | LNET_PID_USERFLAG;
                //the_lnet.ln_pid = getpid();
        }

        CERROR("my pid=%u\n", the_lnet.ln_pid);
#endif

        rc = lnet_descriptor_setup();
        if (rc != 0)
                goto failed0;

        memset(&the_lnet.ln_counters, 0,
               sizeof(the_lnet.ln_counters));

        CFS_INIT_LIST_HEAD (&the_lnet.ln_active_msgs);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_active_mds);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_active_eqs);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_test_peers);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_nis);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_zombie_nis);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_remote_nets);
        CFS_INIT_LIST_HEAD (&the_lnet.ln_routers);

        the_lnet.ln_interface_cookie = lnet_create_interface_cookie();

        lnet_init_rtrpools();

        rc = lnet_setup_handle_hash ();
        if (rc != 0)
{'fraction_non_alphanumeric': 0.12280016541837063, 'fraction_numerical': 0.014474107430041815, 'mean_word_length': 2.3017522566942272, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 123, 'https://': 0, 'lorem ipsum': 0, 'www.': 6, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4787909', 'n_tokens_mistral': 16222, 'n_tokens_neox': 15307, 'n_words': 3061}