Streams: Define a lower level abstraction for queue access

This adds a new pair of interfaces called producer and consumer
which are simpler (in that they don't hold onto the underlying
queue) than the stream interfaces.  This makes it easier to
share a single queue between the endpoints that will manipulate
it.  It was not possible to share a queue between two objects
that implemented the in_stream and out_stream interfaces.

This also adds a pair of adaptors that can convert a producer
or consumer into a stream of the correct type.  These adaptors
will be used for existing code once the usb-stream and usart
drivers are converted over to use the producer/consumer
interfaces instead of the stream interfaces.

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

BRANCH=None
BUG=None
TEST=make buildall -j

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

4 files changed, 147 insertions, 2 deletions

diff --git a/common/build.mk b/common/build.mk
index a6cb7bad8a..f713ea1b5d 100644
--- a/common/build.mk
+++ b/common/build.mk
@@ -7,7 +7,7 @@
 #
 
 common-y=util.o
-common-y+=version.o printf.o queue.o
+common-y+=version.o printf.o queue.o producer.o consumer.o
 
 common-$(CONFIG_ADC)+=adc.o
 common-$(CONFIG_ALS)+=als.o
@@ -75,7 +75,7 @@ common-$(CONFIG_SHA256)+=sha256.o
 common-$(CONFIG_SMBUS)+= smbus.o
 common-$(CONFIG_SOFTWARE_CLZ)+=clz.o
 common-$(CONFIG_SPI_FLASH)+=spi_flash.o
-common-$(CONFIG_STREAM)+=in_stream.o out_stream.o
+common-$(CONFIG_STREAM)+=in_stream.o out_stream.o stream_adaptor.o
 common-$(CONFIG_SWITCH)+=switch.o
 common-$(CONFIG_SW_CRC)+=crc.o
 common-$(CONFIG_TEMP_SENSOR)+=temp_sensor.o thermal.o throttle_ap.o
diff --git a/common/consumer.c b/common/consumer.c
new file mode 100644
index 0000000000..3d424479e4
--- /dev/null
+++ b/common/consumer.c
@@ -0,0 +1,40 @@
+/* Copyright (c) 2015 The Chromium OS Authors. All rights reserved.
+ * Use of this source code is governed by a BSD-style license that can be
+ * found in the LICENSE file.
+ *
+ * Consumer methods
+ */
+#include "consumer.h"
+#include "producer.h"
+
+void consumer_notify_directly(struct consumer const *consumer, size_t count)
+{
+	if (count && consumer->ops->written)
+		consumer->ops->written(consumer, count);
+}
+
+size_t consumer_read_unit(struct consumer const *consumer, void *unit)
+{
+	size_t removed = queue_remove_unit(consumer->queue, unit);
+
+	producer_notify_directly(consumer->producer, removed);
+
+	return removed;
+}
+
+size_t consumer_read_memcpy(struct consumer const *consumer,
+			    void *units,
+			    size_t count,
+			    void *(*memcpy)(void *dest,
+					    void const *src,
+					    size_t n))
+{
+	size_t removed = queue_remove_memcpy(consumer->queue,
+					     units,
+					     count,
+					     memcpy);
+
+	producer_notify_directly(consumer->producer, removed);
+
+	return removed;
+}
diff --git a/common/producer.c b/common/producer.c
new file mode 100644
index 0000000000..855eb4a651
--- /dev/null
+++ b/common/producer.c
@@ -0,0 +1,40 @@
+/* Copyright (c) 2015 The Chromium OS Authors. All rights reserved.
+ * Use of this source code is governed by a BSD-style license that can be
+ * found in the LICENSE file.
+ *
+ * Producer methods
+ */
+#include "consumer.h"
+#include "producer.h"
+
+void producer_notify_directly(struct producer const *producer, size_t count)
+{
+	if (count && producer->ops->read)
+		producer->ops->read(producer, count);
+}
+
+size_t producer_write_unit(struct producer const *producer, void const *unit)
+{
+	size_t added = queue_add_unit(producer->queue, unit);
+
+	consumer_notify_directly(producer->consumer, added);
+
+	return added;
+}
+
+size_t producer_write_memcpy(struct producer const *producer,
+			     void const *units,
+			     size_t count,
+			     void *(*memcpy)(void *dest,
+					     void const *src,
+					     size_t n))
+{
+	size_t added = queue_add_memcpy(producer->queue,
+					units,
+					count,
+					memcpy);
+
+	consumer_notify_directly(producer->consumer, added);
+
+	return added;
+}
diff --git a/common/stream_adaptor.c b/common/stream_adaptor.c
new file mode 100644
index 0000000000..b8ede28e30
--- /dev/null
+++ b/common/stream_adaptor.c
@@ -0,0 +1,65 @@
+/* Copyright (c) 2015 The Chromium OS Authors. All rights reserved.
+ * Use of this source code is governed by a BSD-style license that can be
+ * found in the LICENSE file.
+ *
+ * Stream adaptor implementation.
+ */
+
+#include "producer.h"
+#include "consumer.h"
+#include "stream_adaptor.h"
+#include "util.h"
+
+static size_t in_stream_from_producer_read(struct in_stream const *stream,
+					   uint8_t *buffer,
+					   size_t count)
+{
+	struct in_stream_from_producer const *adaptor =
+		DOWNCAST(stream, struct in_stream_from_producer, in);
+
+	return consumer_read_memcpy(&adaptor->consumer, buffer, count, memcpy);
+}
+
+static void in_stream_from_producer_written(struct consumer const *consumer,
+					    size_t count)
+{
+	struct in_stream_from_producer const *adaptor =
+		DOWNCAST(consumer, struct in_stream_from_producer, consumer);
+
+	in_stream_ready(&adaptor->in);
+}
+
+struct in_stream_ops const in_stream_from_producer_in_stream_ops = {
+	.read = in_stream_from_producer_read,
+};
+
+struct consumer_ops const in_stream_from_producer_consumer_ops = {
+	.written = in_stream_from_producer_written,
+};
+
+static size_t out_stream_from_consumer_write(struct out_stream const *stream,
+					     uint8_t const *buffer,
+					     size_t count)
+{
+	struct out_stream_from_consumer const *adaptor =
+		DOWNCAST(stream, struct out_stream_from_consumer, out);
+
+	return producer_write_memcpy(&adaptor->producer, buffer, count, memcpy);
+}
+
+static void out_stream_from_consumer_read(struct producer const *producer,
+					     size_t count)
+{
+	struct out_stream_from_consumer const *adaptor =
+		DOWNCAST(producer, struct out_stream_from_consumer, producer);
+
+	out_stream_ready(&adaptor->out);
+}
+
+struct out_stream_ops const out_stream_from_consumer_out_stream_ops = {
+	.write = out_stream_from_consumer_write,
+};
+
+struct producer_ops const out_stream_from_consumer_producer_ops = {
+	.read = out_stream_from_consumer_read,
+};