Setup the LIFX client source to get unicast responses.

This should significantly improve both latency and throughput.

CMakeLists.txt

@@ -5,7 +5,7 @@ PROJECT(LIGHTSD C)
 
 SET(CPACK_PACKAGE_VERSION_MAJOR "1")
 SET(CPACK_PACKAGE_VERSION_MINOR "1")
-SET(CPACK_PACKAGE_VERSION_PATCH "0")
+SET(CPACK_PACKAGE_VERSION_PATCH "1")
 SET(LIGHTSD_VERSION "${CPACK_PACKAGE_VERSION_MAJOR}.${CPACK_PACKAGE_VERSION_MINOR}.${CPACK_PACKAGE_VERSION_PATCH}")
 
 MESSAGE(STATUS "lightsd version: ${LIGHTSD_VERSION}")

core/daemon.c

@@ -400,6 +400,28 @@ lgtd_daemon_write_pidfile(const char *filepath)
     return written == pidlen;
 }
 
+uint32_t
+lgtd_daemon_randuint32(void)
+{
+    int fd = open("/dev/urandom", O_RDONLY);
+    if (fd == -1) {
+        lgtd_err(1, "couldn't open /dev/urandom");
+    }
+
+    uint32_t rv;
+    int nbytes = read(fd, &rv, sizeof(rv));
+    if (nbytes != sizeof(rv)) {
+        close(fd);
+        lgtd_err(
+            1, "couln't fetch %ju bytes from /dev/urandom",
+            sizeof((uintmax_t)rv)
+        );
+    }
+
+    close(fd);
+    return rv;
+}
+
 int
 lgtd_daemon_syslog_facilitytoi(const char *facility)
 {

core/daemon.h

@@ -36,6 +36,7 @@ void lgtd_daemon_set_group(const char *);
 bool lgtd_daemon_write_pidfile(const char *);
 void lgtd_daemon_drop_privileges(void);
 bool lgtd_daemon_makedirs(const char *);
+uint32_t lgtd_daemon_randuint32(void);
 
 int lgtd_daemon_syslog_facilitytoi(const char *);
 void lgtd_daemon_syslog_open(const char *, enum lgtd_verbosity, int);

core/lightsd.c

@@ -336,7 +336,7 @@ main(int argc, char *argv[], char *envp[])
 
     lgtd_daemon_die_if_running_as_root_unless_requested(lgtd_opts.user);
 
-    lgtd_lifx_wire_load_packet_info_map();
+    lgtd_lifx_wire_setup();
     if (!lgtd_lifx_discovery_setup() || !lgtd_lifx_broadcast_setup()) {
         lgtd_err(1, "can't setup lightsd");
     }

core/lightsd.h

@@ -131,6 +131,9 @@ char *lgtd_sockaddrtoa(const struct sockaddr *, char *buf, int buflen);
 char *lgtd_print_duration(uint64_t, char *, int);
 #define LGTD_PRINT_DURATION(secs, arr) \
     lgtd_print_duration((secs), (arr), sizeof((arr)))
+char* lgtd_print_nsec_timestamp(uint64_t, char *, int);
+#define LGTD_LIFX_WIRE_PRINT_NSEC_TIMESTAMP(ts, arr) \
+    lgtd_print_nsec_timestamp((ts), (arr), sizeof((arr)))
 
 void lgtd_log_setup(void);
 

core/utils.c

@@ -25,6 +25,7 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
+#include <time.h>
 
 #include "lightsd.h"
 
@@ -117,3 +118,22 @@ lgtd_print_duration(uint64_t secs, char *buf, int bufsz)
     snprintf(&buf[i], bufsz - i, "%02d:%02d", hours, minutes);
     return buf;
 }
+
+char *
+lgtd_print_nsec_timestamp(uint64_t nsec_ts, char *buf, int bufsz)
+{
+    assert(buf);
+    assert(bufsz > 0);
+
+    time_t ts = LGTD_NSECS_TO_SECS(nsec_ts);
+
+    struct tm tm_utc;
+    if (gmtime_r(&ts, &tm_utc)) {
+        int64_t usecs = LGTD_NSECS_TO_USECS(nsec_ts - LGTD_SECS_TO_NSECS(ts));
+        LGTD_TM_TO_ISOTIME(&tm_utc, buf, bufsz, usecs);
+    } else {
+        buf[0] = '\0';
+    }
+
+    return buf;
+}

lifx/wire_proto.c

@@ -27,14 +27,14 @@
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
-#include <time.h>
 
 #include <event2/util.h>
 
 #include "wire_proto.h"
 #include "core/time_monotonic.h"
 #include "bulb.h"
 #include "gateway.h"
+#include "core/daemon.h"
 #include "core/lightsd.h"
 
 const union lgtd_lifx_target LGTD_LIFX_UNSPEC_TARGET = { .tags = 0 };
@@ -50,6 +50,8 @@ const int LGTD_LIFX_DEBRUIJN_SEQUENCE[64] = {
    13, 18,  8, 12,  7,  6,  5, 63
 };
 
+static uint32_t lgtd_lifx_client_id = 0;
+
 static struct lgtd_lifx_packet_info_map lgtd_lifx_packet_info =
     RB_INITIALIZER(&lgtd_lifx_packets_infos);
 
@@ -95,7 +97,7 @@ lgtd_lifx_wire_enosys_packet_handler(struct lgtd_lifx_gateway *gw,
     );
 }
 
-void
+static void
 lgtd_lifx_wire_load_packet_info_map(void)
 {
 #define DECODER(x)  ((void (*)(void *))(x))
@@ -465,6 +467,15 @@ lgtd_lifx_wire_get_packet_info(enum lgtd_lifx_packet_type packet_type)
     return RB_FIND(lgtd_lifx_packet_info_map, &lgtd_lifx_packet_info, &pkt_info);
 }
 
+void
+lgtd_lifx_wire_setup(void)
+{
+    lgtd_lifx_wire_load_packet_info_map();
+    do {
+        lgtd_lifx_client_id = lgtd_daemon_randuint32();
+    } while (!lgtd_lifx_client_id);
+}
+
 
 #define WAVEFORM_ENTRY(e) { .str = e, .len = sizeof(e) - 1 }
 const struct lgtd_lifx_waveform_string_id lgtd_lifx_waveform_table[] = {
@@ -493,25 +504,6 @@ lgtd_lifx_wire_waveform_string_id_to_type(const char *s, int len)
     return LGTD_LIFX_WAVEFORM_INVALID;
 }
 
-char *
-lgtd_lifx_wire_print_nsec_timestamp(uint64_t nsec_ts, char *buf, int bufsz)
-{
-    assert(buf);
-    assert(bufsz > 0);
-
-    time_t ts = LGTD_NSECS_TO_SECS(nsec_ts);
-
-    struct tm tm_utc;
-    if (gmtime_r(&ts, &tm_utc)) {
-        int64_t usecs = LGTD_NSECS_TO_USECS(nsec_ts - LGTD_SECS_TO_NSECS(ts));
-        LGTD_TM_TO_ISOTIME(&tm_utc, buf, bufsz, usecs);
-    } else {
-        buf[0] = '\0';
-    }
-
-    return buf;
-}
-
 static void
 lgtd_lifx_wire_encode_header(struct lgtd_lifx_packet_header *hdr, int flags)
 {
@@ -534,6 +526,7 @@ lgtd_lifx_wire_encode_header(struct lgtd_lifx_packet_header *hdr, int flags)
     }
     hdr->at_time = htole64(hdr->at_time);
     hdr->packet_type = htole16(hdr->packet_type);
+    hdr->source = htole32(hdr->source); // not strictly necessary but eh.
 }
 
 // Convert all the fields in the header to the host endianness.
@@ -554,6 +547,7 @@ lgtd_lifx_wire_decode_header(struct lgtd_lifx_packet_header *hdr)
     }
     hdr->at_time = le64toh(hdr->at_time);
     hdr->packet_type = le16toh(hdr->packet_type);
+    hdr->source = le32toh(hdr->source);
 }
 
 const struct lgtd_lifx_packet_info *
@@ -564,13 +558,15 @@ lgtd_lifx_wire_setup_header(struct lgtd_lifx_packet_header *hdr,
                             enum lgtd_lifx_packet_type packet_type)
 {
     assert(hdr);
+    assert(lgtd_lifx_client_id);
 
     const struct lgtd_lifx_packet_info *pkt_info =
         lgtd_lifx_wire_get_packet_info(packet_type);
 
     assert(pkt_info);
 
     memset(hdr, 0, sizeof(*hdr));
+    hdr->source = lgtd_lifx_client_id;
     hdr->size = pkt_info->size + sizeof(*hdr);
     hdr->packet_type = packet_type;
     if (site) {

lifx/wire_proto.h

@@ -61,8 +61,16 @@ struct lgtd_lifx_packet_header {
     //! - tagged: true when the target field holds tags;
     //! - origin: LIFX internal use, should be 0.
     uint16le_t      protocol;
-    //! Here is what LIFXKit says about it, maybe it's related to zigbee:
-    //! Message source identifier from NAT table (Internal LIFX use)
+    //! http://lan.developer.lifx.com/v2.0/docs/header-description
+    //! The source identifier allows each client to provide an unique value,
+    //! which will be included by the LIFX device in any message that is sent
+    //! in response to a message sent by the client. If the source identifier
+    //! is a non-zero value, then the LIFX device will send a unicast message
+    //! to the source IP address and port that the client used to send the
+    //! originating message. If the source identifier is a zero value, then the
+    //! LIFX device may send a broadcast message that can be received by all
+    //! clients on the same sub-net. See _ack_required_ and _res_required_
+    //! fields in the Frame Address.
     uint32le_t      source;
     union {
         //! All targeted tags ORed together.
@@ -387,12 +395,10 @@ lgtd_lifx_wire_next_tag_id(int current_tag_id, uint64_t tags)
          (tag_id_varname) = lgtd_lifx_wire_next_tag_id((tag_id_varname), (tags)))
 
 enum lgtd_lifx_waveform_type lgtd_lifx_wire_waveform_string_id_to_type(const char *, int);
-char* lgtd_lifx_wire_print_nsec_timestamp(uint64_t, char *, int);
-#define LGTD_LIFX_WIRE_PRINT_NSEC_TIMESTAMP(ts, arr) \
-    lgtd_lifx_wire_print_nsec_timestamp((ts), (arr), sizeof((arr)))
 
 const struct lgtd_lifx_packet_info *lgtd_lifx_wire_get_packet_info(enum lgtd_lifx_packet_type);
-void lgtd_lifx_wire_load_packet_info_map(void);
+
+void lgtd_lifx_wire_setup(void);
 
 const struct lgtd_lifx_packet_info *lgtd_lifx_wire_setup_header(struct lgtd_lifx_packet_header *,
                                                                  enum lgtd_lifx_target_type,

tests/core/client/CMakeLists.txt

@@ -10,7 +10,6 @@ ADD_CORE_LIBRARY(
     ${LIGHTSD_SOURCE_DIR}/core/utils.c
     ${LIGHTSD_SOURCE_DIR}/lifx/bulb.c
     ${LIGHTSD_SOURCE_DIR}/lifx/tagging.c
-    ${LIGHTSD_SOURCE_DIR}/lifx/wire_proto.c
     ${CMAKE_CURRENT_SOURCE_DIR}/../tests_shims.c
     ${CMAKE_CURRENT_SOURCE_DIR}/../tests_utils.c
 )

tests/core/daemon/CMakeLists.txt

@@ -10,7 +10,6 @@ ADD_CORE_LIBRARY(
     ${LIGHTSD_SOURCE_DIR}/core/utils.c
     ${LIGHTSD_SOURCE_DIR}/lifx/bulb.c
     ${LIGHTSD_SOURCE_DIR}/lifx/tagging.c
-    ${LIGHTSD_SOURCE_DIR}/lifx/wire_proto.c
     ${CMAKE_CURRENT_SOURCE_DIR}/../tests_shims.c
     ${CMAKE_CURRENT_SOURCE_DIR}/../tests_utils.c
 )

tests/core/daemon/test_daemon_randuint32.c

@@ -0,0 +1,94 @@
+#include <sys/types.h>
+
+int mock_open(const char *, int, ...);
+ssize_t mock_read(int, void *, size_t);
+int mock_close(int);
+
+#define open(fp, flags, ...) mock_open(fp, flags, ##__VA_ARGS__)
+#define read(fd, buf, sz) mock_read(fd, buf, sz)
+#define close(fd) mock_close(fd)
+
+#include "daemon.c"
+
+#include "mock_gateway.h"
+#include "mock_pipe.h"
+#include "mock_router.h"
+#include "mock_log.h"
+#include "mock_timer.h"
+
+static const int MOCK_RANDOM_NUMBER = 0x72616e64;
+
+int mock_open_call_count = 0;
+
+int
+mock_open(const char *fp, int flags, ...)
+{
+    mock_open_call_count++;
+
+    if (strcmp(fp, "/dev/urandom")) {
+        errx(1, "got fp %s (expected /dev/urandom)", fp);
+    }
+
+    if (flags != O_RDONLY) {
+        errx(1, "got flags %#x (expected %#x)", flags, O_RDONLY);
+    }
+
+    return 42;
+}
+
+int mock_read_call_count = 0;
+
+ssize_t
+mock_read(int fd, void *buf, size_t nbytes)
+{
+    mock_read_call_count++;
+
+    if (fd != 42) {
+        errx(1, "got fd %d (expected 42)", fd);
+    }
+
+    if (!buf) {
+        errx(1, "missing buf");
+    }
+
+    if (nbytes != 4) {
+        errx(1, "got nbytes %ju (expected 4)", (uintmax_t)nbytes);
+    }
+
+    memcpy(buf, &MOCK_RANDOM_NUMBER, sizeof(MOCK_RANDOM_NUMBER));
+
+    return nbytes;
+}
+
+int mock_close_call_count = 0;
+
+int
+mock_close(int fd)
+{
+    mock_close_call_count++;
+
+    if (fd != 42) {
+        errx(1, "got fd %d (expected 42)", fd);
+    }
+
+    return 0;
+}
+
+int
+main(void)
+{
+    if (lgtd_daemon_randuint32() != MOCK_RANDOM_NUMBER) {
+        errx(1, "got unexpected value from randuint32");
+    }
+    if (mock_open_call_count != 1) {
+        errx(1, "open wasn't once");
+    }
+    if (mock_read_call_count != 1) {
+        errx(1, "read wasn't once");
+    }
+    if (mock_close_call_count != 1) {
+        errx(1, "close wasn't once");
+    }
+
+    return 0;
+}

tests/core/daemon/test_daemon_writepidfile.c

@@ -46,7 +46,7 @@ mock_open(const char *fp, int flags, ...)
     va_end(ap);
 
     if (strcmp(fp, "/test.pid")) {
-        lgtd_errx(1, "got fp %s (expected test.pid)", fp);
+        lgtd_errx(1, "got fp %s (expected /test.pid)", fp);
     }
 
     int expected_flags = O_CREAT|O_WRONLY|O_TRUNC;

tests/core/jsonrpc/CMakeLists.txt

@@ -8,7 +8,6 @@ ADD_CORE_LIBRARY(
     ${LIGHTSD_SOURCE_DIR}/core/jsmn.c
     ${LIGHTSD_SOURCE_DIR}/core/stats.c
     ${LIGHTSD_SOURCE_DIR}/core/utils.c
-    ${LIGHTSD_SOURCE_DIR}/lifx/wire_proto.c
     ${LIGHTSD_SOURCE_DIR}/lifx/bulb.c
     ${CMAKE_CURRENT_SOURCE_DIR}/../tests_shims.c
     ${CMAKE_CURRENT_SOURCE_DIR}/../tests_utils.c

tests/core/jsonrpc/test_jsonrpc_batch.c

@@ -5,6 +5,7 @@
 #define MOCKED_LGTD_PROTO_GET_LIGHT_STATE
 #define MOCKED_LGTD_PROTO_POWER_ON
 #include "mock_proto.h"
+#include "mock_wire_proto.h"
 #include "test_jsonrpc_utils.h"
 
 static int power_on_call_count = 0;

tests/core/jsonrpc/test_jsonrpc_batch_notifications_only.c

@@ -5,6 +5,7 @@
 #define MOCKED_LGTD_PROTO_GET_LIGHT_STATE
 #define MOCKED_LGTD_PROTO_POWER_ON
 #include "mock_proto.h"
+#include "mock_wire_proto.h"
 #include "test_jsonrpc_utils.h"
 
 static int power_on_call_count = 0;

tests/core/jsonrpc/test_jsonrpc_batch_one_invalid_request.c

@@ -4,6 +4,7 @@
 #include "mock_log.h"
 #define MOCKED_LGTD_PROTO_POWER_ON
 #include "mock_proto.h"
+#include "mock_wire_proto.h"
 #include "test_jsonrpc_utils.h"
 
 static int power_on_call_count = 0;

tests/core/jsonrpc/test_jsonrpc_batch_one_notification.c

@@ -5,6 +5,7 @@
 #define MOCKED_LGTD_PROTO_GET_LIGHT_STATE
 #define MOCKED_LGTD_PROTO_POWER_ON
 #include "mock_proto.h"
+#include "mock_wire_proto.h"
 #include "test_jsonrpc_utils.h"
 
 static int power_on_call_count = 0;