lcd/ixgbe: Add multiqueue support

Add a hack for alloc and merge a number of 4MB chunks from the buddy allocator.
Ideally we should use a contiguous memory allocator.
Signed-off-by: Vikram Narayanan <vikram186@gmail.com>

lcd-domains/test_mods/ixgbe/ixgbe_common.h

@@ -24,6 +24,7 @@
 #define NAPI_CONSUME_SEND_ONLY
 #define LOCAL_SKB
 #define NAPI_RX_SEND_ONLY
+#define SENDER_DISPATCH_LOOP
 
 enum dispatch_t {
 	__PCI_REGISTER_DRIVER,
@@ -101,6 +102,7 @@ enum dispatch_t {
 	TRIGGER_EXIT,
 	SERVICE_EVENT_SCHED,
 	TRIGGER_DUMP,
+	TRIGGER_CLEAN,
 };
 
 typedef enum {
@@ -126,7 +128,7 @@ typedef enum {
 	printk("%02x\n", addr[_i]); \
 	} while(0)
 
-#define ASYNC_RPC_BUFFER_ORDER 12
+#define ASYNC_RPC_BUFFER_ORDER 15
 
 struct pcidev_info {
 	unsigned int domain, bus, slot, fn;
@@ -215,6 +217,8 @@ async_msg_blocking_recv_start(struct thc_channel *chnl,
 			struct fipc_message** out)
 {
 	int ret;
+	static int count = 0;
+
 	for (;;) {
 		/* Poll until we get a message or error */
 		ret = fipc_recv_msg_start(thc_channel_to_fipc(chnl),
@@ -222,6 +226,8 @@ async_msg_blocking_recv_start(struct thc_channel *chnl,
 		if ( !ret || ret != -EWOULDBLOCK )
 			return ret;
 		cpu_relax();
+		if (count++ % 512 == 0)
+			cond_resched();
 	}
 }
 #endif /* __IXGBE_COMMON_H__ */

lcd-domains/test_mods/ixgbe/ixgbe_glue_helper.h

@@ -63,6 +63,7 @@ struct sk_buff_container {
 	struct cptr other_ref;
 	struct cptr my_ref;
 	struct task_struct *tsk;
+	void *channel;
 };
 
 struct trampoline_hidden_args {

lcd-domains/test_mods/ixgbe/ixgbe_lcd/glue/ixgbe_caller.c

@@ -15,7 +15,7 @@
 #include "../../rdtsc_helper.h"
 #include <lcd_config/post_hook.h>
 
-#define LCD_MEASUREMENT
+//#define LCD_MEASUREMENT
 
 struct cptr sync_ep;
 static struct glue_cspace *c_cspace;
@@ -307,9 +307,12 @@ fail1:
 	return;
 }
 
+struct thc_channel_group_item *ptrs[32];
+
 int create_one_async_channel(struct thc_channel **chnl, cptr_t *tx, cptr_t *rx)
 {
 	int ret;
+	static int idx = 0;
 	struct thc_channel_group_item *xmit_ch_item;
 
 	ret = setup_async_channel(tx, rx, chnl);
@@ -323,8 +326,12 @@ int create_one_async_channel(struct thc_channel **chnl, cptr_t *tx, cptr_t *rx)
 
 	thc_channel_group_item_init(xmit_ch_item, *chnl, NULL);
 
+	xmit_ch_item->xmit_channel = true;
+
 	thc_channel_group_item_add(&ch_grp, xmit_ch_item);
 
+	ptrs[idx++%32] = xmit_ch_item;
+
 	return 0;
 }
 
@@ -1500,19 +1507,24 @@ void napi_consume_skb(struct sk_buff *skb, int budget)
 {
 	int ret;
 	struct fipc_message *_request;
-	unsigned long skb_sz, skb_off, skbh_sz, skbh_off;
-	cptr_t skb_cptr, skbh_cptr;
 	struct sk_buff_container *skb_c;
 #ifndef NAPI_CONSUME_SEND_ONLY
 	struct fipc_message *_response;
 #endif
-
-	ret = async_msg_blocking_send_start(ixgbe_async,
-		&_request);
+	struct thc_channel *channel;
 
 	glue_lookup_skbuff(cptr_table,
 		__cptr((unsigned long)skb), &skb_c);
 
+#ifdef SENDER_DISPATCH_LOOP
+	channel = skb_c->channel;
+#else
+	channel = ixgbe_async;
+#endif
+
+	ret = async_msg_blocking_send_start(channel,
+		&_request);
+
 	if (ret) {
 		LIBLCD_ERR("failed to get a send slot");
 		goto fail_async;
@@ -1523,50 +1535,12 @@ void napi_consume_skb(struct sk_buff *skb, int budget)
 	fipc_set_reg0(_request, skb_c->other_ref.cptr);
 	fipc_set_reg1(_request, budget);
 
-	if (!skb->private) {
-		ret = lcd_virt_to_cptr(__gva((unsigned long)skb),
-			&skb_cptr,
-			&skb_sz,
-			&skb_off);
-		if (ret) {
-			LIBLCD_ERR("lcd_virt_to_cptr");
-			goto fail_virt;
-		}
-
-		ret = lcd_virt_to_cptr(__gva((unsigned long)skb->head),
-			&skbh_cptr,
-			&skbh_sz,
-			&skbh_off);
-		if (ret) {
-			LIBLCD_ERR("lcd_virt_to_cptr");
-			goto fail_virt;
-		}
-
-#ifdef IOMMU_ASSIGN
-		LIBLCD_MSG("%s, iommu unmap page", __func__);
-		ret = lcd_syscall_iommu_unmap_page(lcd_gva2gpa(__gva(
-			(unsigned long)((void*)skb->head - skbh_off))),
-			skb_c->skbd_ord);
-		if (ret)
-			LIBLCD_ERR("unmap iommu failed for addr %p",
-					__pa(skb->data));
-#endif
-
-		lcd_unmap_virt(__gva((unsigned long)skb->head),
-				get_order(skbh_sz));
-		lcd_unmap_virt(__gva((unsigned long)skb),
-				get_order(skb_sz));
-
-		lcd_cap_delete(skb_c->skb_cptr);
-		lcd_cap_delete(skb_c->skbh_cptr);
-	}
-
 #ifdef NAPI_CONSUME_SEND_ONLY
 	thc_set_msg_type(_request, msg_type_request);
-	fipc_send_msg_end(thc_channel_to_fipc(ixgbe_async),
+	fipc_send_msg_end(thc_channel_to_fipc(channel),
 					_request);
 #else
-	ret = thc_ipc_call(ixgbe_async,
+	ret = thc_ipc_call(channel,
 		_request,
 		&_response);
 
@@ -1575,7 +1549,7 @@ void napi_consume_skb(struct sk_buff *skb, int budget)
 		goto fail_ipc;
 	}
 
-	fipc_recv_msg_end(thc_channel_to_fipc(ixgbe_async),
+	fipc_recv_msg_end(thc_channel_to_fipc(channel),
 			_response);
 #endif
 
@@ -1586,7 +1560,6 @@ fail_async:
 #ifndef NAPI_CONSUME_SEND_ONLY
 fail_ipc:
 #endif
-fail_virt:
 	return;
 }
 
@@ -2617,9 +2590,11 @@ int prep_channel_callee(struct fipc_message *_request,
 		struct cptr sync_ep)
 {
 	cptr_t tx, rx;
+#ifdef SOFTIRQ_CHANNELS
 	cptr_t tx_sirq, rx_sirq;
-	struct thc_channel *xmit;
 	struct thc_channel *xmit_sirq;
+#endif
+	struct thc_channel *xmit;
 	unsigned 	int request_cookie;
 	int ret;
 	struct fipc_message *_response;
@@ -2632,23 +2607,28 @@ int prep_channel_callee(struct fipc_message *_request,
 	if (create_one_async_channel(&xmit, &tx, &rx))
 		LIBLCD_ERR("async channel creation failed\n");
 
+	lcd_set_cr0(tx);
+	lcd_set_cr1(rx);
+
+#ifdef SOFTIRQ_CHANNELS
 	printk("Creating one for softirq\n");
 
 	if (create_one_async_channel(&xmit_sirq, &tx_sirq, &rx_sirq))
 		LIBLCD_ERR("async channel creation failed\n");
-
-	lcd_set_cr0(tx);
-	lcd_set_cr1(rx);
 	lcd_set_cr2(tx_sirq);
 	lcd_set_cr3(rx_sirq);
+#endif
 
 	LIBLCD_MSG("%s: Preparing sync send", __func__);
 	ret = lcd_sync_send(ixgbe_sync_endpoint);
 
 	lcd_set_cr0(CAP_CPTR_NULL);
 	lcd_set_cr1(CAP_CPTR_NULL);
+
+#ifdef SOFTIRQ_CHANNELS
 	lcd_set_cr2(CAP_CPTR_NULL);
 	lcd_set_cr3(CAP_CPTR_NULL);
+#endif
 
 	if (ret) {
 		LIBLCD_ERR("failed to send");
@@ -2665,6 +2645,138 @@ int prep_channel_callee(struct fipc_message *_request,
 	return ret;
 }
 
+/* This function is used for testing bare fipc, non-async mtu sized packets */
+int ndo_start_xmit_bare_callee(struct fipc_message *_request, struct thc_channel *channel, struct glue_cspace *cspace, struct cptr sync_ep)
+{
+	struct fipc_message *response;
+	fipc_recv_msg_end(thc_channel_to_fipc(channel), _request);
+
+	if (likely(async_msg_blocking_send_start(channel, &response))) {
+		LIBLCD_ERR("error getting response msg");
+		return -EIO;
+	}
+	fipc_send_msg_end(thc_channel_to_fipc(channel), response);
+
+	return 0;
+}
+
+int ndo_start_xmit_clean_callee(struct fipc_message *_request,
+		struct thc_channel *_channel,
+		struct glue_cspace *cspace,
+		struct cptr sync_ep)
+{
+	struct sk_buff *skb;
+	struct sk_buff_container *skb_c;
+	int ret = 0;
+	struct fipc_message *_response;
+	unsigned 	int request_cookie;
+	int func_ret;
+	cptr_t skb_ref;
+	xmit_type_t xmit_type;
+	unsigned long skbh_offset, skb_end;
+	struct skbuff_members *skb_lcd;
+	__be16 proto;
+	void *mem;
+#ifdef LCD_MEASUREMENT
+	TS_DECL(xmit);
+#endif
+	request_cookie = thc_get_request_cookie(_request);
+	xmit_type = fipc_get_reg0(_request);
+
+	skb_ref = __cptr(fipc_get_reg2(_request));
+
+	skbh_offset = fipc_get_reg3(_request);
+
+	skb_end = fipc_get_reg4(_request);
+	proto = fipc_get_reg5(_request);
+
+	fipc_recv_msg_end(thc_channel_to_fipc(_channel),
+			_request);
+
+	if (_channel == ixgbe_async)
+		printk("%s, Got msg - reqc 0x%x | xmitty %d",
+				__func__, request_cookie,
+				xmit_type);
+
+	skb_c = mem = kmem_cache_alloc(skb_c_cache,
+				GFP_KERNEL);
+	skb = (struct sk_buff*)((char*)mem +
+			sizeof(struct sk_buff_container));
+
+	if (!skb) {
+		LIBLCD_MSG("out of mmeory");
+		goto fail_alloc;
+	}
+
+	skb->head = (char*)data_pool + skbh_offset;
+	skb->end = skb_end;
+
+	skb_lcd = SKB_LCD_MEMBERS(skb);
+	skb->private = true;
+	skb->protocol = proto;
+
+	P(len);
+	P(data_len);
+	P(queue_mapping);
+	P(xmit_more);
+	P(tail);
+	P(truesize);
+	P(ip_summed);
+	P(csum_start);
+	P(network_header);
+	P(csum_offset);
+	P(transport_header);
+
+	if (0)
+	LIBLCD_MSG("lcd-> l: %d | dlen %d | qm %d"
+		   " | xm %d | t %lu |ts %u",
+		skb->len, skb->data_len, skb->queue_mapping,
+		skb->xmit_more, skb->tail, skb->truesize);
+
+	skb->data = skb->head + skb_lcd->head_data_off;
+
+	skb_c->skb = skb;
+	glue_insert_skbuff(cptr_table, skb_c);
+	skb_c->other_ref = skb_ref;
+
+	skb_c->channel = _channel;
+
+#ifdef LCD_MEASUREMENT
+	TS_START_LCD(xmit);
+#endif
+	func_ret = ixgbe_xmit_frame(skb, g_netdev);
+
+	if (func_ret)
+		LIBLCD_MSG("ixgbe xmit failed %p", skb);
+
+#ifdef LCD_MEASUREMENT
+	TS_STOP_LCD(xmit);
+#endif
+	if (async_msg_blocking_send_start(_channel,
+		&_response)) {
+		LIBLCD_ERR("error getting response msg");
+		return -EIO;
+	}
+	fipc_set_reg1(_response,
+			func_ret);
+#ifdef LCD_MEASUREMENT
+	fipc_set_reg2(_response,
+			TS_DIFF(xmit));
+#endif
+	thc_ipc_reply(_channel,
+			request_cookie,
+			_response);
+
+//	printk("%s, posting response for reqc 0x%x", __func__,
+//				request_cookie);
+
+	if (_channel == ixgbe_async)
+		printk("%s, Sending reply for reqc 0x%x", __func__,
+				request_cookie);
+fail_alloc:
+	return ret;
+
+}
 
 int ndo_start_xmit_callee(struct fipc_message *_request,
 		struct thc_channel *_channel,
@@ -4044,3 +4156,20 @@ fail_ipc:
 	return ret;
 }
 #endif
+
+int cleanup_channel_group(struct fipc_message *request, struct thc_channel *channel)
+{
+	int i;
+	fipc_recv_msg_end(thc_channel_to_fipc(channel), request);
+
+	for (i = 0; i < 32; i++) {
+		if (ptrs[i]) {
+			thc_channel_group_item_remove(&ch_grp, ptrs[i]);
+			destroy_async_channel(ptrs[i]->channel);
+			kfree(ptrs[i]);
+			ptrs[i] = NULL;
+		} //if
+	} //for
+
+	return 0;
+}

lcd-domains/test_mods/ixgbe/ixgbe_lcd/glue/ixgbe_caller_dispatch.c

@@ -6,6 +6,7 @@
 #define trace(x) LIBLCD_MSG("net got " #x " msg")
 
 extern void ixgbe_exit_module(void);
+int cleanup_channel_group(struct fipc_message *, struct thc_channel *);
 extern bool poll_start;
 
 int dispatch_async_loop(struct thc_channel *_channel,
@@ -143,6 +144,10 @@ int dispatch_async_loop(struct thc_channel *_channel,
 			 */
 			return -1;
 
+		case TRIGGER_CLEAN:
+			trace(TRIGGER_CLEAN);
+			return cleanup_channel_group(message, _channel);
+
 		case PREP_CHANNEL:
 			trace(PREP_CHANNEL);
 			return prep_channel_callee(message, _channel, cspace, sync_ep);

lcd-domains/test_mods/ixgbe/ixgbe_lcd/ixgbe/ixgbe_lib.c

@@ -510,9 +510,10 @@ static bool ixgbe_set_sriov_queues(struct ixgbe_adapter *adapter)
 	bool pools = (find_first_zero_bit(&adapter->fwd_bitmask, 32) > 1);
 
 	/* only proceed if SR-IOV is enabled */
-	if (!(adapter->flags & IXGBE_FLAG_SRIOV_ENABLED))
+	if (!(adapter->flags & IXGBE_FLAG_SRIOV_ENABLED)) {
+		printk("%s, SRIOV not enabled, won't allocate more queues", __func__);
 		return false;
-
+	}
 	/* Add starting offset to total pool count */
 	vmdq_i += adapter->ring_feature[RING_F_VMDQ].offset;
 
@@ -665,6 +666,10 @@ static bool ixgbe_set_rss_queues(struct ixgbe_adapter *adapter)
 	}
 
 #endif /* IXGBE_FCOE */
+
+	printk("%s:%d, adapter tx=%d rx=%d", __func__, __LINE__,
+			adapter->num_tx_queues, adapter->num_rx_queues);
+
 	adapter->num_rx_queues = rss_i;
 	adapter->num_tx_queues = rss_i;
 
@@ -690,6 +695,8 @@ static void ixgbe_set_num_queues(struct ixgbe_adapter *adapter)
 	adapter->num_rx_pools = adapter->num_rx_queues;
 	adapter->num_rx_queues_per_pool = 1;
 
+	printk("%s:%d, adapter tx=%d rx=%d", __func__, __LINE__,
+			adapter->num_tx_queues, adapter->num_rx_queues);
 #ifdef CONFIG_IXGBE_DCB
 	if (ixgbe_set_dcb_sriov_queues(adapter))
 		return;
@@ -701,6 +708,8 @@ static void ixgbe_set_num_queues(struct ixgbe_adapter *adapter)
 	if (ixgbe_set_sriov_queues(adapter))
 		return;
 
+	printk("%s:%d, after calling sriov queues adapter tx=%d rx=%d", __func__, __LINE__,
+			adapter->num_tx_queues, adapter->num_rx_queues);
 	ixgbe_set_rss_queues(adapter);
 }
 

lcd-domains/test_mods/ixgbe/ixgbe_lcd/ixgbe/ixgbe_main.c

@@ -5852,13 +5852,21 @@ static int ixgbe_sw_init(struct ixgbe_adapter *adapter)
 	hw->subsystem_vendor_id = pdev->subsystem_vendor;
 	hw->subsystem_device_id = pdev->subsystem_device;
 
+#define NUM_HW_QUEUES		4
+	/* XXX: This decides the number of hardware queues
+	 * It checks the number of online cpus and sets the number of
+	 * queues to the number of cpus. Fake it now to enable
+	 * multiqueue support and see if something comes up
+	 */
 	/* Set common capability flags and settings */
-	rss = min_t(int, ixgbe_max_rss_indices(adapter), num_online_cpus());
+	//rss = min_t(int, ixgbe_max_rss_indices(adapter), num_online_cpus());
+	rss = min_t(int, ixgbe_max_rss_indices(adapter), NUM_HW_QUEUES);
 	adapter->ring_feature[RING_F_RSS].limit = rss;
 	adapter->flags2 |= IXGBE_FLAG2_RSC_CAPABLE;
 	adapter->max_q_vectors = MAX_Q_VECTORS_82599;
 	adapter->atr_sample_rate = 20;
-	fdir = min_t(int, IXGBE_MAX_FDIR_INDICES, num_online_cpus());
+	//fdir = min_t(int, IXGBE_MAX_FDIR_INDICES, num_online_cpus());
+	fdir = min_t(int, IXGBE_MAX_FDIR_INDICES, NUM_HW_QUEUES);
 	adapter->ring_feature[RING_F_FDIR].limit = fdir;
 	adapter->fdir_pballoc = IXGBE_FDIR_PBALLOC_64K;
 #ifdef CONFIG_IXGBE_DCA
@@ -8166,6 +8174,7 @@ xmit_fcoe:
 	return NETDEV_TX_OK;
 
 out_drop:
+	printk("%s, dropping packet %p", __func__, first->skb);
 	dev_kfree_skb_any(first->skb);
 	first->skb = NULL;
 

lcd-domains/test_mods/ixgbe/ixgbe_lcd/ixgbe_caller.h

@@ -18,6 +18,17 @@ int ndo_stop_callee(struct fipc_message *_request,
 		struct thc_channel *_channel,
 		struct glue_cspace *cspace,
 		struct cptr sync_ep);
+
+int ndo_start_xmit_bare_callee(struct fipc_message *_request,
+		struct thc_channel *_channel,
+		struct glue_cspace *cspace,
+		struct cptr sync_ep);
+
+int ndo_start_xmit_clean_callee(struct fipc_message *_request,
+		struct thc_channel *_channel,
+		struct glue_cspace *cspace,
+		struct cptr sync_ep);
+
 int ndo_start_xmit_callee(struct fipc_message *_request,
 		struct thc_channel *_channel,
 		struct glue_cspace *cspace,

lcd-domains/test_mods/ixgbe/ixgbe_lcd/main.c

@@ -77,6 +77,19 @@ static void main_and_loop(void)
 					stop = 1; /* stop */
 				}
 			}
+
+			if (async_msg_get_fn_type(msg) == NDO_START_XMIT) {
+				ret = ndo_start_xmit_clean_callee(msg,
+					curr_item->channel,
+					ixgbe_cspace,
+					ixgbe_sync_endpoint);
+
+				if (unlikely(ret)) {
+					LIBLCD_ERR("async dispatch failed");
+					stop = 1;
+				}
+
+			} else {
 			/*
 			 * Got a message. Dispatch.
 			 */
@@ -91,6 +104,7 @@ static void main_and_loop(void)
 				}
 
 				);
+			}
 			/* FIXME: This is pretty naive. This is *NOT* how
 			 * we should do napi polling.
 			 */

lcd-domains/test_mods/ixgbe/net_klcd/main.c

@@ -31,17 +31,19 @@ extern int setup_async_net_ring_channel(cptr_t tx, cptr_t rx,
 extern void destroy_async_net_ring_channel(struct thc_channel *chnl);
 extern int ixgbe_trigger_dump(struct thc_channel *_channel);
 extern int ixgbe_service_event_sched(struct thc_channel *_channel);
-extern int trigger_exit_to_lcd(struct thc_channel *_channel);
+extern int trigger_exit_to_lcd(struct thc_channel *_channel, enum dispatch_t);
 extern struct timer_list service_timer;
 extern struct glue_cspace *c_cspace;
 
 /* mechanism for unloading LCD gracefully */
 static bool unload_lcd =0;
+static bool clean_up = false;
 module_param_named(unload, unload_lcd, bool, S_IWUSR);
 
 /* to dump ixgbe registers */
 static bool ixgbe_dump =0;
 module_param_named(dump_regs, ixgbe_dump, bool, S_IWUSR);
+module_param_named(clean, clean_up, bool, S_IWUSR);
 
 struct net_info *
 add_net(struct thc_channel *chnl, struct glue_cspace *cspace,
@@ -286,78 +288,86 @@ static void loop(cptr_t register_chnl)
 	bool reg_done = false;
 
 	DO_FINISH(
-
-		while (!stop) {
-			if (jiffies >= tics && !reg_done) {
-				/*
-				 * Listen for a register call
-				 */
-				ret = do_one_register(register_chnl);
-				if (ret) {
-					LIBLCD_ERR("register error");
-					break;
-				} else {
-					reg_done = true;
-				}
-				tics = jiffies + REGISTER_FREQ;
-				continue;
-			}
-			if (stop)
-				break;
-
+	while (!stop) {
+		if (jiffies >= tics && !reg_done) {
 			/*
-			 * will be updated by a write into sysfs
-			 * from userspace.
+			 * Listen for a register call
 			 */
-			if (unload_lcd) {
-				unload_lcd = 0;
-				if (__get_net(&net))
-					trigger_exit_to_lcd(net->chnl);
-			}
-			if (ixgbe_dump) {
-				ixgbe_dump = 0;
-				if (__get_net(&net))
-					ixgbe_trigger_dump(net->chnl);
-			}
-			ret = async_loop(&net, &msg);
-			if (!ret) {
-				ASYNC(
-					ret = dispatch_async_loop(
-						net->chnl,
-						msg,
-						net->cspace,
-						net->sync_endpoint);
-					if (ret) {
-						LIBLCD_ERR("net dispatch err");
-						/* (break won't work here) */
-						stop = 1;
-					}
-					);
-			} else if (ret != -EWOULDBLOCK) {
-				LIBLCD_ERR("async loop failed");
-				stop = 1;
+			ret = do_one_register(register_chnl);
+			if (ret) {
+				LIBLCD_ERR("register error");
 				break;
+			} else {
+				reg_done = true;
 			}
-
-			if (kthread_should_stop()) {
-				LIBLCD_MSG("kthread should stop");
-				stop = 1;
-				break;
+			tics = jiffies + REGISTER_FREQ;
+			continue;
+		}
+		if (stop)
+			break;
+
+		/*
+		 * will be updated by a write into sysfs
+		 * from userspace.
+		 */
+		if (unload_lcd || clean_up) {
+			if (__get_net(&net)) {
+				if (unload_lcd) {
+					trigger_exit_to_lcd(net->chnl, TRIGGER_EXIT);
+					unload_lcd ^= unload_lcd;
+				}
+				if (clean_up) {
+					LIBLCD_MSG("cleanup triggered"); 
+					trigger_exit_to_lcd(net->chnl, TRIGGER_CLEAN);
+					clean_up ^= clean_up;
+				}
 			}
+		}
+
+		if (ixgbe_dump) {
+			ixgbe_dump = 0;
+			if (__get_net(&net))
+				ixgbe_trigger_dump(net->chnl);
+		}
+		ret = async_loop(&net, &msg);
+		if (!ret) {
+			ASYNC(
+				ret = dispatch_async_loop(
+					net->chnl,
+					msg,
+					net->cspace,
+					net->sync_endpoint);
+				if (ret) {
+					LIBLCD_ERR("net dispatch err");
+					/* (break won't work here) */
+					stop = 1;
+				}
+				);
+		} else if (ret != -EWOULDBLOCK) {
+			LIBLCD_ERR("async loop failed");
+			stop = 1;
+			break;
+		}
+
+		if (kthread_should_stop()) {
+			LIBLCD_MSG("kthread should stop");
+			stop = 1;
+			break;
+		}