Threaded Dispatcher (#49)

* dispatcher POC

* add host connection

* accept external cv

* recv aio signal from host

* add host url

* create host socket

* move to req/rep

* port dispatcher concept

* comment unsafe print statements from thread + document R function

* re-use aios

* use safe printfs

* use new API layer

* disable retries at daemon level

* record active pipes, allow scale down

* tidy up urls

* prelim threaded dispatcher tls support

* add thread_disp_finalizer

* add safety

* simplify - remove need for pipe notify

* use pipe events to track connections through cv flag value

* doc for internal use

* add more safety

* align APIs

* use more efficient saio structs

* active status concept

* synchronize

* read active status

* support daemon time/task outs

* signal from pipe events

* catch one

* rename to online

* remove one lock

* synchronize daemons

* fix synchronization routine

* move url parse out of loop

* take all urls as inputs

* add tests

* fixes and cleanups

* passes valgrind

* increase coverage

* more coverage

* try more

* run comms over dispatcher

* widen test timings

* sync comms check

Author: shikokuchuo

NAMESPACE

@@ -47,8 +47,10 @@ export("%~>%")
 export("opt<-")
 export(.advance)
 export(.context)
+export(.dispatcher)
 export(.keep)
 export(.mark)
+export(.online)
 export(.unresolved)
 export(call_aio)
 export(call_aio_)

R/sync.R

@@ -306,3 +306,36 @@ unlock <- function(socket) invisible(.Call(rnng_socket_unlock, socket))
 #' @export
 #'
 `%~>%` <- function(cv, cv2) invisible(.Call(rnng_signal_thread_create, cv, cv2))
+
+#' Dispatcher Socket
+#'
+#' Creates a Dispatcher socket, which is a special type of \sQuote{req} socket,
+#'     with FIFO scheduling using a threaded implementation (for internal use
+#'     only).
+#'
+#' @param cv a \sQuote{conditionVariable} object.
+#' @param host \sQuote{inproc://} url connecting the host to the thread.
+#' @param url the URLs at which to listen for rep nodes.
+#' @inheritParams listen
+#'
+#' @return A \sQuote{req} Socket. The thread is attached as an attribute.
+#'
+#' @keywords internal
+#' @export
+#'
+.dispatcher <- function(cv, host, url, tls = quote(expr =))
+  .Call(rnng_dispatcher_socket, cv, host, url, tls)
+
+#' Read Online Status
+#'
+#' Reads the online status of threaded dispatcher sockets (for internal use
+#'     only).
+#'
+#' @param sock a dispatcher Socket.
+#'
+#' @return An vector of integer values.
+#'
+#' @keywords internal
+#' @export
+#'
+.online <- function(sock) .Call(rnng_read_online, sock)

man/dot-dispatcher.Rd

@@ -67,28 +67,6 @@ static void raio_complete(void *arg) {
 
 }
 
-static void raio_complete_signal(void *arg) {
-
-  nano_aio *raio = (nano_aio *) arg;
-  nano_cv *ncv = (nano_cv *) raio->next;
-  nng_cv *cv = ncv->cv;
-  nng_mtx *mtx = ncv->mtx;
-
-  const int res = nng_aio_result(raio->aio);
-  if (res == 0)
-    raio->data = nng_aio_get_msg(raio->aio);
-
-  nng_mtx_lock(mtx);
-  raio->result = res - !res;
-  ncv->condition++;
-  nng_cv_wake(cv);
-  nng_mtx_unlock(mtx);
-
-  if (raio->cb != NULL)
-    later2(raio_invoke_cb, raio->cb);
-
-}
-
 static void iraio_complete(void *arg) {
 
   nano_aio *iaio = (nano_aio *) arg;

man/dot-online.Rd

@@ -122,6 +122,33 @@ static SEXP nano_outHook(SEXP x, SEXP fun) {
 
 // functions with forward definitions in nanonext.h ----------------------------
 
+void raio_complete_signal(void *arg) {
+
+  nano_aio *raio = (nano_aio *) arg;
+  nano_cv *ncv = (nano_cv *) raio->next;
+  nng_cv *cv = ncv->cv;
+  nng_mtx *mtx = ncv->mtx;
+
+  const int res = nng_aio_result(raio->aio);
+  if (res == 0)
+    raio->data = nng_aio_get_msg(raio->aio);
+
+  nng_mtx_lock(mtx);
+  raio->result = res - !res;
+  ncv->condition++;
+  nng_cv_wake(cv);
+  nng_mtx_unlock(mtx);
+
+}
+
+void sendaio_complete(void *arg) {
+
+  nng_aio *aio = ((nano_saio *) arg)->aio;
+  if (nng_aio_result(aio))
+    nng_msg_free(nng_aio_get_msg(aio));
+
+}
+
 void dialer_finalizer(SEXP xptr) {
 
   if (NANO_PTR(xptr) == NULL) return;
@@ -153,6 +180,16 @@ void socket_finalizer(SEXP xptr) {
 
 }
 
+void cv_finalizer(SEXP xptr) {
+
+  if (NANO_PTR(xptr) == NULL) return;
+  nano_cv *xp = (nano_cv *) NANO_PTR(xptr);
+  nng_cv_free(xp->cv);
+  nng_mtx_free(xp->mtx);
+  R_Free(xp);
+
+}
+
 #if R_VERSION < R_Version(4, 1, 0)
 
 inline SEXP R_NewEnv(SEXP parent, int hash, int size) {

src/aio.c

@@ -143,6 +143,7 @@ static const R_CallMethodDef callMethods[] = {
   {"rnng_dial", (DL_FUNC) &rnng_dial, 5},
   {"rnng_dialer_close", (DL_FUNC) &rnng_dialer_close, 1},
   {"rnng_dialer_start", (DL_FUNC) &rnng_dialer_start, 2},
+  {"rnng_dispatcher_socket", (DL_FUNC) &rnng_dispatcher_socket, 4},
   {"rnng_fini", (DL_FUNC) &rnng_fini, 0},
   {"rnng_get_opt", (DL_FUNC) &rnng_get_opt, 2},
   {"rnng_is_error_value", (DL_FUNC) &rnng_is_error_value, 1},
@@ -160,6 +161,7 @@ static const R_CallMethodDef callMethods[] = {
   {"rnng_pipe_notify", (DL_FUNC) &rnng_pipe_notify, 6},
   {"rnng_protocol_open", (DL_FUNC) &rnng_protocol_open, 6},
   {"rnng_random", (DL_FUNC) &rnng_random, 2},
+  {"rnng_read_online", (DL_FUNC) &rnng_read_online, 1},
   {"rnng_reap", (DL_FUNC) &rnng_reap, 1},
   {"rnng_recv", (DL_FUNC) &rnng_recv, 4},
   {"rnng_recv_aio", (DL_FUNC) &rnng_recv_aio, 6},

src/core.c

@@ -212,6 +212,25 @@ typedef struct nano_thread_duo_s {
   nano_cv *cv2;
 } nano_thread_duo;
 
+typedef struct nano_thread_disp_s {
+  nng_thread *thr;
+  nano_cv *cv;
+  nng_tls_config *tls;
+  nano_saio **saio;
+  nano_aio **raio;
+  nano_aio **haio;
+  nng_url *up;
+  const char *host;
+  char **url;
+  int *online;
+  R_xlen_t n;
+} nano_thread_disp;
+
+typedef struct nano_signal_s {
+  nano_cv *cv;
+  int *online;
+} nano_signal;
+
 typedef struct nano_buf_s {
   unsigned char *buf;
   size_t len;
@@ -227,6 +246,9 @@ SEXP R_mkClosure(SEXP, SEXP, SEXP);
 SEXP nano_findVarInFrame(const SEXP, const SEXP);
 SEXP nano_PreserveObject(const SEXP);
 void nano_ReleaseObject(SEXP);
+void raio_complete_signal(void *);
+void sendaio_complete(void *);
+void cv_finalizer(SEXP);
 void dialer_finalizer(SEXP);
 void listener_finalizer(SEXP);
 void socket_finalizer(SEXP);
@@ -275,6 +297,7 @@ SEXP rnng_cv_wait_safe(SEXP);
 SEXP rnng_dial(SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_dialer_close(SEXP);
 SEXP rnng_dialer_start(SEXP, SEXP);
+SEXP rnng_dispatcher_socket(SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_fini(void);
 SEXP rnng_get_opt(SEXP, SEXP);
 SEXP rnng_is_error_value(SEXP);
@@ -293,6 +316,7 @@ SEXP rnng_pipe_close(SEXP);
 SEXP rnng_pipe_notify(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_protocol_open(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_random(SEXP, SEXP);
+SEXP rnng_read_online(SEXP);
 SEXP rnng_reap(SEXP);
 SEXP rnng_recv(SEXP, SEXP, SEXP, SEXP);
 SEXP rnng_recv_aio(SEXP, SEXP, SEXP, SEXP, SEXP, SEXP);

src/init.c

@@ -76,14 +76,6 @@ static void request_complete_signal(void *arg) {
 
 }
 
-static void sendaio_complete(void *arg) {
-
-  nng_aio *aio = ((nano_saio *) arg)->aio;
-  if (nng_aio_result(aio))
-    nng_msg_free(nng_aio_get_msg(aio));
-
-}
-
 static void pipe_cb_signal(nng_pipe p, nng_pipe_ev ev, void *arg) {
 
   int sig;
@@ -161,16 +153,6 @@ static void pipe_cb_dropcon(nng_pipe p, nng_pipe_ev ev, void *arg) {
 
 // finalizers ------------------------------------------------------------------
 
-static void cv_finalizer(SEXP xptr) {
-
-  if (NANO_PTR(xptr) == NULL) return;
-  nano_cv *xp = (nano_cv *) NANO_PTR(xptr);
-  nng_cv_free(xp->cv);
-  nng_mtx_free(xp->mtx);
-  R_Free(xp);
-
-}
-
 static void cv_duo_finalizer(SEXP xptr) {
 
   if (NANO_PTR(xptr) == NULL) return;