xref: /illumos-gate/usr/src/uts/common/io/mac/mac_soft_ring.c (revision 6e375c8351497b82ffa4f33cbf61d712999b4605)
1 /*
2  * CDDL HEADER START
3  *
4  * The contents of this file are subject to the terms of the
5  * Common Development and Distribution License (the "License").
6  * You may not use this file except in compliance with the License.
7  *
8  * You can obtain a copy of the license at usr/src/OPENSOLARIS.LICENSE
9  * or http://www.opensolaris.org/os/licensing.
10  * See the License for the specific language governing permissions
11  * and limitations under the License.
12  *
13  * When distributing Covered Code, include this CDDL HEADER in each
14  * file and include the License file at usr/src/OPENSOLARIS.LICENSE.
15  * If applicable, add the following below this CDDL HEADER, with the
16  * fields enclosed by brackets "[]" replaced with your own identifying
17  * information: Portions Copyright [yyyy] [name of copyright owner]
18  *
19  * CDDL HEADER END
20  */
21 /*
22  * Copyright 2008 Sun Microsystems, Inc.  All rights reserved.
23  * Use is subject to license terms.
24  */
25 
26 /*
27  * General Soft rings - Simulating Rx rings in S/W.
28  *
29  * Soft ring is a data abstraction containing a queue and a worker
30  * thread and represents a hardware Rx ring in software. Each soft
31  * ring set can have a collection of soft rings for separating
32  * L3/L4 specific traffic (IPv4 from IPv6 or TCP from UDP) or for
33  * allowing a higher degree of parallelism by sending traffic to
34  * one of the soft rings for a SRS (using a hash on src IP or port).
35  * Each soft ring worker thread can be bound to a different CPU
36  * allowing the processing for each soft ring to happen in parallel
37  * and independent from each other.
38  *
39  * Protocol soft rings:
40  *
41  * Each SRS has at an minimum 3 softrings. One each for IPv4 TCP,
42  * IPv4 UDP and rest (OTH - for IPv6 and everything else). The
43  * SRS does dynamic polling and enforces link level bandwidth but
44  * it does so for all traffic (IPv4 and IPv6 and all protocols) on
45  * that link. However, each protocol layer wants a different
46  * behaviour. For instance IPv4 TCP has per CPU squeues which
47  * enforce their own polling and flow control so IPv4 TCP traffic
48  * needs to go to a separate soft ring which can be polled by the
49  * TCP squeue. It also allows TCP squeue to push back flow control
50  * all the way to NIC hardware (if it puts its corresponding soft
51  * ring in the poll mode and soft ring queue builds up, the
52  * shared srs_poll_pkt_cnt goes up and SRS automatically stops
53  * more packets from entering the system).
54  *
55  * Similarly, the UDP benefits from a DLS bypass and packet chaining
56  * so sending it to a separate soft ring is desired. All the rest of
57  * the traffic (including IPv6 is sent to OTH softring). The IPv6
58  * traffic current goes through OTH softring and via DLS because
59  * it need more processing to be done. Irrespective of the sap
60  * (IPv4 or IPv6) or the transport, the dynamic polling, B/W enforcement,
61  * cpu assignment, fanout, etc apply to all traffic since they
62  * are implement by the SRS which is agnostic to sap or transport.
63  *
64  * Fanout soft rings:
65  *
66  * On a multithreaded system, we can assign more CPU and multi thread
67  * the stack by creating a soft ring per CPU and spreading traffic
68  * based on a hash computed on src IP etc. Since we still need to
69  * keep the protocol separation, we create a set of 3 soft ring per
70  * CPU (specified by cpu list or degree of fanout).
71  *
72  * NOTE: See the block level comment on top of mac_sched.c
73  */
74 
75 #include <sys/types.h>
76 #include <sys/callb.h>
77 #include <sys/sdt.h>
78 #include <sys/strsubr.h>
79 #include <sys/strsun.h>
80 #include <sys/vlan.h>
81 #include <inet/ipsec_impl.h>
82 #include <inet/ip_impl.h>
83 #include <inet/sadb.h>
84 #include <inet/ipsecesp.h>
85 #include <inet/ipsecah.h>
86 
87 #include <sys/mac_impl.h>
88 #include <sys/mac_client_impl.h>
89 #include <sys/mac_soft_ring.h>
90 #include <sys/mac_flow_impl.h>
91 
92 static void mac_rx_soft_ring_drain(mac_soft_ring_t *);
93 static void mac_soft_ring_fire(void *);
94 static void mac_soft_ring_worker(mac_soft_ring_t *);
95 static void mac_tx_soft_ring_drain(mac_soft_ring_t *);
96 
97 uint32_t mac_tx_soft_ring_max_q_cnt = 100000;
98 uint32_t mac_tx_soft_ring_hiwat = 1000;
99 
100 extern kmem_cache_t *mac_soft_ring_cache;
101 
102 #define	ADD_SOFTRING_TO_SET(mac_srs, softring) {			\
103 	if (mac_srs->srs_soft_ring_head == NULL) {			\
104 		mac_srs->srs_soft_ring_head = softring;			\
105 		mac_srs->srs_soft_ring_tail = softring;			\
106 	} else {							\
107 		/* ADD to the list */					\
108 		softring->s_ring_prev =					\
109 			mac_srs->srs_soft_ring_tail;			\
110 		mac_srs->srs_soft_ring_tail->s_ring_next = softring;	\
111 		mac_srs->srs_soft_ring_tail = softring;			\
112 	}								\
113 	mac_srs->srs_soft_ring_count++;					\
114 }
115 
116 /*
117  * mac_soft_ring_worker_wakeup
118  *
119  * Wake up the soft ring worker thread to process the queue as long
120  * as no one else is processing it and upper layer (client) is still
121  * ready to receive packets.
122  */
123 void
124 mac_soft_ring_worker_wakeup(mac_soft_ring_t *ringp)
125 {
126 	ASSERT(MUTEX_HELD(&ringp->s_ring_lock));
127 	if (!(ringp->s_ring_state & S_RING_PROC) &&
128 	    !(ringp->s_ring_state & S_RING_BLANK) &&
129 	    (ringp->s_ring_tid == NULL)) {
130 		if (ringp->s_ring_wait != 0) {
131 			ringp->s_ring_tid =
132 			    timeout(mac_soft_ring_fire, ringp,
133 			    ringp->s_ring_wait);
134 		} else {
135 			/* Schedule the worker thread. */
136 			cv_signal(&ringp->s_ring_async);
137 		}
138 	}
139 }
140 
141 /*
142  * mac_soft_ring_create
143  *
144  * Create a soft ring, do the necessary setup and bind the worker
145  * thread to the assigned CPU.
146  */
147 mac_soft_ring_t *
148 mac_soft_ring_create(int id, clock_t wait, void *flent, uint16_t type,
149     pri_t pri, mac_client_impl_t *mcip, mac_soft_ring_set_t *mac_srs,
150     processorid_t cpuid, mac_direct_rx_t rx_func, void *x_arg1,
151     mac_resource_handle_t x_arg2)
152 {
153 	mac_soft_ring_t 	*ringp;
154 	char 			name[S_RING_NAMELEN];
155 
156 	bzero(name, 64);
157 	ringp = kmem_cache_alloc(mac_soft_ring_cache, KM_SLEEP);
158 
159 	if (type & ST_RING_TCP) {
160 		(void) snprintf(name, sizeof (name),
161 		    "mac_tcp_soft_ring_%d_%p", id, (void *)mac_srs);
162 	} else if (type & ST_RING_UDP) {
163 		(void) snprintf(name, sizeof (name),
164 		    "mac_udp_soft_ring_%d_%p", id, (void *)mac_srs);
165 	} else {
166 		(void) snprintf(name, sizeof (name),
167 		    "mac_oth_soft_ring_%d_%p", id, (void *)mac_srs);
168 	}
169 
170 	bzero(ringp, sizeof (mac_soft_ring_t));
171 	(void) strncpy(ringp->s_ring_name, name, S_RING_NAMELEN + 1);
172 	ringp->s_ring_name[S_RING_NAMELEN] = '\0';
173 	mutex_init(&ringp->s_ring_lock, NULL, MUTEX_DEFAULT, NULL);
174 	ringp->s_ring_notify_cb_info.mcbi_lockp = &ringp->s_ring_lock;
175 
176 	ringp->s_ring_type = type;
177 	ringp->s_ring_wait = MSEC_TO_TICK(wait);
178 	ringp->s_ring_mcip = mcip;
179 	ringp->s_ring_set = mac_srs;
180 	ringp->s_ring_flent = flent;
181 
182 	/*
183 	 * Protect against access from DR callbacks (mac_walk_srs_bind/unbind)
184 	 * which can't grab the mac perimeter
185 	 */
186 	mutex_enter(&mac_srs->srs_lock);
187 	ADD_SOFTRING_TO_SET(mac_srs, ringp);
188 	mutex_exit(&mac_srs->srs_lock);
189 
190 	/*
191 	 * set the bind CPU to -1 to indicate
192 	 * no thread affinity set
193 	 */
194 	ringp->s_ring_cpuid = ringp->s_ring_cpuid_save = -1;
195 	ringp->s_ring_worker = thread_create(NULL, 0,
196 	    mac_soft_ring_worker, ringp, 0, &p0, TS_RUN, pri);
197 	if (type & ST_RING_TX) {
198 		ringp->s_ring_drain_func = mac_tx_soft_ring_drain;
199 		ringp->s_ring_tx_arg1 = x_arg1;
200 		ringp->s_ring_tx_arg2 = x_arg2;
201 		ringp->s_ring_tx_max_q_cnt = mac_tx_soft_ring_max_q_cnt;
202 		ringp->s_ring_tx_hiwat =
203 		    (mac_tx_soft_ring_hiwat > mac_tx_soft_ring_max_q_cnt) ?
204 		    mac_tx_soft_ring_max_q_cnt : mac_tx_soft_ring_hiwat;
205 	} else {
206 		ringp->s_ring_drain_func = mac_rx_soft_ring_drain;
207 		ringp->s_ring_rx_func = rx_func;
208 		ringp->s_ring_rx_arg1 = x_arg1;
209 		ringp->s_ring_rx_arg2 = x_arg2;
210 	}
211 	if (cpuid != -1)
212 		(void) mac_soft_ring_bind(ringp, cpuid);
213 
214 	return (ringp);
215 }
216 
217 /*
218  * mac_soft_ring_free
219  *
220  * Free the soft ring once we are done with it.
221  */
222 void
223 mac_soft_ring_free(mac_soft_ring_t *softring, boolean_t release_tx_ring)
224 {
225 	ASSERT((softring->s_ring_state &
226 	    (S_RING_CONDEMNED | S_RING_CONDEMNED_DONE | S_RING_PROC)) ==
227 	    (S_RING_CONDEMNED | S_RING_CONDEMNED_DONE));
228 	mac_pkt_drop(NULL, NULL, softring->s_ring_first, B_FALSE);
229 	if (release_tx_ring && softring->s_ring_tx_arg2 != NULL) {
230 		ASSERT(softring->s_ring_type & ST_RING_TX);
231 		mac_release_tx_ring(softring->s_ring_tx_arg2);
232 	}
233 	if (softring->s_ring_ksp)
234 		kstat_delete(softring->s_ring_ksp);
235 	mac_callback_free(softring->s_ring_notify_cb_list);
236 	kmem_cache_free(mac_soft_ring_cache, softring);
237 }
238 
239 int mac_soft_ring_thread_bind = 1;
240 
241 /*
242  * mac_soft_ring_bind
243  *
244  * Bind a soft ring worker thread to supplied CPU.
245  */
246 cpu_t *
247 mac_soft_ring_bind(mac_soft_ring_t *ringp, processorid_t cpuid)
248 {
249 	cpu_t *cp;
250 	boolean_t clear = B_FALSE;
251 
252 	ASSERT(MUTEX_HELD(&cpu_lock));
253 
254 	if (mac_soft_ring_thread_bind == 0) {
255 		DTRACE_PROBE1(mac__soft__ring__no__cpu__bound,
256 		    mac_soft_ring_t *, ringp);
257 		return (NULL);
258 	}
259 
260 	cp = cpu_get(cpuid);
261 	if (cp == NULL || !cpu_is_online(cp))
262 		return (NULL);
263 
264 	mutex_enter(&ringp->s_ring_lock);
265 	ringp->s_ring_state |= S_RING_BOUND;
266 	if (ringp->s_ring_cpuid != -1)
267 		clear = B_TRUE;
268 	ringp->s_ring_cpuid = cpuid;
269 	mutex_exit(&ringp->s_ring_lock);
270 
271 	if (clear)
272 		thread_affinity_clear(ringp->s_ring_worker);
273 
274 	DTRACE_PROBE2(mac__soft__ring__cpu__bound, mac_soft_ring_t *,
275 	    ringp, processorid_t, cpuid);
276 
277 	thread_affinity_set(ringp->s_ring_worker, cpuid);
278 
279 	return (cp);
280 }
281 
282 /*
283  * mac_soft_ring_unbind
284  *
285  * Un Bind a soft ring worker thread.
286  */
287 void
288 mac_soft_ring_unbind(mac_soft_ring_t *ringp)
289 {
290 	ASSERT(MUTEX_HELD(&cpu_lock));
291 
292 	mutex_enter(&ringp->s_ring_lock);
293 	if (!(ringp->s_ring_state & S_RING_BOUND)) {
294 		ASSERT(ringp->s_ring_cpuid == -1);
295 		mutex_exit(&ringp->s_ring_lock);
296 		return;
297 	}
298 
299 	ringp->s_ring_cpuid = -1;
300 	ringp->s_ring_state &= ~S_RING_BOUND;
301 	thread_affinity_clear(ringp->s_ring_worker);
302 	mutex_exit(&ringp->s_ring_lock);
303 }
304 
305 /*
306  * PRIVATE FUNCTIONS
307  */
308 
309 static void
310 mac_soft_ring_fire(void *arg)
311 {
312 	mac_soft_ring_t	*ringp = arg;
313 
314 	mutex_enter(&ringp->s_ring_lock);
315 	if (ringp->s_ring_tid == 0) {
316 		mutex_exit(&ringp->s_ring_lock);
317 		return;
318 	}
319 
320 	ringp->s_ring_tid = 0;
321 
322 	if (!(ringp->s_ring_state & S_RING_PROC)) {
323 		cv_signal(&ringp->s_ring_async);
324 	}
325 	mutex_exit(&ringp->s_ring_lock);
326 }
327 
328 /*
329  * mac_rx_soft_ring_drain
330  *
331  * Called when worker thread model (ST_RING_WORKER_ONLY) of processing
332  * incoming packets is used. s_ring_first contain the queued packets.
333  * s_ring_rx_func contains the upper level (client) routine where the
334  * packets are destined and s_ring_rx_arg1/s_ring_rx_arg2 are the
335  * cookie meant for the client.
336  */
337 /* ARGSUSED */
338 static void
339 mac_rx_soft_ring_drain(mac_soft_ring_t *ringp)
340 {
341 	mblk_t		*mp;
342 	void		*arg1;
343 	mac_resource_handle_t arg2;
344 	timeout_id_t 	tid;
345 	mac_direct_rx_t	proc;
346 	size_t		sz;
347 	int		cnt;
348 	mac_soft_ring_set_t	*mac_srs = ringp->s_ring_set;
349 
350 	ringp->s_ring_run = curthread;
351 	ASSERT(mutex_owned(&ringp->s_ring_lock));
352 	ASSERT(!(ringp->s_ring_state & S_RING_PROC));
353 
354 	if ((tid = ringp->s_ring_tid) != 0)
355 		ringp->s_ring_tid = 0;
356 
357 	ringp->s_ring_state |= S_RING_PROC;
358 
359 	proc = ringp->s_ring_rx_func;
360 	arg1 = ringp->s_ring_rx_arg1;
361 	arg2 = ringp->s_ring_rx_arg2;
362 
363 	while ((ringp->s_ring_first != NULL) &&
364 	    !(ringp->s_ring_state & S_RING_PAUSE)) {
365 		mp = ringp->s_ring_first;
366 		ringp->s_ring_first = NULL;
367 		ringp->s_ring_last = NULL;
368 		cnt = ringp->s_ring_count;
369 		ringp->s_ring_count = 0;
370 		sz = ringp->s_ring_size;
371 		ringp->s_ring_size = 0;
372 		mutex_exit(&ringp->s_ring_lock);
373 
374 		if (tid != 0) {
375 			(void) untimeout(tid);
376 			tid = 0;
377 		}
378 
379 		(*proc)(arg1, arg2, mp, NULL);
380 
381 		/*
382 		 * If we have a soft ring set which is doing
383 		 * bandwidth control, we need to decrement its
384 		 * srs_size so it can have a accurate idea of
385 		 * what is the real data queued between SRS and
386 		 * its soft rings. We decrement the size for a
387 		 * packet only when it gets processed by both
388 		 * SRS and the soft ring.
389 		 */
390 		mutex_enter(&mac_srs->srs_lock);
391 		MAC_UPDATE_SRS_COUNT_LOCKED(mac_srs, cnt);
392 		MAC_UPDATE_SRS_SIZE_LOCKED(mac_srs, sz);
393 		mutex_exit(&mac_srs->srs_lock);
394 
395 		mutex_enter(&ringp->s_ring_lock);
396 	}
397 	ringp->s_ring_state &= ~S_RING_PROC;
398 	if (ringp->s_ring_state & S_RING_CLIENT_WAIT)
399 		cv_signal(&ringp->s_ring_client_cv);
400 	ringp->s_ring_run = NULL;
401 }
402 
403 /*
404  * mac_soft_ring_worker
405  *
406  * The soft ring worker routine to process any queued packets. In
407  * normal case, the worker thread is bound to a CPU. It the soft
408  * ring is dealing with TCP packets, then the worker thread will
409  * be bound to the same CPU as the TCP squeue.
410  */
411 static void
412 mac_soft_ring_worker(mac_soft_ring_t *ringp)
413 {
414 	kmutex_t *lock = &ringp->s_ring_lock;
415 	kcondvar_t *async = &ringp->s_ring_async;
416 	mac_soft_ring_set_t *srs = ringp->s_ring_set;
417 	callb_cpr_t cprinfo;
418 
419 	CALLB_CPR_INIT(&cprinfo, lock, callb_generic_cpr, "mac_soft_ring");
420 	mutex_enter(lock);
421 start:
422 	for (;;) {
423 		while (((ringp->s_ring_first == NULL ||
424 		    (ringp->s_ring_state & S_RING_BLOCK)) &&
425 		    !(ringp->s_ring_state & S_RING_PAUSE)) ||
426 		    (ringp->s_ring_state & S_RING_PROC)) {
427 
428 			CALLB_CPR_SAFE_BEGIN(&cprinfo);
429 			cv_wait(async, lock);
430 			CALLB_CPR_SAFE_END(&cprinfo, lock);
431 		}
432 
433 		/*
434 		 * Either we have work to do, or we have been asked to
435 		 * shutdown temporarily or permanently
436 		 */
437 		if (ringp->s_ring_state & S_RING_PAUSE)
438 			goto done;
439 
440 		ringp->s_ring_drain_func(ringp);
441 	}
442 done:
443 	mutex_exit(lock);
444 	mutex_enter(&srs->srs_lock);
445 	mutex_enter(lock);
446 
447 	ringp->s_ring_state |= S_RING_QUIESCE_DONE;
448 	if (!(ringp->s_ring_state & S_RING_CONDEMNED)) {
449 		srs->srs_soft_ring_quiesced_count++;
450 		cv_broadcast(&srs->srs_async);
451 		mutex_exit(&srs->srs_lock);
452 		while (!(ringp->s_ring_state &
453 		    (S_RING_RESTART | S_RING_CONDEMNED)))
454 			cv_wait(&ringp->s_ring_async, &ringp->s_ring_lock);
455 		mutex_exit(lock);
456 		mutex_enter(&srs->srs_lock);
457 		mutex_enter(lock);
458 		srs->srs_soft_ring_quiesced_count--;
459 		if (ringp->s_ring_state & S_RING_RESTART) {
460 			ASSERT(!(ringp->s_ring_state & S_RING_CONDEMNED));
461 			ringp->s_ring_state &= ~(S_RING_RESTART |
462 			    S_RING_QUIESCE | S_RING_QUIESCE_DONE);
463 			cv_broadcast(&srs->srs_async);
464 			mutex_exit(&srs->srs_lock);
465 			goto start;
466 		}
467 	}
468 	ASSERT(ringp->s_ring_state & S_RING_CONDEMNED);
469 	ringp->s_ring_state |= S_RING_CONDEMNED_DONE;
470 	CALLB_CPR_EXIT(&cprinfo);
471 	srs->srs_soft_ring_condemned_count++;
472 	cv_broadcast(&srs->srs_async);
473 	mutex_exit(&srs->srs_lock);
474 	thread_exit();
475 }
476 
477 /*
478  * mac_soft_ring_intr_enable and mac_soft_ring_intr_disable
479  *
480  * these functions are called to toggle the sending of packets to the
481  * client. They are called by the client. the client gets the name
482  * of these routine and corresponding cookie (pointing to softring)
483  * during capability negotiation at setup time.
484  *
485  * Enabling is allow the processing thread to send packets to the
486  * client while disabling does the opposite.
487  */
488 void
489 mac_soft_ring_intr_enable(void *arg)
490 {
491 	mac_soft_ring_t *ringp = (mac_soft_ring_t *)arg;
492 	mutex_enter(&ringp->s_ring_lock);
493 	ringp->s_ring_state &= ~S_RING_BLANK;
494 	if (ringp->s_ring_first != NULL)
495 		mac_soft_ring_worker_wakeup(ringp);
496 	mutex_exit(&ringp->s_ring_lock);
497 }
498 
499 void
500 mac_soft_ring_intr_disable(void *arg)
501 {
502 	mac_soft_ring_t *ringp = (mac_soft_ring_t *)arg;
503 	/*
504 	 * Stop worker thread from sending packets above.
505 	 * Squeue will poll soft ring when it needs packets.
506 	 */
507 	mutex_enter(&ringp->s_ring_lock);
508 	ringp->s_ring_state |= S_RING_BLANK;
509 	mutex_exit(&ringp->s_ring_lock);
510 }
511 
512 /*
513  * mac_soft_ring_poll
514  *
515  * This routine is called by the client to poll for packets from
516  * the soft ring. The function name and cookie corresponding to
517  * the soft ring is exchanged during capability negotiation during
518  * setup.
519  */
520 mblk_t *
521 mac_soft_ring_poll(mac_soft_ring_t *ringp, int bytes_to_pickup)
522 {
523 	mblk_t	*head, *tail;
524 	mblk_t	*mp;
525 	size_t	sz = 0;
526 	int	cnt = 0;
527 	mac_soft_ring_set_t	*mac_srs = ringp->s_ring_set;
528 
529 	ASSERT(mac_srs != NULL);
530 
531 	mutex_enter(&ringp->s_ring_lock);
532 	head = tail = mp = ringp->s_ring_first;
533 	if (head == NULL) {
534 		mutex_exit(&ringp->s_ring_lock);
535 		return (NULL);
536 	}
537 
538 	if (ringp->s_ring_size <= bytes_to_pickup) {
539 		head = ringp->s_ring_first;
540 		ringp->s_ring_first = NULL;
541 		ringp->s_ring_last = NULL;
542 		cnt = ringp->s_ring_count;
543 		ringp->s_ring_count = 0;
544 		sz = ringp->s_ring_size;
545 		ringp->s_ring_size = 0;
546 	} else {
547 		while (mp && sz <= bytes_to_pickup) {
548 			sz += msgdsize(mp);
549 			cnt++;
550 			tail = mp;
551 			mp = mp->b_next;
552 		}
553 		ringp->s_ring_count -= cnt;
554 		ringp->s_ring_size -= sz;
555 		tail->b_next = NULL;
556 		if (mp == NULL) {
557 			ringp->s_ring_first = NULL;
558 			ringp->s_ring_last = NULL;
559 			ASSERT(ringp->s_ring_count == 0);
560 		} else {
561 			ringp->s_ring_first = mp;
562 		}
563 	}
564 
565 	mutex_exit(&ringp->s_ring_lock);
566 	/*
567 	 * Update the shared count and size counters so
568 	 * that SRS has a accurate idea of queued packets.
569 	 */
570 	mutex_enter(&mac_srs->srs_lock);
571 	MAC_UPDATE_SRS_COUNT_LOCKED(mac_srs, cnt);
572 	MAC_UPDATE_SRS_SIZE_LOCKED(mac_srs, sz);
573 	mutex_exit(&mac_srs->srs_lock);
574 	return (head);
575 }
576 
577 /*
578  * mac_soft_ring_dls_bypass
579  *
580  * Enable direct client (IP) callback function from the softrings.
581  * Callers need to make sure they don't need any DLS layer processing
582  */
583 void
584 mac_soft_ring_dls_bypass(void *arg, mac_direct_rx_t rx_func, void *rx_arg1)
585 {
586 	mac_soft_ring_t		*softring = arg;
587 	mac_soft_ring_set_t	*srs;
588 
589 	ASSERT(rx_func != NULL);
590 
591 	mutex_enter(&softring->s_ring_lock);
592 	softring->s_ring_rx_func = rx_func;
593 	softring->s_ring_rx_arg1 = rx_arg1;
594 	mutex_exit(&softring->s_ring_lock);
595 
596 	srs = softring->s_ring_set;
597 	mutex_enter(&srs->srs_lock);
598 	srs->srs_type |= SRST_DLS_BYPASS;
599 	mutex_exit(&srs->srs_lock);
600 }
601 
602 /*
603  * mac_soft_ring_signal
604  *
605  * Typically used to set the soft ring state to QUIESCE, CONDEMNED, or
606  * RESTART.
607  *
608  * In the Rx side, the quiescing is done bottom up. After the Rx upcalls
609  * from the driver are done, then the Rx SRS is quiesced and only then can
610  * we signal the soft rings. Thus this function can't be called arbitrarily
611  * without satisfying the prerequisites. On the Tx side, the threads from
612  * top need to quiesced, then the Tx SRS and only then can we signal the
613  * Tx soft rings.
614  */
615 void
616 mac_soft_ring_signal(mac_soft_ring_t *softring, uint_t sr_flag)
617 {
618 	mutex_enter(&softring->s_ring_lock);
619 	softring->s_ring_state |= sr_flag;
620 	cv_signal(&softring->s_ring_async);
621 	mutex_exit(&softring->s_ring_lock);
622 }
623 
624 /*
625  * mac_tx_soft_ring_drain
626  *
627  * The transmit side drain routine in case the soft ring was being
628  * used to transmit packets.
629  */
630 static void
631 mac_tx_soft_ring_drain(mac_soft_ring_t *ringp)
632 {
633 	mblk_t 			*mp;
634 	void 			*arg1;
635 	void 			*arg2;
636 	mblk_t 			*tail;
637 	uint_t			saved_pkt_count, saved_size;
638 	boolean_t		is_subflow;
639 	mac_tx_stats_t		stats;
640 	mac_soft_ring_set_t	*mac_srs = ringp->s_ring_set;
641 
642 	saved_pkt_count = saved_size = 0;
643 	ringp->s_ring_run = curthread;
644 	ASSERT(mutex_owned(&ringp->s_ring_lock));
645 	ASSERT(!(ringp->s_ring_state & S_RING_PROC));
646 
647 	ringp->s_ring_state |= S_RING_PROC;
648 	is_subflow = ((mac_srs->srs_type & SRST_FLOW) != 0);
649 	arg1 = ringp->s_ring_tx_arg1;
650 	arg2 = ringp->s_ring_tx_arg2;
651 
652 	while (ringp->s_ring_first != NULL) {
653 		mp = ringp->s_ring_first;
654 		tail = ringp->s_ring_last;
655 		saved_pkt_count = ringp->s_ring_count;
656 		saved_size = ringp->s_ring_size;
657 		ringp->s_ring_first = NULL;
658 		ringp->s_ring_last = NULL;
659 		ringp->s_ring_count = 0;
660 		ringp->s_ring_size = 0;
661 		mutex_exit(&ringp->s_ring_lock);
662 
663 		mp = mac_tx_send(arg1, arg2, mp, &stats);
664 
665 		mutex_enter(&ringp->s_ring_lock);
666 		if (mp != NULL) {
667 			/* Device out of tx desc, set block */
668 			tail->b_next = ringp->s_ring_first;
669 			ringp->s_ring_first = mp;
670 			ringp->s_ring_count +=
671 			    (saved_pkt_count - stats.ts_opackets);
672 			ringp->s_ring_size += (saved_size - stats.ts_obytes);
673 			if (ringp->s_ring_last == NULL)
674 				ringp->s_ring_last = tail;
675 
676 			if (ringp->s_ring_tx_woken_up) {
677 				ringp->s_ring_tx_woken_up = B_FALSE;
678 			} else {
679 				ringp->s_ring_state |= S_RING_BLOCK;
680 				ringp->s_ring_blocked_cnt++;
681 			}
682 
683 			ringp->s_ring_state &= ~S_RING_PROC;
684 			ringp->s_ring_run = NULL;
685 			return;
686 		} else {
687 			ringp->s_ring_tx_woken_up = B_FALSE;
688 			if (is_subflow) {
689 				FLOW_TX_STATS_UPDATE(
690 				    mac_srs->srs_flent, &stats);
691 			}
692 		}
693 	}
694 
695 	if (ringp->s_ring_count == 0 && ringp->s_ring_state &
696 	    (S_RING_TX_HIWAT | S_RING_WAKEUP_CLIENT | S_RING_ENQUEUED)) {
697 		mac_tx_notify_cb_t *mtnfp;
698 		mac_cb_t *mcb;
699 		mac_client_impl_t *mcip =  ringp->s_ring_mcip;
700 		boolean_t wakeup_required = B_FALSE;
701 
702 		if (ringp->s_ring_state &
703 		    (S_RING_TX_HIWAT|S_RING_WAKEUP_CLIENT)) {
704 			wakeup_required = B_TRUE;
705 		}
706 		ringp->s_ring_state &=
707 		    ~(S_RING_TX_HIWAT | S_RING_WAKEUP_CLIENT | S_RING_ENQUEUED);
708 		mutex_exit(&ringp->s_ring_lock);
709 		if (wakeup_required) {
710 			/* Wakeup callback registered clients */
711 			MAC_CALLBACK_WALKER_INC(&mcip->mci_tx_notify_cb_info);
712 			for (mcb = mcip->mci_tx_notify_cb_list; mcb != NULL;
713 			    mcb = mcb->mcb_nextp) {
714 				mtnfp = (mac_tx_notify_cb_t *)mcb->mcb_objp;
715 				mtnfp->mtnf_fn(mtnfp->mtnf_arg,
716 				    (mac_tx_cookie_t)ringp);
717 			}
718 			MAC_CALLBACK_WALKER_DCR(&mcip->mci_tx_notify_cb_info,
719 			    &mcip->mci_tx_notify_cb_list);
720 			/*
721 			 * If the client is not the primary MAC client, then we
722 			 * need to send the notification to the clients upper
723 			 * MAC, i.e. mci_upper_mip.
724 			 */
725 			mac_tx_notify(mcip->mci_upper_mip != NULL ?
726 			    mcip->mci_upper_mip : mcip->mci_mip);
727 		}
728 		mutex_enter(&ringp->s_ring_lock);
729 	}
730 	ringp->s_ring_state &= ~S_RING_PROC;
731 	ringp->s_ring_run = NULL;
732 }
733