luaeventloop/lel_instance.c

   1 #include <stdio.h>
   2 #include <stdlib.h>
   3 #include <stdint.h>
   4 #include <stdbool.h>
   5 #include <unistd.h>
   6 #include <string.h>
   7 #include <errno.h>
   8 #include <time.h>
   9 #include <signal.h>
  10 #include <ctype.h>
  11 #include <sys/types.h>
  12 #include <sys/wait.h>
  13
  14 #include <lua.h>
  15 #include <lauxlib.h>
  16
  17 #include "lel_debug.h"
  18 #include "lel_util.h"
  19 #include "lel_instance.h"
  20
  21 // local hepers
  22 static int loop_handle_event (lua_State *L, struct lel_program *prog);
  23 static void kill_exec (lua_State *L, struct lel_eventloop *el, int fd);
  24
  25 /* ------------------------------------------------------------------------
  26  * utility functions
  27  */
  28
  29 struct lel_eventloop *lel_checkeventloop (lua_State *L, int narg)
  30 {
  31         void *ud = luaL_checkudata (L, narg, L_EVENTLOOP_MT);
  32         luaL_argcheck (L, ud != NULL, 1, "`eventloop' expected");
  33         return (struct lel_eventloop*)ud;
  34 }
  35
  36 int l_eventloop_tostring (lua_State *L)
  37 {
  38         struct lel_eventloop *el = lel_checkeventloop (L, 1);
  39         lua_pushfstring (L, "eventloop instance %p", el);
  40         return 1;
  41 }
  42
  43 /* ------------------------------------------------------------------------
  44  * dealing with program list
  45  */
  46
  47 static int progs_compare (const void *_one, const void *_two)
  48 {
  49         const struct lel_program *const*one = _one;
  50         const struct lel_program *const*two = _two;
  51
  52         return (*one)->fd - (*two)->fd;
  53 }
  54
  55 static void progs_add (struct lel_eventloop *el, struct lel_program *prog)
  56 {
  57         // make more room
  58         if (el->progs_size <= el->progs_count) {
  59                 size_t bytes;
  60
  61                 el->progs_size += LEL_PROGS_ARRAY_GROWS_BY;
  62                 bytes = el->progs_size * sizeof(struct lel_program*);
  63
  64                 el->progs = realloc (el->progs, bytes);
  65                 if (!el->progs) {
  66                         // TODO: we could handle this better then blowing away
  67                         // all of our data if we run out of room.  For now exit.
  68                         perror ("malloc");
  69                         exit (1);
  70                 }
  71         }
  72
  73         el->progs[el->progs_count++] = prog;
  74
  75         qsort (el->progs, el->progs_count, sizeof (struct lel_program*),
  76                         progs_compare);
  77 }
  78
  79 static struct lel_program * progs_remove (struct lel_eventloop *el, int fd)
  80 {
  81         struct lel_program key = {.fd = fd};
  82         struct lel_program *pkey = &key;
  83         struct lel_program *found;
  84         struct lel_program **pfound;
  85         size_t end_bytes;
  86
  87         // find the program
  88         pfound = bsearch (&pkey, el->progs, el->progs_count,
  89                         sizeof (struct lel_program*), progs_compare);
  90         if (!pfound)
  91                 return NULL;
  92
  93         found = *pfound;
  94         if (!found)
  95                 return NULL;
  96
  97         // adjust remaining entries
  98         el->progs_count --;
  99         end_bytes = (char*)(el->progs + el->progs_count) - (char*)pfound;
 100         memmove (pfound, pfound+1, end_bytes);
 101
 102         return found;
 103 }
 104
 105
 106 /* ------------------------------------------------------------------------
 107  * executes a new process to handle events from another source
 108  *
 109  * lua: fd = el:add_exec(cmd, function)
 110  *
 111  *    cmd - a string with program and parameters for execution
 112  *    function - a function to call back with data read
 113  *    fd - returned is the file descriptor or nil on error
 114  */
 115
 116 int l_eventloop_add_exec (lua_State *L)
 117 {
 118         struct lel_eventloop *el;
 119         struct lel_program *prog;
 120         const char *cmd;
 121         int pfds[2];                    // 0 is server, 1 is client
 122         int rc, pid;
 123
 124         el = lel_checkeventloop (L, 1);
 125         cmd = luaL_checkstring (L, 2);
 126         (void)luaL_checktype (L, 3, LUA_TFUNCTION);
 127
 128         DBGF("** eventloop:add_exec (%s, ...) **\n", cmd);
 129
 130         // create a new program entry
 131         prog = (struct lel_program*) malloc (sizeof (struct lel_program)
 132                         // we allocate the buffer and room for a terminator at the end
 133                         + LEL_PROGRAM_IO_BUF_SIZE + 1);
 134         if (!prog)
 135                 return lel_pusherror (L, "failed to allocate program structure");
 136
 137         // spawn off a worker process
 138         rc = pipe(pfds);
 139         if (rc<0) {
 140                 free (prog);
 141                 return lel_pusherror (L, "failed to create a pipe");
 142         }
 143
 144         pid = vfork();
 145         if (pid<0) {                    // fork failed...
 146                 free (prog);
 147                 close (pfds[0]);
 148                 close (pfds[1]);
 149                 return lel_pusherror (L, "failed to fork()");
 150         }
 151
 152         if (! pid) {                    // client...
 153                 close (pfds[0]);        // close the server end
 154                 dup2(pfds[1], 1);       // stdout to client's end of pipe
 155                 close (pfds[1]);        // close the client end
 156                 execlp ("sh", "sh", "-c", cmd, NULL);
 157                 exit (1);
 158         }
 159
 160         // back in server...
 161         close (pfds[1]);                        // close the client end
 162
 163         // time to setup the program entry
 164         memset (prog, 0, sizeof(*prog));
 165
 166         prog->cmd = strdup (cmd);
 167         prog->pid = pid;
 168         prog->fd = pfds[0];
 169
 170         if (el->max_fd < prog->fd)
 171                 el->max_fd = prog->fd;
 172
 173         FD_SET (prog->fd, &el->all_fds);
 174
 175         // add to the list
 176         progs_add (el, prog);
 177
 178         // everything is setup, but we need to get a hold of the function later;
 179         // we add the function to the L_EVENTLOOP_MT with the fd as the key...
 180         luaL_getmetatable (L, L_EVENTLOOP_MT);  // [-3] = get the table
 181         lua_pushinteger (L, prog->fd);          // [-2] = the key
 182         lua_pushvalue (L, 3);                   // [-1] = the function (3rd arg)
 183         lua_settable (L, -3);                   // eventloop[fd] = function
 184
 185         lua_pushinteger (L, pid);
 186         return 1;
 187 }
 188
 189 /* ------------------------------------------------------------------------
 190  * kills off a previously spawned off process and cleans up
 191  * (actually, it just closes the fifo)
 192  *
 193  * lua: el:kill_exec(fd)
 194  *
 195  *    fd - return from add_exec()
 196  */
 197
 198 int l_eventloop_kill_exec (lua_State *L)
 199 {
 200         struct lel_eventloop *el;
 201         int fd;
 202
 203         el = lel_checkeventloop (L, 1);
 204         fd = luaL_checknumber (L, 2);
 205
 206         DBGF("** eventloop:kill_exec (%d) **\n", fd);
 207
 208         kill_exec (L, el, fd);
 209
 210         return 0;
 211 }
 212
 213 static void kill_exec (lua_State *L, struct lel_eventloop *el, int fd)
 214 {
 215         struct lel_program *prog;
 216         int status;
 217
 218         prog = progs_remove (el, fd);
 219         if (! prog)
 220                 return;
 221
 222         if (el->max_fd == prog->fd) {
 223                 if (el->progs_count)
 224                         // last entry is the new max
 225                         el->max_fd = el->progs[el->progs_count-1]->fd;
 226                 else
 227                         // there are no more entries
 228                         el->max_fd = 0;
 229         }
 230
 231         FD_CLR (prog->fd, &el->all_fds);
 232
 233         kill (prog->pid, SIGTERM);
 234         close (prog->fd);
 235         free (prog);
 236
 237         // catchup on programs that quit
 238         while (waitpid (-1, &status, WNOHANG) > 0);
 239
 240         // and we still have to remove it from the table
 241         luaL_getmetatable (L, L_EVENTLOOP_MT);  // [-3] = get the table
 242         lua_pushinteger (L, prog->fd);          // [-2] = the key
 243         lua_pushnil (L);                        // [-1] = nil
 244         lua_settable (L, -3);                   // eventloop[fd] = function
 245
 246         // cleanup
 247         lua_gc (L, LUA_GCSTEP, 10);
 248 }
 249
 250 /* ------------------------------------------------------------------------
 251  * runs the select loop over all registered execs with timeout
 252  *
 253  * lua: el.run_loop (timeout)
 254  */
 255 int l_eventloop_run_loop (lua_State *L)
 256 {
 257         struct lel_eventloop *el;
 258         int timeout, status;
 259         fd_set rfds, xfds;
 260         struct timeval tv;
 261
 262         el = lel_checkeventloop (L, 1);
 263         timeout = luaL_optnumber (L, 2, 0);
 264
 265         DBGF("** eventloop:run_loop (%d) **\n", timeout);
 266
 267         // init for timeout
 268         tv.tv_sec = timeout;
 269         tv.tv_usec = 0;
 270
 271         // run the loop
 272         while (el->progs_count) {
 273                 int i, rc;
 274
 275                 // catchup on programs that quit
 276                 while (waitpid (-1, &status, WNOHANG) > 0);
 277
 278                 // init for select
 279                 rfds = el->all_fds;
 280
 281                 // wait for the next event
 282                 rc = select (el->max_fd+1, &rfds, NULL, &xfds, &tv);
 283                 if (rc<0)
 284                         return lel_pusherror (L, "select failed");
 285
 286                 if (!rc)
 287                         // timeout
 288                         break;
 289
 290                 for (i=(el->progs_count-1); i>=0; i--) {
 291                         struct lel_program *prog;
 292                         bool dead = false;
 293
 294                         prog = el->progs[i];
 295
 296                         if (FD_ISSET (prog->fd, &rfds)) {
 297                                 rc = loop_handle_event (L, prog);
 298                                 if (rc<=0)
 299                                         dead = true;
 300
 301                                 // count could have changed in callback
 302                                 if (i >= el->progs_count)
 303                                         break;
 304                         }
 305
 306                         if (dead /* || FD_ISSET (prog->fd, &xfds) */ ) {
 307                                 DBGF("** killing %d (fd=%d) **\n",
 308                                                 prog->pid, prog->fd);
 309                                 kill_exec(L, el, prog->fd);
 310                         }
 311                 }
 312         }
 313
 314         // catchup on programs that quit
 315         while (waitpid (-1, &status, WNOHANG) > 0);
 316
 317         return 0;
 318 }
 319
 320 /* ------------------------------------------------------------------------
 321  * terminates all executables
 322  */
 323 int l_eventloop_kill_all (lua_State *L)
 324 {
 325         struct lel_eventloop *el;
 326         int i;
 327
 328         el = lel_checkeventloop (L, 1);
 329
 330         for (i=(el->progs_count-1); i>=0; i--) {
 331                 struct lel_program *prog;
 332
 333                 prog = el->progs[i];
 334                 kill_exec (L, el, prog->fd);
 335         }
 336
 337         return 0;
 338 }
 339
 340 /* ------------------------------------------------------------------------
 341  * read more data and call callbacks
 342  */
 343
 344 static void prog_read_more (lua_State *L, struct lel_program *prog)
 345 {
 346         int rc;
 347         char *buf;
 348         ssize_t len;
 349
 350         if (!prog->buf_len) {
 351                 // reset pos to beginning
 352                 prog->buf_pos = 0;
 353         }
 354
 355         buf = prog->buf + prog->buf_pos;
 356         len = LEL_PROGRAM_IO_BUF_SIZE - prog->buf_pos - prog->buf_len;
 357
 358         if (len<=0) {
 359                 // reached the end
 360                 if (! prog->buf_pos) {
 361                         // cannot shift data down
 362                         prog->read_rc = -1;
 363                         prog->read_errno = EBUSY;
 364                         return;
 365                 }
 366
 367                 // shift data down to make some more room
 368                 memmove (prog->buf, buf, prog->buf_len);
 369                 prog->buf_pos = 0;
 370                 buf = prog->buf;
 371         }
 372
 373         // get some data
 374         rc = read (prog->fd, buf, len);
 375
 376         prog->read_rc = rc;
 377         prog->read_errno = errno;
 378
 379         if (rc>0) {
 380                 prog->buf_len = rc;
 381                 buf[rc] = 0;
 382         }
 383
 384 }
 385
 386 static void lua_issue_callback (lua_State *L, struct lel_program *prog,
 387                 char *string)
 388 {
 389         int top;
 390
 391         // backup top of stack
 392         top = lua_gettop (L);
 393
 394         // find the call back function
 395         luaL_getmetatable (L, L_EVENTLOOP_MT);  // [-2] = get the table
 396         lua_pushinteger (L, prog->fd);          // [-1] = the key
 397         lua_gettable (L, -2);                   // push (eventloop[fd])
 398
 399         if (string) {
 400                 // success
 401                 lua_pushstring (L, string);
 402                 lua_call (L, 1, 0);
 403
 404         } else if (prog->read_rc == 0) {
 405                 // stream ended
 406                 lua_pushnil (L);
 407                 lua_pushstring (L, "EOF");
 408                 lua_call (L, 2, 0);
 409
 410         } else if (prog->read_rc < 0) {
 411                 // error reading
 412                 lua_pushnil (L);
 413                 lua_pushstring (L, strerror(prog->read_errno));
 414                 lua_call (L, 2, 0);
 415         }
 416
 417         // restore top of stack
 418         lua_settop (L, top);
 419 }
 420
 421 static int loop_handle_event (lua_State *L, struct lel_program *prog)
 422 {
 423         char *s, *e, *cr;
 424
 425         prog_read_more (L, prog);
 426
 427         while (prog->buf_len) {
 428                 // as long as we have some data we try to find a full line
 429                 s = prog->buf + prog->buf_pos;
 430                 e = s + prog->buf_len;
 431
 432                 cr = strchr (s, '\n');
 433                 if (cr) {
 434                         // we have a match: s..cr is our substring
 435                         int len = (cr-s) + 1;
 436                         *cr = 0;
 437
 438                         lua_issue_callback (L, prog, s);
 439
 440                         prog->buf_pos += len;
 441                         prog->buf_len -= len;
 442
 443                 } else if (!prog->buf_pos) {
 444                         // no match and we cannot even read more out of
 445                         // the buffer; we have to return the partial buffer
 446                         lua_issue_callback (L, prog, s);
 447
 448                         prog->buf_len = 0;
 449
 450                 } else {
 451                         // no match, we will try to read more on next select()
 452                         // read event
 453                         break;
 454                 }
 455         }
 456
 457         return prog->read_rc;
 458 }
 459