citadel/event_client.c

   1 /*
   2  * Copyright (c) 1998-2012 by the citadel.org team
   3  *
   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.
   6  *
   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.
  11  */
  12 #include "sysdep.h"
  13
  14 #include <stdio.h>
  15 #include <string.h>
  16 #include <syslog.h>
  17 #include <assert.h>
  18 #include <sys/socket.h>
  19 #include <netinet/in.h>
  20 #include <arpa/inet.h>
  21 #if HAVE_BACKTRACE
  22 #include <execinfo.h>
  23 #endif
  24
  25 #include <libcitadel.h>
  26
  27 #include "ctdl_module.h"
  28 #include "event_client.h"
  29
  30 ConstStr IOStates[] = {
  31         {HKEY("DB Queue")},
  32         {HKEY("DB Q Next")},
  33         {HKEY("DB Attach")},
  34         {HKEY("DB Next")},
  35         {HKEY("DB Stop")},
  36         {HKEY("DB Exit")},
  37         {HKEY("DB Terminate")},
  38         {HKEY("IO Queue")},
  39         {HKEY("IO Attach")},
  40         {HKEY("IO Connect Socket")},
  41         {HKEY("IO Abort")},
  42         {HKEY("IO Timeout")},
  43         {HKEY("IO ConnFail")},
  44         {HKEY("IO ConnFail Now")},
  45         {HKEY("IO Conn Now")},
  46         {HKEY("IO Conn Wait")},
  47         {HKEY("Curl Q")},
  48         {HKEY("Curl Start")},
  49         {HKEY("Curl Shotdown")},
  50         {HKEY("Curl More IO")},
  51         {HKEY("Curl Got IO")},
  52         {HKEY("Curl Got Data")},
  53         {HKEY("Curl Got Status")},
  54         {HKEY("C-Ares Start")},
  55         {HKEY("C-Ares IO Done")},
  56         {HKEY("C-Ares Finished")},
  57         {HKEY("C-Ares exit")},
  58         {HKEY("Killing")},
  59         {HKEY("Exit")}
  60 };
  61
  62 void SetEVState(AsyncIO *IO, eIOState State)
  63 {
  64
  65         CitContext* CCC = IO->CitContext;
  66         if (CCC != NULL)
  67                 memcpy(CCC->lastcmdname, IOStates[State].Key, IOStates[State].len + 1);
  68
  69 }
  70
  71
  72 static void IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents);
  73 static void IO_abort_shutdown_callback(struct ev_loop *loop,
  74                                        ev_cleanup *watcher,
  75                                        int revents);
  76
  77
  78 /*------------------------------------------------------------------------------
  79  *                              Server DB IO
  80  *----------------------------------------------------------------------------*/
  81 extern int evdb_count;
  82 extern pthread_mutex_t DBEventQueueMutex;
  83 extern pthread_mutex_t DBEventExitQueueMutex;
  84 extern HashList *DBInboundEventQueue;
  85 extern struct ev_loop *event_db;
  86 extern ev_async DBAddJob;
  87 extern ev_async DBExitEventLoop;
  88
  89 eNextState QueueDBOperation(AsyncIO *IO, IO_CallBack CB)
  90 {
  91         IOAddHandler *h;
  92         int i;
  93
  94         SetEVState(IO, eDBQ);
  95         h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
  96         h->IO = IO;
  97         h->EvAttch = CB;
  98         ev_cleanup_init(&IO->db_abort_by_shutdown,
  99                         IO_abort_shutdown_callback);
 100         IO->db_abort_by_shutdown.data = IO;
 101
 102         pthread_mutex_lock(&DBEventQueueMutex);
 103         if (DBInboundEventQueue == NULL)
 104         {
 105                 /* shutting down... */
 106                 free(h);
 107                 EVM_syslog(LOG_DEBUG, "DBEVENT Q exiting.\n");
 108                 pthread_mutex_unlock(&DBEventQueueMutex);
 109                 return eAbort;
 110         }
 111         EVM_syslog(LOG_DEBUG, "DBEVENT Q\n");
 112         i = ++evdb_count ;
 113         Put(DBInboundEventQueue, IKEY(i), h, NULL);
 114         pthread_mutex_unlock(&DBEventQueueMutex);
 115
 116         pthread_mutex_lock(&DBEventExitQueueMutex);
 117         if (event_db == NULL)
 118         {
 119                 pthread_mutex_unlock(&DBEventExitQueueMutex);
 120                 return eAbort;
 121         }
 122         ev_async_send (event_db, &DBAddJob);
 123         pthread_mutex_unlock(&DBEventExitQueueMutex);
 124
 125         EVM_syslog(LOG_DEBUG, "DBEVENT Q Done.\n");
 126         return eDBQuery;
 127 }
 128
 129 void StopDBWatchers(AsyncIO *IO)
 130 {
 131         SetEVState(IO, eDBStop);
 132         ev_cleanup_stop(event_db, &IO->db_abort_by_shutdown);
 133         ev_idle_stop(event_db, &IO->db_unwind_stack);
 134 }
 135
 136 void ShutDownDBCLient(AsyncIO *IO)
 137 {
 138         CitContext *Ctx =IO->CitContext;
 139         become_session(Ctx);
 140
 141         SetEVState(IO, eDBTerm);
 142         EVM_syslog(LOG_DEBUG, "DBEVENT Terminating.\n");
 143         StopDBWatchers(IO);
 144
 145         assert(IO->DBTerminate);
 146         IO->DBTerminate(IO);
 147 }
 148
 149 void
 150 DB_PerformNext(struct ev_loop *loop, ev_idle *watcher, int revents)
 151 {
 152         AsyncIO *IO = watcher->data;
 153
 154         SetEVState(IO, eDBNext);
 155         IO->Now = ev_now(event_db);
 156         EV_syslog(LOG_DEBUG, "%s()", __FUNCTION__);
 157         become_session(IO->CitContext);
 158
 159         ev_idle_stop(event_db, &IO->db_unwind_stack);
 160
 161         assert(IO->NextDBOperation);
 162         switch (IO->NextDBOperation(IO))
 163         {
 164         case eDBQuery:
 165                 break;
 166         case eSendDNSQuery:
 167         case eReadDNSReply:
 168         case eConnect:
 169         case eSendReply:
 170         case eSendMore:
 171         case eSendFile:
 172         case eReadMessage:
 173         case eReadMore:
 174         case eReadPayload:
 175         case eReadFile:
 176                 ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
 177                 break;
 178         case eTerminateConnection:
 179         case eAbort:
 180                 ev_idle_stop(event_db, &IO->db_unwind_stack);
 181                 ev_cleanup_stop(loop, &IO->db_abort_by_shutdown);
 182                 ShutDownDBCLient(IO);
 183         }
 184 }
 185
 186 eNextState NextDBOperation(AsyncIO *IO, IO_CallBack CB)
 187 {
 188         SetEVState(IO, eQDBNext);
 189         IO->NextDBOperation = CB;
 190         ev_idle_init(&IO->db_unwind_stack,
 191                      DB_PerformNext);
 192         IO->db_unwind_stack.data = IO;
 193         ev_idle_start(event_db, &IO->db_unwind_stack);
 194         return eDBQuery;
 195 }
 196
 197 /*------------------------------------------------------------------------------
 198  *                      Client IO
 199  *----------------------------------------------------------------------------*/
 200 extern int evbase_count;
 201 extern pthread_mutex_t EventQueueMutex;
 202 extern pthread_mutex_t EventExitQueueMutex;
 203 extern HashList *InboundEventQueue;
 204 extern struct ev_loop *event_base;
 205 extern ev_async AddJob;
 206 extern ev_async ExitEventLoop;
 207
 208 static void IO_abort_shutdown_callback(struct ev_loop *loop,
 209                                        ev_cleanup *watcher,
 210                                        int revents)
 211 {
 212         AsyncIO *IO = watcher->data;
 213
 214         SetEVState(IO, eIOAbort);
 215         EV_syslog(LOG_DEBUG, "EVENT Q: %s\n", __FUNCTION__);
 216         IO->Now = ev_now(event_base);
 217         assert(IO->ShutdownAbort);
 218         IO->ShutdownAbort(IO);
 219 }
 220
 221
 222 eNextState QueueEventContext(AsyncIO *IO, IO_CallBack CB)
 223 {
 224         IOAddHandler *h;
 225         int i;
 226
 227         SetEVState(IO, eIOQ);
 228         h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
 229         h->IO = IO;
 230         h->EvAttch = CB;
 231         ev_cleanup_init(&IO->abort_by_shutdown,
 232                         IO_abort_shutdown_callback);
 233         IO->abort_by_shutdown.data = IO;
 234
 235         pthread_mutex_lock(&EventQueueMutex);
 236         if (InboundEventQueue == NULL)
 237         {
 238                 free(h);
 239                 /* shutting down... */
 240                 EVM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 241                 pthread_mutex_unlock(&EventQueueMutex);
 242                 return eAbort;
 243         }
 244         EVM_syslog(LOG_DEBUG, "EVENT Q\n");
 245         i = ++evbase_count;
 246         Put(InboundEventQueue, IKEY(i), h, NULL);
 247         pthread_mutex_unlock(&EventQueueMutex);
 248
 249         pthread_mutex_lock(&EventExitQueueMutex);
 250         if (event_base == NULL) {
 251                 pthread_mutex_unlock(&EventExitQueueMutex);
 252                 return eAbort;
 253         }
 254         ev_async_send (event_base, &AddJob);
 255         pthread_mutex_unlock(&EventExitQueueMutex);
 256         EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
 257         return eSendReply;
 258 }
 259
 260 eNextState EventQueueDBOperation(AsyncIO *IO, IO_CallBack CB)
 261 {
 262         StopClientWatchers(IO, 0);
 263         return QueueDBOperation(IO, CB);
 264 }
 265 eNextState DBQueueEventContext(AsyncIO *IO, IO_CallBack CB)
 266 {
 267         StopDBWatchers(IO);
 268         return QueueEventContext(IO, CB);
 269 }
 270
 271 extern eNextState evcurl_handle_start(AsyncIO *IO);
 272
 273 eNextState QueueCurlContext(AsyncIO *IO)
 274 {
 275         IOAddHandler *h;
 276         int i;
 277
 278         SetEVState(IO, eCurlQ);
 279         h = (IOAddHandler*)malloc(sizeof(IOAddHandler));
 280         h->IO = IO;
 281         h->EvAttch = evcurl_handle_start;
 282
 283         pthread_mutex_lock(&EventQueueMutex);
 284         if (InboundEventQueue == NULL)
 285         {
 286                 /* shutting down... */
 287                 free(h);
 288                 EVM_syslog(LOG_DEBUG, "EVENT Q exiting.\n");
 289                 pthread_mutex_unlock(&EventQueueMutex);
 290                 return eAbort;
 291         }
 292
 293         EVM_syslog(LOG_DEBUG, "EVENT Q\n");
 294         i = ++evbase_count;
 295         Put(InboundEventQueue, IKEY(i), h, NULL);
 296         pthread_mutex_unlock(&EventQueueMutex);
 297
 298         pthread_mutex_lock(&EventExitQueueMutex);
 299         if (event_base == NULL) {
 300                 pthread_mutex_unlock(&EventExitQueueMutex);
 301                 return eAbort;
 302         }
 303         ev_async_send (event_base, &AddJob);
 304         pthread_mutex_unlock(&EventExitQueueMutex);
 305
 306         EVM_syslog(LOG_DEBUG, "EVENT Q Done.\n");
 307         return eSendReply;
 308 }
 309
 310 eNextState CurlQueueDBOperation(AsyncIO *IO, IO_CallBack CB)
 311 {
 312         StopCurlWatchers(IO);
 313         return QueueDBOperation(IO, CB);
 314 }
 315
 316
 317 void DestructCAres(AsyncIO *IO);
 318 void FreeAsyncIOContents(AsyncIO *IO)
 319 {
 320         CitContext *Ctx = IO->CitContext;
 321
 322         FreeStrBuf(&IO->IOBuf);
 323         FreeStrBuf(&IO->SendBuf.Buf);
 324         FreeStrBuf(&IO->RecvBuf.Buf);
 325
 326         DestructCAres(IO);
 327
 328         FreeURL(&IO->ConnectMe);
 329         FreeStrBuf(&IO->HttpReq.ReplyData);
 330
 331         if (Ctx) {
 332                 Ctx->state = CON_IDLE;
 333                 Ctx->kill_me = 1;
 334                 IO->CitContext = NULL;
 335         }
 336 }
 337
 338
 339 void StopClientWatchers(AsyncIO *IO, int CloseFD)
 340 {
 341         EVM_syslog(LOG_DEBUG, "EVENT StopClientWatchers");
 342
 343         ev_timer_stop (event_base, &IO->rw_timeout);
 344         ev_timer_stop(event_base, &IO->conn_fail);
 345         ev_idle_stop(event_base, &IO->unwind_stack);
 346         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
 347
 348         ev_io_stop(event_base, &IO->conn_event);
 349         ev_io_stop(event_base, &IO->send_event);
 350         ev_io_stop(event_base, &IO->recv_event);
 351
 352         if (CloseFD && (IO->SendBuf.fd > 0)) {
 353                 close(IO->SendBuf.fd);
 354                 IO->SendBuf.fd = 0;
 355                 IO->RecvBuf.fd = 0;
 356         }
 357 }
 358
 359 void StopCurlWatchers(AsyncIO *IO)
 360 {
 361         EVM_syslog(LOG_DEBUG, "EVENT StopCurlWatchers \n");
 362
 363         ev_timer_stop (event_base, &IO->rw_timeout);
 364         ev_timer_stop(event_base, &IO->conn_fail);
 365         ev_idle_stop(event_base, &IO->unwind_stack);
 366         ev_cleanup_stop(event_base, &IO->abort_by_shutdown);
 367
 368         ev_io_stop(event_base, &IO->conn_event);
 369         ev_io_stop(event_base, &IO->send_event);
 370         ev_io_stop(event_base, &IO->recv_event);
 371
 372         curl_easy_cleanup(IO->HttpReq.chnd);
 373         IO->HttpReq.chnd = NULL;
 374
 375         if (IO->SendBuf.fd != 0) {
 376                 close(IO->SendBuf.fd);
 377         }
 378         IO->SendBuf.fd = 0;
 379         IO->RecvBuf.fd = 0;
 380 }
 381
 382 void ShutDownCLient(AsyncIO *IO)
 383 {
 384         CitContext *Ctx =IO->CitContext;
 385
 386         SetEVState(IO, eExit);
 387         become_session(Ctx);
 388
 389         EVM_syslog(LOG_DEBUG, "EVENT Terminating \n");
 390
 391         StopClientWatchers(IO, 1);
 392
 393         if (IO->DNS.Channel != NULL) {
 394                 ares_destroy(IO->DNS.Channel);
 395                 EV_DNS_LOG_STOP(DNS.recv_event);
 396                 EV_DNS_LOG_STOP(DNS.send_event);
 397                 ev_io_stop(event_base, &IO->DNS.recv_event);
 398                 ev_io_stop(event_base, &IO->DNS.send_event);
 399                 IO->DNS.Channel = NULL;
 400         }
 401         assert(IO->Terminate);
 402         IO->Terminate(IO);
 403 }
 404
 405 void PostInbound(AsyncIO *IO)
 406 {
 407         switch (IO->NextState) {
 408         case eSendFile:
 409                 ev_io_start(event_base, &IO->send_event);
 410                 break;
 411         case eSendReply:
 412         case eSendMore:
 413                 assert(IO->SendDone);
 414                 IO->NextState = IO->SendDone(IO);
 415                 switch (IO->NextState)
 416                 {
 417                 case eSendFile:
 418                 case eSendReply:
 419                 case eSendMore:
 420                 case eReadMessage:
 421                 case eReadPayload:
 422                 case eReadMore:
 423                 case eReadFile:
 424                         ev_io_start(event_base, &IO->send_event);
 425                         break;
 426                 case eDBQuery:
 427                         StopClientWatchers(IO, 0);
 428                 default:
 429                         break;
 430                 }
 431                 break;
 432         case eReadPayload:
 433         case eReadMore:
 434         case eReadFile:
 435                 ev_io_start(event_base, &IO->recv_event);
 436                 break;
 437         case eTerminateConnection:
 438                 ShutDownCLient(IO);
 439                 break;
 440         case eAbort:
 441                 ShutDownCLient(IO);
 442                 break;
 443         case eSendDNSQuery:
 444         case eReadDNSReply:
 445         case eDBQuery:
 446         case eConnect:
 447         case eReadMessage:
 448                 break;
 449         }
 450 }
 451 eReadState HandleInbound(AsyncIO *IO)
 452 {
 453         const char *Err = NULL;
 454         eReadState Finished = eBufferNotEmpty;
 455
 456         become_session(IO->CitContext);
 457
 458         while ((Finished == eBufferNotEmpty) &&
 459                ((IO->NextState == eReadMessage)||
 460                 (IO->NextState == eReadMore)||
 461                 (IO->NextState == eReadFile)||
 462                 (IO->NextState == eReadPayload)))
 463         {
 464                 /* Reading lines...
 465                  * lex line reply in callback,
 466                  * or do it ourselves.
 467                  * i.e. as nnn-blabla means continue reading in SMTP
 468                  */
 469                 if ((IO->NextState == eReadFile) &&
 470                     (Finished == eBufferNotEmpty))
 471                 {
 472                         Finished = WriteIOBAlreadyRead(&IO->IOB, &Err);
 473                         if (Finished == eReadSuccess)
 474                         {
 475                                 IO->NextState = eSendReply;
 476                         }
 477                 }
 478                 else if (IO->LineReader)
 479                         Finished = IO->LineReader(IO);
 480                 else
 481                         Finished = StrBufChunkSipLine(IO->IOBuf,
 482                                                       &IO->RecvBuf);
 483
 484                 switch (Finished) {
 485                 case eMustReadMore: /// read new from socket...
 486                         break;
 487                 case eBufferNotEmpty: /* shouldn't happen... */
 488                 case eReadSuccess: /// done for now...
 489                         break;
 490                 case eReadFail: /// WHUT?
 491                                 ///todo: shut down!
 492                         break;
 493                 }
 494
 495                 if (Finished != eMustReadMore) {
 496                         assert(IO->ReadDone);
 497                         ev_io_stop(event_base, &IO->recv_event);
 498                         IO->NextState = IO->ReadDone(IO);
 499                         Finished = StrBufCheckBuffer(&IO->RecvBuf);
 500                 }
 501         }
 502
 503         PostInbound(IO);
 504
 505         return Finished;
 506 }
 507
 508
 509 static void
 510 IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 511 {
 512         int rc;
 513         AsyncIO *IO = watcher->data;
 514         const char *errmsg = NULL;
 515
 516         IO->Now = ev_now(event_base);
 517         become_session(IO->CitContext);
 518 #ifdef BIGBAD_IODBG
 519         {
 520                 int rv = 0;
 521                 char fn [SIZ];
 522                 FILE *fd;
 523                 const char *pch = ChrPtr(IO->SendBuf.Buf);
 524                 const char *pchh = IO->SendBuf.ReadWritePointer;
 525                 long nbytes;
 526
 527                 if (pchh == NULL)
 528                         pchh = pch;
 529
 530                 nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch);
 531                 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
 532                          ((CitContext*)(IO->CitContext))->ServiceName,
 533                          IO->SendBuf.fd);
 534
 535                 fd = fopen(fn, "a+");
 536                 fprintf(fd, "Send: BufSize: %ld BufContent: [",
 537                         nbytes);
 538                 rv = fwrite(pchh, nbytes, 1, fd);
 539                 if (!rv) printf("failed to write debug to %s!\n", fn);
 540                 fprintf(fd, "]\n");
 541 #endif
 542                 switch (IO->NextState) {
 543                 case eSendFile:
 544                         rc = FileSendChunked(&IO->IOB, &errmsg);
 545                         if (rc < 0)
 546                                 StrBufPlain(IO->ErrMsg, errmsg, -1);
 547                         break;
 548                 default:
 549                         rc = StrBuf_write_one_chunk_callback(IO->SendBuf.fd,
 550                                                              0,
 551                                                              &IO->SendBuf);
 552                 }
 553
 554 #ifdef BIGBAD_IODBG
 555                 fprintf(fd, "Sent: BufSize: %d bytes.\n", rc);
 556                 fclose(fd);
 557         }
 558 #endif
 559         if (rc == 0)
 560         {
 561                 ev_io_stop(event_base, &IO->send_event);
 562                 switch (IO->NextState) {
 563                 case eSendMore:
 564                         assert(IO->SendDone);
 565                         IO->NextState = IO->SendDone(IO);
 566
 567                         if ((IO->NextState == eTerminateConnection) ||
 568                             (IO->NextState == eAbort) )
 569                                 ShutDownCLient(IO);
 570                         else {
 571                                 ev_io_start(event_base, &IO->send_event);
 572                         }
 573                         break;
 574                 case eSendFile:
 575                         if (IO->IOB.ChunkSendRemain > 0) {
 576                                 ev_io_start(event_base, &IO->recv_event);
 577                                 SetNextTimeout(IO, 100.0);
 578
 579                         } else {
 580                                 assert(IO->ReadDone);
 581                                 IO->NextState = IO->ReadDone(IO);
 582                                 switch(IO->NextState) {
 583                                 case eSendDNSQuery:
 584                                 case eReadDNSReply:
 585                                 case eDBQuery:
 586                                 case eConnect:
 587                                         break;
 588                                 case eSendReply:
 589                                 case eSendMore:
 590                                 case eSendFile:
 591                                         ev_io_start(event_base,
 592                                                     &IO->send_event);
 593                                         break;
 594                                 case eReadMessage:
 595                                 case eReadMore:
 596                                 case eReadPayload:
 597                                 case eReadFile:
 598                                         break;
 599                                 case eTerminateConnection:
 600                                 case eAbort:
 601                                         break;
 602                                 }
 603                         }
 604                         break;
 605                 case eSendReply:
 606                     if (StrBufCheckBuffer(&IO->SendBuf) != eReadSuccess)
 607                         break;
 608                     IO->NextState = eReadMore;
 609                 case eReadMore:
 610                 case eReadMessage:
 611                 case eReadPayload:
 612                 case eReadFile:
 613                         if (StrBufCheckBuffer(&IO->RecvBuf) == eBufferNotEmpty)
 614                         {
 615                                 HandleInbound(IO);
 616                         }
 617                         else {
 618                                 ev_io_start(event_base, &IO->recv_event);
 619                         }
 620
 621                         break;
 622                 case eDBQuery:
 623                         /*
 624                          * we now live in another queue,
 625                          * so we have to unregister.
 626                          */
 627                         ev_cleanup_stop(loop, &IO->abort_by_shutdown);
 628                         break;
 629                 case eSendDNSQuery:
 630                 case eReadDNSReply:
 631                 case eConnect:
 632                 case eTerminateConnection:
 633                 case eAbort:
 634                         break;
 635                 }
 636         }
 637         else if (rc < 0) {
 638                 if (errno != EAGAIN) {
 639                         StopClientWatchers(IO, 1);
 640                         EV_syslog(LOG_DEBUG,
 641                                   "IO_send_callback(): Socket Invalid! [%d] [%s] [%d]\n",
 642                                   errno, strerror(errno), IO->SendBuf.fd);
 643                         StrBufPrintf(IO->ErrMsg,
 644                                      "Socket Invalid! [%s]",
 645                                      strerror(errno));
 646                         SetNextTimeout(IO, 0.01);
 647                 }
 648         }
 649         /* else : must write more. */
 650 }
 651 static void
 652 set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents)
 653 {
 654         ev_timer_stop(event_base, &IO->conn_fail);
 655         ev_timer_start(event_base, &IO->rw_timeout);
 656
 657         switch(IO->NextState) {
 658         case eReadMore:
 659         case eReadMessage:
 660         case eReadFile:
 661                 StrBufAppendBufPlain(IO->ErrMsg, HKEY("[while waiting for greeting]"), 0);
 662                 ev_io_start(event_base, &IO->recv_event);
 663                 break;
 664         case eSendReply:
 665         case eSendMore:
 666         case eReadPayload:
 667         case eSendFile:
 668                 become_session(IO->CitContext);
 669                 IO_send_callback(loop, &IO->send_event, revents);
 670                 break;
 671         case eDBQuery:
 672         case eSendDNSQuery:
 673         case eReadDNSReply:
 674         case eConnect:
 675         case eTerminateConnection:
 676         case eAbort:
 677                 /// TODO: WHUT?
 678                 break;
 679         }
 680 }
 681
 682 static void
 683 IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
 684 {
 685         AsyncIO *IO = watcher->data;
 686
 687         SetEVState(IO, eIOTimeout);
 688         IO->Now = ev_now(event_base);
 689         ev_timer_stop (event_base, &IO->rw_timeout);
 690         become_session(IO->CitContext);
 691
 692         if (IO->SendBuf.fd != 0)
 693         {
 694                 ev_io_stop(event_base, &IO->send_event);
 695                 ev_io_stop(event_base, &IO->recv_event);
 696                 ev_timer_stop (event_base, &IO->rw_timeout);
 697                 close(IO->SendBuf.fd);
 698                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 699         }
 700
 701         assert(IO->Timeout);
 702         switch (IO->Timeout(IO))
 703         {
 704         case eAbort:
 705                 ShutDownCLient(IO);
 706         default:
 707                 break;
 708         }
 709 }
 710
 711 static void
 712 IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
 713 {
 714         AsyncIO *IO = watcher->data;
 715
 716         SetEVState(IO, eIOConnfail);
 717         IO->Now = ev_now(event_base);
 718         ev_timer_stop (event_base, &IO->conn_fail);
 719
 720         if (IO->SendBuf.fd != 0)
 721         {
 722                 ev_io_stop(loop, &IO->conn_event);
 723                 ev_io_stop(event_base, &IO->send_event);
 724                 ev_io_stop(event_base, &IO->recv_event);
 725                 ev_timer_stop (event_base, &IO->rw_timeout);
 726                 close(IO->SendBuf.fd);
 727                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 728         }
 729         become_session(IO->CitContext);
 730
 731         assert(IO->ConnFail);
 732         switch (IO->ConnFail(IO))
 733         {
 734         case eAbort:
 735                 ShutDownCLient(IO);
 736         default:
 737                 break;
 738
 739         }
 740 }
 741
 742 static void
 743 IO_connfailimmediate_callback(struct ev_loop *loop,
 744                               ev_idle *watcher,
 745                               int revents)
 746 {
 747         AsyncIO *IO = watcher->data;
 748
 749         SetEVState(IO, eIOConnfailNow);
 750         IO->Now = ev_now(event_base);
 751         ev_idle_stop (event_base, &IO->conn_fail_immediate);
 752
 753         if (IO->SendBuf.fd != 0)
 754         {
 755                 close(IO->SendBuf.fd);
 756                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 757         }
 758         become_session(IO->CitContext);
 759
 760         assert(IO->ConnFail);
 761         switch (IO->ConnFail(IO))
 762         {
 763         case eAbort:
 764                 ShutDownCLient(IO);
 765         default:
 766                 break;
 767
 768         }
 769 }
 770
 771 static void
 772 IO_connestd_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 773 {
 774         AsyncIO *IO = watcher->data;
 775         int             so_err = 0;
 776         socklen_t       lon = sizeof(so_err);
 777         int             err;
 778
 779         SetEVState(IO, eIOConnNow);
 780         IO->Now = ev_now(event_base);
 781         EVM_syslog(LOG_DEBUG, "connect() succeeded.\n");
 782
 783         ev_io_stop(loop, &IO->conn_event);
 784         ev_timer_stop(event_base, &IO->conn_fail);
 785
 786         err = getsockopt(IO->SendBuf.fd,
 787                          SOL_SOCKET,
 788                          SO_ERROR,
 789                          (void*)&so_err,
 790                          &lon);
 791
 792         if ((err == 0) && (so_err != 0))
 793         {
 794                 EV_syslog(LOG_DEBUG, "connect() failed [%d][%s]\n",
 795                           so_err,
 796                           strerror(so_err));
 797                 IO_connfail_callback(loop, &IO->conn_fail, revents);
 798
 799         }
 800         else
 801         {
 802                 EVM_syslog(LOG_DEBUG, "connect() succeeded\n");
 803                 set_start_callback(loop, IO, revents);
 804         }
 805 }
 806
 807 static void
 808 IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 809 {
 810         const char *errmsg;
 811         ssize_t nbytes;
 812         AsyncIO *IO = watcher->data;
 813
 814         IO->Now = ev_now(event_base);
 815         switch (IO->NextState) {
 816         case eReadFile:
 817                 nbytes = FileRecvChunked(&IO->IOB, &errmsg);
 818                 if (nbytes < 0)
 819                         StrBufPlain(IO->ErrMsg, errmsg, -1);
 820                 else
 821                 {
 822                         if (IO->IOB.ChunkSendRemain == 0)
 823                         {
 824                                 IO->NextState = eSendReply;
 825                                 assert(IO->ReadDone);
 826                                 ev_io_stop(event_base, &IO->recv_event);
 827                                 PostInbound(IO);
 828                                 return;
 829                         }
 830                         else
 831                                 return;
 832                 }
 833                 break;
 834         default:
 835                 nbytes = StrBuf_read_one_chunk_callback(IO->RecvBuf.fd,
 836                                                         0,
 837                                                         &IO->RecvBuf);
 838                 break;
 839         }
 840
 841 #ifdef BIGBAD_IODBG
 842         {
 843                 long nbytes;
 844                 int rv = 0;
 845                 char fn [SIZ];
 846                 FILE *fd;
 847                 const char *pch = ChrPtr(IO->RecvBuf.Buf);
 848                 const char *pchh = IO->RecvBuf.ReadWritePointer;
 849
 850                 if (pchh == NULL)
 851                         pchh = pch;
 852
 853                 nbytes = StrLength(IO->RecvBuf.Buf) - (pchh - pch);
 854                 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
 855                          ((CitContext*)(IO->CitContext))->ServiceName,
 856                          IO->SendBuf.fd);
 857
 858                 fd = fopen(fn, "a+");
 859                 fprintf(fd, "Read: BufSize: %ld BufContent: [",
 860                         nbytes);
 861                 rv = fwrite(pchh, nbytes, 1, fd);
 862                 if (!rv) printf("failed to write debug to %s!\n", fn);
 863                 fprintf(fd, "]\n");
 864                 fclose(fd);
 865         }
 866 #endif
 867         if (nbytes > 0) {
 868                 HandleInbound(IO);
 869         } else if (nbytes == 0) {
 870                 StopClientWatchers(IO, 1);
 871                 SetNextTimeout(IO, 0.01);
 872                 return;
 873         } else if (nbytes == -1) {
 874                 if (errno != EAGAIN) {
 875                         // FD is gone. kick it.
 876                         StopClientWatchers(IO, 1);
 877                         EV_syslog(LOG_DEBUG,
 878                                   "IO_recv_callback(): Socket Invalid! [%d] [%s] [%d]\n",
 879                                   errno, strerror(errno), IO->SendBuf.fd);
 880                         StrBufPrintf(IO->ErrMsg,
 881                                      "Socket Invalid! [%s]",
 882                                      strerror(errno));
 883                         SetNextTimeout(IO, 0.01);
 884                 }
 885                 return;
 886         }
 887 }
 888
 889 void
 890 IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents)
 891 {
 892         AsyncIO *IO = watcher->data;
 893
 894         SetEVState(IO, eCaresFinished);
 895         IO->Now = ev_now(event_base);
 896         EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__);
 897         become_session(IO->CitContext);
 898         assert(IO->DNS.Query->PostDNS);
 899         switch (IO->DNS.Query->PostDNS(IO))
 900         {
 901         case eAbort:
 902                 assert(IO->DNS.Fail);
 903                 switch (IO->DNS.Fail(IO)) {
 904                 case eAbort:
 905 ////                    StopClientWatchers(IO);
 906                         ShutDownCLient(IO);
 907                 default:
 908                         break;
 909                 }
 910         default:
 911                 break;
 912         }
 913 }
 914
 915
 916 eNextState EvConnectSock(AsyncIO *IO,
 917                          double conn_timeout,
 918                          double first_rw_timeout,
 919                          int ReadFirst)
 920 {
 921         struct sockaddr_in egress_sin;
 922         int fdflags;
 923         int rc = -1;
 924
 925         SetEVState(IO, eIOConnectSock);
 926         become_session(IO->CitContext);
 927
 928         if (ReadFirst) {
 929                 IO->NextState = eReadMessage;
 930         }
 931         else {
 932                 IO->NextState = eSendReply;
 933         }
 934
 935         IO->SendBuf.fd = IO->RecvBuf.fd =
 936                 socket(
 937                         (IO->ConnectMe->IPv6)?PF_INET6:PF_INET,
 938                         SOCK_STREAM,
 939                         IPPROTO_TCP);
 940
 941         if (IO->SendBuf.fd < 0) {
 942                 EV_syslog(LOG_ERR,
 943                           "EVENT: socket() failed: %s\n",
 944                           strerror(errno));
 945
 946                 StrBufPrintf(IO->ErrMsg,
 947                              "Failed to create socket: %s",
 948                              strerror(errno));
 949                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 950                 return eAbort;
 951         }
 952         fdflags = fcntl(IO->SendBuf.fd, F_GETFL);
 953         if (fdflags < 0) {
 954                 EV_syslog(LOG_ERR,
 955                           "EVENT: unable to get socket %d flags! %s \n",
 956                           IO->SendBuf.fd,
 957                           strerror(errno));
 958                 StrBufPrintf(IO->ErrMsg,
 959                              "Failed to get socket %d flags: %s",
 960                              IO->SendBuf.fd,
 961                              strerror(errno));
 962                 close(IO->SendBuf.fd);
 963                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 964                 return eAbort;
 965         }
 966         fdflags = fdflags | O_NONBLOCK;
 967         if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) {
 968                 EV_syslog(
 969                         LOG_ERR,
 970                         "EVENT: unable to set socket %d nonblocking flags! %s \n",
 971                         IO->SendBuf.fd,
 972                         strerror(errno));
 973                 StrBufPrintf(IO->ErrMsg,
 974                              "Failed to set socket flags: %s",
 975                              strerror(errno));
 976                 close(IO->SendBuf.fd);
 977                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 978                 return eAbort;
 979         }
 980 /* TODO: maye we could use offsetof() to calc the position of data...
 981  * http://doc.dvgu.ru/devel/ev.html#associating_custom_data_with_a_watcher
 982  */
 983         ev_io_init(&IO->recv_event, IO_recv_callback, IO->RecvBuf.fd, EV_READ);
 984         IO->recv_event.data = IO;
 985         ev_io_init(&IO->send_event, IO_send_callback, IO->SendBuf.fd, EV_WRITE);
 986         IO->send_event.data = IO;
 987
 988         ev_timer_init(&IO->conn_fail, IO_connfail_callback, conn_timeout, 0);
 989         IO->conn_fail.data = IO;
 990         ev_timer_init(&IO->rw_timeout, IO_Timeout_callback, first_rw_timeout,0);
 991         IO->rw_timeout.data = IO;
 992
 993
 994
 995
 996         /* for debugging you may bypass it like this:
 997          * IO->Addr.sin_addr.s_addr = inet_addr("127.0.0.1");
 998          * ((struct sockaddr_in)IO->ConnectMe->Addr).sin_addr.s_addr =
 999          *   inet_addr("127.0.0.1");
1000          */
1001         if (IO->ConnectMe->IPv6) {
1002                 rc = connect(IO->SendBuf.fd,
1003                              &IO->ConnectMe->Addr,
1004                              sizeof(struct sockaddr_in6));
1005         }
1006         else {
1007                 /* If citserver is bound to a specific IP address on the host, make
1008                  * sure we use that address for outbound connections.
1009                  */
1010
1011                 memset(&egress_sin, 0, sizeof(egress_sin));
1012                 egress_sin.sin_family = AF_INET;
1013                 if (!IsEmptyStr(config.c_ip_addr)) {
1014                         egress_sin.sin_addr.s_addr = inet_addr(config.c_ip_addr);
1015                         if (egress_sin.sin_addr.s_addr == !INADDR_ANY) {
1016                                 egress_sin.sin_addr.s_addr = INADDR_ANY;
1017                         }
1018
1019                         /* If this bind fails, no problem; we can still use INADDR_ANY */
1020                         bind(IO->SendBuf.fd, (struct sockaddr *)&egress_sin, sizeof(egress_sin));
1021                 }
1022                 rc = connect(IO->SendBuf.fd,
1023                              (struct sockaddr_in *)&IO->ConnectMe->Addr,
1024                              sizeof(struct sockaddr_in));
1025         }
1026
1027         if (rc >= 0){
1028                 SetEVState(IO, eIOConnNow);
1029                 EV_syslog(LOG_DEBUG, "connect() = %d immediate success.\n", IO->SendBuf.fd);
1030                 set_start_callback(event_base, IO, 0);
1031                 return IO->NextState;
1032         }
1033         else if (errno == EINPROGRESS) {
1034                 SetEVState(IO, eIOConnWait);
1035                 EV_syslog(LOG_DEBUG, "connect() = %d have to wait now.\n", IO->SendBuf.fd);
1036
1037                 ev_io_init(&IO->conn_event,
1038                            IO_connestd_callback,
1039                            IO->SendBuf.fd,
1040                            EV_READ|EV_WRITE);
1041
1042                 IO->conn_event.data = IO;
1043
1044                 ev_io_start(event_base, &IO->conn_event);
1045                 ev_timer_start(event_base, &IO->conn_fail);
1046                 return IO->NextState;
1047         }
1048         else {
1049                 SetEVState(IO, eIOConnfail);
1050                 ev_idle_init(&IO->conn_fail_immediate,
1051                              IO_connfailimmediate_callback);
1052                 IO->conn_fail_immediate.data = IO;
1053                 ev_idle_start(event_base, &IO->conn_fail_immediate);
1054
1055                 EV_syslog(LOG_ERR,
1056                           "connect() = %d failed: %s\n",
1057                           IO->SendBuf.fd,
1058                           strerror(errno));
1059
1060                 StrBufPrintf(IO->ErrMsg,
1061                              "Failed to connect: %s",
1062                              strerror(errno));
1063                 return IO->NextState;
1064         }
1065         return IO->NextState;
1066 }
1067
1068 void SetNextTimeout(AsyncIO *IO, double timeout)
1069 {
1070         IO->rw_timeout.repeat = timeout;
1071         ev_timer_again (event_base,  &IO->rw_timeout);
1072 }
1073
1074
1075 eNextState ReAttachIO(AsyncIO *IO,
1076                       void *pData,
1077                       int ReadFirst)
1078 {
1079         SetEVState(IO, eIOAttach);
1080         IO->Data = pData;
1081         become_session(IO->CitContext);
1082         ev_cleanup_start(event_base, &IO->abort_by_shutdown);
1083         if (ReadFirst) {
1084                 IO->NextState = eReadMessage;
1085         }
1086         else {
1087                 IO->NextState = eSendReply;
1088         }
1089         set_start_callback(event_base, IO, 0);
1090
1091         return IO->NextState;
1092 }
1093
1094 void InitIOStruct(AsyncIO *IO,
1095                   void *Data,
1096                   eNextState NextState,
1097                   IO_LineReaderCallback LineReader,
1098                   IO_CallBack DNS_Fail,
1099                   IO_CallBack SendDone,
1100                   IO_CallBack ReadDone,
1101                   IO_CallBack Terminate,
1102                   IO_CallBack DBTerminate,
1103                   IO_CallBack ConnFail,
1104                   IO_CallBack Timeout,
1105                   IO_CallBack ShutdownAbort)
1106 {
1107         IO->Data          = Data;
1108
1109         IO->CitContext    = CloneContext(CC);
1110         IO->CitContext->session_specific_data = Data;
1111         IO->CitContext->IO = IO;
1112
1113         IO->NextState     = NextState;
1114
1115         IO->SendDone      = SendDone;
1116         IO->ReadDone      = ReadDone;
1117         IO->Terminate     = Terminate;
1118         IO->DBTerminate   = DBTerminate;
1119         IO->LineReader    = LineReader;
1120         IO->ConnFail      = ConnFail;
1121         IO->Timeout       = Timeout;
1122         IO->ShutdownAbort = ShutdownAbort;
1123
1124         IO->DNS.Fail      = DNS_Fail;
1125
1126         IO->SendBuf.Buf   = NewStrBufPlain(NULL, 1024);
1127         IO->RecvBuf.Buf   = NewStrBufPlain(NULL, 1024);
1128         IO->IOBuf         = NewStrBuf();
1129         EV_syslog(LOG_DEBUG,
1130                   "EVENT: Session lives at %p IO at %p \n",
1131                   Data, IO);
1132
1133 }
1134
1135 extern int evcurl_init(AsyncIO *IO);
1136
1137 int InitcURLIOStruct(AsyncIO *IO,
1138                      void *Data,
1139                      const char* Desc,
1140                      IO_CallBack SendDone,
1141                      IO_CallBack Terminate,
1142                      IO_CallBack DBTerminate,
1143                      IO_CallBack ShutdownAbort)
1144 {
1145         IO->Data          = Data;
1146
1147         IO->CitContext    = CloneContext(CC);
1148         IO->CitContext->session_specific_data = Data;
1149         IO->CitContext->IO = IO;
1150
1151         IO->SendDone      = SendDone;
1152         IO->Terminate     = Terminate;
1153         IO->DBTerminate   = DBTerminate;
1154         IO->ShutdownAbort = ShutdownAbort;
1155
1156         strcpy(IO->HttpReq.errdesc, Desc);
1157
1158
1159         return  evcurl_init(IO);
1160
1161 }
1162
1163
1164 typedef struct KillOtherSessionContext {
1165         AsyncIO IO;
1166         AsyncIO *OtherOne;
1167 }KillOtherSessionContext;
1168
1169 eNextState KillTerminate(AsyncIO *IO)
1170 {
1171         long id;
1172         KillOtherSessionContext *Ctx = (KillOtherSessionContext*)IO->Data;
1173         EV_syslog(LOG_DEBUG, "%s Exit\n", __FUNCTION__);
1174         id = IO->ID;
1175         FreeAsyncIOContents(IO);
1176         memset(Ctx, 0, sizeof(KillOtherSessionContext));
1177         IO->ID = id; /* just for the case we want to analyze it in a coredump */
1178         free(Ctx);
1179         return eAbort;
1180
1181 }
1182
1183 eNextState KillShutdown(AsyncIO *IO)
1184 {
1185         return eTerminateConnection;
1186 }
1187
1188 eNextState KillOtherContextNow(AsyncIO *IO)
1189 {
1190         KillOtherSessionContext *Ctx = IO->Data;
1191
1192         SetEVState(IO, eKill);
1193
1194         if (Ctx->OtherOne->ShutdownAbort != NULL)
1195                 Ctx->OtherOne->ShutdownAbort(Ctx->OtherOne);
1196         return eTerminateConnection;
1197 }
1198
1199 void KillAsyncIOContext(AsyncIO *IO)
1200 {
1201         KillOtherSessionContext *Ctx;
1202
1203         Ctx = (KillOtherSessionContext*) malloc(sizeof(KillOtherSessionContext));
1204         memset(Ctx, 0, sizeof(KillOtherSessionContext));
1205
1206         InitIOStruct(&Ctx->IO,
1207                      Ctx,
1208                      eReadMessage,
1209                      NULL,
1210                      NULL,
1211                      NULL,
1212                      NULL,
1213                      KillTerminate,
1214                      NULL,
1215                      NULL,
1216                      NULL,
1217                      KillShutdown);
1218
1219         Ctx->OtherOne = IO;
1220
1221         switch(IO->NextState) {
1222         case eSendDNSQuery:
1223         case eReadDNSReply:
1224
1225         case eConnect:
1226         case eSendReply:
1227         case eSendMore:
1228         case eSendFile:
1229
1230         case eReadMessage:
1231         case eReadMore:
1232         case eReadPayload:
1233         case eReadFile:
1234                 QueueEventContext(&Ctx->IO, KillOtherContextNow);
1235                 break;
1236         case eDBQuery:
1237                 QueueDBOperation(&Ctx->IO, KillOtherContextNow);
1238                 break;
1239         case eTerminateConnection:
1240         case eAbort:
1241                 /*hm, its already dying, dunno which Queue its in... */
1242                 free(Ctx);
1243         }
1244
1245 }
1246
1247 extern int DebugEventLoopBacktrace;
1248 void EV_backtrace(AsyncIO *IO)
1249 {
1250 #ifdef HAVE_BACKTRACE
1251         void *stack_frames[50];
1252         size_t size, i;
1253         char **strings;
1254
1255         if ((IO == NULL) || (DebugEventLoopBacktrace == 0))
1256                 return;
1257         size = backtrace(stack_frames, sizeof(stack_frames) / sizeof(void*));
1258         strings = backtrace_symbols(stack_frames, size);
1259         for (i = 0; i < size; i++) {
1260                 if (strings != NULL) {
1261                         EV_syslog(LOG_ALERT, " BT %s\n", strings[i]);
1262                 }
1263                 else {
1264                         EV_syslog(LOG_ALERT, " BT %p\n", stack_frames[i]);
1265                 }
1266         }
1267         free(strings);
1268 #endif
1269 }
1270
1271
1272 ev_tstamp ctdl_ev_now (void)
1273 {
1274         return ev_now(event_base);
1275 }