qswlnd.c

非常经典的一个分布式系统

	CDEBUG (D_MALLOC, "start kmem %d\n", atomic_read(&libcfs_kmemory));
	
	/* ensure all pointers NULL etc */
	memset (&kqswnal_data, 0, sizeof (kqswnal_data));

	kqswnal_data.kqn_ni = ni;
	ni->ni_data = &kqswnal_data;
	ni->ni_peertxcredits = *kqswnal_tunables.kqn_peercredits;
	ni->ni_maxtxcredits = *kqswnal_tunables.kqn_credits;

	INIT_LIST_HEAD (&kqswnal_data.kqn_idletxds);
	INIT_LIST_HEAD (&kqswnal_data.kqn_activetxds);
	spin_lock_init (&kqswnal_data.kqn_idletxd_lock);

	INIT_LIST_HEAD (&kqswnal_data.kqn_delayedtxds);
	INIT_LIST_HEAD (&kqswnal_data.kqn_donetxds);
	INIT_LIST_HEAD (&kqswnal_data.kqn_readyrxds);

	spin_lock_init (&kqswnal_data.kqn_sched_lock);
	init_waitqueue_head (&kqswnal_data.kqn_sched_waitq);

	/* pointers/lists/locks initialised */
	kqswnal_data.kqn_init = KQN_INIT_DATA;
	PORTAL_MODULE_USE;
	
	kqswnal_data.kqn_ep = ep_system();
	if (kqswnal_data.kqn_ep == NULL) {
		CERROR("Can't initialise EKC\n");
		kqswnal_shutdown(ni);
		return (-ENODEV);
	}

	if (ep_waitfor_nodeid(kqswnal_data.kqn_ep) == ELAN_INVALID_NODE) {
		CERROR("Can't get elan ID\n");
		kqswnal_shutdown(ni);
		return (-ENODEV);
	}

	kqswnal_data.kqn_nnodes = ep_numnodes (kqswnal_data.kqn_ep);
	kqswnal_data.kqn_elanid = ep_nodeid (kqswnal_data.kqn_ep);

	ni->ni_nid = LNET_MKNID(LNET_NIDNET(ni->ni_nid), kqswnal_data.kqn_elanid);
	
	/**********************************************************************/
	/* Get the transmitter */

	kqswnal_data.kqn_eptx = ep_alloc_xmtr (kqswnal_data.kqn_ep);
	if (kqswnal_data.kqn_eptx == NULL)
	{
		CERROR ("Can't allocate transmitter\n");
		kqswnal_shutdown (ni);
		return (-ENOMEM);
	}

	/**********************************************************************/
	/* Get the receivers */

	kqswnal_data.kqn_eprx_small = 
		ep_alloc_rcvr (kqswnal_data.kqn_ep,
			       EP_MSG_SVC_PORTALS_SMALL,
			       *kqswnal_tunables.kqn_ep_envelopes_small);
	if (kqswnal_data.kqn_eprx_small == NULL)
	{
		CERROR ("Can't install small msg receiver\n");
		kqswnal_shutdown (ni);
		return (-ENOMEM);
	}

	kqswnal_data.kqn_eprx_large = 
		ep_alloc_rcvr (kqswnal_data.kqn_ep,
			       EP_MSG_SVC_PORTALS_LARGE,
			       *kqswnal_tunables.kqn_ep_envelopes_large);
	if (kqswnal_data.kqn_eprx_large == NULL)
	{
		CERROR ("Can't install large msg receiver\n");
		kqswnal_shutdown (ni);
		return (-ENOMEM);
	}

	/**********************************************************************/
	/* Reserve Elan address space for transmit descriptors NB we may
	 * either send the contents of associated buffers immediately, or
	 * map them for the peer to suck/blow... */
	kqswnal_data.kqn_ep_tx_nmh = 
		ep_dvma_reserve(kqswnal_data.kqn_ep,
				KQSW_NTXMSGPAGES*(*kqswnal_tunables.kqn_ntxmsgs),
				EP_PERM_WRITE);
	if (kqswnal_data.kqn_ep_tx_nmh == NULL) {
		CERROR("Can't reserve tx dma space\n");
		kqswnal_shutdown(ni);
		return (-ENOMEM);
	}

	/**********************************************************************/
	/* Reserve Elan address space for receive buffers */
	kqswnal_data.kqn_ep_rx_nmh =
		ep_dvma_reserve(kqswnal_data.kqn_ep,
				KQSW_NRXMSGPAGES_SMALL * 
				(*kqswnal_tunables.kqn_nrxmsgs_small) +
				KQSW_NRXMSGPAGES_LARGE * 
				(*kqswnal_tunables.kqn_nrxmsgs_large),
				EP_PERM_WRITE);
	if (kqswnal_data.kqn_ep_tx_nmh == NULL) {
		CERROR("Can't reserve rx dma space\n");
		kqswnal_shutdown(ni);
		return (-ENOMEM);
	}

	/**********************************************************************/
	/* Allocate/Initialise transmit descriptors */

	kqswnal_data.kqn_txds = NULL;
	for (i = 0; i < (*kqswnal_tunables.kqn_ntxmsgs); i++)
	{
		int           premapped_pages;
		int           basepage = i * KQSW_NTXMSGPAGES;

		LIBCFS_ALLOC (ktx, sizeof(*ktx));
		if (ktx == NULL) {
			kqswnal_shutdown (ni);
			return (-ENOMEM);
		}

		memset(ktx, 0, sizeof(*ktx));	/* NULL pointers; zero flags */
		ktx->ktx_alloclist = kqswnal_data.kqn_txds;
		kqswnal_data.kqn_txds = ktx;

		LIBCFS_ALLOC (ktx->ktx_buffer, KQSW_TX_BUFFER_SIZE);
		if (ktx->ktx_buffer == NULL)
		{
			kqswnal_shutdown (ni);
			return (-ENOMEM);
		}

		/* Map pre-allocated buffer NOW, to save latency on transmit */
		premapped_pages = kqswnal_pages_spanned(ktx->ktx_buffer,
							KQSW_TX_BUFFER_SIZE);
		ep_dvma_load(kqswnal_data.kqn_ep, NULL, 
			     ktx->ktx_buffer, KQSW_TX_BUFFER_SIZE, 
			     kqswnal_data.kqn_ep_tx_nmh, basepage,
			     &all_rails, &ktx->ktx_ebuffer);

		ktx->ktx_basepage = basepage + premapped_pages; /* message mapping starts here */
		ktx->ktx_npages = KQSW_NTXMSGPAGES - premapped_pages; /* for this many pages */

		INIT_LIST_HEAD (&ktx->ktx_schedlist);

		ktx->ktx_state = KTX_IDLE;
		ktx->ktx_rail = -1;		/* unset rail */

		list_add_tail (&ktx->ktx_list, &kqswnal_data.kqn_idletxds);
	}

	/**********************************************************************/
	/* Allocate/Initialise receive descriptors */
	kqswnal_data.kqn_rxds = NULL;
	elan_page_idx = 0;
	for (i = 0; i < *kqswnal_tunables.kqn_nrxmsgs_small + *kqswnal_tunables.kqn_nrxmsgs_large; i++)
	{
		EP_NMD        elanbuffer;
		int           j;

		LIBCFS_ALLOC(krx, sizeof(*krx));
		if (krx == NULL) {
			kqswnal_shutdown(ni);
			return (-ENOMEM);
		}

		memset(krx, 0, sizeof(*krx)); /* clear flags, null pointers etc */
		krx->krx_alloclist = kqswnal_data.kqn_rxds;
		kqswnal_data.kqn_rxds = krx;

		if (i < *kqswnal_tunables.kqn_nrxmsgs_small)
		{
			krx->krx_npages = KQSW_NRXMSGPAGES_SMALL;
			krx->krx_eprx   = kqswnal_data.kqn_eprx_small;
		}
		else
		{
			krx->krx_npages = KQSW_NRXMSGPAGES_LARGE;
			krx->krx_eprx   = kqswnal_data.kqn_eprx_large;
		}

		LASSERT (krx->krx_npages > 0);
		for (j = 0; j < krx->krx_npages; j++)
		{
			struct page *page = alloc_page(GFP_KERNEL);
			
			if (page == NULL) {
				kqswnal_shutdown (ni);
				return (-ENOMEM);
			}

			krx->krx_kiov[j] = (lnet_kiov_t) {.kiov_page = page,
							  .kiov_offset = 0,
							  .kiov_len = PAGE_SIZE};
			LASSERT(page_address(page) != NULL);

			ep_dvma_load(kqswnal_data.kqn_ep, NULL,
				     page_address(page),
				     PAGE_SIZE, kqswnal_data.kqn_ep_rx_nmh,
				     elan_page_idx, &all_rails, &elanbuffer);
			
			if (j == 0) {
				krx->krx_elanbuffer = elanbuffer;
			} else {
				rc = ep_nmd_merge(&krx->krx_elanbuffer,
						  &krx->krx_elanbuffer, 
						  &elanbuffer);
				/* NB contiguous mapping */
				LASSERT(rc);
			}
			elan_page_idx++;

		}
	}
	LASSERT (elan_page_idx ==
		 (*kqswnal_tunables.kqn_nrxmsgs_small * KQSW_NRXMSGPAGES_SMALL) +
		 (*kqswnal_tunables.kqn_nrxmsgs_large * KQSW_NRXMSGPAGES_LARGE));

	/**********************************************************************/
	/* Queue receives, now that it's OK to run their completion callbacks */

	for (krx = kqswnal_data.kqn_rxds; krx != NULL; krx = krx->krx_alloclist) {
		/* NB this enqueue can allocate/sleep (attr == 0) */
		krx->krx_state = KRX_POSTED;
		rc = ep_queue_receive(krx->krx_eprx, kqswnal_rxhandler, krx,
				      &krx->krx_elanbuffer, 0);
		if (rc != EP_SUCCESS) {
			CERROR ("failed ep_queue_receive %d\n", rc);
			kqswnal_shutdown (ni);
			return (-EIO);
		}
	}

	/**********************************************************************/
	/* Spawn scheduling threads */
	for (i = 0; i < num_online_cpus(); i++) {
		rc = kqswnal_thread_start (kqswnal_scheduler, NULL);
		if (rc != 0)
		{
			CERROR ("failed to spawn scheduling thread: %d\n", rc);
			kqswnal_shutdown (ni);
			return (-ESRCH);
		}
	}

	kqswnal_data.kqn_init = KQN_INIT_ALL;
	return (0);
}

void __exit
kqswnal_finalise (void)
{
	lnet_unregister_lnd(&the_kqswlnd);
	kqswnal_tunables_fini();
}

static int __init
kqswnal_initialise (void)
{
	int   rc = kqswnal_tunables_init();
	
	if (rc != 0)
		return rc;

	lnet_register_lnd(&the_kqswlnd);
	return (0);
}

MODULE_AUTHOR("Cluster File Systems, Inc. <[e-mail]>");
MODULE_DESCRIPTION("Kernel Quadrics/Elan LND v1.01");
MODULE_LICENSE("GPL");

module_init (kqswnal_initialise);
module_exit (kqswnal_finalise);