root/src/stratcon_iep.c

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

fixes #219

This is "significant" as it requires adding a module section to
stratcon.conf and not using the <stomp> stanza, but instead using
<mq type="stomp">. I've tested it with ActiveMQ and RabbitMQ and
both work fine.

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