Periodic tasks

10 files changed, 115 insertions, 10 deletions

diff --git a/doc/manual.tex b/doc/manual.tex
index 5a7110cb..f86ea97e 100644
--- a/doc/manual.tex
+++ b/doc/manual.tex
@@ -2100,15 +2100,17 @@ In many web applications, it's useful to run code at points other than requests
 $$\begin{array}{l}
 \mt{con} \; \mt{task\_kind} :: \mt{Type} \to \mt{Type} \\
 \mt{val} \; \mt{initialize} : \mt{task\_kind} \; \mt{unit} \\
-\mt{val} \; \mt{clientLeaves} : \mt{task\_kind} \; \mt{client}
+\mt{val} \; \mt{clientLeaves} : \mt{task\_kind} \; \mt{client} \\
+\mt{val} \; \mt{periodic} : \mt{int} \to \mt{task\_kind} \; \mt{unit}
 \end{array}$$
 
 A task kind names a particular extension point of generated applications, where the type parameter of a task kind describes which extra input data is available at that extension point.  Add task code with the special declaration form $\mt{task} \; e_1 = e_2$, where $e_1$ is a task kind with data $\tau$, and $e_2$ is a function from $\tau$ to $\mt{transaction} \; \mt{unit}$.
 
 The currently supported task kinds are:
 \begin{itemize}
-\item $\mt{initialize}$: Code that is run in each freshly-allocated request context.
+\item $\mt{initialize}$: Code that is run when the application starts up.
 \item $\mt{clientLeaves}$: Code that is run for each client that the runtime system decides has surfed away.  When a request that generates a new client handle is aborted, that handle will still eventually be passed to $\mt{clientLeaves}$ task code, even though the corresponding browser was never informed of the client handle's existence.  In other words, in general, $\mt{clientLeaves}$ handlers will be called more times than there are actual clients.
+\item $\mt{periodic} \; n$: Code that is run when the application starts up and then every $n$ seconds thereafter.
 \end{itemize}
 
 
diff --git a/include/types.h b/include/types.h
index 2adda753..01776213 100644
--- a/include/types.h
+++ b/include/types.h
@@ -59,6 +59,11 @@ typedef void (*uw_callback_with_retry)(void *, int will_retry);
 typedef void (*uw_logger)(void*, const char *fmt, ...);
 
 typedef struct {
+  void (*callback)(uw_context);
+  unsigned int period;
+} uw_periodic;
+
+typedef struct {
   int inputs_len, timeout;
   char *url_prefix;
 
@@ -80,6 +85,8 @@ typedef struct {
   int (*check_mime)(const char *);
 
   void (*on_error)(uw_context, char *);
+
+  uw_periodic *periodics; // 0-terminated array
 } uw_app;
 
 #define ERROR_BUF_LEN 1024
diff --git a/include/urweb.h b/include/urweb.h
index a0decd11..9314d089 100644
--- a/include/urweb.h
+++ b/include/urweb.h
@@ -293,4 +293,6 @@ uw_Basis_int uw_Basis_rand(uw_context);
 
 extern int uw_time_max, uw_supports_direct_status, uw_min_heap;
 
+failure_kind uw_runCallback(uw_context, void (*callback)(uw_context));
+
 #endif
diff --git a/lib/ur/basis.urs b/lib/ur/basis.urs
index 2a61b701..2a3b9a33 100644
--- a/lib/ur/basis.urs
+++ b/lib/ur/basis.urs
@@ -813,6 +813,7 @@ val show_xml : ctx ::: {Unit} -> use ::: {Type} -> bind ::: {Type} -> show (xml
 con task_kind :: Type -> Type
 val initialize : task_kind unit
 val clientLeaves : task_kind client
+val periodic : int -> task_kind unit
 
 
 (** Information flow security *)
diff --git a/src/c/request.c b/src/c/request.c
index e51f95ae..b49a524e 100644
--- a/src/c/request.c
+++ b/src/c/request.c
@@ -79,9 +79,44 @@ static void *ticker(void *data) {
   return NULL;
 }
 
+typedef struct {
+  uw_app *app;
+  void *logger_data;
+  uw_logger log_error, log_debug;
+} loggers;
+
+typedef struct {
+  loggers *ls;
+  uw_periodic pdic;
+} periodic;
+
+static void *periodic_loop(void *data) {
+  periodic *p = (periodic *)data;
+  uw_context ctx = uw_request_new_context(p->ls->app, p->ls->logger_data, p->ls->log_error, p->ls->log_debug);
+
+  if (!ctx)
+    exit(1);
+
+  while (1) {
+    failure_kind r;
+    do {
+      r = uw_runCallback(ctx, p->pdic.callback);
+    } while (r == UNLIMITED_RETRY);
+
+    sleep(p->pdic.period);
+  };
+}
+
 void uw_request_init(uw_app *app, void *logger_data, uw_logger log_error, uw_logger log_debug) {
   uw_context ctx;
   failure_kind fk;
+  uw_periodic *ps;
+  loggers *ls = malloc(sizeof(loggers));
+
+  ls->app = app;
+  ls->logger_data = logger_data;
+  ls->log_error = log_error;
+  ls->log_debug = log_debug;
 
   uw_global_init();
   uw_app_init(app);
@@ -113,6 +148,18 @@ void uw_request_init(uw_app *app, void *logger_data, uw_logger log_error, uw_log
   }
 
   uw_free(ctx);
+
+  for (ps = app->periodics; ps->callback; ++ps) {
+    pthread_t thread;
+    periodic *arg = malloc(sizeof(periodic));
+    arg->ls = ls;
+    arg->pdic = *ps;
+    
+    if (pthread_create(&thread, NULL, periodic_loop, arg)) {
+      fprintf(stderr, "Error creating periodic thread\n");
+      exit(1);
+    }
+  }  
 }
 
 
@@ -468,12 +515,6 @@ request_result uw_request(uw_request_context rc, uw_context ctx,
   }
 }
 
-typedef struct {
-  uw_app *app;
-  void *logger_data;
-  uw_logger log_error, log_debug;
-} loggers;
-
 void *client_pruner(void *data) {
   loggers *ls = (loggers *)data;
   uw_context ctx = uw_request_new_context(ls->app, ls->logger_data, ls->log_error, ls->log_debug);
diff --git a/src/c/urweb.c b/src/c/urweb.c
index 2b54e87c..0356e0fa 100644
--- a/src/c/urweb.c
+++ b/src/c/urweb.c
@@ -3492,3 +3492,19 @@ uw_Basis_postBody uw_getPostBody(uw_context ctx) {
   else
     uw_error(ctx, FATAL, "Asked for POST body when none exists");
 }
+
+failure_kind uw_runCallback(uw_context ctx, void (*callback)(uw_context)) {
+  int r = setjmp(ctx->jmp_buf);
+
+  if (ctx->app->db_begin(ctx))
+    uw_error(ctx, BOUNDED_RETRY, "Error running SQL BEGIN");
+
+  if (r == 0) {
+    callback(ctx);
+    uw_commit(ctx);
+  }
+  else
+    uw_rollback(ctx, 0);
+
+  return r;
+}
diff --git a/src/cjr.sml b/src/cjr.sml
index c57128cf..7ea665ce 100644
--- a/src/cjr.sml
+++ b/src/cjr.sml
@@ -103,7 +103,7 @@ datatype exp' =
 
 withtype exp = exp' located
 
-datatype task = Initialize | ClientLeaves
+datatype task = Initialize | ClientLeaves | Periodic of Int64.int
 
 datatype decl' =
          DStruct of int * (string * typ) list
diff --git a/src/cjr_print.sml b/src/cjr_print.sml
index 2bb5775e..b4f75eb5 100644
--- a/src/cjr_print.sml
+++ b/src/cjr_print.sml
@@ -2796,6 +2796,7 @@ fun p_file env (ds, ps) =
 
         val initializers = List.mapPartial (fn (DTask (Initialize, x1, x2, e), _) => SOME (x1, x2, e) | _ => NONE) ds
         val expungers = List.mapPartial (fn (DTask (ClientLeaves, x1, x2, e), _) => SOME (x1, x2, e) | _ => NONE) ds
+        val periodics = List.mapPartial (fn (DTask (Periodic n, x1, x2, e), _) => SOME (n, x1, x2, e) | _ => NONE) ds
 
         val onError = ListUtil.search (fn (DOnError n, _) => SOME n | _ => NONE) ds
 
@@ -2887,6 +2888,36 @@ fun p_file env (ds, ps) =
              newline,
              newline,
 
+             box (ListUtil.mapi (fn (i, (_, x1, x2, e)) =>
+                                    box [string "static void uw_periodic",
+                                         string (Int.toString i),
+                                         string "(uw_context ctx) {",
+                                         newline,
+                                         box [string "uw_unit __uwr_",
+                                              string x1,
+                                              string "_0 = uw_unit_v, __uwr_",
+                                              string x2,
+                                              string "_1 = uw_unit_v;",
+                                              newline,
+                                              p_exp (E.pushERel (E.pushERel env x1 dummyt) x2 dummyt) e,
+                                              string ";",
+                                              newline],
+                                         string "}",
+                                         newline,
+                                         newline]) periodics),
+
+             string "static uw_periodic my_periodics[] = {",
+             box (ListUtil.mapi (fn (i, (n, _, _, _)) =>
+                                    box [string "{uw_periodic",
+                                         string (Int.toString i),
+                                         string ",",
+                                         space,
+                                         string (Int64.toString n),
+                                         string "},"]) periodics),
+             string "{NULL}};",
+             newline,
+             newline,
+
              string "static const char begin_xhtml[] = \"<?xml version=\\\"1.0\\\" encoding=\\\"utf-8\\\" ?>\\n<!DOCTYPE html PUBLIC \\\"-//W3C//DTD XHTML 1.0 Transitional//EN\\\" \\\"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd\\\">\\n<html xmlns=\\\"http://www.w3.org/1999/xhtml\\\" xml:lang=\\\"en\\\" lang=\\\"en\\\">\";",
              newline,
              newline,
@@ -3043,7 +3074,7 @@ fun p_file env (ds, ps) =
                          "uw_db_init", "uw_db_begin", "uw_db_commit", "uw_db_rollback", "uw_db_close",
                          "uw_handle",
                          "uw_input_num", "uw_cookie_sig", "uw_check_url", "uw_check_mime",
-                         case onError of NONE => "NULL" | SOME _ => "uw_onError"],
+                         case onError of NONE => "NULL" | SOME _ => "uw_onError", "my_periodics"],
              string "};",
              newline]
     end
diff --git a/src/cjrize.sml b/src/cjrize.sml
index 0505af62..9c297fad 100644
--- a/src/cjrize.sml
+++ b/src/cjrize.sml
@@ -667,6 +667,7 @@ fun cifyDecl ((d, loc), sm) =
                  val tk = case #1 e1 of
                               L.EFfi ("Basis", "initialize") => L'.Initialize
                             | L.EFfi ("Basis", "clientLeaves") => L'.ClientLeaves
+                            | L.EFfiApp ("Basis", "periodic", [(L.EPrim (Prim.Int n), _)]) => L'.Periodic n
                             | _ => (ErrorMsg.errorAt loc "Task kind not fully determined";
                                     L'.Initialize)
                  val (e, sm) = cifyExp (e, sm)
diff --git a/tests/periodic.ur b/tests/periodic.ur
new file mode 100644
index 00000000..baf49b3b
--- /dev/null
+++ b/tests/periodic.ur
@@ -0,0 +1,4 @@
+task periodic 5 = fn () => debug "Every 5 seconds"
+task periodic 13 = fn () => debug "Every 13 seconds"
+
+fun main () : transaction page = return <xml/>