int verbosity = DEFAULT_VERBOSITY; /* Logging level */
struct CitContext masterCC;
-int rescan[2]; /* The Rescan Pipe */
time_t last_purge = 0; /* Last dead session purge */
static int num_threads = 0; /* Current number of threads */
int num_sessions = 0; /* Current number of sessions */
-fd_set masterfds; /* Master sockets etc. */
-int masterhighest;
-
pthread_t initial_thread; /* tid for main() thread */
int syslog_facility = (-1);
* Purge all sessions which have the 'kill_me' flag set.
* This function has code to prevent it from running more than once every
* few seconds, because running it after every single unbind would waste a lot
- * of CPU time and keep the context list locked too much.
+ * of CPU time and keep the context list locked too much. To force it to run
+ * anyway, set "force" to nonzero.
+ *
*
- * After that's done, we raise or lower the size of the worker thread pool
+ * After that's done, we raise the size of the worker thread pool
* if such an action is appropriate.
*/
-void dead_session_purge(void) {
+void dead_session_purge(int force) {
struct CitContext *ptr, *rem;
- if ( (time(NULL) - last_purge) < 5 ) return; /* Too soon, go away */
+ if (force == 0) {
+ if ( (time(NULL) - last_purge) < 5 ) {
+ return; /* Too soon, go away */
+ }
+ }
time(&last_purge);
do {
-/*
- * Set up a fd_set containing all the master sockets to which we
- * always listen. It's computationally less expensive to just copy
- * this to a local fd_set when starting a new select() and then add
- * the client sockets than it is to initialize a new one and then
- * figure out what to put there.
- */
-void init_master_fdset(void) {
- struct ServiceFunctionHook *serviceptr;
- int m;
-
- lprintf(CTDL_DEBUG, "Initializing master fdset\n");
-
- FD_ZERO(&masterfds);
- masterhighest = 0;
-
- lprintf(CTDL_DEBUG, "Will listen on rescan pipe %d\n", rescan[0]);
- FD_SET(rescan[0], &masterfds);
- if (rescan[0] > masterhighest) masterhighest = rescan[0];
- for (serviceptr = ServiceHookTable; serviceptr != NULL;
- serviceptr = serviceptr->next ) {
- m = serviceptr->msock;
- lprintf(CTDL_DEBUG, "Will listen on master socket %d\n", m);
- FD_SET(m, &masterfds);
- if (m > masterhighest) {
- masterhighest = m;
- }
- }
- lprintf(CTDL_DEBUG, "masterhighest = %d\n", masterhighest);
-}
/*
*/
void *worker_thread(void *arg) {
int i;
- char junk;
int highest;
- /* This is synchronized below; it helps implement round robin mode */
- static struct CitContext* next_session = NULL;
struct CitContext *ptr;
struct CitContext *bind_me = NULL;
fd_set readfds;
struct ServiceFunctionHook *serviceptr;
int ssock; /* Descriptor for client socket */
struct timeval tv;
+ int force_purge = 0;
+ int m;
num_threads++;
* which might cause a deadlock.
*/
cdb_check_handles();
+ force_purge = 0;
+ bind_me = NULL; /* Which session shall we handle? */
begin_critical_section(S_I_WANNA_SELECT);
-SETUP_FD: memcpy(&readfds, &masterfds, sizeof masterfds);
- highest = masterhighest;
+
+ /* Initialize the fdset. */
+ FD_ZERO(&readfds);
+ highest = 0;
+
begin_critical_section(S_SESSION_TABLE);
for (ptr = ContextList; ptr != NULL; ptr = ptr->next) {
if (ptr->state == CON_IDLE) {
if (ptr->client_socket > highest)
highest = ptr->client_socket;
}
+ if ((bind_me == NULL) && (ptr->state == CON_READY)) {
+ bind_me = ptr;
+ ptr->state = CON_EXECUTING;
+ }
}
end_critical_section(S_SESSION_TABLE);
- tv.tv_sec = 1; /* wake up every second if no input */
- tv.tv_usec = 0;
+ if (bind_me) goto SKIP_SELECT;
+
+do_select: /* Only select() if the server isn't being told to shut down. */
- do_select:
- if (!time_to_die)
+ /* Add the various master sockets to the fdset. */
+ for (serviceptr = ServiceHookTable; serviceptr != NULL;
+ serviceptr = serviceptr->next ) {
+ m = serviceptr->msock;
+ FD_SET(m, &readfds);
+ if (m > highest) {
+ highest = m;
+ }
+ }
+
+ if (!time_to_die) {
+ tv.tv_sec = 1; /* wake up every second if no input */
+ tv.tv_usec = 0;
retval = select(highest + 1, &readfds, NULL, NULL, &tv);
+ }
else {
end_critical_section(S_I_WANNA_SELECT);
break;
* First, check for an error or exit condition.
*/
if (retval < 0) {
+ if (errno == EBADF) {
+ lprintf(CTDL_NOTICE, "select() failed: (%s)\n",
+ strerror(errno));
+ goto do_select;
+ }
if (errno != EINTR) {
lprintf(CTDL_EMERG, "Exiting (%s)\n", strerror(errno));
time_to_die = 1;
con->client_socket = ssock;
con->h_command_function =
serviceptr->h_command_function;
+ con->h_async_function =
+ serviceptr->h_async_function;
/* Determine whether local socket */
if (serviceptr->sockpath != NULL)
serviceptr->h_greeting_function();
become_session(NULL);
con->state = CON_IDLE;
- goto SETUP_FD;
+ goto do_select;
}
}
}
break;
}
- /* If the rescan pipe went active, someone is telling this
- * thread that the &readfds needs to be refreshed with more
- * current data.
- */
- if (FD_ISSET(rescan[0], &readfds)) {
- read(rescan[0], &junk, 1);
- goto SETUP_FD;
- }
-
/* It must be a client socket. Find a context that has data
- * waiting on its socket *and* is in the CON_IDLE state.
+ * waiting on its socket *and* is in the CON_IDLE state. Any
+ * active sockets other than our chosen one are marked as
+ * CON_READY so the next thread that comes around can just bind
+ * to one without having to select() again.
*/
- else {
- bind_me = NULL;
- begin_critical_section(S_SESSION_TABLE);
- /*
- * We start where we left off. If we get to the end
- * we'll start from the beginning again, then give up
- * if we still don't find anything. This ensures
- * that all contexts get a more-or-less equal chance
- * to run. And yes, I did add a goto to the code. -IO
- */
-find_session: if (next_session == NULL)
- next_session = ContextList;
- for (ptr = next_session;
- ( (ptr != NULL) && (bind_me == NULL) );
- ptr = ptr->next) {
- if ( (FD_ISSET(ptr->client_socket, &readfds))
- && (ptr->state == CON_IDLE) ) {
- bind_me = ptr;
+ begin_critical_section(S_SESSION_TABLE);
+ for (ptr = ContextList; ptr != NULL; ptr = ptr->next) {
+ if ( (FD_ISSET(ptr->client_socket, &readfds))
+ && (ptr->state != CON_EXECUTING) ) {
+ ptr->input_waiting = 1;
+ if (!bind_me) {
+ bind_me = ptr; /* I choose you! */
+ bind_me->state = CON_EXECUTING;
+ }
+ else {
+ ptr->state = CON_READY;
}
}
- if (bind_me != NULL) {
- /* Found one. Stake a claim to it before
- * letting anyone else touch the context list.
- */
- bind_me->state = CON_EXECUTING;
- next_session = bind_me->next;
- } else if (next_session == ContextList) {
- next_session = NULL;
- }
- if (bind_me == NULL && next_session != NULL) {
- next_session = NULL;
- goto find_session;
- }
+ }
+ end_critical_section(S_SESSION_TABLE);
+SKIP_SELECT: end_critical_section(S_I_WANNA_SELECT);
- end_critical_section(S_SESSION_TABLE);
- end_critical_section(S_I_WANNA_SELECT);
+ /* We're bound to a session */
+ if (bind_me != NULL) {
+ become_session(bind_me);
- /* We're bound to a session, now do *one* command */
- if (bind_me != NULL) {
- become_session(bind_me);
+ /* If the client has sent a command, execute it. */
+ if (CC->input_waiting) {
CC->h_command_function();
- become_session(NULL);
- bind_me->state = CON_IDLE;
- write(rescan[1], &junk, 1);
+ CC->input_waiting = 0;
}
+ /* If there are asynchronous messages waiting and the
+ * client supports it, do those now */
+ if ((CC->is_async) && (CC->async_waiting)
+ && (CC->h_async_function != NULL)) {
+ CC->h_async_function();
+ CC->async_waiting = 0;
+ }
+
+ force_purge = CC->kill_me;
+ become_session(NULL);
+ bind_me->state = CON_IDLE;
}
- dead_session_purge();
+
+ dead_session_purge(force_purge);
do_housekeeping();
check_sched_shutdown();
}
projects / citadel.git / blobdiff