2 * Copyright (c) 1998-2012 by the citadel.org team
4 * This program is open source software; you can redistribute it and/or modify
5 * it under the terms of the GNU General Public License, version 3.
7 * This program is distributed in the hope that it will be useful,
8 * but WITHOUT ANY WARRANTY; without even the implied warranty of
9 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
10 * GNU General Public License for more details.
18 #include <sys/socket.h>
19 #include <netinet/in.h>
20 #include <arpa/inet.h>
25 #include <libcitadel.h>
27 #include "ctdl_module.h"
28 #include "event_client.h"
29 #include "citserver.h"
32 ConstStr IOStates[] = {
39 {HKEY("DB Terminate")},
42 {HKEY("IO Connect Socket")},
45 {HKEY("IO ConnFail")},
46 {HKEY("IO ConnFail Now")},
47 {HKEY("IO Conn Now")},
48 {HKEY("IO Conn Wait")},
51 {HKEY("Curl Shotdown")},
52 {HKEY("Curl More IO")},
53 {HKEY("Curl Got IO")},
54 {HKEY("Curl Got Data")},
55 {HKEY("Curl Got Status")},
56 {HKEY("C-Ares Start")},
57 {HKEY("C-Ares IO Done")},
58 {HKEY("C-Ares Finished")},
59 {HKEY("C-Ares exit")},
64 void SetEVState(AsyncIO *IO, eIOState State)
67 CitContext* CCC = IO->CitContext;
69 memcpy(CCC->lastcmdname, IOStates[State].Key, IOStates[State].len + 1);
73 eNextState QueueAnEventContext(AsyncIO *IO);
74 static void IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents);
75 static void IO_abort_shutdown_callback(struct ev_loop *loop,
80 /*------------------------------------------------------------------------------
82 *----------------------------------------------------------------------------*/
83 extern int evdb_count;
84 extern pthread_mutex_t DBEventQueueMutex;
85 extern pthread_mutex_t DBEventExitQueueMutex;
86 extern HashList *DBInboundEventQueue;
87 extern struct ev_loop *event_db;
88 extern ev_async DBAddJob;
89 extern ev_async DBExitEventLoop;
91 eNextState QueueAnDBOperation(AsyncIO *IO)
97 h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
100 assert(IO->ReAttachCB != NULL);
102 h->EvAttch = IO->ReAttachCB;
103 ev_cleanup_init(&IO->db_abort_by_shutdown,
104 IO_abort_shutdown_callback);
105 IO->db_abort_by_shutdown.data = IO;
107 pthread_mutex_lock(&DBEventQueueMutex);
108 if (DBInboundEventQueue == NULL)
110 /* shutting down... */
112 EVM_syslog(LOG_DEBUG, "DBEVENT Q exiting.\n");
113 pthread_mutex_unlock(&DBEventQueueMutex);
116 EVM_syslog(LOG_DEBUG, "DBEVENT Q\n");
118 Put(DBInboundEventQueue, IKEY(i), h, NULL);
119 pthread_mutex_unlock(&DBEventQueueMutex);
121 pthread_mutex_lock(&DBEventExitQueueMutex);
122 if (event_db == NULL)
124 pthread_mutex_unlock(&DBEventExitQueueMutex);
127 ev_async_send (event_db, &DBAddJob);
128 pthread_mutex_unlock(&DBEventExitQueueMutex);
130 EVQM_syslog(LOG_DEBUG, "DBEVENT Q Done.\n");
134 void StopDBWatchers(AsyncIO *IO)
136 SetEVState(IO, eDBStop);
137 ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown);
138 ev_idle_stop(event_db, &IO->db_unwind_stack);
141 void ShutDownDBCLient(AsyncIO *IO)
143 CitContext *Ctx =IO->CitContext;
146 SetEVState(IO, eDBTerm);
147 EVM_syslog(LOG_DEBUG, "DBEVENT Terminating.\n");
150 assert(IO->DBTerminate);
155 DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents)
157 AsyncIO *IO = watcher->data;
159 SetEVState(IO, eDBNext);
160 SET_EV_TIME(IO, event_db);
161 EV_syslog(LOG_DEBUG, "%s()", __FUNCTION__);
162 become_session(IO->CitContext);
164 ev_idle_stop(event_db, &IO->db_unwind_stack);
166 assert(IO->NextDBOperation);
167 switch (IO->NextDBOperation(IO))
170 ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
171 QueueAnEventContext(IO);
184 ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
186 case eTerminateConnection:
188 ev_idle_stop(event_db, &IO->db_unwind_stack);
189 ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
190 ShutDownDBCLient(IO);
194 eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB)
196 SetEVState(IO, eQDBNext);
197 IO->NextDBOperation = CB;
198 ev_idle_init(&IO->db_unwind_stack,
200 IO->db_unwind_stack.data = IO;
201 ev_idle_start(event_db, &IO->db_unwind_stack);
205 /*------------------------------------------------------------------------------
207 *----------------------------------------------------------------------------*/
208 extern int evbase_count;
209 extern pthread_mutex_t EventQueueMutex;
210 extern pthread_mutex_t EventExitQueueMutex;
211 extern HashList *InboundEventQueue;
212 extern struct ev_loop *event_base;
213 extern ev_async AddJob;
214 extern ev_async ExitEventLoop;
216 static void IO_abort_shutdown_callback(struct ev_loop *loop,
220 AsyncIO *IO = watcher->data;
222 SetEVState(IO, eIOAbort);
223 EV_syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__);
224 SET_EV_TIME(IO, event_base);
225 assert(IO->ShutdownAbort);
226 IO->ShutdownAbort(IO);
230 eNextState QueueAnEventContext(AsyncIO *IO)
235 SetEVState(IO, eIOQ);
236 h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
239 assert(IO->ReAttachCB != NULL);
241 h->EvAttch = IO->ReAttachCB;
243 ev_cleanup_init(&IO->abort_by_shutdown,
244 IO_abort_shutdown_callback);
245 IO->abort_by_shutdown.data = IO;
247 pthread_mutex_lock(&EventQueueMutex);
248 if (InboundEventQueue == NULL)
251 /* shutting down... */
252 EVM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
253 pthread_mutex_unlock(&EventQueueMutex);
256 EVM_syslog(LOG_DEBUG, "EVENT Q\n");
258 Put(InboundEventQueue, IKEY(i), h, NULL);
259 pthread_mutex_unlock(&EventQueueMutex);
261 pthread_mutex_lock(&EventExitQueueMutex);
262 if (event_base == NULL) {
263 pthread_mutex_unlock(&EventExitQueueMutex);
266 ev_async_send (event_base, &AddJob);
267 pthread_mutex_unlock(&EventExitQueueMutex);
268 EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
272 eNextState EventQueueDBOperation(AsyncIO *IO, IO_CallBack CB, int CloseFDs)
274 StopClientWatchers(IO, CloseFDs);
278 eNextState DBQueueEventContext(AsyncIO *IO, IO_CallBack CB)
285 eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB)
288 return QueueAnEventContext(IO);
291 extern eNextState evcurl_handle_start(AsyncIO *IO);
293 eNextState QueueCurlContext(AsyncIO *IO)
298 SetEVState(IO, eCurlQ);
299 h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
301 h->EvAttch = evcurl_handle_start;
303 pthread_mutex_lock(&EventQueueMutex);
304 if (InboundEventQueue == NULL)
306 /* shutting down... */
308 EVM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
309 pthread_mutex_unlock(&EventQueueMutex);
313 EVM_syslog(LOG_DEBUG, "EVENT Q\n");
315 Put(InboundEventQueue, IKEY(i), h, NULL);
316 pthread_mutex_unlock(&EventQueueMutex);
318 pthread_mutex_lock(&EventExitQueueMutex);
319 if (event_base == NULL) {
320 pthread_mutex_unlock(&EventExitQueueMutex);
323 ev_async_send (event_base, &AddJob);
324 pthread_mutex_unlock(&EventExitQueueMutex);
326 EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
330 eNextState CurlQueueDBOperation(AsyncIO *IO, IO_CallBack CB)
332 StopCurlWatchers(IO);
338 void FreeAsyncIOContents(AsyncIO *IO)
340 CitContext *Ctx = IO->CitContext;
342 FreeStrBuf(&IO->IOBuf);
343 FreeStrBuf(&IO->SendBuf.Buf);
344 FreeStrBuf(&IO->RecvBuf.Buf);
346 FreeURL(&IO->ConnectMe);
347 FreeStrBuf(&IO->HttpReq.ReplyData);
350 Ctx->state = CON_IDLE;
352 IO->CitContext = NULL;
357 void DestructCAres(AsyncIO *IO);
358 void StopClientWatchers(AsyncIO *IO, int CloseFD)
360 EVM_syslog(LOG_DEBUG, "EVENT StopClientWatchers");
364 ev_timer_stop (event_base, &IO->rw_timeout);
365 ev_timer_stop(event_base, &IO->conn_fail);
366 ev_idle_stop(event_base, &IO->unwind_stack);
367 ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
369 ev_io_stop(event_base, &IO->conn_event);
370 ev_io_stop(event_base, &IO->send_event);
371 ev_io_stop(event_base, &IO->recv_event);
373 if (CloseFD && (IO->SendBuf.fd > 0)) {
374 close(IO->SendBuf.fd);
380 void StopCurlWatchers(AsyncIO *IO)
382 EVM_syslog(LOG_DEBUG, "EVENT StopCurlWatchers \n");
384 ev_timer_stop (event_base, &IO->rw_timeout);
385 ev_timer_stop(event_base, &IO->conn_fail);
386 ev_idle_stop(event_base, &IO->unwind_stack);
387 ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
389 ev_io_stop(event_base, &IO->conn_event);
390 ev_io_stop(event_base, &IO->send_event);
391 ev_io_stop(event_base, &IO->recv_event);
393 curl_easy_cleanup(IO->HttpReq.chnd);
394 IO->HttpReq.chnd = NULL;
396 if (IO->SendBuf.fd != 0) {
397 close(IO->SendBuf.fd);
403 eNextState ShutDownCLient(AsyncIO *IO)
405 CitContext *Ctx =IO->CitContext;
407 SetEVState(IO, eExit);
410 EVM_syslog(LOG_DEBUG, "EVENT Terminating \n");
412 StopClientWatchers(IO, 1);
414 if (IO->DNS.Channel != NULL) {
415 ares_destroy(IO->DNS.Channel);
416 EV_DNS_LOG_STOP(DNS.recv_event);
417 EV_DNS_LOG_STOP(DNS.send_event);
418 ev_io_stop(event_base, &IO->DNS.recv_event);
419 ev_io_stop(event_base, &IO->DNS.send_event);
420 IO->DNS.Channel = NULL;
422 assert(IO->Terminate);
423 return IO->Terminate(IO);
426 void PostInbound(AsyncIO *IO)
429 switch (IO->NextState) {
431 ev_io_start(event_base, &IO->send_event);
435 assert(IO->SendDone);
436 IO->NextState = IO->SendDone(IO);
437 switch (IO->NextState)
446 ev_io_start(event_base, &IO->send_event);
449 StopClientWatchers(IO, 0);
450 QueueAnDBOperation(IO);
458 ev_io_start(event_base, &IO->recv_event);
460 case eTerminateConnection:
462 if (ShutDownCLient(IO) == eDBQuery) {
463 QueueAnDBOperation(IO);
472 QueueAnDBOperation(IO);
475 eReadState HandleInbound(AsyncIO *IO)
477 const char *Err = NULL;
478 eReadState Finished = eBufferNotEmpty;
480 become_session(IO->CitContext);
482 while ((Finished == eBufferNotEmpty) &&
483 ((IO->NextState == eReadMessage)||
484 (IO->NextState == eReadMore)||
485 (IO->NextState == eReadFile)||
486 (IO->NextState == eReadPayload)))
489 * lex line reply in callback,
490 * or do it ourselves.
491 * i.e. as nnn-blabla means continue reading in SMTP
493 if ((IO->NextState == eReadFile) &&
494 (Finished == eBufferNotEmpty))
496 Finished = WriteIOBAlreadyRead(&IO->IOB, &Err);
497 if (Finished == eReadSuccess)
499 IO->NextState = eSendReply;
502 else if (IO->LineReader)
503 Finished = IO->LineReader(IO);
505 Finished = StrBufChunkSipLine(IO->IOBuf,
509 case eMustReadMore: /// read new from socket...
511 case eBufferNotEmpty: /* shouldn't happen... */
512 case eReadSuccess: /// done for now...
514 case eReadFail: /// WHUT?
519 if (Finished != eMustReadMore) {
520 ev_io_stop(event_base, &IO->recv_event);
521 IO->NextState = IO->ReadDone(IO);
522 if (IO->NextState == eDBQuery) {
523 if (QueueAnDBOperation(IO) == eAbort)
529 Finished = StrBufCheckBuffer(&IO->RecvBuf);
541 IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
544 AsyncIO *IO = watcher->data;
545 const char *errmsg = NULL;
547 SET_EV_TIME(IO, event_base);
548 become_session(IO->CitContext);
554 const char *pch = ChrPtr(IO->SendBuf.Buf);
555 const char *pchh = IO->SendBuf.ReadWritePointer;
561 nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch);
562 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
563 ((CitContext*)(IO->CitContext))->ServiceName,
566 fd = fopen(fn, "a+");
568 syslog(LOG_EMERG, "failed to open file %s: %s", fn, strerror(errno));
572 fprintf(fd, "Send: BufSize: %ld BufContent: [",
574 rv = fwrite(pchh, nbytes, 1, fd);
575 if (!rv) printf("failed to write debug to %s!\n", fn);
578 switch (IO->NextState) {
580 rc = FileSendChunked(&IO->IOB, &errmsg);
582 StrBufPlain(IO->ErrMsg, errmsg, -1);
585 rc = StrBuf_write_one_chunk_callback(IO->SendBuf.fd,
591 fprintf(fd, "Sent: BufSize: %d bytes.\n", rc);
597 ev_io_stop(event_base, &IO->send_event);
598 switch (IO->NextState) {
600 assert(IO->SendDone);
601 IO->NextState = IO->SendDone(IO);
603 if ((IO->NextState == eTerminateConnection) ||
604 (IO->NextState == eAbort) )
607 ev_io_start(event_base, &IO->send_event);
611 if (IO->IOB.ChunkSendRemain > 0) {
612 ev_io_start(event_base, &IO->recv_event);
613 SetNextTimeout(IO, 100.0);
616 assert(IO->ReadDone);
617 IO->NextState = IO->ReadDone(IO);
618 switch(IO->NextState) {
627 ev_io_start(event_base,
635 case eTerminateConnection:
642 if (StrBufCheckBuffer(&IO->SendBuf) != eReadSuccess)
644 IO->NextState = eReadMore;
649 if (StrBufCheckBuffer(&IO->RecvBuf) == eBufferNotEmpty)
654 ev_io_start(event_base, &IO->recv_event);
660 * we now live in another queue,
661 * so we have to unregister.
663 ev_cleanup_stop(loop, &IO->abort_by_shutdown);
668 case eTerminateConnection:
674 if (errno != EAGAIN) {
675 StopClientWatchers(IO, 1);
677 "IO_send_callback(): Socket Invalid! [%d] [%s] [%d]\n",
678 errno, strerror(errno), IO->SendBuf.fd);
679 StrBufPrintf(IO->ErrMsg,
680 "Socket Invalid! [%s]",
682 SetNextTimeout(IO, 0.01);
685 /* else : must write more. */
688 set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents)
690 ev_timer_stop(event_base, &IO->conn_fail);
691 ev_timer_start(event_base, &IO->rw_timeout);
693 switch(IO->NextState) {
697 StrBufAppendBufPlain(IO->ErrMsg, HKEY("[while waiting for greeting]"), 0);
698 ev_io_start(event_base, &IO->recv_event);
704 become_session(IO->CitContext);
705 IO_send_callback(loop, &IO->send_event, revents);
711 case eTerminateConnection:
719 IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
721 AsyncIO *IO = watcher->data;
723 SetEVState(IO, eIOTimeout);
724 SET_EV_TIME(IO, event_base);
725 ev_timer_stop (event_base, &IO->rw_timeout);
726 become_session(IO->CitContext);
728 if (IO->SendBuf.fd != 0)
730 ev_io_stop(event_base, &IO->send_event);
731 ev_io_stop(event_base, &IO->recv_event);
732 ev_timer_stop (event_base, &IO->rw_timeout);
733 close(IO->SendBuf.fd);
734 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
738 switch (IO->Timeout(IO))
748 IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
750 AsyncIO *IO = watcher->data;
752 SetEVState(IO, eIOConnfail);
753 SET_EV_TIME(IO, event_base);
754 ev_timer_stop (event_base, &IO->conn_fail);
756 if (IO->SendBuf.fd != 0)
758 ev_io_stop(loop, &IO->conn_event);
759 ev_io_stop(event_base, &IO->send_event);
760 ev_io_stop(event_base, &IO->recv_event);
761 ev_timer_stop (event_base, &IO->rw_timeout);
762 close(IO->SendBuf.fd);
763 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
765 become_session(IO->CitContext);
767 assert(IO->ConnFail);
768 switch (IO->ConnFail(IO))
779 IO_connfailimmediate_callback(struct ev_loop *loop,
783 AsyncIO *IO = watcher->data;
785 SetEVState(IO, eIOConnfailNow);
786 SET_EV_TIME(IO, event_base);
787 ev_idle_stop (event_base, &IO->conn_fail_immediate);
789 if (IO->SendBuf.fd != 0)
791 close(IO->SendBuf.fd);
792 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
794 become_session(IO->CitContext);
796 assert(IO->ConnFail);
797 switch (IO->ConnFail(IO))
808 IO_connestd_callback(struct ev_loop *loop, ev_io *watcher, int revents)
810 AsyncIO *IO = watcher->data;
812 socklen_t lon = sizeof(so_err);
815 SetEVState(IO, eIOConnNow);
816 SET_EV_TIME(IO, event_base);
817 EVM_syslog(LOG_DEBUG, "connect() succeeded.\n");
819 ev_io_stop(loop, &IO->conn_event);
820 ev_timer_stop(event_base, &IO->conn_fail);
822 err = getsockopt(IO->SendBuf.fd,
828 if ((err == 0) && (so_err != 0))
830 EV_syslog(LOG_DEBUG, "connect() failed [%d][%s]\n",
833 IO_connfail_callback(loop, &IO->conn_fail, revents);
838 EVM_syslog(LOG_DEBUG, "connect() succeeded\n");
839 set_start_callback(loop, IO, revents);
844 IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
848 AsyncIO *IO = watcher->data;
850 SET_EV_TIME(IO, event_base);
851 switch (IO->NextState) {
853 nbytes = FileRecvChunked(&IO->IOB, &errmsg);
855 StrBufPlain(IO->ErrMsg, errmsg, -1);
858 if (IO->IOB.ChunkSendRemain == 0)
860 IO->NextState = eSendReply;
861 assert(IO->ReadDone);
862 ev_io_stop(event_base, &IO->recv_event);
871 nbytes = StrBuf_read_one_chunk_callback(IO->RecvBuf.fd,
883 const char *pch = ChrPtr(IO->RecvBuf.Buf);
884 const char *pchh = IO->RecvBuf.ReadWritePointer;
889 nbytes = StrLength(IO->RecvBuf.Buf) - (pchh - pch);
890 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
891 ((CitContext*)(IO->CitContext))->ServiceName,
894 fd = fopen(fn, "a+");
896 syslog(LOG_EMERG, "failed to open file %s: %s", fn, strerror(errno));
900 fprintf(fd, "Read: BufSize: %ld BufContent: [",
902 rv = fwrite(pchh, nbytes, 1, fd);
903 if (!rv) printf("failed to write debug to %s!\n", fn);
910 } else if (nbytes == 0) {
911 StopClientWatchers(IO, 1);
912 SetNextTimeout(IO, 0.01);
914 } else if (nbytes == -1) {
915 if (errno != EAGAIN) {
916 // FD is gone. kick it.
917 StopClientWatchers(IO, 1);
919 "IO_recv_callback(): Socket Invalid! [%d] [%s] [%d]\n",
920 errno, strerror(errno), IO->SendBuf.fd);
921 StrBufPrintf(IO->ErrMsg,
922 "Socket Invalid! [%s]",
924 SetNextTimeout(IO, 0.01);
931 IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents)
933 AsyncIO *IO = watcher->data;
935 SetEVState(IO, eCaresFinished);
936 SET_EV_TIME(IO, event_base);
937 EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__);
938 become_session(IO->CitContext);
939 assert(IO->DNS.Query->PostDNS);
940 switch (IO->DNS.Query->PostDNS(IO))
943 assert(IO->DNS.Fail);
944 switch (IO->DNS.Fail(IO)) {
946 //// StopClientWatchers(IO);
950 StopClientWatchers(IO, 0);
951 QueueAnDBOperation(IO);
957 StopClientWatchers(IO, 0);
958 QueueAnDBOperation(IO);
966 eNextState EvConnectSock(AsyncIO *IO,
968 double first_rw_timeout,
971 struct sockaddr_in egress_sin;
975 SetEVState(IO, eIOConnectSock);
976 become_session(IO->CitContext);
979 IO->NextState = eReadMessage;
982 IO->NextState = eSendReply;
985 IO->SendBuf.fd = IO->RecvBuf.fd =
987 (IO->ConnectMe->IPv6)?PF_INET6:PF_INET,
991 if (IO->SendBuf.fd < 0) {
993 "EVENT: socket() failed: %s\n",
996 StrBufPrintf(IO->ErrMsg,
997 "Failed to create socket: %s",
999 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
1002 fdflags = fcntl(IO->SendBuf.fd, F_GETFL);
1005 "EVENT: unable to get socket %d flags! %s \n",
1008 StrBufPrintf(IO->ErrMsg,
1009 "Failed to get socket %d flags: %s",
1012 close(IO->SendBuf.fd);
1013 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
1016 fdflags = fdflags | O_NONBLOCK;
1017 if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) {
1020 "EVENT: unable to set socket %d nonblocking flags! %s \n",
1023 StrBufPrintf(IO->ErrMsg,
1024 "Failed to set socket flags: %s",
1026 close(IO->SendBuf.fd);
1027 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
1030 /* TODO: maye we could use offsetof() to calc the position of data...
1031 * http://doc.dvgu.ru/devel/ev.html#associating_custom_data_with_a_watcher
1033 ev_io_init(&IO->recv_event, IO_recv_callback, IO->RecvBuf.fd, EV_READ);
1034 IO->recv_event.data = IO;
1035 ev_io_init(&IO->send_event, IO_send_callback, IO->SendBuf.fd, EV_WRITE);
1036 IO->send_event.data = IO;
1038 ev_timer_init(&IO->conn_fail, IO_connfail_callback, conn_timeout, 0);
1039 IO->conn_fail.data = IO;
1040 ev_timer_init(&IO->rw_timeout, IO_Timeout_callback, first_rw_timeout,0);
1041 IO->rw_timeout.data = IO;
1046 /* for debugging you may bypass it like this:
1047 * IO->Addr.sin_addr.s_addr = inet_addr("127.0.0.1");
1048 * ((struct sockaddr_in)IO->ConnectMe->Addr).sin_addr.s_addr =
1049 * inet_addr("127.0.0.1");
1051 if (IO->ConnectMe->IPv6) {
1052 rc = connect(IO->SendBuf.fd,
1053 &IO->ConnectMe->Addr,
1054 sizeof(struct sockaddr_in6));
1057 /* If citserver is bound to a specific IP address on the host, make
1058 * sure we use that address for outbound connections.
1061 memset(&egress_sin, 0, sizeof(egress_sin));
1062 egress_sin.sin_family = AF_INET;
1063 if (!IsEmptyStr(CtdlGetConfigStr("c_ip_addr"))) {
1064 egress_sin.sin_addr.s_addr = inet_addr(CtdlGetConfigStr("c_ip_addr"));
1065 if (egress_sin.sin_addr.s_addr == !INADDR_ANY) {
1066 egress_sin.sin_addr.s_addr = INADDR_ANY;
1069 /* If this bind fails, no problem; we can still use INADDR_ANY */
1070 bind(IO->SendBuf.fd, (struct sockaddr *)&egress_sin, sizeof(egress_sin));
1072 rc = connect(IO->SendBuf.fd,
1073 (struct sockaddr_in *)&IO->ConnectMe->Addr,
1074 sizeof(struct sockaddr_in));
1078 SetEVState(IO, eIOConnNow);
1079 EV_syslog(LOG_DEBUG, "connect() = %d immediate success.\n", IO->SendBuf.fd);
1080 set_start_callback(event_base, IO, 0);
1081 return IO->NextState;
1083 else if (errno == EINPROGRESS) {
1084 SetEVState(IO, eIOConnWait);
1085 EV_syslog(LOG_DEBUG, "connect() = %d have to wait now.\n", IO->SendBuf.fd);
1087 ev_io_init(&IO->conn_event,
1088 IO_connestd_callback,
1092 IO->conn_event.data = IO;
1094 ev_io_start(event_base, &IO->conn_event);
1095 ev_timer_start(event_base, &IO->conn_fail);
1096 return IO->NextState;
1099 SetEVState(IO, eIOConnfail);
1100 ev_idle_init(&IO->conn_fail_immediate,
1101 IO_connfailimmediate_callback);
1102 IO->conn_fail_immediate.data = IO;
1103 ev_idle_start(event_base, &IO->conn_fail_immediate);
1106 "connect() = %d failed: %s\n",
1110 StrBufPrintf(IO->ErrMsg,
1111 "Failed to connect: %s",
1113 return IO->NextState;
1115 return IO->NextState;
1118 void SetNextTimeout(AsyncIO *IO, double timeout)
1120 IO->rw_timeout.repeat = timeout;
1121 ev_timer_again (event_base, &IO->rw_timeout);
1125 eNextState ReAttachIO(AsyncIO *IO,
1129 SetEVState(IO, eIOAttach);
1131 become_session(IO->CitContext);
1132 ev_cleanup_start(event_base, &IO->abort_by_shutdown);
1134 IO->NextState = eReadMessage;
1137 IO->NextState = eSendReply;
1139 set_start_callback(event_base, IO, 0);
1141 return IO->NextState;
1144 void InitIOStruct(AsyncIO *IO,
1146 eNextState NextState,
1147 IO_LineReaderCallback LineReader,
1148 IO_CallBack DNS_Fail,
1149 IO_CallBack SendDone,
1150 IO_CallBack ReadDone,
1151 IO_CallBack Terminate,
1152 IO_CallBack DBTerminate,
1153 IO_CallBack ConnFail,
1154 IO_CallBack Timeout,
1155 IO_CallBack ShutdownAbort)
1159 IO->CitContext = CloneContext(CC);
1160 IO->CitContext->session_specific_data = Data;
1161 IO->CitContext->IO = IO;
1163 IO->NextState = NextState;
1165 IO->SendDone = SendDone;
1166 IO->ReadDone = ReadDone;
1167 IO->Terminate = Terminate;
1168 IO->DBTerminate = DBTerminate;
1169 IO->LineReader = LineReader;
1170 IO->ConnFail = ConnFail;
1171 IO->Timeout = Timeout;
1172 IO->ShutdownAbort = ShutdownAbort;
1174 IO->DNS.Fail = DNS_Fail;
1176 IO->SendBuf.Buf = NewStrBufPlain(NULL, 1024);
1177 IO->RecvBuf.Buf = NewStrBufPlain(NULL, 1024);
1178 IO->IOBuf = NewStrBuf();
1179 EV_syslog(LOG_DEBUG,
1180 "EVENT: Session lives at %p IO at %p \n",
1185 extern int evcurl_init(AsyncIO *IO);
1187 int InitcURLIOStruct(AsyncIO *IO,
1190 IO_CallBack SendDone,
1191 IO_CallBack Terminate,
1192 IO_CallBack DBTerminate,
1193 IO_CallBack ShutdownAbort)
1197 IO->CitContext = CloneContext(CC);
1198 IO->CitContext->session_specific_data = Data;
1199 IO->CitContext->IO = IO;
1201 IO->SendDone = SendDone;
1202 IO->Terminate = Terminate;
1203 IO->DBTerminate = DBTerminate;
1204 IO->ShutdownAbort = ShutdownAbort;
1206 strcpy(IO->HttpReq.errdesc, Desc);
1209 return evcurl_init(IO);
1214 typedef struct KillOtherSessionContext {
1217 }KillOtherSessionContext;
1219 eNextState KillTerminate(AsyncIO *IO)
1222 KillOtherSessionContext *Ctx = (KillOtherSessionContext*)IO->Data;
1223 EV_syslog(LOG_DEBUG, "%s Exit\n", __FUNCTION__);
1225 FreeAsyncIOContents(IO);
1226 memset(Ctx, 0, sizeof(KillOtherSessionContext));
1227 IO->ID = id; /* just for the case we want to analyze it in a coredump */
1233 eNextState KillShutdown(AsyncIO *IO)
1235 return eTerminateConnection;
1238 eNextState KillOtherContextNow(AsyncIO *IO)
1240 KillOtherSessionContext *Ctx = IO->Data;
1242 SetEVState(IO, eKill);
1244 if (Ctx->OtherOne->ShutdownAbort != NULL) {
1245 Ctx->OtherOne->NextState = eAbort;
1246 if (Ctx->OtherOne->ShutdownAbort(Ctx->OtherOne) == eDBQuery) {
1247 StopClientWatchers(Ctx->OtherOne, 0);
1248 QueueAnDBOperation(Ctx->OtherOne);
1251 return eTerminateConnection;
1254 void KillAsyncIOContext(AsyncIO *IO)
1256 KillOtherSessionContext *Ctx;
1258 Ctx = (KillOtherSessionContext*) malloc(sizeof(KillOtherSessionContext));
1259 memset(Ctx, 0, sizeof(KillOtherSessionContext));
1261 InitIOStruct(&Ctx->IO,
1276 switch(IO->NextState) {
1289 Ctx->IO.ReAttachCB = KillOtherContextNow;
1290 QueueAnEventContext(&Ctx->IO);
1293 Ctx->IO.ReAttachCB = KillOtherContextNow;
1294 QueueAnDBOperation(&Ctx->IO);
1296 case eTerminateConnection:
1298 /*hm, its already dying, dunno which Queue its in... */
1304 extern int DebugEventLoopBacktrace;
1305 void EV_backtrace(AsyncIO *IO)
1307 #ifdef HAVE_BACKTRACE
1308 void *stack_frames[50];
1312 if ((IO == NULL) || (DebugEventLoopBacktrace == 0))
1314 size = backtrace(stack_frames, sizeof(stack_frames) / sizeof(void*));
1315 strings = backtrace_symbols(stack_frames, size);
1316 for (i = 0; i < size; i++) {
1317 if (strings != NULL) {
1318 EV_syslog(LOG_ALERT, " BT %s\n", strings[i]);
1321 EV_syslog(LOG_ALERT, " BT %p\n", stack_frames[i]);
1329 ev_tstamp ctdl_ev_now (void)
1331 return ev_now(event_base);