3 * Copyright (c) 1998-2012 by the citadel.org team
5 * This program is open source software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation; either version 3 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License
16 * along with this program; if not, write to the Free Software
17 * Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
29 #include <sys/types.h>
32 #if TIME_WITH_SYS_TIME
33 # include <sys/time.h>
37 # include <sys/time.h>
46 #include <sys/socket.h>
47 #include <netinet/in.h>
48 #include <arpa/inet.h>
54 #include <libcitadel.h>
57 #include "citserver.h"
64 #include "internet_addressing.h"
67 #include "clientsocket.h"
68 #include "locate_host.h"
69 #include "citadel_dirs.h"
71 #include "event_client.h"
72 #include "ctdl_module.h"
74 static void IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents);
76 static void IO_abort_shutdown_callback(struct ev_loop *loop,
80 AsyncIO *IO = watcher->data;
81 EV_syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__);
83 assert(IO->ShutdownAbort);
84 IO->ShutdownAbort(IO);
88 /*------------------------------------------------------------------------------
90 *----------------------------------------------------------------------------*/
91 extern int evdb_count;
92 extern pthread_mutex_t DBEventQueueMutex;
93 extern HashList *DBInboundEventQueue;
94 extern struct ev_loop *event_db;
95 extern ev_async DBAddJob;
96 extern ev_async DBExitEventLoop;
98 eNextState QueueDBOperation(AsyncIO *IO, IO_CallBack CB)
103 h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
106 ev_cleanup_init(&IO->db_abort_by_shutdown,
107 IO_abort_shutdown_callback);
108 IO->db_abort_by_shutdown.data = IO;
109 ev_cleanup_start(event_db, &IO->db_abort_by_shutdown);
111 pthread_mutex_lock(&DBEventQueueMutex);
112 EVM_syslog(LOG_DEBUG, "DBEVENT Q\n");
114 Put(DBInboundEventQueue, IKEY(i), h, NULL);
115 pthread_mutex_unlock(&DBEventQueueMutex);
117 ev_async_send (event_db, &DBAddJob);
118 EVM_syslog(LOG_DEBUG, "DBEVENT Q Done.\n");
122 void ShutDownDBCLient(AsyncIO *IO)
124 CitContext *Ctx =IO->CitContext;
127 EVM_syslog(LOG_DEBUG, "DBEVENT Terminating.\n");
128 ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown);
130 assert(IO->Terminate);
135 DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents)
137 AsyncIO *IO = watcher->data;
138 EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__);
139 become_session(IO->CitContext);
141 ev_idle_stop(event_db, &IO->db_unwind_stack);
143 assert(IO->NextDBOperation);
144 switch (IO->NextDBOperation(IO))
158 ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
160 case eTerminateConnection:
162 ev_idle_stop(event_db, &IO->db_unwind_stack);
163 ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
164 ShutDownDBCLient(IO);
168 eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB)
170 IO->NextDBOperation = CB;
171 ev_idle_init(&IO->db_unwind_stack,
173 IO->db_unwind_stack.data = IO;
174 ev_idle_start(event_db, &IO->db_unwind_stack);
178 /*------------------------------------------------------------------------------
180 *----------------------------------------------------------------------------*/
181 extern int evbase_count;
182 extern pthread_mutex_t EventQueueMutex;
183 extern HashList *InboundEventQueue;
184 extern struct ev_loop *event_base;
185 extern ev_async AddJob;
186 extern ev_async ExitEventLoop;
189 eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB)
194 h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
197 ev_cleanup_init(&IO->abort_by_shutdown,
198 IO_abort_shutdown_callback);
199 IO->abort_by_shutdown.data = IO;
200 ev_cleanup_start(event_base, &IO->abort_by_shutdown);
202 pthread_mutex_lock(&EventQueueMutex);
203 EVM_syslog(LOG_DEBUG, "EVENT Q\n");
205 Put(InboundEventQueue, IKEY(i), h, NULL);
206 pthread_mutex_unlock(&EventQueueMutex);
208 ev_async_send (event_base, &AddJob);
209 EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
213 extern eNextState evcurl_handle_start(AsyncIO *IO);
215 eNextState QueueCurlContext(AsyncIO *IO)
220 h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
222 h->EvAttch = evcurl_handle_start;
224 pthread_mutex_lock(&EventQueueMutex);
225 EVM_syslog(LOG_DEBUG, "EVENT Q\n");
227 Put(InboundEventQueue, IKEY(i), h, NULL);
228 pthread_mutex_unlock(&EventQueueMutex);
230 ev_async_send (event_base, &AddJob);
231 EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
235 void DestructCAres(AsyncIO *IO);
236 void FreeAsyncIOContents(AsyncIO *IO)
238 CitContext *Ctx = IO->CitContext;
240 FreeStrBuf(&IO->IOBuf);
241 FreeStrBuf(&IO->SendBuf.Buf);
242 FreeStrBuf(&IO->RecvBuf.Buf);
246 FreeURL(&IO->ConnectMe);
247 FreeStrBuf(&IO->HttpReq.ReplyData);
250 Ctx->state = CON_IDLE;
256 void StopClientWatchers(AsyncIO *IO)
258 ev_timer_stop (event_base, &IO->rw_timeout);
259 ev_timer_stop(event_base, &IO->conn_fail);
260 ev_idle_stop(event_base, &IO->unwind_stack);
262 ev_io_stop(event_base, &IO->conn_event);
263 ev_io_stop(event_base, &IO->send_event);
264 ev_io_stop(event_base, &IO->recv_event);
266 if (IO->SendBuf.fd != 0) {
267 close(IO->SendBuf.fd);
273 void ShutDownCLient(AsyncIO *IO)
275 CitContext *Ctx =IO->CitContext;
278 EVM_syslog(LOG_DEBUG, "EVENT Terminating \n");
280 ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
281 StopClientWatchers(IO);
283 if (IO->DNS.Channel != NULL) {
284 ares_destroy(IO->DNS.Channel);
285 EV_DNS_LOG_STOP(DNS.recv_event);
286 EV_DNS_LOG_STOP(DNS.send_event);
287 ev_io_stop(event_base, &IO->DNS.recv_event);
288 ev_io_stop(event_base, &IO->DNS.send_event);
289 IO->DNS.Channel = NULL;
291 assert(IO->Terminate);
295 eReadState HandleInbound(AsyncIO *IO)
297 const char *Err = NULL;
298 eReadState Finished = eBufferNotEmpty;
300 become_session(IO->CitContext);
302 while ((Finished == eBufferNotEmpty) &&
303 ((IO->NextState == eReadMessage)||
304 (IO->NextState == eReadMore)||
305 (IO->NextState == eReadFile)||
306 (IO->NextState == eReadPayload)))
309 * lex line reply in callback,
310 * or do it ourselves.
311 * i.e. as nnn-blabla means continue reading in SMTP
313 if ((IO->NextState == eReadFile) &&
314 (Finished == eBufferNotEmpty))
316 Finished = WriteIOBAlreadyRead(&IO->IOB, &Err);
317 if (Finished == eReadSuccess)
319 IO->NextState = eSendReply;
322 else if (IO->LineReader)
323 Finished = IO->LineReader(IO);
325 Finished = StrBufChunkSipLine(IO->IOBuf,
329 case eMustReadMore: /// read new from socket...
331 case eBufferNotEmpty: /* shouldn't happen... */
332 case eReadSuccess: /// done for now...
334 case eReadFail: /// WHUT?
339 if (Finished != eMustReadMore) {
340 assert(IO->ReadDone);
341 ev_io_stop(event_base, &IO->recv_event);
342 IO->NextState = IO->ReadDone(IO);
343 Finished = StrBufCheckBuffer(&IO->RecvBuf);
347 switch (IO->NextState) {
349 ev_io_start(event_base, &IO->send_event);
353 assert(IO->SendDone);
354 IO->NextState = IO->SendDone(IO);
355 ev_io_start(event_base, &IO->send_event);
360 ev_io_start(event_base, &IO->recv_event);
362 case eTerminateConnection:
380 IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
383 AsyncIO *IO = watcher->data;
384 const char *errmsg = NULL;
386 become_session(IO->CitContext);
392 const char *pch = ChrPtr(IO->SendBuf.Buf);
393 const char *pchh = IO->SendBuf.ReadWritePointer;
399 nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch);
400 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
401 ((CitContext*)(IO->CitContext))->ServiceName,
404 fd = fopen(fn, "a+");
405 fprintf(fd, "Send: BufSize: %ld BufContent: [",
407 rv = fwrite(pchh, nbytes, 1, fd);
408 if (!rv) printf("failed to write debug to %s!\n", fn);
411 switch (IO->NextState) {
413 rc = FileSendChunked(&IO->IOB, &errmsg);
415 StrBufPlain(IO->ErrMsg, errmsg, -1);
418 rc = StrBuf_write_one_chunk_callback(watcher->fd,
424 fprintf(fd, "Sent: BufSize: %d bytes.\n", rc);
430 ev_io_stop(event_base, &IO->send_event);
431 switch (IO->NextState) {
433 assert(IO->SendDone);
434 IO->NextState = IO->SendDone(IO);
436 if ((IO->NextState == eTerminateConnection) ||
437 (IO->NextState == eAbort) )
440 ev_io_start(event_base, &IO->send_event);
444 if (IO->IOB.ChunkSendRemain > 0) {
445 ev_io_start(event_base, &IO->recv_event);
446 SetNextTimeout(IO, 100.0);
449 assert(IO->ReadDone);
450 IO->NextState = IO->ReadDone(IO);
451 switch(IO->NextState) {
460 ev_io_start(event_base,
468 case eTerminateConnection:
475 if (StrBufCheckBuffer(&IO->SendBuf) != eReadSuccess)
477 IO->NextState = eReadMore;
482 if (StrBufCheckBuffer(&IO->RecvBuf) == eBufferNotEmpty)
487 ev_io_start(event_base, &IO->recv_event);
493 * we now live in another queue,
494 * so we have to unregister.
496 ev_cleanup_stop(loop, &IO->abort_by_shutdown);
501 case eTerminateConnection:
507 IO_Timeout_callback(loop, &IO->rw_timeout, revents);
509 /* else : must write more. */
512 set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents)
514 switch(IO->NextState) {
518 ev_io_start(event_base, &IO->recv_event);
524 become_session(IO->CitContext);
525 IO_send_callback(loop, &IO->send_event, revents);
531 case eTerminateConnection:
539 IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
541 AsyncIO *IO = watcher->data;
543 ev_timer_stop (event_base, &IO->rw_timeout);
544 become_session(IO->CitContext);
546 if (IO->SendBuf.fd != 0)
548 ev_io_stop(event_base, &IO->send_event);
549 ev_io_stop(event_base, &IO->recv_event);
550 ev_timer_stop (event_base, &IO->rw_timeout);
551 close(IO->SendBuf.fd);
552 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
556 switch (IO->Timeout(IO))
566 IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
568 AsyncIO *IO = watcher->data;
570 ev_timer_stop (event_base, &IO->conn_fail);
572 if (IO->SendBuf.fd != 0)
574 ev_io_stop(loop, &IO->conn_event);
575 ev_io_stop(event_base, &IO->send_event);
576 ev_io_stop(event_base, &IO->recv_event);
577 ev_timer_stop (event_base, &IO->rw_timeout);
578 close(IO->SendBuf.fd);
579 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
581 become_session(IO->CitContext);
583 assert(IO->ConnFail);
584 switch (IO->ConnFail(IO))
595 IO_connfailimmediate_callback(struct ev_loop *loop,
599 AsyncIO *IO = watcher->data;
601 ev_idle_stop (event_base, &IO->conn_fail_immediate);
603 if (IO->SendBuf.fd != 0)
605 close(IO->SendBuf.fd);
606 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
608 become_session(IO->CitContext);
610 assert(IO->ConnFail);
611 switch (IO->ConnFail(IO))
622 IO_connestd_callback(struct ev_loop *loop, ev_io *watcher, int revents)
624 AsyncIO *IO = watcher->data;
626 ev_io_stop(loop, &IO->conn_event);
627 ev_timer_stop (event_base, &IO->conn_fail);
628 set_start_callback(loop, IO, revents);
631 IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
635 AsyncIO *IO = watcher->data;
637 switch (IO->NextState) {
639 nbytes = FileRecvChunked(&IO->IOB, &errmsg);
641 StrBufPlain(IO->ErrMsg, errmsg, -1);
644 if (IO->IOB.ChunkSendRemain == 0)
646 IO->NextState = eSendReply;
653 nbytes = StrBuf_read_one_chunk_callback(watcher->fd,
665 const char *pch = ChrPtr(IO->RecvBuf.Buf);
666 const char *pchh = IO->RecvBuf.ReadWritePointer;
671 nbytes = StrLength(IO->RecvBuf.Buf) - (pchh - pch);
672 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
673 ((CitContext*)(IO->CitContext))->ServiceName,
676 fd = fopen(fn, "a+");
677 fprintf(fd, "Read: BufSize: %ld BufContent: [",
679 rv = fwrite(pchh, nbytes, 1, fd);
680 if (!rv) printf("failed to write debug to %s!\n", fn);
687 } else if (nbytes == 0) {
690 switch (IO->Timeout(IO))
698 } else if (nbytes == -1) {
699 // FD is gone. kick it.
700 StopClientWatchers(IO);
702 "EVENT: Socket Invalid! %s \n",
710 IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents)
712 AsyncIO *IO = watcher->data;
713 EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__);
714 become_session(IO->CitContext);
715 assert(IO->DNS.Query->PostDNS);
716 switch (IO->DNS.Query->PostDNS(IO))
719 assert(IO->DNS.Fail);
720 switch (IO->DNS.Fail(IO)) {
722 //// StopClientWatchers(IO);
733 eNextState EvConnectSock(AsyncIO *IO,
735 double first_rw_timeout,
738 struct sockaddr_in egress_sin;
742 become_session(IO->CitContext);
745 IO->NextState = eReadMessage;
748 IO->NextState = eSendReply;
751 IO->SendBuf.fd = IO->RecvBuf.fd =
753 (IO->ConnectMe->IPv6)?PF_INET6:PF_INET,
757 if (IO->SendBuf.fd < 0) {
759 "EVENT: socket() failed: %s\n",
762 StrBufPrintf(IO->ErrMsg,
763 "Failed to create socket: %s",
765 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
768 fdflags = fcntl(IO->SendBuf.fd, F_GETFL);
771 "EVENT: unable to get socket flags! %s \n",
773 StrBufPrintf(IO->ErrMsg,
774 "Failed to get socket flags: %s",
776 close(IO->SendBuf.fd);
777 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
780 fdflags = fdflags | O_NONBLOCK;
781 if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) {
784 "EVENT: unable to set socket nonblocking flags! %s \n",
786 StrBufPrintf(IO->ErrMsg,
787 "Failed to set socket flags: %s",
789 close(IO->SendBuf.fd);
790 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
793 /* TODO: maye we could use offsetof() to calc the position of data...
794 * http://doc.dvgu.ru/devel/ev.html#associating_custom_data_with_a_watcher
796 ev_io_init(&IO->recv_event, IO_recv_callback, IO->RecvBuf.fd, EV_READ);
797 IO->recv_event.data = IO;
798 ev_io_init(&IO->send_event, IO_send_callback, IO->SendBuf.fd, EV_WRITE);
799 IO->send_event.data = IO;
801 ev_timer_init(&IO->conn_fail, IO_connfail_callback, conn_timeout, 0);
802 IO->conn_fail.data = IO;
803 ev_timer_init(&IO->rw_timeout, IO_Timeout_callback, first_rw_timeout,0);
804 IO->rw_timeout.data = IO;
809 /* for debugging you may bypass it like this:
810 * IO->Addr.sin_addr.s_addr = inet_addr("127.0.0.1");
811 * ((struct sockaddr_in)IO->ConnectMe->Addr).sin_addr.s_addr =
812 * inet_addr("127.0.0.1");
814 if (IO->ConnectMe->IPv6) {
815 rc = connect(IO->SendBuf.fd,
816 &IO->ConnectMe->Addr,
817 sizeof(struct sockaddr_in6));
820 /* If citserver is bound to a specific IP address on the host, make
821 * sure we use that address for outbound connections.
824 memset(&egress_sin, 0, sizeof(egress_sin));
825 egress_sin.sin_family = AF_INET;
826 if (!IsEmptyStr(config.c_ip_addr)) {
827 egress_sin.sin_addr.s_addr = inet_addr(config.c_ip_addr);
828 if (egress_sin.sin_addr.s_addr == !INADDR_ANY) {
829 egress_sin.sin_addr.s_addr = INADDR_ANY;
832 /* If this bind fails, no problem; we can still use INADDR_ANY */
833 bind(IO->SendBuf.fd, (struct sockaddr *)&egress_sin, sizeof(egress_sin));
835 rc = connect(IO->SendBuf.fd,
836 (struct sockaddr_in *)&IO->ConnectMe->Addr,
837 sizeof(struct sockaddr_in));
841 EVM_syslog(LOG_DEBUG, "connect() immediate success.\n");
842 set_start_callback(event_base, IO, 0);
843 ev_timer_start(event_base, &IO->rw_timeout);
844 return IO->NextState;
846 else if (errno == EINPROGRESS) {
847 EVM_syslog(LOG_DEBUG, "connect() have to wait now.\n");
849 ev_io_init(&IO->conn_event,
850 IO_connestd_callback,
854 IO->conn_event.data = IO;
856 ev_io_start(event_base, &IO->conn_event);
857 ev_timer_start(event_base, &IO->conn_fail);
858 return IO->NextState;
861 ev_idle_init(&IO->conn_fail_immediate,
862 IO_connfailimmediate_callback);
863 IO->conn_fail_immediate.data = IO;
864 ev_idle_start(event_base, &IO->conn_fail_immediate);
866 EV_syslog(LOG_ERR, "connect() failed: %s\n", strerror(errno));
867 StrBufPrintf(IO->ErrMsg,
868 "Failed to connect: %s",
870 return IO->NextState;
872 return IO->NextState;
875 void SetNextTimeout(AsyncIO *IO, double timeout)
877 IO->rw_timeout.repeat = timeout;
878 ev_timer_again (event_base, &IO->rw_timeout);
882 eNextState ReAttachIO(AsyncIO *IO,
887 become_session(IO->CitContext);
888 ev_cleanup_start(event_base, &IO->abort_by_shutdown);
890 IO->NextState = eReadMessage;
893 IO->NextState = eSendReply;
895 set_start_callback(event_base, IO, 0);
897 return IO->NextState;
900 void InitIOStruct(AsyncIO *IO,
902 eNextState NextState,
903 IO_LineReaderCallback LineReader,
904 IO_CallBack DNS_Fail,
905 IO_CallBack SendDone,
906 IO_CallBack ReadDone,
907 IO_CallBack Terminate,
908 IO_CallBack ConnFail,
910 IO_CallBack ShutdownAbort)
914 IO->CitContext = CloneContext(CC);
915 ((CitContext *)IO->CitContext)->session_specific_data = (char*) Data;
917 IO->NextState = NextState;
919 IO->SendDone = SendDone;
920 IO->ReadDone = ReadDone;
921 IO->Terminate = Terminate;
922 IO->LineReader = LineReader;
923 IO->ConnFail = ConnFail;
924 IO->Timeout = Timeout;
925 IO->ShutdownAbort = ShutdownAbort;
927 IO->DNS.Fail = DNS_Fail;
929 IO->SendBuf.Buf = NewStrBufPlain(NULL, 1024);
930 IO->RecvBuf.Buf = NewStrBufPlain(NULL, 1024);
931 IO->IOBuf = NewStrBuf();
933 "EVENT: Session lives at %p IO at %p \n",
938 extern int evcurl_init(AsyncIO *IO);
940 int InitcURLIOStruct(AsyncIO *IO,
943 IO_CallBack SendDone,
944 IO_CallBack Terminate,
945 IO_CallBack ShutdownAbort)
949 IO->CitContext = CloneContext(CC);
950 ((CitContext *)IO->CitContext)->session_specific_data = (char*) Data;
952 IO->SendDone = SendDone;
953 IO->Terminate = Terminate;
954 IO->ShutdownAbort = ShutdownAbort;
956 strcpy(IO->HttpReq.errdesc, Desc);
959 return evcurl_init(IO);
963 void EV_backtrace(AsyncIO *IO)
965 #ifdef HAVE_BACKTRACE
966 void *stack_frames[50];
971 size = backtrace(stack_frames, sizeof(stack_frames) / sizeof(void*));
972 strings = backtrace_symbols(stack_frames, size);
973 for (i = 0; i < size; i++) {
975 EV_syslog(LOG_ALERT, " BT %s\n", strings[i]);
977 EV_syslog(LOG_ALERT, " BT %p\n", stack_frames[i]);
984 ev_tstamp ctdl_ev_now (void)
986 return ev_now(event_base);