common: Added a small locking mechanism to simplify how we lock IO

We've done this a number of times already where we're getting
exclusive access to either the out direction of a connection, or we
try to lock out the read side while we are responding to a previous
request. They usually are really cumbersome because we reach around to
the other direction to stop it from proceeding, or we flag our
exclusive access somewhere, and we always need to know whom to notify.

PR ElementsProject/lightning#1970 adds two new instances of this:

 - Streaming a JSON response requires that nothing else should write
   while the stream is active.
 - We also want to stop reading new requests while we are responding
   to one.

To remove the complexity of having to know whom to stop and notify
when we're done, this adds a simple `io_lock` primitive that can be
used to get exclusive access to a connection. This inverts the
requirement for notifications, since everybody registers interest in
the lock and they get notified if the lock holder releases it.

common/Makefile

@@ -23,6 +23,7 @@ COMMON_SRC_NOGEN :=				\
 	common/htlc_wire.c			\
 	common/initial_channel.c		\
 	common/initial_commit_tx.c		\
+	common/io_lock.c			\
 	common/json.c				\
 	common/json_escaped.c			\
 	common/key_derive.c			\

common/io_lock.c

@@ -0,0 +1,82 @@
+#include "io_lock.h"
+#include <assert.h>
+#include <ccan/io/io_plan.h>
+
+struct io_lock {
+	bool locked;
+};
+
+/* Struct to hold information while we wait for the lock to be freed */
+struct io_lock_waiter {
+	struct io_plan *(*next)(struct io_conn *conn, void *next_arg);
+	void *arg;
+	struct io_lock *lock;
+	enum io_direction dir;
+};
+
+struct io_lock *io_lock_new(const tal_t *ctx)
+{
+	struct io_lock *lock = tal(ctx, struct io_lock);
+	lock->locked = false;
+	return lock;
+}
+
+static struct io_plan *io_lock_try_acquire(struct io_conn *conn,
+					   struct io_lock_waiter *waiter)
+{
+	/* Destructure waiter, since we might be freeing it below */
+	struct io_plan *(*next)(struct io_conn *, void *) = waiter->next;
+	void *next_arg = waiter->arg;
+
+	if (!waiter->lock->locked) {
+		waiter->lock->locked = true;
+		tal_free(waiter);
+		return next(conn, next_arg);
+	} else {
+		switch (waiter->dir) {
+		case IO_IN:
+			return io_wait(conn, waiter->lock, io_lock_try_acquire,
+				       waiter);
+		case IO_OUT:
+			return io_out_wait(conn, waiter->lock,
+					   io_lock_try_acquire, waiter);
+		}
+		/* Should not happen if waiter->dir is a valid enum
+		 * value */
+		abort();
+	}
+}
+
+static struct io_plan *io_lock_acquire_dir(
+    struct io_conn *conn, struct io_lock *lock, enum io_direction dir,
+    struct io_plan *(*next)(struct io_conn *, void *), void *arg)
+{
+	/* FIXME: We can avoid one allocation if we lock and call next here directly */
+	struct io_lock_waiter *waiter = tal(lock, struct io_lock_waiter);
+	waiter->next = next;
+	waiter->arg = arg;
+	waiter->lock = lock;
+	waiter->dir = dir;
+	return io_lock_try_acquire(conn, waiter);
+}
+
+struct io_plan *
+io_lock_acquire_out_(struct io_conn *conn, struct io_lock *lock,
+		 struct io_plan *(*next)(struct io_conn *, void *), void *arg)
+{
+	return io_lock_acquire_dir(conn, lock, IO_OUT, next, arg);
+}
+
+struct io_plan *
+io_lock_acquire_in_(struct io_conn *conn, struct io_lock *lock,
+		    struct io_plan *(*next)(struct io_conn *, void *), void *arg)
+{
+	return io_lock_acquire_dir(conn, lock, IO_IN, next, arg);
+}
+
+void io_lock_release(struct io_lock *lock)
+{
+	assert(lock->locked);
+	lock->locked = false;
+	io_wake(lock);
+}

common/io_lock.h

@@ -0,0 +1,50 @@
+#ifndef LIGHTNING_COMMON_IO_LOCK_H
+#define LIGHTNING_COMMON_IO_LOCK_H
+
+#include "config.h"
+#include <ccan/io/io.h>
+#include <ccan/tal/tal.h>
+struct io_lock;
+
+/**
+ * Create a new lock
+ */
+struct io_lock *io_lock_new(const tal_t *ctx);
+
+/**
+ * Acquire lock @lock before proceeding to @next
+ *
+ * Attempts to acquire the lock before proceeding with next. If the
+ * lock is free this reduces to `io_always`, otherwise we put @conn in
+ * wait until we get notified about the lock being released.
+ */
+#define io_lock_acquire_out(conn, lock, next, arg)                             \
+	io_lock_acquire_out_((conn), (lock),                                   \
+			     typesafe_cb_preargs(struct io_plan *, void *,     \
+						 (next), (arg),                \
+						 struct io_conn *),            \
+			     (arg))
+
+struct io_plan *io_lock_acquire_out_(struct io_conn *conn, struct io_lock *lock,
+				     struct io_plan *(*next)(struct io_conn *,
+							     void *),
+				     void *arg);
+
+#define io_lock_acquire_in(conn, lock, next, arg)                             \
+	io_lock_acquire_in_((conn), (lock),                                   \
+			    typesafe_cb_preargs(struct io_plan *, void *, \
+						(next), (arg),		\
+						struct io_conn *),	\
+			    (arg))
+
+struct io_plan *io_lock_acquire_in_(struct io_conn *conn, struct io_lock *lock,
+				    struct io_plan *(*next)(struct io_conn *,
+							    void *),
+				    void *arg);
+
+/**
+ * Release the lock and notify waiters so they can proceed.
+ */
+void io_lock_release(struct io_lock *lock);
+
+#endif /* LIGHTNING_COMMON_IO_LOCK_H */