Skip to content

T
trackrdrd
Commit d4d73dfe authored by Geoff Simmons's avatar Geoff Simmons
Browse files
Options

trackrdrd: unified indentation style, some code cleanup

parent e2163244
Hide whitespace changes
Inline Side-by-side
Showing with 345 additions and 370 deletions
trackrdrd/src/child.c
View file @ d4d73dfe
...@@ -95,7 +95,7 @@ struct hashentry_s { ...@@ -95,7 +95,7 @@ struct hashentry_s {
float insert_time; float insert_time;
VTAILQ_ENTRY(hashentry_s) insert_list; VTAILQ_ENTRY(hashentry_s) insert_list;
dataentry *de; dataentry *de;
}; };
typedef struct hashentry_s hashentry; typedef struct hashentry_s hashentry;
...@@ -144,7 +144,7 @@ typedef struct hashtable_s hashtable; ...@@ -144,7 +144,7 @@ typedef struct hashtable_s hashtable;
static hashtable htbl; static hashtable htbl;
#ifdef WITHOUT_ASSERTS #ifdef WITHOUT_ASSERTS
#define entry_assert(e, cond) do { (void)(e);(void)(cond);} while(0) #define entry_assert(e, cond) do { (void)(e);(void)(cond);} while(0)
#else /* WITH_ASSERTS */ #else /* WITH_ASSERTS */
#define entry_assert(e, cond) \ #define entry_assert(e, cond) \
do { \ do { \
...@@ -317,50 +317,50 @@ jenkmulvey2(uint32_t n) ...@@ -317,50 +317,50 @@ jenkmulvey2(uint32_t n)
static uint32_t static uint32_t
wang(uint32_t n) wang(uint32_t n)
{ {
n = ~n + (n << 15); // n = (n << 15) - n - 1; n = ~n + (n << 15); // n = (n << 15) - n - 1;
n ^= rotr(n,12); n ^= rotr(n,12);
n += (n << 2); n += (n << 2);
n ^= rotr(n,4); n ^= rotr(n,4);
n = (n + (n << 3)) + (n << 11); n = (n + (n << 3)) + (n << 11);
n ^= rotr(n,16); n ^= rotr(n,16);
return n; return n;
} }
void void
HASH_Stats(void) HASH_Stats(void)
{ {
LOG_Log(LOG_INFO, LOG_Log(LOG_INFO,
"Hash table: " "Hash table: "
"len=%u " "len=%u "
"seen=%u " "seen=%u "
"drop_reqstart=%u " "drop_reqstart=%u "
"drop_vcl_log=%u " "drop_vcl_log=%u "
"drop_reqend=%u " "drop_reqend=%u "
"expired=%u " "expired=%u "
"evacuated=%u " "evacuated=%u "
"open=%u " "open=%u "
"load=%.2f " "load=%.2f "
"collisions=%u " "collisions=%u "
"insert_probes=%u " "insert_probes=%u "
"find_probes=%u " "find_probes=%u "
"fail=%u " "fail=%u "
"occ_hi=%u " "occ_hi=%u "
"occ_hi_this=%u ", "occ_hi_this=%u ",
htbl.len, htbl.len,
htbl.seen, htbl.seen,
htbl.drop_reqstart, htbl.drop_reqstart,
htbl.drop_vcl_log, htbl.drop_vcl_log,
htbl.drop_reqend, htbl.drop_reqend,
htbl.expired, htbl.expired,
htbl.evacuated, htbl.evacuated,
htbl.open, htbl.open,
100.0 * htbl.open / htbl.len, 100.0 * htbl.open / htbl.len,
htbl.collisions, htbl.collisions,
htbl.insert_probes, htbl.insert_probes,
htbl.find_probes, htbl.find_probes,
htbl.fail, htbl.fail,
htbl.occ_hi, htbl.occ_hi,
htbl.occ_hi_this); htbl.occ_hi_this);
htbl.occ_hi_this = 0; htbl.occ_hi_this = 0;
} }
...@@ -400,7 +400,7 @@ hash_init(void) ...@@ -400,7 +400,7 @@ hash_init(void)
/* entries init */ /* entries init */
for (int i = 0; i < entries; i++) { for (int i = 0; i < entries; i++) {
htbl.entry[i].magic = HASH_MAGIC; htbl.entry[i].magic = HASH_MAGIC;
htbl.entry[i].state = HASH_EMPTY; htbl.entry[i].state = HASH_EMPTY;
} }
atexit(hash_cleanup); atexit(hash_cleanup);
...@@ -500,7 +500,7 @@ static hashentry ...@@ -500,7 +500,7 @@ static hashentry
he = &htbl.entry[INDEX(h)]; he = &htbl.entry[INDEX(h)];
if (he->state == HASH_EMPTY) if (he->state == HASH_EMPTY)
goto ok; goto ok;
htbl.collisions++; htbl.collisions++;
oldest = he; oldest = he;
...@@ -618,7 +618,7 @@ static inline dataentry ...@@ -618,7 +618,7 @@ static inline dataentry
sprintf(de->data, "XID=%d", xid); sprintf(de->data, "XID=%d", xid);
de->end = strlen(de->data); de->end = strlen(de->data);
if (de->end > dtbl.w_stats.data_hi) if (de->end > dtbl.w_stats.data_hi)
dtbl.w_stats.data_hi = de->end; dtbl.w_stats.data_hi = de->end;
MON_StatsUpdate(STATS_OCCUPANCY); MON_StatsUpdate(STATS_OCCUPANCY);
return (de); return (de);
...@@ -714,21 +714,21 @@ OSL_Track(void *priv, enum VSL_tag_e tag, unsigned fd, unsigned len, ...@@ -714,21 +714,21 @@ OSL_Track(void *priv, enum VSL_tag_e tag, unsigned fd, unsigned len,
AZ(err); AZ(err);
LOG_Log(LOG_DEBUG, "%s: XID=%u", VSL_tags[tag], xid); LOG_Log(LOG_DEBUG, "%s: XID=%u", VSL_tags[tag], xid);
if (xid > last_start_xid) if (xid > last_start_xid)
last_start_xid = xid; last_start_xid = xid;
tim = TIM_mono(); tim = TIM_mono();
if (! insert(xid, fd, tim)) { if (! insert(xid, fd, tim)) {
htbl.drop_reqstart++; htbl.drop_reqstart++;
break; break;
} }
/* configurable ? */ /* configurable ? */
if ((tim - tim_exp_check) > 10) { if ((tim - tim_exp_check) > 10) {
hash_exp(tim - htbl.ttl); hash_exp(tim - htbl.ttl);
tim_exp_check = tim; tim_exp_check = tim;
} }
break; break;
case SLT_VCL_Log: case SLT_VCL_Log:
/* Skip VCL_Log entries without the "track " prefix. */ /* Skip VCL_Log entries without the "track " prefix. */
...@@ -741,15 +741,15 @@ OSL_Track(void *priv, enum VSL_tag_e tag, unsigned fd, unsigned len, ...@@ -741,15 +741,15 @@ OSL_Track(void *priv, enum VSL_tag_e tag, unsigned fd, unsigned len,
LOG_Log(LOG_DEBUG, "%s: XID=%u, data=[%.*s]", VSL_tags[tag], LOG_Log(LOG_DEBUG, "%s: XID=%u, data=[%.*s]", VSL_tags[tag],
xid, datalen, data); xid, datalen, data);
he = hash_find(xid); he = hash_find(xid);
if (! he) { if (! he) {
LOG_Log(LOG_WARNING, "%s: XID %d not found", LOG_Log(LOG_WARNING, "%s: XID %d not found",
VSL_tags[tag], xid); VSL_tags[tag], xid);
htbl.drop_vcl_log++; htbl.drop_vcl_log++;
break; break;
} }
check_entry(he, xid, fd); check_entry(he, xid, fd);
de = he->de; de = he->de;
append(de, tag, xid, data, datalen); append(de, tag, xid, data, datalen);
de->hasdata = true; de->hasdata = true;
break; break;
...@@ -761,21 +761,21 @@ OSL_Track(void *priv, enum VSL_tag_e tag, unsigned fd, unsigned len, ...@@ -761,21 +761,21 @@ OSL_Track(void *priv, enum VSL_tag_e tag, unsigned fd, unsigned len,
LOG_Log(LOG_DEBUG, "%s: XID=%u req_endt=%u.%09lu", VSL_tags[tag], xid, LOG_Log(LOG_DEBUG, "%s: XID=%u req_endt=%u.%09lu", VSL_tags[tag], xid,
(unsigned) reqend_t.tv_sec, reqend_t.tv_nsec); (unsigned) reqend_t.tv_sec, reqend_t.tv_nsec);
if (xid > last_end_xid) if (xid > last_end_xid)
last_end_xid = xid; last_end_xid = xid;
xid_spread_sum += (last_end_xid - last_start_xid); xid_spread_sum += (last_end_xid - last_start_xid);
xid_spread_count++; xid_spread_count++;
he = hash_find(xid); he = hash_find(xid);
if (! he) { if (! he) {
LOG_Log(LOG_WARNING, "%s: XID %d not found", LOG_Log(LOG_WARNING, "%s: XID %d not found",
VSL_tags[tag], xid); VSL_tags[tag], xid);
htbl.drop_reqend++; htbl.drop_reqend++;
break; break;
} }
check_entry(he, xid, fd); check_entry(he, xid, fd);
de = he->de; de = he->de;
sprintf(reqend_str, "%s=%u.%09lu", REQEND_T_VAR, sprintf(reqend_str, "%s=%u.%09lu", REQEND_T_VAR,
(unsigned) reqend_t.tv_sec, reqend_t.tv_nsec); (unsigned) reqend_t.tv_sec, reqend_t.tv_nsec);
...@@ -965,7 +965,6 @@ CHILD_Main(struct VSM_data *vd, int endless, int readconfig) ...@@ -965,7 +965,6 @@ CHILD_Main(struct VSM_data *vd, int endless, int readconfig)
/* Main loop */ /* Main loop */
term = 0; term = 0;
/* XXX: Varnish restart? */
/* XXX: TERM not noticed until request received */ /* XXX: TERM not noticed until request received */
while (VSL_Dispatch(vd, OSL_Track, NULL) > 0) while (VSL_Dispatch(vd, OSL_Track, NULL) > 0)
if (term || !endless) if (term || !endless)
...@@ -984,7 +983,7 @@ CHILD_Main(struct VSM_data *vd, int endless, int readconfig) ...@@ -984,7 +983,7 @@ CHILD_Main(struct VSM_data *vd, int endless, int readconfig)
WRK_Shutdown(); WRK_Shutdown();
AZ(MQ_GlobalShutdown()); AZ(MQ_GlobalShutdown());
if (config.monitor_interval > 0.0) if (config.monitor_interval > 0.0)
MON_StatusShutdown(monitor); MON_StatusShutdown(monitor);
LOG_Log0(LOG_INFO, "Worker process exiting"); LOG_Log0(LOG_INFO, "Worker process exiting");
LOG_Close(); LOG_Close();
exit(EXIT_SUCCESS); exit(EXIT_SUCCESS);
......
trackrdrd/src/data.c
View file @ d4d73dfe
...@@ -83,12 +83,12 @@ DATA_Init(void) ...@@ -83,12 +83,12 @@ DATA_Init(void)
dtbl.nfree = 0; dtbl.nfree = 0;
for (int i = 0; i < entries; i++) { for (int i = 0; i < entries; i++) {
dtbl.entry[i].magic = DATA_MAGIC; dtbl.entry[i].magic = DATA_MAGIC;
dtbl.entry[i].state = DATA_EMPTY; dtbl.entry[i].state = DATA_EMPTY;
dtbl.entry[i].hasdata = false; dtbl.entry[i].hasdata = false;
dtbl.entry[i].data = &dtbl.buf[i * bufsize]; dtbl.entry[i].data = &dtbl.buf[i * bufsize];
VSTAILQ_INSERT_TAIL(&dtbl.freehead, &dtbl.entry[i], freelist); VSTAILQ_INSERT_TAIL(&dtbl.freehead, &dtbl.entry[i], freelist);
dtbl.nfree++; dtbl.nfree++;
} }
assert(dtbl.nfree == entries); assert(dtbl.nfree == entries);
assert(VSTAILQ_FIRST(&dtbl.freehead)); assert(VSTAILQ_FIRST(&dtbl.freehead));
...@@ -105,10 +105,10 @@ DATA_Init(void) ...@@ -105,10 +105,10 @@ DATA_Init(void)
void void
DATA_Take_Freelist(struct freehead_s *dst) DATA_Take_Freelist(struct freehead_s *dst)
{ {
AZ(pthread_mutex_lock(&dtbl.freelist_lock)); AZ(pthread_mutex_lock(&dtbl.freelist_lock));
VSTAILQ_CONCAT(dst, &dtbl.freehead); VSTAILQ_CONCAT(dst, &dtbl.freehead);
dtbl.nfree = 0; dtbl.nfree = 0;
AZ(pthread_mutex_unlock(&dtbl.freelist_lock)); AZ(pthread_mutex_unlock(&dtbl.freelist_lock));
} }
/* /*
...@@ -119,25 +119,25 @@ DATA_Take_Freelist(struct freehead_s *dst) ...@@ -119,25 +119,25 @@ DATA_Take_Freelist(struct freehead_s *dst)
void void
DATA_Return_Freelist(struct freehead_s *returned, unsigned nreturned) DATA_Return_Freelist(struct freehead_s *returned, unsigned nreturned)
{ {
AZ(pthread_mutex_lock(&dtbl.freelist_lock)); AZ(pthread_mutex_lock(&dtbl.freelist_lock));
VSTAILQ_CONCAT(&dtbl.freehead, returned); VSTAILQ_CONCAT(&dtbl.freehead, returned);
dtbl.nfree += nreturned; dtbl.nfree += nreturned;
AZ(pthread_mutex_unlock(&dtbl.freelist_lock)); AZ(pthread_mutex_unlock(&dtbl.freelist_lock));
} }
void void
DATA_Dump1(dataentry *entry, int i) DATA_Dump1(dataentry *entry, int i)
{ {
if (entry->state == DATA_EMPTY) if (entry->state == DATA_EMPTY)
return; return;
LOG_Log(LOG_INFO, "Data entry %d: XID=%d tid=%d state=%s data=[%.*s]", LOG_Log(LOG_INFO, "Data entry %d: XID=%d tid=%d state=%s data=[%.*s]",
i, entry->xid, entry->tid, statename[entry->state], entry->end, i, entry->xid, entry->tid, statename[entry->state], entry->end,
entry->data); entry->data);
} }
void void
DATA_Dump(void) DATA_Dump(void)
{ {
for (int i = 0; i < dtbl.len; i++) for (int i = 0; i < dtbl.len; i++)
DATA_Dump1(&dtbl.entry[i], i); DATA_Dump1(&dtbl.entry[i], i);
} }
trackrdrd/src/handler.c
View file @ d4d73dfe
...@@ -73,17 +73,17 @@ stacktrace(void) ...@@ -73,17 +73,17 @@ stacktrace(void)
depth = backtrace (buf, MAX_STACK_DEPTH); depth = backtrace (buf, MAX_STACK_DEPTH);
if (depth == 0) { if (depth == 0) {
LOG_Log0(LOG_ERR, "Stacktrace empty"); LOG_Log0(LOG_ERR, "Stacktrace empty");
return; return;
} }
strings = backtrace_symbols(buf, depth); strings = backtrace_symbols(buf, depth);
if (strings == NULL) { if (strings == NULL) {
LOG_Log0(LOG_ERR, "Cannot retrieve symbols for stacktrace"); LOG_Log0(LOG_ERR, "Cannot retrieve symbols for stacktrace");
return; return;
} }
/* XXX: get symbol names from nm? cf. cache_panic.c/pan_backtrace */ /* XXX: get symbol names from nm? cf. cache_panic.c/pan_backtrace */
for (i = 0; i < depth; i++) for (i = 0; i < depth; i++)
LOG_Log(LOG_ERR, "%s", strings[i]); LOG_Log(LOG_ERR, "%s", strings[i]);
free(strings); free(strings);
} }
...@@ -92,7 +92,7 @@ void ...@@ -92,7 +92,7 @@ void
HNDL_Abort(int sig) HNDL_Abort(int sig)
{ {
LOG_Log(LOG_ALERT, "Received signal %d (%s), stacktrace follows", sig, LOG_Log(LOG_ALERT, "Received signal %d (%s), stacktrace follows", sig,
strsignal(sig)); strsignal(sig));
stacktrace(); stacktrace();
AZ(sigaction(SIGABRT, &default_action, NULL)); AZ(sigaction(SIGABRT, &default_action, NULL));
LOG_Log0(LOG_ALERT, "Aborting"); LOG_Log0(LOG_ALERT, "Aborting");
......
trackrdrd/src/monitor.c
View file @ d4d73dfe
...@@ -51,38 +51,38 @@ log_output(void) ...@@ -51,38 +51,38 @@ log_output(void)
LOG_Log(LOG_INFO, LOG_Log(LOG_INFO,
"Data table writer: " "Data table writer: "
"len=%u " "len=%u "
"nodata=%u " "nodata=%u "
"submitted=%u " "submitted=%u "
"wait_qfull=%u " "wait_qfull=%u "
"wait_room=%u " "wait_room=%u "
"data_hi=%u " "data_hi=%u "
"data_overflows=%u ", "data_overflows=%u ",
dtbl.len, dtbl.len,
dtbl.w_stats.nodata, dtbl.w_stats.nodata,
dtbl.w_stats.submitted, dtbl.w_stats.submitted,
dtbl.w_stats.wait_qfull, dtbl.w_stats.wait_qfull,
dtbl.w_stats.wait_room, dtbl.w_stats.wait_room,
dtbl.w_stats.data_hi, dtbl.w_stats.data_hi,
dtbl.w_stats.data_overflows); dtbl.w_stats.data_overflows);
LOG_Log(LOG_INFO, LOG_Log(LOG_INFO,
"Data table reader: " "Data table reader: "
"done=%u " "done=%u "
"open=%u " "open=%u "
"load=%.2f " "load=%.2f "
"sent=%u " "sent=%u "
"failed=%u " "failed=%u "
"occ_hi=%u " "occ_hi=%u "
"occ_hi_this=%u ", "occ_hi_this=%u ",
dtbl.r_stats.done, dtbl.r_stats.done,
dtbl.r_stats.open, dtbl.r_stats.open,
(100.0 * (1.0 * dtbl.r_stats.done + 1.0 * dtbl.r_stats.open) / dtbl.len), (100.0 * (1.0 * dtbl.r_stats.done + 1.0 * dtbl.r_stats.open) / dtbl.len),
dtbl.r_stats.sent, dtbl.r_stats.sent,
dtbl.r_stats.failed, dtbl.r_stats.failed,
dtbl.r_stats.occ_hi, dtbl.r_stats.occ_hi,
dtbl.r_stats.occ_hi_this dtbl.r_stats.occ_hi_this
); );
/* locking would be overkill */ /* locking would be overkill */
dtbl.r_stats.occ_hi_this = 0; dtbl.r_stats.occ_hi_this = 0;
...@@ -118,7 +118,7 @@ void ...@@ -118,7 +118,7 @@ void
if (nanosleep(&t, NULL) != 0) { if (nanosleep(&t, NULL) != 0) {
if (errno == EINTR) { if (errno == EINTR) {
if (run == 0) if (run == 0)
break; break;
LOG_Log0(LOG_INFO, "Monitoring thread interrupted"); LOG_Log0(LOG_INFO, "Monitoring thread interrupted");
continue; continue;
} }
...@@ -182,7 +182,7 @@ MON_StatsUpdate(stats_update_t update) ...@@ -182,7 +182,7 @@ MON_StatsUpdate(stats_update_t update)
break; break;
case STATS_NODATA: case STATS_NODATA:
dtbl.w_stats.nodata++; dtbl.w_stats.nodata++;
dtbl.r_stats.done--; dtbl.r_stats.done--;
break; break;
......
trackrdrd/src/spmcq.c
View file @ d4d73dfe
...@@ -50,26 +50,6 @@ spmcq_len(void) ...@@ -50,26 +50,6 @@ spmcq_len(void)
return UINT_MAX - spmcq.head + 1 + spmcq.tail; return UINT_MAX - spmcq.head + 1 + spmcq.tail;
} }
#if 0
/*
* this is only approximately correct and could even become negative when values
* get updated while we read them!
*
*/
int SPMCQ_Len(void) {
unsigned l;
do {
l = spmcq_len();
if (l <= spmcq.mask + 1)
break;
VRMB();
} while (1);
return (l);
}
#endif
static void static void
spmcq_cleanup(void) spmcq_cleanup(void)
{ {
......
trackrdrd/src/test/test_mq.c
View file @ d4d73dfe
...@@ -90,7 +90,7 @@ static const char ...@@ -90,7 +90,7 @@ static const char
{ {
const char *err; const char *err;
printf("... test worker init (including connect to ActiveMQ)\n"); printf("... test worker init\n");
err = MQ_WorkerInit(&worker); err = MQ_WorkerInit(&worker);
sprintf(errmsg, "MQ_WorkerInit: %s", err); sprintf(errmsg, "MQ_WorkerInit: %s", err);
......
trackrdrd/src/test/test_spmcq.c
View file @ d4d73dfe
...@@ -117,32 +117,32 @@ static void ...@@ -117,32 +117,32 @@ static void
unsigned *xid; unsigned *xid;
while (run) { while (run) {
/* run may be stale at this point */ /* run may be stale at this point */
debug_print("Consumer %d: attempt dequeue\n", id); debug_print("Consumer %d: attempt dequeue\n", id);
xid = (unsigned *) SPMCQ_Deq(); xid = (unsigned *) SPMCQ_Deq();
if (xid == NULL) { if (xid == NULL) {
/* grab the CV lock, which also constitutes an implicit memory /* grab the CV lock, which also constitutes an implicit memory
barrier */ barrier */
debug_print("Consumer %d: mutex\n", id); debug_print("Consumer %d: mutex\n", id);
if (pthread_mutex_lock(&spmcq_datawaiter_lock) != 0) if (pthread_mutex_lock(&spmcq_datawaiter_lock) != 0)
consumer_exit(pcdata, CONSUMER_MUTEX); consumer_exit(pcdata, CONSUMER_MUTEX);
/* run is guaranteed to be fresh here */ /* run is guaranteed to be fresh here */
if (run) { if (run) {
debug_print("Consumer %d: wait, run = %d\n", id, run); debug_print("Consumer %d: wait, run = %d\n", id, run);
if (pthread_cond_wait(&spmcq_datawaiter_cond, if (pthread_cond_wait(&spmcq_datawaiter_cond,
&spmcq_datawaiter_lock) != 0) &spmcq_datawaiter_lock) != 0)
consumer_exit(pcdata, CONSUMER_WAIT); consumer_exit(pcdata, CONSUMER_WAIT);
} }
debug_print("Consumer %d: unlock\n", id); debug_print("Consumer %d: unlock\n", id);
if (pthread_mutex_unlock(&spmcq_datawaiter_lock) != 0) if (pthread_mutex_unlock(&spmcq_datawaiter_lock) != 0)
consumer_exit(pcdata, CONSUMER_MUTEX); consumer_exit(pcdata, CONSUMER_MUTEX);
if (! run) { if (! run) {
debug_print("Consumer %d: quit signaled, run = %d\n", id, run); debug_print("Consumer %d: quit signaled, run = %d\n", id, run);
break; break;
} }
} else { } else {
/* xid != NULL */ /* xid != NULL */
debug_print("Consumer %d: dequeue %d (xid = %u)\n", id, ++deqs, debug_print("Consumer %d: dequeue %d (xid = %u)\n", id, ++deqs,
*xid); *xid);
pcdata->sum += *xid; pcdata->sum += *xid;
} }
......
trackrdrd/src/trackrdrd.c
View file @ d4d73dfe
...@@ -184,7 +184,7 @@ parent_main(pid_t child_pid, struct VSM_data *vd, int endless) ...@@ -184,7 +184,7 @@ parent_main(pid_t child_pid, struct VSM_data *vd, int endless)
wpid, WEXITSTATUS(status)); wpid, WEXITSTATUS(status));
if (WIFSIGNALED(status)) if (WIFSIGNALED(status))
LOG_Log(LOG_WARNING, LOG_Log(LOG_WARNING,
"Worker process %d exited due to signal %d (%s)", "Worker process %d exited due to signal %d (%s)",
wpid, WTERMSIG(status), strsignal(WTERMSIG(status))); wpid, WTERMSIG(status), strsignal(WTERMSIG(status)));
if (wpid != child_pid) if (wpid != child_pid)
...@@ -210,169 +210,169 @@ usage(int status) ...@@ -210,169 +210,169 @@ usage(int status)
int int
main(int argc, char * const *argv) main(int argc, char * const *argv)
{ {
int c, d_flag = 0, D_flag = 0, endless = 1, err; int c, d_flag = 0, D_flag = 0, endless = 1, err;
const char *P_arg = NULL, *l_arg = NULL, *n_arg = NULL, *f_arg = NULL, const char *P_arg = NULL, *l_arg = NULL, *n_arg = NULL, *f_arg = NULL,
*y_arg = NULL, *c_arg = NULL, *u_arg = NULL; *y_arg = NULL, *c_arg = NULL, *u_arg = NULL;
struct VSM_data *vd; struct VSM_data *vd;
pid_t child_pid; pid_t child_pid;
vd = VSM_New(); vd = VSM_New();
VSL_Setup(vd); VSL_Setup(vd);
CONF_Init(); CONF_Init();
if ((err = CONF_ReadDefault()) != 0) { if ((err = CONF_ReadDefault()) != 0) {
if (err != -1) if (err != -1)
LOG_Log(LOG_ALERT, "Cannot read %s: %s", DEFAULT_CONFIG, LOG_Log(LOG_ALERT, "Cannot read %s: %s", DEFAULT_CONFIG,
strerror(err)); strerror(err));
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
cli_config_filename[0] = '\0'; cli_config_filename[0] = '\0';
while ((c = getopt(argc, argv, "u:P:Vn:hl:df:y:c:D")) != -1) { while ((c = getopt(argc, argv, "u:P:Vn:hl:df:y:c:D")) != -1) {
switch (c) { switch (c) {
case 'P': case 'P':
P_arg = optarg; P_arg = optarg;
break; break;
case 'V': case 'V':
printf(PACKAGE_STRING " revision " REVISION "\n"); printf(PACKAGE_STRING " revision " REVISION "\n");
exit(EXIT_SUCCESS); exit(EXIT_SUCCESS);
case 'n': case 'n':
n_arg = optarg; n_arg = optarg;
break; break;
case 'l': case 'l':
l_arg = optarg; l_arg = optarg;
break; break;
case 'd': case 'd':
d_flag = 1; d_flag = 1;
break; break;
case 'f': case 'f':
f_arg = optarg; f_arg = optarg;
break; break;
case 'y': case 'y':
y_arg = optarg; y_arg = optarg;
break; break;
case 'c': case 'c':
c_arg = optarg; c_arg = optarg;
break; break;
case 'D': case 'D':
D_flag = 1; D_flag = 1;
break; break;
case 'u': case 'u':
u_arg = optarg; u_arg = optarg;
break; break;
case 'h': case 'h':
usage(EXIT_SUCCESS); usage(EXIT_SUCCESS);
default: default:
usage(EXIT_FAILURE);
}
}
if ((argc - optind) > 0)
usage(EXIT_FAILURE); usage(EXIT_FAILURE);
if (c_arg) {
strcpy(cli_config_filename, c_arg);
printf("Reading config from %s\n", c_arg);
if (CONF_ReadFile(c_arg) != 0)
exit(EXIT_FAILURE);
} }
}
if ((argc - optind) > 0)
usage(EXIT_FAILURE);
if (c_arg) {
strcpy(cli_config_filename, c_arg);
printf("Reading config from %s\n", c_arg);
if (CONF_ReadFile(c_arg) != 0)
exit(EXIT_FAILURE);
}
if (f_arg && n_arg) if (f_arg && n_arg)
usage(EXIT_FAILURE); usage(EXIT_FAILURE);
if (l_arg && y_arg) if (l_arg && y_arg)
usage(EXIT_FAILURE); usage(EXIT_FAILURE);
if (u_arg) { if (u_arg) {
err = CONF_Add("user", u_arg); err = CONF_Add("user", u_arg);
if (err) { if (err) {
fprintf(stderr, "Unknown user: %s\n", u_arg); fprintf(stderr, "Unknown user: %s\n", u_arg);
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
}
} }
}
if (y_arg) { if (y_arg) {
err = CONF_Add("syslog.facility", y_arg); err = CONF_Add("syslog.facility", y_arg);
if (err) { if (err) {
fprintf(stderr, "Unknown syslog facility: %s\n", y_arg); fprintf(stderr, "Unknown syslog facility: %s\n", y_arg);
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
}
} }
}
if (P_arg) if (P_arg)
strcpy(config.pid_file, P_arg); strcpy(config.pid_file, P_arg);
if (n_arg) if (n_arg)
strcpy(config.varnish_name, n_arg); strcpy(config.varnish_name, n_arg);
if (l_arg) if (l_arg)
strcpy(config.log_file, l_arg); strcpy(config.log_file, l_arg);
if (f_arg) { if (f_arg) {
strcpy(config.varnish_bindump, f_arg); strcpy(config.varnish_bindump, f_arg);
endless = 0; endless = 0;
} }
if (f_arg && VSL_Arg(vd, 'r', f_arg) <= 0) if (f_arg && VSL_Arg(vd, 'r', f_arg) <= 0)
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
else if (!EMPTY(config.varnish_name) else if (!EMPTY(config.varnish_name)
&& VSL_Arg(vd, 'n', config.varnish_name) <= 0) && VSL_Arg(vd, 'n', config.varnish_name) <= 0)
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
if (LOG_Open(PACKAGE_NAME) != 0) { if (LOG_Open(PACKAGE_NAME) != 0) {
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
VAS_Fail = ASRT_Fail; VAS_Fail = ASRT_Fail;
if (d_flag) if (d_flag)
LOG_SetLevel(LOG_DEBUG); LOG_SetLevel(LOG_DEBUG);
LOG_Log0(LOG_INFO, LOG_Log0(LOG_INFO,
"initializing (v" PACKAGE_VERSION " revision " REVISION ")"); "initializing (v" PACKAGE_VERSION " revision " REVISION ")");
CONF_Dump(); CONF_Dump();
if (!EMPTY(config.pid_file) if (!EMPTY(config.pid_file)
&& (pfh = VPF_Open(config.pid_file, 0644, NULL)) == NULL) { && (pfh = VPF_Open(config.pid_file, 0644, NULL)) == NULL) {
LOG_Log(LOG_ERR, "Cannot write pid file %s: %s\n", LOG_Log(LOG_ERR, "Cannot write pid file %s: %s\n",
config.pid_file, strerror(errno)); config.pid_file, strerror(errno));
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
if (!D_flag && varnish_daemon(0, 0) == -1) { if (!D_flag && varnish_daemon(0, 0) == -1) {
perror("daemon()"); perror("daemon()");
if (pfh != NULL) if (pfh != NULL)
VPF_Remove(pfh); VPF_Remove(pfh);
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
if (pfh != NULL) if (pfh != NULL)
VPF_Write(pfh); VPF_Write(pfh);
terminate_action.sa_handler = HNDL_Terminate; terminate_action.sa_handler = HNDL_Terminate;
AZ(sigemptyset(&terminate_action.sa_mask)); AZ(sigemptyset(&terminate_action.sa_mask));
terminate_action.sa_flags &= ~SA_RESTART; terminate_action.sa_flags &= ~SA_RESTART;
stacktrace_action.sa_handler = HNDL_Abort; stacktrace_action.sa_handler = HNDL_Abort;
ignore_action.sa_handler = SIG_IGN; ignore_action.sa_handler = SIG_IGN;
default_action.sa_handler = SIG_DFL; default_action.sa_handler = SIG_DFL;
if (!D_flag) { if (!D_flag) {
child_pid = fork(); child_pid = fork();
switch(child_pid) { switch(child_pid) {
case -1: case -1:
LOG_Log(LOG_ALERT, LOG_Log(LOG_ALERT,
"Cannot fork (%s), running as single process", "Cannot fork (%s), running as single process",
strerror(errno)); strerror(errno));
CHILD_Main(vd, endless, 0); CHILD_Main(vd, endless, 0);
break; break;
case 0: case 0:
CHILD_Main(vd, endless, 0);
break;
default:
parent_main(child_pid, vd, endless);
break;
}
}
else {
LOG_Log0(LOG_INFO, "Running as non-demon single process");
CHILD_Main(vd, endless, 0); CHILD_Main(vd, endless, 0);
break;
default:
parent_main(child_pid, vd, endless);
break;
} }
}
else {
LOG_Log0(LOG_INFO, "Running as non-demon single process");
CHILD_Main(vd, endless, 0);
}
} }
trackrdrd/src/trackrdrd.h
View file @ d4d73dfe
...@@ -53,10 +53,10 @@ void ASRT_Fail(const char *func, const char *file, int line, const char *cond, ...@@ -53,10 +53,10 @@ void ASRT_Fail(const char *func, const char *file, int line, const char *cond,
#define SIGDISP(SIG, action) \ #define SIGDISP(SIG, action) \
do { if (UNDEFINED(SIG)) break; \ do { if (UNDEFINED(SIG)) break; \
if (sigaction((SIG), (&action), NULL) != 0) \ if (sigaction((SIG), (&action), NULL) != 0) \
LOG_Log(LOG_ALERT, \ LOG_Log(LOG_ALERT, \
"Cannot install handler for " #SIG ": %s", \ "Cannot install handler for " #SIG ": %s", \
strerror(errno)); \ strerror(errno)); \
} while(0) } while(0)
volatile sig_atomic_t term; volatile sig_atomic_t term;
...@@ -108,7 +108,7 @@ bool SPMCQ_StopWorker(int running); ...@@ -108,7 +108,7 @@ bool SPMCQ_StopWorker(int running);
#define spmcq_wait(what) \ #define spmcq_wait(what) \
do { \ do { \
AZ(pthread_mutex_lock(&spmcq_##what##waiter_lock)); \ AZ(pthread_mutex_lock(&spmcq_##what##waiter_lock)); \
spmcq_##what##waiter++; \ spmcq_##what##waiter++; \
AZ(pthread_cond_wait(&spmcq_##what##waiter_cond, \ AZ(pthread_cond_wait(&spmcq_##what##waiter_cond, \
&spmcq_##what##waiter_lock)); \ &spmcq_##what##waiter_lock)); \
......
trackrdrd/src/worker.c
View file @ d4d73dfe
...@@ -54,12 +54,13 @@ typedef enum { ...@@ -54,12 +54,13 @@ typedef enum {
} wrk_state_e; } wrk_state_e;
static const char* statename[WRK_STATE_E_LIMIT] = { static const char* statename[WRK_STATE_E_LIMIT] = {
[WRK_NOTSTARTED] = "not started", [WRK_NOTSTARTED] = "not started",
[WRK_INITIALIZING] = "initializing", [WRK_INITIALIZING] = "initializing",
[WRK_RUNNING] = "running", [WRK_RUNNING] = "running",
[WRK_WAITING] = "waiting", [WRK_WAITING] = "waiting",
[WRK_SHUTTINGDOWN] = "shutting down", [WRK_SHUTTINGDOWN] = "shutting down",
[WRK_EXITED] = "exited"}; [WRK_EXITED] = "exited"
};
struct worker_data_s { struct worker_data_s {
unsigned magic; unsigned magic;
...@@ -121,9 +122,9 @@ wrk_send(void *amq_worker, dataentry *entry, worker_data_t *wrk) ...@@ -121,9 +122,9 @@ wrk_send(void *amq_worker, dataentry *entry, worker_data_t *wrk)
wrk->wrk_nfree++; wrk->wrk_nfree++;
if (dtbl.nfree == 0) { if (dtbl.nfree == 0) {
DATA_Return_Freelist(&wrk->wrk_freelist, wrk->wrk_nfree); DATA_Return_Freelist(&wrk->wrk_freelist, wrk->wrk_nfree);
wrk->wrk_nfree = 0; wrk->wrk_nfree = 0;
assert(VSTAILQ_EMPTY(&wrk->wrk_freelist)); assert(VSTAILQ_EMPTY(&wrk->wrk_freelist));
} }
spmcq_signal(room); spmcq_signal(room);
...@@ -168,48 +169,43 @@ static void ...@@ -168,48 +169,43 @@ static void
LOG_Log(LOG_INFO, "Worker %d: running (%s)", wrk->id, version); LOG_Log(LOG_INFO, "Worker %d: running (%s)", wrk->id, version);
while (run) { while (run) {
entry = (dataentry *) SPMCQ_Deq(); entry = (dataentry *) SPMCQ_Deq();
if (entry != NULL) { if (entry != NULL) {
wrk->deqs++; wrk->deqs++;
wrk_send(amq_worker, entry, wrk); wrk_send(amq_worker, entry, wrk);
/* should we go to sleep ? */ if (!SPMCQ_StopWorker(running))
if (SPMCQ_StopWorker(running)) continue;
goto sleep;
continue;
} }
sleep: /* return space before sleeping */
/* return space before sleeping */ if (wrk->wrk_nfree > 0) {
if (wrk->wrk_nfree > 0) { DATA_Return_Freelist(&wrk->wrk_freelist, wrk->wrk_nfree);
DATA_Return_Freelist(&wrk->wrk_freelist, wrk->wrk_nfree); wrk->wrk_nfree = 0;
wrk->wrk_nfree = 0; }
}
/* /*
* Queue is empty or we should backoff * Queue is empty or we should backoff
* *
* wait until data are available, or quit is signaled. * wait until data are available, or quit is signaled.
* *
* Grab the CV lock, which also constitutes an implicit memory * Grab the CV lock, which also constitutes an implicit memory
* barrier * barrier
*/ */
AZ(pthread_mutex_lock(&spmcq_datawaiter_lock)); AZ(pthread_mutex_lock(&spmcq_datawaiter_lock));
/* /*
* run is guaranteed to be fresh here * run is guaranteed to be fresh here
* *
* also re-check the stop condition under the lock * also re-check the stop condition under the lock
*/ */
if (run && ((! entry) || SPMCQ_StopWorker(running))) { if (run && ((! entry) || SPMCQ_StopWorker(running))) {
wrk->waits++; wrk->waits++;
spmcq_datawaiter++; spmcq_datawaiter++;
wrk->state = WRK_WAITING; wrk->state = WRK_WAITING;
AZ(pthread_cond_wait(&spmcq_datawaiter_cond, AZ(pthread_cond_wait(&spmcq_datawaiter_cond,
&spmcq_datawaiter_lock)); &spmcq_datawaiter_lock));
spmcq_datawaiter--; spmcq_datawaiter--;
wrk->state = WRK_RUNNING; wrk->state = WRK_RUNNING;
} }
AZ(pthread_mutex_unlock(&spmcq_datawaiter_lock)); AZ(pthread_mutex_unlock(&spmcq_datawaiter_lock));
} }
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment