From 2052dc0a4faf38d792e3fec09d24f896155948b6 Mon Sep 17 00:00:00 2001 From: Vladimir Dubrovin <3proxy@3proxy.ru> Date: Fri, 23 Oct 2026 19:41:48 +0300 Subject: [PATCH] Logs: a thread and a queue for each log; build-time defaults Once 3proxy runs (after the first configuration is in force, so after a daemon's fork), each logger has a thread of its own writing from a queue, and a client never waits for a log: ODBC, RADIUS, a full disk or a pipe nobody reads hold up only that log's thread. A queued line is fixed size: the part of struct clientparam logging reads (everything before the filters: addresses, counters, names and credentials, which hold no pointer it follows but srv), the time it was logged, and its text, 256 bytes inline or a copy of its own if longer. The thread formats it: struct tm is per call now, timechanged() and dologname() included, and the type's format() takes the time. minlogqueue (64) slots are allocated as the thread starts and kept, maxlogqueue (4096) lines are queued at most. A line which finds the queue full is dropped and counted on the last one queued; once that is written, "Log X has its queue full, N records dropped" goes to where errors go. A log which fails is no longer written to the error log instead: its lines are dropped and counted while it is retried every minute, and the count is told when it is open again, or when it was reopened after failed writes. A service which ended is stopped at once (srvstop(): sockets, listener, its clients gone), but freed (srvrelease()) through its log's queue, behind the lines which refer to it and to its configuration: the event is part of struct srvparam, so a full queue never drops it. logfinish() writes what is queued, 10 seconds at most, also as 3proxy exits on Unix. Standalone services write as before, as each line comes. LOGQUEUEMIN, LOGQUEUEMAX, DEFAULT_MAXCHILD, DEFAULT_POOLMAX, DEFAULT_POOLTIMEOUT and POOLPERSERVICE (the automatic thread pool cap for each service, which was 32 in the code) can be set with -D at build time. Tests: a log nobody reads (a FIFO) holds no client up, drops past its queue and says so, and lines queued for a service a reload removed are written as it formatted them; /dev/full says what it dropped. Co-Authored-By: Claude Opus 5.5 --- doc/html/man5/3proxy.cfg.5.html | 28 +- man/3proxy.cfg.5 | 32 +- src/3proxy.c | 9 +- src/conf.c | 14 + src/confinit.h | 6 +- src/log.c | 773 ++++++++++++++++++++++++++------ src/mainclient.c | 6 +- src/mainconf.c | 41 +- src/proxy.h | 19 + src/proxymain.c | 15 +- src/sql.c | 4 +- src/structures.h | 54 ++- tests/cases/logs.py | 87 +++- 13 files changed, 930 insertions(+), 158 deletions(-) diff --git a/doc/html/man5/3proxy.cfg.5.html b/doc/html/man5/3proxy.cfg.5.html index 9103b39..f7a7dee 100644 --- a/doc/html/man5/3proxy.cfg.5.html +++ b/doc/html/man5/3proxy.cfg.5.html @@ -349,12 +349,23 @@ log or by -l, and shared by every service using it; it is kept open over a configuration reload while some service still uses it, and closed once none does. stdout, stderr and syslog are opened at startup and never closed. rotate and archiver apply -to the file of the last log command. A log which fails to be -written is reopened, and if it cannot be, its records go to -stderr while it is retried every minute. Errors and warnings +to the file of the last log command. Errors and warnings with no connection go to stderr; after daemon (or as a Windows service) nothing goes to stdout or stderr: logging to stdout is none, and errors go to the last log configured.
+Once 3proxy runs, each log is written by a thread of its own +from a queue (see minlogqueue and maxlogqueue): +a client never waits for a log, a slow one (ODBC, RADIUS, a +full disk) included. A record which finds the queue full is +dropped. A log which fails to be written is reopened; if it +cannot be, its records are dropped while it is retried every +minute. Each time records are dropped, how many is said where +errors go: "Log name has its queue full, +N records dropped", "could not be written, +opened again", "failed", "is open +again". A service a reload removed is freed only once the +records logged for it are written. Standalone services write +their log as each record comes.
log command supports same format specifications for filename template as "logformat" (if filename contains ´%´ sign it´s believed to be template). As @@ -367,6 +378,17 @@ specificators.

<n>
how many archived log files to keep

+

minlogqueue +<n>
+maxlogqueue <n>
+The records a log´s queue keeps room for, allocated as +its thread starts (64 by default), and holds at most, records +past it being dropped (4096 by default). Room taken by a burst +beyond minlogqueue is freed as the queue empties. A record +takes about 2 KB. The values of the configuration in force +apply to every log. Other defaults can be given at build time +with -DLOGQUEUEMIN=n and -DLOGQUEUEMAX=n.

+

logformat <format>
diff --git a/man/3proxy.cfg.5 b/man/3proxy.cfg.5 index 252b44d..a5b1fe9 100644 --- a/man/3proxy.cfg.5 +++ b/man/3proxy.cfg.5 @@ -353,9 +353,19 @@ alone is stdout. A log is opened once for the same argument, whether given by log or by -l, and shared by every service using it; it is kept open over a configuration reload while some service still uses it, and closed once none does. stdout, stderr and syslog are opened at startup and never closed. rotate and archiver apply to the file of the last log -command. A log which fails to be written is reopened, and if it cannot be, its records go to stderr while it is -retried every minute. Errors and warnings with no connection go to stderr; after daemon (or as a Windows service) +command. Errors and warnings with no connection go to stderr; after daemon (or as a Windows service) nothing goes to stdout or stderr: logging to stdout is none, and errors go to the last log configured. +.br + Once 3proxy runs, each log is written by a thread of its own from a queue (see +.BR minlogqueue +and +.BR maxlogqueue ): +a client never waits for a log, a slow one (ODBC, RADIUS, a full disk) included. A record which finds the +queue full is dropped. A log which fails to be written is reopened; if it cannot be, its records are dropped +while it is retried every minute. Each time records are dropped, how many is said where errors go: "Log +\fIname\fR has its queue full, \fIN\fR records dropped", "could not be written, opened again", "failed", +"is open again". A service a reload removed is freed only once the records logged for it are written. +Standalone services write their log as each record comes. .br log command supports same format specifications for filename template as "logformat" (if filename contains \'%\' sign it\'s believed to be template). @@ -367,6 +377,18 @@ Grinwitch time zone for all time-based format specificators. \fI\fR how many archived log files to keep +.br +.BR minlogqueue +\fI\fR +.br +.BR maxlogqueue +\fI\fR +.br + The records a log's queue keeps room for, allocated as its thread starts (64 by default), and holds at +most, records past it being dropped (4096 by default). Room taken by a burst beyond minlogqueue is freed +as the queue empties. A record takes about 2 KB. The values of the configuration in force apply to every +log. Other defaults can be given at build time with -DLOGQUEUEMIN=\fIn\fR and -DLOGQUEUEMAX=\fIn\fR. + .br .BR logformat \fI\fR @@ -1478,7 +1500,11 @@ instead of ending, and keeps the memory it had for its connection: a new connection starts without creating a thread. At most \fImax_idle\fR threads wait, each at most \fIidle_seconds\fR (60 by default; 0 waits without a limit). By default up to 32 threads wait for each running service, but no more than -the services' maxchild together. \fBthreadpool 0\fR ends every thread with its +the services' maxchild together. Each waiting thread keeps its stack and +its connection's structure, about 60 KB with the buffers. Other defaults can +be given at build time: -DDEFAULT_POOLMAX=\fIn\fR (-1 is the automatic +cap), -DPOOLPERSERVICE=\fIn\fR (the 32), -DDEFAULT_POOLTIMEOUT=\fIseconds\fR +and, for maxconn, -DDEFAULT_MAXCHILD=\fIn\fR (500). \fBthreadpool 0\fR ends every thread with its connection. A thread serves a service only if its stack is at least as large as the service needs (see \fBstacksize\fR), otherwise a new one is created. The admin service shows how many threads wait, how many were created and how diff --git a/src/3proxy.c b/src/3proxy.c index 3ab75d7..60769b5 100644 --- a/src/3proxy.c +++ b/src/3proxy.c @@ -538,12 +538,17 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int { (LPSTR)conf->stringtable[1], (LPSERVICE_MAIN_FUNCTION)ServiceMain}, { NULL, NULL } }; - if(!StartServiceCtrlDispatcher( ste ))cyclestep(); + if(!StartServiceCtrlDispatcher( ste )){ + cyclestep(); + logfinish(); + } } else #endif { cyclestep(); + /* what is queued to the logs written */ + logfinish(); } @@ -552,6 +557,8 @@ int WinMain(HINSTANCE hInstance, HINSTANCE hPrevInstance, LPWSTR lpCmdLine, int signal(SIGTERM, mysigterm); signal(SIGUSR1, mysigusr1); cyclestep(); + /* what is queued to the logs written */ + logfinish(); #endif diff --git a/src/conf.c b/src/conf.c index 2b29750..2ee632b 100644 --- a/src/conf.c +++ b/src/conf.c @@ -401,6 +401,18 @@ static int h_log(int argc, unsigned char ** argv){ return 0; } +static int h_minlogqueue(int argc, unsigned char **argv){ + conf->minlogqueue = atoi((char *)argv[1]); + if(conf->minlogqueue < 0) conf->minlogqueue = 0; + return 0; +} + +static int h_maxlogqueue(int argc, unsigned char **argv){ + conf->maxlogqueue = atoi((char *)argv[1]); + if(conf->maxlogqueue < 1) conf->maxlogqueue = 1; + return 0; +} + static int h_stacksize(int argc, unsigned char **argv){ conf->stacksize = atoi((char *)argv[1]); return 0; @@ -2114,6 +2126,8 @@ struct commands commandhandlers[]={ {NULL, "delimchar",h_delimchar, 2, 2}, {NULL, "authnserver", h_authnserver, 2, 2}, {NULL, "stacksize", h_stacksize, 2, 2}, + {NULL, "minlogqueue", h_minlogqueue, 2, 2}, + {NULL, "maxlogqueue", h_maxlogqueue, 2, 2}, {NULL, "threadpool", h_threadpool, 2, 3}, {NULL, "force", h_force, 1, 1}, {NULL, "noforce", h_noforce, 1, 1}, diff --git a/src/confinit.h b/src/confinit.h index 214ecab..de1c91e 100644 --- a/src/confinit.h +++ b/src/confinit.h @@ -22,12 +22,14 @@ .trafcounter = NULL, .services = NULL, .stacksize = 0, - .poolmax = -1, - .pooltimeout = 60, + .poolmax = DEFAULT_POOLMAX, + .pooltimeout = DEFAULT_POOLTIMEOUT, .counterd = -1, .rotate = 0, .paused = 0, .archiverc = 0, + .minlogqueue = LOGQUEUEMIN, + .maxlogqueue = LOGQUEUEMAX, .demon = 0, .maxchild = DEFAULT_MAXCHILD, .backlog = 0, diff --git a/src/log.c b/src/log.c index 47102b2..6db8484 100644 --- a/src/log.c +++ b/src/log.c @@ -60,37 +60,41 @@ char months[12][4] = { }; -int dobuf2(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec, struct tm* tm, char * format){ +/* The time now, to the millisecond. */ +static void lognow(time_t *sec, unsigned *msec){ +#ifdef _WIN32 + struct timeb tb; + + ftime(&tb); + *sec = (time_t)tb.time; + *msec = (unsigned)tb.millitm; +#else + struct timeval tv; + + gettimeofday(&tv, NULL); + *sec = (time_t)tv.tv_sec; + *msec = (unsigned)(tv.tv_usec / 1000); +#endif +} + +/* The line for what was logged at sec and msec, tm being sec's time. */ +static int dobuf3(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec, struct tm* tm, char * format, time_t sec, unsigned msec){ int i, j; int len; - time_t sec; - unsigned msec; long timezone; unsigned delay; #ifdef _WIN32 struct timeb tb; -#else - struct timeval tv; - struct timezone tz; #endif if(!param) param = &logparam; - - #ifdef _WIN32 ftime(&tb); - sec = (time_t)tb.time; - msec = (unsigned)tb.millitm; timezone = tm->tm_isdst*60 - tb.timezone; - #else - gettimeofday(&tv, &tz); - - sec = (time_t)tv.tv_sec; - msec = tv.tv_usec / 1000; #ifdef _SOLARIS timezone = -altzone / 60; #else @@ -316,14 +320,29 @@ int dobuf2(struct clientparam * param, unsigned char * buf, const unsigned char return i; } +int dobuf2(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec, struct tm* tm, char * format){ + time_t sec; + unsigned msec; + + lognow(&sec, &msec); + return dobuf3(param, buf, s, doublec, tm, format, sec, msec); +} + +/* localtime() for many threads: into tm, NULL if t cannot be converted */ +static struct tm *loglocal(time_t t, struct tm *tm){ +#ifdef _WIN32 + return localtime_s(tm, &t)? NULL : tm; +#else + return localtime_r(&t, tm); +#endif +} + int timechanged (time_t oldtime, time_t newtime, ROTATION lt){ - struct tm tmold; + struct tm tmold, tmnew; struct tm *tm; - tm = localtime(&oldtime); - if(!tm) return 0; - tmold = *tm; - tm = localtime(&newtime); - if(!tm) return 0; + + if(!loglocal(oldtime, &tmold)) return 0; + if(!(tm = loglocal(newtime, &tmnew))) return 0; switch(lt){ case MINUTELY: if(tm->tm_min != tmold.tm_min)return 1; @@ -353,7 +372,7 @@ int timechanged (time_t oldtime, time_t newtime, ROTATION lt){ } unsigned char * dologname (unsigned char *buf, unsigned char *name, const unsigned char *ext, ROTATION lt, time_t t) { - struct tm *ts; + struct tm tm, *ts; static const char * const rot_fmt[] = { [NONE] = "%s", [ANNUALLY] = "%s.%04d", @@ -363,17 +382,19 @@ unsigned char * dologname (unsigned char *buf, unsigned char *name, const unsign [MINUTELY] = "%s.%04d.%02d.%02d-%02d.%02d", }; - ts = localtime(&t); - if(strlen((char *)name) >= 4096){ + if(strlen((char *)name) >= 4096 || !(ts = loglocal(t, &tm))){ *buf = 0; return buf; } if(strchr((char *)name, '%')){ - dobuf2(NULL, buf, NULL, NULL, ts, (char *)name); + dobuf3(NULL, buf, NULL, NULL, ts, (char *)name, t, 0); } else if(lt == WEEKLY){ t = t - (ts->tm_wday * (60*60*24)); - ts = localtime(&t); + if(!(ts = loglocal(t, &tm))){ + *buf = 0; + return buf; + } sprintf((char *)buf, "%s.%04d.%02d.%02d", name, ts->tm_year+1900, ts->tm_mon+1, ts->tm_mday); } else if((unsigned)lt < sizeof(rot_fmt)/sizeof(rot_fmt[0]) && rot_fmt[lt]){ @@ -388,16 +409,33 @@ unsigned char * dologname (unsigned char *buf, unsigned char *name, const unsign /* The line for a connection, its length: with nothing else done, as a log formats it. */ -int logbuf(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec){ - struct tm* tm; +int logbuft(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec, time_t sec, unsigned msec){ + struct tm tm, *tp; char * format; - time_t t; - time(&t); format = (param && param->srv && param->srv->logformat)?(char *)param->srv->logformat : DEFLOGFORMAT; - tm = (*format == 'G' || *format == 'g')? - gmtime(&t) : localtime(&t); - return dobuf2(param, buf, s, doublec, tm, format + 1); + /* the threads which format lines are many: each its own struct tm */ + if(*format == 'G' || *format == 'g'){ +#ifdef _WIN32 + tp = gmtime_s(&tm, &sec)? NULL : &tm; +#else + tp = gmtime_r(&sec, &tm); +#endif + } + else tp = loglocal(sec, &tm); + if(!tp){ + memset(&tm, 0, sizeof(tm)); + tp = &tm; + } + return dobuf3(param, buf, s, doublec, tp, format + 1, sec, msec); +} + +int logbuf(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec){ + time_t sec; + unsigned msec; + + lognow(&sec, &msec); + return logbuft(param, buf, s, doublec, sec, msec); } /* The line, with the traffic counted and the statistics cleared, as logging @@ -416,14 +454,44 @@ int dobuf(struct clientparam * param, unsigned char * buf, const unsigned char * static int logstdoff = 0; /* a daemon or a service: no standard output and error */ static int logticking = 0; /* logtick() runs, and flushes the files */ +static int logthreads = 0; /* loggers get threads of their own: 3proxy runs */ +static int logstartfailed = 0; /* a thread could not be started, not yet told */ +static int logworkers = 0; /* threads running, under log_mutex */ +static _3proxy_sem_t logexited; /* posted by each thread as it ends */ +/* the slots of a queue: kept, and lines held at most */ +static int logqmin = LOGQUEUEMIN, logqmax = LOGQUEUEMAX; static struct logger *loggers = NULL; /* the last log configured, where errors go with no standard error */ static struct logger *logerrlast = NULL; static const struct logtype *logtypes[8]; static int nlogtypes = 0; -static int logtext(struct logger *lg, struct clientparam *param, const unsigned char *s, unsigned char *buf){ - return logbuf(param, buf, s, NULL); +/* A line in a queue. A text longer than text[] is a copy of its own. A + connection's line is followed by what logging reads of the connection: + everything before its filters, names and credentials included, which + holds no pointer it follows but srv. The service is freed only once its + lines are written (logcall()). */ +#define LOGTEXTSIZE 256 +#define LOGPARAMSIZE offsetof(struct clientparam, filters) +#define LOGSLOTSIZE (sizeof(struct logslot) + LOGPARAMSIZE) +#define LOGSLOTPARAM(sl) ((unsigned char *)((sl) + 1)) + +struct logslot { + struct logevent e; + time_t sec; /* when it was logged */ + unsigned msec; + unsigned char *big; /* the text, if too long for text[] */ + unsigned char text[LOGTEXTSIZE]; +}; + +enum {LOGLINE = 1, LOGTEXT, LOGCALL}; + +/* names and credentials, read by logging, come before the filters */ +typedef char logparamcheck[(offsetof(struct clientparam, extpassword) + + sizeof(((struct clientparam *)0)->extpassword) <= LOGPARAMSIZE)? 1 : -1]; + +static int logtext(struct logger *lg, struct clientparam *param, const unsigned char *s, time_t sec, unsigned msec, unsigned char *buf){ + return logbuft(param, buf, s, NULL, sec, msec); } static int matchnull(const unsigned char *selector){ @@ -457,7 +525,7 @@ static const struct logtype logtype_stdout = { "stdout", matchstdout, NULL, logtext, stdoutwrite, NULL, NULL, 0, 1, 1 }; -/* errors and warnings, and the lines of a log which fails */ +/* errors and warnings */ static const struct logtype logtype_stderr = { "stderr", NULL, NULL, logtext, stderrwrite, NULL, NULL, 0, 1, 1 }; @@ -540,8 +608,9 @@ static int filewrite(struct logger *lg, struct clientparam *param, const unsigne if(!fp) return 1; if(fprintf(fp, "%s\n", buf) < 0) return 1; - /* with nobody to flush it later, a standalone service */ - if(!logticking && fflush(fp)) return 1; + /* by its thread once nothing more is queued; with nobody to flush it + later, a standalone service, each line */ + if((lg->running? !lg->more : !logticking) && fflush(fp)) return 1; return 0; } @@ -592,7 +661,7 @@ static void filetick(struct logger *lg, time_t t){ if(*name && (fp = fopen((char *)name, "a"))){ if(lg->data) fclose((FILE *)lg->data); lg->data = fp; - lg->failed = lg->errors = 0; + lg->errors = 0; } lg->rotated = t; lg->reopen = 0; @@ -661,6 +730,12 @@ static const struct logtype *logmatch(const unsigned char *selector){ return &logtype_file; } +/* A logger's own: its mutex and the semaphore its thread waits on. */ +static void logsetup(struct logger *lg){ + _3proxy_mutex_init(&lg->mutex); + _3proxy_sem_init_timed(lg->wake, 0, 1); +} + /* The standard loggers, before anything is logged. */ void loginit(void){ static int done = 0; @@ -668,10 +743,12 @@ void loginit(void){ if(done) return; done = 1; _3proxy_mutex_init(&log_mutex); - _3proxy_mutex_init(&lognull.mutex); - _3proxy_mutex_init(&logstdout.mutex); - _3proxy_mutex_init(&logstderr.mutex); + _3proxy_sem_init_timed(logexited, 0, 0x7fffffff); + logsetup(&lognull); + logsetup(&logstdout); + logsetup(&logstderr); lognull.next = &logstdout; + logstdout.next = &logstderr; loggers = &lognull; srvinit(&logsrv, &logparam); /* not a service: it holds no configuration and no log, and the first @@ -691,25 +768,94 @@ static size_t lognamelen(const struct logtype *type, const unsigned char *select return type->params? strcspn((char *)selector, ",") : strlen((char *)selector); } +static void logfreearchiver(unsigned char **archiver, int archiverc){ + int i; + + if(!archiver) return; + for(i = 0; i < archiverc; i++) if(archiver[i]) free(archiver[i]); + free(archiver); +} + +/* Opens the log again, its parameters or rotation having changed, or after + it failed: 0, or the error of its init. */ +static int logreopen(struct logger *lg){ + if(lg->type->close) lg->type->close(lg); + lg->errors = 0; + return lg->type->init? lg->type->init(lg) : 0; +} + /* Parameters which differ from those a logger has: given to it, and the - logger opened again with them. Called holding a reference. */ + logger opened again with them, by its thread if it has one. Called + holding a reference. */ static void logsetparams(struct logger *lg, const unsigned char *params){ unsigned char *p = NULL; if(params && !(p = (unsigned char *)strdup((char *)params))) return; _3proxy_mutex_lock(&lg->mutex); + if(lg->running){ + if(lg->nparams) free(lg->nparams); + lg->nparams = p; + lg->newparams = 1; + _3proxy_mutex_unlock(&lg->mutex); + _3proxy_sem_unlock(lg->wake); + return; + } if(lg->params) free(lg->params); lg->params = p; - if(lg->type->close) lg->type->close(lg); - lg->errors = 0; lg->failed = 0; - if(lg->type->init && lg->type->init(lg)){ + if(logreopen(lg)){ lg->failed = 1; lg->failtime = time(NULL); } _3proxy_mutex_unlock(&lg->mutex); } +static void *logthread(void *arg); + +/* Gives a logger a thread of its own, its queue's slots allocated first. + Its mutex is held until the thread is there: no line is queued for a + thread which could not be started. Called holding log_mutex: nothing + may be logged from here. */ +static void logstart(struct logger *lg){ + struct logslot *sl; + pthread_t thread; + int i, res; +#ifdef _WIN32 + HANDLE h; +#else + pthread_attr_t pa; +#endif + + if(lg->running || !lg->type->write) return; + _3proxy_mutex_lock(&lg->mutex); + for(i = lg->slots; i < logqmin && (sl = malloc(LOGSLOTSIZE)); i++){ + sl->e.next = (struct logevent *)lg->free; + lg->free = sl; + lg->slots++; + } + lg->stop = 0; +#ifdef _WIN32 +#ifndef _WINCE + h = (HANDLE)_beginthreadex((LPSECURITY_ATTRIBUTES )NULL, (unsigned)(65536 + lg->type->stacksize), (void *)logthread, (void *)lg, 0, &thread); +#else + h = (HANDLE)CreateThread((LPSECURITY_ATTRIBUTES )NULL, (unsigned)(65536 + lg->type->stacksize), (void *)logthread, (void *)lg, 0, &thread); +#endif + res = !h; + if(h) CloseHandle(h); +#else + pthread_attr_init(&pa); + pthread_attr_setstacksize(&pa, threadstacksize(lg->type->stacksize)); + pthread_attr_setdetachstate(&pa, PTHREAD_CREATE_DETACHED); + res = pthread_create(&thread, &pa, logthread, (void *)lg); + pthread_attr_destroy(&pa); +#endif + /* if not, written as it comes */ + if(!res) lg->running = 1; + _3proxy_mutex_unlock(&lg->mutex); + if(res) logstartfailed = 1; + else logworkers++; +} + /* The logger named so, opened if none is: held until logput(). NULL, with the error in *errp (an errno value) unless it is NULL, if it cannot be opened. A logger found with other parameters takes the new ones. */ @@ -717,6 +863,7 @@ struct logger *logget(const unsigned char *selector, ROTATION rotation, int *err struct logger *lg; const struct logtype *type; const unsigned char *params; + unsigned char *cur; size_t len; int err = 0, changed; @@ -727,7 +874,11 @@ struct logger *logget(const unsigned char *selector, ROTATION rotation, int *err for(lg = loggers; lg; lg = lg->next){ if(lg->type == type && !strncmp((char *)lg->selector, (char *)selector, len) && !lg->selector[len]){ lg->refs++; - changed = (params && lg->params)? strcmp((char *)params, (char *)lg->params) : (params != NULL || lg->params != NULL); + /* those last given, which its thread may not have taken yet */ + _3proxy_mutex_lock(&lg->mutex); + cur = lg->newparams? lg->nparams : lg->params; + changed = (params && cur)? strcmp((char *)params, (char *)cur) : (params != NULL || cur != NULL); + _3proxy_mutex_unlock(&lg->mutex); _3proxy_mutex_unlock(&log_mutex); if(changed) logsetparams(lg, params); return lg; @@ -748,7 +899,6 @@ struct logger *logget(const unsigned char *selector, ROTATION rotation, int *err lg->type = type; lg->rotation = rotation; lg->rotated = time(NULL); - _3proxy_mutex_init(&lg->mutex); /* < 0: not to be reached now, and tried again later */ if(lg->type->init && (err = lg->type->init(lg)) < 0){ lg->failed = 1; @@ -756,16 +906,17 @@ struct logger *logget(const unsigned char *selector, ROTATION rotation, int *err } else if(err){ _3proxy_mutex_unlock(&log_mutex); - _3proxy_mutex_destroy(&lg->mutex); if(lg->params) free(lg->params); free(lg->selector); free(lg); if(errp) *errp = err; return NULL; } + logsetup(lg); lg->refs = 1; lg->next = loggers; loggers = lg; + if(logthreads) logstart(lg); _3proxy_mutex_unlock(&log_mutex); return lg; } @@ -777,17 +928,26 @@ void logref(struct logger *lg){ _3proxy_mutex_unlock(&log_mutex); } -static void logfreearchiver(struct logger *lg){ - int i; +/* What a logger holds, the logger itself last; its queue is empty. */ +static void logdestroy(struct logger *lg){ + struct logslot *sl; - if(!lg->archiver) return; - for(i = 0; i < lg->archiverc; i++) if(lg->archiver[i]) free(lg->archiver[i]); - free(lg->archiver); - lg->archiver = NULL; - lg->archiverc = 0; + while((sl = (struct logslot *)lg->free)){ + lg->free = sl->e.next; + free(sl); + } + _3proxy_mutex_destroy(&lg->mutex); + _3proxy_sem_destroy(lg->wake); + logfreearchiver(lg->archiver, lg->archiverc); + logfreearchiver(lg->narchiver, lg->narchiverc); + if(lg->params) free(lg->params); + if(lg->nparams) free(lg->nparams); + free(lg->selector); + free(lg); } -/* The last to let go of a logger closes it, unless it is a standard one. */ +/* The last to let go of a logger closes it, unless it is a standard one: + its thread, if it has one, once it has written what is queued. */ void logput(struct logger *lg){ struct logger **lp; int gone = 0; @@ -805,12 +965,16 @@ void logput(struct logger *lg){ } _3proxy_mutex_unlock(&log_mutex); if(!gone) return; + _3proxy_mutex_lock(&lg->mutex); + if(lg->running){ + lg->stop = 2; + _3proxy_mutex_unlock(&lg->mutex); + _3proxy_sem_unlock(lg->wake); + return; + } + _3proxy_mutex_unlock(&lg->mutex); if(lg->type->close) lg->type->close(lg); - _3proxy_mutex_destroy(&lg->mutex); - logfreearchiver(lg); - if(lg->params) free(lg->params); - free(lg->selector); - free(lg); + logdestroy(lg); } /* Where errors go: standard error, or for a daemon the last log @@ -846,19 +1010,6 @@ void logstdclose(void){ logstdoff = 1; } -static void logwrite(struct logger *lg, struct clientparam *param, const unsigned char *buf, int len); - -/* The line of a log which failed, to standard error, or to where errors go. - Not to another log of the same: once there it would come back. */ -static void logfallback(struct logger *lg, struct clientparam *param, const unsigned char *buf, int len){ - struct logger *to; - - if(len < 0) return; - if(!(to = logerrget())) return; - if(to != lg && !to->failed) logwrite(to, param, buf, len); - logput(to); -} - /* The part of a log string fit to be shown: its name, without parameters (an ODBC user and password), and at most 200 bytes. */ size_t logselen(const unsigned char *selector){ @@ -867,53 +1018,376 @@ size_t logselen(const unsigned char *selector){ return len > 200? 200 : len; } -static void logwrite(struct logger *lg, struct clientparam *param, const unsigned char *buf, int len){ - int res = 0, failed, report = 0; - time_t t; +/* What happened to a log, to where errors go. */ +static void logtell(struct logger *lg, const char *what, unsigned n){ + unsigned char msg[400]; - _3proxy_mutex_lock(&lg->mutex); - if(lg->failed && (t = time(NULL)) - lg->failtime >= LOGRETRY){ - if(lg->type->init && lg->type->init(lg)) lg->failtime = t; - else lg->failed = lg->errors = 0; + sprintf((char *)msg, "Log %.200s %s", lg->selector, what); + if(n) sprintf((char *)msg + strlen((char *)msg), ", %u records dropped", n); + logerror(msg); +} + +/* Writes a line: 0 if written or nothing to write, 1 if it could not be, + 3 if then the log was opened again, 2 if it could not be, and failed. */ +static int logout(struct logger *lg, struct clientparam *param, const unsigned char *s, time_t sec, unsigned msec, unsigned char *buf){ + int len = -1; + + if(lg->type->format && (len = lg->type->format(lg, param, s, sec, msec, buf)) < 0) return 0; + if(!lg->type->write) return 0; + if(!lg->type->write(lg, param, buf, len)){ + lg->errors = 0; + return 0; } - if(!(failed = lg->failed) && lg->type->write){ - if((res = lg->type->write(lg, param, buf, len))){ - /* open again what keeps failing, and if it cannot be - opened, write elsewhere for a while */ - if(++lg->errors >= LOGERRORS){ - lg->errors = 0; - if(lg->type->close) lg->type->close(lg); - if(lg->type->init && lg->type->init(lg)){ - lg->failed = 1; - lg->failtime = time(NULL); - report = 1; - } - } + /* open again what keeps failing */ + if(++lg->errors < LOGERRORS) return 1; + return logreopen(lg)? 2 : 3; +} + +/* A line written as it comes: by a logger with no thread. */ +static void logdirect(struct logger *lg, struct clientparam *param, const unsigned char *s, unsigned char *buf){ + time_t sec; + unsigned msec, n = 0, lost = 0; + int res = 0, reopened = 0; + + lognow(&sec, &msec); + _3proxy_mutex_lock(&lg->mutex); + if(lg->failed && sec - lg->failtime >= LOGRETRY){ + if(logreopen(lg)) lg->failtime = sec; + else { + lg->failed = 0; + n = lg->faildropped; + lg->faildropped = 0; + reopened = 1; + } + } + if(lg->failed) lg->faildropped++; + else if((res = logout(lg, param, s, sec, msec, buf))){ + lg->faildropped++; + if(res == 2){ + lg->failed = 1; + lg->failtime = sec; + } + else if(res == 3){ + lost = lg->faildropped; + lg->faildropped = 0; } - else lg->errors = 0; } _3proxy_mutex_unlock(&lg->mutex); - if(failed || res) logfallback(lg, param, buf, len); - if(report){ - unsigned char msg[300]; + if(reopened) logtell(lg, "is open again", n); + if(res == 2) logtell(lg, "failed: its records are dropped until it is open again", 0); + if(res == 3) logtell(lg, "could not be written, opened again", lost); +} - sprintf((char *)msg, "Log %.200s failed, its lines go to the error log", lg->selector); - logerror(msg); +/* Puts an event at the end of the queue, the thread woken if it was + empty. Under the logger's mutex. */ +static void logappend(struct logger *lg, struct logevent *ev){ + int wake = !lg->head; + + ev->next = NULL; + if(lg->tail) lg->tail->next = ev; + else lg->head = ev; + lg->tail = ev; + if(wake) _3proxy_sem_unlock(lg->wake); +} + +/* Queues a line, for a connection or of text: 0 if the logger has no + thread and it is to be written now, 1 if it was queued, or dropped: the + log failed, or its queue is full, counted on the last line queued. */ +static int logqueue(struct logger *lg, struct clientparam *param, const unsigned char *s, int kind){ + struct logslot *sl; + unsigned char *big = NULL; + size_t len = 0; + time_t sec; + unsigned msec; + + if(s && (len = strlen((char *)s)) >= LOGTEXTSIZE){ + /* cut short if there is no memory for all of it */ + if(!(big = (unsigned char *)malloc(len + 1))) len = LOGTEXTSIZE - 1; + else memcpy(big, s, len + 1); + } + lognow(&sec, &msec); + _3proxy_mutex_lock(&lg->mutex); + if(!lg->running){ + _3proxy_mutex_unlock(&lg->mutex); + if(big) free(big); + return 0; + } + sl = NULL; + if(lg->failed) lg->faildropped++; + else if(lg->lines >= logqmax || (!(sl = (struct logslot *)lg->free) && !(sl = malloc(LOGSLOTSIZE)))){ + sl = NULL; + if(lg->tail) lg->tail->dropped++; + else lg->faildropped++; + } + else if(sl == lg->free) lg->free = sl->e.next; + else lg->slots++; + if(!sl){ + _3proxy_mutex_unlock(&lg->mutex); + if(big) free(big); + return 1; + } + lg->lines++; + sl->e.kind = kind; + sl->e.dropped = 0; + sl->e.func = NULL; + sl->e.arg = NULL; + sl->sec = sec; + sl->msec = msec; + sl->big = big; + if(big || !s) len = 0; + memcpy(sl->text, s? s : (const unsigned char *)"", len); + sl->text[len] = 0; + if(kind == LOGLINE) memcpy(LOGSLOTPARAM(sl), param, LOGPARAMSIZE); + logappend(lg, &sl->e); + _3proxy_mutex_unlock(&lg->mutex); + return 1; +} + +/* func(arg) once the lines queued to the log before are written, ev being + where it waits in the queue: now, if the log has no thread. A full + queue does not drop it. */ +void logcall(struct logger *lg, struct logevent *ev, void (*func)(void *), void *arg){ + if(lg){ + _3proxy_mutex_lock(&lg->mutex); + if(lg->running){ + ev->kind = LOGCALL; + ev->dropped = 0; + ev->func = func; + ev->arg = arg; + logappend(lg, ev); + _3proxy_mutex_unlock(&lg->mutex); + return; + } + _3proxy_mutex_unlock(&lg->mutex); + } + func(arg); +} + +/* The connection of a queued line, for its thread to format: what logging + reads copied in, the pointers it does not follow cleared. */ +static void logparamcopy(struct clientparam *to, const unsigned char *from){ + memcpy(to, from, LOGPARAMSIZE); + to->next = to->prev = NULL; + to->sostate = NULL; + to->newsrv = NULL; + to->connlimconf = NULL; + to->redirectfunc = NULL; + to->bandlimfunc = NULL; + to->trafcountfunc = NULL; + to->clibuf = to->srvbuf = NULL; + memset(to->bandlims, 0, sizeof(to->bandlims)); + memset(to->bandlimsout, 0, sizeof(to->bandlimsout)); + to->lastace = NULL; + memset(to->failedparents, 0, sizeof(to->failedparents)); +} + +/* the credentials of a connection, not left behind */ +static void logparamclear(unsigned char *p){ + memset(p + offsetof(struct clientparam, password), 0, sizeof(((struct clientparam *)0)->password)); + memset(p + offsetof(struct clientparam, extpassword), 0, sizeof(((struct clientparam *)0)->extpassword)); +} + +/* Flushing and rotation, and a log which failed opened again once it is + time to try. By its thread. */ +static void logthreadtick(struct logger *lg, time_t t){ + unsigned n = 0; + int failed; + + _3proxy_mutex_lock(&lg->mutex); + failed = lg->failed; + _3proxy_mutex_unlock(&lg->mutex); + if(!failed){ + if(lg->type->tick) lg->type->tick(lg, t); + return; + } + if(t - lg->failtime < LOGRETRY) return; + if(logreopen(lg)){ + lg->failtime = t; + return; + } + _3proxy_mutex_lock(&lg->mutex); + lg->failed = 0; + n = lg->faildropped; + lg->faildropped = 0; + _3proxy_mutex_unlock(&lg->mutex); + logtell(lg, "is open again", n); +} + +/* What was given to a logger while its thread runs, taken by it: new + parameters, a new rotation. Called holding its mutex, released while + the log is opened again. */ +static void logthreadtake(struct logger *lg){ + unsigned char **archiver; + unsigned char *params; + int archiverc, res; + ROTATION rotation; + + if(lg->newparams){ + params = lg->nparams; + lg->nparams = NULL; + lg->newparams = 0; + _3proxy_mutex_unlock(&lg->mutex); + if(lg->params) free(lg->params); + lg->params = params; + res = logreopen(lg); + _3proxy_mutex_lock(&lg->mutex); + lg->failed = res? 1 : 0; + if(res) lg->failtime = time(NULL); + } + if(lg->newrot){ + rotation = lg->nrotation; + archiver = lg->narchiver; + archiverc = lg->narchiverc; + lg->narchiver = NULL; + lg->narchiverc = 0; + lg->newrot = 0; + if(lg->rotation != rotation) lg->reopen = 1; + lg->rotation = rotation; + lg->rotate = lg->nrotate; + _3proxy_mutex_unlock(&lg->mutex); + logfreearchiver(lg->archiver, lg->archiverc); + lg->archiver = archiver; + lg->archiverc = archiverc; + _3proxy_mutex_lock(&lg->mutex); } } +/* A logger's thread: writes what is queued, in order, and does what is + queued to be done once the lines before it are written. It ends once + told to and the queue is empty, freeing the logger if it is gone. */ +static void *logthread(void *arg){ + struct logger *lg = (struct logger *)arg; + struct clientparam *param; + unsigned char *buf; + struct logevent *ev; + struct logslot *sl; + void (*func)(void *); + void *farg; + time_t lasttick = 0; + unsigned dropped, n, lost; + int res, failed, written = 0, gone; + +#ifndef _WIN32 + { + sigset_t mask; + + sigfillset(&mask); + pthread_sigmask(SIG_SETMASK, &mask, NULL); + } +#endif + param = (struct clientparam *)calloc(1, sizeof(*param)); + buf = (unsigned char *)malloc(CLIENTTMPBUFSIZE); + _3proxy_mutex_lock(&lg->mutex); + for(;;){ + if(lg->newparams || lg->newrot) logthreadtake(lg); + if(!(ev = lg->head)){ + if(lg->stop) break; + _3proxy_mutex_unlock(&lg->mutex); + /* nothing more to write for now: flushed */ + if(written || time(NULL) != lasttick){ + lasttick = time(NULL); + logthreadtick(lg, lasttick); + written = 0; + } + _3proxy_sem_timedlock(lg->wake, 1000); + _3proxy_mutex_lock(&lg->mutex); + continue; + } + if(ev->kind == LOGCALL){ + /* out of the queue first: what it does may free it */ + if(!(lg->head = ev->next)) lg->tail = NULL; + dropped = ev->dropped; + func = ev->func; + farg = ev->arg; + _3proxy_mutex_unlock(&lg->mutex); + (*func)(farg); + if(dropped) logtell(lg, "has its queue full", dropped); + _3proxy_mutex_lock(&lg->mutex); + continue; + } + sl = (struct logslot *)ev; + failed = lg->failed; + lg->more = ev->next != NULL; + _3proxy_mutex_unlock(&lg->mutex); + res = 0; + if(!failed && param && buf){ + if(ev->kind == LOGLINE) logparamcopy(param, LOGSLOTPARAM(sl)); + res = logout(lg, (ev->kind == LOGLINE)? param : &logparam, + sl->big? sl->big : sl->text, sl->sec, sl->msec, buf); + written = 1; + if(ev->kind == LOGLINE) logparamclear((unsigned char *)param); + } + if(ev->kind == LOGLINE) logparamclear(LOGSLOTPARAM(sl)); + if(sl->big){ + free(sl->big); + sl->big = NULL; + } + if(sl->sec != lasttick){ + lasttick = sl->sec; + logthreadtick(lg, lasttick); + } + _3proxy_mutex_lock(&lg->mutex); + if(!(lg->head = ev->next)) lg->tail = NULL; + lg->lines--; + dropped = ev->dropped; + lost = 0; + if(failed || res || !param || !buf) lg->faildropped++; + if(res == 2){ + lg->failed = 1; + lg->failtime = sl->sec; + } + else if(res == 3){ + lost = lg->faildropped; + lg->faildropped = 0; + } + if(lg->slots > logqmin){ + free(sl); + lg->slots--; + } + else { + ev->next = (struct logevent *)lg->free; + lg->free = sl; + } + _3proxy_mutex_unlock(&lg->mutex); + if(dropped) logtell(lg, "has its queue full", dropped); + if(res == 2) logtell(lg, "failed: its records are dropped until it is open again", 0); + if(res == 3) logtell(lg, "could not be written, opened again", lost); + _3proxy_mutex_lock(&lg->mutex); + } + /* from now on written as it comes, what this thread would tell too */ + lg->running = 0; + gone = lg->stop == 2; + n = lg->faildropped; + lg->faildropped = 0; + _3proxy_mutex_unlock(&lg->mutex); + if(n) logtell(lg, "is closed", n); + if(param) free(param); + if(buf) free(buf); + if(lg->type->close) lg->type->close(lg); + if(gone) logdestroy(lg); + else { + /* the process ends: what comes after is dropped */ + _3proxy_mutex_lock(&lg->mutex); + lg->failed = 1; + lg->failtime = time(NULL); + _3proxy_mutex_unlock(&lg->mutex); + } + _3proxy_mutex_lock(&log_mutex); + logworkers--; + _3proxy_mutex_unlock(&log_mutex); + _3proxy_sem_unlock(logexited); + return NULL; +} + /* The log function of the services, unless a plugin puts its own before - it: the traffic counted, the line written to the service's log, the - statistics cleared. */ + it: the traffic counted, the line queued to the service's log, or + written, the statistics cleared. */ void logdo(struct clientparam * param, const unsigned char *s){ struct logger *lg = param->srv? param->srv->logger : NULL; - int len = -1; if(param->trafcountfunc)(*param->trafcountfunc)(param); - if(lg && !param->nolog && lg->type->write){ - if(lg->type->format) len = lg->type->format(lg, param, s, param->tmpbuf); - if(len >= 0 || !lg->type->format) logwrite(lg, param, param->tmpbuf, len); - } + if(lg && !param->nolog && lg->type->write && !logqueue(lg, param, s, LOGLINE)) + logdirect(lg, param, s, param->tmpbuf); clearstat(param); } @@ -921,25 +1395,41 @@ void logdo(struct clientparam * param, const unsigned char *s){ void logerror(const unsigned char *s){ struct logger *lg; unsigned char *buf; - int len; if(!(lg = logerrget())) return; - if(lg->type->format && lg->type->write && (buf = malloc(CLIENTTMPBUFSIZE))){ - len = lg->type->format(lg, &logparam, s, buf); - if(len >= 0) logwrite(lg, &logparam, buf, len); + if(lg->type->format && lg->type->write && !logqueue(lg, NULL, s, LOGTEXT) && + (buf = malloc(CLIENTTMPBUFSIZE))){ + logdirect(lg, &logparam, s, buf); free(buf); } logput(lg); } -/* What a configuration says of its last log, once it is in force: the - rotation, and that errors go there for a daemon. */ +/* What a configuration says of logs, once it is in force: the rotation of + its last log, where errors go for a daemon, the size of the queues. The + first one in force gives the loggers threads: a daemon has forked by + now. */ void logcommit(struct extparam *c){ struct logger *lg = c->logger; unsigned char **archiver = NULL; - int i, archiverc = 0; + int i, archiverc = 0, told = 0; logerrset(lg); + logqmin = c->minlogqueue; + logqmax = c->maxlogqueue; + _3proxy_mutex_lock(&log_mutex); + if(!logthreads){ + struct logger *l; + + logthreads = 1; + for(l = loggers; l; l = l->next) logstart(l); + } + if(logstartfailed){ + logstartfailed = 0; + told = 1; + } + _3proxy_mutex_unlock(&log_mutex); + if(told) logerror((const unsigned char *)"A log has no thread of its own: one could not be started"); if(!lg || lg->type != &logtype_file) return; if(c->archiver && c->archiverc > 0 && (archiver = calloc(c->archiverc, sizeof(*archiver)))){ for(archiverc = 0; archiverc < c->archiverc; archiverc++){ @@ -952,19 +1442,30 @@ void logcommit(struct extparam *c){ archiverc = 0; } } - _3proxy_mutex_lock(&log_mutex); _3proxy_mutex_lock(&lg->mutex); + if(lg->running){ + /* for its thread to take */ + logfreearchiver(lg->narchiver, lg->narchiverc); + lg->nrotation = c->logtype; + lg->nrotate = c->rotate; + lg->narchiver = archiver; + lg->narchiverc = archiverc; + lg->newrot = 1; + _3proxy_mutex_unlock(&lg->mutex); + _3proxy_sem_unlock(lg->wake); + return; + } if(lg->rotation != c->logtype) lg->reopen = 1; lg->rotation = c->logtype; lg->rotate = c->rotate; - logfreearchiver(lg); + logfreearchiver(lg->archiver, lg->archiverc); lg->archiver = archiver; lg->archiverc = archiverc; _3proxy_mutex_unlock(&lg->mutex); - _3proxy_mutex_unlock(&log_mutex); } -/* Rotation and flushing, for every logger: from the main loop. */ +/* Rotation and flushing, from the main loop, for the loggers with no + thread: those with one do it themselves. */ void logtick(time_t t){ struct logger *lg; @@ -973,23 +1474,45 @@ void logtick(time_t t){ for(lg = loggers; lg; lg = lg->next){ if(!lg->type->tick) continue; _3proxy_mutex_lock(&lg->mutex); - lg->type->tick(lg, t); + if(!lg->running) lg->type->tick(lg, t); _3proxy_mutex_unlock(&lg->mutex); } _3proxy_mutex_unlock(&log_mutex); } -/* Closes every log as the process stops, its services done. */ +/* As the process stops, its services done: what is queued written, for + some seconds at most, and every log closed. */ void logfinish(void){ struct logger *lg; + time_t until = time(NULL) + 10; + int n; _3proxy_mutex_lock(&log_mutex); + logthreads = 0; for(lg = loggers; lg; lg = lg->next){ - if(!lg->type->close) continue; _3proxy_mutex_lock(&lg->mutex); - lg->type->close(lg); - lg->failed = 1; - lg->failtime = time(NULL); + if(lg->running && !lg->stop){ + lg->stop = 1; + _3proxy_sem_unlock(lg->wake); + } + _3proxy_mutex_unlock(&lg->mutex); + } + n = logworkers; + _3proxy_mutex_unlock(&log_mutex); + while(n > 0 && time(NULL) < until){ + _3proxy_sem_timedlock(logexited, 100); + _3proxy_mutex_lock(&log_mutex); + n = logworkers; + _3proxy_mutex_unlock(&log_mutex); + } + _3proxy_mutex_lock(&log_mutex); + for(lg = loggers; lg; lg = lg->next){ + _3proxy_mutex_lock(&lg->mutex); + if(!lg->running){ + if(lg->type->close) lg->type->close(lg); + lg->failed = 1; + lg->failtime = time(NULL); + } _3proxy_mutex_unlock(&lg->mutex); } _3proxy_mutex_unlock(&log_mutex); diff --git a/src/mainclient.c b/src/mainclient.c index e18f7de..d205429 100644 --- a/src/mainclient.c +++ b/src/mainclient.c @@ -159,14 +159,14 @@ static void freeworker(struct clientparam *param){ free(param); } -/* The automatic cap: 32 idle workers for each running service, but never - more than all of them could have connections. */ +/* The automatic cap: POOLPERSERVICE idle workers for each running service, + but never more than all of them could have connections. */ static unsigned poollimit(void){ unsigned limit; if(runconf->poolmax >= 0) limit = (unsigned)runconf->poolmax; else { - limit = 32 * tpool.services; + limit = POOLPERSERVICE * tpool.services; if(limit > tpool.maxchild) limit = tpool.maxchild; } tpool.limit = limit; diff --git a/src/mainconf.c b/src/mainconf.c index 12d365d..b4dbf74 100644 --- a/src/mainconf.c +++ b/src/mainconf.c @@ -145,13 +145,28 @@ void setopts(SOCKET s, const struct sockoptset *opts, int maxseg){ } } -/* Releases what the preparation and the run took, s itself excepted. */ -static void srvdiscard(struct srvstart *s){ - srvfree(&s->srv); +static void srvstrings(struct srvstart *s){ if(s->cbc_string) free(s->cbc_string); if(s->cbl_string) free(s->cbl_string); } +/* Releases what the preparation and the run took, s itself excepted. */ +static void srvdiscard(struct srvstart *s){ + srvfree(&s->srv); + srvstrings(s); +} + +#ifndef STDMAIN +/* A service which ran, released once its lines are written, s with it. */ +static void srvlater(void *p){ + struct srvstart *s = (struct srvstart *)p; + + srvrelease(&s->srv); + srvstrings(s); + free(s); +} +#endif + #ifndef STDMAIN static void srvunlink(struct srvparam *srv){ if(srv->next)srv->next->prev = srv->prev; @@ -673,11 +688,9 @@ int mainprepare(int argc, char** argv, void **service){ return 0; } +/* service is freed once its lines are written */ int mainrun(void *service){ - int res = srvrun((struct srvstart *)service); - - free(service); - return res; + return srvrun((struct srvstart *)service); } #endif @@ -782,7 +795,9 @@ void srvinit2(struct srvparam * srv, struct clientparam *param){ param->sinsr = srv->extsa; } -void srvfree(struct srvparam * srv){ +/* A service ended: its sockets closed, its clients gone. What its lines + refer to is kept for srvrelease(). */ +void srvstop(struct srvparam * srv){ int i; #ifndef STDMAIN @@ -810,6 +825,11 @@ void srvfree(struct srvparam * srv){ for(i = srv->maxchild; i; i--) _3proxy_sem_lock(srv->childslots); _3proxy_mutex_lock(&srv->counter_mutex); _3proxy_mutex_unlock(&srv->counter_mutex); +} + +/* What a stopped service holds: once the lines logged for it, which refer + to it and to its configuration, are written. */ +void srvrelease(struct srvparam * srv){ #ifndef STDMAIN if(srv->filter){ while(srv->nfilters){ @@ -861,6 +881,11 @@ void srvfree(struct srvparam * srv){ #endif } +void srvfree(struct srvparam * srv){ + srvstop(srv); + srvrelease(srv); +} + #ifndef STDMAIN static void * itcopy (void * from, size_t size){ void * ret; diff --git a/src/proxy.h b/src/proxy.h index 08cda2b..c738c0e 100644 --- a/src/proxy.h +++ b/src/proxy.h @@ -34,7 +34,22 @@ #define MAXUSERNAME 128 #define _PASSWORD_LEN 256 +/* Defaults, -D at build time sets others: maxconn; threadpool, the idle + workers kept (-1 automatic: POOLPERSERVICE for each service, no more than + their maxconn together) and how long one waits, in seconds. Each idle + worker keeps its thread and its struct clientparam. */ +#ifndef DEFAULT_MAXCHILD #define DEFAULT_MAXCHILD 500 +#endif +#ifndef DEFAULT_POOLMAX +#define DEFAULT_POOLMAX -1 +#endif +#ifndef DEFAULT_POOLTIMEOUT +#define DEFAULT_POOLTIMEOUT 60 +#endif +#ifndef POOLPERSERVICE +#define POOLPERSERVICE 32 +#endif /* How much of a connection's buffer is read into at once: all of it, or less if the service was told so with -b. */ @@ -342,6 +357,7 @@ void dolog(struct clientparam * param, const unsigned char *s); int dobuf(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec); int dobuf2(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec, struct tm* tm, char * format); int logbuf(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec); +int logbuft(struct clientparam * param, unsigned char * buf, const unsigned char *s, const unsigned char * doublec, time_t sec, unsigned msec); void logdo(struct clientparam * param, const unsigned char *s); void logerror(const unsigned char *s); void loginit(void); @@ -355,6 +371,7 @@ void logstdclose(void); size_t logselen(const unsigned char *selector); void logtick(time_t t); void logfinish(void); +void logcall(struct logger *lg, struct logevent *ev, void (*func)(void *), void *arg); int timechanged (time_t oldtime, time_t newtime, ROTATION lt); extern struct logger lognull; #ifdef WITH_ODBC @@ -500,6 +517,8 @@ FILTER_ACTION handledatfltsrv(struct clientparam *param, unsigned char ** buf_p, void srvinit(struct srvparam * srv, struct clientparam *param); void srvinit2(struct srvparam * srv, struct clientparam *param); void srvfree(struct srvparam * srv); +void srvstop(struct srvparam * srv); +void srvrelease(struct srvparam * srv); unsigned char * dologname (unsigned char *buf, unsigned char *name, const unsigned char *ext, ROTATION lt, time_t t); int readconfig(FILE * fp); void initcommands(void); diff --git a/src/proxymain.c b/src/proxymain.c index bf00a3c..b38afe3 100644 --- a/src/proxymain.c +++ b/src/proxymain.c @@ -50,6 +50,10 @@ struct udpstate; Their functions of file scope are declared here, so that they can be in any order. */ static void srvdiscard(struct srvstart *s); +static void srvstrings(struct srvstart *s); +#ifndef STDMAIN +static void srvlater(void *p); +#endif static void srvunlink(struct srvparam *srv); static int srvprepare(struct srvstart *s, int argc, char **argv); static int samestr(const char *a, const char *b); @@ -163,12 +167,19 @@ EXIT: #endif if(!srv->silent) srv->logfunc(defparam, (unsigned char *)"Exiting thread"); - /* srvfree() waits for every slot to come back, this one included */ + /* srvstop() waits for every slot to come back, this one included */ if(haveslot) _3proxy_sem_unlock(srv->childslots); - srvdiscard(s); + srvstop(srv); #ifndef _WIN32 pthread_attr_destroy(&s->pa); +#endif +#ifndef STDMAIN + /* freed, s with it, once its log has written the lines which refer to it */ + logcall(srv->logger, &srv->logrelease, srvlater, s); +#else + srvrelease(srv); + srvstrings(s); #endif return res; } diff --git a/src/sql.c b/src/sql.c index 3289a37..0f0fd99 100644 --- a/src/sql.c +++ b/src/sql.c @@ -105,8 +105,8 @@ static int odbcinit(struct logger *lg){ return odbcconnect(lg)? 0 : -1; } -static int odbcformat(struct logger *lg, struct clientparam *param, const unsigned char *s, unsigned char *buf){ - return logbuf(param, buf, s, (unsigned char *)"\'"); +static int odbcformat(struct logger *lg, struct clientparam *param, const unsigned char *s, time_t sec, unsigned msec, unsigned char *buf){ + return logbuft(param, buf, s, (unsigned char *)"\'", sec, msec); } static int odbcwrite(struct logger *lg, struct clientparam *param, const unsigned char *buf, int len){ diff --git a/src/structures.h b/src/structures.h index 4f40749..a7744ad 100644 --- a/src/structures.h +++ b/src/structures.h @@ -204,6 +204,14 @@ int #define UDPBUFSIZE 16384 #define CLIENTBUFSIZE (16384 + 64) #define CLIENTTMPBUFSIZE 8192 +/* The defaults of minlogqueue and maxlogqueue: the lines a log's queue + keeps slots for, and holds at most; -D at build time sets others. */ +#ifndef LOGQUEUEMIN +#define LOGQUEUEMIN 64 +#endif +#ifndef LOGQUEUEMAX +#define LOGQUEUEMAX 4096 +#endif #define HOSTNAMELEN 256 #define USERNAMELEN 128 @@ -638,20 +646,35 @@ struct logtype { /* opens what it writes to, lg->data: 0, an error (an errno value), or < 0 if it cannot be reached now but is to be tried again */ int (*init)(struct logger *lg); - /* the line for a connection into buf, CLIENTTMPBUFSIZE: its length, -1 - for nothing to write; NULL if write() takes the connection itself */ - int (*format)(struct logger *lg, struct clientparam *param, const unsigned char *s, unsigned char *buf); + /* the line for a connection, logged at sec and msec, into buf, + CLIENTTMPBUFSIZE: its length, -1 for nothing to write; NULL if + write() takes the connection itself */ + int (*format)(struct logger *lg, struct clientparam *param, const unsigned char *s, time_t sec, unsigned msec, unsigned char *buf); /* 0, or an error */ int (*write)(struct logger *lg, struct clientparam *param, const unsigned char *buf, int len); void (*close)(struct logger *lg); /* every second or so: rotation, flushing */ void (*tick)(struct logger *lg, time_t t); - int stacksize; /* the client thread stack its writing needs */ + int stacksize; /* the stack its writing needs */ int standard; /* kept open for the life of the process */ int text; /* writes a line of text: errors can go there */ int params; /* the string after a comma is parameters */ }; +/* What a log's queue holds: a line (log.c's struct logslot, of which this + is the head), or something to do once the lines before it are written, + such as freeing a service they refer to. */ +struct logevent { + struct logevent *next; + int kind; + unsigned dropped; /* lines dropped after this one, the queue full */ + void (*func)(void *arg); + void *arg; +}; + +/* Once 3proxy runs, a logger has a thread of its own, which writes what its + queue holds: the clients do not wait for a log. Until then, and in a + standalone service, lines are written as they come. */ struct logger { struct logger *next; const struct logtype *type; @@ -660,20 +683,38 @@ struct logger { void *data; /* the type's own */ int refs; int errors; /* writes failed in a row */ - int failed; /* could not be reopened: lines go to standard error */ + int failed; /* could not be reopened: its lines are dropped */ time_t failtime; + unsigned faildropped; /* lines dropped as it failed, not yet told */ ROTATION rotation; int rotate, archiverc; unsigned char **archiver; time_t rotated; /* when the file in use was opened */ int reopen; /* the rotation changed: the file name too */ - _3proxy_mutex_t mutex; /* writing, reopening */ + _3proxy_mutex_t mutex; /* the queue; writing, while there is no thread */ + /* the queue: the lines in it counted against maxlogqueue, the slots + allocated kept down to minlogqueue as they come back */ + struct logevent *head, *tail; + void *free; + int lines, slots; + _3proxy_sem_t wake; + int running, /* has its thread */ + stop, /* its thread is to end once the queue is empty */ + more; /* for its thread: lines follow the one written */ + /* given while its thread runs, for the thread to take */ + int newrot, newparams; + ROTATION nrotation; + int nrotate, narchiverc; + unsigned char **narchiver; + unsigned char *nparams; }; struct srvparam { struct sockfuncs so; struct srvparam *next; struct srvparam *prev; + /* freeing it, queued to its log behind the lines which refer to it */ + struct logevent logrelease; struct listener *listener; /* its listening socket, which a newer service may take over on a reload */ struct srvparam *lisnext; /* the next service on the listener */ @@ -941,6 +982,7 @@ struct extparam { struct srvparam *services; int stacksize, counterd, rotate, paused, archiverc, + minlogqueue, maxlogqueue, demon, maxchild, backlog, version, noforce, bandlimver, parentretries; unsigned authcachetype, authcachetime; int filtermaxsize; diff --git a/tests/cases/logs.py b/tests/cases/logs.py index bd189c8..428039c 100644 --- a/tests/cases/logs.py +++ b/tests/cases/logs.py @@ -44,6 +44,81 @@ def _opened(pid, path): return n +def _stalled(t): + """A log nobody reads: a FIFO whose pipe is full. Clients are not held + up, the lines past the queue are dropped and counted, and lines queued + for a service a reload removed are still written as it formatted them.""" + fifo = os.path.join(t.tmpdir, "logs-fifo") + os.mkfifo(fifo) + # opened for reading before 3proxy opens it for writing, then not read + rfd = os.open(fifo, os.O_RDONLY | os.O_NONBLOCK) + origin, port, adm = t.free_port(), t.free_port(), t.free_port() + + # lines of 2 KB, made so by the service's format: the pipe's 64 KB fill + # before the requests are done + pad = "z" * 2000 + + def conf(proxyline): + return f""" + log + maxlogqueue 8 + auth iponly + allow * + http data * /data + httpsrv -p{origin} + flush + admin -p{adm} + flush + logformat "LQ {pad} %T" + auth iponly + allow * + {proxyline} + """ + + server = t.start("logs_stall", conf(f'proxy -p{port} -l{fifo}'), + ports=[origin, port, adm]) + began = time.time() + ok = 0 + for i in range(120): + r = t.http(f"http://127.0.0.1:{origin}/data?size=10&n{i:04d}", + proxy=f"127.0.0.1:{port}") + ok += r.status == 200 + t.eq(120, ok, "a log nobody reads holds no client up") + t.eq(True, time.time() - began < 60, "and they are not slowed to its pace") + # the service goes: its lines still queued are written as it made them + t.write_config("logs_stall", conf("flush")) + t.http(f"http://127.0.0.1:{adm}/R") + got = b"" + deadline = time.time() + 15 + while time.time() < deadline and "has its queue full" not in server.output(): + try: + chunk = os.read(rfd, 65536) + except BlockingIOError: + chunk = b"" + got += chunk + if not chunk: + time.sleep(0.05) + t.eq(True, "has its queue full" in server.output(), + "the lines past its queue are dropped, and said so") + # what is left in the pipe and the queue, read to the end + deadline = time.time() + 5 + while time.time() < deadline: + try: + chunk = os.read(rfd, 65536) + except BlockingIOError: + chunk = b"" + if chunk: + got += chunk + deadline = time.time() + 1 + else: + time.sleep(0.05) + os.close(rfd) + lines = got.decode("utf-8", "replace").splitlines() + t.eq(True, 0 < len(lines) < 120, "some lines are written, not all") + t.eq(True, all(l.startswith("Q " + pad) for l in lines), + "each as its service formats it, the service gone or not") + + def config(origin, p1, p2, p3, adm, main, shared=None): """p2 and p3 log to shared with -l, if it is given, or are not there.""" lines = [f"log {main}", 'logformat "LLOG %N %T"', "auth iponly", "allow *", @@ -108,7 +183,7 @@ def run(t): t.eq(True, _wait(lambda: _proxied(dated, "five")), "goes to the dated file") t.eq(False, _proxied(second, "five"), "not to the undated one") - # --- a log which fails ------------------------------------------------ + # --- a log which fails: its lines dropped, and said so ----------------- if os.path.exists("/dev/full"): o2, q = t.free_port(), t.free_port() full = t.start("logs_full", f""" @@ -124,8 +199,14 @@ def run(t): for i in range(80): t.http(f"http://127.0.0.1:{o2}/data?size=10&{'x' * 40}{i}", proxy=f"127.0.0.1:{q}") - t.eq(True, _wait(lambda: "LOG PROXY" in full.output()), - "the lines a full device refuses go to standard error") + t.eq(True, _wait(lambda: "Log /dev/full could not be written" in full.output()), + "a log a full device refuses says so on standard error") + t.eq(True, "records dropped" in full.output(), "with the records it dropped") + t.eq(False, "LOG PROXY" in full.output(), "its lines are not written elsewhere") + + # --- a log which stalls: clients go on, the queue is bounded ---------- + if hasattr(os, "mkfifo"): + _stalled(t) # --- a log which cannot be opened is an error in the configuration --- out = t.run_config("logs_bad", f"""