root/src/stratcon_iep.c

Revision 0ba16060522984b6f0f311e3b6da4e135ea15433, 20.0 kB (checked in by Theo Schlossnagle <jesus@omniti.com>, 3 years ago)

a varity of strat aliasing problems and avoiding odd includes on Linux

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