X-Git-Url: https://code.citadel.org/?a=blobdiff_plain;f=citadel%2Fmodules%2Feventclient%2Fserv_eventclient.c;h=75e1f785ae905ccecb2c34317ea887e7252ff4b3;hb=3228daa54e93fe68aeb0e08a428795fa6dcc0d28;hp=b0d9a3c3257642a1a3cb11e553708f40f7931fc8;hpb=70d3a7532a0c3a599d5f72f41f408f4d5fe97957;p=citadel.git diff --git a/citadel/modules/eventclient/serv_eventclient.c b/citadel/modules/eventclient/serv_eventclient.c index b0d9a3c32..75e1f785a 100644 --- a/citadel/modules/eventclient/serv_eventclient.c +++ b/citadel/modules/eventclient/serv_eventclient.c @@ -134,13 +134,20 @@ gotstatus(int nnrun) CURLINFO_PRIVATE, &chandle); if (sta) { - EVCURL_syslog(LOG_ERR, - "error asking curl for private" - " cookie of curl handle: %s\n", - curl_easy_strerror(sta)); + syslog(LOG_ERR, + "error asking curl for private" + " cookie of curl handle: %s\n", + curl_easy_strerror(sta)); continue; } IO = (AsyncIO *)chandle; + if (IO->ID == 0) { + EVCURL_syslog(LOG_ERR, + "Error, invalid IO context %p\n", + IO); + continue; + } + SetEVState(IO, eCurlGotStatus); EVCURLM_syslog(LOG_DEBUG, "request complete\n"); @@ -166,7 +173,7 @@ gotstatus(int nnrun) "error asking curl for " "response code from request: %s\n", curl_easy_strerror(sta)); - EVCURL_syslog(LOG_ERR, + EVCURL_syslog(LOG_DEBUG, "http response code was %ld\n", (long)IO->HttpReq.httpcode); @@ -186,9 +193,6 @@ gotstatus(int nnrun) switch(IO->SendDone(IO)) { case eDBQuery: - curl_easy_cleanup(IO->HttpReq.chnd); - IO->HttpReq.chnd = NULL; - break; case eSendDNSQuery: case eReadDNSReply: case eConnect: @@ -199,8 +203,6 @@ gotstatus(int nnrun) case eReadMore: case eReadPayload: case eReadFile: - curl_easy_cleanup(IO->HttpReq.chnd); - IO->HttpReq.chnd = NULL; break; case eTerminateConnection: case eAbort: @@ -266,9 +268,11 @@ got_out(struct ev_loop *loop, ev_io *ioev, int events) } static size_t -gotdata(void *data, size_t size, size_t nmemb, void *cglobal) { +gotdata(void *data, size_t size, size_t nmemb, void *cglobal) +{ AsyncIO *IO = (AsyncIO*) cglobal; + SetEVState(IO, eCurlGotData); if (IO->HttpReq.ReplyData == NULL) { IO->HttpReq.ReplyData = NewStrBufPlain(NULL, SIZ); @@ -310,13 +314,14 @@ gotwatchsock(CURL *easy, if (IO == NULL) { sta = curl_easy_getinfo(easy, CURLINFO_PRIVATE, &f); if (sta) { - EVCURL_syslog(LOG_ERR, - "EVCURL: error asking curl for private " - "cookie of curl handle: %s\n", - curl_easy_strerror(sta)); + CURL_syslog(LOG_ERR, + "EVCURL: error asking curl for private " + "cookie of curl handle: %s\n", + curl_easy_strerror(sta)); return -1; } IO = (AsyncIO *) f; + SetEVState(IO, eCurlNewIO); EVCURL_syslog(LOG_DEBUG, "EVCURL: got socket for URL: %s\n", IO->ConnectMe->PlainUrl); @@ -332,6 +337,7 @@ gotwatchsock(CURL *easy, curl_multi_assign(mhnd, fd, IO); } + SetEVState(IO, eCurlGotIO); IO->Now = ev_now(event_base); Action = ""; @@ -486,6 +492,11 @@ static void IOcurl_abort_shutdown_callback(struct ev_loop *loop, { CURLMcode msta; AsyncIO *IO = watcher->data; + + if (IO == NULL) + return; + + SetEVState(IO, eCurlShutdown); IO->Now = ev_now(event_base); EVCURL_syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__); @@ -514,6 +525,7 @@ evcurl_handle_start(AsyncIO *IO) CURLcode sta; CURL *chnd; + SetEVState(IO, eCurlStart); chnd = IO->HttpReq.chnd; EVCURL_syslog(LOG_DEBUG, "EVCURL: Loading URL: %s\n", IO->ConnectMe->PlainUrl); @@ -549,10 +561,12 @@ evcurl_handle_start(AsyncIO *IO) IO->HttpReq.attached = 1; ev_async_send (event_base, &WakeupCurl); + ev_cleanup_init(&IO->abort_by_shutdown, IOcurl_abort_shutdown_callback); ev_cleanup_start(event_base, &IO->abort_by_shutdown); + return eReadMessage; } @@ -577,17 +591,19 @@ static void evcurl_shutdown (void) * this currently is the main loop (which may change in some future?) */ int evbase_count = 0; -int event_add_pipe[2] = {-1, -1}; pthread_mutex_t EventQueueMutex; /* locks the access to the following vars: */ +pthread_mutex_t EventExitQueueMutex; /* locks the access to the event queue pointer on exit. */ HashList *QueueEvents = NULL; HashList *InboundEventQueue = NULL; HashList *InboundEventQueues[2] = { NULL, NULL }; +extern void ShutDownCLient(AsyncIO *IO); ev_async AddJob; ev_async ExitEventLoop; static void QueueEventAddCallback(EV_P_ ev_async *w, int revents) { + CitContext *Ctx; ev_tstamp Now; HashList *q; void *v; @@ -617,8 +633,32 @@ static void QueueEventAddCallback(EV_P_ ev_async *w, int revents) } if (h->IO->StartIO == 0.0) h->IO->StartIO = Now; + + SetEVState(h->IO, eIOAttach); + + Ctx = h->IO->CitContext; + become_session(Ctx); + h->IO->Now = Now; - h->EvAttch(h->IO); + switch (h->EvAttch(h->IO)) + { + case eReadMore: + case eReadMessage: + case eReadFile: + case eSendReply: + case eSendMore: + case eReadPayload: + case eSendFile: + case eDBQuery: + case eSendDNSQuery: + case eReadDNSReply: + case eConnect: + break; + case eTerminateConnection: + case eAbort: + ShutDownCLient(h->IO); + break; + } } DeleteHashPos(&It); DeleteHashContent(&q); @@ -637,19 +677,8 @@ static void EventExitCallback(EV_P_ ev_async *w, int revents) void InitEventQueue(void) { - struct rlimit LimitSet; - pthread_mutex_init(&EventQueueMutex, NULL); - - if (pipe(event_add_pipe) != 0) { - syslog(LOG_EMERG, - "Unable to create pipe for libev queueing: %s\n", - strerror(errno)); - abort(); - } - LimitSet.rlim_cur = 1; - LimitSet.rlim_max = 1; - setrlimit(event_add_pipe[1], &LimitSet); + pthread_mutex_init(&EventExitQueueMutex, NULL); QueueEvents = NewHash(1, Flathash); InboundEventQueues[0] = NewHash(1, Flathash); @@ -685,19 +714,20 @@ void *client_event_thread(void *arg) EVQM_syslog(LOG_DEBUG, "client_event_thread() exiting\n"); ///what todo here? CtdlClearSystemContext(); + pthread_mutex_lock(&EventExitQueueMutex); ev_loop_destroy (EV_DEFAULT_UC); + event_base = NULL; DeleteHash(&QueueEvents); InboundEventQueue = NULL; DeleteHash(&InboundEventQueues[0]); DeleteHash(&InboundEventQueues[1]); /* citthread_mutex_destroy(&EventQueueMutex); TODO */ evcurl_shutdown(); - close(event_add_pipe[0]); - close(event_add_pipe[1]); CtdlDestroyEVCleanupHooks(); - EVQShutDown = 1; + pthread_mutex_unlock(&EventExitQueueMutex); + EVQShutDown = 1; return(NULL); } @@ -708,8 +738,8 @@ void *client_event_thread(void *arg) */ ev_loop *event_db; int evdb_count = 0; -int evdb_add_pipe[2] = {-1, -1}; pthread_mutex_t DBEventQueueMutex; /* locks the access to the following vars: */ +pthread_mutex_t DBEventExitQueueMutex; /* locks the access to the db-event queue pointer on exit. */ HashList *DBQueueEvents = NULL; HashList *DBInboundEventQueue = NULL; HashList *DBInboundEventQueues[2] = { NULL, NULL }; @@ -721,6 +751,7 @@ extern void ShutDownDBCLient(AsyncIO *IO); static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents) { + CitContext *Ctx; ev_tstamp Now; HashList *q; void *v; @@ -752,6 +783,11 @@ static void DBQueueEventAddCallback(EV_P_ ev_async *w, int revents) if (h->IO->StartDB == 0.0) h->IO->StartDB = Now; h->IO->Now = Now; + + SetEVState(h->IO, eDBAttach); + Ctx = h->IO->CitContext; + become_session(Ctx); + ev_cleanup_start(event_db, &h->IO->db_abort_by_shutdown); rc = h->EvAttch(h->IO); switch (rc) { @@ -777,17 +813,8 @@ static void DBEventExitCallback(EV_P_ ev_async *w, int revents) void DBInitEventQueue(void) { - struct rlimit LimitSet; - pthread_mutex_init(&DBEventQueueMutex, NULL); - - if (pipe(evdb_add_pipe) != 0) { - EVQ_syslog(LOG_EMERG, "Unable to create pipe for libev queueing: %s\n", strerror(errno)); - abort(); - } - LimitSet.rlim_cur = 1; - LimitSet.rlim_max = 1; - setrlimit(evdb_add_pipe[1], &LimitSet); + pthread_mutex_init(&DBEventExitQueueMutex, NULL); DBQueueEvents = NewHash(1, Flathash); DBInboundEventQueues[0] = NewHash(1, Flathash); @@ -800,13 +827,14 @@ void DBInitEventQueue(void) */ void *db_event_thread(void *arg) { + ev_loop *tmp; struct CitContext libev_msg_CC; CtdlFillSystemContext(&libev_msg_CC, "LibEv DB IO Thread"); -// citthread_setspecific(MyConKey, (void *)&smtp_queue_CC); + EVQM_syslog(LOG_DEBUG, "dbevent_thread() initializing\n"); - event_db = ev_loop_new (EVFLAG_AUTO); + tmp = event_db = ev_loop_new (EVFLAG_AUTO); ev_async_init(&DBAddJob, DBQueueEventAddCallback); ev_async_start(event_db, &DBAddJob); @@ -815,20 +843,20 @@ void *db_event_thread(void *arg) ev_run (event_db, 0); - EVQM_syslog(LOG_DEBUG, "dbevent_thread() exiting\n"); + pthread_mutex_lock(&DBEventExitQueueMutex); -//// what to do here? CtdlClearSystemContext(); - ev_loop_destroy (event_db); + event_db = NULL; + EVQM_syslog(LOG_INFO, "dbevent_thread() exiting\n"); DeleteHash(&DBQueueEvents); DBInboundEventQueue = NULL; DeleteHash(&DBInboundEventQueues[0]); DeleteHash(&DBInboundEventQueues[1]); - close(evdb_add_pipe[0]); - close(evdb_add_pipe[1]); /* citthread_mutex_destroy(&DBEventQueueMutex); TODO */ + ev_loop_destroy (tmp); + pthread_mutex_unlock(&DBEventExitQueueMutex); return(NULL); }