/*****************************************************************************
* libevent / curl integration *
*****************************************************************************/
-#define MOPT(s, v)\
+#define DBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (DebugCurl != 0))
+
+#define EVCURL_syslog(LEVEL, FORMAT, ...) \
+ DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT, \
+ IO->ID, CCID, __VA_ARGS__)
+
+#define EVCURLM_syslog(LEVEL, FORMAT) \
+ DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT, \
+ IO->ID, CCID)
+
+#define CURL_syslog(LEVEL, FORMAT, ...) \
+ DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT, __VA_ARGS__)
+
+#define CURLM_syslog(LEVEL, FORMAT) \
+ DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT)
+
+#define MOPT(s, v) \
do { \
sta = curl_multi_setopt(mhnd, (CURLMOPT_##s), (v)); \
if (sta) { \
- syslog(LOG_ERR, "EVCURL: error setting option " \
+ EVQ_syslog(LOG_ERR, "error setting option " \
#s " on curl multi handle: %s\n", \
curl_easy_strerror(sta)); \
exit (1); \
} \
} while (0)
-#define DBGLOG(LEVEL) if ((LEVEL != LOG_DEBUG) || (DebugCurl != 0))
-
-#define EVCURL_syslog(LEVEL, FORMAT, ...) \
- DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT, IO->ID, CCID, __VA_ARGS__)
-
-#define EVCURLM_syslog(LEVEL, FORMAT) \
- DBGLOG (LEVEL) syslog(LEVEL, "EVCURL:IO[%ld]CC[%d] " FORMAT, IO->ID, CCID)
-
-#define CURL_syslog(LEVEL, FORMAT, ...) \
- DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT, __VA_ARGS__)
-
-#define CURLM_syslog(LEVEL, FORMAT) \
- DBGLOG (LEVEL) syslog(LEVEL, "CURL: " FORMAT)
typedef struct _evcurl_global_data {
int magic;
sta = curl_easy_getinfo(chnd,
CURLINFO_PRIVATE,
&chandle);
- EVCURLM_syslog(LOG_DEBUG, "request complete\n");
if (sta) {
EVCURL_syslog(LOG_ERR,
"error asking curl for private"
continue;
}
IO = (AsyncIO *)chandle;
+ if (IO->ID == 0) {
+ EVCURL_syslog(LOG_ERR,
+ "Error, invalid IO context %p\n",
+ IO);
+ continue;
+ }
+
+ EVCURLM_syslog(LOG_DEBUG, "request complete\n");
IO->Now = ev_now(event_base);
"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);
"%s\n",
curl_multi_strerror(msta));
- ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
+ ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
IO->HttpReq.attached = 0;
switch(IO->SendDone(IO))
switch (action)
{
case CURL_POLL_NONE:
- EVCURLM_syslog(LOG_ERR,
+ EVCURLM_syslog(LOG_DEBUG,
"called first time "
"to register this sockwatcker\n");
break;
case CURL_POLL_REMOVE:
- EVCURLM_syslog(LOG_ERR,
+ EVCURLM_syslog(LOG_DEBUG,
"called last time to unregister "
"this sockwatcher\n");
ev_io_stop(event_base, &IO->recv_event);
{
CURLMcode msta;
AsyncIO *IO = watcher->data;
+
+ if (IO == NULL)
+ return;
IO->Now = ev_now(event_base);
EVCURL_syslog(LOG_DEBUG, "EVENT Curl: %s\n", __FUNCTION__);
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;
}
* 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 };
}
if (h->IO->StartIO == 0.0)
h->IO->StartIO = Now;
+
h->IO->Now = Now;
h->EvAttch(h->IO);
}
DeleteHashPos(&It);
DeleteHashContent(&q);
- syslog(LOG_DEBUG, "EVENT Q Add done.\n");
+ EVQM_syslog(LOG_DEBUG, "EVENT Q Add done.\n");
}
{
ev_break(event_base, EVBREAK_ALL);
- syslog(LOG_DEBUG, "EVENT Q exiting.\n");
+ EVQM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
}
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);
CtdlFillSystemContext(&libev_client_CC, "LibEv Thread");
// citthread_setspecific(MyConKey, (void *)&smtp_queue_CC);
- syslog(LOG_DEBUG, "client_event_thread() initializing\n");
+ EVQM_syslog(LOG_DEBUG, "client_event_thread() initializing\n");
event_base = ev_default_loop (EVFLAG_AUTO);
ev_async_init(&AddJob, QueueEventAddCallback);
ev_run (event_base, 0);
- syslog(LOG_DEBUG, "client_event_thread() exiting\n");
+ 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);
}
*/
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 };
if (h->IO->StartDB == 0.0)
h->IO->StartDB = Now;
h->IO->Now = Now;
+ ev_cleanup_start(event_db, &h->IO->db_abort_by_shutdown);
rc = h->EvAttch(h->IO);
switch (rc)
{
}
DeleteHashPos(&It);
DeleteHashContent(&q);
- syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
+ EVQM_syslog(LOG_DEBUG, "DBEVENT Q Add done.\n");
}
static void DBEventExitCallback(EV_P_ ev_async *w, int revents)
{
- syslog(LOG_DEBUG, "DB EVENT Q exiting.\n");
+ EVQM_syslog(LOG_DEBUG, "DB EVENT Q exiting.\n");
ev_break(event_db, EVBREAK_ALL);
}
void DBInitEventQueue(void)
{
- struct rlimit LimitSet;
-
pthread_mutex_init(&DBEventQueueMutex, NULL);
-
- if (pipe(evdb_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(evdb_add_pipe[1], &LimitSet);
+ pthread_mutex_init(&DBEventExitQueueMutex, NULL);
DBQueueEvents = NewHash(1, Flathash);
DBInboundEventQueues[0] = NewHash(1, Flathash);
*/
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);
- syslog(LOG_DEBUG, "dbevent_thread() initializing\n");
- event_db = ev_loop_new (EVFLAG_AUTO);
+ EVQM_syslog(LOG_DEBUG, "dbevent_thread() initializing\n");
+
+ tmp = event_db = ev_loop_new (EVFLAG_AUTO);
ev_async_init(&DBAddJob, DBQueueEventAddCallback);
ev_async_start(event_db, &DBAddJob);
ev_run (event_db, 0);
- 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);
}
void ShutDownEventQueues(void)
{
- syslog(LOG_DEBUG, "EVENT Qs triggering exits.\n");
+ EVQM_syslog(LOG_DEBUG, "EVENT Qs triggering exits.\n");
pthread_mutex_lock(&DBEventQueueMutex);
ev_async_send (event_db, &DBExitEventLoop);
projects / citadel.git / blobdiff