erl_node_tables.c

OTP是开放电信平台的简称

/* ``The contents of this file are subject to the Erlang Public License,
 * Version 1.1, (the "License"); you may not use this file except in
 * compliance with the License. You should have received a copy of the
 * Erlang Public License along with this software. If not, it can be
 * retrieved via the world wide web at http://www.erlang.org/.
 * 
 * Software distributed under the License is distributed on an "AS IS"
 * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See
 * the License for the specific language governing rights and limitations
 * under the License.
 * 
 * The Initial Developer of the Original Code is Ericsson Utvecklings AB.
 * Portions created by Ericsson are Copyright 1999, Ericsson Utvecklings
 * AB. All Rights Reserved.''
 * 
 *     $Id$
 */

#ifdef HAVE_CONFIG_H
#  include "config.h"
#endif

#include "global.h"
#include "erl_node_tables.h"
#include "dist.h"
#include "big.h"
#include "error.h"

Hash erts_dist_table;
Hash erts_node_table;
erts_smp_mtx_t erts_dist_table_mtx;
erts_smp_mtx_t erts_node_table_mtx;

DistEntry *erts_hidden_dist_entries;
DistEntry *erts_visible_dist_entries;
DistEntry *erts_not_connected_dist_entries;
Sint erts_no_of_hidden_dist_entries;
Sint erts_no_of_visible_dist_entries;
Sint erts_no_of_not_connected_dist_entries;

DistEntry *erts_this_dist_entry;
ErlNode *erts_this_node;

static Uint node_entries;
static Uint dist_entries;

#ifdef ERTS_SMP
#define ERTS_NO_OF_DIST_ENTRY_MUTEXES 32
static erts_smp_mtx_t dist_entry_mutexes[ERTS_NO_OF_DIST_ENTRY_MUTEXES];
#endif

static int references_atoms_need_init = 1;

/* -- The distribution table ---------------------------------------------- */

#ifdef DEBUG
static int
is_in_de_list(DistEntry *dep, DistEntry *dep_list)
{
    DistEntry *tdep;
    for(tdep = dep_list; tdep; tdep = tdep->next)
	if(tdep == dep)
	    return 1;
    return 0;
}
#endif

static HashValue
dist_table_hash(void *dep)
{
    return atom_tab(atom_val(((DistEntry *) dep)->sysname))->slot.bucket.hvalue;
}

static int
dist_table_cmp(void *dep1, void *dep2)
{
    return (((DistEntry *) dep1)->sysname == ((DistEntry *) dep2)->sysname
	    ? 0 : 1);
}

static void*
dist_table_alloc(void *dep_tmpl)
{
    DistEntry *dep;

    if(((DistEntry *) dep_tmpl) == erts_this_dist_entry)
	return dep_tmpl;

    dep = (DistEntry *) erts_alloc(ERTS_ALC_T_DIST_ENTRY, sizeof(DistEntry));

    dist_entries++;

    dep->prev       = NULL;
    erts_refc_init(&dep->refc, -1);
    dep->sysname    = ((DistEntry *) dep_tmpl)->sysname;
    dep->cid        = NIL;
    dep->nlinks     = NULL;
    dep->node_links = NULL;
    dep->monitors   = NULL;
    dep->status     = 0;
    dep->flags      = 0;
    dep->cache      = NULL;
    dep->version    = 0;
#ifdef ERTS_SMP
    dep->mtxp       = &dist_entry_mutexes[(atom_val(dep->sysname)
					   % ERTS_NO_OF_DIST_ENTRY_MUTEXES)];
#endif

    /* Link in */

    /* All new dist entries are "not connected" */
    dep->next = erts_not_connected_dist_entries;
    if(erts_not_connected_dist_entries) {
	ASSERT(erts_not_connected_dist_entries->prev == NULL);
	erts_not_connected_dist_entries->prev = dep;
    }
    erts_not_connected_dist_entries = dep;
    erts_no_of_not_connected_dist_entries++;

    return (void *) dep;
}

static void
dist_table_free(void *vdep)
{
    DistEntry *dep = (DistEntry *) vdep;

    if(dep == erts_this_dist_entry)
	return;

    ASSERT(is_nil(dep->cid));
    ASSERT(dep->nlinks == NULL);
    ASSERT(dep->node_links == NULL);
    ASSERT(dep->monitors == NULL);

    /* Link out */

    /* All dist entries about to be removed are "not connected" */

    if(dep->prev) {
	ASSERT(is_in_de_list(dep, erts_not_connected_dist_entries));
	dep->prev->next = dep->next;
    }
    else {
	ASSERT(erts_not_connected_dist_entries == dep);
	erts_not_connected_dist_entries = dep->next;
    }

    if(dep->next)
	dep->next->prev = dep->prev;

    ASSERT(erts_no_of_not_connected_dist_entries > 0);
    erts_no_of_not_connected_dist_entries--;

    delete_cache((DistEntry *) dep);
#ifdef DEBUG
    sys_memset(vdep, 0x77, sizeof(DistEntry));
#endif
    erts_free(ERTS_ALC_T_DIST_ENTRY, (void *) dep);

    ASSERT(dist_entries > 1);
    dist_entries--;
}


void
erts_dist_table_info(int to, void *to_arg)
{
    int lock = !ERTS_IS_CRASH_DUMPING;
    if (lock)
	erts_smp_mtx_lock(&erts_dist_table_mtx);
    hash_info(to, to_arg, &erts_dist_table);
    if (lock)
	erts_smp_mtx_unlock(&erts_dist_table_mtx);
}

DistEntry *
erts_channel_no_to_dist_entry(Uint cno)
{
/*
 * For this node (and previous incarnations of this node),
 * ERST_INTERNAL_CHANNEL_NO (will always be 0 I guess) is used as
 * channel no. For other nodes, the atom index of the atom corresponding
 * to the node name is used as channel no.
 */
    if(cno == ERST_INTERNAL_CHANNEL_NO) {
	erts_refc_inc(&erts_this_dist_entry->refc, 2);
	return erts_this_dist_entry;
    }

    if((cno > MAX_ATOM_INDEX)
       || (cno >= atom_table_size())
       ||  (atom_tab(cno) == NULL))
	return NULL;
        
    /* cno is a valid atom index; find corresponding dist entry (if there
       is one) */
    return erts_find_dist_entry(make_atom(cno));
}


DistEntry *
erts_sysname_to_connected_dist_entry(Eterm sysname)
{
    DistEntry de;
    DistEntry *res_dep;
    de.sysname = sysname;
  
    if(erts_this_dist_entry->sysname == sysname) {
	erts_refc_inc(&erts_this_dist_entry->refc, 2);
	return erts_this_dist_entry;
    }

    erts_smp_mtx_lock(&erts_dist_table_mtx);
    res_dep = (DistEntry *) hash_get(&erts_dist_table, (void *) &de);
    if (res_dep) {
	long refc = erts_refc_inctest(&res_dep->refc, 1);
	if (refc < 2) /* Pending delete */
	    erts_refc_inc(&res_dep->refc, 1);
    }
    erts_smp_mtx_unlock(&erts_dist_table_mtx);
    if (res_dep) {
	erts_smp_mtx_t *mtxp;
#ifdef ERTS_SMP
	mtxp = res_dep->mtxp;
#else
	mtxp = NULL;
#endif
	erts_smp_mtx_lock(mtxp);
	if (is_nil(res_dep->cid)) {
	    erts_deref_dist_entry(res_dep);
	    res_dep = NULL;
	}
	erts_smp_mtx_unlock(mtxp);
    }
    return res_dep;
}

DistEntry *erts_find_or_insert_dist_entry(Eterm sysname)
{
    DistEntry *res;
    DistEntry de;
    long refc;
    de.sysname = sysname;
    erts_smp_mtx_lock(&erts_dist_table_mtx);
    res = hash_put(&erts_dist_table, (void *) &de);
    refc = erts_refc_inctest(&res->refc, 0);
    if (refc < 2) /* New or pending delete */
	erts_refc_inc(&res->refc, 1);
    erts_smp_mtx_unlock(&erts_dist_table_mtx);
    return res;
}

DistEntry *erts_find_dist_entry(Eterm sysname)
{
    DistEntry *res;
    DistEntry de;
    de.sysname = sysname;
    erts_smp_mtx_lock(&erts_dist_table_mtx);
    res = hash_get(&erts_dist_table, (void *) &de);
    if (res) {
	long refc = erts_refc_inctest(&res->refc, 1);
	if (refc < 2) /* Pending delete */
	    erts_refc_inc(&res->refc, 1);
    }
    erts_smp_mtx_unlock(&erts_dist_table_mtx);
    return res;
}

void erts_delete_dist_entry(DistEntry *dep)
{
    ASSERT(dep != erts_this_dist_entry);
    if(dep != erts_this_dist_entry) {
	erts_smp_mtx_lock(&erts_dist_table_mtx);
	/*
	 * Another thread might have looked up this dist entry after
	 * we decided to delete it (refc became zero). If so, the other
	 * thread incremented refc twice. Once for the new reference
	 * and once for this thread. Therefore, delete dist entry if
	 * refc is 0 or -1 after a decrement.
	 */
	if (erts_refc_dectest(&dep->refc, -1) <= 0)
	    (void) hash_erase(&erts_dist_table, (void *) dep);
	erts_smp_mtx_unlock(&erts_dist_table_mtx);
    }
}

Uint
erts_dist_table_size(void)
{
    Uint res;
#ifdef DEBUG
    HashInfo hi;
    DistEntry *dep;
    int i;
#endif
    int lock = !ERTS_IS_CRASH_DUMPING;

    if (lock)
	erts_smp_mtx_lock(&erts_dist_table_mtx);
#ifdef DEBUG
    hash_get_info(&hi, &erts_dist_table);
    ASSERT(dist_entries == hi.objs);

    i = 0;
    for(dep = erts_visible_dist_entries; dep; dep = dep->next)
	i++;
    ASSERT(i == erts_no_of_visible_dist_entries);
    i = 0;
    for(dep = erts_hidden_dist_entries; dep; dep = dep->next)
	i++;
    ASSERT(i == erts_no_of_hidden_dist_entries);
    i = 0;
    for(dep = erts_not_connected_dist_entries; dep; dep = dep->next)
	i++;
    ASSERT(i == erts_no_of_not_connected_dist_entries);

    ASSERT(dist_entries == (erts_no_of_visible_dist_entries
			    + erts_no_of_hidden_dist_entries
			    + erts_no_of_not_connected_dist_entries
			    + 1 /* erts_this_dist_entry */));
#endif

    res = (hash_table_sz(&erts_dist_table)
	   + dist_entries*sizeof(DistEntry)
	   + erts_dist_cache_size());
    if (lock)
	erts_smp_mtx_unlock(&erts_dist_table_mtx);
    return res;
}

void
erts_set_dist_entry_not_connected(DistEntry *dep)
{
    ERTS_SMP_LC_ASSERT(erts_lc_is_dist_entry_locked(dep));
    erts_smp_mtx_lock(&erts_dist_table_mtx);

    ASSERT(dep != erts_this_dist_entry);
    ASSERT(is_internal_port(dep->cid));

    if(dep->flags & DFLAG_PUBLISHED) {
	if(dep->prev) {
	    ASSERT(is_in_de_list(dep, erts_visible_dist_entries));
	    dep->prev->next = dep->next;
	}
	else {
	    ASSERT(erts_visible_dist_entries == dep);
	    erts_visible_dist_entries = dep->next;
	}

	ASSERT(erts_no_of_visible_dist_entries > 0);
	erts_no_of_visible_dist_entries--;
    }
    else {
	if(dep->prev) {
	    ASSERT(is_in_de_list(dep, erts_hidden_dist_entries));
	    dep->prev->next = dep->next;
	}
	else {
	    ASSERT(erts_hidden_dist_entries == dep);
	    erts_hidden_dist_entries = dep->next;
	}

	ASSERT(erts_no_of_hidden_dist_entries > 0);
	erts_no_of_hidden_dist_entries--;
    }

    if(dep->next)
	dep->next->prev = dep->prev;

    dep->status &= ~ERTS_DE_SFLG_CONNECTED;
    dep->flags = 0;
    dep->prev = NULL;
    dep->cid = NIL;

    dep->next = erts_not_connected_dist_entries;
    if(erts_not_connected_dist_entries) {
	ASSERT(erts_not_connected_dist_entries->prev == NULL);
	erts_not_connected_dist_entries->prev = dep;
    }
    erts_not_connected_dist_entries = dep;
    erts_no_of_not_connected_dist_entries++;
    erts_smp_mtx_unlock(&erts_dist_table_mtx);
}

void
erts_set_dist_entry_connected(DistEntry *dep, Eterm cid, Uint flags)
{
    ERTS_SMP_LC_ASSERT(erts_lc_is_dist_entry_locked(dep));
    erts_smp_mtx_lock(&erts_dist_table_mtx);

    ASSERT(dep != erts_this_dist_entry);
    ASSERT(is_nil(dep->cid));
    ASSERT(is_internal_port(cid));

    if(dep->prev) {
	ASSERT(is_in_de_list(dep, erts_not_connected_dist_entries));
	dep->prev->next = dep->next;
    }
    else {
	ASSERT(erts_not_connected_dist_entries == dep);
	erts_not_connected_dist_entries = dep->next;
    }

    if(dep->next)
	dep->next->prev = dep->prev;

    ASSERT(erts_no_of_not_connected_dist_entries > 0);
    erts_no_of_not_connected_dist_entries--;

    dep->status |= ERTS_DE_SFLG_CONNECTED;
    dep->flags = flags;
    dep->cid = cid;
    dep->prev = NULL;

    if(flags & DFLAG_PUBLISHED) {
	dep->next = erts_visible_dist_entries;
	if(erts_visible_dist_entries) {
	    ASSERT(erts_visible_dist_entries->prev == NULL);
	    erts_visible_dist_entries->prev = dep;
	}
	erts_visible_dist_entries = dep;
	erts_no_of_visible_dist_entries++;
    }
    else {
	dep->next = erts_hidden_dist_entries;
	if(erts_hidden_dist_entries) {
	    ASSERT(erts_hidden_dist_entries->prev == NULL);
	    erts_hidden_dist_entries->prev = dep;
	}
	erts_hidden_dist_entries = dep;
	erts_no_of_hidden_dist_entries++;
    }
    erts_smp_mtx_unlock(&erts_dist_table_mtx);
}

/* -- Node table --------------------------------------------------------- */

/* Some large primes */
#define PRIME0 ((HashValue) 268438039)
#define PRIME1 ((HashValue) 268440479)
#define PRIME2 ((HashValue) 268439161)
#define PRIME3 ((HashValue) 268437017)

static HashValue
node_table_hash(void *venp)
{
    Uint32 cre = ((ErlNode *) venp)->creation;
    HashValue h = atom_tab(atom_val(((ErlNode *) venp)->sysname))->slot.bucket.hvalue;

    h *= PRIME0;
    h += cre & 0xff;

#if MAX_CREATION >= (1 << 8)
    h *= PRIME1;
    h += (cre >> 8) & 0xff;
#endif

#if MAX_CREATION >= (1 << 16)
    h *= PRIME2;
    h += (cre >> 16) & 0xff;
#endif

#if MAX_CREATION >= (1 << 24)
    h *= PRIME3;
    h += (cre >> 24) & 0xff;
#endif

#if 0
/* XXX Problems in older versions of GCC */
 #if MAX_CREATION >= (1UL << 32)
 #error "MAX_CREATION larger than size of expected creation storage (Uint32)"
 #endif
#endif
    return h;
}

static int
node_table_cmp(void *venp1, void *venp2)
{
    return ((((ErlNode *) venp1)->sysname == ((ErlNode *) venp2)->sysname
	     && ((ErlNode *) venp1)->creation == ((ErlNode *) venp2)->creation)
	    ? 0
	    : 1);
}

static void*
node_table_alloc(void *venp_tmpl)
{
    ErlNode *enp;

    if(((ErlNode *) venp_tmpl) == erts_this_node)
	return venp_tmpl;

    enp = (ErlNode *) erts_alloc(ERTS_ALC_T_NODE_ENTRY, sizeof(ErlNode));

    node_entries++;

    erts_refc_init(&enp->refc, -1);
    enp->creation = ((ErlNode *) venp_tmpl)->creation;
    enp->sysname = ((ErlNode *) venp_tmpl)->sysname;
    enp->dist_entry = erts_find_or_insert_dist_entry(((ErlNode *) venp_tmpl)->sysname);

    return (void *) enp;
}

static void
node_table_free(void *venp)
{
    ErlNode *enp = (ErlNode *) venp;

    if(enp == erts_this_node)
	return;

    erts_deref_dist_entry(enp->dist_entry);
#ifdef DEBUG
    sys_memset(venp, 0x55, sizeof(ErlNode));
#endif
    erts_free(ERTS_ALC_T_NODE_ENTRY, venp);

    ASSERT(node_entries > 1);
    node_entries--;
}

Uint
erts_node_table_size(void)
{
    Uint res;
#ifdef DEBUG
    HashInfo hi;
#endif
    int lock = !ERTS_IS_CRASH_DUMPING;
    if (lock)
	erts_smp_mtx_lock(&erts_node_table_mtx);
#ifdef DEBUG
    hash_get_info(&hi, &erts_node_table);
    ASSERT(node_entries == hi.objs);