--- a/fix_command_pipe_read_loop.patch	Sat Aug 15 15:39:21 2015 -0700
+++ b/fix_command_pipe_read_loop.patch	Sat Aug 15 16:21:19 2015 -0700
@@ -1,5 +1,14 @@
 # HG changeset patch
-# Parent  61d444c8e81f6c00f8347fd2d3fa9cbc9f7faef4
+# Parent  870e94c0386d9595949a21c3486875759c4a1ff6
+Fix infinite loop in the command pipe read callback
+
+Looks like I suck at Unix, EOF & EAGAIN weren't handled properly. The
+pipe is now re-opened on EOF.
+
+We could probably make the command pipe bi-directional at this point.
+
+I wish conky wasn't fucked nowadays, that would have let me spot this
+earlier.
 
 diff --git a/core/pipe.c b/core/pipe.c
 --- a/core/pipe.c
@@ -57,10 +66,10 @@
 -                lgtd_command_pipe_close(pipe);
 -                lgtd_command_pipe_open(path);
 -                return;
-+                break;  // go back to the event loop
++                break;
              }
 -            continue;
-+            return;
++            return; // EAGAIN, go back to the event loop
          }
  
          if (!drain) {
@@ -123,7 +132,7 @@
      SLIST_INSERT_HEAD(&lgtd_command_pipes, pipe, link);
  
      return true;
-@@ -239,6 +237,26 @@
+@@ -239,6 +237,27 @@
      return false;
  }
  
@@ -131,6 +140,7 @@
 +lgtd_command_pipe_reset(struct lgtd_command_pipe *pipe)
 +{
 +    const char *path = pipe->path;
++    // we could optimize a bit to avoid re-allocations here:
 +    _lgtd_command_pipe_close(pipe);
 +    if (!_lgtd_command_pipe_open(path)) {
 +        lgtd_warn("can't re-open pipe %s", path);
@@ -150,3 +160,330 @@
  void
  lgtd_command_pipe_close_all(void)
  {
+diff --git a/tests/core/pipe/test_pipe_read_callback.c b/tests/core/pipe/test_pipe_read_callback.c
+--- a/tests/core/pipe/test_pipe_read_callback.c
++++ b/tests/core/pipe/test_pipe_read_callback.c
+@@ -7,6 +7,7 @@
+ #include "lifx/wire_proto.h"
+ 
+ #define MOCKED_EVENT_NEW
++#define MOCKED_EVENT_DEL
+ #define MOCKED_EVBUFFER_NEW
+ #define MOCKED_EVBUFFER_READ
+ #define MOCKED_EVBUFFER_PULLUP
+@@ -69,6 +70,16 @@
+     return (void *)1;
+ }
+ 
++static int event_del_call_count = 0;
++
++int
++event_del(struct event *ev)
++{
++    (void)ev;
++    event_del_call_count++;
++    return 0;
++}
++
+ static int
+ get_nbytes_read(int call_count)
+ {
+@@ -186,8 +197,22 @@
+     }
+ 
+     struct lgtd_command_pipe *pipe = SLIST_FIRST(&lgtd_command_pipes);
++    lgtd_command_pipe_read_callback(pipe->fd, EV_READ, pipe);
++    if (event_del_call_count != 1) {
++        errx(1, "the pipe wasn't reset");
++    }
+ 
++    jsonrpc_dispatch_request_call_count = 0;
++    evbuffer_drain_call_count = 0;
++    evbuffer_read_call_count = 0;
++    evbuffer_pullup_call_count = 0;
++    evbuffer_get_length_call_count = 0;
++    event_del_call_count = 0;
++    pipe = SLIST_FIRST(&lgtd_command_pipes);
+     lgtd_command_pipe_read_callback(pipe->fd, EV_READ, pipe);
++    if (event_del_call_count != 1) {
++        errx(1, "the pipe wasn't reset");
++    }
+ 
+     return 0;
+ }
+diff --git a/tests/core/pipe/test_pipe_read_callback_yield_on_eagain.c b/tests/core/pipe/test_pipe_read_callback_yield_on_eagain.c
+new file mode 100644
+--- /dev/null
++++ b/tests/core/pipe/test_pipe_read_callback_yield_on_eagain.c
+@@ -0,0 +1,271 @@
++#include "pipe.c"
++
++#include <sys/tree.h>
++#include <endian.h>
++#include <limits.h>
++
++#include "lifx/wire_proto.h"
++
++#define MOCKED_EVENT_NEW
++#define MOCKED_EVBUFFER_NEW
++#define MOCKED_EVBUFFER_READ
++#define MOCKED_EVBUFFER_PULLUP
++#define MOCKED_EVBUFFER_GET_LENGTH
++#define MOCKED_EVBUFFER_DRAIN
++#include "mock_event2.h"
++#include "mock_gateway.h"
++#include "mock_router.h"
++#include "mock_timer.h"
++
++#include "tests_utils.h"
++#define MOCKED_JSONRPC_DISPATCH_REQUEST
++#include "tests_pipe_utils.h"
++
++#define REQUEST_1 "{"                   \
++    "\"jsonrpc\": \"2.0\","             \
++    "\"method\": \"get_light_state\","  \
++    "\"params\": [\"*\"],"              \
++    "\"id\": 42"                        \
++"}"
++
++#define REQUEST_2 "{"           \
++    "\"jsonrpc\": \"2.0\","     \
++    "\"method\": \"power_on\"," \
++    "\"params\": [\"*\"],"      \
++    "\"id\": 43"                \
++"}"
++
++static unsigned char request[] = (
++    REQUEST_1
++    REQUEST_2
++);
++
++static char *tmpdir = NULL;
++
++void
++cleanup_tmpdir(void)
++{
++    lgtd_tests_remove_temp_dir(tmpdir);
++}
++
++static int jsonrpc_dispatch_request_call_count = 0;
++
++void
++lgtd_jsonrpc_dispatch_request(struct lgtd_client *client, int parsed)
++{
++    (void)client;
++    (void)parsed;
++
++    if (!parsed) {
++        errx(1, "number of parsed json tokens not passed in");
++    }
++
++    switch (jsonrpc_dispatch_request_call_count) {
++    case 0:
++        if (memcmp(client->json, request, sizeof(request) - 1)) {
++            errx(
++                1, "got unexpected json %s (expected %s)",
++                client->json, request
++            );
++        }
++        break;
++    case 1:
++        if (memcmp(client->json, REQUEST_2, sizeof(REQUEST_2) - 1)) {
++            errx(
++                1, "got unexpected json %s (expected %s)",
++                client->json, REQUEST_2
++            );
++        }
++        break;
++    default:
++        errx(
++            1, "jsonrpc_dispatch_request_call_count = %d",
++            jsonrpc_dispatch_request_call_count
++        );
++        break;
++    }
++
++    jsonrpc_dispatch_request_call_count++;
++}
++
++struct event *
++event_new(struct event_base *base,
++          evutil_socket_t fd,
++          short events,
++          event_callback_fn cb,
++          void *ctx)
++{
++    (void)base;
++    (void)fd;
++    (void)events;
++    (void)cb;
++    (void)ctx;
++
++    return (void *)1;
++}
++
++struct evbuffer *
++evbuffer_new(void)
++{
++    return (void *)2;
++}
++
++static int evbuffer_drain_call_count = 0;
++
++int
++evbuffer_drain(struct evbuffer *buf, size_t len)
++{
++    if (buf != (void *)2) {
++        errx(1, "got unexpected buf %p (expected %p)", buf, (void *)2);
++    }
++
++    jsmn_parser jsmn_ctx;
++    jsmn_init(&jsmn_ctx);
++    struct lgtd_command_pipe *pipe = SLIST_FIRST(&lgtd_command_pipes);
++    if (memcmp(&pipe->client.jsmn_ctx, &jsmn_ctx, sizeof(jsmn_ctx))) {
++        errx(1, "the client json parser context wasn't re-initialized");
++    }
++
++    switch (evbuffer_drain_call_count) {
++    case 0:
++        if (len != sizeof(REQUEST_1) - 1) {
++            errx(
++                1, "trying to drain %ju bytes (expected %ju)",
++                (uintmax_t)len, (uintmax_t)sizeof(REQUEST_1) - 1
++            );
++        }
++        break;
++    case 1:
++        if (len != sizeof(REQUEST_2) - 1) {
++            errx(
++                1, "trying to drain %ju bytes (expected %ju)",
++                (uintmax_t)len, (uintmax_t)sizeof(REQUEST_2) - 1
++            );
++        }
++        break;
++    default:
++        errx(1, "evbuffer_drain_call_count = %d", evbuffer_drain_call_count);
++        break;
++    }
++    evbuffer_drain_call_count++;
++
++    return 0;
++}
++
++static int evbuffer_pullup_call_count = 0;
++
++unsigned char *
++evbuffer_pullup(struct evbuffer *buf, ev_ssize_t size)
++{
++    if (buf != (void *)2) {
++        errx(1, "got unexpected buf %p (expected %p)", buf, (void *)2);
++    }
++
++    if (size != -1) {
++        errx(
++            1, "got unexpected size %jd in pullup (expected -1)", (intmax_t)size
++        );
++    }
++
++    int offset;
++    switch (evbuffer_pullup_call_count) {
++    case 0:
++        offset = 0;
++        break;
++    case 1:
++        offset = sizeof(REQUEST_1) - 1;
++        break;
++    default:
++        offset = sizeof(request);
++        break;
++    }
++    evbuffer_pullup_call_count++;
++
++    return &request[offset];
++}
++
++static int evbuffer_get_length_call_count = 0;
++
++size_t
++evbuffer_get_length(const struct evbuffer *buf)
++{
++    if (buf != (void *)2) {
++        errx(1, "got unexpected buf %p (expected %p)", buf, (void *)2);
++    }
++
++    size_t len;
++    switch (evbuffer_get_length_call_count) {
++    case 0:
++        len = sizeof(REQUEST_1) - 1;
++        break;
++    case 1:
++        len = sizeof(request) - sizeof(REQUEST_1);
++        break;
++    default:
++        len = 0;
++        break;
++    }
++    evbuffer_get_length_call_count++;
++
++    return len;
++}
++
++static int evbuffer_read_call_count = 0;
++
++int
++evbuffer_read(struct evbuffer *buf, evutil_socket_t fd, int howmuch)
++{
++    if (buf != (void *)2) {
++        errx(1, "got unexpected buf %p (expected %p)", buf, (void *)2);
++    }
++
++    struct lgtd_command_pipe *pipe = SLIST_FIRST(&lgtd_command_pipes);
++    if (fd != pipe->fd) {
++        errx(1, "got unexpected fd %d (expected %d)", fd, pipe->fd);
++    }
++
++    if (howmuch != -1) {
++        errx(
++            1, "got unexpected howmuch bytes to read %d (expected -1)", howmuch
++        );
++    }
++
++    int rv = 0;
++    switch (evbuffer_read_call_count) {
++    case 0:
++        rv = sizeof(REQUEST_1) - 1;
++        break;
++    case 1:
++        rv = -1;
++        errno = EAGAIN;
++        break;
++    case 2:
++        rv = sizeof(request) - sizeof(REQUEST_1);
++        break;
++    default:
++        break;
++    }
++    evbuffer_read_call_count++;
++
++    return rv;
++}
++
++int
++main(void)
++{
++    tmpdir = lgtd_tests_make_temp_dir();
++    atexit(cleanup_tmpdir);
++
++    char path[PATH_MAX] = { 0 };
++    snprintf(path, sizeof(path), "%s/lightsd.pipe", tmpdir);
++    if (!lgtd_command_pipe_open(path)) {
++        errx(1, "couldn't open pipe");
++    }
++
++    struct lgtd_command_pipe *pipe = SLIST_FIRST(&lgtd_command_pipes);
++
++    lgtd_command_pipe_read_callback(pipe->fd, EV_READ, pipe);
++    lgtd_command_pipe_read_callback(pipe->fd, EV_READ, pipe);
++
++    return 0;
++}