coap_io.c: Add in timeout support

CMakeLists.txt

@@ -284,6 +284,7 @@ check_include_file(sys/stat.h HAVE_SYS_STAT_H)
 check_include_file(sys/time.h HAVE_SYS_TIME_H)
 check_include_file(sys/types.h HAVE_SYS_TYPES_H)
 check_include_file(sys/unistd.h HAVE_SYS_UNISTD_H)
+check_include_file(fcntl.h HAVE_FCNTL_H)
 check_include_file(time.h HAVE_TIME_H)
 check_include_file(unistd.h HAVE_UNISTD_H)
 check_include_file(float.h HAVE_FLOAT_H)

cmake_coap_config.h.in

@@ -142,6 +142,9 @@
 /* Define to 1 if you have the <sys/unistd.h> header file. */
 #cmakedefine HAVE_SYS_UNISTD_H @HAVE_SYS_UNISTD_H@
 
+/* Define to 1 if you have the <fcntl.h> header file. */
+#cmakedefine HAVE_FCNTL_H @HAVE_FCNTL_H@
+
 /* Define to 1 if you have the <time.h> header file. */
 #cmakedefine HAVE_TIME_H @HAVE_TIME_H@
 

coap_config.h.riot

@@ -349,6 +349,9 @@
 /* Define to 1 if you have the <sys/unistd.h> header file. */
 #define HAVE_SYS_UNISTD_H 1
 
+/* Define to 1 if you have the <fcntl.h> header file. */
+#define HAVE_FCNTL_H 1
+
 /* Define to 1 if you have the <time.h> header file. */
 #define HAVE_TIME_H 1
 

configure.ac

@@ -1102,7 +1102,7 @@ fi
 
 # Checks for header files.
 AC_CHECK_HEADERS([assert.h arpa/inet.h limits.h netdb.h netinet/in.h \
-                  pthread.h errno.h winsock2.h ws2tcpip.h \
+                  pthread.h errno.h winsock2.h ws2tcpip.h fcntl.h \
                   stdlib.h string.h strings.h sys/socket.h sys/time.h \
                   time.h unistd.h sys/unistd.h sys/ioctl.h net/if.h ifaddrs.h])
 

include/coap3/coap_net_internal.h

@@ -178,6 +178,7 @@ struct coap_context_t {
   int epfd;                        /**< External FD for epoll */
   int eptimerfd;                   /**< Internal FD for timeout */
 #else /* ! COAP_EPOLL_SUPPORT */
+  int pipefd[2];                   /**< Used to speed up select() timeouts */
 #if !defined(RIOT_VERSION) && !defined(WITH_CONTIKI)
   fd_set readfds, writefds, exceptfds; /**< Used for select call
                                             in coap_io_process_with_fds_lkd() */

include/coap3/coap_subscribe.h

@@ -63,6 +63,15 @@ void coap_resource_set_get_observable(coap_resource_t *resource, int mode);
 COAP_API int coap_resource_notify_observers(coap_resource_t *resource,
                                             const coap_string_t *query);
 
+/**
+ * Check whether resource has any subscribers.
+ *
+ * @param resource The CoAP resource to check.
+ *
+ * @return         @c 1 if there are subscribers, @c 0 otherwise.
+ */
+int coap_resource_has_observers(coap_resource_t *resource);
+
 /**
  * Checks all known resources to see if they are dirty and then notifies
  * subscribed observers.

libcoap-3.map

@@ -223,6 +223,7 @@ global:
   coap_resolve_address_info;
   coap_resource_get_uri_path;
   coap_resource_get_userdata;
+  coap_resource_has_observers;
   coap_resource_init;
   coap_resource_notify_observers;
   coap_resource_proxy_uri_init2;

libcoap-3.sym

@@ -221,6 +221,7 @@ coap_resize_binary
 coap_resolve_address_info
 coap_resource_get_uri_path
 coap_resource_get_userdata
+coap_resource_has_observers
 coap_resource_init
 coap_resource_notify_observers
 coap_resource_proxy_uri_init

src/coap_io.c

@@ -538,6 +538,20 @@ coap_update_io_timer(coap_context_t *context, coap_tick_t delay) {
   coap_tick_t now;
 
   coap_ticks(&now);
+  if (context->pipefd[0] != -1) {
+    if (delay == 0 || (context->next_timeout && context->next_timeout > now + delay)) {
+      /* No delay, or context->next_timeout needs reducing */
+      char byte = 0;
+
+#ifdef _WIN32
+#include <io.h>
+#define write(a,b,c) _write(a,b,c)
+#endif
+      if (write(context->pipefd[1], &byte, 1) == -1) {
+        /* Ignore error - need to test write(function) */
+      }
+    }
+  }
   if (context->next_timeout == 0 || context->next_timeout > now + delay) {
     context->next_timeout = now + delay;
   }
@@ -1644,6 +1658,12 @@ coap_io_get_fds_lkd(coap_context_t *ctx,
   coap_session_t *s, *rtmp;
   coap_tick_t now;
   unsigned int timeout_ms;
+
+  if (ctx->pipefd[0] != -1) {
+    if (!coap_add_fd(ctx->pipefd[0], read_fds, have_read_fds, max_read_fds))
+      return 0;
+  }
+
 #if COAP_SERVER_SUPPORT
   coap_endpoint_t *ep;
 
@@ -1687,6 +1707,26 @@ coap_io_get_fds_lkd(coap_context_t *ctx,
                               ctx->next_timeout - now : 0 : 0) *
                1000 / COAP_TICKS_PER_SECOND;
   *rem_timeout_ms = timeout_ms;
+
+#ifdef COAP_DEBUG_WAKEUP_TIMES
+  char buff[1023];
+  int len;
+  unsigned int i;
+
+  snprintf(buff, sizeof(buff), "timeout_ms: %u read:", timeout_ms);
+  for (i = 0 ; i < *have_read_fds; i++) {
+    len = strlen(buff);
+    snprintf(&buff[len], sizeof(buff) - len, " %d", read_fds[i]);
+  }
+  len = strlen(buff);
+  snprintf(&buff[len], sizeof(buff) - len, " write:");
+  for (i = 0 ; i < *have_write_fds; i++) {
+    len = strlen(buff);
+    snprintf(&buff[len], sizeof(buff) - len, " %d", write_fds[i]);
+  }
+  coap_log_debug("%s\n", buff);
+#endif /* COAP_DEBUG_WAKEUP_TIMES */
+
   return 1;
 #endif /* ! COAP_EPOLL_SUPPORT */
 }
@@ -1791,6 +1831,10 @@ coap_io_process_with_fds_lkd(coap_context_t *ctx, uint32_t timeout_ms,
   } else {
     FD_ZERO(&ctx->exceptfds);
   }
+  if (ctx->pipefd[0] != -1) {
+    nfds = ctx->pipefd[0] + 1;
+    FD_SET(ctx->pipefd[0], &ctx->readfds);
+  }
   for (i = 0; i < ctx->num_sockets; i++) {
     if (ctx->sockets[i]->fd + 1 > nfds)
       nfds = ctx->sockets[i]->fd + 1;
@@ -1855,6 +1899,21 @@ coap_io_process_with_fds_lkd(coap_context_t *ctx, uint32_t timeout_ms,
     tv.tv_sec = 0;
     select((int)nfds, &ctx->readfds, &ctx->writefds, &ctx->exceptfds, &tv);
 #endif /* COAP_THREAD_SAFE */
+    if (ctx->pipefd[0] != -1) {
+      if (FD_ISSET(ctx->pipefd[0], &ctx->readfds)) {
+        /* Timeout has been reduced */
+        char byte;
+
+        /* Check the result from read() to suppress the warning on
+         * systems that declare read() with warn_unused_result. */
+#ifdef _WIN32
+#define read(a,b,c) _read(a,b,c)
+#endif
+        if (read(ctx->pipefd[0], &byte, sizeof(byte)) == -1) {
+          /* do nothing */;
+        }
+      }
+    }
     for (i = 0; i < ctx->num_sockets; i++) {
       if ((ctx->sockets[i]->flags & COAP_SOCKET_WANT_READ) &&
           FD_ISSET(ctx->sockets[i]->fd, &ctx->readfds))

src/coap_net.c

@@ -46,6 +46,9 @@
 #ifdef HAVE_NET_IF_H
 #include <net/if.h>
 #endif
+#ifdef HAVE_FCNTL_H
+#include <fcntl.h>
+#endif
 #ifdef COAP_EPOLL_SUPPORT
 #include <sys/epoll.h>
 #include <sys/timerfd.h>
@@ -694,6 +697,30 @@ coap_new_context(const coap_address_t *listen_addr) {
       }
     }
   }
+#else /* COAP_EPOLL_SUPPORT */
+#if !defined(WITH_LWIP) && !defined(WITH_CONTIKI)
+#ifdef _WIN32
+#include <io.h>
+#include <fcntl.h>
+#define pipe(a) _pipe(a, 128, _O_BINARY)
+#endif
+  if (pipe(c->pipefd) == -1) {
+    coap_log_err("coap_new_context: Unable to pipe(): %s (%d)\n",
+                 coap_socket_strerror(),
+                 errno);
+  } else {
+#ifndef _WIN32
+    int flags;
+
+    flags = fcntl(c->pipefd[0], F_GETFL);
+    flags |= O_NONBLOCK;
+    fcntl(c->pipefd[0], F_SETFL, flags);
+    flags = fcntl(c->pipefd[1], F_GETFL);
+    flags |= O_NONBLOCK;
+    fcntl(c->pipefd[1], F_SETFL, flags);
+#endif
+  }
+#endif /* ! WITH_LWIP && ! WITH_CONTIKI */
 #endif /* COAP_EPOLL_SUPPORT */
 
   if (coap_dtls_is_supported() || coap_tls_is_supported()) {
@@ -828,7 +855,20 @@ coap_free_context_lkd(coap_context_t *context) {
     close(context->epfd);
     context->epfd = -1;
   }
+#else /* COAP_EPOLL_SUPPORT */
+#if !defined(WITH_LWIP) && !defined(WITH_CONTIKI)
+  if (context->pipefd[0] != -1) {
+#ifdef _WIN32
+#define close(a) _close(a)
+#endif
+    close(context->pipefd[0]);
+    close(context->pipefd[0]);
+    context->pipefd[0] = -1;
+    context->pipefd[1] = -1;
+  }
+#endif /* ! WITH_LWIP && ! WITH_CONTIKI */
 #endif /* COAP_EPOLL_SUPPORT */
+
 #if COAP_SERVER_SUPPORT
 #if COAP_WITH_OBSERVE_PERSIST
   coap_persist_cleanup(context);

src/coap_resource.c

@@ -1342,6 +1342,11 @@ coap_resource_notify_observers_lkd(coap_resource_t *r,
   return 1;
 }
 
+int
+coap_resource_has_observers(coap_resource_t *r) {
+  return r && r->observable && r->subscribers;
+}
+
 void
 coap_resource_set_mode(coap_resource_t *resource, int mode) {
   resource->flags = (resource->flags &