Producer/Consumer: Convert USART and USB Stream drivers

Previously the USART and USB Stream drivers exposed in_stream
and out_stream interfaces, which don't allow for sharing their
queues easily.  This change converts these drivers over to the
producer/consumer model and updates the two uses.

Signed-off-by: Anton Staaf <robotboy@chromium.org>

BRANCH=None
BUG=None
TEST=make buildall -j
     Verify that the discovery echo functionality is unchanged.

Change-Id: I29f043ab1712373f638e1621378df98647d736cf
Reviewed-on: https://chromium-review.googlesource.com/252820
Trybot-Ready: Anton Staaf <robotboy@chromium.org>
Tested-by: Anton Staaf <robotboy@chromium.org>
Reviewed-by: Randall Spangler <rspangler@chromium.org>
Reviewed-by: Todd Broch <tbroch@chromium.org>
Tested-by: Todd Broch <tbroch@chromium.org>
Commit-Queue: Anton Staaf <robotboy@chromium.org>

6 files changed, 221 insertions, 296 deletions

diff --git a/board/discovery-stm32f072/echo.c b/board/discovery-stm32f072/echo.c
index f8396ba3a6..56f99a1ddf 100644
--- a/board/discovery-stm32f072/echo.c
+++ b/board/discovery-stm32f072/echo.c
@@ -13,6 +13,7 @@
 #include "console.h"
 #include "panic.h"
 #include "task.h"
+#include "stream_adaptor.h"
 #include "timer.h"
 #include "usart-stm32f0.h"
 #include "usb-stream.h"
@@ -28,17 +29,56 @@ static void out_ready(struct out_stream const *stream)
 	task_wake(TASK_ID_ECHO);
 }
 
-USART_CONFIG(usart1, usart1_hw, 115200, 64, 64, in_ready, NULL)
-USART_CONFIG(usart3, usart3_hw, 115200, 64, 64, in_ready, NULL)
-USART_CONFIG(usart4, usart4_hw, 115200, 64, 64, in_ready, NULL)
+#define USART_STREAM_CONFIG(NAME,					\
+			    HW,						\
+			    BAUD,					\
+			    RX_SIZE,					\
+			    TX_SIZE,					\
+			    IN_READY,					\
+			    OUT_READY)					\
+									\
+	QUEUE_CONFIG(CONCAT2(NAME, _rx_queue), RX_SIZE, uint8_t);	\
+	QUEUE_CONFIG(CONCAT2(NAME, _tx_queue), TX_SIZE, uint8_t);	\
+									\
+	struct usart_config const NAME;					\
+									\
+	IN_STREAM_FROM_PRODUCER(CONCAT2(NAME, _in),			\
+				NAME.producer,				\
+				CONCAT2(NAME, _rx_queue),		\
+				IN_READY)				\
+	OUT_STREAM_FROM_CONSUMER(CONCAT2(NAME, _out),			\
+				 NAME.consumer,				\
+				 CONCAT2(NAME, _tx_queue),		\
+				 OUT_READY)				\
+									\
+	USART_CONFIG(NAME,						\
+		     HW,						\
+		     BAUD,						\
+		     CONCAT2(NAME, _rx_queue),				\
+		     CONCAT2(NAME, _tx_queue),				\
+		     CONCAT2(NAME, _in).consumer,			\
+		     CONCAT2(NAME, _out).producer)
+
+USART_STREAM_CONFIG(usart1, usart1_hw, 115200, 64, 64, in_ready, NULL);
+USART_STREAM_CONFIG(usart3, usart3_hw, 115200, 64, 64, in_ready, NULL);
+USART_STREAM_CONFIG(usart4, usart4_hw, 115200, 64, 64, in_ready, NULL);
+
+QUEUE_CONFIG(usb_rx_queue, 256, uint8_t);
+QUEUE_CONFIG(usb_tx_queue, 256, uint8_t);
+
+struct usb_stream_config const usb_stream1;
+
+IN_STREAM_FROM_PRODUCER(usb_in, usb_stream1.producer, usb_rx_queue, in_ready)
+OUT_STREAM_FROM_CONSUMER(usb_out, usb_stream1.consumer, usb_tx_queue, out_ready)
+
 USB_STREAM_CONFIG(usb_stream1,
 		  USB_IFACE_STREAM,
 		  USB_STR_STREAM_NAME,
 		  USB_EP_STREAM,
-		  256,
-		  256,
-		  in_ready,
-		  out_ready)
+		  usb_rx_queue,
+		  usb_tx_queue,
+		  usb_in.consumer,
+		  usb_out.producer)
 
 struct stream_console_state {
 	size_t wrote;
@@ -59,10 +99,10 @@ struct stream_console_config {
 		.out   = OUT,					\
 	};
 
-STREAM_CONSOLE_CONFIG(usart1_stream_console, &usart1.in, &usart1.out)
-STREAM_CONSOLE_CONFIG(usart3_stream_console, &usart3.in, &usart3.out)
-STREAM_CONSOLE_CONFIG(usart4_stream_console, &usart4.in, &usart4.out)
-STREAM_CONSOLE_CONFIG(usb_stream1_console, &usb_stream1.in, &usb_stream1.out)
+STREAM_CONSOLE_CONFIG(usart1_stream_console, &usart1_in.in, &usart1_out.out)
+STREAM_CONSOLE_CONFIG(usart3_stream_console, &usart3_in.in, &usart3_out.out)
+STREAM_CONSOLE_CONFIG(usart4_stream_console, &usart4_in.in, &usart4_out.out)
+STREAM_CONSOLE_CONFIG(usb_stream1_console, &usb_in.in, &usb_out.out)
 
 static struct stream_console_config const *const consoles[] = {
 	&usart1_stream_console,
diff --git a/chip/stm32/usart.c b/chip/stm32/usart.c
index d97eb45565..087843de18 100644
--- a/chip/stm32/usart.c
+++ b/chip/stm32/usart.c
@@ -14,60 +14,51 @@
 #include "usart.h"
 #include "util.h"
 
-static size_t usart_read(struct in_stream const *stream,
-			 uint8_t *buffer,
-			 size_t count)
+static void usart_written(struct consumer const *consumer, size_t count)
 {
 	struct usart_config const *config =
-		DOWNCAST(stream, struct usart_config, in);
-
-	return QUEUE_REMOVE_UNITS(&config->rx, buffer, count);
-}
-
-static size_t usart_write(struct out_stream const *stream,
-			  uint8_t const *buffer,
-			  size_t count)
-{
-	struct usart_config const *config =
-		DOWNCAST(stream, struct usart_config, out);
-
-	size_t wrote = QUEUE_ADD_UNITS(&config->tx, buffer, count);
+		DOWNCAST(consumer, struct usart_config, consumer);
 
 	/*
 	 * Enable USART interrupt.  This causes the USART interrupt handler to
 	 * start fetching from the TX queue if it wasn't already.
 	 */
-	if (wrote)
+	if (count)
 		STM32_USART_CR1(config->hw->base) |= STM32_USART_CR1_TXEIE;
-
-	return wrote;
 }
 
-static void usart_flush(struct out_stream const *stream)
+static void usart_flush(struct consumer const *consumer)
 {
 	struct usart_config const *config =
-		DOWNCAST(stream, struct usart_config, out);
+		DOWNCAST(consumer, struct usart_config, consumer);
+
+	/*
+	 * Enable USART interrupt.  This causes the USART interrupt handler to
+	 * start fetching from the TX queue if it wasn't already.
+	 */
+	STM32_USART_CR1(config->hw->base) |= STM32_USART_CR1_TXEIE;
 
-	while (queue_count(&config->tx))
+	while (queue_count(consumer->queue))
 		;
 }
 
-struct in_stream_ops const usart_in_stream_ops = {
-	.read = usart_read,
+struct producer_ops const usart_producer_ops = {
+	/*
+	 * Nothing to do here, we either had enough space in the queue when
+	 * a character came in or we dropped it already.
+	 */
+	.read = NULL,
 };
 
-struct out_stream_ops const usart_out_stream_ops = {
-	.write = usart_write,
-	.flush = usart_flush,
+struct consumer_ops const usart_consumer_ops = {
+	.written = usart_written,
+	.flush   = usart_flush,
 };
 
 void usart_init(struct usart_config const *config)
 {
 	intptr_t base = config->hw->base;
 
-	queue_init(&config->tx);
-	queue_init(&config->rx);
-
 	/*
 	 * Enable clock to USART, this must be done first, before attempting
 	 * to configure the USART.
@@ -151,11 +142,9 @@ static void usart_interrupt_tx(struct usart_config const *config)
 	intptr_t base = config->hw->base;
 	uint8_t  byte;
 
-	if (queue_remove_unit(&config->tx, &byte)) {
+	if (consumer_read_unit(&config->consumer, &byte)) {
 		STM32_USART_TDR(base) = byte;
 
-		out_stream_ready(&config->out);
-
 		/*
 		 * Make sure the TXE interrupt is enabled and that we won't go
 		 * into deep sleep.  This invocation of the USART interrupt
@@ -179,19 +168,11 @@ static void usart_interrupt_tx(struct usart_config const *config)
 
 static void usart_interrupt_rx(struct usart_config const *config)
 {
-	intptr_t base    = config->hw->base;
-	uint8_t  byte    = STM32_USART_RDR(base);
-	uint32_t dropped = 1 - queue_add_unit(&config->rx, &byte);
-
-	atomic_add((uint32_t *) &config->state->rx_dropped, dropped);
+	intptr_t base = config->hw->base;
+	uint8_t  byte = STM32_USART_RDR(base);
 
-	/*
-	 * Wake up whoever is listening on the other end of the queue.  The
-	 * queue_add_units call above may have failed due to a full queue, but
-	 * it doesn't really matter to the ready callback because there will be
-	 * something in the queue to consume either way.
-	 */
-	in_stream_ready(&config->in);
+	if (!producer_write_unit(&config->producer, &byte))
+		atomic_add((uint32_t *) &config->state->rx_dropped, 1);
 }
 
 void usart_interrupt(struct usart_config const *config)
diff --git a/chip/stm32/usart.h b/chip/stm32/usart.h
index 3293dd5722..3ed16a8796 100644
--- a/chip/stm32/usart.h
+++ b/chip/stm32/usart.h
@@ -8,21 +8,23 @@
 /* STM32 USART driver for Chrome EC */
 
 #include "common.h"
+#include "consumer.h"
 #include "in_stream.h"
 #include "out_stream.h"
+#include "producer.h"
 #include "queue.h"
 
 #include <stdint.h>
 
 /*
  * Per-USART state stored in RAM.  This structure will be zero initialized by
- * BSS init.  Most importantly, irq_lock will be zero, ensuring that shared
- * interrupts don't cause problems.
+ * BSS init.
  */
 struct usart_state {
-	struct queue_state rx;
-	struct queue_state tx;
-
+	/*
+	 * Counter of bytes receieved and then dropped because of lack of space
+	 * in the RX queue.
+	 */
 	uint32_t rx_dropped;
 };
 
@@ -38,6 +40,9 @@ struct usart_hw_ops {
 	void (*enable)(struct usart_config const *config);
 
 	/*
+	 * The generic USART shutdown code calls this function, allowing the
+	 * variant specific code an opportunity to do any variant specific
+	 * shutdown tasks.
 	 */
 	void (*disable)(struct usart_config const *config);
 };
@@ -71,8 +76,7 @@ struct usart_config {
 
 	/*
 	 * Pointer to USART state structure.  The state structure maintains per
-	 * USART information (head and tail pointers for the queues for
-	 * instance).
+	 * USART information.
 	 */
 	struct usart_state volatile *state;
 
@@ -81,82 +85,68 @@ struct usart_config {
 	 */
 	int baud;
 
-	/*
-	 * TX and RX queue configs.  The state for the queue is stored
-	 * separately in the usart_state structure.
-	 */
-	struct queue rx;
-	struct queue tx;
-
-	/*
-	 * In and Out streams, these contain pointers to the virtual function
-	 * tables that implement in and out streams.  They can be used by any
-	 * code that wants to read or write to a stream interface.
-	 */
-	struct in_stream  in;
-	struct out_stream out;
+	struct consumer consumer;
+	struct producer producer;
 };
 
 /*
  * These function tables are defined by the USART driver and are used to
- * initialize the in and out streams in the usart_config.
+ * initialize the consumer and producer in the usart_config.
  */
-extern struct in_stream_ops const usart_in_stream_ops;
-extern struct out_stream_ops const usart_out_stream_ops;
+extern struct consumer_ops const usart_consumer_ops;
+extern struct producer_ops const usart_producer_ops;
 
 /*
  * Convenience macro for defining USARTs and their associated state and buffers.
- * NAME is used to construct the names of the queue buffers, usart_state struct,
- * and usart_config struct, the latter is just called NAME.  RX_SIZE and TX_SIZE
- * are the size in bytes of the RX and TX buffers respectively.  RX_READY and
- * TX_READY are the callback functions for the in and out streams.  The USART
- * baud rate is specified by the BAUD parameter.
+ * NAME is used to construct the names of the usart_state struct, and
+ * usart_config struct, the latter is just called NAME.
+ *
+ * HW is the name of the usart_hw_config provided by the variant specific code.
  *
- * If you want to share a queue with other code, you can manually initialize a
- * usart_config to use the shared queue, or you can use this macro and then get
- * the queue buffers as <NAME>_tx_buffer, and <NAME>_rx_buffer.
+ * RX_QUEUE and TX_QUEUE are the names of the RX and TX queues that this USART
+ * should write to and read from respectively.  They must match the queues
+ * that the CONSUMER and PRODUCER read from and write to respectively.
+ *
+ * CONSUMER and PRODUCER are the names of the consumer and producer objects at
+ * the other ends of the RX and TX queues respectively.
  */
-#define USART_CONFIG(NAME,						\
-		     HW,						\
-		     BAUD,						\
-		     RX_SIZE,						\
-		     TX_SIZE,						\
-		     RX_READY,						\
-		     TX_READY)						\
-	static uint8_t CONCAT2(NAME, _tx_buffer)[TX_SIZE];		\
-	static uint8_t CONCAT2(NAME, _rx_buffer)[RX_SIZE];		\
-									\
-	static struct usart_state CONCAT2(NAME, _state);		\
-	struct usart_config const NAME = {				\
-		.hw    = &HW,						\
-		.state = &CONCAT2(NAME, _state),			\
-		.baud  = BAUD,						\
-		.rx = {							\
-			.state        = &CONCAT2(NAME, _state.rx),	\
-			.buffer_units = RX_SIZE,			\
-			.unit_bytes   = 1,				\
-			.buffer       = CONCAT2(NAME, _rx_buffer),	\
-		},							\
-		.tx = {							\
-			.state        = &CONCAT2(NAME, _state.tx),	\
-			.buffer_units = TX_SIZE,			\
-			.unit_bytes   = 1,				\
-			.buffer       = CONCAT2(NAME, _tx_buffer),	\
-		},							\
-		.in  = {						\
-			.ready = RX_READY,				\
-			.ops   = &usart_in_stream_ops,			\
-		},							\
-		.out = {						\
-			.ready = TX_READY,				\
-			.ops   = &usart_out_stream_ops,			\
-		},							\
+/*
+ * The following assertions can not be made because they require access to
+ * non-const fields, but should be kept in mind.
+ *
+ * BUILD_ASSERT(RX_QUEUE.unit_bytes == 1);
+ * BUILD_ASSERT(TX_QUEUE.unit_bytes == 1);
+ * BUILD_ASSERT(PRODUCER.queue == &TX_QUEUE);
+ * BUILD_ASSERT(CONSUMER.queue == &RX_QUEUE);
+ */
+#define USART_CONFIG(NAME,					\
+		     HW,					\
+		     BAUD,					\
+		     RX_QUEUE,					\
+		     TX_QUEUE,					\
+		     CONSUMER,					\
+		     PRODUCER)					\
+								\
+	static struct usart_state CONCAT2(NAME, _state);	\
+	struct usart_config const NAME = {			\
+		.hw       = &HW,				\
+		.state    = &CONCAT2(NAME, _state),		\
+		.baud     = BAUD,				\
+		.consumer = {					\
+			.producer = &PRODUCER,			\
+			.queue    = &TX_QUEUE,			\
+			.ops      = &usart_consumer_ops,	\
+		},						\
+		.producer = {					\
+			.consumer = &CONSUMER,			\
+			.queue    = &RX_QUEUE,			\
+			.ops      = &usart_producer_ops,	\
+		},						\
 	};
 
 /*
  * Initialize the given USART.  Once init is finished the USART streams are
- * available for operating on, and the stream ready callbacks could be called
- * at any time.
+ * available for operating on.
  */
 void usart_init(struct usart_config const *config);
 
@@ -169,8 +159,6 @@ void usart_shutdown(struct usart_config const *config);
  * Handle a USART interrupt.  The per-variant USART code creates bindings
  * for the variants interrupts to call this generic USART interrupt handler
  * with the appropriate usart_config.
- *
- * This function could also be called manually to poll the USART hardware.
  */
 void usart_interrupt(struct usart_config const *config);
 
diff --git a/chip/stm32/usb-stream.c b/chip/stm32/usb-stream.c
index 644ac88dbb..42d9d3774c 100644
--- a/chip/stm32/usb-stream.c
+++ b/chip/stm32/usb-stream.c
@@ -15,112 +15,45 @@
 #include "usb.h"
 #include "usb-stream.h"
 
-/*
- * The USB packet RAM is attached to the processor via the AHB2APB bridge.  This
- * bridge performs manipulations of read and write accesses as per the note in
- * section 2.1 of RM0091.  The upshot is that it is OK to read from the packet
- * RAM using 8-bit or 16-bit accesses, but not 32-bit, and it is only really OK
- * to write to the packet RAM using 16-bit accesses.  Thus custom memcpy like
- * routines need to be employed.  Furthermore, reading from and writing to the
- * RX and TX queues uses memcpy, which will try to do 32-bit accesses if it can.
- * so we must read and write single bytes at a time and construct 16-bit
- * accesses to the packet RAM.
- *
- * This could be improved by adding a set of operations on the queue to get
- * a pointer and size of the largest contiguous free/full region, then that
- * region could be operated on and a commit operation could be performed on
- * the queue.
- */
 static size_t rx_read(struct usb_stream_config const *config)
 {
 	size_t count = btable_ep[config->endpoint].rx_count & 0x3ff;
 
-	if (count < queue_space(&config->rx)) {
-		usb_uint *buffer = config->rx_ram;
-		size_t    i;
-
-		for (i = 0; i < count / 2; i++) {
-			usb_uint word = *buffer++;
-			uint8_t  lsb  = (word >> 0) & 0xff;
-			uint8_t  msb  = (word >> 8) & 0xff;
-
-			queue_add_unit(&config->rx, &lsb);
-			queue_add_unit(&config->rx, &msb);
-		}
-
-		if (count & 1) {
-			usb_uint word = *buffer++;
-			uint8_t  lsb  = (word >> 0) & 0xff;
-
-			queue_add_unit(&config->rx, &lsb);
-		}
-
-		return count;
-	}
+	/*
+	 * Only read the received USB packet if there is enough space in the
+	 * receive queue.
+	 */
+	if (count >= queue_space(config->producer.queue))
+		return 0;
 
-	return 0;
+	return producer_write_memcpy(&config->producer,
+				     config->rx_ram,
+				     count,
+				     memcpy_from_usbram);
 }
 
 static size_t tx_write(struct usb_stream_config const *config)
 {
-	usb_uint *buffer = config->tx_ram;
-	size_t    count  = MIN(USB_MAX_PACKET_SIZE, queue_count(&config->tx));
-	size_t    i;
-
-	for (i = 0; i < count / 2; i++) {
-		uint8_t lsb;
-		uint8_t msb;
-
-		queue_remove_unit(&config->tx, &lsb);
-		queue_remove_unit(&config->tx, &msb);
-
-		*buffer++ = (msb << 8) | lsb;
-	}
-
-	if (count & 1) {
-		uint8_t lsb;
-
-		queue_remove_unit(&config->tx, &lsb);
-
-		*buffer++ = lsb;
-	}
+	size_t count = consumer_read_memcpy(&config->consumer,
+					    config->tx_ram,
+					    USB_MAX_PACKET_SIZE,
+					    memcpy_to_usbram);
 
 	btable_ep[config->endpoint].tx_count = count;
 
 	return count;
 }
 
-static size_t usb_read(struct in_stream const *stream,
-		       uint8_t *buffer,
-		       size_t count)
+static void usb_read(struct producer const *producer, size_t count)
 {
 	struct usb_stream_config const *config =
-		DOWNCAST(stream, struct usb_stream_config, in);
-
-	size_t read = QUEUE_REMOVE_UNITS(&config->rx, buffer, count);
+		DOWNCAST(producer, struct usb_stream_config, producer);
 
 	if (config->state->rx_waiting && rx_read(config)) {
 		config->state->rx_waiting = 0;
 
 		STM32_TOGGLE_EP(config->endpoint, EP_RX_MASK, EP_RX_VALID, 0);
-
-		/*
-		 * Make sure that the reader of this queue knows that there is
-		 * more to read.
-		 */
-		in_stream_ready(&config->in);
-
-		/*
-		 * If there is still space left in the callers buffer fill it
-		 * up with the additional bytes just added to the queue.
-		 */
-		if (count - read > 0)
-			read += QUEUE_REMOVE_UNITS(&config->rx,
-						   buffer + read,
-						   count - read);
 	}
-
-	return read;
 }
 
 static int tx_valid(struct usb_stream_config const *config)
@@ -128,14 +61,10 @@ static int tx_valid(struct usb_stream_config const *config)
 	return (STM32_USB_EP(config->endpoint) & EP_TX_MASK) == EP_TX_VALID;
 }
 
-static size_t usb_write(struct out_stream const *stream,
-			uint8_t const *buffer,
-			size_t count)
+static void usb_written(struct consumer const *consumer, size_t count)
 {
 	struct usb_stream_config const *config =
-		DOWNCAST(stream, struct usb_stream_config, out);
-
-	size_t wrote = QUEUE_ADD_UNITS(&config->tx, buffer, count);
+		DOWNCAST(consumer, struct usb_stream_config, consumer);
 
 	/*
 	 * If we are not currently in a valid transmission state and we had
@@ -143,26 +72,24 @@ static size_t usb_write(struct out_stream const *stream,
 	 */
 	if (!tx_valid(config) && tx_write(config))
 		STM32_TOGGLE_EP(config->endpoint, EP_TX_MASK, EP_TX_VALID, 0);
-
-	return wrote;
 }
 
-static void usb_flush(struct out_stream const *stream)
+static void usb_flush(struct consumer const *consumer)
 {
 	struct usb_stream_config const *config =
-		DOWNCAST(stream, struct usb_stream_config, out);
+		DOWNCAST(consumer, struct usb_stream_config, consumer);
 
-	while (tx_valid(config) || queue_count(&config->tx))
+	while (tx_valid(config) || queue_count(consumer->queue))
 		;
 }
 
-struct in_stream_ops const usb_stream_in_stream_ops = {
+struct producer_ops const usb_stream_producer_ops = {
 	.read = usb_read,
 };
 
-struct out_stream_ops const usb_stream_out_stream_ops = {
-	.write = usb_write,
-	.flush = usb_flush,
+struct consumer_ops const usb_stream_consumer_ops = {
+	.written = usb_written,
+	.flush   = usb_flush,
 };
 
 void usb_stream_tx(struct usb_stream_config const *config)
@@ -171,8 +98,6 @@ void usb_stream_tx(struct usb_stream_config const *config)
 		STM32_TOGGLE_EP(config->endpoint, EP_TX_MASK, EP_TX_VALID, 0);
 	else
 		STM32_TOGGLE_EP(config->endpoint, 0, 0, 0);
-
-	out_stream_ready(&config->out);
 }
 
 void usb_stream_rx(struct usb_stream_config const *config)
@@ -193,8 +118,6 @@ void usb_stream_rx(struct usb_stream_config const *config)
 		config->state->rx_waiting = 1;
 		STM32_TOGGLE_EP(config->endpoint, 0, 0, 0);
 	}
-
-	in_stream_ready(&config->in);
 }
 
 void usb_stream_reset(struct usb_stream_config const *config)
diff --git a/chip/stm32/usb-stream.h b/chip/stm32/usb-stream.h
index 796eacaa17..06374676f9 100644
--- a/chip/stm32/usb-stream.h
+++ b/chip/stm32/usb-stream.h
@@ -8,8 +8,10 @@
 /* STM32 USB STREAM driver for Chrome EC */
 
 #include "compile_time_macros.h"
+#include "consumer.h"
 #include "in_stream.h"
 #include "out_stream.h"
+#include "producer.h"
 #include "queue.h"
 #include "usb.h"
 
@@ -19,19 +21,13 @@
  * Per-USB stream state stored in RAM.  Zero initialization of this structure
  * by the BSS initialization leaves it in a valid and correctly initialized
  * state, so there is no need currently for a usb_stream_init style function.
- *
- * If this structure is changed to require non-zero initialization such a
- * function should be added.
  */
 struct usb_stream_state {
-	struct queue_state rx;
-	struct queue_state tx;
-
 	/*
 	 * Flag indicating that there is a full RX buffer in the USB packet RAM
 	 * that we were not able to move into the RX queue because there was
 	 * not enough room when the packet was initially received.  The
-	 * in_stream read operation checks this flag so that once there is
+	 * producer read operation checks this flag so that once there is
 	 * room in the queue it can copy the RX buffer into the queue and
 	 * restart USB reception by marking the RX buffer as VALID.
 	 */
@@ -46,8 +42,7 @@ struct usb_stream_state {
 struct usb_stream_config {
 	/*
 	 * Pointer to usb_stream_state structure.  The state structure
-	 * maintains per USB stream information (head and tail pointers for
-	 * the queues for instance).
+	 * maintains per USB stream information.
 	 */
 	struct usb_stream_state volatile *state;
 
@@ -59,62 +54,60 @@ struct usb_stream_config {
 	usb_uint *rx_ram;
 	usb_uint *tx_ram;
 
-	/*
-	 * RX and TX queue config.  The state for the queue is stored
-	 * separately in the usb_stream_state structure.
-	 */
-	struct queue rx;
-	struct queue tx;
-
-	/*
-	 * In and Out streams, these contain pointers to the virtual function
-	 * tables that implement in and out streams.  They can be used by any
-	 * code that wants to read or write to a stream interface.
-	 */
-	struct in_stream  in;
-	struct out_stream out;
+	struct consumer consumer;
+	struct producer producer;
 };
 
 /*
- * These function tables are defined by the USB stream driver and are used to
- * initialize the in and out streams in the usb_stream_config.
+ * These function tables are defined by the USB Stream driver and are used to
+ * initialize the consumer and producer in the usb_stream_config.
  */
-extern struct in_stream_ops const usb_stream_in_stream_ops;
-extern struct out_stream_ops const usb_stream_out_stream_ops;
+extern struct consumer_ops const usb_stream_consumer_ops;
+extern struct producer_ops const usb_stream_producer_ops;
 
 /*
  * Convenience macro for defining USB streams and their associated state and
  * buffers.
  *
- * NAME is used to construct the names of the queue buffers, trampoline
+ * NAME is used to construct the names of the packet RAM buffers, trampoline
  * functions, usb_stream_state struct, and usb_stream_config struct, the
  * latter is just called NAME.
  *
  * INTERFACE is the index of the USB interface to associate with this
  * stream.
  *
+ * INTERFACE_NAME is the index of the USB string descriptor (iInterface).
+ *
  * ENDPOINT is the index of the USB bulk endpoint used for receiving and
  * transmitting bytes.
  *
- * RX_SIZE and TX_SIZE are the size in bytes of the RX and TX queue buffers
- * respectively.
+ * RX_QUEUE and TX_QUEUE are the names of the RX and TX queues that this driver
+ * should write to and read from respectively.  They must match the queues
+ * that the CONSUMER and PRODUCER read from and write to respectively.
+ *
+ * CONSUMER and PRODUCER are the names of the consumer and producer objects at
+ * the other ends of the RX and TX queues respectively.
+ */
+/*
+ * The following assertions can not be made because they require access to
+ * non-const fields, but should be kept in mind.
  *
- * RX_READY and TX_READY are the callback functions for the in and out streams.
- * These functions are called when there are bytes to read or space for bytes
- * to write respectively.
+ * BUILD_ASSERT(RX_QUEUE.buffer_units >= USB_MAX_PACKET_SIZE);
+ * BUILD_ASSERT(TX_QUEUE.buffer_units >= USB_MAX_PACKET_SIZE);
+ * BUILD_ASSERT(RX_QUEUE.unit_bytes == 1);
+ * BUILD_ASSERT(TX_QUEUE.unit_bytes == 1);
+ * BUILD_ASSERT(PRODUCER.queue == &TX_QUEUE);
+ * BUILD_ASSERT(CONSUMER.queue == &RX_QUEUE);
  */
 #define USB_STREAM_CONFIG(NAME,						\
 			  INTERFACE,					\
 			  INTERFACE_NAME,				\
 			  ENDPOINT,					\
-			  RX_SIZE,					\
-			  TX_SIZE,					\
-			  RX_READY,					\
-			  TX_READY)					\
-	BUILD_ASSERT(RX_SIZE >= USB_MAX_PACKET_SIZE);			\
-	BUILD_ASSERT(TX_SIZE >= USB_MAX_PACKET_SIZE);			\
-	static uint8_t  CONCAT2(NAME, _rx_buffer)[RX_SIZE];		\
-	static uint8_t  CONCAT2(NAME, _tx_buffer)[TX_SIZE];		\
+			  RX_QUEUE,					\
+			  TX_QUEUE,					\
+			  CONSUMER,					\
+			  PRODUCER)					\
+									\
 	static usb_uint CONCAT2(NAME, _ep_rx_buffer)[USB_MAX_PACKET_SIZE / 2] __usb_ram; \
 	static usb_uint CONCAT2(NAME, _ep_tx_buffer)[USB_MAX_PACKET_SIZE / 2] __usb_ram; \
 	static struct usb_stream_state CONCAT2(NAME, _state);		\
@@ -123,25 +116,15 @@ extern struct out_stream_ops const usb_stream_out_stream_ops;
 		.endpoint = ENDPOINT,					\
 		.rx_ram   = CONCAT2(NAME, _ep_rx_buffer),		\
 		.tx_ram   = CONCAT2(NAME, _ep_tx_buffer),		\
-		.rx = {							\
-			.state        = &CONCAT2(NAME, _state.rx),	\
-			.buffer_units = RX_SIZE,			\
-			.unit_bytes   = 1,				\
-			.buffer       = CONCAT2(NAME, _rx_buffer),	\
-		},							\
-		.tx = {							\
-			.state        = &CONCAT2(NAME, _state.tx),	\
-			.buffer_units = TX_SIZE,			\
-			.unit_bytes   = 1,				\
-			.buffer       = CONCAT2(NAME, _tx_buffer),	\
-		},							\
-		.in  = {						\
-			.ready = RX_READY,				\
-			.ops   = &usb_stream_in_stream_ops,		\
+		.consumer = {						\
+			.producer = &PRODUCER,				\
+			.queue    = &TX_QUEUE,				\
+			.ops      = &usb_stream_consumer_ops,		\
 		},							\
-		.out = {						\
-			.ready = TX_READY,				\
-			.ops   = &usb_stream_out_stream_ops,		\
+		.producer = {						\
+			.consumer = &CONSUMER,				\
+			.queue    = &RX_QUEUE,				\
+			.ops      = &usb_stream_producer_ops,		\
 		},							\
 	};								\
 	const struct usb_interface_descriptor				\
diff --git a/driver/mcdp28x0.c b/driver/mcdp28x0.c
index de2572c87e..7fba0bd035 100644
--- a/driver/mcdp28x0.c
+++ b/driver/mcdp28x0.c
@@ -10,6 +10,7 @@
 #include "common.h"
 #include "ec_commands.h"
 #include "mcdp28x0.h"
+#include "stream_adaptor.h"
 #include "timer.h"
 #include "usart-stm32f0.h"
 #include "util.h"
@@ -34,8 +35,17 @@ static inline void print_buffer(uint8_t *buf, int cnt)
 static inline void print_buffer(uint8_t *buf, int cnt) {}
 #endif
 
-USART_CONFIG(usart_mcdp, CONFIG_MCDP28X0, 115200, MCDP_INBUF_MAX,
-	     MCDP_OUTBUF_MAX, NULL, NULL);
+QUEUE_CONFIG(rx_queue, MCDP_INBUF_MAX,  uint8_t);
+QUEUE_CONFIG(tx_queue, MCDP_OUTBUF_MAX, uint8_t);
+
+struct usart_config const usart_mcdp;
+
+IN_STREAM_FROM_PRODUCER(usart_in, usart_mcdp.producer, rx_queue, NULL)
+OUT_STREAM_FROM_CONSUMER(usart_out, usart_mcdp.consumer, tx_queue, NULL)
+
+USART_CONFIG(usart_mcdp, CONFIG_MCDP28X0, 115200, rx_queue, tx_queue,
+	     usart_in.consumer, usart_out.producer);
+
 
 /**
  * Compute checksum.
@@ -73,16 +83,16 @@ static int tx_serial(const uint8_t *msg, int cnt)
 	uint8_t out = cnt + 2;
 	uint8_t chksum = compute_checksum(msg, cnt);
 
-	if (out_stream_write(&usart_mcdp.out, &out, 1) != 1)
+	if (out_stream_write(&usart_out.out, &out, 1) != 1)
 		return EC_ERROR_UNKNOWN;
 
-	if (out_stream_write(&usart_mcdp.out, msg, cnt) != cnt)
+	if (out_stream_write(&usart_out.out, msg, cnt) != cnt)
 		return EC_ERROR_UNKNOWN;
 
-	if (out_stream_write(&usart_mcdp.out, &chksum, 1) != 1)
+	if (out_stream_write(&usart_out.out, &chksum, 1) != 1)
 		return EC_ERROR_UNKNOWN;
 
-	print_buffer(usart_mcdp_tx_buffer, cnt + 2);
+	print_buffer(tx_queue_buffer, cnt + 2);
 
 	return EC_SUCCESS;
 }
@@ -99,10 +109,10 @@ static int rx_serial(uint8_t *msg, int cnt)
 	size_t read;
 	int retry = 2;
 
-	read = in_stream_read(&usart_mcdp.in, msg, cnt);
+	read = in_stream_read(&usart_in.in, msg, cnt);
 	while ((read < cnt) && retry) {
 		usleep(100*MSEC);
-		read += in_stream_read(&usart_mcdp.in, msg + read,
+		read += in_stream_read(&usart_in.in, msg + read,
 				       cnt - read);
 		retry--;
 	}