router.c

来自「lustre 1.6.5 source code」· C语言 代码 · 共 1,077 行 · 第 1/3 页

        /* The unlink event callback will signal final completion */

	return 0;
}


void
lnet_wait_known_routerstate(void)
{
        lnet_peer_t         *rtr;
        struct list_head    *entry;
        int                  all_known;

        for (;;) {
                LNET_LOCK();
                
                all_known = 1;
                list_for_each (entry, &the_lnet.ln_routers) {
                        rtr = list_entry(entry, lnet_peer_t, lp_rtr_list);
                
                        if (rtr->lp_alive_count == 0) {
                                all_known = 0;
                                break;
                        }
                }

                LNET_UNLOCK();

                if (all_known)
                        return;

                cfs_pause(cfs_time_seconds(1));
        }
}

void
lnet_router_checker_stop(void)
{
        int       rc;

        LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING ||
                 the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN);

        if (the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN)
                return;

        the_lnet.ln_rc_state = LNET_RC_STATE_STOPTHREAD;
	/* block until event callback signals exit */
	mutex_down(&the_lnet.ln_rc_signal);

        LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_UNLINKED);

        rc = LNetEQFree(the_lnet.ln_rc_eqh);
        LASSERT (rc == 0);
        
        the_lnet.ln_rc_state = LNET_RC_STATE_SHUTDOWN;
}

int
lnet_router_checker_start(void)
{
        int  rc;

        LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_SHUTDOWN);

        if (check_routers_before_use &&
            dead_router_check_interval <= 0) {
                LCONSOLE_ERROR_MSG(0x10a, "'dead_router_check_interval' must be"
                                   " set if 'check_routers_before_use' is set"
                                   "\n");
                return -EINVAL;
        }
        
        if (live_router_check_interval <= 0 &&
            dead_router_check_interval <= 0)
                return 0;

	init_mutex_locked(&the_lnet.ln_rc_signal);

        /* EQ size doesn't matter; the callback is guaranteed to get every
         * event */
        rc = LNetEQAlloc(1, lnet_router_checker_event,
                         &the_lnet.ln_rc_eqh);
        if (rc != 0) {
                CERROR("Can't allocate EQ: %d\n", rc);
                return -ENOMEM;
        }

	rc = (int)cfs_kernel_thread(lnet_router_checker, NULL, 0);
	if (rc < 0) {
		CERROR("Can't start router checker thread: %d\n", rc);
                goto failed;
	}

	mutex_down(&the_lnet.ln_rc_signal);     /* wait for checker to startup */

        rc = the_lnet.ln_rc_state;
        if (rc < 0) {
                the_lnet.ln_rc_state = LNET_RC_STATE_SHUTDOWN;
                goto failed;
        }
        
        LASSERT (the_lnet.ln_rc_state == LNET_RC_STATE_RUNNING);

        if (check_routers_before_use) {
                /* Note that a helpful side-effect of pinging all known routers
                 * at startup is that it makes them drop stale connections they
                 * may have to a previous instance of me. */
                lnet_wait_known_routerstate();
        }
        
        return 0;
        
 failed:
        rc = LNetEQFree(the_lnet.ln_rc_eqh);
        LASSERT (rc == 0);
        return rc;
}

void
lnet_destroy_rtrbuf(lnet_rtrbuf_t *rb, int npages)
{
        int sz = offsetof(lnet_rtrbuf_t, rb_kiov[npages]);

        while (--npages >= 0)
                cfs_free_page(rb->rb_kiov[npages].kiov_page);

        LIBCFS_FREE(rb, sz);
}

lnet_rtrbuf_t *
lnet_new_rtrbuf(lnet_rtrbufpool_t *rbp)
{
        int            npages = rbp->rbp_npages;
        int            sz = offsetof(lnet_rtrbuf_t, rb_kiov[npages]);
        struct page   *page;
        lnet_rtrbuf_t *rb;
        int            i;

        LIBCFS_ALLOC(rb, sz);
        if (rb == NULL)
                return NULL;

        rb->rb_pool = rbp;

        for (i = 0; i < npages; i++) {
                page = cfs_alloc_page(CFS_ALLOC_ZERO | CFS_ALLOC_STD);
                if (page == NULL) {
                        while (--i >= 0)
                                cfs_free_page(rb->rb_kiov[i].kiov_page);

                        LIBCFS_FREE(rb, sz);
                        return NULL;
                }

                rb->rb_kiov[i].kiov_len = CFS_PAGE_SIZE;
                rb->rb_kiov[i].kiov_offset = 0;
                rb->rb_kiov[i].kiov_page = page;
        }

        return rb;
}

void
lnet_rtrpool_free_bufs(lnet_rtrbufpool_t *rbp)
{
        int            npages = rbp->rbp_npages;
        int            nbuffers = 0;
        lnet_rtrbuf_t *rb;

        LASSERT (list_empty(&rbp->rbp_msgs));
        LASSERT (rbp->rbp_credits == rbp->rbp_nbuffers);

        while (!list_empty(&rbp->rbp_bufs)) {
                LASSERT (rbp->rbp_credits > 0);

                rb = list_entry(rbp->rbp_bufs.next,
                                lnet_rtrbuf_t, rb_list);
                list_del(&rb->rb_list);
                lnet_destroy_rtrbuf(rb, npages);
                nbuffers++;
        }

        LASSERT (rbp->rbp_nbuffers == nbuffers);
        LASSERT (rbp->rbp_credits == nbuffers);

        rbp->rbp_nbuffers = rbp->rbp_credits = 0;
}

int
lnet_rtrpool_alloc_bufs(lnet_rtrbufpool_t *rbp, int nbufs)
{
        lnet_rtrbuf_t *rb;
        int            i;

        if (rbp->rbp_nbuffers != 0) {
                LASSERT (rbp->rbp_nbuffers == nbufs);
                return 0;
        }
        
        for (i = 0; i < nbufs; i++) {
                rb = lnet_new_rtrbuf(rbp);

                if (rb == NULL) {
                        CERROR("Failed to allocate %d router bufs of %d pages\n",
                               nbufs, rbp->rbp_npages);
                        return -ENOMEM;
                }

                rbp->rbp_nbuffers++;
                rbp->rbp_credits++;
                rbp->rbp_mincredits++;
                list_add(&rb->rb_list, &rbp->rbp_bufs);

                /* No allocation "under fire" */
                /* Otherwise we'd need code to schedule blocked msgs etc */
                LASSERT (!the_lnet.ln_routing);
        }

        LASSERT (rbp->rbp_credits == nbufs);
        return 0;
}

void
lnet_rtrpool_init(lnet_rtrbufpool_t *rbp, int npages)
{
        CFS_INIT_LIST_HEAD(&rbp->rbp_msgs);
        CFS_INIT_LIST_HEAD(&rbp->rbp_bufs);

        rbp->rbp_npages = npages;
        rbp->rbp_credits = 0;
        rbp->rbp_mincredits = 0;
}

void
lnet_free_rtrpools(void)
{
        lnet_rtrpool_free_bufs(&the_lnet.ln_rtrpools[0]);
        lnet_rtrpool_free_bufs(&the_lnet.ln_rtrpools[1]);
        lnet_rtrpool_free_bufs(&the_lnet.ln_rtrpools[2]);
}

void
lnet_init_rtrpools(void)
{
        int small_pages = 1;
        int large_pages = (LNET_MTU + CFS_PAGE_SIZE - 1) >> CFS_PAGE_SHIFT;

        lnet_rtrpool_init(&the_lnet.ln_rtrpools[0], 0);
        lnet_rtrpool_init(&the_lnet.ln_rtrpools[1], small_pages);
        lnet_rtrpool_init(&the_lnet.ln_rtrpools[2], large_pages);
}


int
lnet_alloc_rtrpools(int im_a_router)
{
        int       rc;
        
        if (!strcmp(forwarding, "")) {
                /* not set either way */
                if (!im_a_router)
                        return 0;
        } else if (!strcmp(forwarding, "disabled")) {
                /* explicitly disabled */
                return 0;
        } else if (!strcmp(forwarding, "enabled")) {
                /* explicitly enabled */
        } else {
                LCONSOLE_ERROR_MSG(0x10b, "'forwarding' not set to either "
                                   "'enabled' or 'disabled'\n");
                return -EINVAL;
        }
        
        if (tiny_router_buffers <= 0) {
                LCONSOLE_ERROR_MSG(0x10c, "tiny_router_buffers=%d invalid when "
                                   "routing enabled\n", tiny_router_buffers);
                rc = -EINVAL;
                goto failed;
        }

        rc = lnet_rtrpool_alloc_bufs(&the_lnet.ln_rtrpools[0],
                                     tiny_router_buffers);
        if (rc != 0)
                goto failed;

        if (small_router_buffers <= 0) {
                LCONSOLE_ERROR_MSG(0x10d, "small_router_buffers=%d invalid when"
                                   " routing enabled\n", small_router_buffers);
                rc = -EINVAL;
                goto failed;
        }

        rc = lnet_rtrpool_alloc_bufs(&the_lnet.ln_rtrpools[1],
                                     small_router_buffers);
        if (rc != 0)
                goto failed;

        if (large_router_buffers <= 0) {
                LCONSOLE_ERROR_MSG(0x10e, "large_router_buffers=%d invalid when"
                                   " routing enabled\n", large_router_buffers);
                rc = -EINVAL;
                goto failed;
        }

        rc = lnet_rtrpool_alloc_bufs(&the_lnet.ln_rtrpools[2],
                                     large_router_buffers);
        if (rc != 0)
                goto failed;

        LNET_LOCK();
        the_lnet.ln_routing = 1;
        LNET_UNLOCK();
        
        return 0;

 failed:
        lnet_free_rtrpools();
        return rc;
}

#else

int
lnet_peers_start_down(void)
{
        return 0;
}

void
lnet_router_checker_stop(void)
{
        return;
}

int
lnet_router_checker_start(void)
{
        return 0;
}

void
lnet_free_rtrpools (void)
{
}

void
lnet_init_rtrpools (void)
{
}

int
lnet_alloc_rtrpools (int im_a_arouter)
{
        return 0;
}

#endif