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         EVQM_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, int CloseFDs)
 261 {
 262         StopClientWatchers(IO, CloseFDs);
 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 FreeAsyncIOContents(AsyncIO *IO)
 318 {
 319         CitContext *Ctx = IO->CitContext;
 320
 321         FreeStrBuf(&IO->IOBuf);
 322         FreeStrBuf(&IO->SendBuf.Buf);
 323         FreeStrBuf(&IO->RecvBuf.Buf);
 324
 325         FreeURL(&IO->ConnectMe);
 326         FreeStrBuf(&IO->HttpReq.ReplyData);
 327
 328         if (Ctx) {
 329                 Ctx->state = CON_IDLE;
 330                 Ctx->kill_me = 1;
 331                 IO->CitContext = NULL;
 332         }
 333 }
 334
 335
 336 void DestructCAres(AsyncIO *IO);
 337 void StopClientWatchers(AsyncIO *IO, int CloseFD)
 338 {
 339         EVM_syslog(LOG_DEBUG, "EVENT StopClientWatchers");
 340
 341         DestructCAres(IO);
 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                         eNextState rc;
 497                         assert(IO->ReadDone);
 498                         ev_io_stop(event_base, &IO->recv_event);
 499                         rc = IO->ReadDone(IO);
 500                         if  (rc != eDBQuery) {
 501                                 IO->NextState = rc;
 502                                 Finished = StrBufCheckBuffer(&IO->RecvBuf);
 503                         }
 504                         else {
 505                                 return rc;
 506
 507                         }
 508                 }
 509         }
 510
 511         PostInbound(IO);
 512
 513         return Finished;
 514 }
 515
 516
 517 static void
 518 IO_send_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 519 {
 520         int rc;
 521         AsyncIO *IO = watcher->data;
 522         const char *errmsg = NULL;
 523
 524         IO->Now = ev_now(event_base);
 525         become_session(IO->CitContext);
 526 #ifdef BIGBAD_IODBG
 527         {
 528                 int rv = 0;
 529                 char fn [SIZ];
 530                 FILE *fd;
 531                 const char *pch = ChrPtr(IO->SendBuf.Buf);
 532                 const char *pchh = IO->SendBuf.ReadWritePointer;
 533                 long nbytes;
 534
 535                 if (pchh == NULL)
 536                         pchh = pch;
 537
 538                 nbytes = StrLength(IO->SendBuf.Buf) - (pchh - pch);
 539                 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
 540                          ((CitContext*)(IO->CitContext))->ServiceName,
 541                          IO->SendBuf.fd);
 542
 543                 fd = fopen(fn, "a+");
 544                 fprintf(fd, "Send: BufSize: %ld BufContent: [",
 545                         nbytes);
 546                 rv = fwrite(pchh, nbytes, 1, fd);
 547                 if (!rv) printf("failed to write debug to %s!\n", fn);
 548                 fprintf(fd, "]\n");
 549 #endif
 550                 switch (IO->NextState) {
 551                 case eSendFile:
 552                         rc = FileSendChunked(&IO->IOB, &errmsg);
 553                         if (rc < 0)
 554                                 StrBufPlain(IO->ErrMsg, errmsg, -1);
 555                         break;
 556                 default:
 557                         rc = StrBuf_write_one_chunk_callback(IO->SendBuf.fd,
 558                                                              0,
 559                                                              &IO->SendBuf);
 560                 }
 561
 562 #ifdef BIGBAD_IODBG
 563                 fprintf(fd, "Sent: BufSize: %d bytes.\n", rc);
 564                 fclose(fd);
 565         }
 566 #endif
 567         if (rc == 0)
 568         {
 569                 ev_io_stop(event_base, &IO->send_event);
 570                 switch (IO->NextState) {
 571                 case eSendMore:
 572                         assert(IO->SendDone);
 573                         IO->NextState = IO->SendDone(IO);
 574
 575                         if ((IO->NextState == eTerminateConnection) ||
 576                             (IO->NextState == eAbort) )
 577                                 ShutDownCLient(IO);
 578                         else {
 579                                 ev_io_start(event_base, &IO->send_event);
 580                         }
 581                         break;
 582                 case eSendFile:
 583                         if (IO->IOB.ChunkSendRemain > 0) {
 584                                 ev_io_start(event_base, &IO->recv_event);
 585                                 SetNextTimeout(IO, 100.0);
 586
 587                         } else {
 588                                 assert(IO->ReadDone);
 589                                 IO->NextState = IO->ReadDone(IO);
 590                                 switch(IO->NextState) {
 591                                 case eSendDNSQuery:
 592                                 case eReadDNSReply:
 593                                 case eDBQuery:
 594                                 case eConnect:
 595                                         break;
 596                                 case eSendReply:
 597                                 case eSendMore:
 598                                 case eSendFile:
 599                                         ev_io_start(event_base,
 600                                                     &IO->send_event);
 601                                         break;
 602                                 case eReadMessage:
 603                                 case eReadMore:
 604                                 case eReadPayload:
 605                                 case eReadFile:
 606                                         break;
 607                                 case eTerminateConnection:
 608                                 case eAbort:
 609                                         break;
 610                                 }
 611                         }
 612                         break;
 613                 case eSendReply:
 614                     if (StrBufCheckBuffer(&IO->SendBuf) != eReadSuccess)
 615                         break;
 616                     IO->NextState = eReadMore;
 617                 case eReadMore:
 618                 case eReadMessage:
 619                 case eReadPayload:
 620                 case eReadFile:
 621                         if (StrBufCheckBuffer(&IO->RecvBuf) == eBufferNotEmpty)
 622                         {
 623                                 HandleInbound(IO);
 624                         }
 625                         else {
 626                                 ev_io_start(event_base, &IO->recv_event);
 627                         }
 628
 629                         break;
 630                 case eDBQuery:
 631                         /*
 632                          * we now live in another queue,
 633                          * so we have to unregister.
 634                          */
 635                         ev_cleanup_stop(loop, &IO->abort_by_shutdown);
 636                         break;
 637                 case eSendDNSQuery:
 638                 case eReadDNSReply:
 639                 case eConnect:
 640                 case eTerminateConnection:
 641                 case eAbort:
 642                         break;
 643                 }
 644         }
 645         else if (rc < 0) {
 646                 if (errno != EAGAIN) {
 647                         StopClientWatchers(IO, 1);
 648                         EV_syslog(LOG_DEBUG,
 649                                   "IO_send_callback(): Socket Invalid! [%d] [%s] [%d]\n",
 650                                   errno, strerror(errno), IO->SendBuf.fd);
 651                         StrBufPrintf(IO->ErrMsg,
 652                                      "Socket Invalid! [%s]",
 653                                      strerror(errno));
 654                         SetNextTimeout(IO, 0.01);
 655                 }
 656         }
 657         /* else : must write more. */
 658 }
 659 static void
 660 set_start_callback(struct ev_loop *loop, AsyncIO *IO, int revents)
 661 {
 662         ev_timer_stop(event_base, &IO->conn_fail);
 663         ev_timer_start(event_base, &IO->rw_timeout);
 664
 665         switch(IO->NextState) {
 666         case eReadMore:
 667         case eReadMessage:
 668         case eReadFile:
 669                 StrBufAppendBufPlain(IO->ErrMsg, HKEY("[while waiting for greeting]"), 0);
 670                 ev_io_start(event_base, &IO->recv_event);
 671                 break;
 672         case eSendReply:
 673         case eSendMore:
 674         case eReadPayload:
 675         case eSendFile:
 676                 become_session(IO->CitContext);
 677                 IO_send_callback(loop, &IO->send_event, revents);
 678                 break;
 679         case eDBQuery:
 680         case eSendDNSQuery:
 681         case eReadDNSReply:
 682         case eConnect:
 683         case eTerminateConnection:
 684         case eAbort:
 685                 /// TODO: WHUT?
 686                 break;
 687         }
 688 }
 689
 690 static void
 691 IO_Timeout_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
 692 {
 693         AsyncIO *IO = watcher->data;
 694
 695         SetEVState(IO, eIOTimeout);
 696         IO->Now = ev_now(event_base);
 697         ev_timer_stop (event_base, &IO->rw_timeout);
 698         become_session(IO->CitContext);
 699
 700         if (IO->SendBuf.fd != 0)
 701         {
 702                 ev_io_stop(event_base, &IO->send_event);
 703                 ev_io_stop(event_base, &IO->recv_event);
 704                 ev_timer_stop (event_base, &IO->rw_timeout);
 705                 close(IO->SendBuf.fd);
 706                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 707         }
 708
 709         assert(IO->Timeout);
 710         switch (IO->Timeout(IO))
 711         {
 712         case eAbort:
 713                 ShutDownCLient(IO);
 714         default:
 715                 break;
 716         }
 717 }
 718
 719 static void
 720 IO_connfail_callback(struct ev_loop *loop, ev_timer *watcher, int revents)
 721 {
 722         AsyncIO *IO = watcher->data;
 723
 724         SetEVState(IO, eIOConnfail);
 725         IO->Now = ev_now(event_base);
 726         ev_timer_stop (event_base, &IO->conn_fail);
 727
 728         if (IO->SendBuf.fd != 0)
 729         {
 730                 ev_io_stop(loop, &IO->conn_event);
 731                 ev_io_stop(event_base, &IO->send_event);
 732                 ev_io_stop(event_base, &IO->recv_event);
 733                 ev_timer_stop (event_base, &IO->rw_timeout);
 734                 close(IO->SendBuf.fd);
 735                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 736         }
 737         become_session(IO->CitContext);
 738
 739         assert(IO->ConnFail);
 740         switch (IO->ConnFail(IO))
 741         {
 742         case eAbort:
 743                 ShutDownCLient(IO);
 744         default:
 745                 break;
 746
 747         }
 748 }
 749
 750 static void
 751 IO_connfailimmediate_callback(struct ev_loop *loop,
 752                               ev_idle *watcher,
 753                               int revents)
 754 {
 755         AsyncIO *IO = watcher->data;
 756
 757         SetEVState(IO, eIOConnfailNow);
 758         IO->Now = ev_now(event_base);
 759         ev_idle_stop (event_base, &IO->conn_fail_immediate);
 760
 761         if (IO->SendBuf.fd != 0)
 762         {
 763                 close(IO->SendBuf.fd);
 764                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 765         }
 766         become_session(IO->CitContext);
 767
 768         assert(IO->ConnFail);
 769         switch (IO->ConnFail(IO))
 770         {
 771         case eAbort:
 772                 ShutDownCLient(IO);
 773         default:
 774                 break;
 775
 776         }
 777 }
 778
 779 static void
 780 IO_connestd_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 781 {
 782         AsyncIO *IO = watcher->data;
 783         int             so_err = 0;
 784         socklen_t       lon = sizeof(so_err);
 785         int             err;
 786
 787         SetEVState(IO, eIOConnNow);
 788         IO->Now = ev_now(event_base);
 789         EVM_syslog(LOG_DEBUG, "connect() succeeded.\n");
 790
 791         ev_io_stop(loop, &IO->conn_event);
 792         ev_timer_stop(event_base, &IO->conn_fail);
 793
 794         err = getsockopt(IO->SendBuf.fd,
 795                          SOL_SOCKET,
 796                          SO_ERROR,
 797                          (void*)&so_err,
 798                          &lon);
 799
 800         if ((err == 0) && (so_err != 0))
 801         {
 802                 EV_syslog(LOG_DEBUG, "connect() failed [%d][%s]\n",
 803                           so_err,
 804                           strerror(so_err));
 805                 IO_connfail_callback(loop, &IO->conn_fail, revents);
 806
 807         }
 808         else
 809         {
 810                 EVM_syslog(LOG_DEBUG, "connect() succeeded\n");
 811                 set_start_callback(loop, IO, revents);
 812         }
 813 }
 814
 815 static void
 816 IO_recv_callback(struct ev_loop *loop, ev_io *watcher, int revents)
 817 {
 818         const char *errmsg;
 819         ssize_t nbytes;
 820         AsyncIO *IO = watcher->data;
 821
 822         IO->Now = ev_now(event_base);
 823         switch (IO->NextState) {
 824         case eReadFile:
 825                 nbytes = FileRecvChunked(&IO->IOB, &errmsg);
 826                 if (nbytes < 0)
 827                         StrBufPlain(IO->ErrMsg, errmsg, -1);
 828                 else
 829                 {
 830                         if (IO->IOB.ChunkSendRemain == 0)
 831                         {
 832                                 IO->NextState = eSendReply;
 833                                 assert(IO->ReadDone);
 834                                 ev_io_stop(event_base, &IO->recv_event);
 835                                 PostInbound(IO);
 836                                 return;
 837                         }
 838                         else
 839                                 return;
 840                 }
 841                 break;
 842         default:
 843                 nbytes = StrBuf_read_one_chunk_callback(IO->RecvBuf.fd,
 844                                                         0,
 845                                                         &IO->RecvBuf);
 846                 break;
 847         }
 848
 849 #ifdef BIGBAD_IODBG
 850         {
 851                 long nbytes;
 852                 int rv = 0;
 853                 char fn [SIZ];
 854                 FILE *fd;
 855                 const char *pch = ChrPtr(IO->RecvBuf.Buf);
 856                 const char *pchh = IO->RecvBuf.ReadWritePointer;
 857
 858                 if (pchh == NULL)
 859                         pchh = pch;
 860
 861                 nbytes = StrLength(IO->RecvBuf.Buf) - (pchh - pch);
 862                 snprintf(fn, SIZ, "/tmp/foolog_ev_%s.%d",
 863                          ((CitContext*)(IO->CitContext))->ServiceName,
 864                          IO->SendBuf.fd);
 865
 866                 fd = fopen(fn, "a+");
 867                 fprintf(fd, "Read: BufSize: %ld BufContent: [",
 868                         nbytes);
 869                 rv = fwrite(pchh, nbytes, 1, fd);
 870                 if (!rv) printf("failed to write debug to %s!\n", fn);
 871                 fprintf(fd, "]\n");
 872                 fclose(fd);
 873         }
 874 #endif
 875         if (nbytes > 0) {
 876                 HandleInbound(IO);
 877         } else if (nbytes == 0) {
 878                 StopClientWatchers(IO, 1);
 879                 SetNextTimeout(IO, 0.01);
 880                 return;
 881         } else if (nbytes == -1) {
 882                 if (errno != EAGAIN) {
 883                         // FD is gone. kick it.
 884                         StopClientWatchers(IO, 1);
 885                         EV_syslog(LOG_DEBUG,
 886                                   "IO_recv_callback(): Socket Invalid! [%d] [%s] [%d]\n",
 887                                   errno, strerror(errno), IO->SendBuf.fd);
 888                         StrBufPrintf(IO->ErrMsg,
 889                                      "Socket Invalid! [%s]",
 890                                      strerror(errno));
 891                         SetNextTimeout(IO, 0.01);
 892                 }
 893                 return;
 894         }
 895 }
 896
 897 void
 898 IO_postdns_callback(struct ev_loop *loop, ev_idle *watcher, int revents)
 899 {
 900         AsyncIO *IO = watcher->data;
 901
 902         SetEVState(IO, eCaresFinished);
 903         IO->Now = ev_now(event_base);
 904         EV_syslog(LOG_DEBUG, "event: %s\n", __FUNCTION__);
 905         become_session(IO->CitContext);
 906         assert(IO->DNS.Query->PostDNS);
 907         switch (IO->DNS.Query->PostDNS(IO))
 908         {
 909         case eAbort:
 910                 assert(IO->DNS.Fail);
 911                 switch (IO->DNS.Fail(IO)) {
 912                 case eAbort:
 913 ////                    StopClientWatchers(IO);
 914                         ShutDownCLient(IO);
 915                 default:
 916                         break;
 917                 }
 918         default:
 919                 break;
 920         }
 921 }
 922
 923
 924 eNextState EvConnectSock(AsyncIO *IO,
 925                          double conn_timeout,
 926                          double first_rw_timeout,
 927                          int ReadFirst)
 928 {
 929         struct sockaddr_in egress_sin;
 930         int fdflags;
 931         int rc = -1;
 932
 933         SetEVState(IO, eIOConnectSock);
 934         become_session(IO->CitContext);
 935
 936         if (ReadFirst) {
 937                 IO->NextState = eReadMessage;
 938         }
 939         else {
 940                 IO->NextState = eSendReply;
 941         }
 942
 943         IO->SendBuf.fd = IO->RecvBuf.fd =
 944                 socket(
 945                         (IO->ConnectMe->IPv6)?PF_INET6:PF_INET,
 946                         SOCK_STREAM,
 947                         IPPROTO_TCP);
 948
 949         if (IO->SendBuf.fd < 0) {
 950                 EV_syslog(LOG_ERR,
 951                           "EVENT: socket() failed: %s\n",
 952                           strerror(errno));
 953
 954                 StrBufPrintf(IO->ErrMsg,
 955                              "Failed to create socket: %s",
 956                              strerror(errno));
 957                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 958                 return eAbort;
 959         }
 960         fdflags = fcntl(IO->SendBuf.fd, F_GETFL);
 961         if (fdflags < 0) {
 962                 EV_syslog(LOG_ERR,
 963                           "EVENT: unable to get socket %d flags! %s \n",
 964                           IO->SendBuf.fd,
 965                           strerror(errno));
 966                 StrBufPrintf(IO->ErrMsg,
 967                              "Failed to get socket %d flags: %s",
 968                              IO->SendBuf.fd,
 969                              strerror(errno));
 970                 close(IO->SendBuf.fd);
 971                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 972                 return eAbort;
 973         }
 974         fdflags = fdflags | O_NONBLOCK;
 975         if (fcntl(IO->SendBuf.fd, F_SETFL, fdflags) < 0) {
 976                 EV_syslog(
 977                         LOG_ERR,
 978                         "EVENT: unable to set socket %d nonblocking flags! %s \n",
 979                         IO->SendBuf.fd,
 980                         strerror(errno));
 981                 StrBufPrintf(IO->ErrMsg,
 982                              "Failed to set socket flags: %s",
 983                              strerror(errno));
 984                 close(IO->SendBuf.fd);
 985                 IO->SendBuf.fd = IO->RecvBuf.fd = 0;
 986                 return eAbort;
 987         }
 988 /* TODO: maye we could use offsetof() to calc the position of data...
 989  * http://doc.dvgu.ru/devel/ev.html#associating_custom_data_with_a_watcher
 990  */
 991         ev_io_init(&IO->recv_event, IO_recv_callback, IO->RecvBuf.fd, EV_READ);
 992         IO->recv_event.data = IO;
 993         ev_io_init(&IO->send_event, IO_send_callback, IO->SendBuf.fd, EV_WRITE);
 994         IO->send_event.data = IO;
 995
 996         ev_timer_init(&IO->conn_fail, IO_connfail_callback, conn_timeout, 0);
 997         IO->conn_fail.data = IO;
 998         ev_timer_init(&IO->rw_timeout, IO_Timeout_callback, first_rw_timeout,0);
 999         IO->rw_timeout.data = IO;
1000
1001
1002
1003
1004         /* for debugging you may bypass it like this:
1005          * IO->Addr.sin_addr.s_addr = inet_addr("127.0.0.1");
1006          * ((struct sockaddr_in)IO->ConnectMe->Addr).sin_addr.s_addr =
1007          *   inet_addr("127.0.0.1");
1008          */
1009         if (IO->ConnectMe->IPv6) {
1010                 rc = connect(IO->SendBuf.fd,
1011                              &IO->ConnectMe->Addr,
1012                              sizeof(struct sockaddr_in6));
1013         }
1014         else {
1015                 /* If citserver is bound to a specific IP address on the host, make
1016                  * sure we use that address for outbound connections.
1017                  */
1018
1019                 memset(&egress_sin, 0, sizeof(egress_sin));
1020                 egress_sin.sin_family = AF_INET;
1021                 if (!IsEmptyStr(config.c_ip_addr)) {
1022                         egress_sin.sin_addr.s_addr = inet_addr(config.c_ip_addr);
1023                         if (egress_sin.sin_addr.s_addr == !INADDR_ANY) {
1024                                 egress_sin.sin_addr.s_addr = INADDR_ANY;
1025                         }
1026
1027                         /* If this bind fails, no problem; we can still use INADDR_ANY */
1028                         bind(IO->SendBuf.fd, (struct sockaddr *)&egress_sin, sizeof(egress_sin));
1029                 }
1030                 rc = connect(IO->SendBuf.fd,
1031                              (struct sockaddr_in *)&IO->ConnectMe->Addr,
1032                              sizeof(struct sockaddr_in));
1033         }
1034
1035         if (rc >= 0){
1036                 SetEVState(IO, eIOConnNow);
1037                 EV_syslog(LOG_DEBUG, "connect() = %d immediate success.\n", IO->SendBuf.fd);
1038                 set_start_callback(event_base, IO, 0);
1039                 return IO->NextState;
1040         }
1041         else if (errno == EINPROGRESS) {
1042                 SetEVState(IO, eIOConnWait);
1043                 EV_syslog(LOG_DEBUG, "connect() = %d have to wait now.\n", IO->SendBuf.fd);
1044
1045                 ev_io_init(&IO->conn_event,
1046                            IO_connestd_callback,
1047                            IO->SendBuf.fd,
1048                            EV_READ|EV_WRITE);
1049
1050                 IO->conn_event.data = IO;
1051
1052                 ev_io_start(event_base, &IO->conn_event);
1053                 ev_timer_start(event_base, &IO->conn_fail);
1054                 return IO->NextState;
1055         }
1056         else {
1057                 SetEVState(IO, eIOConnfail);
1058                 ev_idle_init(&IO->conn_fail_immediate,
1059                              IO_connfailimmediate_callback);
1060                 IO->conn_fail_immediate.data = IO;
1061                 ev_idle_start(event_base, &IO->conn_fail_immediate);
1062
1063                 EV_syslog(LOG_ERR,
1064                           "connect() = %d failed: %s\n",
1065                           IO->SendBuf.fd,
1066                           strerror(errno));
1067
1068                 StrBufPrintf(IO->ErrMsg,
1069                              "Failed to connect: %s",
1070                              strerror(errno));
1071                 return IO->NextState;
1072         }
1073         return IO->NextState;
1074 }
1075
1076 void SetNextTimeout(AsyncIO *IO, double timeout)
1077 {
1078         IO->rw_timeout.repeat = timeout;
1079         ev_timer_again (event_base,  &IO->rw_timeout);
1080 }
1081
1082
1083 eNextState ReAttachIO(AsyncIO *IO,
1084                       void *pData,
1085                       int ReadFirst)
1086 {
1087         SetEVState(IO, eIOAttach);
1088         IO->Data = pData;
1089         become_session(IO->CitContext);
1090         ev_cleanup_start(event_base, &IO->abort_by_shutdown);
1091         if (ReadFirst) {
1092                 IO->NextState = eReadMessage;
1093         }
1094         else {
1095                 IO->NextState = eSendReply;
1096         }
1097         set_start_callback(event_base, IO, 0);
1098
1099         return IO->NextState;
1100 }
1101
1102 void InitIOStruct(AsyncIO *IO,
1103                   void *Data,
1104                   eNextState NextState,
1105                   IO_LineReaderCallback LineReader,
1106                   IO_CallBack DNS_Fail,
1107                   IO_CallBack SendDone,
1108                   IO_CallBack ReadDone,
1109                   IO_CallBack Terminate,
1110                   IO_CallBack DBTerminate,
1111                   IO_CallBack ConnFail,
1112                   IO_CallBack Timeout,
1113                   IO_CallBack ShutdownAbort)
1114 {
1115         IO->Data          = Data;
1116
1117         IO->CitContext    = CloneContext(CC);
1118         IO->CitContext->session_specific_data = Data;
1119         IO->CitContext->IO = IO;
1120
1121         IO->NextState     = NextState;
1122
1123         IO->SendDone      = SendDone;
1124         IO->ReadDone      = ReadDone;
1125         IO->Terminate     = Terminate;
1126         IO->DBTerminate   = DBTerminate;
1127         IO->LineReader    = LineReader;
1128         IO->ConnFail      = ConnFail;
1129         IO->Timeout       = Timeout;
1130         IO->ShutdownAbort = ShutdownAbort;
1131
1132         IO->DNS.Fail      = DNS_Fail;
1133
1134         IO->SendBuf.Buf   = NewStrBufPlain(NULL, 1024);
1135         IO->RecvBuf.Buf   = NewStrBufPlain(NULL, 1024);
1136         IO->IOBuf         = NewStrBuf();
1137         EV_syslog(LOG_DEBUG,
1138                   "EVENT: Session lives at %p IO at %p \n",
1139                   Data, IO);
1140
1141 }
1142
1143 extern int evcurl_init(AsyncIO *IO);
1144
1145 int InitcURLIOStruct(AsyncIO *IO,
1146                      void *Data,
1147                      const char* Desc,
1148                      IO_CallBack SendDone,
1149                      IO_CallBack Terminate,
1150                      IO_CallBack DBTerminate,
1151                      IO_CallBack ShutdownAbort)
1152 {
1153         IO->Data          = Data;
1154
1155         IO->CitContext    = CloneContext(CC);
1156         IO->CitContext->session_specific_data = Data;
1157         IO->CitContext->IO = IO;
1158
1159         IO->SendDone      = SendDone;
1160         IO->Terminate     = Terminate;
1161         IO->DBTerminate   = DBTerminate;
1162         IO->ShutdownAbort = ShutdownAbort;
1163
1164         strcpy(IO->HttpReq.errdesc, Desc);
1165
1166
1167         return  evcurl_init(IO);
1168
1169 }
1170
1171
1172 typedef struct KillOtherSessionContext {
1173         AsyncIO IO;
1174         AsyncIO *OtherOne;
1175 }KillOtherSessionContext;
1176
1177 eNextState KillTerminate(AsyncIO *IO)
1178 {
1179         long id;
1180         KillOtherSessionContext *Ctx = (KillOtherSessionContext*)IO->Data;
1181         EV_syslog(LOG_DEBUG, "%s Exit\n", __FUNCTION__);
1182         id = IO->ID;
1183         FreeAsyncIOContents(IO);
1184         memset(Ctx, 0, sizeof(KillOtherSessionContext));
1185         IO->ID = id; /* just for the case we want to analyze it in a coredump */
1186         free(Ctx);
1187         return eAbort;
1188
1189 }
1190
1191 eNextState KillShutdown(AsyncIO *IO)
1192 {
1193         return eTerminateConnection;
1194 }
1195
1196 eNextState KillOtherContextNow(AsyncIO *IO)
1197 {
1198         KillOtherSessionContext *Ctx = IO->Data;
1199
1200         SetEVState(IO, eKill);
1201
1202         if (Ctx->OtherOne->ShutdownAbort != NULL)
1203                 Ctx->OtherOne->ShutdownAbort(Ctx->OtherOne);
1204         return eTerminateConnection;
1205 }
1206
1207 void KillAsyncIOContext(AsyncIO *IO)
1208 {
1209         KillOtherSessionContext *Ctx;
1210
1211         Ctx = (KillOtherSessionContext*) malloc(sizeof(KillOtherSessionContext));
1212         memset(Ctx, 0, sizeof(KillOtherSessionContext));
1213
1214         InitIOStruct(&Ctx->IO,
1215                      Ctx,
1216                      eReadMessage,
1217                      NULL,
1218                      NULL,
1219                      NULL,
1220                      NULL,
1221                      KillTerminate,
1222                      NULL,
1223                      NULL,
1224                      NULL,
1225                      KillShutdown);
1226
1227         Ctx->OtherOne = IO;
1228
1229         switch(IO->NextState) {
1230         case eSendDNSQuery:
1231         case eReadDNSReply:
1232
1233         case eConnect:
1234         case eSendReply:
1235         case eSendMore:
1236         case eSendFile:
1237
1238         case eReadMessage:
1239         case eReadMore:
1240         case eReadPayload:
1241         case eReadFile:
1242                 QueueEventContext(&Ctx->IO, KillOtherContextNow);
1243                 break;
1244         case eDBQuery:
1245                 QueueDBOperation(&Ctx->IO, KillOtherContextNow);
1246                 break;
1247         case eTerminateConnection:
1248         case eAbort:
1249                 /*hm, its already dying, dunno which Queue its in... */
1250                 free(Ctx);
1251         }
1252
1253 }
1254
1255 extern int DebugEventLoopBacktrace;
1256 void EV_backtrace(AsyncIO *IO)
1257 {
1258 #ifdef HAVE_BACKTRACE
1259         void *stack_frames[50];
1260         size_t size, i;
1261         char **strings;
1262
1263         if ((IO == NULL) || (DebugEventLoopBacktrace == 0))
1264                 return;
1265         size = backtrace(stack_frames, sizeof(stack_frames) / sizeof(void*));
1266         strings = backtrace_symbols(stack_frames, size);
1267         for (i = 0; i < size; i++) {
1268                 if (strings != NULL) {
1269                         EV_syslog(LOG_ALERT, " BT %s\n", strings[i]);
1270                 }
1271                 else {
1272                         EV_syslog(LOG_ALERT, " BT %p\n", stack_frames[i]);
1273                 }
1274         }
1275         free(strings);
1276 #endif
1277 }
1278
1279
1280 ev_tstamp ctdl_ev_now (void)
1281 {
1282         return ev_now(event_base);
1283 }