Producer/Consumer: Refactor to use Queue policies

Previously the Producer and Consumer interfaces tracked the
Consumer and Producer respectively at the other end of the
queue that they interacted with.  This was done to avoid
modifying the queue implementation, but resulted in a rougher
interface that required additional initialization steps and
prevented alternative configurations; many producers and one
consumer for example.

This commit uses the new queue policies to track this
information.  The new direct policy behaves as the old producer
and consumers did.  Now the producers and consumers are just
named references to the queue that they work on and a convenient
location for a notification callback when the queue is updated in
a way that is relevent to the producer or consumer.

All users of Producer and Consumer have been updated including the
stream adaptors which are in use by the echo test code and the
mcdp28x0 driver.  Use of the stream adaptors has also been
simplified.

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

BRANCH=None
BUG=None
TEST=make buildall -j
     Manual testing of Ryu (P5) and discovery board echo task

Change-Id: I704be6378a31b4e20f5063295eff9943e4900409
Reviewed-on: https://chromium-review.googlesource.com/271792
Reviewed-by: Anton Staaf <robotboy@chromium.org>
Commit-Queue: Anton Staaf <robotboy@chromium.org>
Trybot-Ready: Anton Staaf <robotboy@chromium.org>
Tested-by: Anton Staaf <robotboy@chromium.org>

5 files changed, 64 insertions, 110 deletions

diff --git a/common/build.mk b/common/build.mk
index 395844c527..ee6c3975e2 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 producer.o consumer.o
+common-y+=version.o printf.o queue.o queue_policies.o
 
 common-$(CONFIG_ADC)+=adc.o
 common-$(CONFIG_ALS)+=als.o
diff --git a/common/consumer.c b/common/consumer.c
deleted file mode 100644
index 3d424479e4..0000000000
--- a/common/consumer.c
+++ /dev/null
@@ -1,40 +0,0 @@
-/* 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
deleted file mode 100644
index 855eb4a651..0000000000
--- a/common/producer.c
+++ /dev/null
@@ -1,40 +0,0 @@
-/* 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/queue_policies.c b/common/queue_policies.c
new file mode 100644
index 0000000000..130dee52f0
--- /dev/null
+++ b/common/queue_policies.c
@@ -0,0 +1,28 @@
+/* Copyright 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.
+ *
+ * Queue policies.
+ */
+#include "queue_policies.h"
+#include "util.h"
+
+#include <stddef.h>
+
+void queue_add_direct(struct queue_policy const *policy, size_t count)
+{
+	struct queue_policy_direct const *direct =
+		DOWNCAST(policy, struct queue_policy_direct, policy);
+
+	if (count && direct->consumer->ops->written)
+		direct->consumer->ops->written(direct->consumer, count);
+}
+
+void queue_remove_direct(struct queue_policy const *policy, size_t count)
+{
+	struct queue_policy_direct const *direct =
+		DOWNCAST(policy, struct queue_policy_direct, policy);
+
+	if (count && direct->producer->ops->read)
+		direct->producer->ops->read(direct->producer, count);
+}
diff --git a/common/stream_adaptor.c b/common/stream_adaptor.c
index b8ede28e30..8a8a414f55 100644
--- a/common/stream_adaptor.c
+++ b/common/stream_adaptor.c
@@ -1,4 +1,4 @@
-/* Copyright (c) 2015 The Chromium OS Authors. All rights reserved.
+/* Copyright 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.
  *
@@ -10,56 +10,62 @@
 #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)
+static size_t in_stream_from_queue_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);
+	struct in_stream_from_queue const *adaptor =
+		DOWNCAST(stream, struct in_stream_from_queue, in);
 
-	return consumer_read_memcpy(&adaptor->consumer, buffer, count, memcpy);
+	return queue_remove_memcpy(adaptor->consumer.queue,
+				   buffer,
+				   count,
+				   memcpy);
 }
 
-static void in_stream_from_producer_written(struct consumer const *consumer,
-					    size_t count)
+static void in_stream_from_queue_written(struct consumer const *consumer,
+					 size_t count)
 {
-	struct in_stream_from_producer const *adaptor =
-		DOWNCAST(consumer, struct in_stream_from_producer, consumer);
+	struct in_stream_from_queue const *adaptor =
+		DOWNCAST(consumer, struct in_stream_from_queue, 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 in_stream_ops const in_stream_from_queue_in_stream_ops = {
+	.read = in_stream_from_queue_read,
 };
 
-struct consumer_ops const in_stream_from_producer_consumer_ops = {
-	.written = in_stream_from_producer_written,
+struct consumer_ops const in_stream_from_queue_consumer_ops = {
+	.written = in_stream_from_queue_written,
 };
 
-static size_t out_stream_from_consumer_write(struct out_stream const *stream,
-					     uint8_t const *buffer,
-					     size_t count)
+static size_t out_stream_from_queue_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);
+	struct out_stream_from_queue const *adaptor =
+		DOWNCAST(stream, struct out_stream_from_queue, out);
 
-	return producer_write_memcpy(&adaptor->producer, buffer, count, memcpy);
+	return queue_add_memcpy(adaptor->producer.queue,
+				buffer,
+				count,
+				memcpy);
 }
 
-static void out_stream_from_consumer_read(struct producer const *producer,
-					     size_t count)
+static void out_stream_from_queue_read(struct producer const *producer,
+				       size_t count)
 {
-	struct out_stream_from_consumer const *adaptor =
-		DOWNCAST(producer, struct out_stream_from_consumer, producer);
+	struct out_stream_from_queue const *adaptor =
+		DOWNCAST(producer, struct out_stream_from_queue, 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 out_stream_ops const out_stream_from_queue_out_stream_ops = {
+	.write = out_stream_from_queue_write,
 };
 
-struct producer_ops const out_stream_from_consumer_producer_ops = {
-	.read = out_stream_from_consumer_read,
+struct producer_ops const out_stream_from_queue_producer_ops = {
+	.read = out_stream_from_queue_read,
 };