root/src/stratcon_iep.c

Revision a559c4e0ffb0db177961118cc4e7928a441374ac, 19.7 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 4 years ago)

refs #34

  • Property mode set to 100644
Line 
1 /*
2  * Copyright (c) 2007, OmniTI Computer Consulting, Inc.
3  * All rights reserved.
4  *
5  * Redistribution and use in source and binary forms, with or without
6  * modification, are permitted provided that the following conditions are
7  * met:
8  *
9  *     * Redistributions of source code must retain the above copyright
10  *       notice, this list of conditions and the following disclaimer.
11  *     * Redistributions in binary form must reproduce the above
12  *       copyright notice, this list of conditions and the following
13  *       disclaimer in the documentation and/or other materials provided
14  *       with the distribution.
15  *     * Neither the name OmniTI Computer Consulting, Inc. nor the names
16  *       of its contributors may be used to endorse or promote products
17  *       derived from this software without specific prior written
18  *       permission.
19  *
20  * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21  * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22  * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23  * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24  * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25  * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26  * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27  * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28  * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29  * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30  * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
31  */
32
33 #include "noit_defines.h"
34 #include "eventer/eventer.h"
35 #include "utils/noit_log.h"
36 #include "utils/noit_b64.h"
37 #include "noit_jlog_listener.h"
38 #include "stratcon_jlog_streamer.h"
39 #include "stratcon_datastore.h"
40 #include "stratcon_iep.h"
41 #include "noit_conf.h"
42 #include "noit_check.h"
43
44 #include <sys/types.h>
45 #ifdef HAVE_SYS_WAIT_H
46 #include <sys/wait.h>
47 #endif
48 #include <sys/stat.h>
49 #include <fcntl.h>
50 #include <unistd.h>
51 #include <sys/fcntl.h>
52 #ifdef HAVE_SYS_FILIO_H
53 #include <sys/filio.h>
54 #endif
55 #include <signal.h>
56 #include <errno.h>
57 #include <assert.h>
58
59 eventer_jobq_t iep_jobq;
60 static noit_log_stream_t noit_iep = NULL;
61 static noit_spinlock_t iep_conn_cnt = 0;
62
63 static pthread_key_t iep_connection;
64 static noit_hash_table mq_drivers = NOIT_HASH_EMPTY;
65 static mq_driver_t *mq_driver = NULL;
66
67 struct iep_job_closure {
68   char *line;       /* This is a copy and gets trashed during processing */
69   char *remote;
70   char *doc_str;
71 };
72
73 static void
74 start_iep_daemon();
75
76 static float
77 stratcon_iep_age_from_line(char *data, struct timeval now) {
78   float n, t;
79   if(data && (*data == 'S' || *data == 'M')) {
80     if(data[1] != '\t') return 0;
81     t = strtof(data + 2, NULL);
82     n = (float)now.tv_sec + (float)now.tv_usec / 1000000.0;
83     return n - t;
84   }
85   return 0;
86 }
87
88 struct statement_node {
89   char *id;
90   char *statement;
91   char *provides;
92   int marked; /* helps with identifying cycles */
93   int nrequires;
94   struct statement_node **requires;
95 };
96 static void
97 statement_node_free(void *vstmt) {
98   struct statement_node *stmt = vstmt;
99   if(stmt->id) free(stmt->id);
100   if(stmt->statement) free(stmt->statement);
101   if(stmt->provides) free(stmt->provides);
102   if(stmt->requires) free(stmt->requires);
103 }
104 static int
105 stmt_mark_dag(struct statement_node *stmt, int mgen) {
106   int i;
107   assert(stmt->marked <= mgen);
108   if(stmt->marked == mgen) return -1;
109   if(stmt->marked > 0) return 0; /* validated in a previous sweep */
110   stmt->marked = mgen;
111   for(i=0; i<stmt->nrequires; i++)
112     if(stmt_mark_dag(stmt->requires[i], mgen) < 0) return -1;
113   return 0;
114 }
115 static void
116 submit_statement_node(struct statement_node *stmt) {
117   int line_len, i;
118   char *line, *cp;
119
120   if(stmt->marked) return;
121   for(i=0; i<stmt->nrequires; i++)
122     submit_statement_node(stmt->requires[i]);
123
124   line_len = 3 /* 2 tabs + \0 */ +
125              1 /* 'D' */ + 1 /* '\n' */ +
126              strlen(stmt->id) + strlen(stmt->statement);
127   line = malloc(line_len);
128   snprintf(line, line_len, "D\t%s\t%s\n", stmt->id, stmt->statement);
129   cp = line;
130   while(cp[0] && cp[1]) {
131     if(*cp == '\n') *cp = ' ';
132     cp++;
133   }
134   noitL(noit_error, "submitting statement: %s\n", line);
135   stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, line, NULL);
136   stmt->marked = 1;
137 }
138 void stratcon_iep_submit_statements() {
139   int i, cnt = 0;
140   noit_conf_section_t *statement_configs;
141   char path[256];
142   struct statement_node *stmt;
143   void *vstmt;
144   noit_hash_table stmt_by_id = NOIT_HASH_EMPTY;
145   noit_hash_table stmt_by_provider = NOIT_HASH_EMPTY;
146   noit_hash_iter iter = NOIT_HASH_ITER_ZERO;
147   const char *key;
148   int klen, mgen = 0;
149
150   snprintf(path, sizeof(path), "/stratcon/iep/queries//statement");
151   statement_configs = noit_conf_get_sections(NULL, path, &cnt);
152   noitL(noit_debug, "Found %d %s stanzas\n", cnt, path);
153
154   /* Phase 1: sweep in all the statements */
155   for(i=0; i<cnt; i++) {
156     char id[UUID_STR_LEN+1];
157     char provides[256];
158     char *statement;
159
160     if(!noit_conf_get_stringbuf(statement_configs[i],
161                                 "self::node()/@id",
162                                 id, sizeof(id))) {
163       noitL(noit_iep, "No uuid specified in query\n");
164       continue;
165     }
166     if(!noit_conf_get_stringbuf(statement_configs[i],
167                                 "ancestor-or-self::node()/@provides",
168                                 provides, sizeof(provides))) {
169       provides[0] = '\0';
170     }
171     if(!noit_conf_get_string(statement_configs[i], "self::node()/epl",
172                              &statement)) {
173       noitL(noit_iep, "No contents specified in statement\n");
174       continue;
175     }
176     stmt = calloc(1, sizeof(*stmt));
177     stmt->id = strdup(id);
178     stmt->statement = statement;
179     stmt->provides = provides[0] ? strdup(provides) : NULL;
180     if(!noit_hash_store(&stmt_by_id, stmt->id, strlen(stmt->id), stmt)) {
181       noitL(noit_error, "Duplicate statement id: %s\n", stmt->id);
182       exit(-1);
183     }
184     if(stmt->provides) {
185       if(!noit_hash_store(&stmt_by_provider, stmt->provides,
186                           strlen(stmt->provides), stmt)) {
187         noitL(noit_error, "Two statements provide: '%s'\n", stmt->provides);
188         exit(-1);
189       }
190     }
191   }
192
193   /* Phase 2: load the requires graph */
194   for(i=0; i<cnt; i++) {
195     char id[UUID_STR_LEN+1];
196     int rcnt, j;
197     char *requires;
198     noit_conf_section_t *reqs;
199
200     if(!noit_conf_get_stringbuf(statement_configs[i],
201                                 "self::node()/@id",
202                                 id, sizeof(id))) {
203       noitL(noit_iep, "No uuid specified in query\n");
204       continue;
205     }
206     if(!noit_hash_retrieve(&stmt_by_id, id, strlen(id), &vstmt)) {
207       noitL(noit_error, "Cannot find statement: %s\n", id);
208       exit(-1);
209     }
210     stmt = vstmt;
211     reqs = noit_conf_get_sections(statement_configs[i],
212                                   "self::node()/requires", &rcnt);
213     if(rcnt > 0) {
214       stmt->requires = malloc(rcnt * sizeof(*(stmt->requires)));
215       for(j=0; j<rcnt; j++) {
216         void *vrstmt;
217         if(!noit_conf_get_string(reqs[j], "self::node()",
218                                  &requires) || requires[0] == '\0') {
219           continue;
220         }
221         if(!noit_hash_retrieve(&stmt_by_provider, requires, strlen(requires),
222                                &vrstmt)) {
223           noitL(noit_error,
224                 "Statement %s requires %s which no one provides.\n",
225                 stmt->id, requires);
226           exit(-1);
227         }
228         stmt->requires[stmt->nrequires++] = vrstmt;
229       }
230     }
231   }
232
233   /* Phase 3: Recursive sweep and mark to detect cycles.
234      We're walking the graph backwards here from dependent to provider,
235      but a cycle is a cycle, so this validates the graph. */
236   while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
237     stmt = vstmt;
238     if(stmt_mark_dag(stmt, ++mgen) < 0) {
239       noitL(noit_error, "Statement %s has a cyclic requirement\n", stmt->id);
240       exit(-1);
241     }
242   }
243
244   /* Phase 4: clean the markings */
245   mgen = 0;
246   memset(&iter, 0, sizeof(iter));
247   while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
248     stmt = vstmt;
249     stmt->marked = 0;
250   }
251
252   /* Phase 5: do the load */
253   memset(&iter, 0, sizeof(iter));
254   while(noit_hash_next(&stmt_by_id, &iter, &key, &klen, &vstmt)) {
255     stmt = vstmt;
256     submit_statement_node(stmt);
257   }
258
259   noit_hash_destroy(&stmt_by_provider, NULL, NULL);
260   noit_hash_destroy(&stmt_by_id, NULL, statement_node_free);
261   free(statement_configs);
262 }
263
264 void stratcon_iep_submit_queries() {
265   int i, cnt = 0;
266   noit_conf_section_t *query_configs;
267   char path[256];
268
269   snprintf(path, sizeof(path), "/stratcon/iep/queries//query");
270   query_configs = noit_conf_get_sections(NULL, path, &cnt);
271   noitL(noit_debug, "Found %d %s stanzas\n", cnt, path);
272   for(i=0; i<cnt; i++) {
273     char id[UUID_STR_LEN+1];
274     char topic[256];
275     char *query;
276     char *line;
277     int line_len;
278
279     if(!noit_conf_get_stringbuf(query_configs[i],
280                                 "self::node()/@id",
281                                 id, sizeof(id))) {
282       noitL(noit_iep, "No uuid specified in query\n");
283       continue;
284     }
285     if(!noit_conf_get_stringbuf(query_configs[i],
286                                 "ancestor-or-self::node()/@topic",
287                                 topic, sizeof(topic))) {
288       noitL(noit_iep, "No topic specified in query\n");
289       continue;
290     }
291     if(!noit_conf_get_string(query_configs[i], "self::node()/epl",
292                              &query)) {
293       noitL(noit_iep, "No contents specified in query\n");
294       continue;
295     }
296     line_len = 4 /* 3 tabs + \0 */ +
297                1 /* 'Q' */ + 1 /* '\n' */ +
298                strlen(id) + strlen(topic) + strlen(query);
299     line = malloc(line_len);
300     snprintf(line, line_len, "Q\t%s\t%s\t%s\n", id, topic, query);
301     free(query);
302     query = line;
303     while(query[0] && query[1]) {
304       if(*query == '\n') *query = ' ';
305       query++;
306     }
307     stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, line, NULL);
308   }
309   free(query_configs);
310 }
311
312 static
313 struct iep_thread_driver *stratcon_iep_get_connection() {
314   int rc;
315   struct iep_thread_driver *driver;
316   driver = pthread_getspecific(iep_connection);
317   if(!driver) {
318     driver = mq_driver->allocate();
319     pthread_setspecific(iep_connection, driver);
320   }
321
322   rc = mq_driver->connect(driver);
323   if(rc < 0) return NULL;
324   if(rc == 0) {
325     /* Initial connect */
326     /* TODO: this should be requested by Esper, not blindly pushed */
327     stratcon_iep_submit_statements();
328     stratcon_datastore_iep_check_preload();
329     stratcon_iep_submit_queries();
330   }
331
332   return driver;
333 }
334
335 static int
336 setup_iep_connection_callback(eventer_t e, int mask, void *closure,
337                               struct timeval *now) {
338   noit_spinlock_unlock(&iep_conn_cnt);
339   stratcon_iep_line_processor(DS_OP_INSERT, NULL, NULL, NULL, NULL);
340   return 0;
341 }
342
343 static void
344 setup_iep_connection_later(int seconds) {
345   eventer_t newe;
346   if(!noit_spinlock_trylock(&iep_conn_cnt)) return;
347   newe = eventer_alloc();
348   gettimeofday(&newe->whence, NULL);
349   newe->whence.tv_sec += seconds;
350   newe->mask = EVENTER_TIMER;
351   newe->callback = setup_iep_connection_callback;
352   newe->closure = NULL;
353   eventer_add(newe);
354 }
355
356 static int
357 stratcon_iep_submitter(eventer_t e, int mask, void *closure,
358                        struct timeval *now) {
359   float age;
360   struct iep_job_closure *job = closure;
361   struct iep_thread_driver *driver;
362   /* We only play when it is an asynch event */
363   if(!(mask & EVENTER_ASYNCH_WORK)) return 0;
364
365   if(mask & EVENTER_ASYNCH_CLEANUP) {
366     /* free all the memory associated with the batch */
367     if(job) {
368       if(job->line) free(job->line);
369       if(job->remote) free(job->remote);
370       if(job->doc_str) free(job->doc_str);
371       free(job);
372     }
373     return 0;
374   }
375   driver = stratcon_iep_get_connection();
376   if(!driver) setup_iep_connection_later(1);
377
378   if(!job->line || job->line[0] == '\0') return 0;
379
380   if((age = stratcon_iep_age_from_line(job->line, *now)) > 60) {
381     noitL(noit_debug, "Skipping old event %f second old.\n", age);
382     return 0;
383   }
384   /* Submit */
385   if(driver) {
386     int line_len = strlen(job->line);
387     int remote_len = strlen(job->remote);
388
389     job->doc_str = (char*)calloc(line_len + 1 /* \t */ +
390         remote_len + 2, 1);
391     strncpy(job->doc_str, job->line, 2);
392     strncat(job->doc_str, job->remote, remote_len);
393     strncat(job->doc_str, "\t", 1);
394     strncat(job->doc_str, job->line + 2, line_len - 2);
395
396     /* Don't need to catch error here, next submit will catch it */
397     if(mq_driver->submit(driver, job->doc_str, line_len + remote_len + 1) != 0) {
398       noitL(noit_debug, "failed to MQ submit.\n");
399     }
400   }
401   else {
402     noitL(noit_iep, "no iep connection, skipping line: '%s'\n", job->line);
403   }
404   return 0;
405 }
406
407 void
408 stratcon_iep_line_processor(stratcon_datastore_op_t op,
409                             struct sockaddr *remote, const char *remote_cn,
410                             void *operand, eventer_t completion) {
411   int len;
412   char remote_str[128];
413   struct iep_job_closure *jc;
414   eventer_t newe;
415   struct timeval __now, iep_timeout = { 20L, 0L };
416   /* We only care about inserts */
417
418   if(op == DS_OP_CHKPT) {
419     eventer_add(completion);
420     return;
421   }
422   if(op != DS_OP_INSERT) return;
423
424   snprintf(remote_str, sizeof(remote_str), "%s", "0.0.0.0");
425   if(remote) {
426     switch(remote->sa_family) {
427       case AF_INET:
428         len = sizeof(struct sockaddr_in);
429         inet_ntop(remote->sa_family, &((struct sockaddr_in *)remote)->sin_addr,
430                   remote_str, len);
431         break;
432       case AF_INET6:
433        len = sizeof(struct sockaddr_in6);
434         inet_ntop(remote->sa_family, &((struct sockaddr_in6 *)remote)->sin6_addr,
435                   remote_str, len);
436        break;
437       case AF_UNIX:
438         len = SUN_LEN(((struct sockaddr_un *)remote));
439         snprintf(remote_str, sizeof(remote_str), "%s", ((struct sockaddr_un *)remote)->sun_path);
440         break;
441     }
442   }
443
444   /* process operand and push onto queue */
445   gettimeofday(&__now, NULL);
446   newe = eventer_alloc();
447   newe->mask = EVENTER_ASYNCH;
448   add_timeval(__now, iep_timeout, &newe->whence);
449   newe->callback = stratcon_iep_submitter;
450   jc = calloc(1, sizeof(*jc));
451   jc->line = operand;
452   jc->remote = strdup(remote_str);
453   newe->closure = jc;
454
455   eventer_add_asynch(&iep_jobq, newe);
456 }
457
458 static void connection_destroy(void *vd) {
459   struct iep_thread_driver *driver = vd;
460   mq_driver->disconnect(driver);
461   mq_driver->deallocate(driver);
462 }
463
464 jlog_streamer_ctx_t *
465 stratcon_jlog_streamer_iep_ctx_alloc(void) {
466   jlog_streamer_ctx_t *ctx;
467   ctx = stratcon_jlog_streamer_ctx_alloc();
468   ctx->jlog_feed_cmd = htonl(NOIT_JLOG_DATA_TEMP_FEED);
469   ctx->push = stratcon_iep_line_processor;
470   return ctx;
471 }
472
473 struct iep_daemon_info {
474   pid_t child;
475   int stdin_pipe[2];
476   int stderr_pipe[2];
477   char *directory;
478   char *command;
479 };
480
481 static void
482 iep_daemon_info_free(struct iep_daemon_info *info) {
483   if(!info) return;
484   if(info->directory) free(info->directory);
485   if(info->command) free(info->command);
486   if(info->stdin_pipe[0] >= 0) close(info->stdin_pipe[0]);
487   if(info->stdin_pipe[1] >= 0) close(info->stdin_pipe[1]);
488   if(info->stderr_pipe[0] >= 0) close(info->stderr_pipe[0]);
489   if(info->stderr_pipe[1] >= 0) close(info->stderr_pipe[1]);
490   free(info);
491 }
492
493 static int
494 stratcon_iep_err_handler(eventer_t e, int mask, void *closure,
495                          struct timeval *now) {
496   int len, newmask;
497   char buff[4096];
498   struct iep_daemon_info *info = (struct iep_daemon_info *)closure;
499
500   if(mask & EVENTER_EXCEPTION) {
501     int rv;
502    read_error:
503     kill(info->child, SIGKILL);
504     if(waitpid(info->child, &rv, 0) != info->child) {
505       noitL(noit_error, "Failed to reap IEP daemon\n");
506       exit(-1);
507     }
508     noitL(noit_error, "IEP daemon is done, starting a new one\n");
509     start_iep_daemon();
510     eventer_remove_fd(e->fd);
511     iep_daemon_info_free(info);
512     return 0;
513   }
514   while(1) {
515     len = e->opset->read(e->fd, buff, sizeof(buff)-1, &newmask, e);
516     if(len == -1 && (errno == EAGAIN || errno == EINTR))
517       return newmask | EVENTER_EXCEPTION;
518     if(len <= 0) goto read_error;
519     assert(len < sizeof(buff));
520     buff[len] = '\0';
521     noitL(noit_iep, "%s", buff);
522   }
523 }
524
525 static void
526 start_iep_daemon() {
527   eventer_t newe;
528   struct iep_daemon_info *info;
529
530   info = calloc(1, sizeof(*info));
531   info->stdin_pipe[0] = info->stdin_pipe[1] = -1;
532   info->stderr_pipe[0] = info->stderr_pipe[1] = -1;
533
534   if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@directory",
535                            &info->directory))
536     info->directory = strdup(".");
537   if(!noit_conf_get_string(NULL, "/stratcon/iep/start/@command",
538                            &info->command)) {
539     noitL(noit_error, "No IEP start command provided.  You're on your own.\n");
540     setup_iep_connection_later(0);
541     return;
542   }
543   if(pipe(info->stdin_pipe) != 0 ||
544      pipe(info->stderr_pipe) != 0) {
545     noitL(noit_error, "pipe: %s\n", strerror(errno));
546     goto bail;
547   }
548   info->child = fork();
549   if(info->child == -1) {
550     noitL(noit_error, "fork: %s\n", strerror(errno));
551     goto bail;
552   }
553   if(info->child == 0) {
554     char *argv[3] = { "run-iep", NULL, NULL };
555     int stdout_fileno;
556
557     argv[1] = noit_conf_config_filename();
558
559     if(chdir(info->directory) != 0) {
560       noitL(noit_error, "Starting IEP daemon, chdir failed: %s\n",
561             strerror(errno));
562       exit(-1);
563     }
564
565     close(info->stdin_pipe[1]);
566     close(info->stderr_pipe[0]);
567     dup2(info->stdin_pipe[0], 0);
568     dup2(info->stderr_pipe[1], 2);
569     stdout_fileno = open("/dev/null", O_WRONLY);
570     dup2(stdout_fileno, 1);
571
572     exit(execv(info->command, argv));
573   }
574   /* in the parent */
575   close(info->stdin_pipe[0]);
576   info->stdin_pipe[0] = -1;
577   close(info->stderr_pipe[1]);
578   info->stderr_pipe[1] = -1;
579   if(eventer_set_fd_nonblocking(info->stderr_pipe[0])) {
580     goto bail;
581   }
582
583   newe = eventer_alloc();
584   newe->fd = info->stderr_pipe[0];
585   newe->mask = EVENTER_READ | EVENTER_EXCEPTION;
586   newe->callback = stratcon_iep_err_handler;
587   newe->closure = info;
588   eventer_add(newe);
589   info = NULL;
590
591   setup_iep_connection_later(1);
592
593   return;
594
595  bail:
596   if(info) {
597     iep_daemon_info_free(info);
598   }
599   noitL(noit_error, "Failed to start IEP daemon\n");
600   exit(-1);
601   return;
602 }
603
604 void
605 stratcon_iep_mq_driver_register(const char *name, mq_driver_t *d) {
606   noit_hash_replace(&mq_drivers, strdup(name), strlen(name), d, free, NULL);
607 }
608
609 void
610 stratcon_iep_init() {
611   noit_boolean disabled = noit_false;
612   char mq_type[128] = "stomp";
613   void *vdriver;
614
615   if(noit_conf_get_boolean(NULL, "/stratcon/iep/@disabled", &disabled) &&
616      disabled == noit_true) {
617     noitL(noit_error, "IEP system is disabled!\n");
618     return;
619   }
620
621   if(!noit_conf_get_stringbuf(NULL, "/stratcon/iep/mq/@type",
622                               mq_type, sizeof(mq_type))) {
623     noitL(noit_error, "You must specify an <mq type=\"...\"> that is valid.\n");
624     exit(-2);
625   }
626   if(!noit_hash_retrieve(&mq_drivers, mq_type, strlen(mq_type), &vdriver) ||
627      vdriver == NULL) {
628     noitL(noit_error, "Cannot find MQ driver type: %s\n", mq_type);
629     noitL(noit_error, "Did you forget to load a module?\n");
630     exit(-2);
631   }
632   mq_driver = (mq_driver_t *)vdriver;
633
634   noit_iep = noit_log_stream_find("error/iep");
635   if(!noit_iep) noit_iep = noit_error;
636
637   eventer_name_callback("stratcon_iep_submitter", stratcon_iep_submitter);
638   eventer_name_callback("setup_iep_connection_callback", setup_iep_connection_callback);
639   pthread_key_create(&iep_connection, connection_destroy);
640
641   /* start up a thread pool of one */
642   memset(&iep_jobq, 0, sizeof(iep_jobq));
643   eventer_jobq_init(&iep_jobq, "iep_submitter");
644   iep_jobq.backq = eventer_default_backq();
645   eventer_jobq_increase_concurrency(&iep_jobq);
646
647   start_iep_daemon();
648
649   /* setup our live jlog stream */
650   stratcon_streamer_connection(NULL, NULL,
651                                stratcon_jlog_recv_handler,
652                                (void *(*)())stratcon_jlog_streamer_iep_ctx_alloc,
653                                NULL,
654                                jlog_streamer_ctx_free);
655 }
656
Note: See TracBrowser for help on using the browser.