src/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, prog->fd);
 186         return 1;
 187 }
 188
 189 /* ------------------------------------------------------------------------
 190  * checks if an executable is still running
 191  *
 192  * lua: running = el:check_exec(fd)
 193  *
 194  *    fd - file descriptor returned from el:add_exec()
 195  *    running - boolean indicating if it's still running
 196  */
 197
 198 int l_eventloop_check_exec (lua_State *L)
 199 {
 200         struct lel_eventloop *el;
 201         int fd, i;
 202         bool found = false;
 203
 204         el = lel_checkeventloop(L, 1);
 205         fd = luaL_checknumber(L, 2);
 206
 207         DBGF("** eventloop:check_exec (%d) **\n", fd);
 208
 209         for (i=(el->progs_count-1); i>=0; i--) {
 210                 struct lel_program *prog;
 211
 212                 prog = el->progs[i];
 213
 214                 if (prog->fd != fd)
 215                         continue;
 216
 217                 found = true;
 218                 break;
 219         }
 220
 221         lua_pushboolean(L, found);
 222         return 1;
 223 }
 224
 225
 226 /* ------------------------------------------------------------------------
 227  * kills off a previously spawned off process and cleans up
 228  * (actually, it just closes the fifo)
 229  *
 230  * lua: el:kill_exec(fd)
 231  *
 232  *    fd - return from add_exec()
 233  */
 234
 235 int l_eventloop_kill_exec (lua_State *L)
 236 {
 237         struct lel_eventloop *el;
 238         int fd;
 239
 240         el = lel_checkeventloop (L, 1);
 241         fd = luaL_checknumber (L, 2);
 242
 243         DBGF("** eventloop:kill_exec (%d) **\n", fd);
 244
 245         kill_exec (L, el, fd);
 246
 247         return 0;
 248 }
 249
 250 static void kill_exec (lua_State *L, struct lel_eventloop *el, int fd)
 251 {
 252         struct lel_program *prog;
 253         int status;
 254
 255         prog = progs_remove (el, fd);
 256         if (! prog)
 257                 return;
 258
 259         if (el->max_fd == prog->fd) {
 260                 if (el->progs_count)
 261                         // last entry is the new max
 262                         el->max_fd = el->progs[el->progs_count-1]->fd;
 263                 else
 264                         // there are no more entries
 265                         el->max_fd = 0;
 266         }
 267
 268         FD_CLR (prog->fd, &el->all_fds);
 269
 270         kill (prog->pid, SIGTERM);
 271         close (prog->fd);
 272         free (prog);
 273
 274         // catchup on programs that quit
 275         while (waitpid (-1, &status, WNOHANG) > 0);
 276
 277         // and we still have to remove it from the table
 278         luaL_getmetatable (L, L_EVENTLOOP_MT);  // [-3] = get the table
 279         lua_pushinteger (L, prog->fd);          // [-2] = the key
 280         lua_pushnil (L);                        // [-1] = nil
 281         lua_settable (L, -3);                   // eventloop[fd] = nil
 282
 283         // cleanup
 284         lua_gc (L, LUA_GCSTEP, 10);
 285 }
 286
 287 /* ------------------------------------------------------------------------
 288  * runs the select loop over all registered execs with timeout
 289  *
 290  * lua: el.run_loop (timeout)
 291  */
 292 int l_eventloop_run_loop (lua_State *L)
 293 {
 294         struct lel_eventloop *el;
 295         int timeout, status;
 296         fd_set rfds, xfds;
 297         struct timeval tv;
 298
 299         el = lel_checkeventloop (L, 1);
 300         timeout = luaL_optnumber (L, 2, 0);
 301
 302         DBGF("** eventloop:run_loop (%d) **\n", timeout);
 303
 304         // init for timeout
 305         tv.tv_sec = timeout;
 306         tv.tv_usec = 0;
 307
 308         // run the loop
 309         while (el->progs_count) {
 310                 int i, rc;
 311
 312                 // catchup on programs that quit
 313                 while (waitpid (-1, &status, WNOHANG) > 0);
 314
 315                 // init for select
 316                 rfds = el->all_fds;
 317                 xfds = el->all_fds;
 318
 319                 // wait for the next event
 320                 rc = select (el->max_fd+1, &rfds, NULL, &xfds, &tv);
 321                 if (rc<0)
 322                         return lel_pusherror (L, "select failed");
 323
 324                 if (!rc)
 325                         // timeout
 326                         break;
 327
 328                 for (i=(el->progs_count-1); i>=0; i--) {
 329                         struct lel_program *prog;
 330                         bool dead = false;
 331
 332                         prog = el->progs[i];
 333
 334                         if (FD_ISSET (prog->fd, &rfds)) {
 335                                 rc = loop_handle_event (L, prog);
 336                                 if (rc<=0)
 337                                         dead = true;
 338
 339                                 // count could have changed in callback
 340                                 if (i >= el->progs_count)
 341                                         break;
 342                         }
 343
 344                         if (dead /* || FD_ISSET (prog->fd, &xfds) */ ) {
 345                                 DBGF("** killing %d (fd=%d) **\n",
 346                                                 prog->pid, prog->fd);
 347                                 kill_exec(L, el, prog->fd);
 348                         }
 349                 }
 350         }
 351
 352         // catchup on programs that quit
 353         while (waitpid (-1, &status, WNOHANG) > 0);
 354
 355         return 0;
 356 }
 357
 358 /* ------------------------------------------------------------------------
 359  * terminates all executables
 360  */
 361 int l_eventloop_kill_all (lua_State *L)
 362 {
 363         struct lel_eventloop *el;
 364         int i;
 365
 366         el = lel_checkeventloop (L, 1);
 367
 368         for (i=(el->progs_count-1); i>=0; i--) {
 369                 struct lel_program *prog;
 370
 371                 prog = el->progs[i];
 372                 kill_exec (L, el, prog->fd);
 373         }
 374
 375         return 0;
 376 }
 377
 378 /* ------------------------------------------------------------------------
 379  * read more data and call callbacks
 380  */
 381
 382 static void prog_read_more (lua_State *L, struct lel_program *prog)
 383 {
 384         int rc;
 385         char *buf;
 386         ssize_t len;
 387
 388         if (!prog->buf_len) {
 389                 // reset pos to beginning
 390                 prog->buf_pos = 0;
 391         }
 392
 393         buf = prog->buf + prog->buf_pos;
 394         len = LEL_PROGRAM_IO_BUF_SIZE - prog->buf_pos - prog->buf_len;
 395
 396         if (len<=0) {
 397                 // reached the end
 398                 if (! prog->buf_pos) {
 399                         // cannot shift data down
 400                         prog->read_rc = -1;
 401                         prog->read_errno = EBUSY;
 402                         return;
 403                 }
 404
 405                 // shift data down to make some more room
 406                 memmove (prog->buf, buf, prog->buf_len);
 407                 prog->buf_pos = 0;
 408                 buf = prog->buf;
 409         }
 410
 411         // get some data
 412         rc = read (prog->fd, buf, len);
 413
 414         prog->read_rc = rc;
 415         prog->read_errno = errno;
 416
 417         if (rc>0) {
 418                 prog->buf_len = rc;
 419                 buf[rc] = 0;
 420         }
 421
 422 }
 423
 424 static void lua_issue_callback (lua_State *L, struct lel_program *prog,
 425                 char *string)
 426 {
 427         int top;
 428
 429         // backup top of stack
 430         top = lua_gettop (L);
 431
 432         // find the call back function
 433         luaL_getmetatable (L, L_EVENTLOOP_MT);  // [-2] = get the table
 434         lua_pushinteger (L, prog->fd);          // [-1] = the key
 435         lua_gettable (L, -2);                   // push (eventloop[fd])
 436
 437         if (string) {
 438                 // success
 439                 lua_pushstring (L, string);
 440                 lua_call (L, 1, 0);
 441
 442         } else if (prog->read_rc == 0) {
 443                 // stream ended
 444                 lua_pushnil (L);
 445                 lua_pushstring (L, "EOF");
 446                 lua_call (L, 2, 0);
 447
 448         } else if (prog->read_rc < 0) {
 449                 // error reading
 450                 lua_pushnil (L);
 451                 lua_pushstring (L, strerror(prog->read_errno));
 452                 lua_call (L, 2, 0);
 453         }
 454
 455         // restore top of stack
 456         lua_settop (L, top);
 457 }
 458
 459 static int loop_handle_event (lua_State *L, struct lel_program *prog)
 460 {
 461         char *s, *e, *cr;
 462
 463         prog_read_more (L, prog);
 464
 465         while (prog->buf_len) {
 466                 // as long as we have some data we try to find a full line
 467                 s = prog->buf + prog->buf_pos;
 468                 e = s + prog->buf_len;
 469
 470                 cr = strchr (s, '\n');
 471                 if (cr) {
 472                         // we have a match: s..cr is our substring
 473                         int len = (cr-s) + 1;
 474                         *cr = 0;
 475
 476                         lua_issue_callback (L, prog, s);
 477
 478                         prog->buf_pos += len;
 479                         prog->buf_len -= len;
 480
 481                 } else if (!prog->buf_pos) {
 482                         // no match and we cannot even read more out of
 483                         // the buffer; we have to return the partial buffer
 484                         lua_issue_callback (L, prog, s);
 485
 486                         prog->buf_len = 0;
 487
 488                 } else {
 489                         // no match, we will try to read more on next select()
 490                         // read event
 491                         break;
 492                 }
 493         }
 494
 495         return prog->read_rc;
 496 }
 497